Infrastructure improvements and bugfixes for vMCP (#3439)

* Infrastructure improvements and bugfixes for vMCP

- Add OpenTelemetry tracing to capability aggregation
- Add singleflight deduplication for discovery requests
- Add health checker self-check prevention
- Add HTTP client timeout fixes
- Improve E2E test reliability
- Various build and infrastructure improvements

* fix: Update CallTool and GetPrompt signatures to match BackendClient interface

- Add conversion import for meta field handling
- Update CallTool to accept meta parameter and return *vmcp.ToolCallResult
- Update GetPrompt to return *vmcp.PromptGetResult
- Add convertContent helper function

* fix: Update ReadResource signature to match BackendClient interface

- Update ReadResource to return *vmcp.ResourceReadResult instead of []byte
- Extract and include meta field from backend response
- Include MIME type in result

* fix: Pass selfURL parameter to health.NewMonitor

- Construct selfURL from Host, Port, and EndpointPath
- Prevents health checker from checking itself

* Fix NewHealthChecker calls in checker_test.go to include selfURL parameter

* Fix NewMonitor calls in monitor_test.go to include selfURL parameter

All 10 calls to NewMonitor in monitor_test.go were missing the new selfURL parameter that was added to the function signature. This was causing compilation failures in CI.

* Fix Go import formatting issues (gci linter)

Fixed import ordering in:
- pkg/vmcp/client/client.go
- pkg/vmcp/health/checker_test.go
- pkg/vmcp/health/monitor_test.go

* Fix Chart.yaml version - restore to 0.0.103

The version was incorrectly downgraded to 0.0.102. Restore it to 0.0.103 to match main branch.

* Bump Chart.yaml version to 0.0.104

The chart-testing tool requires version bumps to be higher than the base branch version (0.0.103).

* Update README.md version badge to 0.0.104

Match the Chart.yaml version update to satisfy helm-docs pre-commit hook.

* Refactor vMCP tracing and remove health checker self-check

Move telemetry provider initialization earlier in vmcp serve command to
enable distributed tracing in the aggregator. The aggregator now accepts
an explicit tracer provider parameter instead of using the global otel
tracer, following dependency injection best practices.

Improve tracing error handling by using named return values and deferred
error recording in aggregator methods, ensuring errors are properly
captured in traces.

Remove health checker self-check functionality that prevented the server
from checking its own health endpoint. This simplifies the implementation
and removes unnecessary URL normalization logic.

Changes:
- Add tracerProvider parameter to aggregator.NewDefaultAggregator
- Use noop tracer when provider is nil
- Improve span error handling with deferred recording
- Remove selfURL parameter from health.NewHealthChecker
- Delete pkg/vmcp/health/checker_selfcheck_test.go
- Update all tests to match new function signatures
- Add debug logging for auth strategy application in client

* Add explanatory comment for MCP SDK Meta limitations

Restores comment explaining why Meta field preservation is important
for ReadResource, in anticipation of future SDK improvements.

This addresses PR feedback to maintain context about the SDK's
current limitations regarding Meta field handling.

* Update test helper comments to clarify pod readiness contract

- Clarify that checkPodsReady waits for at least one pod (not all pods)
- Add context that helpers are used for single replica deployments
- Update comments on WaitForPodsReady and WaitForVirtualMCPServerReady

Addresses code review feedback from PR review.

* Complete error capture pattern in MergeCapabilities defer

- Add named return value (retErr error) to MergeCapabilities
- Add error capture in defer statement with span.RecordError and span.SetStatus
- Ensures consistent error handling pattern across all aggregator methods

This completes the implementation of the error capture pattern suggested
in code review for all methods with tracing spans.

* Remove singleflight race condition fix

Moving the singleflight deduplication logic to a separate PR
as it addresses a different race condition from the one fixed in #3450.

The fix prevents duplicate capability aggregation when multiple
concurrent requests arrive simultaneously at startup.

* Add SPDX license headers to manager.go

Author: Nigel Brown

.gitignore

@@ -42,4 +42,5 @@ cmd/thv-operator/.task/checksum/crdref-gen
 # Test coverage
 coverage*
 
-crd-helm-wrapper
+crd-helm-wrapper
+cmd/vmcp/__debug_bin*

.golangci.yml

@@ -139,6 +139,7 @@ linters:
       - third_party$
       - builtin$
       - examples$
+      - scripts$
 formatters:
   enable:
     - gci
@@ -155,3 +156,4 @@ formatters:
       - third_party$
       - builtin$
       - examples$
+      - scripts$

cmd/vmcp/app/commands.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
+	"go.opentelemetry.io/otel/trace"
 	"k8s.io/client-go/rest"
 
 	"github.com/stacklok/toolhive/pkg/audit"
@@ -310,8 +311,27 @@ func runServe(cmd *cobra.Command, _ []string) error {
 		return fmt.Errorf("failed to create conflict resolver: %w", err)
 	}
 
-	// Create aggregator
-	agg := aggregator.NewDefaultAggregator(backendClient, conflictResolver, cfg.Aggregation.Tools)
+	// If telemetry is configured, create the provider early so aggregator can use it
+	var telemetryProvider *telemetry.Provider
+	if cfg.Telemetry != nil {
+		telemetryProvider, err = telemetry.NewProvider(ctx, *cfg.Telemetry)
+		if err != nil {
+			return fmt.Errorf("failed to create telemetry provider: %w", err)
+		}
+		defer func() {
+			err := telemetryProvider.Shutdown(ctx)
+			if err != nil {
+				logger.Errorf("failed to shutdown telemetry provider: %v", err)
+			}
+		}()
+	}
+
+	// Create aggregator with tracer provider (nil if telemetry not configured)
+	var tracerProvider trace.TracerProvider
+	if telemetryProvider != nil {
+		tracerProvider = telemetryProvider.TracerProvider()
+	}
+	agg := aggregator.NewDefaultAggregator(backendClient, conflictResolver, cfg.Aggregation.Tools, tracerProvider)
 
 	// Use DynamicRegistry for version-based cache invalidation
 	// Works in both standalone (CLI with YAML config) and Kubernetes (operator-deployed) modes
@@ -381,21 +401,8 @@ func runServe(cmd *cobra.Command, _ []string) error {
 	host, _ := cmd.Flags().GetString("host")
 	port, _ := cmd.Flags().GetInt("port")
 
-	// If telemetry is configured, create the provider.
-	var telemetryProvider *telemetry.Provider
-	if cfg.Telemetry != nil {
-		var err error
-		telemetryProvider, err = telemetry.NewProvider(ctx, *cfg.Telemetry)
-		if err != nil {
-			return fmt.Errorf("failed to create telemetry provider: %w", err)
-		}
-		defer func() {
-			err := telemetryProvider.Shutdown(ctx)
-			if err != nil {
-				logger.Errorf("failed to shutdown telemetry provider: %v", err)
-			}
-		}()
-	}
+	// Note: telemetryProvider was already created earlier (before aggregator creation)
+	// to enable tracing in the aggregator
 
 	// Configure health monitoring if enabled
 	var healthMonitorConfig *health.MonitorConfig

codecov.yaml

@@ -13,6 +13,8 @@ coverage:
     - "**/mocks/**/*"
     - "**/mock_*.go"
     - "**/zz_generated.deepcopy.go"
+    - "**/*_test.go"
+    - "**/*_test_coverage.go"
   status:
     project:
       default:

pkg/runner/config_builder_test.go

@@ -1079,8 +1079,8 @@ func TestRunConfigBuilder_WithRegistryProxyPort(t *testing.T) {
 				ProxyPort:  testPort,
 				TargetPort: testPort,
 			},
-			cliProxyPort:      9000,
-			expectedProxyPort: 9000,
+			cliProxyPort:      9999,
+			expectedProxyPort: 9999,
 		},
 		{
 			name: "random port when neither CLI nor registry specified",

pkg/vmcp/aggregator/default_aggregator.go

@@ -8,6 +8,10 @@ import (
 	"fmt"
 	"sync"
 
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/codes"
+	"go.opentelemetry.io/otel/trace"
+	"go.opentelemetry.io/otel/trace/noop"
 	"golang.org/x/sync/errgroup"
 
 	"github.com/stacklok/toolhive/pkg/logger"
@@ -21,15 +25,18 @@ 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.
 // conflictResolver handles tool name conflicts across backends.
 // workloadConfigs specifies per-backend tool filtering and overrides.
+// tracerProvider is used to create a tracer for distributed tracing (pass nil for no tracing).
 func NewDefaultAggregator(
 	backendClient vmcp.BackendClient,
 	conflictResolver ConflictResolver,
 	workloadConfigs []*config.WorkloadToolConfig,
+	tracerProvider trace.TracerProvider,
 ) Aggregator {
 	// Build tool config map for quick lookup by backend ID
 	toolConfigMap := make(map[string]*config.WorkloadToolConfig)
@@ -39,16 +46,38 @@ func NewDefaultAggregator(
 		}
 	}
 
+	// Create tracer from provider (use noop tracer if provider is nil)
+	var tracer trace.Tracer
+	if tracerProvider != nil {
+		tracer = tracerProvider.Tracer("github.com/stacklok/toolhive/pkg/vmcp/aggregator")
+	} else {
+		tracer = noop.NewTracerProvider().Tracer("github.com/stacklok/toolhive/pkg/vmcp/aggregator")
+	}
+
 	return &defaultAggregator{
 		backendClient:    backendClient,
 		conflictResolver: conflictResolver,
 		toolConfigMap:    toolConfigMap,
+		tracer:           tracer,
 	}
 }
 
 // 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) {
+func (a *defaultAggregator) QueryCapabilities(ctx context.Context, backend vmcp.Backend) (_ *BackendCapabilities, retErr error) {
+	ctx, span := a.tracer.Start(ctx, "aggregator.QueryCapabilities",
+		trace.WithAttributes(
+			attribute.String("backend.id", backend.ID),
+		),
+	)
+	defer func() {
+		if retErr != nil {
+			span.RecordError(retErr)
+			span.SetStatus(codes.Error, retErr.Error())
+		}
+		span.End()
+	}()
+
 	logger.Debugf("Querying capabilities from backend %s", backend.ID)
 
 	// Create a BackendTarget from the Backend
@@ -74,6 +103,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))
 
@@ -85,7 +120,20 @@ func (a *defaultAggregator) QueryCapabilities(ctx context.Context, backend vmcp.
 func (a *defaultAggregator) QueryAllCapabilities(
 	ctx context.Context,
 	backends []vmcp.Backend,
-) (map[string]*BackendCapabilities, error) {
+) (_ map[string]*BackendCapabilities, retErr error) {
+	ctx, span := a.tracer.Start(ctx, "aggregator.QueryAllCapabilities",
+		trace.WithAttributes(
+			attribute.Int("backends.count", len(backends)),
+		),
+	)
+	defer func() {
+		if retErr != nil {
+			span.RecordError(retErr)
+			span.SetStatus(codes.Error, retErr.Error())
+		}
+		span.End()
+	}()
+
 	logger.Infof("Querying capabilities from %d backends", len(backends))
 
 	// Use errgroup for parallel queries with context cancellation
@@ -125,6 +173,10 @@ func (a *defaultAggregator) QueryAllCapabilities(
 		return nil, fmt.Errorf("no backends returned capabilities")
 	}
 
+	span.SetAttributes(
+		attribute.Int("successful.backends", len(capabilities)),
+	)
+
 	logger.Infof("Successfully queried %d/%d backends", len(capabilities), len(backends))
 	return capabilities, nil
 }
@@ -134,7 +186,20 @@ func (a *defaultAggregator) QueryAllCapabilities(
 func (a *defaultAggregator) ResolveConflicts(
 	ctx context.Context,
 	capabilities map[string]*BackendCapabilities,
-) (*ResolvedCapabilities, error) {
+) (_ *ResolvedCapabilities, retErr error) {
+	ctx, span := a.tracer.Start(ctx, "aggregator.ResolveConflicts",
+		trace.WithAttributes(
+			attribute.Int("backends.count", len(capabilities)),
+		),
+	)
+	defer func() {
+		if retErr != nil {
+			span.RecordError(retErr)
+			span.SetStatus(codes.Error, retErr.Error())
+		}
+		span.End()
+	}()
+
 	logger.Debugf("Resolving conflicts across %d backends", len(capabilities))
 
 	// Group tools by backend for conflict resolution
@@ -191,6 +256,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 +270,26 @@ 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) {
+) (_ *AggregatedCapabilities, retErr 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 func() {
+		if retErr != nil {
+			span.RecordError(retErr)
+			span.SetStatus(codes.Error, retErr.Error())
+		}
+		span.End()
+	}()
+
 	logger.Debugf("Merging capabilities into final view")
 
 	// Create routing table
@@ -304,6 +390,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)
 
@@ -315,7 +408,23 @@ func (*defaultAggregator) MergeCapabilities(
 // 2. Query all backends
 // 3. Resolve conflicts
 // 4. Merge into final view with full backend information
-func (a *defaultAggregator) AggregateCapabilities(ctx context.Context, backends []vmcp.Backend) (*AggregatedCapabilities, error) {
+func (a *defaultAggregator) AggregateCapabilities(
+	ctx context.Context,
+	backends []vmcp.Backend,
+) (_ *AggregatedCapabilities, retErr error) {
+	ctx, span := a.tracer.Start(ctx, "aggregator.AggregateCapabilities",
+		trace.WithAttributes(
+			attribute.Int("backends.count", len(backends)),
+		),
+	)
+	defer func() {
+		if retErr != nil {
+			span.RecordError(retErr)
+			span.SetStatus(codes.Error, retErr.Error())
+		}
+		span.End()
+	}()
+
 	logger.Infof("Starting capability aggregation for %d backends", len(backends))
 
 	// Step 1: Create registry from discovered backends
@@ -343,6 +452,14 @@ func (a *defaultAggregator) AggregateCapabilities(ctx context.Context, backends
 	// 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)