Kafka module

CHANGELOG.asciidoc

@@ -68,6 +68,7 @@ https://github.com/elastic/beats/compare/v5.0.0...master[Check the HEAD diff]
 - Add system core metricset for Windows. {pull}2883[2883]
 - Add a sample Redis Kibana dashboard. {pull}2916[2916]
 - Add support for MongoDB 3.4 and WiredTiger metrics. {pull}2999[2999]
+- Add kafkka module with partition metricset. {pull}2969[2969]
 
 *Packetbeat*
 - Define `client_geoip.location` as geo_point in the mappings to be used by the GeoIP processor in the Ingest Node pipeline.

metricbeat/_meta/beat.full.yml

@@ -98,6 +98,14 @@ metricbeat.modules:
   #period: 10s
   #hosts: ["tcp://127.0.0.1:14567"]
 
+#-------------------------------- kafka Module -------------------------------
+#- module: kafka
+  #metricsets: ["partition"]
+  #enabled: true
+  #period: 10s
+  #hosts: ["localhost:9092"]
+
+
 #------------------------------- MongoDB Module ------------------------------
 #- module: mongodb
   #metricsets: ["status"]

metricbeat/docker-compose.yml

@@ -6,6 +6,7 @@ services:
       - apache
       - filebeat
       - mongodb
+      - kafka
       - mysql
       - nginx
       - postgresql
@@ -17,6 +18,8 @@ services:
       - APACHE_PORT=80
       - FILEBEAT_HOST=filebeat
       - FILEBEAT_PORT=6060
+      - KAFKA_HOST=kafka
+      - KAFKA_PORT=9092
       - NGINX_HOST=nginx
       - NGINX_PORT=80
       - REDIS_HOST=redis
@@ -54,6 +57,14 @@ services:
   filebeat:
     build: ${PWD}/module/beats/filebeat/_meta
 
+  kafka:
+    image: spotify/kafka
+    expose:
+      - 9092
+      - 2181
+    environment:
+      - ADVERTISED_HOST=kafka
+
   mongodb:
     image: mongo:3.4
 
@@ -77,3 +88,4 @@ services:
   zookeeper:
     image: jplock/zookeeper:3.4.8
 
+

metricbeat/docker-entrypoint.sh

@@ -24,10 +24,11 @@ waitFor() {
 # Main
 waitFor ${APACHE_HOST} ${APACHE_PORT} Apache
 waitFor ${FILEBEAT_HOST} ${FILEBEAT_PORT} Filebeat
+waitFor ${HAPROXY_HOST} ${HAPROXY_PORT} HAProxy
+waitFor ${KAFKA_HOST} ${KAFKA_PORT} Kafka
 waitFor ${MYSQL_HOST} ${MYSQL_PORT} MySQL
 waitFor ${NGINX_HOST} ${NGINX_PORT} Nginx
 waitFor ${POSTGRESQL_HOST} ${POSTGRESQL_PORT} Postgresql
 waitFor ${REDIS_HOST} ${REDIS_PORT} Redis
 waitFor ${ZOOKEEPER_HOST} ${ZOOKEEPER_PORT} Zookeeper
-waitFor ${HAPROXY_HOST} ${HAPROXY_PORT} HAProxy
 exec "$@"

metricbeat/docs/fields.asciidoc

@@ -19,6 +19,7 @@ grouped in the following categories:
 * <<exported-fields-common>>
 * <<exported-fields-docker>>
 * <<exported-fields-haproxy>>
+* <<exported-fields-kafka>>
 * <<exported-fields-mongodb>>
 * <<exported-fields-mysql>>
 * <<exported-fields-nginx>>
@@ -2147,6 +2148,81 @@ type: integer
 the average total session time in ms over the 1024 last requests
 
 
+[[exported-fields-kafka]]
+== kafka Fields
+
+kafka Module
+
+
+
+[float]
+== kafka Fields
+
+
+
+
+[float]
+== partition Fields
+
+partition
+
+
+
+[float]
+== offset Fields
+
+Available offsets of the given partition.
+
+
+
+[float]
+=== kafka.partition.offset.newest
+
+type: long
+
+Newest offset of the partition.
+
+
+[float]
+=== kafka.partition.offset.oldest
+
+type: long
+
+Oldest offset of the partition.
+
+
+[float]
+=== kafka.partition.partition
+
+type: long
+
+Partition id.
+
+
+[float]
+=== kafka.partition.topic
+
+type: keyword
+
+Topic name
+
+
+[float]
+=== kafka.partition.broker.id
+
+type: long
+
+Broker id
+
+
+[float]
+=== kafka.partition.broker.address
+
+type: keyword
+
+Broker address
+
+
 [[exported-fields-mongodb]]
 == MongoDB Fields
 

metricbeat/docs/modules/kafka.asciidoc

@@ -0,0 +1,37 @@
+////
+This file is generated! See scripts/docs_collector.py
+////
+
+[[metricbeat-module-kafka]]
+== kafka Module
+
+This is the kafka Module.
+
+
+
+[float]
+=== Example Configuration
+
+The kafka module supports the standard configuration options that are described
+in <<configuration-metricbeat>>. Here is an example configuration:
+
+[source,yaml]
+----
+metricbeat.modules:
+#- module: kafka
+  #metricsets: ["partition"]
+  #enabled: true
+  #period: 10s
+  #hosts: ["localhost:9092"]
+
+----
+
+[float]
+=== Metricsets
+
+The following metricsets are available:
+
+* <<metricbeat-metricset-kafka-partition,partition>>
+
+include::kafka/partition.asciidoc[]
+

metricbeat/docs/modules/kafka/partition.asciidoc

@@ -0,0 +1,19 @@
+////
+This file is generated! See scripts/docs_collector.py
+////
+
+[[metricbeat-metricset-kafka-partition]]
+include::../../../module/kafka/partition/_meta/docs.asciidoc[]
+
+
+==== Fields
+
+For a description of each field in the metricset, see the
+<<exported-fields-kafka,exported fields>> section.
+
+Here is an example document generated by this metricset:
+
+[source,json]
+----
+include::../../../module/kafka/partition/_meta/data.json[]
+----

metricbeat/include/list.go

@@ -23,6 +23,8 @@ import (
 	_ "github.com/elastic/beats/metricbeat/module/haproxy"
 	_ "github.com/elastic/beats/metricbeat/module/haproxy/info"
 	_ "github.com/elastic/beats/metricbeat/module/haproxy/stat"
+	_ "github.com/elastic/beats/metricbeat/module/kafka"
+	_ "github.com/elastic/beats/metricbeat/module/kafka/partition"
 	_ "github.com/elastic/beats/metricbeat/module/mongodb"
 	_ "github.com/elastic/beats/metricbeat/module/mongodb/status"
 	_ "github.com/elastic/beats/metricbeat/module/mysql"

metricbeat/metricbeat.full.yml

@@ -98,6 +98,14 @@ metricbeat.modules:
   #period: 10s
   #hosts: ["tcp://127.0.0.1:14567"]
 
+#-------------------------------- kafka Module -------------------------------
+#- module: kafka
+  #metricsets: ["partition"]
+  #enabled: true
+  #period: 10s
+  #hosts: ["localhost:9092"]
+
+
 #------------------------------- MongoDB Module ------------------------------
 #- module: mongodb
   #metricsets: ["status"]

metricbeat/metricbeat.template-es2x.json

@@ -917,6 +917,44 @@
             }
           }
         },
+        "kafka": {
+          "properties": {
+            "partition": {
+              "properties": {
+                "broker": {
+                  "properties": {
+                    "address": {
+                      "ignore_above": 1024,
+                      "index": "not_analyzed",
+                      "type": "string"
+                    },
+                    "id": {
+                      "type": "long"
+                    }
+                  }
+                },
+                "offset": {
+                  "properties": {
+                    "newest": {
+                      "type": "long"
+                    },
+                    "oldest": {
+                      "type": "long"
+                    }
+                  }
+                },
+                "partition": {
+                  "type": "long"
+                },
+                "topic": {
+                  "ignore_above": 1024,
+                  "index": "not_analyzed",
+                  "type": "string"
+                }
+              }
+            }
+          }
+        },
         "meta": {
           "properties": {
             "cloud": {

metricbeat/metricbeat.template.json

@@ -919,6 +919,42 @@
             }
           }
         },
+        "kafka": {
+          "properties": {
+            "partition": {
+              "properties": {
+                "broker": {
+                  "properties": {
+                    "address": {
+                      "ignore_above": 1024,
+                      "type": "keyword"
+                    },
+                    "id": {
+                      "type": "long"
+                    }
+                  }
+                },
+                "offset": {
+                  "properties": {
+                    "newest": {
+                      "type": "long"
+                    },
+                    "oldest": {
+                      "type": "long"
+                    }
+                  }
+                },
+                "partition": {
+                  "type": "long"
+                },
+                "topic": {
+                  "ignore_above": 1024,
+                  "type": "keyword"
+                }
+              }
+            }
+          }
+        },
         "meta": {
           "properties": {
             "cloud": {

metricbeat/module/kafka/_meta/config.yml

@@ -0,0 +1,6 @@
+#- module: kafka
+  #metricsets: ["partition"]
+  #enabled: true
+  #period: 10s
+  #hosts: ["localhost:9092"]
+

metricbeat/module/kafka/_meta/docs.asciidoc

@@ -0,0 +1,4 @@
+== kafka Module
+
+This is the kafka Module.
+

metricbeat/module/kafka/_meta/fields.yml

@@ -0,0 +1,10 @@
+- key: kafka
+  title: "kafka"
+  description: >
+    kafka Module
+  short_config: false
+  fields:
+    - name: kafka
+      type: group
+      description: >
+      fields:

metricbeat/module/kafka/doc.go

@@ -0,0 +1,20 @@
+/*
+Package kafka is a Metricbeat module that contains MetricSets.
+
+Kafka is organised as following
+
+- Topic
+- Partition
+- Producer
+- Consumer
+- Consumer Groups
+- Broker
+
+Notes
+- Topics has a list of partitions
+- Each partition has an offset
+- Topic can be across brokers
+- Each broker has a list of paritions
+
+*/
+package kafka