feature(gcp_pubsub): add extract_tracing_map / inject_tracing_map support

internal/impl/gcp/input_pubsub.go

@@ -117,6 +117,8 @@ You can access these metadata fields using [function interpolation](/docs/config
 			).
 				Description("Allows you to configure the input subscription and creates if it doesn't exist.").
 				Advanced(),
+			service.NewExtractTracingSpanMappingField(),
+			service.NewRootSpanWithLinkField().Version("1.20.0"),
 		)
 }
 
@@ -127,7 +129,11 @@ func init() {
 			if err != nil {
 				return nil, err
 			}
-			return newGCPPubSubReader(pConf, mgr)
+			r, err := newGCPPubSubReader(pConf, mgr)
+			if err != nil {
+				return nil, err
+			}
+			return conf.WrapInputExtractTracingSpanMapping("gcp_pubsub", r)
 		})
 	if err != nil {
 		panic(err)

internal/impl/gcp/output_pubsub.go

@@ -74,6 +74,7 @@ pipeline:
 			service.NewMetadataExcludeFilterField("metadata").
 				Optional().
 				Description("Specify criteria for which metadata values are sent as attributes, all are sent by default."),
+			service.NewInjectTracingSpanMappingField(),
 			service.NewObjectField(
 				"flow_control",
 				service.NewIntField("max_outstanding_bytes").
@@ -360,7 +361,11 @@ func init() {
 			return
 		}
 
-		out, err = newPubSubOutput(conf)
+		if out, err = newPubSubOutput(conf); err != nil {
+			return
+		}
+
+		out, err = conf.WrapBatchOutputExtractTracingSpanMapping("gcp_pubsub", out)
 
 		return
 	}); err != nil {

website/docs/components/inputs/gcp_pubsub.md

@@ -55,6 +55,8 @@ input:
     create_subscription:
       enabled: false
       topic: ""
+    extract_tracing_map: root = @ # No default (optional)
+    new_root_span_with_link: false # No default (optional)
 ```
 
 </TabItem>
@@ -156,4 +158,27 @@ Defines the topic that the subscription should be vinculated to.
 Type: `string`  
 Default: `""`  
 
+### `extract_tracing_map`
+
+EXPERIMENTAL: A [Bloblang mapping](/docs/guides/bloblang/about) 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`  
+
+```yml
+# Examples
+
+extract_tracing_map: root = @
+
+extract_tracing_map: root = this.meta.span
+```
+
+### `new_root_span_with_link`
+
+EXPERIMENTAL: Starts a new root span with link to parent.
+
+
+Type: `bool`  
+Requires version 1.20.0 or newer  
+
 

website/docs/components/outputs/gcp_pubsub.md

@@ -66,6 +66,7 @@ output:
     publish_timeout: 1m0s
     metadata:
       exclude_prefixes: []
+    inject_tracing_map: meta = @.assign(this) # No default (optional)
     flow_control:
       max_outstanding_bytes: -1
       max_outstanding_messages: 1000
@@ -211,6 +212,21 @@ Provide a list of explicit metadata key prefixes to be excluded when adding meta
 Type: `array`  
 Default: `[]`  
 
+### `inject_tracing_map`
+
+EXPERIMENTAL: A [Bloblang mapping](/docs/guides/bloblang/about) 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`  
+
+```yml
+# Examples
+
+inject_tracing_map: meta = @.assign(this)
+
+inject_tracing_map: root.meta.span = this
+```
+
 ### `flow_control`
 
 For a given topic, configures the PubSub client's internal buffer for messages to be published.