exporter/kafkaexporter: Ability to add custom Log and Metric Marshalers

[bgranetzke]

Is your feature request related to a problem? Please describe.

There is currently no way to configure custom Log and Metric marshalers without forking the code line.

Describe the solution you'd like

I'd like two additional additional export factory options: WithLogsMarshalers and WithMetricsMarshalers which work exactly the same as the existing WithTracesMarshalers option (except with with logs and metrics, respectively)

Describe alternatives you've considered

I was going to contribute the custom marshalers I was building for my use case, but there was at least one other PR which attempted to contribute keyed messages which had trouble getting pushed through. Also, when searching related issues, there are several issues which involve site-specific log formats which could benefit from the more general option to extend without forking.

Additional context

This PR was exactly what I needed, but it ended with:

The minimal required testing of custom encodings is that the encoded data matches a known output.
Testing that the set encoding matches the expected result is just the start.
...and the author abandoned the PR.

This PR started where that PR left off except it attempted to solve the testing requirement above. I tried to limit impact radius to make the basic options testable, but I could not come up with an elegant way. I'm open to suggestions.

[github-actions]

Pinging code owners: @pavolloffay @MovieStoreGuy. See Adding Labels via Comments if you do not have permissions to add labels yourself.

[MovieStoreGuy]

There is currently no way to configure custom Log and Metric marshalers without forking the code line.

Can I ask you to expand upon this? What would you consider custom?

To do you mean, I can convert from one existing logging format (OTLP) to elastic schema. (Not sure if it is a real thing)

[bgranetzke]

Any implementation of kafkaexporter.MetricsMarshaler/LogsMarshaler cannot be added to the exporter without contributing it to otel-collector-contrib or forking the repository to modify the private functions: metricsMarshalers()/logsMarshalers().

Use cases include:

• (my use case) creating keyed kafka messages (the jager protos do, but otlp do not) related issue: could kafka export hash parittion #12318
• Logs with custom schema: [exporter/kafka] Logs with custom schema #12621
• Elastic schema: Add elastic json encoding to kafka log exporter #7416

I briefly considered just abstracting away a function which could produce the key for the message, however, I also needed to (potentially) split up the plog.Logs/pmetric.Metrics into multiple messages. The current *Marshaler interfaces already account for this.

[MovieStoreGuy]

(my use case) creating keyed kafka messages (the jager protos do, but otlp do not) related issue: #12318

Ah yes, I knew about the key partitioning but haven't scheduled any of my time to address it.

Logs with custom schema: #12621

I am gonna push back on this since the collector still needs to retain the ability to send data to kafka with reasonable speed and once it is committed to Kafka, it should be simple to apply transforms to the data.

#7416

Unfortunately, this had fallen off my radar as well, I promise to get back to it working with a solution.

[bgranetzke]

I tend to agree with you about having the capability to transform downstream. Like I said, my immediate need is just to have the option to inject my own marshal behavior with the factory option. I'm really not excited about having to create the producer interface, but I could not find another way to test the factory option properly (without involving a Kafka broker). I'd love to rip that code out if you could provide an alternative or relax the test requirement to what is already in place for TraceMarshalers.

[bgranetzke]

@MovieStoreGuy I have created a new PR using a different approach to testing. I consider it minimally invasive to the existing design. Please provide feedback.