[filebeat] Add a Kafka input

[faec]

Add a Kafka input to Filebeat (#7641).

[urso]

can we have duplicate headers?

Would it make sense to combine headers and only have the return type map[string]string?

[faec]

I thought about this, but as far as I can tell duplicate header keys are valid (if not in the spec then at least in the implementation -- when I post events with duplicate headers, it passes them on unchanged to the consumer)

[urso]

Checking the kafka implementation it is indeed just a list of headers, allowing for duplicates.
This is not optimal for querying/filtering events via kibana. A simple map[string]string would fit kibana UI better. I wonder if map[string][]string might be a better compromise.

We should test how the differnce []map[string]string and map[string][]string affect the usability.

[urso]

For adding end-to-end ACK support you will need this: #12997

end-to-end ACK becomes possible via:

    ...
    out, err := outlet.ConnectWith(cfg, beat.ClientConfig{
		Processing: beat.ProcessingConfig{
			DynamicFields: context.DynamicFields,
		},
        ACKEvents: func(privates []interface{}) {
            for _, priv := range []privates {
                if cm, ok := priv.(*sarama.ConsumerMessage); ok {
                    sess.MarkMessage(cm, "")
                }
            }
        },
	})
    ...

    var msg *sarama.ConsumerMessage
    ...
    out.OnEvent(beat.Event{
        Timestamp: ...
        Fields: ...
        Meta: ...
        Private: msg
    })

The ACKEvents callback will be called with the 'Private' field of all events recently ACKed. The pipeline ensures that ACKs are returned in the same order as you published your events.

Note: I have the sessions available in the callback. This is motivated by the fact that a pipeline.Client is not necessarily multithreading safe (we just made it safe because of the logs input in filebeat).

[faec]

I've revised the core run loop to be more robust, fixed end-to-end ACK per @urso's suggestions about groupHandler, and fixed a shutdown bug that I came across in the process. The integration test now waits for the input to shutdown and confirms it takes <30sec.

I also looked at the header issue in a live index, and currently it's definitely not what we want:

curl "http://localhost:9200/filebeat-8.0.0-2019.08.01-000001/_search?pretty"
...
            "headers" : [
              {
                "key" : [
                  107,
                  101,
                  121,
                  115,
                  32,
                  97,
                  110,
...

I initially assumed a data blob was the safest since that's what's in the spec, but given this outcome, absent better suggestions, I'll make them strings and leave it to the receiver to re-extract a data blob if that's truly their intent.

[urso]

We normally apply some exponential backoff with jitter. See package libbeat/common/backoff.

[faec]

The backoff package works only based on the Wait function, not a channel, so I had to wrap it in an auxiliary channel to handle shutdown signals, let me know if you see a prettier approach.

[urso]

Internally the implementation uses a channel. We could introduce a method like C() <- chan time.Time, which is created by a timer. If the current timer is not nil, it should be stopped and reset when calling C.

[faec]

Update on the headers issue, following several tests and an offline discussion: the suggestion above about writing them to map[string][]string doesn't allow for the kind of searching we'd like. The current container, []struct{ key, value: string }, can work in theory as long as it's indexed as a nested type. I've updated filebeat/_meta/fields.common.yml with something near the correct config, but for some reason the properties inside headers (key, value) do not appear in the index template.

I can create index templates for fields like this manually, but when the template is built by filebeat the subfields are omitted. None of our beats / modules / etc currently use a nested type with defined subfields, so it's not clear to me yet whether the definition in fields.common.yml is somehow incorrect, or if the template generator doesn't handle subfields of a nested field.

[urso]

@dedemorton Can you have a look at the docs?

[jsoriano]

@faec @urso I wonder if we should use the same fields metricbeat uses for this, in case we want to correlate and/or monitor the kafka topics used. Metricbeat uses kafka.topic.name for the topic and kafka.partition.id for the partition.

And I wonder if message specific fields (offset, key, headers) should be under an specific namespace as kafka.message, at least to differentiate the offset from other offsets.