Modified README and fixed minor bugs

README.md

@@ -5,12 +5,12 @@
 This tool allows exporting the often missed Kubernetes events to various outputs so that  they can be used for 
 observability or alerting purposes. You won't believe what you are missing. 
 
-# Deployment
+## Deployment
 
 Head on to `deploy/` folder and apply the YAMLs in the given filename order. Do not forget to modify the 
 `deploy/01-config.yaml` file to your configuration needs. The additional information for configuration is as follows:
 
-# Configuration
+## Configuration
 
 Configuration is done via a YAML file, when run in Kubernetes, it's in ConfigMap. The tool watches all the events and
 user has to option to filter out some events, according to their properties. Critical events can be routed to alerting
@@ -25,8 +25,8 @@ route:
     # This route allows dumping all events because it has no fields to match and no drop rules.
     - match:
       - receiver: dump
-    # This starts another route, drops all the events in *test* namespaces and Normal events for capturing 
-    # critical events 
+    # This starts another route, drops all the events in *test* namespaces and Normal events 
+    # for capturing critical events 
     - drop:
       - namespace: "*test*"
       - type: "Normal"
@@ -37,7 +37,7 @@ route:
         kind: "Pod|Deployment|ReplicaSet"
         labels:
           version: "dev"    
-        reciever: "slack"    
+        receiver: "slack"    
 receivers:
 # See below for configuring the receivers
 ```
@@ -49,9 +49,8 @@ receivers:
 * A route can have many sub-routes, forming a tree. 
 * Routing starts from the root route.
 
-# Receivers
 
-## Opsgenie
+### Opsgenie
 
 [Opsgenie](https://www.opsgenie.com) is an alerting and on-call management tool. kubernetes-event-exporter can push to 
 events to Opsgenie so that you can notify the on-call when something critical happens. Alerting should be precise and 
@@ -63,7 +62,7 @@ filtering out Normal type of events, while some additional filtering can help. B
 receivers:
   - name: "alerts"
     opsgenie:
-      apikey: xxx
+      apiKey: xxx
       priority: "P3"
       message: "Event {{ .Reason }} for {{ .InvolvedObject.Namespace }}/{{ .InvolvedObject.Name }} on K8s cluster"
       alias: "{{ .UID }}"
@@ -75,7 +74,7 @@ receivers:
         - "{{ .InvolvedObject.Name }}"
 ```
 
-# Webhooks/HTTP
+### Webhooks/HTTP
 
 Webhooks are te easiest way of integrating this tool to external systems. It allows templating & custom headers
 which allows you to push events to many possible sources out there. See [Customizing Payload] for more information.
@@ -92,7 +91,7 @@ receivers:
       layout: # Optional
 ```
 
-# Elasticsearch
+### Elasticsearch
 
 [Elasticsearch](https://www.elastic.co/) is a full-text, distributed search engine which can also do powerful aggregations.
 You may decide to push all events to Elasticsearch and do some interesting queries over time to find out which images
@@ -104,18 +103,19 @@ in Kubecon to see what else you can do with aggregation and reporting.
 receivers:
   - name: "dump"
     elasticsearch:
-      addresses:
+      hosts:
       - http://localhost:9200
       index: kube-events            
       username: # optional
       password: # optional
       cloudID: # optional
       apiKey: # optional
-      useEventID: true|false # If set to true, it allows updating the same document in ES (might be useful handling count)
+      # If set to true, it allows updating the same document in ES (might be useful handling count)
+      useEventID: true|false
   	  layout: # Optional
 ```
 
-# Slack
+### Slack
 
 Slack is a cloud-based instant messaging platform where many people use it for integrations and getting notified by 
 software such as Jira, Opsgenie, Google Calendar etc. and even some implement ChatOps on it. This tool also allows
@@ -138,7 +138,7 @@ receivers:
         
 ```
 
-# Kinesis
+### Kinesis
 
 Kinesis is an AWS service allows to collect high throughput messages and allow it to be used in stream processing.
 
@@ -152,7 +152,7 @@ receivers:
       layout: # Optional
 ```
 
-# SNS
+### SNS
 
 SNS is an AWS service for highly durable pub/sub messaging system.
 
@@ -161,12 +161,12 @@ SNS is an AWS service for highly durable pub/sub messaging system.
 receivers:
   - name: "sns"
     sns:
-      topicARN: "arn:sns."
-      region: us-west-2
+      topicARN: "arn:aws:sns:us-east-1:1234567890123456:mytopic"
+      region: "us-west-2"
       layout: # Optional
 ```
 
-# SQS
+### SQS
 
 SQS is an AWS service for message queuing that allows high throughput messaging.
 
@@ -180,7 +180,7 @@ receivers:
       layout: # Optional
 ```
 
-# File
+### File
 
 For some debugging purposes, you might want to push the events to files. Or you can already have a logging tool that can
 ingest these files and it might be a good idea to just use plain old school files as an integration point.
@@ -194,7 +194,7 @@ receivers:
       layout: # Optional
 ```
 
-## Customizing Payload
+### Customizing Payload
 
 Some receivers allow customizing the payload. This can be useful to integrate it to external systems that require 
 the data be in some format. It is designed to reduce the need for code writing. It allows mapping an event using
@@ -223,4 +223,15 @@ receivers:
           component: "{{ .Source.Component }}"
           host: "{{ .Source.Host }}"
           labels: "{{ toJson .InvolvedObject.Labels}}" 
-```
+```
+
+### Planned Receivers
+
+- Big Query
+- PubSub
+- AWS Firehose
+- Splunk
+- Kafka
+- Redis
+- Logstash
+- Console

config.yaml

@@ -1,11 +1,33 @@
-loglevel: DEBUG
+logLevel: info
 route:
-  match:
-    - receiver: "alerts"
+  # Main route
+  routes:
+    # This route allows dumping all events because it has no fields to match and no drop rules.
+    - match:
+        - receiver: "dump"
+    # This starts another route, drops all the events in *test* namespaces and Normal events 
+    # for capturing critical events 
+    - drop:
+        - namespace: "*test*"
+        - type: "Normal"
+      match:
+        - receiver: "alert"
+        - receiver: "pipe"
+    # This a final route for user messages
+    - match:
+      - kind: "Pod|Deployment|ReplicaSet"
+        labels:
+          version: "dev"
+        receiver: "slack"
 receivers:
-  - name: "alerts"
+  - name: "dump"
+    elasticsearch:
+      hosts:
+        - "http://localhost:9200"
+      index: "kube-events"
+  - name: "alert"
     opsgenie:
-      apikey: xxx
+      apiKey: ""
       priority: "P3"
       message: "Event {{ .Reason }} for {{ .InvolvedObject.Namespace }}/{{ .InvolvedObject.Name }} on K8s cluster"
       alias: "{{ .UID }}"
@@ -15,18 +37,31 @@ receivers:
         - "{{ .Reason }}"
         - "{{ .InvolvedObject.Kind }}"
         - "{{ .InvolvedObject.Name }}"
-  - name: appMetric
-    kinesis:
-      region: us-west-2
-      streamName: event-pipeline
+  - name: "slack"
+    slack:
+      token: ""
+      channel: "#mustafa-test"
+      message: "Received a Kubernetes Event {{ .Message}}"
+      fields:
+        message: "{{ .Message }}"
+        namespace: "{{ .Namespace }}"
+        reason: "{{ .Reason }}"
+        object: "{{ .Namespace }}"
+  - name: "pipe"
+    webhook:
+      endpoint: "http://localhost:3000"
+      headers:
+        X-API-KEY: "123-456-OPSGENIE-789-ABC"
+        User-Agent: "kube-event-exporter 1.0"
+      streamName: "applicationMetric"
       layout:
-        region: "us-west-2"
-        eventType: "kubeevent"
+        endpoint: "localhost2"
+        eventType: "kube-event"
         createdAt: "{{ .GetTimestampMs }}"
         details:
           message: "{{ .Message }}"
           reason: "{{ .Reason }}"
-          type: "{{ .Type }}"
+          tip: "{{ .Type }}"
           count: "{{ .Count }}"
           kind: "{{ .InvolvedObject.Kind }}"
           name: "{{ .InvolvedObject.Name }}"

deploy/02-deployment.yaml

@@ -14,7 +14,7 @@ spec:
       serviceAccountName: event-exporter
       containers:
         - name: event-exporter
-          image: docker.pkg.github.com/opsgenie/kubernetes-event-exporter/exporter:0.1
+          image: docker.pkg.github.com/opsgenie/kubernetes-event-exporter/exporter:0.2
           imagePullPolicy: IfNotPresent
           args:
             - -conf=/data/config.yaml

main.go

@@ -20,27 +20,36 @@ var (
 
 func main() {
 	flag.Parse()
+	b, err := ioutil.ReadFile(*conf)
+
+	if err != nil {
+		log.Fatal().Err(err).Msg("cannot read config file")
+	}
+
+	var cfg exporter.Config
+	err = yaml.Unmarshal(b, &cfg)
+	if err != nil {
+		log.Fatal().Err(err).Msg("cannot parse config to YAML")
+	}
 
 	log.Logger = log.With().Caller().Logger().Output(zerolog.ConsoleWriter{
 		Out:        os.Stdout,
 		TimeFormat: time.RFC3339,
 	}).Level(zerolog.DebugLevel)
 
+	if cfg.LogLevel != "" {
+		level, err := zerolog.ParseLevel(cfg.LogLevel)
+		if err != nil {
+			log.Fatal().Err(err).Str("level", cfg.LogLevel).Msg("Invalid log level")
+		}
+		log.Logger = log.Logger.Level(level)
+	}
+
 	kubeconfig, err := kube.GetKubernetesConfig()
 	if err != nil {
 		log.Fatal().Err(err).Msg("cannot get kubeconfig")
 	}
 
-	b, err := ioutil.ReadFile(*conf)
-	if err != nil {
-		log.Fatal().Err(err).Msg("cannot read config file")
-	}
-	var cfg exporter.Config
-	err = yaml.Unmarshal(b, &cfg)
-	if err != nil {
-		log.Fatal().Err(err).Msg("cannot parse config to YAML")
-	}
-
 	engine := exporter.NewEngine(&cfg, &exporter.ChannelBasedReceiverRegistry{})
 	w := kube.NewEventWatcher(kubeconfig, engine.OnEvent)
 	w.Start()

pkg/exporter/channel_registry.go

@@ -18,6 +18,9 @@ type ChannelBasedReceiverRegistry struct {
 
 func (r *ChannelBasedReceiverRegistry) SendEvent(name string, event *kube.EnhancedEvent) {
 	ch := r.ch[name]
+	if ch == nil {
+		log.Error().Str("name", name).Msg("There is no channel")
+	}
 
 	go func() {
 		ch <- *event
@@ -40,9 +43,10 @@ func (r *ChannelBasedReceiverRegistry) Register(name string, receiver sinks.Sink
 		for {
 			select {
 			case ev := <-ch:
+				log.Debug().Str("sink", name).Str("event", ev.Message).Msg("sending event to sink")
 				err := receiver.Send(context.Background(), &ev)
 				if err != nil {
-					log.Debug().Err(err).Str("sink", name).Str("event", string(ev.UID)).Msg("Cannot send event")
+					log.Debug().Err(err).Str("sink", name).Str("event", ev.Message).Msg("Cannot send event")
 				}
 			case <-exitCh:
 				log.Info().Str("receiver", name).Msg("Killing the receiver")

pkg/exporter/config.go

@@ -9,8 +9,9 @@ type Config struct {
 	// Route is the top route that the events will match
 	// TODO: There is currently a tight coupling with route and config, but not with receiver config and sink so
 	// TODO: I am not sure what to do here.
-	Route     Route
-	Receivers []sinks.ReceiverConfig
+	LogLevel  string                 `yaml:"logLevel"`
+	Route     Route                  `yaml:"route"`
+	Receivers []sinks.ReceiverConfig `yaml:"receivers"`
 }
 
 func (c *Config) Validate() error {

pkg/exporter/route.go

@@ -29,7 +29,6 @@ func (r *Route) ProcessEvent(ev *kube.EnhancedEvent, registry ReceiverRegistry)
 			}
 		} else {
 			matchesAll = false
-			break
 		}
 	}
 

pkg/sinks/elasticsearch.go

@@ -11,19 +11,19 @@ import (
 
 type ElasticsearchConfig struct {
 	// Connection specific
-	Addresses []string
-	Username  string
-	Password  string
-	CloudID   string
-	APIKey    string
+	Hosts    []string `yaml:"hosts"`
+	Username string   `yaml:"username"`
+	Password string   `yaml:"password"`
+	CloudID  string   `yaml:"cloudID"`
+	APIKey   string   `yaml:"apiKey"`
 	// Indexing preferences
-	UseEventID bool
-	Index      string
+	UseEventID bool   `yaml:"useEventID"`
+	Index      string `yaml:"index"`
 }
 
 func NewElasticsearch(cfg *ElasticsearchConfig) (*Elasticsearch, error) {
 	client, err := elasticsearch.NewClient(elasticsearch.Config{
-		Addresses: cfg.Addresses,
+		Addresses: cfg.Hosts,
 		Username:  cfg.Username,
 		Password:  cfg.Password,
 		CloudID:   cfg.CloudID,

pkg/sinks/file.go

@@ -8,8 +8,8 @@ import (
 )
 
 type FileConfig struct {
-	Path   string
-	Layout map[string]interface{}
+	Path   string                 `yaml:"file"`
+	Layout map[string]interface{} `yaml:"layout"`
 }
 
 func (f *FileConfig) Validate() error {

pkg/sinks/kinesis.go

@@ -10,9 +10,9 @@ import (
 )
 
 type KinesisConfig struct {
-	StreamName string
-	Region     string
-	Layout     map[string]interface{}
+	StreamName string                 `yaml:"streamName"`
+	Region     string                 `yaml:"region"`
+	Layout     map[string]interface{} `yaml:"layout"`
 }
 
 type KinesisSink struct {

pkg/sinks/opsgenie.go

@@ -8,14 +8,14 @@ import (
 )
 
 type OpsgenieConfig struct {
-	ApiKey      string
-	URL         client.ApiUrl
-	Priority    string
-	Message     string
-	Alias       string
-	Description string
-	Tags        []string
-	Details     map[string]string
+	ApiKey      string            `yaml:"apiKey"`
+	URL         client.ApiUrl     `yaml:"URL"`
+	Priority    string            `yaml:"priority"`
+	Message     string            `yaml:"message"`
+	Alias       string            `yaml:"alias"`
+	Description string            `yaml:"description"`
+	Tags        []string          `yaml:"tags"`
+	Details     map[string]string `yaml:"details"`
 }
 
 type OpsgenieSink struct {

pkg/sinks/reciever.go

@@ -4,16 +4,16 @@ import "errors"
 
 // Receiver allows receiving
 type ReceiverConfig struct {
-	Name          string
-	Webhook       *WebhookConfig
-	InMemory      *InMemoryConfig
-	File          *FileConfig
-	Elasticsearch *ElasticsearchConfig
-	Kinesis       *KinesisConfig
-	Opsgenie      *OpsgenieConfig
-	SQS           *SQSConfig
-	SNS           *SNSConfig
-	Slack         *SlackConfig
+	Name          string               `yaml:"name"`
+	InMemory      *InMemoryConfig      `yaml:"inMemory"`
+	Webhook       *WebhookConfig       `yaml:"webhook"`
+	File          *FileConfig          `yaml:"file"`
+	Elasticsearch *ElasticsearchConfig `yaml:"elasticsearch"`
+	Kinesis       *KinesisConfig       `yaml:"kinesis"`
+	Opsgenie      *OpsgenieConfig      `yaml:"opsgenie"`
+	SQS           *SQSConfig           `yaml:"sqs"`
+	SNS           *SNSConfig           `yaml:"sns"`
+	Slack         *SlackConfig         `yaml:"slack"`
 }
 
 func (r *ReceiverConfig) Validate() error {

pkg/sinks/slack.go

@@ -4,13 +4,14 @@ import (
 	"context"
 	"github.com/nlopes/slack"
 	"github.com/opsgenie/kubernetes-event-exporter/pkg/kube"
+	"github.com/rs/zerolog/log"
 )
 
 type SlackConfig struct {
-	Token   string
-	Channel string
-	Message string
-	Fields  map[string]string
+	Token   string            `yaml:"token"`
+	Channel string            `yaml:"channel"`
+	Message string            `yaml:"message"`
+	Fields  map[string]string `yaml:"fields"`
 }
 
 type SlackSink struct {
@@ -54,7 +55,8 @@ func (s *SlackSink) Send(ctx context.Context, ev *kube.EnhancedEvent) error {
 		options = append(options, slack.MsgOptionAttachments(slack.Attachment{Fields: fields}))
 	}
 
-	_, _, _, err = s.client.SendMessageContext(ctx, channel, options...)
+	_ch, _ts, _text, err := s.client.SendMessageContext(ctx, channel, options...)
+	log.Debug().Str("ch", _ch).Str("ts", _ts).Str("text", _text).Err(err).Msg("Slack Response")
 	return err
 }
 

pkg/sinks/sns.go

@@ -9,9 +9,9 @@ import (
 )
 
 type SNSConfig struct {
-	TopicARN string
-	Region   string
-	Layout   map[string]interface{}
+	TopicARN string                 `yaml:"topicARN"`
+	Region   string                 `yaml:"region"`
+	Layout   map[string]interface{} `yaml:"layout"`
 }
 
 type SNSSink struct {

pkg/sinks/sqs.go

@@ -9,9 +9,9 @@ import (
 )
 
 type SQSConfig struct {
-	QueueName string
-	Region    string
-	Layout    map[string]interface{}
+	QueueName string                 `yaml:"queueName"`
+	Region    string                 `yaml:"region"`
+	Layout    map[string]interface{} `yaml:"layout"`
 }
 
 type SQSSink struct {
@@ -58,7 +58,6 @@ func (s *SQSSink) Send(ctx context.Context, ev *kube.EnhancedEvent) error {
 	return err
 }
 
-
 func (s *SQSSink) Close() {
 	// No-op
 }

pkg/sinks/webhook.go

@@ -10,9 +10,9 @@ import (
 )
 
 type WebhookConfig struct {
-	Endpoint string
-	Layout   map[string]interface{}
-	Headers  map[string]string
+	Endpoint string                 `yaml:"endpoint"`
+	Layout   map[string]interface{} `yaml:"layout"`
+	Headers  map[string]string      `yaml:"headers"`
 }
 
 func NewWebhook(cfg *WebhookConfig) (Sink, error) {