unite traceid

Signed-off-by: jyjiangkai <[email protected]>

Author: hwjiangkai

internal/gateway/gateway.go

@@ -106,7 +106,7 @@ func (ga *ceGateway) receive(ctx context.Context, event v2.Event) (re *v2.Event,
 	ctx, span := ga.tracer.Start(ctx, "receive")
 	defer span.End()
 
-	event.SetExtension("spancontext", ctx)
+	event.SetExtension("traceid", span.SpanContext().TraceID().String())
 	span.SetName("EventTracing")
 	span.SetAttributes(attribute.String("event_id", event.ID()))
 	span.AddEvent("received from source", trace.WithTimestamp(time.Now()))

internal/trigger/reader/reader.go

@@ -206,12 +206,11 @@ func (elReader *eventlogReader) loop(ctx context.Context, lr api.BusReader) erro
 	}
 	for i := range events {
 		_, span := elReader.newSpan(ctx, *events[i])
-		// _, span := elReader.config.Tracer.Start(ctx, events[i].ID())
 		span.SetName("EventTracing")
 		span.SetAttributes(attribute.String("event_id", events[i].ID()))
 		span.AddEvent("read from eventbus", trace.WithTimestamp(time.Now()))
 		span.SetAttributes(attribute.String("eventbus_id", elReader.config.EventbusID.String()))
-		// span.SetAttributes(attribute.String("eventbus_name", elReader.config.EventbusID.String()))
+		span.SetAttributes(attribute.String("subscription_id", elReader.config.SubscriptionIDStr))
 		ec, _ := events[i].Context.(*ce.EventContextV1)
 		offsetByte, _ := ec.Extensions[eventlog.XVanusLogOffset].([]byte)
 		offset := binary.BigEndian.Uint64(offsetByte)
@@ -240,14 +239,14 @@ func (elReader *eventlogReader) newSpan(ctx context.Context, event ce.Event) (co
 	if event.Extensions() == nil {
 		return elReader.config.Tracer.Start(ctx, event.ID())
 	}
-	if _, ok := event.Extensions()["spancontext"]; !ok {
+	if _, ok := event.Extensions()["traceid"]; !ok {
 		return elReader.config.Tracer.Start(ctx, event.ID())
 	}
-	var spanCtx context.Context
-	if err := event.ExtensionAs("spancontext", &spanCtx); err != nil {
-		return elReader.config.Tracer.Start(ctx, event.ID())
-	}
-	return elReader.config.Tracer.Start(spanCtx, event.ID())
+	traceid, _ := trace.TraceIDFromHex(event.Extensions()["traceid"].(string))
+	_ctx := trace.ContextWithSpanContext(ctx, trace.NewSpanContext(trace.SpanContextConfig{
+		TraceID: traceid,
+	}))
+	return elReader.config.Tracer.Start(_ctx, event.ID())
 }
 
 func (elReader *eventlogReader) putEvent(ctx context.Context, event info.EventRecord) error {

observability/tracing/exporter/event_trace_exporter.go

@@ -124,9 +124,9 @@ func newEvent(span tracesdk.ReadOnlySpan) v2.Event {
 	data["end_time"] = span.EndTime()
 	data["status"] = span.Status()
 	for _, attr := range span.Attributes() {
-		if attr.Key == attribute.Key("offset") {
+		if attr.Value.Type() == attribute.INT64 {
 			data[string(attr.Key)] = attr.Value.AsInt64()
-		} else {
+		} else if attr.Value.Type() == attribute.STRING {
 			data[string(attr.Key)] = attr.Value.AsString()
 		}
 	}