update

open-telemetry · Feb 15, 2025 · df83318 · df83318
1 parent 5998f4e
commit df83318
Show file tree

Hide file tree

Showing 4 changed files with 20 additions and 9 deletions.
diff --git a/exporter/exporterhelper/internal/batcher/batch_context.go b/exporter/exporterhelper/internal/batcher/batch_context.go
@@ -20,12 +20,20 @@ func LinksFromContext(ctx context.Context) []trace.Link {
 	if links, ok := ctx.Value(batchSpanLinksKey).([]trace.Link); ok {
 		return links
 	}
-	return []trace.Link{trace.LinkFromContext(ctx)}
+	return []trace.Link{}
 }
 
+func contextWithLink(ctx context.Context) context.Context {
+	return context.WithValue(
+		context.Background(),
+		batchSpanLinksKey,
+		[]trace.Link{trace.LinkFromContext(ctx)})
+}
+
+// contextWithMergedLinks expects ctx1 to have
 func contextWithMergedLinks(ctx1 context.Context, ctx2 context.Context) context.Context {
 	return context.WithValue(
 		context.Background(),
 		batchSpanLinksKey,
-		append(LinksFromContext(ctx1), LinksFromContext(ctx2)...))
+		append(LinksFromContext(ctx1), trace.LinkFromContext(ctx2)))
 }
diff --git a/exporter/exporterhelper/internal/batcher/batch_context_test.go b/exporter/exporterhelper/internal/batcher/batch_context_test.go
@@ -28,7 +28,8 @@ func TestBatchContextLink(t *testing.T) {
 	ctx4, span4 := tracer.Start(ctx1, "span4")
 	defer span4.End()
 
-	batchContext := contextWithMergedLinks(ctx2, ctx3)
+	batchContext := contextWithLink(ctx2)
+	batchContext = contextWithMergedLinks(batchContext, ctx3)
 	batchContext = contextWithMergedLinks(batchContext, ctx4)
 
 	actualLinks := LinksFromContext(batchContext)

diff --git a/exporter/exporterhelper/internal/batcher/default_batcher.go b/exporter/exporterhelper/internal/batcher/default_batcher.go
@@ -86,7 +86,7 @@ func (qb *defaultBatcher) Consume(ctx context.Context, req request.Request, done
 			// Do not flush the last item and add it to the current batch.
 			reqList = reqList[:len(reqList)-1]
 			qb.currentBatch = &batch{
-				ctx:  ctx,
+				ctx:  contextWithLink(ctx),
 				req:  lastReq,
 				done: multiDone{done},
 			}
@@ -142,7 +142,7 @@ func (qb *defaultBatcher) Consume(ctx context.Context, req request.Request, done
 			// Do not flush the last item and add it to the current batch.
 			reqList = reqList[:len(reqList)-1]
 			qb.currentBatch = &batch{
-				ctx:  ctx,
+				ctx:  contextWithLink(ctx),
 				req:  lastReq,
 				done: multiDone{done},
 			}

diff --git a/exporter/exporterhelper/internal/queue_sender.go b/exporter/exporterhelper/internal/queue_sender.go
@@ -113,11 +113,7 @@ func NewQueueSender(
 		}
 
 		q, err := newObsQueue(qSet, qf(context.Background(), qSet, qCfg, func(ctx context.Context, req request.Request, done exporterqueue.Done) {
-			// TODO: move start of span to enqueue instead to dequeue.
-			// Figure out how to preserve span context across persistent storage.
-			ctx, _ = metadata.Tracer(qSet.ExporterSettings.TelemetrySettings).Start(ctx, "exporter/enqueue")
 			done.OnDone(exportFunc(ctx, req))
-			trace.SpanFromContext(ctx).End()
 		}))
 		if err != nil {
 			return nil, err
@@ -129,7 +125,13 @@ func NewQueueSender(
 		// Have to read the number of items before sending the request since the request can
 		// be modified by the downstream components like the batcher.
 		itemsCount := req.ItemsCount()
+
+		// TODO: move start of span to enqueue instead to dequeue.
+		// Figure out how to preserve span context across persistent storage.
+		ctx, _ = metadata.Tracer(qSet.ExporterSettings.TelemetrySettings).Start(ctx, "exporter/enqueue")
 		err := next.Send(ctx, req)
+		trace.SpanFromContext(ctx).End()
+
 		if err != nil {
 			qSet.ExporterSettings.Logger.Error("Exporting failed. Dropping data."+exportFailureMessage,
 				zap.Error(err), zap.Int("dropped_items", itemsCount))