zeroshade commented on code in PR #1655: URL: https://github.com/apache/iceberg-go/pull/1655#discussion_r3737958963
########## catalog/rest/metrics_reporter.go: ########## @@ -0,0 +1,246 @@ +// 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" + "sync" + "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 total auth + request + + // response cycle, in milliseconds. Read from the client-supplied properties + // only. + keyReportMetricsTimeoutMs = "rest-metrics-reporting-timeout-ms" + + // defaultReportMetricsTimeout bounds a single report's total auth + request + // + 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 logged) rather + // than queued without limit, so a stalled endpoint cannot make reporting grow + // without bound. + metricsDispatchQueueSize = 128 +) + +// 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. +func reportMetricsEnabled(props iceberg.Properties) bool { + return props.GetBool(keyReportMetricsEnabled, false) || + 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 { + 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 logged — +// 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 +} + +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, + } + + d.wg.Add(workers) + for range workers { + go d.worker() + } + + 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) + } + }() + + // Derive from the dispatcher context so Close cancels in-flight requests, + // and add the per-report deadline so a stalled endpoint cannot pin a worker + // (and its connection) indefinitely. The deadline covers auth plus the full + // request and response cycle. + ctx, cancel := context.WithTimeout(d.ctx, d.timeout) Review Comment: The deadline is real and it does bound the request and response cycle — but it does not bound authentication, which runs before the transport ever sees this context. `sessionTransport.RoundTrip` calls `AuthHeader()` with no context (`rest.go:264-269`), `oauthAuthManager` calls `TokenSource.Token()` with no context (`auth.go:40-43`), and the built-in token source is constructed against `context.Background()` (`rest.go:880-903`) on a client with no `Timeout`. Nothing in that chain observes `ctx`. With one worker, `timeout` at 30ms, and an `AuthManager` blocked inside `AuthHeader`, the worker was still pinned at 80ms; `close` returned only when its own timer expired, and the worker plus the `wg.Wait` goroutine survived until auth was released by hand. That is the same unbounded-hang shape as the previous round, just relocated from the metrics endpoint to the token endpoint — and a stalled token endpoint is at least as likely as a stalled metrics endpoint. Worth noting that the comment on lines 160-161 asserts the deadline covers auth. It is the one thing it does not cover, so the comment will mislead the next reader. Suggested fix: thread a context through auth retrieval so `AuthHeader`/`Token` honor the caller's deadline, including built-in OAuth refresh. Wrapping the context-free call in another goroutine only moves the leak somewhere harder to see. Failing that, a `Timeout` on the client used for token refresh would at least bound it. ########## catalog/rest/metrics_reporter.go: ########## @@ -0,0 +1,246 @@ +// 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" + "sync" + "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 total auth + request + + // response cycle, in milliseconds. Read from the client-supplied properties + // only. + keyReportMetricsTimeoutMs = "rest-metrics-reporting-timeout-ms" + + // defaultReportMetricsTimeout bounds a single report's total auth + request + // + 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 logged) rather + // than queued without limit, so a stalled endpoint cannot make reporting grow + // without bound. + metricsDispatchQueueSize = 128 +) + +// 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. +func reportMetricsEnabled(props iceberg.Properties) bool { + return props.GetBool(keyReportMetricsEnabled, false) || + 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 { + 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 logged — +// 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 +} + +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, + } + + d.wg.Add(workers) + for range workers { + go d.worker() + } + + 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) + } + }() + + // Derive from the dispatcher context so Close cancels in-flight requests, + // and add the per-report deadline so a stalled endpoint cannot pin a worker + // (and its connection) indefinitely. The deadline covers auth plus the full + // request and response cycle. + ctx, cancel := context.WithTimeout(d.ctx, d.timeout) + defer cancel() + + 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 logging. A per-report timeout + // leaves the dispatcher context live, so genuine timeouts still surface. + if d.ctx.Err() != nil { + return + } + d.log().Warn("iceberg: failed to report metrics to REST catalog", "error", err) + } +} + +// submit offers a job to the queue without blocking. It drops the report (and +// logs the drop, so back-pressure is visible rather than silent) when the queue +// is full, and ignores reports once the dispatcher is closed. +func (d *metricsDispatcher) submit(job metricsJob) { + select { + case <-d.ctx.Done(): + return + default: + } + + select { + case d.jobs <- job: + case <-d.ctx.Done(): + default: + d.log().Warn("iceberg: metrics report dropped; dispatch queue full") + } +} + +// 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. +func (d *metricsDispatcher) close() { + d.cancel() + + done := make(chan struct{}) + go func() { Review Comment: `close` is bounded now, which was the main thing, and because `d.jobs` is never closed a concurrent `Report` cannot panic on a send to a closed channel. Cooperative concurrent `Report`/`close` and double-close stress came back race-clean. Two residual issues: Each call to `close` spawns another waiter goroutine here. If a worker is stuck in the uncancellable auth path described above, every repeated `close` leaves one more `wg.Wait` goroutine behind permanently. Making shutdown one-shot — a `sync.Once` plus a shared completion channel that later calls select on — makes repeated `Close` cheap and leak-free. More importantly, the bound is on how long `close` *waits*, not on the work itself. When a worker is pinned in auth, `close` returns after `d.timeout` and reports success while the worker and its connection are still live. A caller that closes the catalog and expects outbound network activity to have stopped does not get that guarantee. Fixing the auth cancellation above is what makes this bound meaningful. ########## catalog/rest/metrics_reporter.go: ########## @@ -0,0 +1,246 @@ +// 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" + "sync" + "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 total auth + request + + // response cycle, in milliseconds. Read from the client-supplied properties + // only. + keyReportMetricsTimeoutMs = "rest-metrics-reporting-timeout-ms" + + // defaultReportMetricsTimeout bounds a single report's total auth + request + // + 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 logged) rather + // than queued without limit, so a stalled endpoint cannot make reporting grow + // without bound. + metricsDispatchQueueSize = 128 +) + +// 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. +func reportMetricsEnabled(props iceberg.Properties) bool { + return props.GetBool(keyReportMetricsEnabled, false) || + 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 { + 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 logged — +// 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 +} + +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, + } + + d.wg.Add(workers) + for range workers { + go d.worker() + } + + 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) + } + }() + + // Derive from the dispatcher context so Close cancels in-flight requests, + // and add the per-report deadline so a stalled endpoint cannot pin a worker + // (and its connection) indefinitely. The deadline covers auth plus the full + // request and response cycle. + ctx, cancel := context.WithTimeout(d.ctx, d.timeout) + defer cancel() + + 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 logging. A per-report timeout + // leaves the dispatcher context live, so genuine timeouts still surface. + if d.ctx.Err() != nil { + return + } + d.log().Warn("iceberg: failed to report metrics to REST catalog", "error", err) + } +} + +// submit offers a job to the queue without blocking. It drops the report (and +// logs the drop, so back-pressure is visible rather than silent) when the queue +// is full, and ignores reports once the dispatcher is closed. +func (d *metricsDispatcher) submit(job metricsJob) { + select { + case <-d.ctx.Done(): + return + default: + } + + select { + case d.jobs <- job: + case <-d.ctx.Done(): + default: + d.log().Warn("iceberg: metrics report dropped; dispatch queue full") Review Comment: This log call is synchronous, and it runs on the caller's goroutine — the scan or commit being observed. A slow or blocking `slog.Handler` therefore blocks `Report`, which contradicts the contract at `metrics/reporter.go:66-70` that reporting returns promptly. A full-queue probe with a deliberately blocking handler held `Report` until the handler was released. This is my fault as much as anything: logging the drops was my suggestion last round, to keep back-pressure visible rather than silent. Putting that log on the synchronous submit path is what turned the safety valve into a blocking path. The goal stands, the placement needs to move. There is a secondary concern even with a fast handler: one warning per drop means log amplification precisely during overload, when the queue is full and drops are arriving at scan/commit rate. Suggested fix: keep `submit` to non-blocking operations only — bump an atomic drop counter, or do a non-blocking send on a small notification channel — and have a dispatcher-owned goroutine emit an aggregated, rate-limited warning ("N reports dropped in the last interval"). That preserves visibility, keeps the caller unblocked, and removes the amplification. ########## catalog/rest/metrics_reporter_test.go: ########## @@ -0,0 +1,487 @@ +// 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" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "net/url" + "sync/atomic" + "testing" + "time" + + iceberg "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/metrics" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type capturedRequest struct { + method string + path string // decoded path + escapedPath string // percent-encoded path + 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(), + 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"}, Review Comment: This is better than the previous version — the prefix is now real and the escaping assertions on lines 256-257 are meaningful. Two things still keep it from guarding the fragile part. `path` is supplied pre-built here, so the test never runs the production composition: `tableFromResponse`, `splitIdentForPath`, `encodeNamespace`, and `endpointReportMetrics.reqPath` (`rest.go:1140-1146`) are all bypassed. If that composition broke, this test would still pass. I checked the real path separately and it does produce the correct nested result, so this is about protecting a working behavior rather than a live bug. The namespace is also single-level (`"a b"`). Multi-level namespaces are where separator handling and escaping interact, and that is the case most likely to regress. Suggested fix: build the reporter through `tableFromResponse` (or at minimum call `endpointReportMetrics.reqPath`) with a configured prefix and a nested namespace such as `{"a b", "c/d"}`, then assert the complete escaped path. Worth coordinating with `9accf20` (#1614), which touched `encodeNamespace` and is part of the rebase. ########## catalog/rest/metrics_reporter_test.go: ########## @@ -0,0 +1,487 @@ +// 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" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "net/url" + "sync/atomic" + "testing" + "time" + + iceberg "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/metrics" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type capturedRequest struct { Review Comment: Still open from last round: `capturedRequest` records method, decoded path, escaped path and body, but not headers, so nothing asserts that the metrics POST actually goes out authenticated. That reuse of the catalog's already-authenticated client is one of the better properties of this design, and it is currently load-bearing but untested — a refactor that sent metrics through a bare `http.Client` would keep every existing test green. Suggested fix: add `header http.Header` to `capturedRequest`, populate it in `captureTransport.RoundTrip` (clone it — the SDK may reuse the request), and assert `Authorization` is present, `Content-Type: application/json`, and any catalog-configured headers are forwarded. A case where the token refreshes between two reports, asserting the second carries the new credential, would cover the auth path this PR depends on and currently exercises nowhere. ########## catalog/rest/metrics_reporter.go: ########## @@ -0,0 +1,246 @@ +// 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" + "sync" + "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 total auth + request + + // response cycle, in milliseconds. Read from the client-supplied properties + // only. + keyReportMetricsTimeoutMs = "rest-metrics-reporting-timeout-ms" + + // defaultReportMetricsTimeout bounds a single report's total auth + request + // + 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 logged) rather + // than queued without limit, so a stalled endpoint cannot make reporting grow + // without bound. + metricsDispatchQueueSize = 128 +) + +// 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. +func reportMetricsEnabled(props iceberg.Properties) bool { + return props.GetBool(keyReportMetricsEnabled, false) || Review Comment: Non-blocking. Accepting both spellings with the canonical one first, and the alias documented as legacy, is exactly right. One edge worth pinning down: because this is an `||`, a canonical `rest-metrics-reporting-enabled=false` combined with a legacy `rest.metrics-reporting-enabled=true` enables reporting. Someone migrating to the canonical key and explicitly setting it to `false` while an old dotted key lingers in their config would get telemetry they just tried to turn off. Suggested fix: let the canonical key win when it is present at all, and fall back to the legacy key only when the canonical one is absent. A short comment plus a test for the conflicting case would settle it. ########## catalog/rest/metrics_reporter.go: ########## @@ -0,0 +1,246 @@ +// 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" + "sync" + "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 total auth + request + + // response cycle, in milliseconds. Read from the client-supplied properties + // only. + keyReportMetricsTimeoutMs = "rest-metrics-reporting-timeout-ms" + + // defaultReportMetricsTimeout bounds a single report's total auth + request + // + 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 logged) rather + // than queued without limit, so a stalled endpoint cannot make reporting grow + // without bound. + metricsDispatchQueueSize = 128 +) + +// 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. +func reportMetricsEnabled(props iceberg.Properties) bool { + return props.GetBool(keyReportMetricsEnabled, false) || + 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 { + 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 logged — +// 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 +} + +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, + } + + d.wg.Add(workers) + for range workers { + go d.worker() + } + + 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) + } + }() + + // Derive from the dispatcher context so Close cancels in-flight requests, + // and add the per-report deadline so a stalled endpoint cannot pin a worker + // (and its connection) indefinitely. The deadline covers auth plus the full + // request and response cycle. + ctx, cancel := context.WithTimeout(d.ctx, d.timeout) + defer cancel() + + 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 logging. A per-report timeout + // leaves the dispatcher context live, so genuine timeouts still surface. + if d.ctx.Err() != nil { + return + } + d.log().Warn("iceberg: failed to report metrics to REST catalog", "error", err) + } +} + +// submit offers a job to the queue without blocking. It drops the report (and +// logs the drop, so back-pressure is visible rather than silent) when the queue +// is full, and ignores reports once the dispatcher is closed. +func (d *metricsDispatcher) submit(job metricsJob) { + select { + case <-d.ctx.Done(): + return + default: + } + + select { + case d.jobs <- job: + case <-d.ctx.Done(): + default: + d.log().Warn("iceberg: metrics report dropped; dispatch queue full") + } +} + +// 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. +func (d *metricsDispatcher) close() { + d.cancel() + + done := make(chan struct{}) + go func() { + d.wg.Wait() + close(done) + }() + + select { + case <-done: + case <-time.After(d.timeout): + } +} + +// restMetricsReporter POSTs metrics reports for a single table to its REST +// metrics endpoint (POST .../tables/{table}/metrics) via the catalog-owned +// dispatcher. It satisfies metrics.Reporter. +type restMetricsReporter struct { + baseURI *url.URL + cl *http.Client + path []string // table metrics path, relative to baseURI + dispatcher *metricsDispatcher +} + +var _ metrics.Reporter = (*restMetricsReporter)(nil) + +// Report wraps the report in a ReportMetricsRequest and hands it to the +// catalog's dispatcher, returning immediately. Per the Reporter contract it +// never blocks or fails the observed scan/commit: dispatch is asynchronous and +// bounded, the send is detached from the caller's cancellation (the scan/commit +// is already done), and any error is logged and swallowed by the dispatcher. +func (rep *restMetricsReporter) Report(_ context.Context, report metrics.MetricsReport) { Review Comment: Non-blocking, but a small regression against the previous revision. Discarding the caller's context drops its *values* as well as its cancellation; the earlier `context.WithoutCancel(ctx)` detached cancellation while preserving values, and the dispatcher now starts from `context.Background()` (`:108`, `:162`). A probe confirmed a value set by the caller no longer reaches the transport. That matters for anything propagated contextually rather than through the request — tracing spans and request-scoped logging attributes being the usual cases. Detaching from cancellation is right; detaching from values is an unrelated side effect. Suggested fix: keep `context.WithoutCancel(ctx)` on the queued job, and in the worker combine it with dispatcher cancellation and the per-report deadline. ########## catalog/rest/metrics_reporter.go: ########## @@ -0,0 +1,246 @@ +// 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" + "sync" + "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 total auth + request + + // response cycle, in milliseconds. Read from the client-supplied properties + // only. + keyReportMetricsTimeoutMs = "rest-metrics-reporting-timeout-ms" + + // defaultReportMetricsTimeout bounds a single report's total auth + request + // + 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 logged) rather + // than queued without limit, so a stalled endpoint cannot make reporting grow + // without bound. + metricsDispatchQueueSize = 128 +) + +// 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. +func reportMetricsEnabled(props iceberg.Properties) bool { + return props.GetBool(keyReportMetricsEnabled, false) || + 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 { + 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 logged — +// 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 +} + +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, + } + + d.wg.Add(workers) + for range workers { + go d.worker() + } + + 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) + } + }() + + // Derive from the dispatcher context so Close cancels in-flight requests, + // and add the per-report deadline so a stalled endpoint cannot pin a worker + // (and its connection) indefinitely. The deadline covers auth plus the full + // request and response cycle. + ctx, cancel := context.WithTimeout(d.ctx, d.timeout) + defer cancel() + + 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 logging. A per-report timeout + // leaves the dispatcher context live, so genuine timeouts still surface. + if d.ctx.Err() != nil { + return + } + d.log().Warn("iceberg: failed to report metrics to REST catalog", "error", err) Review Comment: Non-blocking. Distinguishing shutdown cancellation from genuine failures via `d.ctx.Err()` is a nice touch. One caution: the raw transport error typically embeds the full request URL, which here carries the catalog host, any configured prefix, the namespace and the table name, plus any query parameters on the base URI. For deployments where table names or URI parameters are sensitive, that lands in operator logs on every failed report. Authorization headers are not logged, so this is limited to URL content. Also, since a persistently unavailable endpoint fails one report per scan and per commit, this warning can become steady-state noise. Rate limiting or aggregating it — alongside the drop counter suggested on the `submit` path — would keep it useful. ########## catalog/rest/metrics_reporter_test.go: ########## @@ -0,0 +1,487 @@ +// 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" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "net/url" + "sync/atomic" + "testing" + "time" + + iceberg "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/metrics" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type capturedRequest struct { + method string + path string // decoded path + escapedPath string // percent-encoded path + 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(), + 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") + } +} + +// 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"})) +} + +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) { Review Comment: This test covers the important cases well: default-off, client opt-in, server `defaults`, and server `overrides`. Given that server-controlled enablement was the security-relevant defect, having it pinned here is the most valuable test in the file. The gap is that the doc comment on lines 407-408 asserts table-response properties cannot enable reporting, but no case exercises that. It is currently true by construction — enablement resolves once at init from client properties — and that structural argument is exactly what a future refactor could invalidate, silently, while every test here still passes. Suggested fix: add a case that loads a table whose *table response* carries `rest-metrics-reporting-enabled=true` with the client not opting in, and assert no metrics traffic is produced. That converts the comment into an enforced invariant and completes the precedence matrix from the last round. -- 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]
