marshal attributes

Author: ashwanthgoli

pkg/engine/engine.go

@@ -435,6 +435,17 @@ func linkCaptureUsingPlan(capture *xcap.Capture, plan *physical.Plan) {
 			if len(parents) > 0 {
 				// TODO: This is assuming a single parent which is not always true.
 				// Fix this when we have plans with multiple parents.
+
+				if parents[0].Type() == physical.NodeTypeParallelize {
+					// Skip Parallelize nodes as they are just not actual work.
+					pp := plan.Graph().Parents(parents[0])
+					if len(pp) > 0 {
+						parents = pp
+					} else {
+						return nil
+					}
+				}
+
 				idToParentID[n.ID().String()] = parents[0].ID().String()
 			}
 			return nil

pkg/xcap/exporter.go

@@ -58,17 +58,10 @@ func createSpans(ctx context.Context, region *Region, parentToChildren map[ident
 		opts = append(opts, trace.WithAttributes(observationToAttribute(key, obs)))
 	}
 
-	if region.parentID.IsZero() {
-		// Creates a new trace for the root region.
-		opts = append(opts, trace.WithNewRoot())
-	}
-
 	// Create the span
 	ctx, span := tracer.Start(ctx, region.name, opts...)
 
-	// Get children directly from the map (O(1) lookup)
 	children := parentToChildren[region.id]
-
 	// Recursively create spans for children
 	for _, child := range children {
 		if err := createSpans(ctx, child, parentToChildren); err != nil {
@@ -83,7 +76,7 @@ func createSpans(ctx context.Context, region *Region, parentToChildren map[ident
 
 // observationToAttribute converts an observation to an OpenTelemetry attribute.
 // The attribute key is the statistic name, and the value depends on the data type.
-func observationToAttribute(key StatisticKey, obs AggregatedObservation) attribute.KeyValue {
+func observationToAttribute(key StatisticKey, obs *AggregatedObservation) attribute.KeyValue {
 	attrKey := attribute.Key(key.Name)
 
 	switch key.DataType {

pkg/xcap/marshal.go

@@ -2,6 +2,8 @@ package xcap
 
 import (
 	"fmt"
+
+	"go.opentelemetry.io/otel/attribute"
 )
 
 // ToProtoCapture converts a Capture to its protobuf representation.
@@ -63,13 +65,24 @@ func toProtoRegion(region *Region, statsIndex map[StatisticKey]uint32) (*ProtoRe
 		})
 	}
 
+	// Convert attributes to proto attributes
+	protoAttributes := make([]*ProtoAttribute, 0, len(region.attributes))
+	for _, attr := range region.attributes {
+		protoAttr, err := marshalAttribute(attr)
+		if err != nil {
+			return nil, fmt.Errorf("failed to marshal attribute %s: %w", attr.Key, err)
+		}
+		protoAttributes = append(protoAttributes, protoAttr)
+	}
+
 	return &ProtoRegion{
 		Name:         region.name,
 		StartTime:    region.startTime,
 		EndTime:      region.endTime,
 		Observations: protoObservations,
 		Id:           region.id[:],
 		ParentId:     region.parentID[:],
+		Attributes:   protoAttributes,
 	}, nil
 }
 
@@ -128,3 +141,30 @@ func marshalAggregationType(agg AggregationType) ProtoAggregationType {
 		return PROTO_AGGREGATION_TYPE_INVALID
 	}
 }
+
+// marshalAttribute converts an OpenTelemetry attribute to its protobuf representation.
+func marshalAttribute(attr attribute.KeyValue) (*ProtoAttribute, error) {
+	if !attr.Valid() {
+		return nil, fmt.Errorf("invalid attribute")
+	}
+
+	protoValue := &ProtoAttributeValue{}
+	switch attr.Value.Type() {
+	case attribute.STRING:
+		protoValue.Kind = &ProtoAttributeValue_StringValue{StringValue: attr.Value.AsString()}
+	case attribute.INT64:
+		protoValue.Kind = &ProtoAttributeValue_IntValue{IntValue: attr.Value.AsInt64()}
+	case attribute.FLOAT64:
+		protoValue.Kind = &ProtoAttributeValue_FloatValue{FloatValue: attr.Value.AsFloat64()}
+	case attribute.BOOL:
+		protoValue.Kind = &ProtoAttributeValue_BoolValue{BoolValue: attr.Value.AsBool()}
+	default:
+		// For unsupported types (like slices), convert to string
+		protoValue.Kind = &ProtoAttributeValue_StringValue{StringValue: attr.Value.Emit()}
+	}
+
+	return &ProtoAttribute{
+		Key:   string(attr.Key),
+		Value: protoValue,
+	}, nil
+}