[exporter/kafka] Add factory options for custom metric and log marshalers

.chloggen/exporter-kafka-marshaler-options.yaml

@@ -0,0 +1,18 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: 'enhancement'
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: kafkaexporter 
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Add ExporterFactory options for custom metric and log marshalers"
+
+# One or more tracking issues related to the change
+issues: [14514]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+

exporter/kafkaexporter/factory.go

@@ -53,7 +53,9 @@ const (
 // FactoryOption applies changes to kafkaExporterFactory.
 type FactoryOption func(factory *kafkaExporterFactory)
 
-// WithTracesMarshalers adds tracesMarshalers.
+// WithTracesMarshalers adds tracesMarshalers to the exporter factory.
+// This allows custom-built collectors to configure custom Kafka marshaler(s) for trace data.
+// An example use case is keying the message by trace ID so downstream collectors could do tail-based sampling with horizontal scale
 func WithTracesMarshalers(tracesMarshalers ...TracesMarshaler) FactoryOption {
 	return func(factory *kafkaExporterFactory) {
 		for _, marshaler := range tracesMarshalers {
@@ -62,12 +64,43 @@ func WithTracesMarshalers(tracesMarshalers ...TracesMarshaler) FactoryOption {
 	}
 }
 
+// WithMetricsMarshalers adds metricsMarshalers to the exporter factory.
+// This allows custom-built collectors to configure custom Kafka marshaler(s) for metric data.
+// An example use case is keying the message by resource attribute values so downstream collectors can be horizontally scaled and still deliver metrics in the correct order to the backend.
+// Another use case might be pre-aggregating metrics to reduce backend update throughput.
+func WithMetricsMarshalers(metricsMarshalers ...MetricsMarshaler) FactoryOption {
+	return func(factory *kafkaExporterFactory) {
+		for _, marshaler := range metricsMarshalers {
+			factory.metricsMarshalers[marshaler.Encoding()] = marshaler
+		}
+	}
+}
+
+// WithLogsMarshalers adds logsMarshalers to the exporter factory.
+// This allows custom-built collectors to configure custom Kafka marshaler(s) for log data.
+// An example use case is keying the message by resource attribute values so downstream collectors can be horizontally scaled and still deliver logs in the correct order to the backend.
+func WithLogsMarshalers(logsMarshalers ...LogsMarshaler) FactoryOption {
+	return func(factory *kafkaExporterFactory) {
+		for _, marshaler := range logsMarshalers {
+			factory.logsMarshalers[marshaler.Encoding()] = marshaler
+		}
+	}
+}
+
+// WithProducerFactory sets an alternative producer factory.  Primarily for injecting mocks/noop producers for testing
+func WithProducerFactory(producerFactory ProducerFactoryFunc) FactoryOption {
+	return func(factory *kafkaExporterFactory) {
+		factory.producerFactory = producerFactory
+	}
+}
+
 // NewFactory creates Kafka exporter factory.
 func NewFactory(options ...FactoryOption) component.ExporterFactory {
 	f := &kafkaExporterFactory{
 		tracesMarshalers:  tracesMarshalers(),
 		metricsMarshalers: metricsMarshalers(),
 		logsMarshalers:    logsMarshalers(),
+		producerFactory:   newSaramaProducer,
 	}
 	for _, o := range options {
 		o(f)
@@ -107,10 +140,15 @@ func createDefaultConfig() component.Config {
 	}
 }
 
+// Factory function resposible for creating the SyncProducer
+// Primarily used for injecting mock producers for testing
+type ProducerFactoryFunc func(config Config) (sarama.SyncProducer, error)
+
 type kafkaExporterFactory struct {
 	tracesMarshalers  map[string]TracesMarshaler
 	metricsMarshalers map[string]MetricsMarshaler
 	logsMarshalers    map[string]LogsMarshaler
+	producerFactory   ProducerFactoryFunc
 }
 
 func (f *kafkaExporterFactory) createTracesExporter(
@@ -125,7 +163,7 @@ func (f *kafkaExporterFactory) createTracesExporter(
 	if oCfg.Encoding == "otlp_json" {
 		set.Logger.Info("otlp_json is considered experimental and should not be used in a production environment")
 	}
-	exp, err := newTracesExporter(oCfg, set, f.tracesMarshalers)
+	exp, err := newTracesExporter(oCfg, set, f.producerFactory, f.tracesMarshalers)
 	if err != nil {
 		return nil, err
 	}
@@ -155,7 +193,7 @@ func (f *kafkaExporterFactory) createMetricsExporter(
 	if oCfg.Encoding == "otlp_json" {
 		set.Logger.Info("otlp_json is considered experimental and should not be used in a production environment")
 	}
-	exp, err := newMetricsExporter(oCfg, set, f.metricsMarshalers)
+	exp, err := newMetricsExporter(oCfg, set, f.producerFactory, f.metricsMarshalers)
 	if err != nil {
 		return nil, err
 	}
@@ -185,7 +223,7 @@ func (f *kafkaExporterFactory) createLogsExporter(
 	if oCfg.Encoding == "otlp_json" {
 		set.Logger.Info("otlp_json is considered experimental and should not be used in a production environment")
 	}
-	exp, err := newLogsExporter(oCfg, set, f.logsMarshalers)
+	exp, err := newLogsExporter(oCfg, set, f.producerFactory, f.logsMarshalers)
 	if err != nil {
 		return nil, err
 	}