diff --git a/duckdbservice/commit_stats.go b/duckdbservice/commit_stats.go new file mode 100644 index 00000000..c30449a2 --- /dev/null +++ b/duckdbservice/commit_stats.go @@ -0,0 +1,257 @@ +package duckdbservice + +// Polls the posthog ducklake extension's ducklake_commit_stats() table +// function and re-exports its counters as the +// duckgres_worker_ducklake_commit_* Prometheus metrics (metrics.go). +// +// The function contract (shared with the extension): +// - Signature: ducklake_commit_stats() — no arguments. +// - Columns: catalog VARCHAR, stat VARCHAR, value BIGINT. +// - One row per (catalog, stat); values are process-global, cumulative, and +// monotonic for the life of the process. +// - Stat names: attempts, successes, retries_exhausted, nonretryable_errors, +// backoff_ms, total_commit_ms, and conflicts. with cause one of +// primary_key, unique, conflict, concurrent (an unknown cause from a newer +// extension is exported under its own cause label rather than dropped). +// +// The function only exists in ducklake extension builds that expose the +// commit retry loop counters — no currently-released extension binary does — +// so availability is probed once per DuckDB engine and, when absent, polling +// is disabled with a single info log. Everything degrades gracefully with the +// pinned extension. + +import ( + "context" + "database/sql" + "errors" + "log/slog" + "strings" + "time" + + "github.com/prometheus/client_golang/prometheus" +) + +const ( + commitStatsInterval = 30 * time.Second + commitStatsScrapeTimeout = 5 * time.Second + + commitStatsProbeQuery = "SELECT 1 FROM duckdb_functions() WHERE function_name = 'ducklake_commit_stats' LIMIT 1" + commitStatsQuery = "SELECT catalog, stat, value FROM ducklake_commit_stats()" + + commitStatsConflictPrefix = "conflicts." +) + +// commitStatRow is one (catalog, stat, value) row from ducklake_commit_stats(). +type commitStatRow struct { + Catalog string + Stat string + Value int64 +} + +type commitStatKey struct { + catalog string + stat string +} + +// commitStatsExporter converts the cumulative counters reported by +// ducklake_commit_stats() into Prometheus counter increments by tracking the +// last-seen value per (catalog, stat). It also caches the per-engine +// capability probe verdict. Only ever touched from the commitStatsLoop +// goroutine, so it needs no locking. The probe and query indirections default +// to the real SQL implementations; tests inject stubs. +type commitStatsExporter struct { + lastSeen map[commitStatKey]int64 + + // probedDB is the *sql.DB the capability probe last delivered a verdict + // for; available is that verdict. A different DB (engine replaced) is + // re-probed once. + probedDB *sql.DB + available bool + + probe func(ctx context.Context, db *sql.DB) (bool, error) + query func(ctx context.Context, db *sql.DB) ([]commitStatRow, error) +} + +func newCommitStatsExporter() *commitStatsExporter { + return &commitStatsExporter{ + lastSeen: make(map[commitStatKey]int64), + probe: probeCommitStatsFunction, + query: queryCommitStats, + } +} + +// delta returns the Prometheus counter increment for a newly observed +// cumulative value, updating the last-seen state. The values are monotonic +// per process and we poll the same process, so a decrease should never +// happen — but be defensive: a value below the last-seen one is treated as a +// fresh baseline and the current value is the delta. A negative delta is never +// returned (prometheus Counter.Add panics on negative values). +func (e *commitStatsExporter) delta(catalog, stat string, value int64) int64 { + key := commitStatKey{catalog: catalog, stat: stat} + if value < 0 { + // Contract violation — emit nothing and keep the existing baseline + // untouched: resetting it would make the next valid cumulative value + // count in full, double-exporting everything already recorded. + return 0 + } + last := e.lastSeen[key] + e.lastSeen[key] = value + if delta := value - last; delta >= 0 { + return delta + } + return value +} + +// record applies one poll's rows to the Prometheus counters. Unknown stat +// names are ignored (a newer extension may add stats this build doesn't map). +// A zero delta still touches the series so every known series exists from the +// first successful poll onward. +func (e *commitStatsExporter) record(rows []commitStatRow) { + for _, row := range rows { + counter, cause, ok := commitStatCounter(row.Stat) + if !ok { + continue + } + delta := float64(e.delta(row.Catalog, row.Stat, row.Value)) + if cause != "" { + counter.WithLabelValues(row.Catalog, cause).Add(delta) + } else { + counter.WithLabelValues(row.Catalog).Add(delta) + } + } +} + +// commitStatCounter maps a ducklake_commit_stats() stat name to its Prometheus +// counter. For conflicts.* stats the second return is the cause label value; +// it is empty for all other stats. ok is false for stat names this build does +// not know how to export. +func commitStatCounter(stat string) (counter *prometheus.CounterVec, cause string, ok bool) { + if c, found := strings.CutPrefix(stat, commitStatsConflictPrefix); found { + if c == "" { + return nil, "", false + } + return ducklakeCommitConflictsTotal, c, true + } + switch stat { + case "attempts": + return ducklakeCommitAttemptsTotal, "", true + case "successes": + return ducklakeCommitSuccessesTotal, "", true + case "retries_exhausted": + return ducklakeCommitRetriesExhaustedTotal, "", true + case "nonretryable_errors": + return ducklakeCommitNonretryableErrorsTotal, "", true + case "backoff_ms": + return ducklakeCommitBackoffMsTotal, "", true + case "total_commit_ms": + return ducklakeCommitDurationMsTotal, "", true + } + return nil, "", false +} + +// ensureAvailable answers whether ducklake_commit_stats() exists on this +// engine, probing at most once per *sql.DB. Absence is the common case today +// (no released ducklake extension binary provides the function), so it is +// logged once at info and polling stays disabled for that engine. A probe +// error records no verdict — the next tick retries. +func (e *commitStatsExporter) ensureAvailable(ctx context.Context, db *sql.DB) bool { + if db == e.probedDB { + return e.available + } + available, err := e.probe(ctx, db) + if err != nil { + slog.Debug("DuckLake commit-stats capability probe failed; will retry.", "error", err) + return false + } + e.probedDB = db + e.available = available + if available { + slog.Info("ducklake_commit_stats() detected; exporting DuckLake commit-loop metrics.") + } else { + slog.Info("ducklake_commit_stats() not provided by the loaded ducklake extension; DuckLake commit-loop metrics disabled.") + } + return available +} + +func probeCommitStatsFunction(ctx context.Context, db *sql.DB) (bool, error) { + var one int + err := db.QueryRowContext(ctx, commitStatsProbeQuery).Scan(&one) + if errors.Is(err, sql.ErrNoRows) { + return false, nil + } + if err != nil { + return false, err + } + return true, nil +} + +func queryCommitStats(ctx context.Context, db *sql.DB) ([]commitStatRow, error) { + rows, err := db.QueryContext(ctx, commitStatsQuery) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + + var out []commitStatRow + for rows.Next() { + var row commitStatRow + if err := rows.Scan(&row.Catalog, &row.Stat, &row.Value); err != nil { + return nil, err + } + out = append(out, row) + } + return out, rows.Err() +} + +// commitStatsLoop periodically polls ducklake_commit_stats() and re-exports +// the counters as Prometheus metrics. Same lifecycle as metadataMetricsLoop: +// started by NewDuckDBService, stops on pool close, and every tick is +// best-effort — a failure is logged and the next tick retries; the loop never +// holds pool locks across a query and never blocks user queries. +func (p *SessionPool) commitStatsLoop() { + exporter := newCommitStatsExporter() + ticker := time.NewTicker(commitStatsInterval) + defer ticker.Stop() + + for { + select { + case <-ticker.C: + p.scrapeCommitStats(exporter) + case <-p.stopCh: + return + } + } +} + +// scrapeCommitStats runs one poll. Prefer controlDB (side connection sharing +// the same DuckDB instance with its own pool) so a long-running user query on +// the main DB does not queue the scrape; fall back to the activation DB only +// when controlDB has not been wired (test paths). Unlike scrapeMetadataMetrics +// this does not require an activation: the counters are process-global to the +// ducklake extension and already carry the catalog name. +func (p *SessionPool) scrapeCommitStats(e *commitStatsExporter) { + p.mu.RLock() + db := p.controlDB + p.mu.RUnlock() + if db == nil { + if act := p.currentActivation(); act != nil { + db = act.db + } + } + if db == nil { + return + } + + ctx, cancel := context.WithTimeout(context.Background(), commitStatsScrapeTimeout) + defer cancel() + + if !e.ensureAvailable(ctx, db) { + return + } + rows, err := e.query(ctx, db) + if err != nil { + slog.Debug("Failed to poll ducklake_commit_stats(); will retry.", "error", err) + return + } + e.record(rows) +} diff --git a/duckdbservice/commit_stats_test.go b/duckdbservice/commit_stats_test.go new file mode 100644 index 00000000..8d0c588f --- /dev/null +++ b/duckdbservice/commit_stats_test.go @@ -0,0 +1,236 @@ +package duckdbservice + +import ( + "context" + "database/sql" + "testing" + "time" + + _ "github.com/duckdb/duckdb-go/v2" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/testutil" +) + +func TestCommitStatsExporterDelta(t *testing.T) { + e := newCommitStatsExporter() + + steps := []struct { + name string + catalog string + stat string + value int64 + expected int64 + }{ + {"first observation counts in full", "lake_a", "attempts", 5, 5}, + {"monotonic increase yields diff", "lake_a", "attempts", 12, 7}, + {"unchanged yields zero", "lake_a", "attempts", 12, 0}, + {"decrease treats current value as delta", "lake_a", "attempts", 3, 3}, + {"recovers after decrease", "lake_a", "attempts", 10, 7}, + {"negative current value yields zero and keeps baseline", "lake_a", "attempts", -4, 0}, + {"post-glitch value counts only the increase over the kept baseline", "lake_a", "attempts", 11, 1}, + {"stats tracked independently", "lake_a", "successes", 4, 4}, + {"catalogs tracked independently", "lake_b", "attempts", 9, 9}, + } + + for _, step := range steps { + if got := e.delta(step.catalog, step.stat, step.value); got != step.expected { + t.Errorf("%s: delta(%q, %q, %d) = %d, want %d", + step.name, step.catalog, step.stat, step.value, got, step.expected) + } + } +} + +func TestCommitStatCounterMapping(t *testing.T) { + tests := []struct { + stat string + counter *prometheus.CounterVec + cause string + ok bool + }{ + {"attempts", ducklakeCommitAttemptsTotal, "", true}, + {"successes", ducklakeCommitSuccessesTotal, "", true}, + {"retries_exhausted", ducklakeCommitRetriesExhaustedTotal, "", true}, + {"nonretryable_errors", ducklakeCommitNonretryableErrorsTotal, "", true}, + {"backoff_ms", ducklakeCommitBackoffMsTotal, "", true}, + {"total_commit_ms", ducklakeCommitDurationMsTotal, "", true}, + {"conflicts.primary_key", ducklakeCommitConflictsTotal, "primary_key", true}, + {"conflicts.unique", ducklakeCommitConflictsTotal, "unique", true}, + {"conflicts.conflict", ducklakeCommitConflictsTotal, "conflict", true}, + {"conflicts.concurrent", ducklakeCommitConflictsTotal, "concurrent", true}, + // Forward compatibility: a new conflict cause from a newer extension + // is exported under its own cause label rather than dropped. + {"conflicts.future_cause", ducklakeCommitConflictsTotal, "future_cause", true}, + {"conflicts.", nil, "", false}, + {"unknown_stat", nil, "", false}, + {"", nil, "", false}, + } + + for _, tt := range tests { + t.Run(tt.stat, func(t *testing.T) { + counter, cause, ok := commitStatCounter(tt.stat) + if ok != tt.ok { + t.Fatalf("commitStatCounter(%q) ok = %v, want %v", tt.stat, ok, tt.ok) + } + if counter != tt.counter { + t.Errorf("commitStatCounter(%q) returned unexpected counter", tt.stat) + } + if cause != tt.cause { + t.Errorf("commitStatCounter(%q) cause = %q, want %q", tt.stat, cause, tt.cause) + } + }) + } +} + +func TestCommitStatsExporterRecord(t *testing.T) { + // Package-level counters live in the default registry, so this test uses + // a catalog label no other test touches and asserts absolute values. + const catalog = "record_test_lake" + e := newCommitStatsExporter() + + e.record([]commitStatRow{ + {Catalog: catalog, Stat: "attempts", Value: 10}, + {Catalog: catalog, Stat: "successes", Value: 8}, + {Catalog: catalog, Stat: "total_commit_ms", Value: 1500}, + {Catalog: catalog, Stat: "conflicts.primary_key", Value: 2}, + {Catalog: catalog, Stat: "not_a_known_stat", Value: 99}, + }) + e.record([]commitStatRow{ + {Catalog: catalog, Stat: "attempts", Value: 14}, + {Catalog: catalog, Stat: "successes", Value: 8}, + {Catalog: catalog, Stat: "total_commit_ms", Value: 400}, // went DOWN → current value is the delta + {Catalog: catalog, Stat: "conflicts.primary_key", Value: 3}, + }) + + assertCounter := func(name string, c prometheus.Counter, want float64) { + t.Helper() + if got := testutil.ToFloat64(c); got != want { + t.Errorf("%s = %v, want %v", name, got, want) + } + } + assertCounter("attempts", ducklakeCommitAttemptsTotal.WithLabelValues(catalog), 14) + assertCounter("successes", ducklakeCommitSuccessesTotal.WithLabelValues(catalog), 8) + assertCounter("duration_ms", ducklakeCommitDurationMsTotal.WithLabelValues(catalog), 1900) + assertCounter("conflicts.primary_key", ducklakeCommitConflictsTotal.WithLabelValues(catalog, "primary_key"), 3) +} + +func TestProbeCommitStatsFunction(t *testing.T) { + db, err := sql.Open("duckdb", "") + if err != nil { + t.Fatalf("open duckdb: %v", err) + } + defer func() { _ = db.Close() }() + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + // Stock DuckDB (like every currently-released ducklake extension binary) + // does not provide ducklake_commit_stats(). + available, err := probeCommitStatsFunction(ctx, db) + if err != nil { + t.Fatalf("probe: %v", err) + } + if available { + t.Fatal("probe reported ducklake_commit_stats() available on stock DuckDB") + } + + // A table macro of the same name is indistinguishable from the extension's + // table function for both the probe and the scrape query. + if _, err := db.ExecContext(ctx, `CREATE MACRO ducklake_commit_stats() AS TABLE + SELECT * FROM (VALUES + ('ducklake', 'attempts', 7::BIGINT), + ('ducklake', 'conflicts.concurrent', 1::BIGINT) + ) AS t(catalog, stat, value)`); err != nil { + t.Fatalf("create macro: %v", err) + } + + available, err = probeCommitStatsFunction(ctx, db) + if err != nil { + t.Fatalf("probe after macro: %v", err) + } + if !available { + t.Fatal("probe did not find ducklake_commit_stats()") + } + + rows, err := queryCommitStats(ctx, db) + if err != nil { + t.Fatalf("query commit stats: %v", err) + } + want := []commitStatRow{ + {Catalog: "ducklake", Stat: "attempts", Value: 7}, + {Catalog: "ducklake", Stat: "conflicts.concurrent", Value: 1}, + } + if len(rows) != len(want) { + t.Fatalf("got %d rows, want %d: %+v", len(rows), len(want), rows) + } + for i, w := range want { + if rows[i] != w { + t.Errorf("row %d = %+v, want %+v", i, rows[i], w) + } + } +} + +func TestScrapeCommitStatsDegradesWhenFunctionAbsent(t *testing.T) { + db, err := sql.Open("duckdb", "") + if err != nil { + t.Fatalf("open duckdb: %v", err) + } + defer func() { _ = db.Close() }() + + p := &SessionPool{controlDB: db} + e := newCommitStatsExporter() + + queries := 0 + e.query = func(ctx context.Context, db *sql.DB) ([]commitStatRow, error) { + queries++ + return nil, nil + } + probes := 0 + realProbe := e.probe + e.probe = func(ctx context.Context, db *sql.DB) (bool, error) { + probes++ + return realProbe(ctx, db) + } + + p.scrapeCommitStats(e) + p.scrapeCommitStats(e) + + if probes != 1 { + t.Errorf("probe ran %d times, want 1 (once per engine)", probes) + } + if queries != 0 { + t.Errorf("query ran %d times, want 0 when the function is absent", queries) + } +} + +func TestScrapeCommitStatsPollsWhenFunctionPresent(t *testing.T) { + db, err := sql.Open("duckdb", "") + if err != nil { + t.Fatalf("open duckdb: %v", err) + } + defer func() { _ = db.Close() }() + + p := &SessionPool{controlDB: db} + e := newCommitStatsExporter() + e.probe = func(ctx context.Context, db *sql.DB) (bool, error) { return true, nil } + e.query = func(ctx context.Context, db *sql.DB) ([]commitStatRow, error) { + return []commitStatRow{{Catalog: "scrape_test_lake", Stat: "attempts", Value: 3}}, nil + } + + p.scrapeCommitStats(e) + p.scrapeCommitStats(e) + + got := testutil.ToFloat64(ducklakeCommitAttemptsTotal.WithLabelValues("scrape_test_lake")) + if got != 3 { + t.Errorf("attempts counter = %v, want 3 (cumulative value unchanged across polls)", got) + } +} + +func TestScrapeCommitStatsSkipsWithoutDB(t *testing.T) { + p := &SessionPool{} + e := newCommitStatsExporter() + e.probe = func(ctx context.Context, db *sql.DB) (bool, error) { + t.Fatal("probe must not run without a DB") + return false, nil + } + p.scrapeCommitStats(e) +} diff --git a/duckdbservice/metrics.go b/duckdbservice/metrics.go index 52efac80..5238a892 100644 --- a/duckdbservice/metrics.go +++ b/duckdbservice/metrics.go @@ -26,3 +26,40 @@ var ( Help: "Total number of DuckLake transaction conflicts where all retries were exhausted (worker)", }) ) + +// Commit-loop stats re-exported from the ducklake extension's +// ducklake_commit_stats() table function (see commit_stats.go). These see +// INSIDE the extension's internal commit retry loop, unlike the +// duckgres_worker_ducklake_conflict_* counters above, which only see conflicts +// that escape it. Labelled by DuckLake catalog name; conflicts additionally by +// conflict cause. +var ( + ducklakeCommitAttemptsTotal = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "duckgres_worker_ducklake_commit_attempts_total", + Help: "Total DuckLake commit attempts inside the ducklake extension's commit retry loop (worker)", + }, []string{"catalog"}) + ducklakeCommitSuccessesTotal = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "duckgres_worker_ducklake_commit_successes_total", + Help: "Total DuckLake commits that succeeded inside the ducklake extension's commit retry loop (worker)", + }, []string{"catalog"}) + ducklakeCommitRetriesExhaustedTotal = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "duckgres_worker_ducklake_commit_retries_exhausted_total", + Help: "Total DuckLake commits that exhausted the ducklake extension's internal retries (worker)", + }, []string{"catalog"}) + ducklakeCommitNonretryableErrorsTotal = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "duckgres_worker_ducklake_commit_nonretryable_errors_total", + Help: "Total DuckLake commit failures the ducklake extension classified as non-retryable (worker)", + }, []string{"catalog"}) + ducklakeCommitBackoffMsTotal = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "duckgres_worker_ducklake_commit_backoff_ms_total", + Help: "Total milliseconds spent backing off between DuckLake commit retries inside the ducklake extension (worker)", + }, []string{"catalog"}) + ducklakeCommitDurationMsTotal = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "duckgres_worker_ducklake_commit_duration_ms_total", + Help: "Total milliseconds spent in DuckLake commits inside the ducklake extension's commit loop (worker)", + }, []string{"catalog"}) + ducklakeCommitConflictsTotal = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "duckgres_worker_ducklake_commit_conflicts_total", + Help: "Total DuckLake commit conflicts observed inside the ducklake extension's commit retry loop, by cause (worker)", + }, []string{"catalog", "cause"}) +) diff --git a/duckdbservice/service.go b/duckdbservice/service.go index 0f976608..702a8125 100644 --- a/duckdbservice/service.go +++ b/duckdbservice/service.go @@ -514,6 +514,7 @@ func NewDuckDBService(cfg ServiceConfig) *DuckDBService { pool.activateTenantFunc = pool.activateTenant go pool.reapLoop() go pool.metadataMetricsLoop() + go pool.commitStatsLoop() return &DuckDBService{ cfg: cfg,