diff --git a/router/core/graph_server.go b/router/core/graph_server.go index af206997be..46ed7915ff 100644 --- a/router/core/graph_server.go +++ b/router/core/graph_server.go @@ -151,7 +151,7 @@ type reusedGraphMux struct { } // newGraphServer creates a new server instance. -func newGraphServer(routerCtx context.Context, r *Router, response *routerconfig.Response, proxy ProxyFunc) (*graphServer, error) { +func newGraphServer(routerCtx context.Context, r *Router, response *routerconfig.Response, proxy ProxyFunc) (_ *graphServer, resErr error) { /* Older versions of composition will not populate a compatibility version. * Currently, all "old" router execution configurations are compatible as there have been no breaking * changes. @@ -231,6 +231,16 @@ func newGraphServer(routerCtx context.Context, r *Router, response *routerconfig headerPropagation: r.headerPropagation, } + defer func() { + if resErr == nil { + return + } + // Shutdown the graph server to clean up resources + if err := s.Shutdown(routerCtx); err != nil { + s.logger.Error("Failed to shut down graph server during failed-build cleanup", zap.Error(err)) + } + }() + baseOtelAttributes := []attribute.KeyValue{ otel.WgRouterVersion.String(Version), otel.WgRouterClusterName.String(r.clusterName), @@ -678,8 +688,9 @@ type graphMux struct { ctx context.Context cancel context.CancelFunc - mux *chi.Mux - reused atomic.Bool + mux *chi.Mux + reused atomic.Bool + finalized atomic.Bool planCache *ristretto.Cache[uint64, *planWithMetaData] planFallbackCache *slowplancache.Cache[*planWithMetaData] @@ -958,6 +969,11 @@ func (s *graphMux) configureCacheMetrics(srv *graphServer, baseOtelAttributes [] } func (s *graphMux) Shutdown(ctx context.Context) error { + // Make sure we do not shutdown the mux multiple times + if !s.finalized.CompareAndSwap(false, true) { + return nil + } + // cancel the graph muxes context to close its resources like websocket connections, resolvers, etc. s.cancel() @@ -997,12 +1013,6 @@ func (s *graphMux) Shutdown(ctx context.Context) error { } } - if s.streamMetricStore != nil { - if aErr := s.streamMetricStore.Shutdown(ctx); aErr != nil { - err = errors.Join(err, aErr) - } - } - if s.prometheusMetricsExporter != nil { if aErr := s.prometheusMetricsExporter.Shutdown(ctx); aErr != nil { err = errors.Join(err, aErr) @@ -1021,7 +1031,7 @@ func (s *graphMux) Shutdown(ctx context.Context) error { // The mux is appended internally to the graph server's list of muxes to clean up later when the server is swapped. func (s *graphServer) buildGraphMux( opts BuildGraphMuxOptions, -) (*graphMux, error) { +) (_ *graphMux, resErr error) { graphMuxCtx, graphMuxCancel := context.WithCancel(s.routerCtx) gm := &graphMux{ @@ -1031,6 +1041,21 @@ func (s *graphServer) buildGraphMux( streamMetricStore: rmetric.NewNoopStreamMetricStore(), } + // A failed mux isn't in s.graphMuxList yet (added on success below), so the graph + // server's Shutdown won't see it. Clean it up here to avoid leaking its callbacks, + // context and caches. + defer func() { + if resErr == nil { + return + } + + cleanupCtx, cancel := context.WithTimeout(s.routerCtx, time.Second*30) + defer cancel() + if cErr := gm.Shutdown(cleanupCtx); cErr != nil { + s.logger.Error("Failed to clean up partially-built graph mux after build error", zap.Error(cErr)) + } + }() + httpRouter := chi.NewRouter() // we only enable the attribute mapper if we are not using the default cloud exporter @@ -2100,6 +2125,38 @@ func (s *graphServer) wait(ctx context.Context) error { } } +// metricsFlushTimeout bounds the single, central flush of the shared meter +// providers during graph server shutdown. +const metricsFlushTimeout = 30 * time.Second + +// flushMeterProviders flushes the OTLP and Prometheus meter providers once. These +// providers are shared by every metric store (request, connection, stream, +// engine, runtime), so a single flush drains all of their metrics. +func (s *graphServer) flushMeterProviders(ctx context.Context) error { + wg := &sync.WaitGroup{} + + var otlpErr error + var promErr error + + if s.otlpMeterProvider != nil { + wg.Go(func() { + if err := s.otlpMeterProvider.ForceFlush(ctx); err != nil { + otlpErr = errors.Join(otlpErr, fmt.Errorf("failed to flush otlp metrics: %w", err)) + } + }) + } + if s.promMeterProvider != nil { + wg.Go(func() { + if err := s.promMeterProvider.ForceFlush(ctx); err != nil { + promErr = errors.Join(promErr, fmt.Errorf("failed to flush prometheus metrics: %w", err)) + } + }) + } + wg.Wait() + + return errors.Join(otlpErr, promErr) +} + // Shutdown gracefully shutdown the server and waits for all in-flight requests to finish. // After all requests are done, it will shut down the metric store and runtime metrics. // Shutdown does cancel the context after all non-hijacked requests such as WebSockets has been handled. @@ -2125,6 +2182,16 @@ func (s *graphServer) Shutdown(ctx context.Context) error { zap.String("config_version", s.baseRouterConfigVersion), ) + // Flush the meter providers exactly once, with their own timeout, + // before tearing down the individual metric stores. + // As all the stores share the same meter providers, we only need to flush once + // before initiating the shutdown of the individual stores. + flushCtx, flushCancel := context.WithTimeout(ctx, metricsFlushTimeout) + if err := s.flushMeterProviders(flushCtx); err != nil { + finalErr = errors.Join(finalErr, fmt.Errorf("failed to flush metrics: %w", err)) + } + flushCancel() + // Ensure that we don't wait indefinitely for shutdown if s.routerGracePeriod > 0 { newCtx, cancel := context.WithTimeout(ctx, s.routerGracePeriod) diff --git a/router/core/graphql_prehandler.go b/router/core/graphql_prehandler.go index 5f2652136c..d60f48c090 100644 --- a/router/core/graphql_prehandler.go +++ b/router/core/graphql_prehandler.go @@ -1244,21 +1244,18 @@ func (h *PreHandler) flushMetrics(ctx context.Context, requestLogger *zap.Logger now := time.Now() wg := &sync.WaitGroup{} - wg.Add(1) - go func() { - defer wg.Done() + + wg.Go(func() { if err := h.metrics.MetricStore().Flush(ctx); err != nil { requestLogger.Error("Failed to flush OTEL metrics", zap.Error(err)) } - }() + }) - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { if err := h.tracerProvider.ForceFlush(ctx); err != nil { requestLogger.Error("Failed to flush OTEL tracer", zap.Error(err)) } - }() + }) wg.Wait() diff --git a/router/internal/exporter/exporter.go b/router/internal/exporter/exporter.go index ee4dbda827..c1e8071b25 100644 --- a/router/internal/exporter/exporter.go +++ b/router/internal/exporter/exporter.go @@ -9,6 +9,7 @@ import ( "github.com/cloudflare/backoff" "go.uber.org/atomic" "go.uber.org/zap" + "golang.org/x/sync/semaphore" ) // Exporter is a generic, thread-safe batch exporter that queues items and sends them @@ -23,7 +24,8 @@ type Exporter[T any] struct { acceptTrafficSema chan struct{} queue chan T inflightBatches *atomic.Int64 - batchBufferPool *sync.Pool // Pool for batch slice buffers to reduce allocations + batchBufferPool *sync.Pool // Pool for batch slice buffers to reduce allocations + exportSema *semaphore.Weighted // Bounds the number of concurrent in-flight export goroutines // exportRequestContext is used to cancel all requests that started before the shutdown exportRequestContext context.Context @@ -47,6 +49,7 @@ const ( defaultMaxBatchItems = 1024 defaultMaxQueueSize = 1024 * 10 defaultBatchInterval = time.Duration(10) * time.Second + defaultMaxConcurrentExports = 100 ) type ExporterSettings struct { @@ -60,14 +63,19 @@ type ExporterSettings struct { RetryOptions RetryOptions // ExportTimeout is the timeout for the export request. ExportTimeout time.Duration + // MaxConcurrentExports caps the number of in-flight export goroutines. + // When <= 0, defaultMaxConcurrentExports is used. Bounding this prevents a slow + // or failing sink from accumulating unbounded retrying batches in memory. + MaxConcurrentExports int } func NewDefaultExporterSettings() *ExporterSettings { return &ExporterSettings{ - BatchSize: defaultMaxBatchItems, - QueueSize: defaultMaxQueueSize, - Interval: defaultBatchInterval, - ExportTimeout: defaultExportTimeout, + BatchSize: defaultMaxBatchItems, + QueueSize: defaultMaxQueueSize, + Interval: defaultBatchInterval, + ExportTimeout: defaultExportTimeout, + MaxConcurrentExports: defaultMaxConcurrentExports, RetryOptions: RetryOptions{ Enabled: true, MaxRetry: defaultExportMaxRetryAttempts, @@ -93,6 +101,13 @@ func NewExporter[T any](logger *zap.Logger, sink Sink[T], isRetryableError SinkE isRetryableError = func(err error) bool { return true } } + // Bound concurrent exports; fall back to the default when unset so callers + // that construct ExporterSettings directly don't get an unbounded pipeline. + maxConcurrentExports := settings.MaxConcurrentExports + if maxConcurrentExports <= 0 { + maxConcurrentExports = defaultMaxConcurrentExports + } + e := &Exporter[T]{ logger: logger.With(zap.String("component", "exporter")), settings: settings, @@ -102,6 +117,7 @@ func NewExporter[T any](logger *zap.Logger, sink Sink[T], isRetryableError SinkE shutdownSignal: make(chan struct{}), acceptTrafficSema: make(chan struct{}), inflightBatches: atomic.NewInt64(0), + exportSema: semaphore.NewWeighted(int64(maxConcurrentExports)), exportRequestContext: ctx, cancelAllExportRequests: cancel, batchBufferPool: &sync.Pool{ @@ -161,9 +177,18 @@ func (e *Exporter[T]) getBatchBuffer() []T { } // putBatchBuffer returns a batch buffer to the pool for reuse. -// The buffer is reset to zero length before being pooled. +// Element references are cleared and the slice is reset to zero length before pooling. func (e *Exporter[T]) putBatchBuffer(buffer []T) { - // Reset the slice to zero length while keeping capacity + // Drop buffers that grew beyond the configured batch size instead of pooling + // them, so the pool doesn't retain oversized backing arrays indefinitely. + if cap(buffer) > e.settings.BatchSize { + return + } + // Clear element references before pooling. Reslicing to [:0] only changes the + // length: the backing array still holds the (already-exported) item pointers, + // and the GC scans the full capacity, so without this the pooled buffer would + // pin every item from the last batch until those slots are overwritten. + clear(buffer) buffer = buffer[:0] e.batchBufferPool.Put(&buffer) } @@ -232,8 +257,20 @@ func (e *Exporter[T]) exportBatch(batch []T) error { func (e *Exporter[T]) prepareAndSendBatch(batch []T) { e.logger.Debug("Preparing to send batch", zap.Int("batch_size", len(batch))) e.inflightBatches.Inc() + // Acquire a slot before spawning the goroutine. When all slots are taken this + // blocks the caller (the start/drain loop), applying backpressure all the way + // up to Record's queue instead of accumulating retrying batches in memory. + // The context is only cancelled on forced shutdown, in which case exports would + // fail anyway: return the buffer to the pool and drop the batch. + if err := e.exportSema.Acquire(e.exportRequestContext, 1); err != nil { + e.logger.Debug("Skipping batch export, exporter is shutting down", zap.Error(err)) + e.putBatchBuffer(batch) + e.inflightBatches.Dec() + return + } go func() { defer e.inflightBatches.Dec() + defer e.exportSema.Release(1) // Release the slot once the export (and any retries) completes defer e.putBatchBuffer(batch) // Return buffer to pool after export completes e.exportBatchWithRetry(batch) }() diff --git a/router/internal/exporter/exporter_test.go b/router/internal/exporter/exporter_test.go new file mode 100644 index 0000000000..6b6836c8f4 --- /dev/null +++ b/router/internal/exporter/exporter_test.go @@ -0,0 +1,207 @@ +package exporter + +import ( + "context" + "runtime" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" + graphqlmetrics "github.com/wundergraph/cosmo/router/gen/proto/wg/cosmo/graphqlmetrics/v1" + "go.uber.org/zap" +) + +// newIdleExporter creates an exporter that won't flush on its own (long interval, +// no records), so tests can exercise the buffer pool deterministically. +func newIdleExporter(t *testing.T, batchSize int) *Exporter[*graphqlmetrics.SchemaUsageInfo] { + t.Helper() + e, err := NewExporter(zap.NewNop(), &mockSink{}, nil, &ExporterSettings{ + BatchSize: batchSize, + QueueSize: 16, + Interval: time.Hour, + ExportTimeout: time.Second, + MaxConcurrentExports: 2, + RetryOptions: RetryOptions{ + Enabled: false, + MaxRetry: 1, + MaxDuration: time.Second, + Interval: time.Second, + }, + }) + require.NoError(t, err) + t.Cleanup(func() { _ = e.Shutdown(context.Background()) }) + return e +} + +// TestPutBatchBufferClearsElements verifies a buffer returned to the pool comes +// back with no lingering element references (the clear-on-return fix). A dirty +// slot here would mean an already-exported item is still pinned by the pool. +func TestPutBatchBufferClearsElements(t *testing.T) { + e := newIdleExporter(t, 4) + + buf := e.getBatchBuffer() + buf = append(buf, &graphqlmetrics.SchemaUsageInfo{}, &graphqlmetrics.SchemaUsageInfo{}, &graphqlmetrics.SchemaUsageInfo{}) + e.putBatchBuffer(buf) + + full := buf[:cap(buf)] + for i := range full { + require.Nilf(t, full[i], "pooled buffer not cleared at slot %d", i) + } +} + +// TestPutBatchBufferDropsOversized verifies buffers that grew past BatchSize are +// discarded instead of pooled, so the pool can't retain oversized backing arrays. +func TestPutBatchBufferDropsOversized(t *testing.T) { + e := newIdleExporter(t, 4) + + oversized := make([]*graphqlmetrics.SchemaUsageInfo, 0, e.settings.BatchSize*4) + oversized = append(oversized, &graphqlmetrics.SchemaUsageInfo{}) + e.putBatchBuffer(oversized) + + got := e.getBatchBuffer() + require.LessOrEqualf(t, cap(got), e.settings.BatchSize, "oversized buffer was pooled: cap=%d, want <= %d", cap(got), e.settings.BatchSize) +} + +// blockingSink blocks every Export until release is closed, tracking how many +// exports run concurrently so a test can assert the inflight cap is respected. +type blockingSink struct { + active atomic.Int64 + maxSeen atomic.Int64 + release chan struct{} +} + +func (s *blockingSink) Export(_ context.Context, _ []*graphqlmetrics.SchemaUsageInfo) error { + cur := s.active.Add(1) + for { + m := s.maxSeen.Load() + if cur <= m || s.maxSeen.CompareAndSwap(m, cur) { + break + } + } + <-s.release + s.active.Add(-1) + return nil +} + +func (s *blockingSink) Close(_ context.Context) error { return nil } + +// TestConcurrentExportsRespectCap verifies that no more than MaxConcurrentExports +// export goroutines run at once, and that the cap is actually reached (parallelism +// isn't accidentally serialized). +func TestConcurrentExportsRespectCap(t *testing.T) { + const cap = 3 + sink := &blockingSink{release: make(chan struct{})} + + e, err := NewExporter(zap.NewNop(), sink, nil, &ExporterSettings{ + BatchSize: 1, // one item per batch, so each record can spawn an export + QueueSize: 100, + Interval: time.Hour, + ExportTimeout: time.Minute, + MaxConcurrentExports: cap, + RetryOptions: RetryOptions{ + Enabled: false, + MaxRetry: 1, + MaxDuration: time.Second, + Interval: time.Second, + }, + }) + require.NoError(t, err) + // Always free blocked export goroutines and shut down, even if an assertion + // below fails. Release first to unblock the goroutines Shutdown waits on. + t.Cleanup(func() { + close(sink.release) + _ = e.Shutdown(context.Background()) + }) + + for range 10 { + e.Record(&graphqlmetrics.SchemaUsageInfo{}, false) + } + + require.Eventuallyf(t, func() bool { return sink.active.Load() >= cap }, 3*time.Second, 10*time.Millisecond, + "cap not saturated: only %d concurrent exports", sink.active.Load()) + // Give the drain loop a chance to (wrongly) start a 4th export if the cap + // weren't enforced; backpressure should keep it blocked on Acquire. + time.Sleep(100 * time.Millisecond) + require.Equal(t, int64(cap), sink.active.Load(), "active exports") + require.Equal(t, int64(cap), sink.maxSeen.Load(), "max observed concurrency") +} + +// TestExporterDoesNotRetainExportedItems is the end-to-end guarantee behind the +// clear fix: once a batch is exported and its goroutine finishes, the items must +// be collectable. If the pooled buffer (or the drain loop's reused buffer) still +// pinned them, the finalizers would never run and this would time out. +func TestExporterDoesNotRetainExportedItems(t *testing.T) { + const n = 8 + sink := &mockSink{} + + e, err := NewExporter(zap.NewNop(), sink, nil, &ExporterSettings{ + BatchSize: n, // fill exactly one batch so it flushes immediately + QueueSize: 1024, + Interval: time.Hour, + ExportTimeout: time.Second, + MaxConcurrentExports: 4, + RetryOptions: RetryOptions{ + Enabled: false, + MaxRetry: 1, + MaxDuration: time.Second, + Interval: time.Second, + }, + }) + require.NoError(t, err) + t.Cleanup(func() { _ = e.Shutdown(context.Background()) }) + + var finalized atomic.Int64 + items := make([]*graphqlmetrics.SchemaUsageInfo, n) + for i := range items { + it := &graphqlmetrics.SchemaUsageInfo{} + runtime.SetFinalizer(it, func(*graphqlmetrics.SchemaUsageInfo) { finalized.Add(1) }) + items[i] = it + require.Truef(t, e.Record(it, false), "record %d unexpectedly dropped", i) + } + + // Wait until the batch is exported and the export goroutine has finished, so + // the buffer has been returned to the pool (and cleared, if the fix is in). + require.Eventually(t, func() bool { + return sink.exportCount.Load() >= 1 && e.inflightBatches.Load() == 0 + }, 5*time.Second, 10*time.Millisecond, "batch was not exported in time") + + // Drop our references; nothing in the exporter should keep the items alive. + for i := range items { + items[i] = nil + } + items = nil + + require.Eventuallyf(t, func() bool { + runtime.GC() + return finalized.Load() == int64(n) + }, 5*time.Second, 10*time.Millisecond, "exported items were retained: want %d finalized", n) +} + +// TestNewExporterDefaultsMaxConcurrentExports verifies that leaving the cap unset +// (<= 0) falls back to the default rather than producing a zero-capacity semaphore +// that would deadlock every export. +func TestNewExporterDefaultsMaxConcurrentExports(t *testing.T) { + sink := &mockSink{} + e, err := NewExporter(zap.NewNop(), sink, nil, &ExporterSettings{ + BatchSize: 4, + QueueSize: 64, + Interval: time.Hour, + ExportTimeout: time.Second, + // MaxConcurrentExports intentionally left 0. + RetryOptions: RetryOptions{ + Enabled: false, + MaxRetry: 1, + MaxDuration: time.Second, + Interval: time.Second, + }, + }) + require.NoError(t, err) + t.Cleanup(func() { _ = e.Shutdown(context.Background()) }) + + for range 12 { + e.Record(&graphqlmetrics.SchemaUsageInfo{}, false) + } + require.Eventually(t, func() bool { return sink.exportCount.Load() >= 1 }, 3*time.Second, 10*time.Millisecond, + "exporter with defaulted cap did not export (possible deadlock)") +} diff --git a/router/pkg/metric/cache_metrics.go b/router/pkg/metric/cache_metrics.go index 1d238aea5e..501c713e26 100644 --- a/router/pkg/metric/cache_metrics.go +++ b/router/pkg/metric/cache_metrics.go @@ -206,7 +206,7 @@ func (c *CacheMetrics) Shutdown() error { for _, reg := range c.instrumentRegistrations { if regErr := reg.Unregister(); regErr != nil { - err = errors.Join(regErr) + err = errors.Join(err, regErr) } } diff --git a/router/pkg/metric/connection_metric_store.go b/router/pkg/metric/connection_metric_store.go index 2ed2ba9b15..13ad40bbd8 100644 --- a/router/pkg/metric/connection_metric_store.go +++ b/router/pkg/metric/connection_metric_store.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "github.com/wundergraph/cosmo/router/pkg/otel" "go.opentelemetry.io/otel/attribute" @@ -17,14 +18,12 @@ import ( type ConnectionMetricProvider interface { MeasureConnectionAcquireDuration(ctx context.Context, duration float64, opts ...otelmetric.RecordOption) MeasureMaxConnections(ctx context.Context, count int64, opts ...otelmetric.RecordOption) - Flush(ctx context.Context) error Shutdown() error } // ConnectionMetricStore is the interface for connection and pool metrics only. type ConnectionMetricStore interface { MeasureConnectionAcquireDuration(ctx context.Context, duration float64, attrs ...attribute.KeyValue) - Flush(ctx context.Context) error Shutdown(ctx context.Context) error } @@ -90,31 +89,12 @@ func (c *ConnectionMetrics) MeasureConnectionAcquireDuration(ctx context.Context c.promConnectionMetrics.MeasureConnectionAcquireDuration(ctx, duration, opts) } -// Flush flushes the metrics to the backend synchronously. -func (h *ConnectionMetrics) Flush(ctx context.Context) error { - var err error - - errOtlp := h.otlpConnectionMetrics.Flush(ctx) - if errOtlp != nil { - err = errors.Join(err, fmt.Errorf("failed to flush otlp metrics: %w", errOtlp)) - } - - errProm := h.promConnectionMetrics.Flush(ctx) - if errProm != nil { - err = errors.Join(err, fmt.Errorf("failed to flush prometheus metrics: %w", errProm)) - } - - return err -} - -// Shutdown flushes the metrics and stops the runtime metrics. -func (h *ConnectionMetrics) Shutdown(ctx context.Context) error { +// Shutdown stops the metric instruments. It does not flush: the shared meter +// providers are flushed once centrally during graph server shutdown to avoid +// redundant ForceFlush calls that all compete for a single shutdown deadline. +func (h *ConnectionMetrics) Shutdown(_ context.Context) error { var err error - if errFlush := h.Flush(ctx); errFlush != nil { - err = errors.Join(err, fmt.Errorf("failed to flush metrics: %w", errFlush)) - } - if errProm := h.promConnectionMetrics.Shutdown(); errProm != nil { err = errors.Join(err, fmt.Errorf("failed to shutdown prom metrics: %w", errProm)) } diff --git a/router/pkg/metric/engine_metrics.go b/router/pkg/metric/engine_metrics.go index 9f0c5775de..0e8270f158 100644 --- a/router/pkg/metric/engine_metrics.go +++ b/router/pkg/metric/engine_metrics.go @@ -174,7 +174,7 @@ func (e *EngineMetrics) Shutdown() error { for _, reg := range e.instrumentRegistrations { if regErr := reg.Unregister(); regErr != nil { - err = errors.Join(regErr) + err = errors.Join(err, regErr) } } diff --git a/router/pkg/metric/metric_store.go b/router/pkg/metric/metric_store.go index 96707fda1d..46ce236116 100644 --- a/router/pkg/metric/metric_store.go +++ b/router/pkg/metric/metric_store.go @@ -479,21 +479,16 @@ func (h *Metrics) Flush(ctx context.Context) error { return err } -// Shutdown flushes the metrics and stops the runtime metrics. -func (h *Metrics) Shutdown(ctx context.Context) error { - +// Shutdown stops the metric instruments. It does not flush: the shared meter +// providers are flushed once centrally during graph server shutdown to avoid +// redundant ForceFlush calls that all compete for a single shutdown deadline. +func (h *Metrics) Shutdown(_ context.Context) error { var err error - if errFlush := h.Flush(ctx); errFlush != nil { - err = errors.Join(err, fmt.Errorf("failed to flush metrics: %w", errFlush)) - } - - errProm := h.promRequestMetrics.Shutdown() - if err != nil { + if errProm := h.promRequestMetrics.Shutdown(); errProm != nil { err = errors.Join(err, fmt.Errorf("failed to shutdown prom metrics: %w", errProm)) } - errOtlp := h.otlpRequestMetrics.Shutdown() - if err != nil { + if errOtlp := h.otlpRequestMetrics.Shutdown(); errOtlp != nil { err = errors.Join(err, fmt.Errorf("failed to shutdown otlp metrics: %w", errOtlp)) } diff --git a/router/pkg/metric/oltp_connection_metric_store.go b/router/pkg/metric/oltp_connection_metric_store.go index e92f43b72b..8518aca730 100644 --- a/router/pkg/metric/oltp_connection_metric_store.go +++ b/router/pkg/metric/oltp_connection_metric_store.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "github.com/wundergraph/cosmo/router/pkg/otel" "go.opentelemetry.io/otel/attribute" @@ -92,16 +93,12 @@ func (h *otlpConnectionMetrics) MeasureMaxConnections(ctx context.Context, count h.instruments.maxConnections.Record(ctx, count, opts...) } -func (h *otlpConnectionMetrics) Flush(ctx context.Context) error { - return h.meterProvider.ForceFlush(ctx) -} - func (h *otlpConnectionMetrics) Shutdown() error { var err error for _, reg := range h.instrumentRegistrations { if regErr := reg.Unregister(); regErr != nil { - err = errors.Join(regErr) + err = errors.Join(err, regErr) } } diff --git a/router/pkg/metric/oltp_stream_metric_store.go b/router/pkg/metric/oltp_stream_metric_store.go index 3d7a8573e9..8d30c15364 100644 --- a/router/pkg/metric/oltp_stream_metric_store.go +++ b/router/pkg/metric/oltp_stream_metric_store.go @@ -46,7 +46,3 @@ func (o *otlpStreamEventMetrics) Produce(ctx context.Context, opts ...otelmetric func (o *otlpStreamEventMetrics) Consume(ctx context.Context, opts ...otelmetric.AddOption) { o.instruments.consumedMessages.Add(ctx, 1, opts...) } - -func (o *otlpStreamEventMetrics) Flush(ctx context.Context) error { - return o.meterProvider.ForceFlush(ctx) -} diff --git a/router/pkg/metric/otlp_metric_store.go b/router/pkg/metric/otlp_metric_store.go index f784e8b309..022e56a73b 100644 --- a/router/pkg/metric/otlp_metric_store.go +++ b/router/pkg/metric/otlp_metric_store.go @@ -165,7 +165,7 @@ func (h *OtlpMetricStore) Shutdown() error { for _, reg := range h.instrumentRegistrations { if regErr := reg.Unregister(); regErr != nil { - err = errors.Join(regErr) + err = errors.Join(err, regErr) } } diff --git a/router/pkg/metric/prom_connection_metric_store.go b/router/pkg/metric/prom_connection_metric_store.go index a4a58b14fa..51b311ea94 100644 --- a/router/pkg/metric/prom_connection_metric_store.go +++ b/router/pkg/metric/prom_connection_metric_store.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "github.com/wundergraph/cosmo/router/pkg/otel" "go.opentelemetry.io/otel/attribute" @@ -91,16 +92,12 @@ func (m *promConnectionMetrics) MeasureMaxConnections(ctx context.Context, count m.instruments.maxConnections.Record(ctx, count, opts...) } -func (m *promConnectionMetrics) Flush(ctx context.Context) error { - return m.meterProvider.ForceFlush(ctx) -} - func (h *promConnectionMetrics) Shutdown() error { var err error for _, reg := range h.instrumentRegistrations { if regErr := reg.Unregister(); regErr != nil { - err = errors.Join(regErr) + err = errors.Join(err, regErr) } } diff --git a/router/pkg/metric/prom_metric_store.go b/router/pkg/metric/prom_metric_store.go index 6596fd0ccc..fb6e06fb15 100644 --- a/router/pkg/metric/prom_metric_store.go +++ b/router/pkg/metric/prom_metric_store.go @@ -172,7 +172,7 @@ func (h *PromMetricStore) Shutdown() error { for _, reg := range h.instrumentRegistrations { if regErr := reg.Unregister(); regErr != nil { - err = errors.Join(regErr) + err = errors.Join(err, regErr) } } diff --git a/router/pkg/metric/router_runtime_metrics.go b/router/pkg/metric/router_runtime_metrics.go index 1a29fee931..103149dac3 100644 --- a/router/pkg/metric/router_runtime_metrics.go +++ b/router/pkg/metric/router_runtime_metrics.go @@ -309,7 +309,7 @@ func (r *RuntimeMetrics) Shutdown() error { for _, reg := range r.instrumentRegistrations { if regErr := reg.Unregister(); regErr != nil { - err = errors.Join(regErr) + err = errors.Join(err, regErr) } } diff --git a/router/pkg/metric/stream_metric_store.go b/router/pkg/metric/stream_metric_store.go index 2034d2bc70..361f49388d 100644 --- a/router/pkg/metric/stream_metric_store.go +++ b/router/pkg/metric/stream_metric_store.go @@ -2,7 +2,6 @@ package metric import ( "context" - "errors" "fmt" "go.opentelemetry.io/otel/attribute" @@ -34,16 +33,11 @@ type StreamsEvent struct { type StreamMetricProvider interface { Produce(ctx context.Context, opts ...otelmetric.AddOption) Consume(ctx context.Context, opts ...otelmetric.AddOption) - - Flush(ctx context.Context) error } type StreamMetricStore interface { Produce(ctx context.Context, event StreamsEvent) Consume(ctx context.Context, event StreamsEvent) - - Flush(ctx context.Context) error - Shutdown(ctx context.Context) error } // StreamMetrics is the store for Event (Kafka/Redis/NATS) metrics. @@ -127,27 +121,3 @@ func (e *StreamMetrics) Consume(ctx context.Context, event StreamsEvent) { provider.Consume(ctx, opt) } } - -// Flush flushes the metrics to the backend synchronously. -func (e *StreamMetrics) Flush(ctx context.Context) error { - var err error - - for _, provider := range e.providers { - if errOtlp := provider.Flush(ctx); errOtlp != nil { - err = errors.Join(err, fmt.Errorf("failed to flush metrics: %w", errOtlp)) - } - } - - return err -} - -// Shutdown flushes the metrics and stops observers if any. -func (e *StreamMetrics) Shutdown(ctx context.Context) error { - var err error - - if errFlush := e.Flush(ctx); errFlush != nil { - err = errors.Join(err, fmt.Errorf("failed to flush metrics: %w", errFlush)) - } - - return err -}