You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by ze...@apache.org on 2026/08/12 18:42:54 UTC
(iceberg-go) branch main updated: feat(metrics): add OpenTelemetry reporter (#1236) (#1767)
This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new faa320c1c feat(metrics): add OpenTelemetry reporter (#1236) (#1767)
faa320c1c is described below
commit faa320c1c019744ad06382a0de5a4884e84ceabf
Author: Tanmay Rauth <t_...@apple.com>
AuthorDate: Wed Aug 12 11:42:49 2026 -0700
feat(metrics): add OpenTelemetry reporter (#1236) (#1767)
Seventh PR of the metrics reporting framework. Adds an in-module
metrics/otel package with an OpenTelemetry-backed reporter that maps
every field of a ScanReport's ScanMetricsResult and a CommitReport's
CommitMetricsResult to an OpenTelemetry instrument, so scan/commit
metrics can flow to any OTLP backend.
The host owns the OpenTelemetry SDK: the reporter takes a Meter from the
global MeterProvider (or one supplied via WithMeter); with no SDK
registered the instruments are no-ops. Durations are histograms (ms);
every other field is a monotonic counter (bytes carry the By unit,
counts are unit-less). By default scan metrics carry iceberg.table.name
and commit metrics carry iceberg.table.name and iceberg.operation;
iceberg.schema.id is opt-in via WithAttributes and snapshot id is never
a metric attribute. It is an in-module package rather than a separate
module because go.opentelemetry.io is already a transitive dependency.
Names, units, and the attribute set mirror Java's OtelMetricsReporter
(apache/iceberg#16250) so metrics stay consistent across
implementations; treat them as provisional until that PR merges.
---
go.mod | 6 +-
metrics/otel/reporter.go | 314 ++++++++++++++++++++++++++++++++++
metrics/otel/reporter_test.go | 384 ++++++++++++++++++++++++++++++++++++++++++
3 files changed, 701 insertions(+), 3 deletions(-)
diff --git a/go.mod b/go.mod
index d2502fb82..b831d6dd7 100644
--- a/go.mod
+++ b/go.mod
@@ -58,6 +58,9 @@ require (
github.com/uptrace/bun/dialect/sqlitedialect v1.2.18
github.com/uptrace/bun/driver/sqliteshim v1.2.18
github.com/uptrace/bun/extra/bundebug v1.2.18
+ go.opentelemetry.io/otel v1.44.0
+ go.opentelemetry.io/otel/metric v1.44.0
+ go.opentelemetry.io/otel/sdk/metric v1.44.0
gocloud.dev v0.46.0
golang.org/x/exp v0.0.0-20260218203240-3dfff04db8fa
golang.org/x/oauth2 v0.36.0
@@ -252,15 +255,12 @@ require (
go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.68.0 // indirect
go.opentelemetry.io/contrib/instrumentation/net/http/httptrace/otelhttptrace v0.63.0 // indirect
go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.67.0 // indirect
- go.opentelemetry.io/otel v1.44.0 // indirect
go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v1.42.0 // indirect
go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v1.43.0 // indirect
go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.43.0 // indirect
go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.42.0 // indirect
go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.43.0 // indirect
- go.opentelemetry.io/otel/metric v1.44.0 // indirect
go.opentelemetry.io/otel/sdk v1.44.0 // indirect
- go.opentelemetry.io/otel/sdk/metric v1.44.0 // indirect
go.opentelemetry.io/otel/trace v1.44.0 // indirect
go.opentelemetry.io/proto/otlp v1.10.0 // indirect
go.yaml.in/yaml/v3 v3.0.4 // indirect
diff --git a/metrics/otel/reporter.go b/metrics/otel/reporter.go
new file mode 100644
index 000000000..8f443ed83
--- /dev/null
+++ b/metrics/otel/reporter.go
@@ -0,0 +1,314 @@
+// 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 otel provides an OpenTelemetry-backed [metrics.Reporter] for
+// iceberg-go. It maps every field of a ScanReport's ScanMetricsResult and a
+// CommitReport's CommitMetricsResult to an OpenTelemetry instrument, so
+// scan/commit metrics can flow to any OTLP backend (Prometheus, Datadog,
+// CloudWatch, ...). Durations are histograms (ms); every other field is a
+// monotonic counter (bytes carry the "By" unit, counts are unit-less). Fields
+// that are unset in a given report are skipped, so no empty time series is
+// created.
+//
+// The host application owns the OpenTelemetry SDK: the reporter obtains a Meter
+// from the global MeterProvider (or one supplied via [WithMeter]). If no SDK is
+// registered, the global provider returns no-op instruments and metric calls
+// are silently dropped — the standard OpenTelemetry contract.
+//
+// Cardinality: by default scan metrics carry iceberg.table.name and commit
+// metrics carry iceberg.table.name and iceberg.operation. iceberg.schema.id is
+// opt-in via [WithAttributes]. The snapshot id is never emitted as an attribute
+// — it is unique per commit, so attaching it would create a new time series for
+// every commit. Because iceberg.table.name is on by default, the number of time
+// series grows with the number of tables; the OpenTelemetry metrics SDK caps a
+// stream at 2000 attribute combinations by default and folds the overflow into a
+// single otel.metric.overflow=true point, so drop iceberg.table.name (via
+// [WithAttributes]) in deployments with a very large number of tables.
+//
+// The names, units, and attribute set mirror Java's OtelMetricsReporter
+// (apache/iceberg#16250) so metrics are consistent across implementations.
+// Treat them as provisional until that PR merges.
+package otel
+
+import (
+ "context"
+
+ "github.com/apache/iceberg-go/metrics"
+ "go.opentelemetry.io/otel"
+ "go.opentelemetry.io/otel/attribute"
+ "go.opentelemetry.io/otel/metric"
+)
+
+// instrumentationName scopes the emitted instruments, mirroring Java's
+// "org.apache.iceberg" meter name.
+const instrumentationName = "github.com/apache/iceberg-go"
+
+// unitBytes is the UCUM unit emitted for byte-valued counters, matching Java.
+const unitBytes = "By"
+
+// Attribute short names accepted by [WithAttributes].
+const (
+ AttrTableName = "table-name"
+ AttrSchemaID = "schema-id"
+ AttrOperation = "operation"
+)
+
+// Emitted OTel attribute keys. Snapshot ID is deliberately never an attribute
+// (unbounded cardinality).
+const (
+ keyTableName = "iceberg.table.name"
+ keySchemaID = "iceberg.schema.id"
+ keyOperation = "iceberg.operation"
+)
+
+// scanCounterSpec describes one scan counter instrument and how to read its
+// value out of a ScanMetricsResult.
+type scanCounterSpec struct {
+ name, desc, unit string
+ get func(metrics.ScanMetricsResult) *metrics.CounterResult
+}
+
+// commitCounterSpec describes one commit counter instrument and how to read its
+// value out of a CommitMetricsResult.
+type commitCounterSpec struct {
+ name, desc, unit string
+ get func(metrics.CommitMetricsResult) *metrics.CounterResult
+}
+
+// scanCounterSpecs enumerates every counter derived from ScanMetricsResult. The
+// order, names, units, and descriptions mirror Java's ScanInstruments.
+var scanCounterSpecs = []scanCounterSpec{
+ {"iceberg.scan.result.data_files", "Number of data files included in scan result", "", func(m metrics.ScanMetricsResult) *metrics.CounterResult { return m.ResultDataFiles }},
+ {"iceberg.scan.result.delete_files", "Number of delete files included in scan result", "", func(m metrics.ScanMetricsResult) *metrics.CounterResult { return m.ResultDeleteFiles }},
+ {"iceberg.scan.data_manifests.total", "Total number of data manifests", "", func(m metrics.ScanMetricsResult) *metrics.CounterResult { return m.TotalDataManifests }},
+ {"iceberg.scan.delete_manifests.total", "Total number of delete manifests", "", func(m metrics.ScanMetricsResult) *metrics.CounterResult { return m.TotalDeleteManifests }},
+ {"iceberg.scan.data_manifests.scanned", "Number of data manifests scanned", "", func(m metrics.ScanMetricsResult) *metrics.CounterResult { return m.ScannedDataManifests }},
+ {"iceberg.scan.data_manifests.skipped", "Number of data manifests skipped", "", func(m metrics.ScanMetricsResult) *metrics.CounterResult { return m.SkippedDataManifests }},
+ {"iceberg.scan.file_size.bytes", "Total file size of data files in scan result", unitBytes, func(m metrics.ScanMetricsResult) *metrics.CounterResult { return m.TotalFileSizeInBytes }},
+ {"iceberg.scan.delete_file_size.bytes", "Total file size of delete files in scan result", unitBytes, func(m metrics.ScanMetricsResult) *metrics.CounterResult { return m.TotalDeleteFileSizeInBytes }},
+ {"iceberg.scan.data_files.skipped", "Number of data files skipped during scan", "", func(m metrics.ScanMetricsResult) *metrics.CounterResult { return m.SkippedDataFiles }},
+ {"iceberg.scan.delete_files.skipped", "Number of delete files skipped during scan", "", func(m metrics.ScanMetricsResult) *metrics.CounterResult { return m.SkippedDeleteFiles }},
+ {"iceberg.scan.delete_manifests.scanned", "Number of delete manifests scanned", "", func(m metrics.ScanMetricsResult) *metrics.CounterResult { return m.ScannedDeleteManifests }},
+ {"iceberg.scan.delete_manifests.skipped", "Number of delete manifests skipped", "", func(m metrics.ScanMetricsResult) *metrics.CounterResult { return m.SkippedDeleteManifests }},
+ {"iceberg.scan.delete_files.indexed", "Number of indexed delete files", "", func(m metrics.ScanMetricsResult) *metrics.CounterResult { return m.IndexedDeleteFiles }},
+ {"iceberg.scan.delete_files.equality", "Number of equality delete files", "", func(m metrics.ScanMetricsResult) *metrics.CounterResult { return m.EqualityDeleteFiles }},
+ {"iceberg.scan.delete_files.positional", "Number of positional delete files", "", func(m metrics.ScanMetricsResult) *metrics.CounterResult { return m.PositionalDeleteFiles }},
+ {"iceberg.scan.dvs", "Number of deletion vectors in scan result", "", func(m metrics.ScanMetricsResult) *metrics.CounterResult { return m.DVs }},
+}
+
+// commitCounterSpecs enumerates every counter derived from CommitMetricsResult.
+// The order, names, units, and descriptions mirror Java's CommitInstruments.
+var commitCounterSpecs = []commitCounterSpec{
+ {"iceberg.commit.attempts", "Number of commit attempts", "", func(m metrics.CommitMetricsResult) *metrics.CounterResult { return m.Attempts }},
+ {"iceberg.commit.data_files.added", "Number of data files added by commit", "", func(m metrics.CommitMetricsResult) *metrics.CounterResult { return m.AddedDataFiles }},
+ {"iceberg.commit.data_files.removed", "Number of data files removed by commit", "", func(m metrics.CommitMetricsResult) *metrics.CounterResult { return m.RemovedDataFiles }},
+ {"iceberg.commit.data_files.total", "Total number of data files after commit", "", func(m metrics.CommitMetricsResult) *metrics.CounterResult { return m.TotalDataFiles }},
+ {"iceberg.commit.delete_files.added", "Number of delete files added by commit", "", func(m metrics.CommitMetricsResult) *metrics.CounterResult { return m.AddedDeleteFiles }},
+ {"iceberg.commit.delete_files.equality.added", "Number of equality delete files added by commit", "", func(m metrics.CommitMetricsResult) *metrics.CounterResult { return m.AddedEqualityDeleteFiles }},
+ {"iceberg.commit.delete_files.positional.added", "Number of positional delete files added by commit", "", func(m metrics.CommitMetricsResult) *metrics.CounterResult { return m.AddedPositionalDeleteFiles }},
+ {"iceberg.commit.dvs.added", "Number of deletion vectors added by commit", "", func(m metrics.CommitMetricsResult) *metrics.CounterResult { return m.AddedDVs }},
+ {"iceberg.commit.delete_files.removed", "Number of delete files removed by commit", "", func(m metrics.CommitMetricsResult) *metrics.CounterResult { return m.RemovedDeleteFiles }},
+ {"iceberg.commit.delete_files.equality.removed", "Number of equality delete files removed by commit", "", func(m metrics.CommitMetricsResult) *metrics.CounterResult { return m.RemovedEqualityDeleteFiles }},
+ {"iceberg.commit.delete_files.positional.removed", "Number of positional delete files removed by commit", "", func(m metrics.CommitMetricsResult) *metrics.CounterResult { return m.RemovedPositionalDeleteFiles }},
+ {"iceberg.commit.dvs.removed", "Number of deletion vectors removed by commit", "", func(m metrics.CommitMetricsResult) *metrics.CounterResult { return m.RemovedDVs }},
+ {"iceberg.commit.delete_files.total", "Total number of delete files after commit", "", func(m metrics.CommitMetricsResult) *metrics.CounterResult { return m.TotalDeleteFiles }},
+ {"iceberg.commit.records.added", "Number of records added by commit", "", func(m metrics.CommitMetricsResult) *metrics.CounterResult { return m.AddedRecords }},
+ {"iceberg.commit.records.removed", "Number of records removed by commit", "", func(m metrics.CommitMetricsResult) *metrics.CounterResult { return m.RemovedRecords }},
+ {"iceberg.commit.records.total", "Total number of records after commit", "", func(m metrics.CommitMetricsResult) *metrics.CounterResult { return m.TotalRecords }},
+ {"iceberg.commit.file_size.added_bytes", "Total size of data files added by commit", unitBytes, func(m metrics.CommitMetricsResult) *metrics.CounterResult { return m.AddedFilesSizeBytes }},
+ {"iceberg.commit.file_size.removed_bytes", "Total size of data files removed by commit", unitBytes, func(m metrics.CommitMetricsResult) *metrics.CounterResult { return m.RemovedFilesSizeBytes }},
+ {"iceberg.commit.file_size.total_bytes", "Total size of all data files after commit", unitBytes, func(m metrics.CommitMetricsResult) *metrics.CounterResult { return m.TotalFilesSizeBytes }},
+ {"iceberg.commit.positional_deletes.added", "Number of positional deletes added by commit", "", func(m metrics.CommitMetricsResult) *metrics.CounterResult { return m.AddedPositionalDeletes }},
+ {"iceberg.commit.positional_deletes.removed", "Number of positional deletes removed by commit", "", func(m metrics.CommitMetricsResult) *metrics.CounterResult { return m.RemovedPositionalDeletes }},
+ {"iceberg.commit.positional_deletes.total", "Total number of positional deletes after commit", "", func(m metrics.CommitMetricsResult) *metrics.CounterResult { return m.TotalPositionalDeletes }},
+ {"iceberg.commit.equality_deletes.added", "Number of equality deletes added by commit", "", func(m metrics.CommitMetricsResult) *metrics.CounterResult { return m.AddedEqualityDeletes }},
+ {"iceberg.commit.equality_deletes.removed", "Number of equality deletes removed by commit", "", func(m metrics.CommitMetricsResult) *metrics.CounterResult { return m.RemovedEqualityDeletes }},
+ {"iceberg.commit.equality_deletes.total", "Total number of equality deletes after commit", "", func(m metrics.CommitMetricsResult) *metrics.CounterResult { return m.TotalEqualityDeletes }},
+ {"iceberg.commit.manifests.created", "Number of manifests created by commit", "", func(m metrics.CommitMetricsResult) *metrics.CounterResult { return m.ManifestsCreated }},
+ {"iceberg.commit.manifests.replaced", "Number of manifests replaced by commit", "", func(m metrics.CommitMetricsResult) *metrics.CounterResult { return m.ManifestsReplaced }},
+ {"iceberg.commit.manifests.kept", "Number of manifests kept by commit", "", func(m metrics.CommitMetricsResult) *metrics.CounterResult { return m.ManifestsKept }},
+ {"iceberg.commit.manifest_entries.processed", "Number of manifest entries processed by commit", "", func(m metrics.CommitMetricsResult) *metrics.CounterResult { return m.ManifestEntriesProcessed }},
+}
+
+// Reporter records iceberg metrics reports to OpenTelemetry instruments.
+type Reporter struct {
+ allow map[string]bool
+
+ scanPlanningDuration metric.Float64Histogram
+ scanCounters []metric.Int64Counter // parallel to scanCounterSpecs
+
+ commitDuration metric.Float64Histogram
+ commitCounters []metric.Int64Counter // parallel to commitCounterSpecs
+}
+
+var _ metrics.Reporter = (*Reporter)(nil)
+
+type config struct {
+ meter metric.Meter
+ allow map[string]bool
+}
+
+// Option configures a [Reporter].
+type Option func(*config)
+
+// WithMeter sets the Meter used to create instruments. By default the global
+// MeterProvider's Meter is used.
+func WithMeter(m metric.Meter) Option {
+ return func(c *config) { c.meter = m }
+}
+
+// WithAttributes sets the allowlist of attribute short names (AttrTableName,
+// AttrSchemaID, AttrOperation) attached to emitted metrics. Names not listed are
+// omitted, which is how cardinality is bounded. The default is AttrTableName and
+// AttrOperation; AttrSchemaID is opt-in. Passing no names emits metrics with no
+// attributes at all.
+func WithAttributes(short ...string) Option {
+ return func(c *config) {
+ c.allow = make(map[string]bool, len(short))
+ for _, s := range short {
+ c.allow[s] = true
+ }
+ }
+}
+
+// NewReporter creates an OpenTelemetry reporter, creating its instruments on the
+// configured (or global) Meter.
+func NewReporter(opts ...Option) (*Reporter, error) {
+ cfg := config{allow: map[string]bool{AttrTableName: true, AttrOperation: true}}
+ for _, o := range opts {
+ o(&cfg)
+ }
+ meter := cfg.meter
+ if meter == nil {
+ meter = otel.Meter(instrumentationName)
+ }
+
+ r := &Reporter{allow: cfg.allow}
+ var err error
+ mkHist := func(h *metric.Float64Histogram, name, desc string) {
+ if err != nil {
+ return
+ }
+ *h, err = meter.Float64Histogram(name, metric.WithDescription(desc), metric.WithUnit("ms"))
+ }
+
+ mkHist(&r.scanPlanningDuration, "iceberg.scan.planning.duration", "Time spent planning a table scan")
+ r.scanCounters = make([]metric.Int64Counter, len(scanCounterSpecs))
+ for i, s := range scanCounterSpecs {
+ if err != nil {
+ break
+ }
+ r.scanCounters[i], err = newCounter(meter, s.name, s.desc, s.unit)
+ }
+
+ mkHist(&r.commitDuration, "iceberg.commit.duration", "Time spent on commit operation")
+ r.commitCounters = make([]metric.Int64Counter, len(commitCounterSpecs))
+ for i, s := range commitCounterSpecs {
+ if err != nil {
+ break
+ }
+ r.commitCounters[i], err = newCounter(meter, s.name, s.desc, s.unit)
+ }
+
+ if err != nil {
+ return nil, err
+ }
+
+ return r, nil
+}
+
+// newCounter builds an Int64Counter, attaching a unit only when one is set
+// (counts are unit-less; byte counters carry "By").
+func newCounter(meter metric.Meter, name, desc, unit string) (metric.Int64Counter, error) {
+ opts := []metric.Int64CounterOption{metric.WithDescription(desc)}
+ if unit != "" {
+ opts = append(opts, metric.WithUnit(unit))
+ }
+
+ return meter.Int64Counter(name, opts...)
+}
+
+// Report implements [metrics.Reporter].
+func (r *Reporter) Report(ctx context.Context, report metrics.MetricsReport) {
+ switch rep := report.(type) {
+ case metrics.ScanReport:
+ r.reportScan(ctx, rep)
+ case *metrics.ScanReport:
+ if rep != nil {
+ r.reportScan(ctx, *rep)
+ }
+ case metrics.CommitReport:
+ r.reportCommit(ctx, rep)
+ case *metrics.CommitReport:
+ if rep != nil {
+ r.reportCommit(ctx, *rep)
+ }
+ }
+}
+
+// Close implements [io.Closer]. The reporter is stateless — it does not own the
+// OpenTelemetry SDK or MeterProvider (the host application does), so there is
+// nothing to release and Close always returns nil.
+func (r *Reporter) Close() error { return nil }
+
+func (r *Reporter) reportScan(ctx context.Context, sr metrics.ScanReport) {
+ var attrs []attribute.KeyValue
+ if r.allow[AttrTableName] {
+ attrs = append(attrs, attribute.String(keyTableName, sr.TableName))
+ }
+ if r.allow[AttrSchemaID] {
+ attrs = append(attrs, attribute.Int(keySchemaID, sr.SchemaID))
+ }
+ opt := metric.WithAttributes(attrs...)
+
+ m := sr.Metrics
+ recordDuration(ctx, r.scanPlanningDuration, m.TotalPlanningDuration, opt)
+ for i, s := range scanCounterSpecs {
+ addCounter(ctx, r.scanCounters[i], s.get(m), opt)
+ }
+}
+
+func (r *Reporter) reportCommit(ctx context.Context, cr metrics.CommitReport) {
+ var attrs []attribute.KeyValue
+ if r.allow[AttrTableName] {
+ attrs = append(attrs, attribute.String(keyTableName, cr.TableName))
+ }
+ if r.allow[AttrOperation] {
+ attrs = append(attrs, attribute.String(keyOperation, cr.Operation))
+ }
+ opt := metric.WithAttributes(attrs...)
+
+ m := cr.Metrics
+ recordDuration(ctx, r.commitDuration, m.TotalDuration, opt)
+ for i, s := range commitCounterSpecs {
+ addCounter(ctx, r.commitCounters[i], s.get(m), opt)
+ }
+}
+
+func addCounter(ctx context.Context, c metric.Int64Counter, cr *metrics.CounterResult, opt metric.AddOption) {
+ if cr != nil {
+ c.Add(ctx, cr.Value, opt)
+ }
+}
+
+// recordDuration records a timer's total duration in milliseconds. Durations are
+// emitted in nanoseconds (see metrics.NewNanosTimerResult), so convert to ms.
+func recordDuration(ctx context.Context, h metric.Float64Histogram, tr *metrics.TimerResult, opt metric.RecordOption) {
+ if tr != nil {
+ h.Record(ctx, float64(tr.TotalDuration)/1e6, opt)
+ }
+}
diff --git a/metrics/otel/reporter_test.go b/metrics/otel/reporter_test.go
new file mode 100644
index 000000000..f22c7de5b
--- /dev/null
+++ b/metrics/otel/reporter_test.go
@@ -0,0 +1,384 @@
+// 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 otel
+
+import (
+ "context"
+ "testing"
+
+ "github.com/apache/iceberg-go/metrics"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+ "go.opentelemetry.io/otel/attribute"
+ sdkmetric "go.opentelemetry.io/otel/sdk/metric"
+ "go.opentelemetry.io/otel/sdk/metric/metricdata"
+)
+
+func newTestReporter(t *testing.T, opts ...Option) (*Reporter, func() metricdata.ResourceMetrics) {
+ t.Helper()
+ reader := sdkmetric.NewManualReader()
+ mp := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader))
+ rep, err := NewReporter(append([]Option{WithMeter(mp.Meter("test"))}, opts...)...)
+ require.NoError(t, err)
+
+ return rep, func() metricdata.ResourceMetrics {
+ var rm metricdata.ResourceMetrics
+ require.NoError(t, reader.Collect(context.Background(), &rm))
+
+ return rm
+ }
+}
+
+func findMetric(t *testing.T, rm metricdata.ResourceMetrics, name string) metricdata.Metrics {
+ t.Helper()
+ for _, sm := range rm.ScopeMetrics {
+ for _, m := range sm.Metrics {
+ if m.Name == name {
+ return m
+ }
+ }
+ }
+ t.Fatalf("metric %q not emitted", name)
+
+ return metricdata.Metrics{}
+}
+
+func metricNames(rm metricdata.ResourceMetrics) map[string]bool {
+ names := map[string]bool{}
+ for _, sm := range rm.ScopeMetrics {
+ for _, m := range sm.Metrics {
+ names[m.Name] = true
+ }
+ }
+
+ return names
+}
+
+func sumValue(t *testing.T, m metricdata.Metrics) (int64, attribute.Set) {
+ t.Helper()
+ s, ok := m.Data.(metricdata.Sum[int64])
+ require.True(t, ok, "metric %q is not an int64 sum", m.Name)
+ require.Len(t, s.DataPoints, 1)
+
+ return s.DataPoints[0].Value, s.DataPoints[0].Attributes
+}
+
+func histogramSum(t *testing.T, m metricdata.Metrics) float64 {
+ t.Helper()
+ h, ok := m.Data.(metricdata.Histogram[float64])
+ require.True(t, ok, "metric %q is not a float64 histogram", m.Name)
+ require.Len(t, h.DataPoints, 1)
+
+ return h.DataPoints[0].Sum
+}
+
+func attrStr(t *testing.T, set attribute.Set, key string) string {
+ t.Helper()
+ v, ok := set.Value(attribute.Key(key))
+ require.True(t, ok, "attribute %q missing", key)
+
+ return v.AsString()
+}
+
+func attrInt(t *testing.T, set attribute.Set, key string) int64 {
+ t.Helper()
+ v, ok := set.Value(attribute.Key(key))
+ require.True(t, ok, "attribute %q missing", key)
+
+ return v.AsInt64()
+}
+
+// attributeKeys returns the union of attribute keys across every data point in
+// the collection, mirroring Java's per-test attribute-set assertions.
+func attributeKeys(rm metricdata.ResourceMetrics) map[string]bool {
+ keys := map[string]bool{}
+ for _, sm := range rm.ScopeMetrics {
+ for _, m := range sm.Metrics {
+ if s, ok := m.Data.(metricdata.Sum[int64]); ok {
+ for _, dp := range s.DataPoints {
+ for _, kv := range dp.Attributes.ToSlice() {
+ keys[string(kv.Key)] = true
+ }
+ }
+ }
+ }
+ }
+
+ return keys
+}
+
+// fullScanMetrics is a ScanMetricsResult with every field populated, used to
+// assert full instrument coverage.
+func fullScanMetrics() metrics.ScanMetricsResult {
+ c := func(v int64) *metrics.CounterResult { return metrics.NewCounterResult(metrics.UnitCount, v) }
+
+ return metrics.ScanMetricsResult{
+ TotalPlanningDuration: metrics.NewNanosTimerResult(1, 150_000_000), // 150ms
+ ResultDataFiles: c(10),
+ ResultDeleteFiles: c(2),
+ TotalDataManifests: c(8),
+ TotalDeleteManifests: c(4),
+ ScannedDataManifests: c(5),
+ SkippedDataManifests: c(3),
+ TotalFileSizeInBytes: metrics.NewCounterResult(metrics.UnitBytes, 1024000),
+ TotalDeleteFileSizeInBytes: metrics.NewCounterResult(metrics.UnitBytes, 2048),
+ SkippedDataFiles: c(7),
+ SkippedDeleteFiles: c(1),
+ ScannedDeleteManifests: c(3),
+ SkippedDeleteManifests: c(1),
+ IndexedDeleteFiles: c(6),
+ EqualityDeleteFiles: c(2),
+ PositionalDeleteFiles: c(4),
+ DVs: c(9),
+ }
+}
+
+// fullCommitMetrics is a CommitMetricsResult with every field populated.
+func fullCommitMetrics() metrics.CommitMetricsResult {
+ c := func(v int64) *metrics.CounterResult { return metrics.NewCounterResult(metrics.UnitCount, v) }
+
+ return metrics.CommitMetricsResult{
+ TotalDuration: metrics.NewNanosTimerResult(1, 200_000_000), // 200ms
+ Attempts: c(1),
+ AddedDataFiles: c(5),
+ RemovedDataFiles: c(2),
+ TotalDataFiles: c(15),
+ AddedDeleteFiles: c(3),
+ AddedEqualityDeleteFiles: c(1),
+ AddedPositionalDeleteFiles: c(2),
+ AddedDVs: c(4),
+ RemovedDeleteFiles: c(1),
+ RemovedEqualityDeleteFiles: c(0),
+ RemovedPositionalDeleteFiles: c(1),
+ RemovedDVs: c(0),
+ TotalDeleteFiles: c(8),
+ AddedRecords: c(1000),
+ RemovedRecords: c(50),
+ TotalRecords: c(5000),
+ AddedFilesSizeBytes: metrics.NewCounterResult(metrics.UnitBytes, 512000),
+ RemovedFilesSizeBytes: metrics.NewCounterResult(metrics.UnitBytes, 10000),
+ TotalFilesSizeBytes: metrics.NewCounterResult(metrics.UnitBytes, 2000000),
+ AddedPositionalDeletes: c(100),
+ RemovedPositionalDeletes: c(20),
+ TotalPositionalDeletes: c(300),
+ AddedEqualityDeletes: c(50),
+ RemovedEqualityDeletes: c(10),
+ TotalEqualityDeletes: c(150),
+ ManifestsCreated: c(3),
+ ManifestsReplaced: c(2),
+ ManifestsKept: c(7),
+ ManifestEntriesProcessed: c(500),
+ }
+}
+
+func TestReporterScanFullCoverage(t *testing.T) {
+ rep, collect := newTestReporter(t)
+ rep.Report(context.Background(), metrics.ScanReport{TableName: "db.t", SchemaID: 3, Metrics: fullScanMetrics()})
+
+ rm := collect()
+ names := metricNames(rm)
+
+ // The planning-duration histogram plus one counter per scan spec must all exist.
+ assert.True(t, names["iceberg.scan.planning.duration"], "planning duration missing")
+ for _, s := range scanCounterSpecs {
+ assert.True(t, names[s.name], "scan metric %q not emitted", s.name)
+ }
+
+ // Spot-check values and units across the data / delete / dv paths.
+ files, _ := sumValue(t, findMetric(t, rm, "iceberg.scan.result.data_files"))
+ assert.Equal(t, int64(10), files)
+ skippedDelManifests, _ := sumValue(t, findMetric(t, rm, "iceberg.scan.delete_manifests.skipped"))
+ assert.Equal(t, int64(1), skippedDelManifests)
+ dvs, _ := sumValue(t, findMetric(t, rm, "iceberg.scan.dvs"))
+ assert.Equal(t, int64(9), dvs)
+
+ bytesMetric := findMetric(t, rm, "iceberg.scan.file_size.bytes")
+ assert.Equal(t, "By", bytesMetric.Unit)
+ countMetric := findMetric(t, rm, "iceberg.scan.result.data_files")
+ assert.Empty(t, countMetric.Unit, "count instruments carry no unit")
+
+ assert.InDelta(t, 150.0, histogramSum(t, findMetric(t, rm, "iceberg.scan.planning.duration")), 0.0001)
+}
+
+func TestReporterCommitFullCoverage(t *testing.T) {
+ rep, collect := newTestReporter(t)
+ rep.Report(context.Background(), metrics.CommitReport{TableName: "db.t", Operation: "append", Metrics: fullCommitMetrics()})
+
+ rm := collect()
+ names := metricNames(rm)
+
+ assert.True(t, names["iceberg.commit.duration"], "commit duration missing")
+ for _, s := range commitCounterSpecs {
+ assert.True(t, names[s.name], "commit metric %q not emitted", s.name)
+ }
+
+ recs, _ := sumValue(t, findMetric(t, rm, "iceberg.commit.records.added"))
+ assert.Equal(t, int64(1000), recs)
+ eqAdded, _ := sumValue(t, findMetric(t, rm, "iceberg.commit.delete_files.equality.added"))
+ assert.Equal(t, int64(1), eqAdded)
+ manifestsKept, _ := sumValue(t, findMetric(t, rm, "iceberg.commit.manifests.kept"))
+ assert.Equal(t, int64(7), manifestsKept)
+
+ bytesMetric := findMetric(t, rm, "iceberg.commit.file_size.added_bytes")
+ assert.Equal(t, "By", bytesMetric.Unit)
+
+ assert.InDelta(t, 200.0, histogramSum(t, findMetric(t, rm, "iceberg.commit.duration")), 0.0001)
+}
+
+// TestReporterNilMetricsEmitNothing mirrors Java's testNullMetricsAreHandled: an
+// empty metrics result produces no time series.
+func TestReporterNilMetricsEmitNothing(t *testing.T) {
+ rep, collect := newTestReporter(t)
+ rep.Report(context.Background(), metrics.ScanReport{TableName: "db.t"})
+
+ assert.Empty(t, metricNames(collect()), "empty ScanMetricsResult must emit no metrics")
+}
+
+// TestReporterMultipleReports verifies counters accumulate across reports.
+func TestReporterMultipleReports(t *testing.T) {
+ rep, collect := newTestReporter(t)
+ for i := 0; i < 3; i++ {
+ rep.Report(context.Background(), metrics.CommitReport{
+ TableName: "db.t",
+ Operation: "append",
+ Metrics: metrics.CommitMetricsResult{
+ Attempts: metrics.NewCounterResult(metrics.UnitCount, 1),
+ AddedDataFiles: metrics.NewCounterResult(metrics.UnitCount, 10),
+ },
+ })
+ }
+
+ rm := collect()
+ files, _ := sumValue(t, findMetric(t, rm, "iceberg.commit.data_files.added"))
+ assert.Equal(t, int64(30), files)
+ attempts, _ := sumValue(t, findMetric(t, rm, "iceberg.commit.attempts"))
+ assert.Equal(t, int64(3), attempts)
+}
+
+// TestReporterDefaultAttributeSet asserts the default attribute set matches Java:
+// table-name + operation, with schema-id and snapshot-id excluded.
+func TestReporterDefaultAttributeSet(t *testing.T) {
+ rep, collect := newTestReporter(t)
+ rep.Report(context.Background(), metrics.ScanReport{
+ TableName: "db.t", SchemaID: 3,
+ Metrics: metrics.ScanMetricsResult{ResultDataFiles: metrics.NewCounterResult(metrics.UnitCount, 1)},
+ })
+ rep.Report(context.Background(), metrics.CommitReport{
+ TableName: "db.t", Operation: "append",
+ Metrics: metrics.CommitMetricsResult{Attempts: metrics.NewCounterResult(metrics.UnitCount, 1)},
+ })
+
+ keys := attributeKeys(collect())
+ assert.Contains(t, keys, keyTableName)
+ assert.Contains(t, keys, keyOperation)
+ assert.NotContains(t, keys, keySchemaID, "schema-id is opt-in, not in the default set")
+ assert.NotContains(t, keys, "iceberg.snapshot.id", "snapshot-id is never an attribute")
+}
+
+func TestReporterScanDefaultAttributes(t *testing.T) {
+ rep, collect := newTestReporter(t)
+ rep.Report(context.Background(), metrics.ScanReport{
+ TableName: "db.t", SchemaID: 3,
+ Metrics: metrics.ScanMetricsResult{ResultDataFiles: metrics.NewCounterResult(metrics.UnitCount, 5)},
+ })
+
+ _, attrs := sumValue(t, findMetric(t, collect(), "iceberg.scan.result.data_files"))
+ assert.Equal(t, "db.t", attrStr(t, attrs, keyTableName))
+ _, ok := attrs.Value(attribute.Key(keySchemaID))
+ assert.False(t, ok, "schema-id must be omitted by default")
+}
+
+func TestReporterAttributeAllowlistOptInSchemaID(t *testing.T) {
+ rep, collect := newTestReporter(t, WithAttributes(AttrTableName, AttrSchemaID, AttrOperation))
+ rep.Report(context.Background(), metrics.ScanReport{
+ TableName: "db.t", SchemaID: 7,
+ Metrics: metrics.ScanMetricsResult{ResultDataFiles: metrics.NewCounterResult(metrics.UnitCount, 1)},
+ })
+
+ _, attrs := sumValue(t, findMetric(t, collect(), "iceberg.scan.result.data_files"))
+ assert.Equal(t, int64(7), attrInt(t, attrs, keySchemaID))
+ assert.Equal(t, "db.t", attrStr(t, attrs, keyTableName))
+}
+
+func TestReporterAttributeAllowlistExcludesTableName(t *testing.T) {
+ rep, collect := newTestReporter(t, WithAttributes(AttrOperation))
+ rep.Report(context.Background(), metrics.CommitReport{
+ TableName: "db.t", Operation: "append",
+ Metrics: metrics.CommitMetricsResult{Attempts: metrics.NewCounterResult(metrics.UnitCount, 1)},
+ })
+
+ _, attrs := sumValue(t, findMetric(t, collect(), "iceberg.commit.attempts"))
+ assert.Equal(t, "append", attrStr(t, attrs, keyOperation))
+ _, ok := attrs.Value(attribute.Key(keyTableName))
+ assert.False(t, ok, "table-name must be omitted when not in the allowlist")
+}
+
+func TestReporterAttributeAllowlistNoneEmitsNoAttributes(t *testing.T) {
+ rep, collect := newTestReporter(t, WithAttributes())
+ rep.Report(context.Background(), metrics.CommitReport{
+ TableName: "db.t", Operation: "append",
+ Metrics: metrics.CommitMetricsResult{Attempts: metrics.NewCounterResult(metrics.UnitCount, 1)},
+ })
+
+ assert.Empty(t, attributeKeys(collect()), "empty allowlist must emit no attributes")
+}
+
+func TestReporterPointerReports(t *testing.T) {
+ rep, collect := newTestReporter(t)
+ rep.Report(context.Background(), &metrics.ScanReport{
+ TableName: "db.t",
+ Metrics: metrics.ScanMetricsResult{ResultDataFiles: metrics.NewCounterResult(metrics.UnitCount, 7)},
+ })
+ rep.Report(context.Background(), &metrics.CommitReport{
+ TableName: "db.t",
+ Metrics: metrics.CommitMetricsResult{AddedRecords: metrics.NewCounterResult(metrics.UnitCount, 9)},
+ })
+
+ rm := collect()
+ files, _ := sumValue(t, findMetric(t, rm, "iceberg.scan.result.data_files"))
+ assert.Equal(t, int64(7), files)
+ recs, _ := sumValue(t, findMetric(t, rm, "iceberg.commit.records.added"))
+ assert.Equal(t, int64(9), recs)
+}
+
+func TestReporterIgnoresNilReport(t *testing.T) {
+ rep, _ := newTestReporter(t)
+ assert.NotPanics(t, func() {
+ rep.Report(context.Background(), nil)
+ // Typed-nil pointers are non-nil interface values; they must be ignored,
+ // not dereferenced (mirrors metrics.isNilReport).
+ rep.Report(context.Background(), (*metrics.ScanReport)(nil))
+ rep.Report(context.Background(), (*metrics.CommitReport)(nil))
+ })
+}
+
+func TestReporterClose(t *testing.T) {
+ rep, _ := newTestReporter(t)
+ assert.NoError(t, rep.Close())
+}
+
+// TestReporterDefaultGlobalMeter exercises the WithMeter-unset path: with no SDK
+// registered the global provider yields no-op instruments, so construction
+// succeeds and Report is a silent no-op rather than a panic.
+func TestReporterDefaultGlobalMeter(t *testing.T) {
+ rep, err := NewReporter()
+ require.NoError(t, err)
+ assert.NotPanics(t, func() {
+ rep.Report(context.Background(), metrics.ScanReport{TableName: "db.t"})
+ })
+}