Add OpenTelemetry tracing to capability aggregation

Add tracing spans to all aggregator methods to enable visibility
of capability aggregation in Jaeger. This includes spans for:
- AggregateCapabilities (parent span)
- QueryAllCapabilities (parallel backend queries)
- QueryCapabilities (per-backend queries)
- ResolveConflicts (conflict resolution)
- MergeCapabilities (final merge)

All spans include relevant attributes like backend counts, tool/resource/prompt
counts, and error recording. This fixes the issue where capability aggregation
logs appeared but no spans were visible in Jaeger.

Author: nigel brown

pkg/vmcp/aggregator/default_aggregator.go

@@ -8,6 +8,10 @@ import (
 	"fmt"
 	"sync"
 
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/codes"
+	"go.opentelemetry.io/otel/trace"
 	"golang.org/x/sync/errgroup"
 
 	"github.com/stacklok/toolhive/pkg/logger"
@@ -21,6 +25,7 @@ type defaultAggregator struct {
 	backendClient    vmcp.BackendClient
 	conflictResolver ConflictResolver
 	toolConfigMap    map[string]*config.WorkloadToolConfig // Maps backend ID to tool config
+	tracer           trace.Tracer
 }
 
 // NewDefaultAggregator creates a new default aggregator implementation.
@@ -43,12 +48,20 @@ func NewDefaultAggregator(
 		backendClient:    backendClient,
 		conflictResolver: conflictResolver,
 		toolConfigMap:    toolConfigMap,
+		tracer:           otel.Tracer("github.com/stacklok/toolhive/pkg/vmcp/aggregator"),
 	}
 }
 
 // QueryCapabilities queries a single backend for its MCP capabilities.
 // Returns the raw capabilities (tools, resources, prompts) from the backend.
 func (a *defaultAggregator) QueryCapabilities(ctx context.Context, backend vmcp.Backend) (*BackendCapabilities, error) {
+	ctx, span := a.tracer.Start(ctx, "aggregator.QueryCapabilities",
+		trace.WithAttributes(
+			attribute.String("backend.id", backend.ID),
+		),
+	)
+	defer span.End()
+
 	logger.Debugf("Querying capabilities from backend %s", backend.ID)
 
 	// Create a BackendTarget from the Backend
@@ -58,6 +71,8 @@ func (a *defaultAggregator) QueryCapabilities(ctx context.Context, backend vmcp.
 	// Query capabilities using the backend client
 	capabilities, err := a.backendClient.ListCapabilities(ctx, target)
 	if err != nil {
+		span.RecordError(err)
+		span.SetStatus(codes.Error, err.Error())
 		return nil, fmt.Errorf("%w: %s: %w", ErrBackendQueryFailed, backend.ID, err)
 	}
 
@@ -74,6 +89,12 @@ func (a *defaultAggregator) QueryCapabilities(ctx context.Context, backend vmcp.
 		SupportsSampling: capabilities.SupportsSampling,
 	}
 
+	span.SetAttributes(
+		attribute.Int("tools.count", len(result.Tools)),
+		attribute.Int("resources.count", len(result.Resources)),
+		attribute.Int("prompts.count", len(result.Prompts)),
+	)
+
 	logger.Debugf("Backend %s: %d tools (after filtering/overrides), %d resources, %d prompts",
 		backend.ID, len(result.Tools), len(result.Resources), len(result.Prompts))
 
@@ -86,6 +107,13 @@ func (a *defaultAggregator) QueryAllCapabilities(
 	ctx context.Context,
 	backends []vmcp.Backend,
 ) (map[string]*BackendCapabilities, error) {
+	ctx, span := a.tracer.Start(ctx, "aggregator.QueryAllCapabilities",
+		trace.WithAttributes(
+			attribute.Int("backends.count", len(backends)),
+		),
+	)
+	defer span.End()
+
 	logger.Infof("Querying capabilities from %d backends", len(backends))
 
 	// Use errgroup for parallel queries with context cancellation
@@ -118,13 +146,22 @@ func (a *defaultAggregator) QueryAllCapabilities(
 
 	// Wait for all queries to complete
 	if err := g.Wait(); err != nil {
+		span.RecordError(err)
+		span.SetStatus(codes.Error, err.Error())
 		return nil, fmt.Errorf("capability queries failed: %w", err)
 	}
 
 	if len(capabilities) == 0 {
-		return nil, fmt.Errorf("no backends returned capabilities")
+		err := fmt.Errorf("no backends returned capabilities")
+		span.RecordError(err)
+		span.SetStatus(codes.Error, err.Error())
+		return nil, err
 	}
 
+	span.SetAttributes(
+		attribute.Int("successful.backends", len(capabilities)),
+	)
+
 	logger.Infof("Successfully queried %d/%d backends", len(capabilities), len(backends))
 	return capabilities, nil
 }
@@ -135,6 +172,13 @@ func (a *defaultAggregator) ResolveConflicts(
 	ctx context.Context,
 	capabilities map[string]*BackendCapabilities,
 ) (*ResolvedCapabilities, error) {
+	ctx, span := a.tracer.Start(ctx, "aggregator.ResolveConflicts",
+		trace.WithAttributes(
+			attribute.Int("backends.count", len(capabilities)),
+		),
+	)
+	defer span.End()
+
 	logger.Debugf("Resolving conflicts across %d backends", len(capabilities))
 
 	// Group tools by backend for conflict resolution
@@ -150,6 +194,8 @@ func (a *defaultAggregator) ResolveConflicts(
 	if a.conflictResolver != nil {
 		resolvedTools, err = a.conflictResolver.ResolveToolConflicts(ctx, toolsByBackend)
 		if err != nil {
+			span.RecordError(err)
+			span.SetStatus(codes.Error, err.Error())
 			return nil, fmt.Errorf("conflict resolution failed: %w", err)
 		}
 	} else {
@@ -191,6 +237,12 @@ func (a *defaultAggregator) ResolveConflicts(
 		resolved.SupportsSampling = resolved.SupportsSampling || caps.SupportsSampling
 	}
 
+	span.SetAttributes(
+		attribute.Int("resolved.tools", len(resolved.Tools)),
+		attribute.Int("resolved.resources", len(resolved.Resources)),
+		attribute.Int("resolved.prompts", len(resolved.Prompts)),
+	)
+
 	logger.Debugf("Resolved %d unique tools, %d resources, %d prompts",
 		len(resolved.Tools), len(resolved.Resources), len(resolved.Prompts))
 
@@ -199,11 +251,20 @@ func (a *defaultAggregator) ResolveConflicts(
 
 // MergeCapabilities creates the final unified capability view and routing table.
 // Uses the backend registry to populate full BackendTarget information for routing.
-func (*defaultAggregator) MergeCapabilities(
+func (a *defaultAggregator) MergeCapabilities(
 	ctx context.Context,
 	resolved *ResolvedCapabilities,
 	registry vmcp.BackendRegistry,
 ) (*AggregatedCapabilities, error) {
+	ctx, span := a.tracer.Start(ctx, "aggregator.MergeCapabilities",
+		trace.WithAttributes(
+			attribute.Int("resolved.tools", len(resolved.Tools)),
+			attribute.Int("resolved.resources", len(resolved.Resources)),
+			attribute.Int("resolved.prompts", len(resolved.Prompts)),
+		),
+	)
+	defer span.End()
+
 	logger.Debugf("Merging capabilities into final view")
 
 	// Create routing table
@@ -304,6 +365,13 @@ func (*defaultAggregator) MergeCapabilities(
 		},
 	}
 
+	span.SetAttributes(
+		attribute.Int("aggregated.tools", aggregated.Metadata.ToolCount),
+		attribute.Int("aggregated.resources", aggregated.Metadata.ResourceCount),
+		attribute.Int("aggregated.prompts", aggregated.Metadata.PromptCount),
+		attribute.String("conflict.strategy", string(aggregated.Metadata.ConflictStrategy)),
+	)
+
 	logger.Infof("Merged capabilities: %d tools, %d resources, %d prompts",
 		aggregated.Metadata.ToolCount, aggregated.Metadata.ResourceCount, aggregated.Metadata.PromptCount)
 
@@ -316,6 +384,13 @@ func (*defaultAggregator) MergeCapabilities(
 // 3. Resolve conflicts
 // 4. Merge into final view with full backend information
 func (a *defaultAggregator) AggregateCapabilities(ctx context.Context, backends []vmcp.Backend) (*AggregatedCapabilities, error) {
+	ctx, span := a.tracer.Start(ctx, "aggregator.AggregateCapabilities",
+		trace.WithAttributes(
+			attribute.Int("backends.count", len(backends)),
+		),
+	)
+	defer span.End()
+
 	logger.Infof("Starting capability aggregation for %d backends", len(backends))
 
 	// Step 1: Create registry from discovered backends
@@ -325,24 +400,38 @@ func (a *defaultAggregator) AggregateCapabilities(ctx context.Context, backends
 	// Step 2: Query all backends
 	capabilities, err := a.QueryAllCapabilities(ctx, backends)
 	if err != nil {
+		span.RecordError(err)
+		span.SetStatus(codes.Error, err.Error())
 		return nil, fmt.Errorf("failed to query backends: %w", err)
 	}
 
 	// Step 3: Resolve conflicts
 	resolved, err := a.ResolveConflicts(ctx, capabilities)
 	if err != nil {
+		span.RecordError(err)
+		span.SetStatus(codes.Error, err.Error())
 		return nil, fmt.Errorf("failed to resolve conflicts: %w", err)
 	}
 
 	// Step 4: Merge into final view with full backend information
 	aggregated, err := a.MergeCapabilities(ctx, resolved, registry)
 	if err != nil {
+		span.RecordError(err)
+		span.SetStatus(codes.Error, err.Error())
 		return nil, fmt.Errorf("failed to merge capabilities: %w", err)
 	}
 
 	// Update metadata with backend count
 	aggregated.Metadata.BackendCount = len(backends)
 
+	span.SetAttributes(
+		attribute.Int("aggregated.backends", aggregated.Metadata.BackendCount),
+		attribute.Int("aggregated.tools", aggregated.Metadata.ToolCount),
+		attribute.Int("aggregated.resources", aggregated.Metadata.ResourceCount),
+		attribute.Int("aggregated.prompts", aggregated.Metadata.PromptCount),
+		attribute.String("conflict.strategy", string(aggregated.Metadata.ConflictStrategy)),
+	)
+
 	logger.Infof("Capability aggregation complete: %d backends, %d tools, %d resources, %d prompts",
 		aggregated.Metadata.BackendCount, aggregated.Metadata.ToolCount,
 		aggregated.Metadata.ResourceCount, aggregated.Metadata.PromptCount)