Add option to start new root span with link to remote context

Author: soluchok

internal/impl/kafka/input_sarama_kafka.go

@@ -126,6 +126,7 @@ Unfortunately this error message will appear for a wide range of connection prob
 				Description("A maximum estimate for the time taken to process a message, this is used for tuning consumer group synchronization.").
 				Advanced().Default("100ms"),
 			service.NewExtractTracingSpanMappingField(),
+			service.NewRootSpanWithLinkField(),
 			service.NewObjectField(iskFieldGroup,
 				service.NewDurationField(iskFieldGroupSessionTimeout).
 					Description("A period after which a consumer of the group is kicked after no heartbeats.").

internal/impl/nats/docs.go

@@ -24,9 +24,13 @@ NATS component, so that monitoring tools between NATS and bento can stay in sync
 `
 }
 
-func inputTracingDocs() *service.ConfigField {
-	return service.NewExtractTracingSpanMappingField().Version(tracingVersion)
+func inputTracingDocs() []*service.ConfigField {
+	return []*service.ConfigField{
+		service.NewExtractTracingSpanMappingField().Version(tracingVersion),
+		service.NewRootSpanWithLinkField().Version(tracingVersion),
+	}
 }
+
 func outputTracingDocs() *service.ConfigField {
 	return service.NewInjectTracingSpanMappingField().Version(tracingVersion)
 }

internal/impl/nats/input.go

@@ -49,7 +49,7 @@ You can access these metadata fields using [function interpolation](/docs/config
 			Default(nats.DefaultSubPendingMsgsLimit).
 			LintRule(`root = if this < 0 { ["prefetch count must be greater than or equal to zero"] }`)).
 		Fields(connectionTailFields()...).
-		Field(inputTracingDocs())
+		Fields(inputTracingDocs()...)
 }
 
 func init() {

internal/impl/nats/input_jetstream.go

@@ -80,7 +80,7 @@ You can access these metadata fields using
 			Advanced().
 			Default(1024)).
 		Fields(connectionTailFields()...).
-		Field(inputTracingDocs())
+		Fields(inputTracingDocs()...)
 }
 
 func init() {

internal/impl/nats/input_stream.go

@@ -135,7 +135,7 @@ You can access these metadata fields using [function interpolation](/docs/config
 				Default("30s"),
 		).
 		Fields(connectionTailFields()...).
-		Field(inputTracingDocs())
+		Fields(inputTracingDocs()...)
 }
 
 func init() {

public/service/config_extract_tracing.go

@@ -7,13 +7,15 @@ import (
 
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/propagation"
+	"go.opentelemetry.io/otel/trace"
 
 	"github.com/warpstreamlabs/bento/internal/bundle"
 	"github.com/warpstreamlabs/bento/public/bloblang"
 )
 
 const (
-	etsField = "extract_tracing_map"
+	etsField   = "extract_tracing_map"
+	nrswlField = "new_root_span_with_link"
 )
 
 // NewExtractTracingSpanMappingField returns a config field for mapping messages
@@ -27,6 +29,14 @@ func NewExtractTracingSpanMappingField() *ConfigField {
 		Advanced()
 }
 
+func NewRootSpanWithLinkField() *ConfigField {
+	return NewBoolField(nrswlField).
+		Description("EXPERIMENTAL: Starts a new root span with link to parent.").
+		Version("1.0.0").
+		Optional().
+		Advanced()
+}
+
 // WrapBatchInputExtractTracingSpanMapping wraps a BatchInput with a mechanism
 // for extracting tracing spans using a bloblang mapping.
 func (p *ParsedConfig) WrapBatchInputExtractTracingSpanMapping(inputName string, i BatchInput) (
@@ -38,7 +48,13 @@ func (p *ParsedConfig) WrapBatchInputExtractTracingSpanMapping(inputName string,
 	if err != nil {
 		return nil, err
 	}
-	return &spanInjectBatchInput{inputName: inputName, mgr: p.mgr, mapping: exe, rdr: i}, nil
+
+	newSpan, err := p.FieldBool(nrswlField)
+	if err != nil {
+		return nil, err
+	}
+
+	return &spanInjectBatchInput{inputName: inputName, mgr: p.mgr, mapping: exe, rdr: i, newSpan: newSpan}, nil
 }
 
 // WrapInputExtractTracingSpanMapping wraps a Input with a mechanism for
@@ -51,7 +67,13 @@ func (p *ParsedConfig) WrapInputExtractTracingSpanMapping(inputName string, i In
 	if err != nil {
 		return nil, err
 	}
-	return &spanInjectInput{inputName: inputName, mgr: p.mgr, mapping: exe, rdr: i}, nil
+
+	newSpan, err := p.FieldBool(nrswlField)
+	if err != nil {
+		return nil, err
+	}
+
+	return &spanInjectInput{inputName: inputName, mgr: p.mgr, mapping: exe, rdr: i, newSpan: newSpan}, nil
 }
 
 func getPropMapCarrier(spanPart *Message) (propagation.MapCarrier, error) {
@@ -82,6 +104,7 @@ type spanInjectBatchInput struct {
 
 	mapping *bloblang.Executor
 	rdr     BatchInput
+	newSpan bool
 }
 
 func (s *spanInjectBatchInput) Connect(ctx context.Context) error {
@@ -112,7 +135,16 @@ func (s *spanInjectBatchInput) ReadBatch(ctx context.Context) (MessageBatch, Ack
 	textProp := otel.GetTextMapPropagator()
 	for i, p := range m {
 		ctx := textProp.Extract(p.Context(), c)
-		pCtx, _ := prov.Tracer("bento").Start(ctx, operationName)
+
+		var opts []trace.SpanStartOption
+		if s.newSpan {
+			opts = []trace.SpanStartOption{
+				trace.WithNewRoot(),
+				trace.WithLinks(trace.LinkFromContext(ctx)),
+			}
+		}
+
+		pCtx, _ := prov.Tracer("bento").Start(ctx, operationName, opts...)
 		m[i] = p.WithContext(pCtx)
 	}
 	return m, afn, nil
@@ -130,6 +162,7 @@ type spanInjectInput struct {
 
 	mapping *bloblang.Executor
 	rdr     Input
+	newSpan bool
 }
 
 func (s *spanInjectInput) Connect(ctx context.Context) error {
@@ -159,7 +192,17 @@ func (s *spanInjectInput) Read(ctx context.Context) (*Message, AckFunc, error) {
 
 	textProp := otel.GetTextMapPropagator()
 
-	pCtx, _ := prov.Tracer("bento").Start(textProp.Extract(m.Context(), c), operationName)
+	ctx = textProp.Extract(m.Context(), c)
+
+	var opts []trace.SpanStartOption
+	if s.newSpan {
+		opts = []trace.SpanStartOption{
+			trace.WithNewRoot(),
+			trace.WithLinks(trace.LinkFromContext(ctx)),
+		}
+	}
+
+	pCtx, _ := prov.Tracer("bento").Start(ctx, operationName, opts...)
 	m = m.WithContext(pCtx)
 
 	return m, afn, nil

website/docs/components/inputs/kafka.md

@@ -83,6 +83,7 @@ input:
     commit_period: 1s
     max_processing_period: 100ms
     extract_tracing_map: root = @ # No default (optional)
+    new_root_span_with_link: false # No default (optional)
     group:
       session_timeout: 10s
       heartbeat_interval: 3s
@@ -599,6 +600,14 @@ 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.0.0 or newer  
+
 ### `group`
 
 Tuning parameters for consumer group synchronization.

website/docs/components/inputs/nats.md

@@ -63,6 +63,7 @@ input:
       user_jwt: "" # No default (optional)
       user_nkey_seed: "" # No default (optional)
     extract_tracing_map: root = @ # No default (optional)
+    new_root_span_with_link: false # No default (optional)
 ```
 
 </TabItem>
@@ -401,4 +402,12 @@ 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 4.23.0 or newer  
+
 

website/docs/components/inputs/nats_jetstream.md

@@ -71,6 +71,7 @@ input:
       user_jwt: "" # No default (optional)
       user_nkey_seed: "" # No default (optional)
     extract_tracing_map: root = @ # No default (optional)
+    new_root_span_with_link: false # No default (optional)
 ```
 
 </TabItem>
@@ -450,4 +451,12 @@ 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 4.23.0 or newer  
+
 

website/docs/components/inputs/nats_stream.md

@@ -70,6 +70,7 @@ input:
       user_jwt: "" # No default (optional)
       user_nkey_seed: "" # No default (optional)
     extract_tracing_map: root = @ # No default (optional)
+    new_root_span_with_link: false # No default (optional)
 ```
 
 </TabItem>
@@ -423,4 +424,12 @@ 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 4.23.0 or newer  
+