[filebeat] Add a Kafka input

filebeat/_meta/fields.common.yml

@@ -150,3 +150,37 @@
       type: keyword
       description: >
         Name of organization associated with the autonomous system.
+
+    - name: kafka
+      type: group
+      fields:
+        - name: topic
+          type: keyword
+          description: >
+            Kafka topic
+
+        - name: partition
+          type: long
+          description: >
+            Kafka partition number
+
+        - name: offset
+          type: long
+          description: >
+            Kafka offset of this message
+
+        - name: key
+          type: keyword
+          description: >
+            Kafka key, corresponding to the Kafka value stored in the message
+
+        - name: block_timestamp
+          type: date
+          description: >
+            Kafka outer (compressed) block timestamp
+
+        - name: headers
+          type: array
+          description: >
+            An array of Kafka header strings for this message, in the form
+            "<key>: <value>".

filebeat/docker-compose.yml

@@ -12,6 +12,8 @@ services:
       - ES_PORT=9200
       - ES_USER=beats
       - ES_PASS=testing
+      - KAFKA_HOST=kafka
+      - KAFKA_PORT=9092
       - KIBANA_HOST=kibana
       - KIBANA_PORT=5601
     working_dir: /go/src/github.com/elastic/beats/filebeat
@@ -27,6 +29,7 @@ services:
     image: busybox
     depends_on:
       elasticsearch: { condition: service_healthy }
+      kafka:         { condition: service_healthy }
       kibana:        { condition: service_healthy }
       redis:         { condition: service_healthy }
 
@@ -35,6 +38,14 @@ services:
       file: ${ES_BEATS}/testing/environments/${TESTING_ENVIRONMENT}.yml
       service: elasticsearch
 
+  kafka:
+    build: ${ES_BEATS}/testing/environments/docker/kafka
+    expose:
+      - 9092
+      - 2181
+    environment:
+      - ADVERTISED_HOST=kafka
+
   kibana:
     extends:
       file: ${ES_BEATS}/testing/environments/${TESTING_ENVIRONMENT}.yml

filebeat/docs/fields.asciidoc

@@ -7827,6 +7827,67 @@ type: keyword
 
 --
 
+
+*`kafka.topic`*::
++
+--
+Kafka topic
+
+
+type: keyword
+
+--
+
+*`kafka.partition`*::
++
+--
+Kafka partition number
+
+
+type: long
+
+--
+
+*`kafka.offset`*::
++
+--
+Kafka offset of this message
+
+
+type: long
+
+--
+
+*`kafka.key`*::
++
+--
+Kafka key, corresponding to the Kafka value stored in the message
+
+
+type: keyword
+
+--
+
+*`kafka.block_timestamp`*::
++
+--
+Kafka outer (compressed) block timestamp
+
+
+type: date
+
+--
+
+*`kafka.headers`*::
++
+--
+Kafka headers for this message.
+
+
+type: nested
+
+--
+
 [[exported-fields-logstash]]
 == logstash fields
 

filebeat/docs/inputs/input-kafka.asciidoc

@@ -0,0 +1,121 @@
+:type: kafka
+
+[id="{beatname_lc}-input-{type}"]
+=== Kafka input
+
+++++
+<titleabbrev>Kafka</titleabbrev>
+++++
+
+Use the `kafka` input to read from topics in a Kafka cluster.
+
+To configure this input, specify a list of one or more <<hosts,`hosts`>> in the
+cluster to bootstrap the connection with, a list of <<topics,`topics`>> to
+track, and a <<groupid,`group_id`>> for the connection.
+
+Example configuration:
+
+["source","yaml",subs="attributes"]
+----
+{beatname_lc}.inputs:
+- type: kafka
+  hosts:
+    - kafka-broker-1:9092
+    - kafka-broker-2:9092
+  topics: ["my-topic"]
+  group_id: "filebeat"
+
+----
+
+
+[id="{beatname_lc}-input-{type}-options"]
+==== Configuration options
+
+The `kafka` input supports the following configuration options plus the
+<<{beatname_lc}-input-{type}-common-options>> described later.
+
+[float]
+[[hosts]]
+===== `hosts`
+
+A list of Kafka bootstrapping hosts (brokers) for this cluster.
+
+[float]
+[[topics]]
+===== `topics`
+
+A list of topics to read from.
+
+[float]
+[[groupid]]
+===== `group_id`
+
+The Kafka consumer group id.
+
+[float]
+===== `client_id`
+
+The Kafka client id (optional).
+
+[float]
+===== `version`
+
+The version of the Kafka protocol to use (defaults to `"1.0.0"`).
+
+[float]
+===== `initial_offset`
+
+The initial offset to start reading, either "oldest" or "newest". Defaults to
+"oldest".
+
+===== `connect_backoff`
+
+How long to wait before trying to reconnect to the kafka cluster after a
+fatal error. Default is 30s.
+
+===== `consume_backoff`
+
+How long to wait before retrying a failed read. Default is 2s.
+
+===== `max_wait_time`
+
+How long to wait for the minimum number of input bytes while reading. Default
+is 250ms.
+
+===== `isolation_level`
+
+This configures the Kafka group isolation level:
+
+- `"read_uncommitted"` returns _all_ messages in the message channel.
+- `"read_committed"` hides messages that are part of an aborted transaction.
+
+The default is `"read_uncommitted"`.
+
+===== `fetch`
+
+Kafka fetch settings:
+
+*`min`*:: The minimum number of bytes to wait for. Defaults to 1.
+
+*`default`*:: The default number of bytes to read per request. Defaults to 1MB.
+
+*`max`*:: The maximum number of bytes to read per request. Defaults to 0
+(no limit).
+
+===== `rebalance`
+
+Kafka rebalance settings:
+
+*`strategy`*:: Either `"range"` or `"roundrobin"`. Defaults to `"range"`.
+
+*`timeout`*:: How long to wait for an attempted rebalance. Defaults to 60s.
+
+*`max_retries`*:: How many times to retry if rebalancing fails. Defaults to 4.
+
+*`retry_backoff`*:: How long to wait after an unsuccessful rebalance attempt.
+Defaults to 2s.
+
+[id="{beatname_lc}-input-{type}-common-options"]
+include::../inputs/input-common-options.asciidoc[]
+
+:type!: