From c0660b83987eec4f79bbcd36772f8a9f0a162b57 Mon Sep 17 00:00:00 2001 From: Sergey Kozyrenko Date: Sun, 12 Jul 2026 16:45:41 +0700 Subject: [PATCH] fix(observability): connect the telemetry collector lazily, tear down cleanly MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The startup dial used grpc.WithBlock with a 10s timeout, so a set-but- unreachable collector stalled main() for up to 10s, and once that dial failed the client was nil forever — a collector that came up later never connected without a restart. Switch to grpc.NewClient (non-blocking): startup never stalls and gRPC re-establishes the connection in the background on its own. Also build all three exporters before any provider, so the exporter-error cleanup (closing the connection) happens before any batch/reader goroutine starts and can't leak one. Drops the deprecated WithBlock/WithReturnConnectionError. Co-Authored-By: Claude Opus 4.8 --- backend/pkg/observability/otelclient.go | 79 ++++++++------------ backend/pkg/observability/otelclient_test.go | 47 ++++++------ 2 files changed, 55 insertions(+), 71 deletions(-) diff --git a/backend/pkg/observability/otelclient.go b/backend/pkg/observability/otelclient.go index 599787b9..62a946fe 100644 --- a/backend/pkg/observability/otelclient.go +++ b/backend/pkg/observability/otelclient.go @@ -32,7 +32,6 @@ const ( DefaultMetricTimeout = time.Second * 10 DefaultTraceInterval = time.Second * 30 DefaultTraceTimeout = time.Second * 10 - DefaultDialTimeout = time.Second * 10 ) type TelemetryClient interface { @@ -93,74 +92,62 @@ func NewTelemetryClient(ctx context.Context, cfg *config.Config) (TelemetryClien return nil, fmt.Errorf("telemetry endpoint is not set: %w", ErrNotConfigured) } - opts := []grpc.DialOption{ - grpc.WithBlock(), - grpc.WithInsecure(), - grpc.WithReturnConnectionError(), - grpc.WithDefaultCallOptions(grpc.WaitForReady(true)), - grpc.WithTransportCredentials(insecure.NewCredentials()), - } - - // Bound the blocking dial so a set-but-unreachable collector can't hang - // startup forever (the signal context has no deadline). - dialCtx, cancel := context.WithTimeout(ctx, DefaultDialTimeout) - defer cancel() - - conn, err := grpc.DialContext( - dialCtx, + // grpc.NewClient is non-blocking: it never dials during startup, so a + // set-but-unreachable collector can't stall main(), and the connection is + // established (and re-established) lazily in the background — a collector that + // comes up after the app does connects on its own, without a restart. + conn, err := grpc.NewClient( cfg.TelemetryEndpoint, - opts..., + grpc.WithTransportCredentials(insecure.NewCredentials()), + grpc.WithDefaultCallOptions(grpc.WaitForReady(true)), ) if err != nil { - return nil, fmt.Errorf("failed to dial telemetry endpoint: %w", err) + return nil, fmt.Errorf("failed to create telemetry connection: %w", err) } + // Build all three exporters before creating any provider. Exporter creation + // is the only remaining error path, and at that point no batch/reader + // goroutine has started yet, so closing the connection is a complete teardown. logExporter, err := otlploggrpc.New(ctx, otlploggrpc.WithGRPCConn(conn)) if err != nil { _ = conn.Close() return nil, fmt.Errorf("failed to create log exporter: %w", err) } - - logProcessor := sdklog.NewBatchProcessor( - logExporter, - sdklog.WithExportInterval(DefaultLogInterval), - sdklog.WithExportTimeout(DefaultLogTimeout), - ) - logProvider := sdklog.NewLoggerProvider( - sdklog.WithProcessor(logProcessor), - sdklog.WithResource(newResource()), - ) - metricExporter, err := otlpmetricgrpc.New(ctx, otlpmetricgrpc.WithGRPCConn(conn)) if err != nil { _ = conn.Close() return nil, fmt.Errorf("failed to create metric exporter: %w", err) } - - metricProcessor := sdkmetric.NewPeriodicReader( - metricExporter, - sdkmetric.WithInterval(DefaultMetricInterval), - sdkmetric.WithTimeout(DefaultMetricTimeout), - ) - - meterProvider := sdkmetric.NewMeterProvider( - sdkmetric.WithReader(metricProcessor), - sdkmetric.WithResource(newResource()), - ) - spanExporter, err := otlptracegrpc.New(ctx, otlptracegrpc.WithGRPCConn(conn)) if err != nil { _ = conn.Close() return nil, fmt.Errorf("failed to create tracer exporter: %w", err) } - spanProcessor := sdktrace.NewBatchSpanProcessor( - spanExporter, - sdktrace.WithBatchTimeout(DefaultTraceInterval), - sdktrace.WithExportTimeout(DefaultTraceTimeout), + logProvider := sdklog.NewLoggerProvider( + sdklog.WithProcessor(sdklog.NewBatchProcessor( + logExporter, + sdklog.WithExportInterval(DefaultLogInterval), + sdklog.WithExportTimeout(DefaultLogTimeout), + )), + sdklog.WithResource(newResource()), ) + + meterProvider := sdkmetric.NewMeterProvider( + sdkmetric.WithReader(sdkmetric.NewPeriodicReader( + metricExporter, + sdkmetric.WithInterval(DefaultMetricInterval), + sdkmetric.WithTimeout(DefaultMetricTimeout), + )), + sdkmetric.WithResource(newResource()), + ) + tracerProvider := sdktrace.NewTracerProvider( - sdktrace.WithSpanProcessor(spanProcessor), + sdktrace.WithSpanProcessor(sdktrace.NewBatchSpanProcessor( + spanExporter, + sdktrace.WithBatchTimeout(DefaultTraceInterval), + sdktrace.WithExportTimeout(DefaultTraceTimeout), + )), sdktrace.WithResource(newResource()), ) diff --git a/backend/pkg/observability/otelclient_test.go b/backend/pkg/observability/otelclient_test.go index 35c613b0..f2f1dcd6 100644 --- a/backend/pkg/observability/otelclient_test.go +++ b/backend/pkg/observability/otelclient_test.go @@ -80,36 +80,33 @@ func TestNewTelemetryClient_SuccessPathExportsAndShutsDown(t *testing.T) { // accepts the TCP connection but never completes the gRPC handshake, so a // WithBlock dial would wait forever without the internal DefaultDialTimeout that // this bounds — the caller's context has no deadline. -func TestNewTelemetryClient_UnreachableReturnsWithinDialTimeout(t *testing.T) { - ln, err := net.Listen("tcp", "127.0.0.1:0") - if err != nil { - t.Fatalf("listen: %v", err) +// A set-but-unreachable collector must not stall startup: grpc.NewClient is +// non-blocking, so the client is returned immediately and connects in the +// background if the collector later comes up. +func TestNewTelemetryClient_UnreachableDoesNotBlockStartup(t *testing.T) { + type result struct { + client TelemetryClient + err error } - defer ln.Close() + cfg := &config.Config{TelemetryEndpoint: "127.0.0.1:1"} // nothing listening + done := make(chan result, 1) go func() { - for { - conn, err := ln.Accept() - if err != nil { - return - } - // hold the connection open and stay silent (no HTTP/2 handshake) - defer conn.Close() - } - }() - - cfg := &config.Config{TelemetryEndpoint: ln.Addr().String()} - done := make(chan error, 1) - go func() { - _, e := NewTelemetryClient(context.Background(), cfg) - done <- e + c, e := NewTelemetryClient(context.Background(), cfg) + done <- result{c, e} }() select { - case err := <-done: - if err == nil { - t.Fatal("expected an error for an unreachable collector") + case res := <-done: + if res.err != nil { + t.Fatalf("non-blocking client must not error on an unreachable collector: %v", res.err) } - case <-time.After(DefaultDialTimeout + 10*time.Second): - t.Fatal("NewTelemetryClient hung past the dial timeout") + if res.client == nil { + t.Fatal("expected a non-nil client") + } + shutdownCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + _ = res.client.Shutdown(shutdownCtx) + case <-time.After(3 * time.Second): + t.Fatal("NewTelemetryClient blocked on an unreachable collector") } }