Merge pull request #48 from seznam/dv-kafka

feat: kafka ingester

CHANGELOG.md

@@ -6,6 +6,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [v6.8.0] 2021-03-24
+### Added
+- [#48](https://github.com/seznam/slo-exporter/pull/48) New ingester module [`kafkaIngester`](docs/modules/kafka_ingester.md)
+
 ## [v6.7.1] 2021-02-15
 ### Fixed
 - [#44](https://github.com/seznam/slo-exporter/issues/44) Install missing ca-certificates to docker base image

cmd/slo_exporter.go

@@ -9,6 +9,7 @@ import (
 	"github.com/seznam/slo-exporter/pkg/dynamic_classifier"
 	"github.com/seznam/slo-exporter/pkg/envoy_access_log_server"
 	"github.com/seznam/slo-exporter/pkg/event_key_generator"
+	"github.com/seznam/slo-exporter/pkg/kafka_ingester"
 	"github.com/seznam/slo-exporter/pkg/metadata_classifier"
 	"github.com/seznam/slo-exporter/pkg/pipeline"
 	"github.com/seznam/slo-exporter/pkg/prometheus_exporter"
@@ -64,6 +65,8 @@ func moduleFactory(moduleName string, logger logrus.FieldLogger, conf *viper.Vip
 		return tailer.NewFromViper(conf, logger)
 	case "prometheusIngester":
 		return prometheus_ingester.NewFromViper(conf, logger)
+	case "kafkaIngester":
+		return kafka_ingester.NewFromViper(conf, logger)
 	case "envoyAccessLogServer":
 		return envoy_access_log_server.NewFromViper(conf, logger)
 	case "relabel":

docs/modules/kafka_ingester.md

@@ -0,0 +1,59 @@
+# Kafka ingester
+
+|                |                         |
+|----------------|-------------------------|
+| `moduleName`   | `kafkaIngester`         |
+| Module type    | `producer`              |
+| Output event   | `raw`                   |
+
+Kafka ingester generates events from Kafka messages.
+
+`moduleConfig`
+```yaml
+# Allow verbose logging of events within Kafka library. Global logger with its configured log level is used.
+logKafkaEvents: false
+# Allow logging of errors within Kafka library. Global logger with its configured log level is used.
+logKafkaErrors: true
+# List of Kafka brokers
+brokers:
+  - <string> # e.g. kafka-1.example.com:9092
+topic: ""
+groupId: ""
+# commitInterval indicates the interval at which offsets are committed to the broker.
+# If 0 (default), commits will be handled synchronously.
+commitInterval: <duration> # e.g. 0, 5s, 10m
+# retentionTime optionally sets the length of time the consumer group will be saved by the broker.
+# Default: 24h
+retentionTime: <duration>
+# fallbackStartOffset determines from whence the consumer group should begin consuming when it finds a partition without a committed offset.
+# Default: FirstOffset
+fallbackStartOffset: <LastOffset|FirstOffset>
+```
+
+
+For every received message from Kafka:
+- data in Key is ignored
+- data in Value is unmarshalled according to the schema version specified in Kafka message header `slo-exporter-schema-version` (defaults to `v1` if none specified).
+
+### Supported data schemas
+#### `v1`
+```
+{
+    "metadata": {
+        "name": "eventName"
+        ...
+    },
+    # Defaults to 1 if none specified
+    "quantity": "10",
+    "slo_classification": {
+        "app": "testApp",
+        "class": "critical",
+        "domain": "testDomain"
+    }
+}
+```
+
+Strictly speaking, none of the keys is mandatory, however please note that:
+- Event with explicitly set Quantity=0 is basically noop for Producer module. To give an example, prometheusExporter does not increment any SLO metric for such events.
+- Event with empty Metadata does not allow much logic in following modules.
+- In case you want to allow ingesting events without SLO classification, you need to make sure that all events are classified within rest of the SLO exporter pipeline.

go.mod

@@ -16,9 +16,10 @@ require (
 	github.com/prometheus/client_golang v1.6.0
 	github.com/prometheus/common v0.10.0
 	github.com/prometheus/prometheus v1.8.2-0.20200213233353-b90be6f32a33
+	github.com/segmentio/kafka-go v0.4.11
 	github.com/sirupsen/logrus v1.6.0
 	github.com/spf13/viper v1.7.0
-	github.com/stretchr/testify v1.6.0
+	github.com/stretchr/testify v1.6.1
 	golang.org/x/exp v0.0.0-20200513190911-00229845015e
 	golang.org/x/sys v0.0.0-20200828194041-157a740278f4 // indirect
 	golang.org/x/tools v0.0.0-20200828161849-5deb26317202 // indirect