Adding changes for kafka tailer integration

Author: hartfordfive

CONFIG.md

@@ -200,6 +200,37 @@ input:
 
 This configuration example may be found in the examples directory [here](example/config_logstash_http_input_ipv6.yml).
 
+### Kafka Input Type
+
+The `grok_exporter` is also capable of consuming log entries from Kafka.  Currently, only plain-text encoded messages are supported.
+
+```yaml
+input:
+  type: kafka
+  # Version corresponding to the kafka cluster
+  kafka_version: 2.1.0
+
+  # The list of the Kafka brokers part of the Kafka cluster.  Please note that you need an instance of grok_exporter per Kafka cluster if you plan on consuming from topics from multiple clusters.
+  kafka_brokers:
+    - localhost:9092
+
+  # The list of Kafka topics to consume from.
+  kafka_topics: 
+    - grok_exporter_test
+
+  # The assignor to use, which can be either range, roundrobin, sticky (range by default)
+  kafka_partition_assignor: range
+
+  # The name of the consumer group to register as on the broker.  If not specified, the default is 'grok_exporter'
+  kafka_consumer_group_name: grok_exporter
+
+  # Indicates if the exporter should start consuming as of the most recent messages in the topic (true), or consume from the earliest messages in the topic (false).
+  kafka_consume_from_oldest: false
+```
+
+This configuration example may be found in the examples directory [here](example/config-kafka.yml).
+
+
 imports Section
 ---------------
 

config/v3/configV3.go

@@ -16,21 +16,24 @@ package v3
 
 import (
 	"fmt"
-	v2 "github.com/fstab/grok_exporter/config/v2"
-	"github.com/fstab/grok_exporter/tailer/glob"
-	"github.com/fstab/grok_exporter/template"
-	"gopkg.in/yaml.v2"
 	"os"
+	"regexp"
 	"strconv"
 	"strings"
 	"time"
+
+	v2 "github.com/fstab/grok_exporter/config/v2"
+	"github.com/fstab/grok_exporter/tailer/glob"
+	"github.com/fstab/grok_exporter/template"
+	"gopkg.in/yaml.v2"
 )
 
 const (
 	defaultRetentionCheckInterval = 53 * time.Second
 	inputTypeStdin                = "stdin"
 	inputTypeFile                 = "file"
 	inputTypeWebhook              = "webhook"
+	inputTypeKafka                = "kafka"
 	importMetricsType             = "metrics"
 	importPatternsType            = "grok_patterns"
 )
@@ -102,6 +105,12 @@ type InputConfig struct {
 	WebhookFormat              string        `yaml:"webhook_format,omitempty"`
 	WebhookJsonSelector        string        `yaml:"webhook_json_selector,omitempty"`
 	WebhookTextBulkSeparator   string        `yaml:"webhook_text_bulk_separator,omitempty"`
+	KafkaVersion               string        `yaml:"kafka_version,omitempty"`
+	KafkaBrokers               []string      `yaml:"kafka_brokers,omitempty"`
+	KafkaTopics                []string      `yaml:"kafka_topics,omitempty"`
+	KafkaPartitionAssignor     string        `yaml:"kafka_partition_assignor,omitempty"`
+	KafkaConsumerGroupName     string        `yaml:"kafka_consumer_group_name,omitempty"`
+	KafkaConsumeFromOldest     bool          `yaml:"kafka_consume_from_oldest,omitempty"`
 }
 
 type GrokPatternsConfig []string
@@ -268,6 +277,19 @@ func (c *InputConfig) addDefaults() {
 			c.WebhookTextBulkSeparator = "\n\n"
 		}
 	}
+	if c.Type == inputTypeKafka {
+		c.KafkaConsumeFromOldest = false
+
+		if c.KafkaPartitionAssignor == "" {
+			c.KafkaPartitionAssignor = "range"
+		}
+		if c.KafkaVersion == "" {
+			c.KafkaVersion = "2.1.0"
+		}
+		if c.KafkaConsumerGroupName == "" {
+			c.KafkaConsumerGroupName = "grok_exporter"
+		}
+	}
 }
 
 func (c *GrokPatternsConfig) addDefaults() {}
@@ -403,6 +425,26 @@ func (c *InputConfig) validate() error {
 		if c.WebhookFormat == "text_bulk" && c.WebhookTextBulkSeparator == "" {
 			return fmt.Errorf("invalid input configuration: 'input.webhook_text_bulk_separator' is required for input type \"webhook\" and webhook_format \"text_bulk\"")
 		}
+	case c.Type == inputTypeKafka:
+		if len(c.KafkaBrokers) == 0 {
+			return fmt.Errorf("invalid input configuration: Kafka 'input.kafka_brokers' cannot be empty")
+		}
+		if len(c.KafkaTopics) == 0 {
+			return fmt.Errorf("invalid input configuration: Kafka 'input.kafka_topics' cannot be empty")
+		}
+
+		matched, _ := regexp.MatchString(`^[0-9]\.[0-9]\.[0-9]$`, c.KafkaVersion)
+		if !matched {
+			return fmt.Errorf("invalid input configuration: Kafka 'input.kafka_version' must a valid semantic version X.Y.Z")
+		}
+
+		versionParts := strings.Split(c.KafkaVersion, ".")
+		vMajor, vMajorErr := strconv.Atoi(versionParts[0])
+		vMinor, vMinorErr := strconv.Atoi(versionParts[1])
+		if vMajorErr != nil && vMinorErr != nil && vMajor < 1 && vMinor < 8 {
+			return fmt.Errorf("invalid input configuration: Kafka 'input.kafka_version' must be >= 0.8.0")
+		}
+
 	default:
 		return fmt.Errorf("unsupported 'input.type': %v", c.Type)
 	}

example/config-kafka.yml

@@ -0,0 +1,25 @@
+global:
+  config_version: 3
+input:
+  type: kafka
+  kafka_version: 2.1.0
+  kafka_brokers:
+    - localhost:9092
+  kafka_topics: 
+    - grok_exporter_test
+  kafka_consumer_group_name: grok_exporter
+  kafka_consume_from_oldest: true
+imports:
+  - type: grok_patterns
+    dir: ./logstash-patterns-core/patterns
+metrics:
+  - type: counter
+    name: test_strings
+    help: Test string detected.
+    match: 'test'
+    labels:
+      logfile: '{{base .logfile}}'
+server:
+  protocol: http
+  port: 9144
+

go.mod

@@ -1,13 +1,18 @@
 module github.com/fstab/grok_exporter
 
 require (
+	github.com/Shopify/sarama v1.27.0
 	github.com/bitly/go-simplejson v0.5.0
 	github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 // indirect
+	github.com/ofabry/go-callvis v0.6.1 // indirect
 	github.com/prometheus/client_golang v1.7.1
 	github.com/prometheus/client_model v0.2.0
 	github.com/prometheus/common v0.10.0
 	github.com/sirupsen/logrus v1.6.0
 	golang.org/x/exp v0.0.0-20191227195350-da58074b4299
+	golang.org/x/image v0.0.0-20200801110659-972c09e46d76 // indirect
+	golang.org/x/tools v0.0.0-20200828161849-5deb26317202 // indirect
+	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
 	gopkg.in/yaml.v2 v2.3.0
 )