laskoviymishka commented on code in PR #1655: URL: https://github.com/apache/iceberg-go/pull/1655#discussion_r3740299798
########## catalog/rest/metrics_reporter.go: ########## @@ -0,0 +1,333 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package rest + +import ( + "context" + "log/slog" + "net/http" + "net/url" + "strconv" + "sync" + "sync/atomic" + "time" + + iceberg "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/metrics" +) + +const ( + // keyReportMetricsEnabled opts a REST catalog into POSTing scan/commit + // reports to the catalog's metrics endpoint. It is disabled by default so + // existing users see no new network traffic unless they turn it on. This is + // the canonical, cross-implementation spelling used by Iceberg Java and the + // Iceberg docs; keyReportMetricsEnabledLegacy is accepted as an alias. + keyReportMetricsEnabled = "rest-metrics-reporting-enabled" + // keyReportMetricsEnabledLegacy is the historical dotted spelling, accepted + // as an alias so existing configs keep working. + keyReportMetricsEnabledLegacy = "rest.metrics-reporting-enabled" + // keyReportMetricsTimeoutMs bounds a single report's request and response + // cycle, in milliseconds. Read from the client-supplied properties only. Auth + // (token refresh) is bounded separately by the OAuth refresh client Timeout. + keyReportMetricsTimeoutMs = "rest-metrics-reporting-timeout-ms" + + // defaultReportMetricsTimeout bounds a single report's request and response + // cycle when keyReportMetricsTimeoutMs is unset. Telemetry is safe to drop, so + // the bound is deliberately short. + defaultReportMetricsTimeout = 10 * time.Second + // metricsDispatchWorkers is the fixed number of goroutines draining the + // dispatch queue, capping the reporter's concurrent connection use. + metricsDispatchWorkers = 4 + // metricsDispatchQueueSize bounds how many reports may await dispatch. + // Reports offered while the queue is full are dropped (and counted, then + // logged in aggregate) rather than queued without limit, so a stalled + // endpoint cannot make reporting grow without bound. + metricsDispatchQueueSize = 128 + // metricsStatsInterval is how often the dispatcher emits an aggregated + // warning summarizing dropped and failed reports. Aggregating keeps a stalled + // or unavailable endpoint from amplifying into one log line per report while + // still surfacing back-pressure. + metricsStatsInterval = 30 * time.Second +) + +// reportMetricsEnabled reports whether the client opted into REST metrics +// reporting, accepting both the canonical and the legacy dotted key. It must be +// given the client-supplied properties only (never the server-merged config) so +// a server cannot flip the default and turn on outbound telemetry the client +// never asked for. +// +// The canonical key wins whenever it is present, so a client that has migrated +// to it and explicitly set it to false is honored even if a stale legacy dotted +// key lingers in the config. The legacy key is consulted only when the canonical +// key is absent. +func reportMetricsEnabled(props iceberg.Properties) bool { + if v, ok := props[keyReportMetricsEnabled]; ok { + enabled, err := strconv.ParseBool(v) + + return err == nil && enabled + } + + return props.GetBool(keyReportMetricsEnabledLegacy, false) +} + +// reportMetricsTimeout resolves the per-report deadline from the client-supplied +// properties, falling back to defaultReportMetricsTimeout for a missing or +// non-positive value. +func reportMetricsTimeout(props iceberg.Properties) time.Duration { + ms := props.GetInt(keyReportMetricsTimeoutMs, int(defaultReportMetricsTimeout/time.Millisecond)) + if ms <= 0 { + return defaultReportMetricsTimeout + } + + return time.Duration(ms) * time.Millisecond +} + +// metricsJob is a single report awaiting dispatch to a table's metrics endpoint. +type metricsJob struct { + // ctx carries the observed scan/commit's context values (trace spans, + // request-scoped attributes) with its cancellation detached — the scan/commit + // is already done, so its cancellation must not abort the report, but its + // values should still propagate to the outbound request. + ctx context.Context + baseURI *url.URL + cl *http.Client + path []string + req metrics.ReportMetricsRequest +} + +// metricsDispatcher POSTs metrics reports to REST metrics endpoints on a fixed +// pool of workers draining a bounded queue. It is owned by the catalog and +// shared across that catalog's table reporters, so concurrent report volume +// stays bounded no matter how many tables are loaded or how often they are +// scanned. A stalled endpoint sheds load — reports are dropped and counted — +// rather than accumulating goroutines and connections. Close cancels in-flight +// reports and drains the workers. +type metricsDispatcher struct { + jobs chan metricsJob + timeout time.Duration + ctx context.Context Review Comment: Storing `ctx`/`cancel` on the struct will draw a lint eye since the Go docs steer away from contexts in struct fields. I think it's the right call here — the dispatcher genuinely owns a lifecycle context that Close cancels — but I'd add a one-line comment saying exactly that, so the next reader doesn't try to "fix" it. ########## catalog/rest/metrics_reporter.go: ########## @@ -0,0 +1,333 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package rest + +import ( + "context" + "log/slog" + "net/http" + "net/url" + "strconv" + "sync" + "sync/atomic" + "time" + + iceberg "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/metrics" +) + +const ( + // keyReportMetricsEnabled opts a REST catalog into POSTing scan/commit + // reports to the catalog's metrics endpoint. It is disabled by default so + // existing users see no new network traffic unless they turn it on. This is + // the canonical, cross-implementation spelling used by Iceberg Java and the + // Iceberg docs; keyReportMetricsEnabledLegacy is accepted as an alias. + keyReportMetricsEnabled = "rest-metrics-reporting-enabled" + // keyReportMetricsEnabledLegacy is the historical dotted spelling, accepted + // as an alias so existing configs keep working. + keyReportMetricsEnabledLegacy = "rest.metrics-reporting-enabled" + // keyReportMetricsTimeoutMs bounds a single report's request and response + // cycle, in milliseconds. Read from the client-supplied properties only. Auth + // (token refresh) is bounded separately by the OAuth refresh client Timeout. + keyReportMetricsTimeoutMs = "rest-metrics-reporting-timeout-ms" + + // defaultReportMetricsTimeout bounds a single report's request and response + // cycle when keyReportMetricsTimeoutMs is unset. Telemetry is safe to drop, so + // the bound is deliberately short. + defaultReportMetricsTimeout = 10 * time.Second + // metricsDispatchWorkers is the fixed number of goroutines draining the + // dispatch queue, capping the reporter's concurrent connection use. + metricsDispatchWorkers = 4 + // metricsDispatchQueueSize bounds how many reports may await dispatch. + // Reports offered while the queue is full are dropped (and counted, then + // logged in aggregate) rather than queued without limit, so a stalled + // endpoint cannot make reporting grow without bound. + metricsDispatchQueueSize = 128 + // metricsStatsInterval is how often the dispatcher emits an aggregated + // warning summarizing dropped and failed reports. Aggregating keeps a stalled + // or unavailable endpoint from amplifying into one log line per report while + // still surfacing back-pressure. + metricsStatsInterval = 30 * time.Second +) + +// reportMetricsEnabled reports whether the client opted into REST metrics +// reporting, accepting both the canonical and the legacy dotted key. It must be +// given the client-supplied properties only (never the server-merged config) so +// a server cannot flip the default and turn on outbound telemetry the client +// never asked for. +// +// The canonical key wins whenever it is present, so a client that has migrated +// to it and explicitly set it to false is honored even if a stale legacy dotted +// key lingers in the config. The legacy key is consulted only when the canonical +// key is absent. +func reportMetricsEnabled(props iceberg.Properties) bool { Review Comment: I'm on board with defaulting this off — the no-surprise-outbound-traffic instinct is the right one, and the comment above makes the intent clear. The thing I'd flag is that Java defaults `METRICS_REPORTING_ENABLED_DEFAULT` to true, so anyone porting a Java client that never set the flag explicitly will silently stop getting metrics here. I'd call that out in CHANGES.md or the catalog property docs so the divergence reads as a documented choice rather than a surprise. wdyt? ########## catalog/rest/rest.go: ########## @@ -1143,6 +1183,21 @@ func (r *Catalog) tableFromResponse(_ context.Context, identifier []string, meta if err != nil { return nil, fmt.Errorf("failed to initialize metrics reporter: %w", err) } + // Opt-in: POST scan/commit reports to the catalog's metrics endpoint via the + // catalog-owned dispatcher. It is non-nil only when the client enabled + // reporting and the server advertises the endpoint (see init). + if r.metricsDispatcher != nil { + if ns, tbl, idErr := r.splitIdentForPath(identifier); idErr == nil { Review Comment: Both `splitIdentForPath` and `reqPath` errors get swallowed here — if either fails, the REST reporter just isn't composed in, with no log. It's unreachable for a valid table today, but if it ever regresses, metrics silently vanishing with no signal is a rough thing to debug. I'd drop a debug-level `slog` on the two error branches, matching the warn-on-drop pattern the dispatcher already uses. Even a comment on why these can't fail would help the next reader. ########## catalog/rest/rest.go: ########## @@ -929,6 +955,16 @@ func (r *Catalog) init(ctx context.Context, ops *options, uri string) error { } r.props = toProps(ops) + // Stand up the metrics dispatcher only when the client opted in and the + // server advertises the endpoint. Enablement is read from the + // client-supplied properties alone (see reporterProps): a server-vended + // default, override, or table-response property must never turn on outbound + // telemetry the client never asked for. + if reportMetricsEnabled(r.reporterProps) && r.endpoints.check(endpointReportMetrics) == nil { Review Comment: Reading enablement from `r.reporterProps` is exactly right for the direction that matters — a server must not be able to turn *on* telemetry the client never asked for. No argument there. The asymmetry I'd think about is the other direction: because this only ever looks at client props, a server that sets `rest-metrics-reporting-enabled=false` in its overrides can't turn reporting *off* for a Go client the way it can for a Java one (Java resolves this from the merged config). Server-side disable is always safe — it never creates surprise outbound traffic — so honoring an explicit `=false` from overrides seems like the friendlier behavior for an operator trying to suppress all clients at once. Either honor a server override that flips it off, or leave it client-only and document that server-side disable is intentionally ignored. I'd lean toward the former, but either's defensible as long as it's a deliberate call. wdyt? ########## catalog/rest/metrics_reporter_test.go: ########## @@ -0,0 +1,844 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package rest + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "log/slog" + "maps" + "net/http" + "net/http/httptest" + "net/url" + "sync" + "sync/atomic" + "testing" + "time" + + iceberg "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/metrics" + "github.com/apache/iceberg-go/table" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type capturedRequest struct { + method string + path string // decoded path + escapedPath string // percent-encoded path + header http.Header + body []byte +} + +// captureTransport records the request it receives and returns 204 No Content, +// avoiding any real network listener. +type captureTransport struct { + ch chan capturedRequest + block <-chan struct{} // if non-nil, RoundTrip waits on it before responding + err error // if non-nil, returned instead of a response +} + +func (c *captureTransport) RoundTrip(r *http.Request) (*http.Response, error) { + body, _ := io.ReadAll(r.Body) + if c.ch != nil { + c.ch <- capturedRequest{ + method: r.Method, + path: r.URL.Path, + escapedPath: r.URL.EscapedPath(), + header: r.Header.Clone(), // the SDK may reuse the request; snapshot it + body: body, + } + } + if c.block != nil { + <-c.block + } + if c.err != nil { + return nil, c.err + } + + return &http.Response{ + StatusCode: http.StatusNoContent, + Body: io.NopCloser(bytes.NewReader(nil)), + Header: make(http.Header), + }, nil +} + +// ctxBlockTransport blocks until the request context is done, then reports the +// context error. It lets a test prove that a report has a finite deadline +// (timeout) and that Close cancels an in-flight request. entered counts how many +// requests reached the transport. +type ctxBlockTransport struct { + entered atomic.Int32 + started chan struct{} // signalled once when RoundTrip is entered + ctxErr chan error // receives the context error once it fires +} + +func (c *ctxBlockTransport) RoundTrip(r *http.Request) (*http.Response, error) { + c.entered.Add(1) + select { + case c.started <- struct{}{}: + default: + } + <-r.Context().Done() + err := r.Context().Err() + select { + case c.ctxErr <- err: + default: + } + + return nil, err +} + +// concurrencyTransport blocks every RoundTrip on release, recording how many run +// concurrently and how many were delivered in total. It lets a test prove the +// dispatcher bounds concurrency and drops excess reports. +type concurrencyTransport struct { + inFlight atomic.Int32 + maxSeen atomic.Int32 + delivered atomic.Int32 + release chan struct{} +} + +func (c *concurrencyTransport) RoundTrip(r *http.Request) (*http.Response, error) { + n := c.inFlight.Add(1) + for { + m := c.maxSeen.Load() + if n <= m || c.maxSeen.CompareAndSwap(m, n) { + break + } + } + c.delivered.Add(1) + <-c.release + c.inFlight.Add(-1) + + return &http.Response{ + StatusCode: http.StatusNoContent, + Body: io.NopCloser(bytes.NewReader(nil)), + Header: make(http.Header), + }, nil +} + +// newTestDispatcher builds a dispatcher with the production pool sizing and a +// discarding logger, registering Close as cleanup. +func newTestDispatcher(t *testing.T, timeout time.Duration) *metricsDispatcher { + t.Helper() + d := newMetricsDispatcher(metricsDispatchWorkers, metricsDispatchQueueSize, timeout, + slog.New(slog.NewTextHandler(io.Discard, nil))) + t.Cleanup(d.close) + + return d +} + +func newTestReporter(t *testing.T, tr http.RoundTripper) *restMetricsReporter { + t.Helper() + + return reporterWith(t, tr, newTestDispatcher(t, 5*time.Second), nil) +} + +// reporterWith builds a reporter bound to the given transport and dispatcher. A +// nil path uses the default single-level namespace path. +func reporterWith(t *testing.T, tr http.RoundTripper, d *metricsDispatcher, path []string) *restMetricsReporter { + t.Helper() + base, err := url.Parse("http://catalog.invalid";) + require.NoError(t, err) + if path == nil { + path = []string{"namespaces", "db", "tables", "t", "metrics"} + } + + return &restMetricsReporter{ + baseURI: base, + cl: &http.Client{Transport: tr}, + path: path, + dispatcher: d, + } +} + +func TestRESTMetricsReporterPostsScanReport(t *testing.T) { + received := make(chan capturedRequest, 1) + rep := newTestReporter(t, &captureTransport{ch: received}) + + rep.Report(context.Background(), metrics.ScanReport{TableName: "db.t", SnapshotID: 99}) + + select { + case req := <-received: + assert.Equal(t, http.MethodPost, req.method) + assert.Equal(t, "/namespaces/db/tables/t/metrics", req.path) + var m map[string]any + require.NoError(t, json.Unmarshal(req.body, &m)) + assert.Equal(t, "scan-report", m["report-type"]) + assert.Equal(t, "db.t", m["table-name"]) + assert.Contains(t, m, "metrics", "report fields are flattened alongside report-type") + case <-time.After(3 * time.Second): + t.Fatal("timed out waiting for the async metrics POST") + } +} + +func TestRESTMetricsReporterPostsCommitReport(t *testing.T) { + received := make(chan capturedRequest, 1) + rep := newTestReporter(t, &captureTransport{ch: received}) + + rep.Report(context.Background(), metrics.CommitReport{TableName: "db.t", Operation: "append"}) + + select { + case req := <-received: + var m map[string]any + require.NoError(t, json.Unmarshal(req.body, &m)) + assert.Equal(t, "commit-report", m["report-type"]) + assert.Equal(t, "append", m["operation"]) + case <-time.After(3 * time.Second): + t.Fatal("timed out waiting for the async metrics POST") + } +} + +func TestRESTMetricsReporterNilReportIsNoop(t *testing.T) { + received := make(chan capturedRequest, 1) + rep := newTestReporter(t, &captureTransport{ch: received}) + + rep.Report(context.Background(), nil) + + select { + case <-received: + t.Fatal("nil report must not produce a POST") + case <-time.After(200 * time.Millisecond): + // expected: nothing sent + } +} + +func TestRESTMetricsReporterReportDoesNotBlockOnSlowServer(t *testing.T) { + block := make(chan struct{}) + defer close(block) + rep := newTestReporter(t, &captureTransport{block: block}) + + done := make(chan struct{}) + go func() { + rep.Report(context.Background(), metrics.ScanReport{TableName: "db.t"}) + close(done) + }() + + select { + case <-done: + // Report returned promptly despite the hanging transport. + case <-time.After(time.Second): + t.Fatal("Report blocked on a slow server") + } +} + +// TestRESTMetricsReporterEscapesFullPath proves the request goes to the full +// /v1/{prefix}/... URL and that namespace and table segments needing escaping +// are percent-encoded. +func TestRESTMetricsReporterEscapesFullPath(t *testing.T) { + received := make(chan capturedRequest, 1) + base, err := url.Parse("http://catalog.invalid/v1/my-prefix";) + require.NoError(t, err) + rep := &restMetricsReporter{ + baseURI: base, + cl: &http.Client{Transport: &captureTransport{ch: received}}, + path: []string{"namespaces", "a b", "tables", "t x", "metrics"}, + dispatcher: newTestDispatcher(t, 5*time.Second), + } + + rep.Report(context.Background(), metrics.ScanReport{TableName: "a b.t x"}) + + select { + case req := <-received: + assert.Equal(t, "/v1/my-prefix/namespaces/a b/tables/t x/metrics", req.path) + assert.Equal(t, "/v1/my-prefix/namespaces/a%20b/tables/t%20x/metrics", req.escapedPath) + case <-time.After(3 * time.Second): + t.Fatal("timed out waiting for the async metrics POST") + } +} + +// rotatingAuthManager returns a distinct bearer header on each call, standing in +// for a token source that refreshes between reports. +type rotatingAuthManager struct{ n atomic.Int32 } + +func (m *rotatingAuthManager) AuthHeader() (string, string, error) { + return "Authorization", fmt.Sprintf("Bearer tok-%d", m.n.Add(1)), nil +} + +// ctxAuthManager records whether the context-aware path was taken and returns a +// header value identifying which method produced it. +type ctxAuthManager struct{ usedContext atomic.Bool } + +func (m *ctxAuthManager) AuthHeader() (string, string, error) { + return "Authorization", "context-free", nil +} + +func (m *ctxAuthManager) AuthHeaderWithContext(ctx context.Context) (string, string, error) { + if err := ctx.Err(); err != nil { + return "", "", err + } + m.usedContext.Store(true) + + return "Authorization", "with-context", nil +} + +// blockingHandler blocks in Handle until released, so a test can prove the drop +// path never logs synchronously on the caller's goroutine. +type blockingHandler struct{ release <-chan struct{} } + +func (h *blockingHandler) Enabled(context.Context, slog.Level) bool { return true } +func (h *blockingHandler) Handle(context.Context, slog.Record) error { + <-h.release + + return nil +} +func (h *blockingHandler) WithAttrs([]slog.Attr) slog.Handler { return h } +func (h *blockingHandler) WithGroup(string) slog.Handler { return h } + +// recordingHandler captures emitted records so a test can assert the aggregated +// warning content. +type recordingHandler struct { + mu sync.Mutex + records []slog.Record +} + +func (h *recordingHandler) Enabled(context.Context, slog.Level) bool { return true } +func (h *recordingHandler) Handle(_ context.Context, r slog.Record) error { + h.mu.Lock() + defer h.mu.Unlock() + h.records = append(h.records, r.Clone()) + + return nil +} +func (h *recordingHandler) WithAttrs([]slog.Attr) slog.Handler { return h } +func (h *recordingHandler) WithGroup(string) slog.Handler { return h } + +func (h *recordingHandler) warnings() []slog.Record { + h.mu.Lock() + defer h.mu.Unlock() + var out []slog.Record + for _, r := range h.records { + if r.Level == slog.LevelWarn { + out = append(out, r) + } + } + + return out +} + +// reporterWithSession builds a reporter whose client runs through a +// sessionTransport (auth + default headers) wrapping the given transport, so a +// test can assert the metrics POST goes out authenticated. +func reporterWithSession(t *testing.T, auth AuthManager, tr http.RoundTripper) *restMetricsReporter { + t.Helper() + session := &sessionTransport{ + RoundTripper: tr, + authManager: auth, + defaultHeaders: http.Header{}, + } + session.defaultHeaders.Set("Content-Type", "application/json") + + return reporterWith(t, session, newTestDispatcher(t, 5*time.Second), nil) +} + +// TestRESTMetricsReporterSendsAuthenticatedRequest pins that the metrics POST +// reuses the catalog's authenticated client: the Authorization header the auth +// manager produces and the catalog's default Content-Type both go out on the +// wire. Without this a refactor that dispatched metrics through a bare +// http.Client would keep every other test green. +func TestRESTMetricsReporterSendsAuthenticatedRequest(t *testing.T) { + received := make(chan capturedRequest, 1) + rep := reporterWithSession(t, staticAuthManager{key: "Authorization", value: "Bearer tok-abc"}, &captureTransport{ch: received}) + + rep.Report(context.Background(), metrics.ScanReport{TableName: "db.t"}) + + select { + case req := <-received: + assert.Equal(t, "Bearer tok-abc", req.header.Get("Authorization"), + "metrics POST must carry the catalog's Authorization header") + assert.Equal(t, "application/json", req.header.Get("Content-Type"), + "catalog default headers must be forwarded") + case <-time.After(3 * time.Second): + t.Fatal("timed out waiting for the async metrics POST") + } +} + +// TestRESTMetricsReporterRefetchesCredentialPerReport proves auth runs per +// request, so a credential that rotates between reports is picked up rather than +// captured once. +func TestRESTMetricsReporterRefetchesCredentialPerReport(t *testing.T) { + received := make(chan capturedRequest, 2) + // One worker serializes the two reports so credentials are handed out in order. + d := newMetricsDispatcher(1, 8, 5*time.Second, slog.New(slog.NewTextHandler(io.Discard, nil))) + t.Cleanup(d.close) + session := &sessionTransport{ + RoundTripper: &captureTransport{ch: received}, + authManager: &rotatingAuthManager{}, + defaultHeaders: http.Header{}, + } + rep := reporterWith(t, session, d, nil) + + rep.Report(context.Background(), metrics.ScanReport{TableName: "db.t"}) + rep.Report(context.Background(), metrics.ScanReport{TableName: "db.t"}) + + seen := make(map[string]bool) + for range 2 { + select { + case req := <-received: + seen[req.header.Get("Authorization")] = true + case <-time.After(3 * time.Second): + t.Fatal("timed out waiting for a metrics POST") + } + } + assert.True(t, seen["Bearer tok-1"] && seen["Bearer tok-2"], + "each report must fetch a fresh credential, got %v", seen) +} + +// TestSessionTransportPrefersContextAuthManager proves sessionTransport uses the +// context-aware auth path when the manager implements ContextAuthManager, so a +// request deadline can bound the auth step. +func TestSessionTransportPrefersContextAuthManager(t *testing.T) { + received := make(chan capturedRequest, 1) + auth := &ctxAuthManager{} + rep := reporterWithSession(t, auth, &captureTransport{ch: received}) + + rep.Report(context.Background(), metrics.ScanReport{TableName: "db.t"}) + + select { + case req := <-received: + assert.Equal(t, "with-context", req.header.Get("Authorization"), + "the context-aware auth path must be preferred") + assert.True(t, auth.usedContext.Load()) + case <-time.After(3 * time.Second): + t.Fatal("timed out waiting for the async metrics POST") + } +} + +// TestRESTMetricsReporterPreservesContextValues proves the caller's context +// values (trace spans, request-scoped attributes) reach the outbound report even +// though the caller's cancellation is detached. +func TestRESTMetricsReporterPreservesContextValues(t *testing.T) { + type ctxKey struct{} + got := make(chan any, 1) + tr := roundTripFunc(func(r *http.Request) (*http.Response, error) { + got <- r.Context().Value(ctxKey{}) + + return &http.Response{StatusCode: http.StatusNoContent, Body: io.NopCloser(bytes.NewReader(nil)), Header: make(http.Header)}, nil + }) + rep := newTestReporter(t, tr) + + ctx := context.WithValue(context.Background(), ctxKey{}, "trace-123") + rep.Report(ctx, metrics.ScanReport{TableName: "db.t"}) + + select { + case v := <-got: + assert.Equal(t, "trace-123", v, "caller context values must propagate to the report") + case <-time.After(3 * time.Second): + t.Fatal("timed out waiting for the async metrics POST") + } +} + +// TestMetricsDispatcherDropCountsWithoutBlocking proves that dropping a report on +// a full queue neither blocks the caller nor logs synchronously: the logger here +// blocks in Handle, yet every Report returns promptly and the drops are counted. +func TestMetricsDispatcherDropCountsWithoutBlocking(t *testing.T) { + release := make(chan struct{}) + handlerRelease := make(chan struct{}) + tr := &concurrencyTransport{release: release} + d := newMetricsDispatcher(1, 1, time.Minute, slog.New(&blockingHandler{release: handlerRelease})) + t.Cleanup(func() { + close(handlerRelease) + close(release) + d.close() + }) + rep := reporterWith(t, tr, d, nil) + + // Occupy the single worker; the next report fills the queue, the rest drop. + rep.Report(context.Background(), metrics.ScanReport{TableName: "db.t"}) + require.Eventually(t, func() bool { + return tr.inFlight.Load() == 1 + }, 3*time.Second, 5*time.Millisecond, "worker never saturated") + + for range 4 { // 1 fills the queue, 3 are dropped + done := make(chan struct{}) + go func() { + rep.Report(context.Background(), metrics.ScanReport{TableName: "db.t"}) + close(done) + }() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("Report blocked; the drop path must not log synchronously") + } + } + + require.Eventually(t, func() bool { + return d.dropped.Load() == 3 + }, 3*time.Second, 5*time.Millisecond, "expected exactly the surplus reports to be dropped") +} + +// TestMetricsDispatcherAggregatesDropWarning proves the drop/failure counters +// surface as a single aggregated warning (carrying the counts) rather than one +// log line per dropped report. The shutdown flush is used to make the assertion +// deterministic without waiting on the stats ticker. +func TestMetricsDispatcherAggregatesDropWarning(t *testing.T) { + handler := &recordingHandler{} + release := make(chan struct{}) + tr := &concurrencyTransport{release: release} + d := newMetricsDispatcher(1, 1, time.Minute, slog.New(handler)) + rep := reporterWith(t, tr, d, nil) + + // Occupy the worker, fill the queue, then overflow: three reports drop. + rep.Report(context.Background(), metrics.ScanReport{TableName: "db.t"}) + require.Eventually(t, func() bool { + return tr.inFlight.Load() == 1 + }, 3*time.Second, 5*time.Millisecond, "worker never saturated") + for range 4 { + rep.Report(context.Background(), metrics.ScanReport{TableName: "db.t"}) + } + require.Eventually(t, func() bool { + return d.dropped.Load() == 3 + }, 3*time.Second, 5*time.Millisecond) + + close(release) + d.close() // triggers the final aggregated flush + + warnings := handler.warnings() + require.Len(t, warnings, 1, "drops must aggregate into a single warning, not one per drop") + attrs := map[string]any{} + warnings[0].Attrs(func(a slog.Attr) bool { + attrs[a.Key] = a.Value.Any() + + return true + }) + assert.Equal(t, uint64(3), attrs["dropped"], "the aggregated warning must carry the drop count") +} + +// TestMetricsDispatcherCloseIsIdempotent proves repeated and concurrent Close +// calls are safe and cheap: they do not panic and each returns, spawning no +// extra waiter goroutine per call. +func TestMetricsDispatcherCloseIsIdempotent(t *testing.T) { + d := newMetricsDispatcher(2, 4, 5*time.Second, slog.New(slog.NewTextHandler(io.Discard, nil))) + + var wg sync.WaitGroup + for range 5 { + wg.Add(1) + go func() { + defer wg.Done() + d.close() + }() + } + + done := make(chan struct{}) + go func() { + wg.Wait() + close(done) + }() + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("concurrent Close calls did not all return") + } + + // A further Close after the fact is still safe. + d.close() +} + +// TestRESTMetricsReporterBuildsPathThroughProduction runs the real path +// composition — splitIdentForPath (encodeNamespace) and endpointReportMetrics +// .reqPath — for a multi-level namespace with segments needing escaping, so a +// regression in that composition or in the separator handling is caught rather +// than bypassed by a hand-built path. +func TestRESTMetricsReporterBuildsPathThroughProduction(t *testing.T) { + c := &Catalog{namespaceSeparator: defaultNamespaceSeparator} + ns, tbl, err := c.splitIdentForPath(table.Identifier{"a b", "d e", "t x"}) + require.NoError(t, err) + path, err := endpointReportMetrics.reqPath(ns, tbl) + require.NoError(t, err) + + received := make(chan capturedRequest, 1) + base, err := url.Parse("http://catalog.invalid/v1/my-prefix";) + require.NoError(t, err) + rep := reporterWith(t, &captureTransport{ch: received}, newTestDispatcher(t, 5*time.Second), path) + rep.baseURI = base + + rep.Report(context.Background(), metrics.ScanReport{TableName: "a b.d e.t x"}) + + select { + case req := <-received: + // The namespace levels are percent-encoded and joined by the encoded + // separator (%1F); the table segment is escaped by the URL builder. + assert.Equal(t, "/v1/my-prefix/namespaces/a%20b%1Fd%20e/tables/t%20x/metrics", req.escapedPath) + case <-time.After(3 * time.Second): + t.Fatal("timed out waiting for the async metrics POST") + } +} + +// TestMetricsDispatcherReportHasFiniteTimeout proves a report does not hang +// forever against a black-holing endpoint: the per-report deadline fires and the +// request context reports DeadlineExceeded. +func TestMetricsDispatcherReportHasFiniteTimeout(t *testing.T) { + tr := &ctxBlockTransport{started: make(chan struct{}, 1), ctxErr: make(chan error, 1)} + d := newMetricsDispatcher(1, 1, 100*time.Millisecond, slog.New(slog.NewTextHandler(io.Discard, nil))) + t.Cleanup(d.close) + rep := reporterWith(t, tr, d, nil) + + rep.Report(context.Background(), metrics.ScanReport{TableName: "db.t"}) + + select { + case err := <-tr.ctxErr: + assert.ErrorIs(t, err, context.DeadlineExceeded, "report must time out rather than hang") + case <-time.After(3 * time.Second): + t.Fatal("report did not observe a finite timeout") + } +} + +// TestMetricsDispatcherBoundsConcurrencyAndDrops proves the worker pool caps +// concurrent sends and sheds excess reports rather than queueing them without +// limit. +func TestMetricsDispatcherBoundsConcurrencyAndDrops(t *testing.T) { + const ( + workers = 2 + queue = 3 + extra = 3 // reports offered once the pool and queue are saturated + ) + tr := &concurrencyTransport{release: make(chan struct{})} + d := newMetricsDispatcher(workers, queue, 5*time.Second, slog.New(slog.NewTextHandler(io.Discard, nil))) + t.Cleanup(d.close) + rep := reporterWith(t, tr, d, nil) + + // Fill every worker; each blocks in RoundTrip, leaving the queue empty. + for range workers { + rep.Report(context.Background(), metrics.ScanReport{TableName: "db.t"}) + } + require.Eventually(t, func() bool { + return tr.inFlight.Load() == workers + }, 3*time.Second, 5*time.Millisecond, "workers never saturated") + + // Fill the queue (workers are blocked, so nothing drains it), then offer more + // than fits — the surplus must be dropped, not queued. + for range queue + extra { + rep.Report(context.Background(), metrics.ScanReport{TableName: "db.t"}) + } + + // Concurrency stayed within the pool while everything was blocked. + assert.LessOrEqual(t, tr.maxSeen.Load(), int32(workers), "concurrency exceeded the worker pool") + + close(tr.release) + + // Only workers + queue reports are ever delivered; the extra were dropped. + require.Eventually(t, func() bool { + return tr.inFlight.Load() == 0 && tr.delivered.Load() == workers+queue + }, 3*time.Second, 5*time.Millisecond, "expected exactly workers+queue reports delivered") + + assert.LessOrEqual(t, tr.maxSeen.Load(), int32(workers), "concurrency exceeded the worker pool") +} + +// TestMetricsDispatcherCloseCancelsInFlight proves Close cancels an in-flight +// report and returns promptly rather than waiting on the stalled endpoint. +func TestMetricsDispatcherCloseCancelsInFlight(t *testing.T) { + tr := &ctxBlockTransport{started: make(chan struct{}, 1), ctxErr: make(chan error, 1)} + // A long per-report timeout so it is Close, not the deadline, that unblocks. + d := newMetricsDispatcher(1, 1, time.Minute, slog.New(slog.NewTextHandler(io.Discard, nil))) + rep := reporterWith(t, tr, d, nil) + + rep.Report(context.Background(), metrics.ScanReport{TableName: "db.t"}) + + select { + case <-tr.started: + case <-time.After(3 * time.Second): + t.Fatal("report never reached the transport") + } + + closed := make(chan struct{}) + go func() { + d.close() + close(closed) + }() + + select { + case <-closed: + case <-time.After(3 * time.Second): + t.Fatal("Close did not return; in-flight report was not cancelled") + } + + select { + case err := <-tr.ctxErr: + assert.ErrorIs(t, err, context.Canceled, "Close must cancel in-flight reports") + case <-time.After(time.Second): + t.Fatal("in-flight request context was not cancelled") + } +} + +// TestMetricsDispatcherCloseDropsQueuedReports proves that reports still waiting +// in the queue when Close is called are dropped rather than sent to the endpoint. +func TestMetricsDispatcherCloseDropsQueuedReports(t *testing.T) { + tr := &ctxBlockTransport{started: make(chan struct{}, 1), ctxErr: make(chan error, 8)} + // One worker, roomy queue: the worker is occupied by the first report while + // the rest pile up behind it. + d := newMetricsDispatcher(1, 8, time.Minute, slog.New(slog.NewTextHandler(io.Discard, nil))) + rep := reporterWith(t, tr, d, nil) + + rep.Report(context.Background(), metrics.ScanReport{TableName: "db.t"}) + select { + case <-tr.started: + case <-time.After(3 * time.Second): + t.Fatal("first report never reached the transport") + } + + // Queue several more behind the busy worker. + for range 5 { + rep.Report(context.Background(), metrics.ScanReport{TableName: "db.t"}) + } + + d.close() + + // Only the in-flight report ever reached the transport; the queued reports + // were discarded on shutdown rather than sent. + assert.Equal(t, int32(1), tr.entered.Load(), + "queued reports must be dropped on Close, not sent") +} + +func TestReportMetricsEnabled(t *testing.T) { + assert.False(t, reportMetricsEnabled(iceberg.Properties{})) + assert.True(t, reportMetricsEnabled(iceberg.Properties{keyReportMetricsEnabled: "true"})) + assert.True(t, reportMetricsEnabled(iceberg.Properties{keyReportMetricsEnabledLegacy: "true"}), + "the historical dotted key is accepted as an alias") + assert.False(t, reportMetricsEnabled(iceberg.Properties{keyReportMetricsEnabled: "false"})) + + // The canonical key wins whenever it is present: a client that migrated to it + // and explicitly disabled reporting is honored even if a stale legacy dotted + // key still lingers in the config. + assert.False(t, reportMetricsEnabled(iceberg.Properties{ + keyReportMetricsEnabled: "false", + keyReportMetricsEnabledLegacy: "true", + }), "an explicit canonical false must override a lingering legacy true") + assert.True(t, reportMetricsEnabled(iceberg.Properties{ + keyReportMetricsEnabled: "true", + keyReportMetricsEnabledLegacy: "false", + }), "an explicit canonical true wins over a legacy false") +} + +func TestReportMetricsTimeout(t *testing.T) { + assert.Equal(t, defaultReportMetricsTimeout, reportMetricsTimeout(iceberg.Properties{})) + assert.Equal(t, 250*time.Millisecond, reportMetricsTimeout(iceberg.Properties{keyReportMetricsTimeoutMs: "250"})) + assert.Equal(t, defaultReportMetricsTimeout, reportMetricsTimeout(iceberg.Properties{keyReportMetricsTimeoutMs: "0"}), + "a non-positive timeout falls back to the default") + assert.Equal(t, defaultReportMetricsTimeout, reportMetricsTimeout(iceberg.Properties{keyReportMetricsTimeoutMs: "bogus"})) +} + +// TestMetricsReportingEnablementPrecedence pins that reporting is enabled only +// by client-supplied configuration: server-vended defaults and overrides setting +// the key must not turn it on, and the server must advertise the endpoint. +// Table-response properties likewise cannot enable it, since enablement is +// resolved once at init from the client properties alone. +func TestMetricsReportingEnablementPrecedence(t *testing.T) { + cfg := func(defaults, overrides map[string]any, endpoints []string) map[string]any { + m := map[string]any{ + "defaults": orEmpty(defaults), + "overrides": orEmpty(overrides), + } + if endpoints != nil { + m["endpoints"] = endpoints + } + + return m + } + + tests := []struct { + name string + serverCfg map[string]any + clientProps iceberg.Properties + wantEnabled bool + }{ + { + name: "off by default", + serverCfg: cfg(nil, nil, nil), + }, + { + name: "server default cannot enable", + serverCfg: cfg(map[string]any{keyReportMetricsEnabled: "true"}, nil, nil), + }, + { + name: "server override cannot enable", + serverCfg: cfg(nil, map[string]any{keyReportMetricsEnabled: "true"}, nil), + }, + { + name: "client enables", + serverCfg: cfg(nil, nil, nil), + clientProps: iceberg.Properties{keyReportMetricsEnabled: "true"}, + wantEnabled: true, + }, + { + name: "client legacy key enables", + serverCfg: cfg(nil, nil, nil), + clientProps: iceberg.Properties{keyReportMetricsEnabledLegacy: "true"}, + wantEnabled: true, + }, + { + name: "client enables but endpoint not advertised", Review Comment: This proves the negative — endpoints advertised but not the metrics one — but there's no case advertising a non-empty list that *does* include the metrics endpoint. The "client enables" cases above only hit the fallback-to-`defaultEndpoints` path (server returns no endpoints key), so the production path where `resolveEndpoints` parses an explicitly-advertised metrics endpoint template never runs. I'd add a sibling entry with `endpoints: []string{"POST /v1/{prefix}/namespaces/{namespace}/tables/{table}/metrics"}` and `wantEnabled: true`. That's the case that catches `endpointFromString` silently failing to parse the template and disabling reporting for servers that actually advertise it. ########## catalog/rest/metrics_reporter_test.go: ########## @@ -0,0 +1,844 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package rest + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "log/slog" + "maps" + "net/http" + "net/http/httptest" + "net/url" + "sync" + "sync/atomic" + "testing" + "time" + + iceberg "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/metrics" + "github.com/apache/iceberg-go/table" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type capturedRequest struct { + method string + path string // decoded path + escapedPath string // percent-encoded path + header http.Header + body []byte +} + +// captureTransport records the request it receives and returns 204 No Content, +// avoiding any real network listener. +type captureTransport struct { + ch chan capturedRequest + block <-chan struct{} // if non-nil, RoundTrip waits on it before responding + err error // if non-nil, returned instead of a response +} + +func (c *captureTransport) RoundTrip(r *http.Request) (*http.Response, error) { + body, _ := io.ReadAll(r.Body) + if c.ch != nil { + c.ch <- capturedRequest{ + method: r.Method, + path: r.URL.Path, + escapedPath: r.URL.EscapedPath(), + header: r.Header.Clone(), // the SDK may reuse the request; snapshot it + body: body, + } + } + if c.block != nil { + <-c.block + } + if c.err != nil { + return nil, c.err + } + + return &http.Response{ + StatusCode: http.StatusNoContent, + Body: io.NopCloser(bytes.NewReader(nil)), + Header: make(http.Header), + }, nil +} + +// ctxBlockTransport blocks until the request context is done, then reports the +// context error. It lets a test prove that a report has a finite deadline +// (timeout) and that Close cancels an in-flight request. entered counts how many +// requests reached the transport. +type ctxBlockTransport struct { + entered atomic.Int32 + started chan struct{} // signalled once when RoundTrip is entered + ctxErr chan error // receives the context error once it fires +} + +func (c *ctxBlockTransport) RoundTrip(r *http.Request) (*http.Response, error) { + c.entered.Add(1) + select { + case c.started <- struct{}{}: + default: + } + <-r.Context().Done() + err := r.Context().Err() + select { + case c.ctxErr <- err: + default: + } + + return nil, err +} + +// concurrencyTransport blocks every RoundTrip on release, recording how many run +// concurrently and how many were delivered in total. It lets a test prove the +// dispatcher bounds concurrency and drops excess reports. +type concurrencyTransport struct { + inFlight atomic.Int32 + maxSeen atomic.Int32 + delivered atomic.Int32 + release chan struct{} +} + +func (c *concurrencyTransport) RoundTrip(r *http.Request) (*http.Response, error) { + n := c.inFlight.Add(1) + for { + m := c.maxSeen.Load() + if n <= m || c.maxSeen.CompareAndSwap(m, n) { + break + } + } + c.delivered.Add(1) + <-c.release + c.inFlight.Add(-1) + + return &http.Response{ + StatusCode: http.StatusNoContent, + Body: io.NopCloser(bytes.NewReader(nil)), + Header: make(http.Header), + }, nil +} + +// newTestDispatcher builds a dispatcher with the production pool sizing and a +// discarding logger, registering Close as cleanup. +func newTestDispatcher(t *testing.T, timeout time.Duration) *metricsDispatcher { + t.Helper() + d := newMetricsDispatcher(metricsDispatchWorkers, metricsDispatchQueueSize, timeout, + slog.New(slog.NewTextHandler(io.Discard, nil))) + t.Cleanup(d.close) + + return d +} + +func newTestReporter(t *testing.T, tr http.RoundTripper) *restMetricsReporter { + t.Helper() + + return reporterWith(t, tr, newTestDispatcher(t, 5*time.Second), nil) +} + +// reporterWith builds a reporter bound to the given transport and dispatcher. A +// nil path uses the default single-level namespace path. +func reporterWith(t *testing.T, tr http.RoundTripper, d *metricsDispatcher, path []string) *restMetricsReporter { + t.Helper() + base, err := url.Parse("http://catalog.invalid";) + require.NoError(t, err) + if path == nil { + path = []string{"namespaces", "db", "tables", "t", "metrics"} + } + + return &restMetricsReporter{ + baseURI: base, + cl: &http.Client{Transport: tr}, + path: path, + dispatcher: d, + } +} + +func TestRESTMetricsReporterPostsScanReport(t *testing.T) { + received := make(chan capturedRequest, 1) + rep := newTestReporter(t, &captureTransport{ch: received}) + + rep.Report(context.Background(), metrics.ScanReport{TableName: "db.t", SnapshotID: 99}) + + select { + case req := <-received: + assert.Equal(t, http.MethodPost, req.method) + assert.Equal(t, "/namespaces/db/tables/t/metrics", req.path) + var m map[string]any + require.NoError(t, json.Unmarshal(req.body, &m)) + assert.Equal(t, "scan-report", m["report-type"]) + assert.Equal(t, "db.t", m["table-name"]) + assert.Contains(t, m, "metrics", "report fields are flattened alongside report-type") + case <-time.After(3 * time.Second): + t.Fatal("timed out waiting for the async metrics POST") + } +} + +func TestRESTMetricsReporterPostsCommitReport(t *testing.T) { + received := make(chan capturedRequest, 1) + rep := newTestReporter(t, &captureTransport{ch: received}) + + rep.Report(context.Background(), metrics.CommitReport{TableName: "db.t", Operation: "append"}) + + select { + case req := <-received: + var m map[string]any + require.NoError(t, json.Unmarshal(req.body, &m)) + assert.Equal(t, "commit-report", m["report-type"]) + assert.Equal(t, "append", m["operation"]) + case <-time.After(3 * time.Second): + t.Fatal("timed out waiting for the async metrics POST") + } +} + +func TestRESTMetricsReporterNilReportIsNoop(t *testing.T) { + received := make(chan capturedRequest, 1) + rep := newTestReporter(t, &captureTransport{ch: received}) + + rep.Report(context.Background(), nil) + + select { + case <-received: + t.Fatal("nil report must not produce a POST") + case <-time.After(200 * time.Millisecond): + // expected: nothing sent + } +} + +func TestRESTMetricsReporterReportDoesNotBlockOnSlowServer(t *testing.T) { + block := make(chan struct{}) + defer close(block) + rep := newTestReporter(t, &captureTransport{block: block}) + + done := make(chan struct{}) + go func() { + rep.Report(context.Background(), metrics.ScanReport{TableName: "db.t"}) + close(done) + }() + + select { + case <-done: + // Report returned promptly despite the hanging transport. + case <-time.After(time.Second): + t.Fatal("Report blocked on a slow server") + } +} + +// TestRESTMetricsReporterEscapesFullPath proves the request goes to the full +// /v1/{prefix}/... URL and that namespace and table segments needing escaping +// are percent-encoded. +func TestRESTMetricsReporterEscapesFullPath(t *testing.T) { + received := make(chan capturedRequest, 1) + base, err := url.Parse("http://catalog.invalid/v1/my-prefix";) + require.NoError(t, err) + rep := &restMetricsReporter{ + baseURI: base, + cl: &http.Client{Transport: &captureTransport{ch: received}}, + path: []string{"namespaces", "a b", "tables", "t x", "metrics"}, + dispatcher: newTestDispatcher(t, 5*time.Second), + } + + rep.Report(context.Background(), metrics.ScanReport{TableName: "a b.t x"}) + + select { + case req := <-received: + assert.Equal(t, "/v1/my-prefix/namespaces/a b/tables/t x/metrics", req.path) + assert.Equal(t, "/v1/my-prefix/namespaces/a%20b/tables/t%20x/metrics", req.escapedPath) + case <-time.After(3 * time.Second): + t.Fatal("timed out waiting for the async metrics POST") + } +} + +// rotatingAuthManager returns a distinct bearer header on each call, standing in +// for a token source that refreshes between reports. +type rotatingAuthManager struct{ n atomic.Int32 } + +func (m *rotatingAuthManager) AuthHeader() (string, string, error) { + return "Authorization", fmt.Sprintf("Bearer tok-%d", m.n.Add(1)), nil +} + +// ctxAuthManager records whether the context-aware path was taken and returns a +// header value identifying which method produced it. +type ctxAuthManager struct{ usedContext atomic.Bool } + +func (m *ctxAuthManager) AuthHeader() (string, string, error) { + return "Authorization", "context-free", nil +} + +func (m *ctxAuthManager) AuthHeaderWithContext(ctx context.Context) (string, string, error) { + if err := ctx.Err(); err != nil { + return "", "", err + } + m.usedContext.Store(true) + + return "Authorization", "with-context", nil +} + +// blockingHandler blocks in Handle until released, so a test can prove the drop +// path never logs synchronously on the caller's goroutine. +type blockingHandler struct{ release <-chan struct{} } + +func (h *blockingHandler) Enabled(context.Context, slog.Level) bool { return true } +func (h *blockingHandler) Handle(context.Context, slog.Record) error { + <-h.release + + return nil +} +func (h *blockingHandler) WithAttrs([]slog.Attr) slog.Handler { return h } +func (h *blockingHandler) WithGroup(string) slog.Handler { return h } + +// recordingHandler captures emitted records so a test can assert the aggregated +// warning content. +type recordingHandler struct { + mu sync.Mutex + records []slog.Record +} + +func (h *recordingHandler) Enabled(context.Context, slog.Level) bool { return true } +func (h *recordingHandler) Handle(_ context.Context, r slog.Record) error { + h.mu.Lock() + defer h.mu.Unlock() + h.records = append(h.records, r.Clone()) + + return nil +} +func (h *recordingHandler) WithAttrs([]slog.Attr) slog.Handler { return h } +func (h *recordingHandler) WithGroup(string) slog.Handler { return h } + +func (h *recordingHandler) warnings() []slog.Record { + h.mu.Lock() + defer h.mu.Unlock() + var out []slog.Record + for _, r := range h.records { + if r.Level == slog.LevelWarn { + out = append(out, r) + } + } + + return out +} + +// reporterWithSession builds a reporter whose client runs through a +// sessionTransport (auth + default headers) wrapping the given transport, so a +// test can assert the metrics POST goes out authenticated. +func reporterWithSession(t *testing.T, auth AuthManager, tr http.RoundTripper) *restMetricsReporter { + t.Helper() + session := &sessionTransport{ + RoundTripper: tr, + authManager: auth, + defaultHeaders: http.Header{}, + } + session.defaultHeaders.Set("Content-Type", "application/json") + + return reporterWith(t, session, newTestDispatcher(t, 5*time.Second), nil) +} + +// TestRESTMetricsReporterSendsAuthenticatedRequest pins that the metrics POST +// reuses the catalog's authenticated client: the Authorization header the auth +// manager produces and the catalog's default Content-Type both go out on the +// wire. Without this a refactor that dispatched metrics through a bare +// http.Client would keep every other test green. +func TestRESTMetricsReporterSendsAuthenticatedRequest(t *testing.T) { + received := make(chan capturedRequest, 1) + rep := reporterWithSession(t, staticAuthManager{key: "Authorization", value: "Bearer tok-abc"}, &captureTransport{ch: received}) + + rep.Report(context.Background(), metrics.ScanReport{TableName: "db.t"}) + + select { + case req := <-received: + assert.Equal(t, "Bearer tok-abc", req.header.Get("Authorization"), + "metrics POST must carry the catalog's Authorization header") + assert.Equal(t, "application/json", req.header.Get("Content-Type"), + "catalog default headers must be forwarded") + case <-time.After(3 * time.Second): + t.Fatal("timed out waiting for the async metrics POST") + } +} + +// TestRESTMetricsReporterRefetchesCredentialPerReport proves auth runs per +// request, so a credential that rotates between reports is picked up rather than +// captured once. +func TestRESTMetricsReporterRefetchesCredentialPerReport(t *testing.T) { + received := make(chan capturedRequest, 2) + // One worker serializes the two reports so credentials are handed out in order. + d := newMetricsDispatcher(1, 8, 5*time.Second, slog.New(slog.NewTextHandler(io.Discard, nil))) + t.Cleanup(d.close) + session := &sessionTransport{ + RoundTripper: &captureTransport{ch: received}, + authManager: &rotatingAuthManager{}, + defaultHeaders: http.Header{}, + } + rep := reporterWith(t, session, d, nil) + + rep.Report(context.Background(), metrics.ScanReport{TableName: "db.t"}) + rep.Report(context.Background(), metrics.ScanReport{TableName: "db.t"}) + + seen := make(map[string]bool) + for range 2 { + select { + case req := <-received: + seen[req.header.Get("Authorization")] = true + case <-time.After(3 * time.Second): + t.Fatal("timed out waiting for a metrics POST") + } + } + assert.True(t, seen["Bearer tok-1"] && seen["Bearer tok-2"], + "each report must fetch a fresh credential, got %v", seen) +} + +// TestSessionTransportPrefersContextAuthManager proves sessionTransport uses the +// context-aware auth path when the manager implements ContextAuthManager, so a +// request deadline can bound the auth step. +func TestSessionTransportPrefersContextAuthManager(t *testing.T) { + received := make(chan capturedRequest, 1) + auth := &ctxAuthManager{} + rep := reporterWithSession(t, auth, &captureTransport{ch: received}) + + rep.Report(context.Background(), metrics.ScanReport{TableName: "db.t"}) + + select { + case req := <-received: + assert.Equal(t, "with-context", req.header.Get("Authorization"), + "the context-aware auth path must be preferred") + assert.True(t, auth.usedContext.Load()) + case <-time.After(3 * time.Second): + t.Fatal("timed out waiting for the async metrics POST") + } +} + +// TestRESTMetricsReporterPreservesContextValues proves the caller's context +// values (trace spans, request-scoped attributes) reach the outbound report even +// though the caller's cancellation is detached. +func TestRESTMetricsReporterPreservesContextValues(t *testing.T) { + type ctxKey struct{} + got := make(chan any, 1) + tr := roundTripFunc(func(r *http.Request) (*http.Response, error) { + got <- r.Context().Value(ctxKey{}) + + return &http.Response{StatusCode: http.StatusNoContent, Body: io.NopCloser(bytes.NewReader(nil)), Header: make(http.Header)}, nil + }) + rep := newTestReporter(t, tr) + + ctx := context.WithValue(context.Background(), ctxKey{}, "trace-123") + rep.Report(ctx, metrics.ScanReport{TableName: "db.t"}) + + select { + case v := <-got: + assert.Equal(t, "trace-123", v, "caller context values must propagate to the report") + case <-time.After(3 * time.Second): + t.Fatal("timed out waiting for the async metrics POST") + } +} + +// TestMetricsDispatcherDropCountsWithoutBlocking proves that dropping a report on +// a full queue neither blocks the caller nor logs synchronously: the logger here +// blocks in Handle, yet every Report returns promptly and the drops are counted. +func TestMetricsDispatcherDropCountsWithoutBlocking(t *testing.T) { + release := make(chan struct{}) + handlerRelease := make(chan struct{}) + tr := &concurrencyTransport{release: release} + d := newMetricsDispatcher(1, 1, time.Minute, slog.New(&blockingHandler{release: handlerRelease})) + t.Cleanup(func() { + close(handlerRelease) + close(release) + d.close() + }) + rep := reporterWith(t, tr, d, nil) + + // Occupy the single worker; the next report fills the queue, the rest drop. + rep.Report(context.Background(), metrics.ScanReport{TableName: "db.t"}) + require.Eventually(t, func() bool { + return tr.inFlight.Load() == 1 + }, 3*time.Second, 5*time.Millisecond, "worker never saturated") + + for range 4 { // 1 fills the queue, 3 are dropped + done := make(chan struct{}) + go func() { + rep.Report(context.Background(), metrics.ScanReport{TableName: "db.t"}) + close(done) + }() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("Report blocked; the drop path must not log synchronously") + } + } + + require.Eventually(t, func() bool { + return d.dropped.Load() == 3 + }, 3*time.Second, 5*time.Millisecond, "expected exactly the surplus reports to be dropped") +} + +// TestMetricsDispatcherAggregatesDropWarning proves the drop/failure counters +// surface as a single aggregated warning (carrying the counts) rather than one +// log line per dropped report. The shutdown flush is used to make the assertion +// deterministic without waiting on the stats ticker. +func TestMetricsDispatcherAggregatesDropWarning(t *testing.T) { + handler := &recordingHandler{} + release := make(chan struct{}) + tr := &concurrencyTransport{release: release} + d := newMetricsDispatcher(1, 1, time.Minute, slog.New(handler)) + rep := reporterWith(t, tr, d, nil) + + // Occupy the worker, fill the queue, then overflow: three reports drop. + rep.Report(context.Background(), metrics.ScanReport{TableName: "db.t"}) + require.Eventually(t, func() bool { + return tr.inFlight.Load() == 1 + }, 3*time.Second, 5*time.Millisecond, "worker never saturated") + for range 4 { + rep.Report(context.Background(), metrics.ScanReport{TableName: "db.t"}) + } + require.Eventually(t, func() bool { + return d.dropped.Load() == 3 + }, 3*time.Second, 5*time.Millisecond) + + close(release) + d.close() // triggers the final aggregated flush + + warnings := handler.warnings() + require.Len(t, warnings, 1, "drops must aggregate into a single warning, not one per drop") + attrs := map[string]any{} + warnings[0].Attrs(func(a slog.Attr) bool { + attrs[a.Key] = a.Value.Any() + + return true + }) + assert.Equal(t, uint64(3), attrs["dropped"], "the aggregated warning must carry the drop count") +} + +// TestMetricsDispatcherCloseIsIdempotent proves repeated and concurrent Close +// calls are safe and cheap: they do not panic and each returns, spawning no +// extra waiter goroutine per call. +func TestMetricsDispatcherCloseIsIdempotent(t *testing.T) { + d := newMetricsDispatcher(2, 4, 5*time.Second, slog.New(slog.NewTextHandler(io.Discard, nil))) + + var wg sync.WaitGroup + for range 5 { + wg.Add(1) + go func() { + defer wg.Done() + d.close() + }() + } + + done := make(chan struct{}) + go func() { + wg.Wait() + close(done) + }() + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("concurrent Close calls did not all return") + } + + // A further Close after the fact is still safe. + d.close() +} + +// TestRESTMetricsReporterBuildsPathThroughProduction runs the real path +// composition — splitIdentForPath (encodeNamespace) and endpointReportMetrics +// .reqPath — for a multi-level namespace with segments needing escaping, so a +// regression in that composition or in the separator handling is caught rather +// than bypassed by a hand-built path. +func TestRESTMetricsReporterBuildsPathThroughProduction(t *testing.T) { + c := &Catalog{namespaceSeparator: defaultNamespaceSeparator} + ns, tbl, err := c.splitIdentForPath(table.Identifier{"a b", "d e", "t x"}) + require.NoError(t, err) + path, err := endpointReportMetrics.reqPath(ns, tbl) + require.NoError(t, err) + + received := make(chan capturedRequest, 1) + base, err := url.Parse("http://catalog.invalid/v1/my-prefix";) + require.NoError(t, err) + rep := reporterWith(t, &captureTransport{ch: received}, newTestDispatcher(t, 5*time.Second), path) + rep.baseURI = base + + rep.Report(context.Background(), metrics.ScanReport{TableName: "a b.d e.t x"}) + + select { + case req := <-received: + // The namespace levels are percent-encoded and joined by the encoded + // separator (%1F); the table segment is escaped by the URL builder. + assert.Equal(t, "/v1/my-prefix/namespaces/a%20b%1Fd%20e/tables/t%20x/metrics", req.escapedPath) + case <-time.After(3 * time.Second): + t.Fatal("timed out waiting for the async metrics POST") + } +} + +// TestMetricsDispatcherReportHasFiniteTimeout proves a report does not hang +// forever against a black-holing endpoint: the per-report deadline fires and the +// request context reports DeadlineExceeded. +func TestMetricsDispatcherReportHasFiniteTimeout(t *testing.T) { + tr := &ctxBlockTransport{started: make(chan struct{}, 1), ctxErr: make(chan error, 1)} + d := newMetricsDispatcher(1, 1, 100*time.Millisecond, slog.New(slog.NewTextHandler(io.Discard, nil))) + t.Cleanup(d.close) + rep := reporterWith(t, tr, d, nil) + + rep.Report(context.Background(), metrics.ScanReport{TableName: "db.t"}) + + select { + case err := <-tr.ctxErr: + assert.ErrorIs(t, err, context.DeadlineExceeded, "report must time out rather than hang") + case <-time.After(3 * time.Second): + t.Fatal("report did not observe a finite timeout") + } +} + +// TestMetricsDispatcherBoundsConcurrencyAndDrops proves the worker pool caps +// concurrent sends and sheds excess reports rather than queueing them without +// limit. +func TestMetricsDispatcherBoundsConcurrencyAndDrops(t *testing.T) { + const ( + workers = 2 + queue = 3 + extra = 3 // reports offered once the pool and queue are saturated + ) + tr := &concurrencyTransport{release: make(chan struct{})} + d := newMetricsDispatcher(workers, queue, 5*time.Second, slog.New(slog.NewTextHandler(io.Discard, nil))) + t.Cleanup(d.close) + rep := reporterWith(t, tr, d, nil) + + // Fill every worker; each blocks in RoundTrip, leaving the queue empty. + for range workers { + rep.Report(context.Background(), metrics.ScanReport{TableName: "db.t"}) + } + require.Eventually(t, func() bool { + return tr.inFlight.Load() == workers + }, 3*time.Second, 5*time.Millisecond, "workers never saturated") + + // Fill the queue (workers are blocked, so nothing drains it), then offer more + // than fits — the surplus must be dropped, not queued. + for range queue + extra { + rep.Report(context.Background(), metrics.ScanReport{TableName: "db.t"}) + } + + // Concurrency stayed within the pool while everything was blocked. + assert.LessOrEqual(t, tr.maxSeen.Load(), int32(workers), "concurrency exceeded the worker pool") + + close(tr.release) + + // Only workers + queue reports are ever delivered; the extra were dropped. + require.Eventually(t, func() bool { + return tr.inFlight.Load() == 0 && tr.delivered.Load() == workers+queue + }, 3*time.Second, 5*time.Millisecond, "expected exactly workers+queue reports delivered") + + assert.LessOrEqual(t, tr.maxSeen.Load(), int32(workers), "concurrency exceeded the worker pool") +} + +// TestMetricsDispatcherCloseCancelsInFlight proves Close cancels an in-flight +// report and returns promptly rather than waiting on the stalled endpoint. +func TestMetricsDispatcherCloseCancelsInFlight(t *testing.T) { + tr := &ctxBlockTransport{started: make(chan struct{}, 1), ctxErr: make(chan error, 1)} + // A long per-report timeout so it is Close, not the deadline, that unblocks. + d := newMetricsDispatcher(1, 1, time.Minute, slog.New(slog.NewTextHandler(io.Discard, nil))) Review Comment: This test and `TestMetricsDispatcherCloseDropsQueuedReports` construct the dispatcher directly without registering `t.Cleanup(d.close)` the way `newTestDispatcher` does. Both use a 1-minute timeout, so if an early `t.Fatal` fires before the close runs — the "report never reached the transport" path, say — the workers keep running and under `-race`/goleak that shows up as a flake in some later test. I'd add `t.Cleanup(d.close)` right after each `newMetricsDispatcher` call here and keep the explicit close too. ########## catalog/rest/metrics_reporter.go: ########## @@ -0,0 +1,333 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package rest + +import ( + "context" + "log/slog" + "net/http" + "net/url" + "strconv" + "sync" + "sync/atomic" + "time" + + iceberg "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/metrics" +) + +const ( + // keyReportMetricsEnabled opts a REST catalog into POSTing scan/commit + // reports to the catalog's metrics endpoint. It is disabled by default so + // existing users see no new network traffic unless they turn it on. This is + // the canonical, cross-implementation spelling used by Iceberg Java and the + // Iceberg docs; keyReportMetricsEnabledLegacy is accepted as an alias. + keyReportMetricsEnabled = "rest-metrics-reporting-enabled" + // keyReportMetricsEnabledLegacy is the historical dotted spelling, accepted + // as an alias so existing configs keep working. + keyReportMetricsEnabledLegacy = "rest.metrics-reporting-enabled" + // keyReportMetricsTimeoutMs bounds a single report's request and response + // cycle, in milliseconds. Read from the client-supplied properties only. Auth + // (token refresh) is bounded separately by the OAuth refresh client Timeout. + keyReportMetricsTimeoutMs = "rest-metrics-reporting-timeout-ms" + + // defaultReportMetricsTimeout bounds a single report's request and response + // cycle when keyReportMetricsTimeoutMs is unset. Telemetry is safe to drop, so + // the bound is deliberately short. + defaultReportMetricsTimeout = 10 * time.Second + // metricsDispatchWorkers is the fixed number of goroutines draining the + // dispatch queue, capping the reporter's concurrent connection use. + metricsDispatchWorkers = 4 + // metricsDispatchQueueSize bounds how many reports may await dispatch. + // Reports offered while the queue is full are dropped (and counted, then + // logged in aggregate) rather than queued without limit, so a stalled + // endpoint cannot make reporting grow without bound. + metricsDispatchQueueSize = 128 + // metricsStatsInterval is how often the dispatcher emits an aggregated + // warning summarizing dropped and failed reports. Aggregating keeps a stalled + // or unavailable endpoint from amplifying into one log line per report while + // still surfacing back-pressure. + metricsStatsInterval = 30 * time.Second +) + +// reportMetricsEnabled reports whether the client opted into REST metrics +// reporting, accepting both the canonical and the legacy dotted key. It must be +// given the client-supplied properties only (never the server-merged config) so +// a server cannot flip the default and turn on outbound telemetry the client +// never asked for. +// +// The canonical key wins whenever it is present, so a client that has migrated +// to it and explicitly set it to false is honored even if a stale legacy dotted +// key lingers in the config. The legacy key is consulted only when the canonical +// key is absent. +func reportMetricsEnabled(props iceberg.Properties) bool { + if v, ok := props[keyReportMetricsEnabled]; ok { + enabled, err := strconv.ParseBool(v) + + return err == nil && enabled + } + + return props.GetBool(keyReportMetricsEnabledLegacy, false) +} + +// reportMetricsTimeout resolves the per-report deadline from the client-supplied +// properties, falling back to defaultReportMetricsTimeout for a missing or +// non-positive value. +func reportMetricsTimeout(props iceberg.Properties) time.Duration { + ms := props.GetInt(keyReportMetricsTimeoutMs, int(defaultReportMetricsTimeout/time.Millisecond)) + if ms <= 0 { + return defaultReportMetricsTimeout + } + + return time.Duration(ms) * time.Millisecond +} + +// metricsJob is a single report awaiting dispatch to a table's metrics endpoint. +type metricsJob struct { + // ctx carries the observed scan/commit's context values (trace spans, + // request-scoped attributes) with its cancellation detached — the scan/commit + // is already done, so its cancellation must not abort the report, but its + // values should still propagate to the outbound request. + ctx context.Context + baseURI *url.URL + cl *http.Client + path []string + req metrics.ReportMetricsRequest +} + +// metricsDispatcher POSTs metrics reports to REST metrics endpoints on a fixed +// pool of workers draining a bounded queue. It is owned by the catalog and +// shared across that catalog's table reporters, so concurrent report volume +// stays bounded no matter how many tables are loaded or how often they are +// scanned. A stalled endpoint sheds load — reports are dropped and counted — +// rather than accumulating goroutines and connections. Close cancels in-flight +// reports and drains the workers. +type metricsDispatcher struct { + jobs chan metricsJob + timeout time.Duration + ctx context.Context + cancel context.CancelFunc + wg sync.WaitGroup + logger *slog.Logger // nil means resolve slog.Default at call time + dropped atomic.Uint64 + failed atomic.Uint64 + closeOnce sync.Once + closeDone chan struct{} +} + +func newMetricsDispatcher(workers, queueSize int, timeout time.Duration, logger *slog.Logger) *metricsDispatcher { + ctx, cancel := context.WithCancel(context.Background()) + d := &metricsDispatcher{ + jobs: make(chan metricsJob, queueSize), + timeout: timeout, + ctx: ctx, + cancel: cancel, + logger: logger, + closeDone: make(chan struct{}), + } + + d.wg.Add(workers + 1) + for range workers { + go d.worker() + } + go d.reportStats() + + return d +} + +func (d *metricsDispatcher) log() *slog.Logger { + if d.logger != nil { + return d.logger + } + + return slog.Default() +} + +func (d *metricsDispatcher) worker() { + defer d.wg.Done() + for { + select { + case <-d.ctx.Done(): + return + case job := <-d.jobs: + d.send(job) + } + } +} + +func (d *metricsDispatcher) send(job metricsJob) { + // If the dispatcher is already shutting down, drop the report before doing + // any work: the derived context would be cancelled immediately anyway. + if d.ctx.Err() != nil { + return + } + + defer func() { + if r := recover(); r != nil { + d.log().Warn("iceberg: panic while reporting metrics to REST catalog", "recovered", r) + } + }() + + // Bound the report by the per-report deadline, starting from the job's + // context so the caller's values (trace spans, request-scoped attributes) + // still reach the transport, and tie the derived context to dispatcher + // shutdown so Close cancels in-flight requests. The deadline covers the + // request and response cycle; a token refresh triggered by auth is bounded + // separately by the OAuth refresh client's Timeout. + ctx, cancel := context.WithTimeout(job.ctx, d.timeout) + defer cancel() + stop := context.AfterFunc(d.ctx, cancel) + defer stop() + + if _, err := doPost[metrics.ReportMetricsRequest, struct{}]( + ctx, job.baseURI, job.path, job.req, job.cl, nil, allowNoContent()); err != nil { + // A report interrupted by Close (dispatcher context cancelled) is expected + // shutdown behavior, not a failure worth counting. A per-report timeout + // leaves the dispatcher context live, so genuine failures still surface. + if d.ctx.Err() != nil { + return + } + d.failed.Add(1) + // The transport error embeds the request URL, which can carry a sensitive + // host, prefix, namespace or table name, so keep the detail at Debug and + // let reportStats surface the aggregate count at Warn. + d.log().Debug("iceberg: failed to report metrics to REST catalog", "error", err) + } +} + +// submit offers a job to the queue without blocking. It drops the report (and +// counts the drop, aggregated and logged by reportStats so back-pressure is +// visible without amplifying into one log line per drop) when the queue is +// full, and ignores reports once the dispatcher is closed. Every path here is +// non-blocking so Report never blocks the observed scan/commit. +func (d *metricsDispatcher) submit(job metricsJob) { + select { + case <-d.ctx.Done(): + return + default: + } + + select { + case d.jobs <- job: + case <-d.ctx.Done(): + default: + d.dropped.Add(1) + } +} + +// reportStats periodically emits an aggregated warning summarizing reports +// dropped (queue full) and failed (endpoint errors) since the last tick, so a +// stalled or unavailable endpoint surfaces as bounded, rate-limited log volume +// rather than one line per report. A final summary is emitted on shutdown. +func (d *metricsDispatcher) reportStats() { + defer d.wg.Done() + + ticker := time.NewTicker(metricsStatsInterval) + defer ticker.Stop() + + var lastDropped, lastFailed uint64 + flush := func() { + dropped, failed := d.dropped.Load(), d.failed.Load() + if dropped == lastDropped && failed == lastFailed { + return + } + d.log().Warn("iceberg: REST metrics reports dropped or failed", + "dropped", dropped-lastDropped, "dropped_total", dropped, + "failed", failed-lastFailed, "failed_total", failed) + lastDropped, lastFailed = dropped, failed + } + + for { + select { + case <-d.ctx.Done(): + flush() + + return + case <-ticker.C: + flush() + } + } +} + +// close cancels in-flight reports and waits for the workers to return, bounded +// by the report timeout so shutdown cannot hang on a stalled endpoint. It is +// one-shot: repeated calls block until the first completes, then return, so no +// extra waiter goroutine is spawned per call. +func (d *metricsDispatcher) close() { + d.closeOnce.Do(func() { + d.cancel() + + done := make(chan struct{}) + go func() { + d.wg.Wait() + close(done) + }() + + select { + case <-done: + case <-time.After(d.timeout): Review Comment: `time.After(d.timeout)` leaves a live timer around until it fires whenever the `<-done` arm wins, which is the common path — with the 10s default that's a 10s-lived timer per catalog Close, and it adds up in a process that cycles through many catalogs. I'd switch to `time.NewTimer` with a `defer Stop`: ```go t := time.NewTimer(d.timeout) defer t.Stop() select { case <-done: case <-t.C: } ``` While we're here: on the timeout arm this returns while the `wg.Wait()` goroutine (and maybe a worker) is still alive. It self-exits once `d.cancel()` takes effect so it's not a leak, but it does mean Close can return before workers have fully stopped — worth a one-line comment so nobody later tears down a shared transport assuming the workers are done. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
