[Meta] Filebeat mqtt input

[faec]

This is a tracking issue for the in-progress MQTT input .

Pending requirements:

• Initial draft checkin (Add MQTT input to Filebeat #15287)
• Reconnection after errors (MQTT: Establish new connection after error and clean shutdown #15982)
• Clean shutdown / reliable Stop and Wait (MQTT: Establish new connection after error and clean shutdown #15982)
• Ack handling ([Filebeat] MQTT input: Implement end-to-end ACK #15610)
  • (Need more background here, it's not clear to me yet if / how we need to handle this)
• Documentation (for all new config fields and emitted event fields) (MQTT: Write input documentation #16126)
• Tests
  • Unit tests (MQTT: Establish new connection after error and clean shutdown #15982)
    • code that translates configuration + raw MQTT input to beats events should be refactored into a standalone function and unit tested.
  • System tests (MQTT: Add integration test #16125)
    • bring up the input with a live connection to localhost and pass it test events. This is usually done in the python tests, but if we can get good coverage in go instead I'd be happy (that usually makes verification simpler since you can pass in a mocked outlet for outgoing events instead of requiring a real backend index, but if library support or other context makes it easier in python then that's fine).
• Merge to master and update changelog (Filebeat: Merge "mqtt" input to master #16204)

[mtojek]

I will take a look at ACK handling.

EDIT:

Speaking about sending acknowledgement packets - paho.mqtt.golang should support them out of the box: matchAndDispatch method calls an appropriate handler (if defined) and m.Ack() to confirm handling (https://github.com/eclipse/paho.mqtt.golang/blob/master/router.go#L136). The function body is defined in: https://github.com/eclipse/paho.mqtt.golang/blob/master/net.go#L291 .

I'm not quite sure if the library implementation allows for any modifications (or passing custom ack functions) here. It calls Ack() without checking handler status.

Do you think it's sufficient in this case?

EDIT2:

Agreed offline to implement an end-to-end ACK flow

[mtojek]

The branch has been rebased and merged (also licensing script has been updated). I'm proceeding with #15982 .

[mtojek]

I'm working on the integration test to verify the module.

[tonygunter]

Why not just allow the handler to call Ack() manually? Hander { unmarshall message -> process message -> forward to next receiver -> ACK}? At this point, the mqtt.Message.Ack() function, wrapping a sync.once.Do function like it does, never actually executes any code except once inside matchAndDispatch.