From 45cf57c7f939d8fdff02a72b61cd2e2ff9450824 Mon Sep 17 00:00:00 2001 From: Mustafa Akin Date: Mon, 9 Dec 2019 14:09:26 +0300 Subject: [PATCH] Modified README and fixed minor bugs --- README.md | 53 ++++++++++++++++------------ config.yaml | 59 +++++++++++++++++++++++++------- deploy/02-deployment.yaml | 2 +- main.go | 29 ++++++++++------ pkg/exporter/channel_registry.go | 6 +++- pkg/exporter/config.go | 5 +-- pkg/exporter/route.go | 1 - pkg/sinks/elasticsearch.go | 16 ++++----- pkg/sinks/file.go | 4 +-- pkg/sinks/kinesis.go | 6 ++-- pkg/sinks/opsgenie.go | 16 ++++----- pkg/sinks/reciever.go | 20 +++++------ pkg/sinks/slack.go | 12 ++++--- pkg/sinks/sns.go | 6 ++-- pkg/sinks/sqs.go | 7 ++-- pkg/sinks/webhook.go | 6 ++-- 16 files changed, 154 insertions(+), 94 deletions(-) diff --git a/README.md b/README.md index 8104a63..83b68e1 100644 --- a/README.md +++ b/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}}" -``` \ No newline at end of file +``` + +### Planned Receivers + +- Big Query +- PubSub +- AWS Firehose +- Splunk +- Kafka +- Redis +- Logstash +- Console \ No newline at end of file diff --git a/config.yaml b/config.yaml index 68e7235..5b8374a 100644 --- a/config.yaml +++ b/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 }}" diff --git a/deploy/02-deployment.yaml b/deploy/02-deployment.yaml index 60074ed..f13bb63 100644 --- a/deploy/02-deployment.yaml +++ b/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 diff --git a/main.go b/main.go index bb2cd1a..5575bfc 100644 --- a/main.go +++ b/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() diff --git a/pkg/exporter/channel_registry.go b/pkg/exporter/channel_registry.go index f3395d5..226bce8 100644 --- a/pkg/exporter/channel_registry.go +++ b/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") diff --git a/pkg/exporter/config.go b/pkg/exporter/config.go index 81415b8..9a33404 100644 --- a/pkg/exporter/config.go +++ b/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 { diff --git a/pkg/exporter/route.go b/pkg/exporter/route.go index 89084bb..7ce9488 100644 --- a/pkg/exporter/route.go +++ b/pkg/exporter/route.go @@ -29,7 +29,6 @@ func (r *Route) ProcessEvent(ev *kube.EnhancedEvent, registry ReceiverRegistry) } } else { matchesAll = false - break } } diff --git a/pkg/sinks/elasticsearch.go b/pkg/sinks/elasticsearch.go index e6bf1a3..78b3a4a 100644 --- a/pkg/sinks/elasticsearch.go +++ b/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, diff --git a/pkg/sinks/file.go b/pkg/sinks/file.go index aa5d130..d0628ea 100644 --- a/pkg/sinks/file.go +++ b/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 { diff --git a/pkg/sinks/kinesis.go b/pkg/sinks/kinesis.go index f540ded..efc5fa5 100644 --- a/pkg/sinks/kinesis.go +++ b/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 { diff --git a/pkg/sinks/opsgenie.go b/pkg/sinks/opsgenie.go index 1059e62..2f802a4 100644 --- a/pkg/sinks/opsgenie.go +++ b/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 { diff --git a/pkg/sinks/reciever.go b/pkg/sinks/reciever.go index a0175ca..3164f07 100644 --- a/pkg/sinks/reciever.go +++ b/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 { diff --git a/pkg/sinks/slack.go b/pkg/sinks/slack.go index f0db259..9428991 100644 --- a/pkg/sinks/slack.go +++ b/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 } diff --git a/pkg/sinks/sns.go b/pkg/sinks/sns.go index f28b5bb..92c2730 100644 --- a/pkg/sinks/sns.go +++ b/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 { diff --git a/pkg/sinks/sqs.go b/pkg/sinks/sqs.go index 7ec19e3..4f0cc97 100644 --- a/pkg/sinks/sqs.go +++ b/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 } diff --git a/pkg/sinks/webhook.go b/pkg/sinks/webhook.go index 5f800e9..9d287ce 100644 --- a/pkg/sinks/webhook.go +++ b/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) {