Add tracing mapping fields into kafka_franz

docs/modules/components/pages/inputs/kafka_franz.adoc

@@ -81,6 +81,7 @@ input:
       period: ""
       check: ""
       processors: [] # No default (optional)
+    extract_tracing_map: root = @ # No default (optional)
 ```
 
 --
@@ -694,4 +695,21 @@ processors:
       format: json_array
 ```
 
+=== `extract_tracing_map`
+
+EXPERIMENTAL: A xref:guides:bloblang/about.adoc[Bloblang mapping] that attempts to extract an object containing tracing propagation information, which will then be used as the root tracing span for the message. The specification of the extracted fields must match the format used by the service wide tracer.
+
+
+*Type*: `string`
+
+Requires version 3.45.0 or newer
+
+```yml
+# Examples
+
+extract_tracing_map: root = @
+
+extract_tracing_map: root = this.meta.span
+```
+
 

docs/modules/components/pages/inputs/ockam_kafka.adoc

@@ -90,6 +90,7 @@ input:
         period: ""
         check: ""
         processors: [] # No default (optional)
+      extract_tracing_map: root = @ # No default (optional)
       seed_brokers: [] # No default (optional)
     disable_content_encryption: false
     enrollment_ticket: "" # No default (optional)
@@ -701,6 +702,23 @@ processors:
       format: json_array
 ```
 
+=== `kafka.extract_tracing_map`
+
+EXPERIMENTAL: A xref:guides:bloblang/about.adoc[Bloblang mapping] that attempts to extract an object containing tracing propagation information, which will then be used as the root tracing span for the message. The specification of the extracted fields must match the format used by the service wide tracer.
+
+
+*Type*: `string`
+
+Requires version 3.45.0 or newer
+
+```yml
+# Examples
+
+extract_tracing_map: root = @
+
+extract_tracing_map: root = this.meta.span
+```
+
 === `kafka.seed_brokers`
 
 A list of broker addresses to connect to in order to establish connections. If an item of the list contains commas it will be expanded into multiple addresses.

docs/modules/components/pages/outputs/kafka_franz.adoc

@@ -95,6 +95,7 @@ output:
       client_certs: []
     sasl: [] # No default (optional)
     timestamp: ${! timestamp_unix() } # No default (optional)
+    inject_tracing_map: meta = @.merge(this) # No default (optional)
 ```
 
 --
@@ -794,4 +795,21 @@ timestamp: ${! timestamp_unix() }
 timestamp: ${! metadata("kafka_timestamp_unix") }
 ```
 
+=== `inject_tracing_map`
+
+EXPERIMENTAL: A xref:guides:bloblang/about.adoc[Bloblang mapping] used to inject an object containing tracing propagation information into outbound messages. The specification of the injected fields will match the format used by the service wide tracer.
+
+
+*Type*: `string`
+
+Requires version 3.45.0 or newer
+
+```yml
+# Examples
+
+inject_tracing_map: meta = @.merge(this)
+
+inject_tracing_map: root.meta.span = this
+```
+
 

docs/modules/components/pages/outputs/ockam_kafka.adoc

@@ -103,6 +103,7 @@ output:
         client_certs: []
       sasl: [] # No default (optional)
       timestamp: ${! timestamp_unix() } # No default (optional)
+      inject_tracing_map: meta = @.merge(this) # No default (optional)
       seed_brokers: [] # No default (optional)
     disable_content_encryption: false
     enrollment_ticket: "" # No default (optional)
@@ -813,6 +814,23 @@ timestamp: ${! timestamp_unix() }
 timestamp: ${! metadata("kafka_timestamp_unix") }
 ```
 
+=== `kafka.inject_tracing_map`
+
+EXPERIMENTAL: A xref:guides:bloblang/about.adoc[Bloblang mapping] used to inject an object containing tracing propagation information into outbound messages. The specification of the injected fields will match the format used by the service wide tracer.
+
+
+*Type*: `string`
+
+Requires version 3.45.0 or newer
+
+```yml
+# Examples
+
+inject_tracing_map: meta = @.merge(this)
+
+inject_tracing_map: root.meta.span = this
+```
+
 === `kafka.seed_brokers`
 
 A list of broker addresses to connect to in order to establish connections. If an item of the list contains commas it will be expanded into multiple addresses.

internal/impl/kafka/input_kafka_franz.go

@@ -126,6 +126,7 @@ Finally, it's also possible to specify an explicit offset to consume from by add
 		service.NewBatchPolicyField("batching").
 			Description("Allows you to configure a xref:configuration:batching.adoc[batching policy] that applies to individual topic partitions in order to batch messages together before flushing them for processing. Batching can be beneficial for performance as well as useful for windowed processing, and doing so this way preserves the ordering of topic partitions.").
 			Advanced(),
+		service.NewExtractTracingSpanMappingField(),
 	}
 }
 
@@ -136,7 +137,13 @@ func init() {
 			if err != nil {
 				return nil, err
 			}
-			return service.AutoRetryNacksBatchedToggled(conf, rdr)
+
+			r, err := service.AutoRetryNacksBatchedToggled(conf, rdr)
+			if err != nil {
+				return nil, err
+			}
+
+			return conf.WrapBatchInputExtractTracingSpanMapping("kafka_franz", r)
 		})
 	if err != nil {
 		panic(err)

internal/impl/kafka/output_kafka_franz.go

@@ -124,6 +124,7 @@ func FranzKafkaOutputConfigFields() []*service.ConfigField {
 			Example(`${! metadata("kafka_timestamp_unix") }`).
 			Optional().
 			Advanced(),
+		service.NewInjectTracingSpanMappingField(),
 	}
 }
 
@@ -141,7 +142,11 @@ func init() {
 			if batchPolicy, err = conf.FieldBatchPolicy("batching"); err != nil {
 				return
 			}
-			output, err = NewFranzKafkaWriterFromConfig(conf, mgr.Logger())
+			if output, err = NewFranzKafkaWriterFromConfig(conf, mgr.Logger()); err != nil {
+				return
+			}
+
+			output, err = conf.WrapBatchOutputExtractTracingSpanMapping("kafka_franz", output)
 			return
 		})
 	if err != nil {