[filebeat] Add a Kafka input

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

@@ -31,6 +31,7 @@ grouped in the following categories:
 * <<exported-fields-iptables>>
 * <<exported-fields-jolokia-autodiscover>>
 * <<exported-fields-kafka>>
+* <<exported-fields-kafka-input>>
 * <<exported-fields-kibana>>
 * <<exported-fields-kubernetes-processor>>
 * <<exported-fields-log>>
@@ -7318,6 +7319,73 @@ type: text
 
 --
 
+[[exported-fields-kafka-input]]
+== Kafka Input fields
+
+Kafka metadata added by the kafka input
+
+
+
+*`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`*::
++
+--
+The array of kafka headers, each an object containing string subfields "key" and "value".
+
+
+type: array
+
+--
+
 [[exported-fields-kibana]]
 == kibana 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!:

filebeat/input/kafka/_meta/fields.yml

@@ -0,0 +1,37 @@
+- key: kafka-input
+  title: Kafka Input
+  description: >
+    Kafka metadata added by the kafka input
+  short_config: false
+  anchor: kafka-input
+  fields:
+    - name: kafka.topic
+      type: keyword
+      description: >
+        Kafka topic
+
+    - name: kafka.partition
+      type: long
+      description: >
+        Kafka partition number
+
+    - name: kafka.offset
+      type: long
+      description: >
+        Kafka offset of this message
+
+    - name: kafka.key
+      type: keyword
+      description: >
+        Kafka key, corresponding to the Kafka value stored in the message
+
+    - name: kafka.block_timestamp
+      type: date
+      description: >
+        Kafka outer (compressed) block timestamp
+
+    - name: kafka.headers
+      type: array
+      description: >
+        The array of kafka headers, each an object containing string subfields
+        "key" and "value".