datadog/datadog-kafka-connect-logs

A plugin for Kafka Connect to send Kafka records as logs to Datadog.

datadog-kafka-connect-logs's Issues

Implement exactly-once support

Kafka Connect allows sink connectors to optionally implement exactly once behavior
by tracking in the external system the offsets for each topic partition. Most users
would prefer exactly once if given the choice, so where possible and where feasible
sink connectors should implement this behavior.
To handle exactly-once semantics for message delivery, the Source Connector must
correctly map the committed offsets to the Kafka cluster with some analog within
the source data system, and then handle the necessary rewinding should messages
need to be re-delivered. For example, consider a trivial Source connector that
publishes the lines from an input file to a Kafka topic one line at a time ... prefixed by
the line number. The commit* methods for that connector would save the line
number of the posted record ... and then pick up at that location upon a restart.

Hardcoded URL doesn't allow to send logs to different endpoint like EU

Describe what happened:
I am using this datadog kafka connector but my datadog system is operating in EU

Describe what you expected:
To be able to configure the URL from parameters in case a different URL than the US one is wanted to be used.

Steps to reproduce the issue:
Trying to send logs to the EU datadog endpoint

Additional environment details (Operating System, Cloud provider, etc):

Implement specialized logic for different HTTP error codes

The Logs Intake API returns different error codes for different situations:

413: For messages exceeding the 5MB limit
408: If the connection times out.

We should check if with some error codes, a simple retry is not the correct way to go and instead there could be logic to handle them better.

Truncate messages if compressed size exceeds 5MB

The Logs Intake API has a limit for the total size of a payload (5MB). We should figure out a logic to truncate those messages while retaining their JSON structure so they can be ingested properly.

Example of structure:

[
  {
    "ddsource": "kafka-connect",
    "message": "hello"
  },
  {
    "ddsource": "kafka-connect",
    "message": "world"
  }
]

Change HTTP Connection to open at startup and close on shutdown

Currently, the connector opens and closes an HTTP connection or every batch it sends. To reduce network load, we can change the logic to open a connection once and keep it alive until either:

a) The task stops
b) The connect worker stops

Special care needs to be given so that the connection is closed even in error states to avoid resource leaks.

bytes conversion to decimal type

We have kafka messages specified based on Avro schema. Some fields are defined as bytes but having Decimal as the logical type, like this:

{"name": "myField", "type": ["null", {"type": "bytes", "logicalType": "decimal", "precision": 11, "scale": 2}], "default": null, "doc": "BigDecimal with precision 11 scale 2"},

When messages imported into datadog as logs, it appears that myField are set as bytes instead of its logical type Decimal.

Am I missing some configuration, or is this currently not supported?

Implement graceful backoff