discovery: fold first QueryError per Execute span (Honeycomb cost) (#4645)

<!-- CURSOR_AGENT_PR_BODY_BEGIN -->
## Summary

Reduces Honeycomb **exception event** volume by **not** calling
`span.RecordError` for the **first** `*sdp.QueryError` on each `Execute`
span. Second and subsequent `*sdp.QueryError` values still emit
exception events (without stack traces). Non-`QueryError` stream errors
use the previous behavior (full `RecordError` with stack trace).

## Details

- **Attributes:** `ovm.adapter.sdpQueryErrorCount` (total
`*sdp.QueryError` from stream) and
`ovm.adapter.sdpQueryErrorRecordErrorCount` (how many became
`RecordError`, i.e. 2nd+).
- **Tracer:** `getTracer()` resolves the global `TracerProvider` at span
creation time so tests can install `sdktrace` with an in-memory
exporter.
- **Bench adapter:** `BenchmarkListAdapter.List` lazily initializes
`sdpcache.NewNoOpCache()` when `cache` is nil (avoids nil deref when
spans are fully recorded).

## Tests

- New `execute_query_trace_test.go`: embedded NATS + in-memory span
exporter; covers first QueryError (no exception event), second
QueryError (one exception event), plain error (one exception event).

## Docs

- `go/discovery/README.md`: behavior + post-deploy Honeycomb validation
hint.

## Commit message

The commit body includes the Honeycomb analysis summary and expected
savings as requested by the implementation plan.
<!-- CURSOR_AGENT_PR_BODY_END -->

<div><a
href="https://cursor.com/agents/bc-65cfd344-f928-48b1-bab0-2fb9e06aac15"><picture><source
media="(prefers-color-scheme: dark)"
srcset="https://cursor.com/assets/images/open-in-web-dark.png"><source
media="(prefers-color-scheme: light)"
srcset="https://cursor.com/assets/images/open-in-web-light.png"><img
alt="Open in Web" width="114" height="28"
src="https://cursor.com/assets/images/open-in-web-dark.png"></picture></a>&nbsp;<a
href="https://cursor.com/background-agent?bcId=bc-65cfd344-f928-48b1-bab0-2fb9e06aac15"><picture><source
media="(prefers-color-scheme: dark)"
srcset="https://cursor.com/assets/images/open-in-cursor-dark.png"><source
media="(prefers-color-scheme: light)"
srcset="https://cursor.com/assets/images/open-in-cursor-light.png"><img
alt="Open in Cursor" width="131" height="28"
src="https://cursor.com/assets/images/open-in-cursor-dark.png"></picture></a>&nbsp;</div>
GitOrigin-RevId: a64db47824f706de83ac4a335372fc85b1fb3ce1

Author: DavidS-ovm

go/discovery/adapterhost_bench_test.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/overmindtech/cli/go/sdp-go"
+	"github.com/overmindtech/cli/go/sdpcache"
 	"github.com/sourcegraph/conc/pool"
 )
 
@@ -245,6 +246,9 @@ type BenchmarkListAdapter struct {
 
 // List returns exactly 10 items (or itemsPerList if set) for each LIST query
 func (b *BenchmarkListAdapter) List(ctx context.Context, scope string, ignoreCache bool) ([]*sdp.Item, error) {
+	if b.cache == nil {
+		b.cache = sdpcache.NewNoOpCache()
+	}
 	// Use the embedded TestAdapter's List method logic but return multiple items
 	// We'll call the parent's cache lookup, but then generate multiple items
 	itemsPerList := b.itemsPerList

go/discovery/enginerequests.go

@@ -98,7 +98,7 @@ func (e *Engine) HandleQuery(ctx context.Context, query *sdp.Query) {
 	u, uuidErr := uuid.FromBytes(query.GetUUID())
 
 	// Only start the span if we actually have something that will respond
-	ctx, span := tracer.Start(ctx, "HandleQuery", trace.WithAttributes(
+	ctx, span := getTracer().Start(ctx, "HandleQuery", trace.WithAttributes(
 		attribute.Int("ovm.discovery.numExpandedQueries", numExpandedQueries),
 		attribute.Bool("ovm.sdp.deadlineOverridden", deadlineOverride),
 		attribute.String("ovm.sdp.source_name", e.EngineConfig.SourceName),
@@ -400,7 +400,7 @@ func (e *Engine) ExecuteQuery(ctx context.Context, query *sdp.Query, responses c
 // closed by this function, the caller should do that as this will likely be
 // called in parallel with other queries and the results should be merged
 func (e *Engine) Execute(ctx context.Context, q *sdp.Query, adapter Adapter, responses chan<- *sdp.QueryResponse) {
-	ctx, span := tracer.Start(ctx, "Execute", trace.WithAttributes(
+	ctx, span := getTracer().Start(ctx, "Execute", trace.WithAttributes(
 		attribute.String("ovm.adapter.name", adapter.Name()),
 		attribute.String("ovm.engine.type", e.EngineConfig.EngineType),
 		attribute.String("ovm.engine.version", e.EngineConfig.Version),
@@ -458,6 +458,11 @@ func (e *Engine) Execute(ctx context.Context, q *sdp.Query, adapter Adapter, res
 	// are passed back to the caller
 	var numItems atomic.Int32
 	var numErrs atomic.Int32
+	// Per-Execute *sdp.QueryError telemetry: fold the first into span aggregates only
+	// (no RecordError) to reduce Honeycomb exception-event volume; record 2+ as
+	// exception events so rare multi-error tails keep detail.
+	var numSDPQueryErrors atomic.Int32
+	var numSDPQueryErrorRecordErrors atomic.Int32
 	var channelSendMaxNs atomic.Int64
 	var channelSendTotalNs atomic.Int64
 	var itemHandler ItemHandler = func(item *sdp.Item) {
@@ -519,8 +524,20 @@ func (e *Engine) Execute(ctx context.Context, q *sdp.Query, adapter Adapter, res
 		// add a recover to prevent panic from stream error handler.
 		defer tracing.LogRecoverToReturn(ctx, "StreamErrorHandler")
 
-		// Record the error in the trace
-		span.RecordError(err, trace.WithStackTrace(true))
+		var sdpErr *sdp.QueryError
+		if errors.As(err, &sdpErr) && sdpErr != nil {
+			n := numSDPQueryErrors.Add(1)
+			if n == 1 {
+				// Fold first QueryError: do not emit a per-error exception event.
+			} else {
+				// Rare multi-error Execute: keep per-error exception rows without stacks
+				// (stacks are expensive and the first error is the high-volume case).
+				span.RecordError(sdpErr, trace.WithStackTrace(false))
+				numSDPQueryErrorRecordErrors.Add(1)
+			}
+		} else {
+			span.RecordError(err, trace.WithStackTrace(true))
+		}
 
 		// Send the error back to the caller
 		numErrs.Add(1)
@@ -616,6 +633,8 @@ func (e *Engine) Execute(ctx context.Context, q *sdp.Query, adapter Adapter, res
 	span.SetAttributes(
 		attribute.Int("ovm.adapter.numItems", int(numItems.Load())),
 		attribute.Int("ovm.adapter.numErrors", int(numErrs.Load())),
+		attribute.Int("ovm.adapter.sdpQueryErrorCount", int(numSDPQueryErrors.Load())),
+		attribute.Int("ovm.adapter.sdpQueryErrorRecordErrorCount", int(numSDPQueryErrorRecordErrors.Load())),
 		attribute.Float64("ovm.discovery.channelSendMaxMs", float64(channelSendMaxNs.Load())/1e6),
 		attribute.Float64("ovm.discovery.channelSendTotalMs", float64(channelSendTotalNs.Load())/1e6),
 	)

go/discovery/execute_query_trace_test.go

@@ -0,0 +1,223 @@
+package discovery
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/nats-io/nats-server/v2/test"
+	"github.com/overmindtech/cli/go/auth"
+	"github.com/overmindtech/cli/go/sdp-go"
+	"github.com/overmindtech/cli/go/sdpcache"
+	"go.opentelemetry.io/otel"
+	sdktrace "go.opentelemetry.io/otel/sdk/trace"
+	"go.opentelemetry.io/otel/sdk/trace/tracetest"
+	semconv "go.opentelemetry.io/otel/semconv/v1.26.0"
+	"google.golang.org/protobuf/types/known/timestamppb"
+)
+
+// startEmbeddedNATSServer runs an in-process NATS for tests that need a live Engine Start.
+func startEmbeddedNATSServer(t *testing.T) string {
+	t.Helper()
+	opts := test.DefaultTestOptions
+	opts.Port = 4739
+	s := test.RunServer(&opts)
+	if !s.ReadyForConnections(10 * time.Second) {
+		s.Shutdown()
+		t.Fatal("could not start embedded NATS server")
+	}
+	t.Cleanup(func() {
+		s.Shutdown()
+	})
+	return s.ClientURL()
+}
+
+func setupTestTracer(t *testing.T) *tracetest.InMemoryExporter {
+	t.Helper()
+	exp := tracetest.NewInMemoryExporter()
+	tp := sdktrace.NewTracerProvider(
+		sdktrace.WithSyncer(exp),
+		sdktrace.WithSampler(sdktrace.AlwaysSample()),
+	)
+	prev := otel.GetTracerProvider()
+	otel.SetTracerProvider(tp)
+	t.Cleanup(func() {
+		_ = tp.Shutdown(context.Background())
+		otel.SetTracerProvider(prev)
+	})
+	return exp
+}
+
+func countExceptionEvents(spans []tracetest.SpanStub) int {
+	n := 0
+	for _, s := range spans {
+		if s.Name != "Execute" {
+			continue
+		}
+		for _, ev := range s.Events {
+			if ev.Name == semconv.ExceptionEventName {
+				n++
+			}
+		}
+	}
+	return n
+}
+
+// streamTwoSDPQueryErrorsAdapter implements ListStreamableAdapter and emits two *sdp.QueryError
+// values on LIST (for multi-error Execute telemetry tests).
+type streamTwoSDPQueryErrorsAdapter struct {
+	*TestAdapter
+}
+
+func (a *streamTwoSDPQueryErrorsAdapter) ListStream(ctx context.Context, scope string, ignoreCache bool, stream QueryResultStream) {
+	_ = ctx
+	_ = scope
+	_ = ignoreCache
+	stream.SendError(&sdp.QueryError{
+		ErrorType:   sdp.QueryError_OTHER,
+		ErrorString: "first sdp query error",
+	})
+	stream.SendError(&sdp.QueryError{
+		ErrorType:   sdp.QueryError_OTHER,
+		ErrorString: "second sdp query error",
+	})
+}
+
+// plainErrOnGetAdapter returns a non-QueryError from Get for every call.
+type plainErrOnGetAdapter struct {
+	*TestAdapter
+}
+
+func (a *plainErrOnGetAdapter) Get(ctx context.Context, scope string, query string, ignoreCache bool) (*sdp.Item, error) {
+	_ = ctx
+	_ = scope
+	_ = query
+	_ = ignoreCache
+	return nil, fmt.Errorf("plain non-sdp error")
+}
+
+func TestExecute_FirstSDPQueryErrorDoesNotRecordExceptionEvent(t *testing.T) {
+	exp := setupTestTracer(t)
+	natsURL := startEmbeddedNATSServer(t)
+
+	adapter := TestAdapter{
+		ReturnType:   "person",
+		ReturnScopes: []string{"test", "error"},
+		cache:        sdpcache.NewNoOpCache(),
+	}
+
+	e := newStartedEngine(t, "TestExecuteTraceSDPQueryError", &auth.NATSOptions{
+		Servers:           []string{natsURL},
+		ConnectionName:    "test-connection",
+		ConnectionTimeout: time.Second,
+		MaxReconnects:     5,
+	}, nil, &adapter)
+
+	u := uuid.New()
+	q := &sdp.Query{
+		UUID:     u[:],
+		Type:     "person",
+		Method:   sdp.QueryMethod_GET,
+		Query:    "foo",
+		Scope:    "error",
+		Deadline: timestamppb.New(time.Now().Add(time.Minute)),
+		RecursionBehaviour: &sdp.Query_RecursionBehaviour{
+			LinkDepth: 3,
+		},
+	}
+
+	ch := make(chan *sdp.QueryResponse, 10)
+	err := e.ExecuteQuery(context.Background(), q, ch)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if n := countExceptionEvents(exp.GetSpans()); n != 0 {
+		t.Fatalf("expected 0 exception events on Execute for first *sdp.QueryError, got %d", n)
+	}
+}
+
+func TestExecute_SecondSDPQueryErrorRecordsExceptionEvent(t *testing.T) {
+	exp := setupTestTracer(t)
+	natsURL := startEmbeddedNATSServer(t)
+
+	base := &TestAdapter{
+		ReturnType:   "person",
+		ReturnScopes: []string{"test"},
+		cache:        sdpcache.NewNoOpCache(),
+	}
+	adapter := &streamTwoSDPQueryErrorsAdapter{TestAdapter: base}
+
+	e := newStartedEngine(t, "TestExecuteTraceMultiSDPQueryError", &auth.NATSOptions{
+		Servers:           []string{natsURL},
+		ConnectionName:    "test-connection",
+		ConnectionTimeout: time.Second,
+		MaxReconnects:     5,
+	}, nil, adapter)
+
+	u := uuid.New()
+	q := &sdp.Query{
+		UUID:     u[:],
+		Type:     "person",
+		Method:   sdp.QueryMethod_LIST,
+		Scope:    "test",
+		Deadline: timestamppb.New(time.Now().Add(time.Minute)),
+		RecursionBehaviour: &sdp.Query_RecursionBehaviour{
+			LinkDepth: 3,
+		},
+	}
+
+	ch := make(chan *sdp.QueryResponse, 10)
+	err := e.ExecuteQuery(context.Background(), q, ch)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if n := countExceptionEvents(exp.GetSpans()); n != 1 {
+		t.Fatalf("expected 1 exception event on Execute (2nd *sdp.QueryError only), got %d", n)
+	}
+}
+
+func TestExecute_PlainErrorStillRecordsExceptionEvent(t *testing.T) {
+	exp := setupTestTracer(t)
+	natsURL := startEmbeddedNATSServer(t)
+
+	base := &TestAdapter{
+		ReturnType:   "person",
+		ReturnScopes: []string{"test"},
+		cache:        sdpcache.NewNoOpCache(),
+	}
+	adapter := &plainErrOnGetAdapter{TestAdapter: base}
+
+	e := newStartedEngine(t, "TestExecuteTracePlainErr", &auth.NATSOptions{
+		Servers:           []string{natsURL},
+		ConnectionName:    "test-connection",
+		ConnectionTimeout: time.Second,
+		MaxReconnects:     5,
+	}, nil, adapter)
+
+	u := uuid.New()
+	q := &sdp.Query{
+		UUID:     u[:],
+		Type:     "person",
+		Method:   sdp.QueryMethod_GET,
+		Query:    "foo",
+		Scope:    "test",
+		Deadline: timestamppb.New(time.Now().Add(time.Minute)),
+		RecursionBehaviour: &sdp.Query_RecursionBehaviour{
+			LinkDepth: 3,
+		},
+	}
+
+	ch := make(chan *sdp.QueryResponse, 10)
+	err := e.ExecuteQuery(context.Background(), q, ch)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if n := countExceptionEvents(exp.GetSpans()); n != 1 {
+		t.Fatalf("expected 1 exception event for plain error, got %d", n)
+	}
+}

go/discovery/heartbeat.go

@@ -25,7 +25,7 @@ var ErrNoHealthcheckDefined = errors.New("no healthcheck defined")
 // to indicate that the engine is in an error state, this will be sent to the
 // management API and will be displayed in the UI.
 func (e *Engine) SendHeartbeat(ctx context.Context, customErr error) error {
-	ctx, span := tracer.Start(ctx, "SendHeartbeat")
+	ctx, span := getTracer().Start(ctx, "SendHeartbeat")
 	defer span.End()
 
 	// Read memory stats and add them to the span

go/discovery/tracing.go

@@ -11,10 +11,13 @@ const (
 	instrumentationVersion = "0.0.1"
 )
 
-var (
-	tracer = otel.GetTracerProvider().Tracer(
+// getTracer returns the discovery tracer from the current global TracerProvider.
+// Call this at span creation time (not once at init) so tests can install an
+// in-memory TracerProvider before running discovery code.
+func getTracer() trace.Tracer {
+	return otel.GetTracerProvider().Tracer(
 		instrumentationName,
 		trace.WithInstrumentationVersion(instrumentationVersion),
 		trace.WithSchemaURL(semconv.SchemaURL),
 	)
-)
+}