From eb0aa5814fd9a87be43e675aa6f0b0e69c9a79d6 Mon Sep 17 00:00:00 2001 From: Hai He Date: Tue, 18 Aug 2026 14:48:30 +0800 Subject: [PATCH 1/3] feat: add klio.plugin.wal.restore_duration end-to-end metric The CNPG-I plugin's RESTORE_WAL path measured its end-to-end duration but only logged it. Expose it as an OTel histogram (klio.plugin.wal.restore_duration, ns, per-file buckets) tagged with outcome, cache_hit, tier and cluster_name. This is the latency PostgreSQL actually experiences when it asks for a WAL segment. In a replica cluster whose designated primary replicates from an external Klio source, it is the speed of the replication path itself and so drives replica lag; during recovery it drives how fast a cluster catches up. Neither is visible today, and the server-side get_duration cannot stand in for it: that times only one WAL.Get call, while prefetch cache hits are served from the local spool and never reach the server at all. cache_hit and tier are threaded up from the prefetcher through restoreWAL via a small restoreOutcome; Restore records on every exit path via defer so failures are measured too. Assisted-by: Claude Signed-off-by: Hai He --- core/internal/cnpgi/metrics.go | 38 +++++++++++++ core/internal/cnpgi/metrics_test.go | 1 + core/internal/cnpgi/prefetcher.go | 45 ++++++++++----- core/internal/cnpgi/prefetcher_test.go | 28 ++++++++++ core/internal/cnpgi/wal.go | 49 ++++++++++++---- core/internal/cnpgi/wal_test.go | 56 +++++++++++++++++++ core/internal/opentelemetry/attributes.go | 34 +++++++++++ core/internal/opentelemetry/catalog.go | 33 +++++++++++ core/internal/opentelemetry/metrics.go | 1 + .../opentelemetry/wal_duration_test.go | 40 +++++++++++++ 10 files changed, 300 insertions(+), 25 deletions(-) diff --git a/core/internal/cnpgi/metrics.go b/core/internal/cnpgi/metrics.go index fe0141f8..36f8be99 100644 --- a/core/internal/cnpgi/metrics.go +++ b/core/internal/cnpgi/metrics.go @@ -53,6 +53,44 @@ func recordBackupSuccess(ctx context.Context, duration time.Duration) { metric.WithAttributes(opentelemetry.OutcomeSuccess.Attribute())) } +// recordWalRestore records the end-to-end duration of one plugin WAL restore, +// tagged with outcome, cache_hit, tier and cluster_name. A restore that fails +// early knows neither the tier nor the cluster, so both fall back to "unknown" +// rather than an empty attribute value: an empty label would add a second, +// near-invisible series to every per-tier or per-cluster panel. +func recordWalRestore( + ctx context.Context, + duration time.Duration, + success bool, + info restoreOutcome, + clusterName string, +) { + outcome := opentelemetry.OutcomeSuccess + if !success { + outcome = opentelemetry.OutcomeFailure + } + + restoreTier := info.tier + if restoreTier == "" { + restoreTier = tierUnknown + } + if clusterName == "" { + clusterName = unknownAttributeValue + } + + opentelemetry.PluginWal.RestoreDuration.Record(ctx, duration.Nanoseconds(), + metric.WithAttributes( + outcome.Attribute(), + opentelemetry.CacheHitOf(info.cacheHit).Attribute(), + opentelemetry.AttributeKeyTier.Of(string(restoreTier)), + opentelemetry.AttributeKeyClusterName.Of(clusterName), + )) +} + +// unknownAttributeValue tags an attribute whose real value was not yet known +// when the metric was recorded. +const unknownAttributeValue = "unknown" + // recordBackupFailure records a failed backup. func recordBackupFailure(ctx context.Context, duration time.Duration, err error) { category := classifyRunBackupError(ctx, err) diff --git a/core/internal/cnpgi/metrics_test.go b/core/internal/cnpgi/metrics_test.go index 666e7e5f..395cd2de 100644 --- a/core/internal/cnpgi/metrics_test.go +++ b/core/internal/cnpgi/metrics_test.go @@ -52,6 +52,7 @@ func setupTestMeter(t *testing.T) *sdkmetric.ManualReader { }) opentelemetry.InitPluginBackupMetrics() + opentelemetry.InitPluginWalMetrics() return reader } diff --git a/core/internal/cnpgi/prefetcher.go b/core/internal/cnpgi/prefetcher.go index 0f7105a5..e917ba3a 100644 --- a/core/internal/cnpgi/prefetcher.go +++ b/core/internal/cnpgi/prefetcher.go @@ -73,6 +73,15 @@ type walEntry struct { isPrefetch bool // true if this was a speculative prefetch, false if PG requested it } +// isReadyPrefetch reports whether this entry is a speculative prefetch that has +// already finished downloading, the only case a restore can be served straight +// from the spool. A prefetch still in flight is not a hit: the caller has to +// wait for its download just as it would for its own. Callers must hold the +// prefetcher lock, since it reads state. +func (e *walEntry) isReadyPrefetch() bool { + return e.isPrefetch && e.state == walStateReady +} + // walPrefetcher manages prefetching of WAL files for faster recovery. type walPrefetcher struct { mu sync.Mutex @@ -120,12 +129,14 @@ func newWALPrefetcher( } // Request retrieves a WAL file, using the prefetch cache if available. -// It also triggers prefetching of subsequent WAL files. -func (p *walPrefetcher) Request(ctx context.Context, walName, targetPath string) error { +// It also triggers prefetching of subsequent WAL files. The returned bool +// reports whether the WAL was served from the prefetch spool (a cache hit); +// it is only meaningful when the error is nil. +func (p *walPrefetcher) Request(ctx context.Context, walName, targetPath string) (bool, error) { contextLogger := log.FromContext(ctx).WithValues("walName", walName) // Try to get complete WAL from cache or download. - err := p.getCompleteWAL(ctx, walName, targetPath) + cacheHit, err := p.getCompleteWAL(ctx, walName, targetPath) if err == nil { // Success - trigger prefetch of next N complete WALs. p.mu.Lock() @@ -136,24 +147,25 @@ func (p *walPrefetcher) Request(ctx context.Context, walName, targetPath string) p.triggerPrefetch(walName) } - return nil + return cacheHit, nil } if !errors.Is(err, errWALNotFound) { - return err + return false, err } // Only a bare WAL segment can have a .partial variant, so don't fabricate a // nonsensical ".partial" request for a history or backup-label file: // report it as missing and let the caller move on. if !canHavePartial(walName) { - return err + return false, err } // Complete WAL not found - try partial (direct to target, no cache). contextLogger.Debug("Complete WAL not found, trying partial") - return p.getPartialWAL(ctx, walName, targetPath) + // Partial WALs are never cached, so this is never a cache hit. + return false, p.getPartialWAL(ctx, walName, targetPath) } // canHavePartial reports whether walName could have a .partial variant. Only a @@ -172,10 +184,13 @@ func (p *walPrefetcher) Close() error { return p.downloadPool.Wait() } -// getCompleteWAL retrieves a complete WAL file from cache or downloads it. +// getCompleteWAL retrieves a complete WAL file from cache or downloads it. The +// returned bool reports whether the file was served from a speculative prefetch +// already waiting in the spool (a cache hit); it is only meaningful when the +// error is nil. A rename fallback to a direct download is not a cache hit. // //nolint:cyclop // complexity is slightly over limit but refactoring would hurt readability -func (p *walPrefetcher) getCompleteWAL(ctx context.Context, walName, targetPath string) error { +func (p *walPrefetcher) getCompleteWAL(ctx context.Context, walName, targetPath string) (bool, error) { contextLogger := log.FromContext(ctx).WithValues("walName", walName) p.mu.Lock() @@ -190,7 +205,7 @@ func (p *walPrefetcher) getCompleteWAL(ctx context.Context, walName, targetPath } // A cache hit is when we have a prefetched entry that's already ready. - prefetchHit := exists && entry.isPrefetch && entry.state == walStateReady + prefetchHit := exists && entry.isReadyPrefetch() if !exists { // Not prefetched - start download now (direct request from PG). @@ -206,11 +221,11 @@ func (p *walPrefetcher) getCompleteWAL(ctx context.Context, walName, targetPath select { case <-entry.done: case <-ctx.Done(): - return ctx.Err() + return false, ctx.Err() } if entry.err != nil { - return entry.err + return false, entry.err } // Rename from spool to target (atomic on same filesystem). @@ -227,14 +242,14 @@ func (p *walPrefetcher) getCompleteWAL(ctx context.Context, walName, targetPath p.cleanupEntry(walName) _ = os.Remove(entry.spoolPath) - // Download directly to target. - return p.downloadDirect(ctx, walName, targetPath) + // Download directly to target - no longer a cache hit. + return false, p.downloadDirect(ctx, walName, targetPath) } // Cleanup entry from map (file already moved). p.cleanupEntry(walName) - return nil + return prefetchHit, nil } // downloadWALToFile downloads a WAL file to the specified path. diff --git a/core/internal/cnpgi/prefetcher_test.go b/core/internal/cnpgi/prefetcher_test.go index 5807f83a..a1bbac98 100644 --- a/core/internal/cnpgi/prefetcher_test.go +++ b/core/internal/cnpgi/prefetcher_test.go @@ -389,6 +389,34 @@ func TestTriggerPrefetch(t *testing.T) { }) } +// TestIsReadyPrefetch checks which entries count as a prefetch cache hit. Only +// a speculative prefetch that already finished downloading qualifies: an +// in-flight prefetch makes the caller wait for the download, and an entry the +// caller started itself was never a hit to begin with. +func TestIsReadyPrefetch(t *testing.T) { + tests := []struct { + name string + isPrefetch bool + state walState + want bool + }{ + {name: "prefetch finished is a hit", isPrefetch: true, state: walStateReady, want: true}, + {name: "prefetch still downloading is not a hit", isPrefetch: true, state: walStateDownloading, want: false}, + {name: "failed prefetch is not a hit", isPrefetch: true, state: walStateFailed, want: false}, + {name: "direct download ready is not a hit", isPrefetch: false, state: walStateReady, want: false}, + {name: "direct download in flight is not a hit", isPrefetch: false, state: walStateDownloading, want: false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + entry := &walEntry{isPrefetch: tt.isPrefetch, state: tt.state} + if got := entry.isReadyPrefetch(); got != tt.want { + t.Errorf("isReadyPrefetch() = %v, want %v", got, tt.want) + } + }) + } +} + func TestWalState(t *testing.T) { // Verify state constants have expected values. assert.Equal(t, walStateDownloading, walState(0)) diff --git a/core/internal/cnpgi/wal.go b/core/internal/cnpgi/wal.go index 8365870f..7817605c 100644 --- a/core/internal/cnpgi/wal.go +++ b/core/internal/cnpgi/wal.go @@ -48,6 +48,9 @@ type tier string const ( tier1 tier = "tier1" tier2 tier = "tier2" + // tierUnknown tags a restore that failed before any tier served it, so the + // metric never carries an empty attribute value. + tierUnknown tier = "unknown" ) type walServiceImplementation struct { @@ -97,6 +100,18 @@ func (w *walServiceImplementation) Restore( walName := request.GetSourceWalName() destinationPath := request.GetDestinationFileName() + // Record the end-to-end restore duration on every exit path. The + // closure reads the final values of success/info/clusterName, so failures — + // including the fast validation bail-outs below — are measured too. + var ( + success bool + info restoreOutcome + clusterName string + ) + defer func() { + recordWalRestore(ctx, time.Since(startCall), success, info, clusterName) + }() + if walName == "" || destinationPath == "" { contextLogger.Warning("WAL restore operation failed. WAL name and destination file name must be specified") return nil, errors.New("source WAL name and destination file name must be provided") @@ -107,6 +122,7 @@ func (w *walServiceImplementation) Restore( if err := json.Unmarshal(request.GetClusterDefinition(), &cluster); err != nil { return nil, fmt.Errorf("failed to unmarshal cluster definition: %w", err) } + clusterName = cluster.Name podName, ok := os.LookupEnv("POD_NAME") // Ensure PODNAME is set in the environment if !ok { return nil, errors.New("POD_NAME environment variable is not set") @@ -120,7 +136,7 @@ func (w *walServiceImplementation) Restore( return nil, errors.New("no WAL repository found for the cluster") } - err = w.restoreWAL(ctx, walName, destinationPath, confPath) + info, err = w.restoreWAL(ctx, walName, destinationPath, confPath) if errors.Is(err, errWALNotFound) { return &wal.WALRestoreResult{}, status.Errorf(codes.NotFound, "WAL file not found: %q", walName) } @@ -128,24 +144,34 @@ func (w *walServiceImplementation) Restore( return nil, err } + success = true contextLogger.Info("WAL.Restore", "walName", request.GetSourceWalName(), "duration", time.Since(startCall)) return &wal.WALRestoreResult{}, nil } +// restoreOutcome carries the observable facts about a completed restore that +// are only known deep in the restore path, so Restore can tag its end-to-end +// duration metric. On failure the fields hold whatever was known so far (tier +// is the last one attempted; cacheHit is false). +type restoreOutcome struct { + tier tier + cacheHit bool +} + func (w *walServiceImplementation) restoreWAL( ctx context.Context, walName, destinationPath string, configPath string, -) error { +) (restoreOutcome, error) { cfg, err := config.NewFromFile(afero.NewOsFs(), configPath) if err != nil { - return fmt.Errorf("while loading configuration from file %q: %w", configPath, err) + return restoreOutcome{}, fmt.Errorf("while loading configuration from file %q: %w", configPath, err) } tiers := availableTiers(cfg) if len(tiers) == 0 { - return errors.New("no WAL tier configured") + return restoreOutcome{}, errors.New("no WAL tier configured") } // Try the previously-successful tier first, when both are available. @@ -155,20 +181,20 @@ func (w *walServiceImplementation) restoreWAL( } for _, t := range tiers { - err := w.mgr.restoreWAL(ctx, walRestoreOptions{ + cacheHit, err := w.mgr.restoreWAL(ctx, walRestoreOptions{ configFile: configPath, targetTier: t, }, walName, destinationPath) if err == nil { w.currentTier.Store(t) - return nil + return restoreOutcome{tier: t, cacheHit: cacheHit}, nil } if !errors.Is(err, errWALNotFound) { - return err + return restoreOutcome{tier: t}, err } } - return errWALNotFound + return restoreOutcome{}, errWALNotFound } // availableTiers returns the tiers the user has opted in to as recovery @@ -297,15 +323,18 @@ func (mgr *grpcClientManager) setupSpoolDir(ctx context.Context, opts walRestore return spoolDir, nil } +// restoreWAL restores a single WAL file via the given tier's client. The +// returned bool reports whether the file was served from the prefetch spool +// (a cache hit); it is only meaningful when the error is nil. func (mgr *grpcClientManager) restoreWAL( ctx context.Context, opts walRestoreOptions, walName string, targetFileName string, -) error { +) (bool, error) { client, err := mgr.getClient(ctx, opts) if err != nil { - return err + return false, err } return client.prefetcher.Request(ctx, walName, targetFileName) diff --git a/core/internal/cnpgi/wal_test.go b/core/internal/cnpgi/wal_test.go index f14d78c2..d1248c41 100644 --- a/core/internal/cnpgi/wal_test.go +++ b/core/internal/cnpgi/wal_test.go @@ -27,6 +27,12 @@ import ( "strings" "testing" + "github.com/cloudnative-pg/cnpg-i/pkg/wal" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/sdk/metric/metricdata" + + "github.com/cloudnative-pg/klio/core/internal/opentelemetry" "github.com/cloudnative-pg/klio/core/pkg/config" ) @@ -126,6 +132,56 @@ func TestAvailableTiers(t *testing.T) { } } +// findInt64HistogramDataPoints returns the data points of the named Int64 +// Histogram instrument, or nil when the instrument recorded nothing. +func findInt64HistogramDataPoints( + rm metricdata.ResourceMetrics, name string, +) []metricdata.HistogramDataPoint[int64] { + for _, sm := range rm.ScopeMetrics { + for _, m := range sm.Metrics { + if m.Name == name { + if h, ok := m.Data.(metricdata.Histogram[int64]); ok { + return h.DataPoints + } + } + } + } + + return nil +} + +// TestRestoreRecordsEarlyFailure checks that a restore rejected before any tier +// is chosen is still measured. The duration is recorded from a defer, so the +// validation bail-outs at the top of Restore must produce a data point too, +// otherwise the failure rate panel would silently under-count. The tier and +// cluster name are unknown that early, and must be reported as "unknown" rather +// than as an empty attribute value. +func TestRestoreRecordsEarlyFailure(t *testing.T) { + reader := setupTestMeter(t) + + w := newWalServiceImplementation(newGRPCClientManager(), WALCapabilityOptions{}) + + // No source WAL name: Restore rejects the request before touching a tier. + _, err := w.Restore(context.Background(), &wal.WALRestoreRequest{}) + require.Error(t, err) + + rm := collectOTelMetrics(t, reader) + dps := findInt64HistogramDataPoints(rm, opentelemetry.PluginWalRestoreDurationMetric) + require.Len(t, dps, 1, "an early failure must still record one restore duration") + + outcome, ok := dps[0].Attributes.Value("outcome") + require.True(t, ok, "data point must carry an outcome attribute") + assert.Equal(t, string(opentelemetry.OutcomeFailure), outcome.AsString()) + + tier, ok := dps[0].Attributes.Value("tier") + require.True(t, ok, "data point must carry a tier attribute") + assert.Equal(t, string(tierUnknown), tier.AsString()) + + cluster, ok := dps[0].Attributes.Value("cluster_name") + require.True(t, ok, "data point must carry a cluster_name attribute") + assert.Equal(t, unknownAttributeValue, cluster.AsString()) +} + func TestGetConnectionErrors(t *testing.T) { tests := []struct { name string diff --git a/core/internal/opentelemetry/attributes.go b/core/internal/opentelemetry/attributes.go index 8052385a..17a63dfd 100644 --- a/core/internal/opentelemetry/attributes.go +++ b/core/internal/opentelemetry/attributes.go @@ -58,6 +58,36 @@ func (o Outcome) Attribute() attribute.KeyValue { return AttributeKeyOutcome.Of(string(o)) } +// CacheHit identifies whether a plugin WAL restore was served straight from the +// local prefetch spool (`true`) or had to wait on a download from the Klio +// server (`false`), used as the value of the `cache_hit` attribute. A prefetch +// still in flight when PostgreSQL asks counts as `false`, because the restore +// waits for that download to finish. +type CacheHit string + +const ( + // CacheHitTrue marks a restore served from a speculative prefetch already + // waiting in the local spool. + CacheHitTrue CacheHit = "true" + // CacheHitFalse marks a restore that had to download the WAL (no prefetch + // hit, or a partial/fallback download). + CacheHitFalse CacheHit = "false" +) + +// CacheHitOf maps a boolean prefetch-hit result to its CacheHit value. +func CacheHitOf(hit bool) CacheHit { + if hit { + return CacheHitTrue + } + + return CacheHitFalse +} + +// Attribute returns the `cache_hit` attribute for this value. +func (c CacheHit) Attribute() attribute.KeyValue { + return AttributeKeyCacheHit.Of(string(c)) +} + // Stage identifies a single step in the per-block WAL pipeline, used as the // value of the `stage` attribute to split a block-duration histogram across // its constituent steps instead of emitting one instrument per step. @@ -138,6 +168,10 @@ const ( // AttributeKeyPath is the attribute key for the WAL data-flow path (put or // get) of a per-block WAL duration histogram. AttributeKeyPath AttributeKey = "path" + // AttributeKeyCacheHit is the attribute key for whether a plugin WAL restore + // was served from a prefetch already complete in the spool (true) or had to + // wait on a download (false). + AttributeKeyCacheHit AttributeKey = "cache_hit" ) // Of builds an OTEL string attribute with the attribute key and the given value. diff --git a/core/internal/opentelemetry/catalog.go b/core/internal/opentelemetry/catalog.go index eb1eb66a..eed7d709 100644 --- a/core/internal/opentelemetry/catalog.go +++ b/core/internal/opentelemetry/catalog.go @@ -95,6 +95,8 @@ const ( PluginBackupInProgressMetric = "klio.plugin.backup.in_progress" PluginBackupRunsMetric = "klio.plugin.backup.runs" + PluginWalRestoreDurationMetric = "klio.plugin.wal.restore_duration" + ServerWalWrittenSizeMetric = "klio.server.wal.written_size" ServerWalWrittenMetric = "klio.server.wal.written" ServerWalLatestWrittenTimeMetric = "klio.server.wal.latest_written_time" @@ -226,6 +228,18 @@ type ClientWalMetrics struct { Timeline metric.Int64Gauge } +// PluginWalMetrics holds OTel instruments the CNPG-I plugin records for WAL +// restore. RestoreDuration is the end-to-end time the plugin takes to satisfy +// one RESTORE_WAL request (config resolution, tier failover, prefetch lookup or +// download, and the spool→destination rename). Each recording carries an +// `outcome` (`success` / `failure`), a `cache_hit` (`true` when the segment was +// already complete in the prefetch spool when PostgreSQL asked for it), a `tier` +// (which storage tier served it), and a `cluster_name`. A restore that fails +// before a tier is chosen reports `tier` and `cluster_name` as `unknown`. +type PluginWalMetrics struct { + RestoreDuration metric.Int64Histogram +} + // Centralized metric instrument instances. // //nolint:gochecknoglobals @@ -234,6 +248,7 @@ var ( ServerBackup ServerBackupMetrics ServerWal ServerWalMetrics ClientWal ClientWalMetrics + PluginWal PluginWalMetrics ) // All metric instruments are created when this package is loaded, in the @@ -247,6 +262,7 @@ func init() { InitServerBackupMetrics() InitServerWalMetrics() InitClientWalMetrics() + InitPluginWalMetrics() } // InitPluginBackupMetrics creates OTel instruments for backup lifecycle tracking. @@ -472,3 +488,20 @@ func InitClientWalMetrics() { metric.WithDescription("Timeline ID the WAL streaming client is currently streaming."), ) } + +// InitPluginWalMetrics creates the OTel instrument for the plugin end-to-end WAL +// restore path. It is called once when this package is loaded; tests can call it +// again after swapping the meter provider to rebind the instrument. +func InitPluginWalMetrics() { + meter := otel.Meter(Meter) + + PluginWal.RestoreDuration, _ = meter.Int64Histogram(PluginWalRestoreDurationMetric, + metric.WithDescription("Distribution of end-to-end WAL restore durations measured by the "+ + "CNPG-I plugin (the full RESTORE_WAL request: config resolution, tier failover, prefetch "+ + "lookup or download, and spool→destination rename). The `outcome` attribute is `success` "+ + "or `failure`; `cache_hit` is `true` when the segment was already complete in the "+ + "prefetch spool when PostgreSQL asked for it, so no download wait was needed; `tier` is the "+ + "storage tier that served the restore; `cluster_name` identifies the PostgreSQL cluster."), + metric.WithUnit("ns"), + ) +} diff --git a/core/internal/opentelemetry/metrics.go b/core/internal/opentelemetry/metrics.go index 64757a11..9c3095a6 100644 --- a/core/internal/opentelemetry/metrics.go +++ b/core/internal/opentelemetry/metrics.go @@ -79,6 +79,7 @@ func WalDurationViews() []metric.View { explicitBucketView(ClientWalBlockDurationMetric, walBlockDurationBuckets()), explicitBucketView(ServerWalGetDurationMetric, walFileDurationBuckets()), explicitBucketView(ServerWalUploadDurationMetric, walFileDurationBuckets()), + explicitBucketView(PluginWalRestoreDurationMetric, walFileDurationBuckets()), } } diff --git a/core/internal/opentelemetry/wal_duration_test.go b/core/internal/opentelemetry/wal_duration_test.go index f7ccd594..263d6ff7 100644 --- a/core/internal/opentelemetry/wal_duration_test.go +++ b/core/internal/opentelemetry/wal_duration_test.go @@ -56,6 +56,7 @@ func setupWalDurationProvider(t *testing.T) *sdkmetric.ManualReader { opentelemetry.InitServerWalMetrics() opentelemetry.InitClientWalMetrics() + opentelemetry.InitPluginWalMetrics() return reader } @@ -170,6 +171,45 @@ func TestClientWalBlockDurationSend(t *testing.T) { assert.Equal(t, string(opentelemetry.StageSend), stage.AsString()) } +// TestPluginWalRestoreDuration verifies the plugin end-to-end WAL restore +// histogram records with the outcome, cache_hit, tier and cluster_name +// attributes and that the per-file bucket boundaries are applied. +func TestPluginWalRestoreDuration(t *testing.T) { + reader := setupWalDurationProvider(t) + ctx := context.Background() + + opentelemetry.PluginWal.RestoreDuration.Record(ctx, 250_000_000, + metric.WithAttributes( + opentelemetry.OutcomeSuccess.Attribute(), + opentelemetry.CacheHitOf(true).Attribute(), + opentelemetry.Tier1.Attribute(), + opentelemetry.AttributeKeyClusterName.Of("cluster-a"), + )) + + dps := collectHistogram(t, reader, opentelemetry.PluginWalRestoreDurationMetric) + require.Len(t, dps, 1) + + outcome, ok := dps[0].Attributes.Value(attribute.Key("outcome")) + require.True(t, ok, "data point must carry an outcome attribute") + assert.Equal(t, string(opentelemetry.OutcomeSuccess), outcome.AsString()) + + cacheHit, ok := dps[0].Attributes.Value(attribute.Key("cache_hit")) + require.True(t, ok, "data point must carry a cache_hit attribute") + assert.Equal(t, "true", cacheHit.AsString()) + + tier, ok := dps[0].Attributes.Value(attribute.Key("tier")) + require.True(t, ok, "data point must carry a tier attribute") + assert.Equal(t, string(opentelemetry.Tier1), tier.AsString()) + + cluster, ok := dps[0].Attributes.Value(attribute.Key("cluster_name")) + require.True(t, ok, "data point must carry a cluster_name attribute") + assert.Equal(t, "cluster-a", cluster.AsString()) + + // The per-file ladder must be in force (not a single +Inf bucket). + assert.Greater(t, len(dps[0].Bounds), 1, + "explicit per-file bucket boundaries must be applied to the restore histogram") +} + // TestClientWalTimeline verifies the client timeline gauge reports the most // recently recorded timeline value. func TestClientWalTimeline(t *testing.T) { From 01b185781875fb6aeb45508e455a6c7e88365f5e Mon Sep 17 00:00:00 2001 From: Hai He Date: Tue, 18 Aug 2026 14:48:30 +0800 Subject: [PATCH 2/3] feat(grafana): add WAL restore latency and rate panels Add two panels to the Client / Plugin section for the new klio.plugin.wal.restore_duration metric: p95 end-to-end restore latency split by cache_hit and restore rate by outcome. A prefetch hit is a local rename while a miss waits on a download, so the two are orders of magnitude apart and are shown split rather than pooled; a falling hit ratio means prefetch is not keeping ahead of replay. Regenerated klio-dashboard.json via the builder. Assisted-by: Claude Signed-off-by: Hai He --- .../web/docs/user/grafana-dashboards.md | 13 +- observability/grafana/client.go | 29 ++- observability/grafana/klio-dashboard.json | 170 ++++++++++++++---- 3 files changed, 175 insertions(+), 37 deletions(-) diff --git a/documentation/web/docs/user/grafana-dashboards.md b/documentation/web/docs/user/grafana-dashboards.md index 4010b9a4..52b5fd01 100644 --- a/documentation/web/docs/user/grafana-dashboards.md +++ b/documentation/web/docs/user/grafana-dashboards.md @@ -25,7 +25,9 @@ The dashboard is a single dashboard split into row sections: and the backup success ratio. Also the WAL streaming client the sidecar supervises as a child process: the PostgreSQL timeline it is currently streaming and the p50/p95/p99 latency of sending - a WAL block to the server. + a WAL block to the server. Finally the WAL restores the plugin serves back + to PostgreSQL: the p95 end-to-end restore latency split by prefetch cache + hit, and the restore rate by outcome. ![Klio client and plugin metrics](images/klio_client_and_plugin_metrics.png) @@ -70,6 +72,15 @@ histogram percentiles that need enough recent samples to be reliable: sent too infrequently for the underlying `histogram_quantile` to produce a reliable percentile, so the line can look sparse or noisy rather than simply absent. +- **WAL restore latency (p95) by cache hit** splits on whether the segment was + already sitting complete in the prefetch spool when PostgreSQL asked for it. + A hit is a local rename and a miss waits on a download, so the two lines sit + orders of magnitude apart and are deliberately not pooled: a single line + would drift with the hit rate rather than describe either case. Because a hit + is usually well under the histogram's smallest bucket, read that line as + "fast" rather than as a precise value; the miss line is the one that carries + detail. A hit rate falling over time means prefetch is no longer keeping up + with replay. - **Backup duration (p50/p95/p99)** has the same limitation, more acutely: backups are infrequent, so this panel is computed over the whole selected range (rather than a short rate window) to stay populated between runs. diff --git a/observability/grafana/client.go b/observability/grafana/client.go index 4dcda25f..64843b8b 100644 --- a/observability/grafana/client.go +++ b/observability/grafana/client.go @@ -35,8 +35,10 @@ const clientWalMatcher = `k8s_namespace_name=~"$namespace",cluster_name=~"$clust // backup lifecycle (`klio.plugin.backup.*`, exported to Prometheus as // `klio_plugin_backup_*`) and the WAL streaming client it supervises as a // child process (`klio.client.wal.*`, exported as `klio_client_wal_*`). -// Backup queries are scoped by $namespace; WAL streaming queries additionally -// carry cluster_name and are scoped by $cluster. +// It also records the end-to-end WAL restore latency the plugin serves to +// PostgreSQL (`klio.plugin.wal.*`, exported as `klio_plugin_wal_*`). +// Backup and plugin WAL queries are scoped by $namespace; WAL streaming +// queries additionally carry cluster_name and are scoped by $cluster. func clientPanels() []sizedPanel { return []sizedPanel{ // Current backup state. @@ -131,5 +133,28 @@ func clientPanels() []sizedPanel { "cluster. Most meaningful under active write load; on an idle or low-write cluster, WAL "+ "blocks are sent too infrequently for the underlying histogram_quantile to be reliable, so "+ "the line may look sparse or noisy rather than absent.")), + + // WAL restore latency and throughput. The plugin records + // klio.plugin.wal.restore_duration for every RESTORE_WAL request it serves + // to PostgreSQL, exported as the klio_plugin_wal_restore_duration_nanoseconds + // histogram. Scoped by $namespace like the backup family. + sized(8, panelHeight, timeseriesPanel("WAL restore latency (p95) by cache hit", "ns", + query( + fmt.Sprintf("histogram_quantile(0.95, sum by (le, cache_hit) "+ + "(rate(klio_plugin_wal_restore_duration_nanoseconds_bucket{%s}[$__rate_interval])))", nsMatcher), + "p95 cache_hit={{cache_hit}}"), + ).Description("95th-percentile end-to-end WAL restore latency measured by the plugin: the latency "+ + "PostgreSQL actually experiences, and in a replica cluster the speed of the replication path "+ + "itself. cache_hit=true means the segment was already in the prefetch spool and the restore "+ + "was a local rename; cache_hit=false means PostgreSQL had to wait on a download, so a falling "+ + "hit ratio means prefetch is not keeping ahead of replay and the prefetch count may need "+ + "raising. The two are orders of magnitude apart, hence the split rather than one pooled line.")), + sized(8, panelHeight, timeseriesPanel("WAL restore rate by outcome", "ops", + query( + fmt.Sprintf("sum by (outcome) "+ + "(rate(klio_plugin_wal_restore_duration_nanoseconds_count{%s}[$__rate_interval]))", nsMatcher), + "{{outcome}}"), + ).Description("Rate of WAL restore requests handled by the plugin, split by outcome (success or "+ + "failure).")), } } diff --git a/observability/grafana/klio-dashboard.json b/observability/grafana/klio-dashboard.json index 90b2cc82..a928a878 100644 --- a/observability/grafana/klio-dashboard.json +++ b/observability/grafana/klio-dashboard.json @@ -820,7 +820,7 @@ }, "gridPos": { "h": 6, - "w": 24, + "w": 16, "x": 0, "y": 19 }, @@ -848,6 +848,108 @@ "overrides": [] } }, + { + "type": "timeseries", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "expr": "histogram_quantile(0.95, sum by (le, cache_hit) (rate(klio_plugin_wal_restore_duration_nanoseconds_bucket{k8s_namespace_name=~\"$namespace\"}[$__rate_interval])))", + "instant": false, + "legendFormat": "p95 cache_hit={{cache_hit}}", + "range": true + } + ], + "title": "WAL restore latency (p95) by cache hit", + "description": "95th-percentile end-to-end WAL restore latency measured by the plugin: the latency PostgreSQL actually experiences, and in a replica cluster the speed of the replication path itself. cache_hit=true means the segment was already in the prefetch spool and the restore was a local rename; cache_hit=false means PostgreSQL had to wait on a download, so a falling hit ratio means prefetch is not keeping ahead of replay and the prefetch count may need raising. The two are orders of magnitude apart, hence the split rather than one pooled line.", + "transparent": false, + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "gridPos": { + "h": 6, + "w": 8, + "x": 16, + "y": 19 + }, + "repeatDirection": "h", + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom", + "showLegend": true, + "calcs": [] + }, + "tooltip": { + "mode": "", + "sort": "" + } + }, + "fieldConfig": { + "defaults": { + "unit": "ns", + "custom": { + "gradientMode": "opacity", + "fillOpacity": 10 + } + }, + "overrides": [] + } + }, + { + "type": "timeseries", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "expr": "sum by (outcome) (rate(klio_plugin_wal_restore_duration_nanoseconds_count{k8s_namespace_name=~\"$namespace\"}[$__rate_interval]))", + "instant": false, + "legendFormat": "{{outcome}}", + "range": true + } + ], + "title": "WAL restore rate by outcome", + "description": "Rate of WAL restore requests handled by the plugin, split by outcome (success or failure).", + "transparent": false, + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "gridPos": { + "h": 6, + "w": 24, + "x": 0, + "y": 25 + }, + "repeatDirection": "h", + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom", + "showLegend": true, + "calcs": [] + }, + "tooltip": { + "mode": "", + "sort": "" + } + }, + "fieldConfig": { + "defaults": { + "unit": "ops", + "custom": { + "gradientMode": "opacity", + "fillOpacity": 10 + } + }, + "overrides": [] + } + }, { "type": "row", "collapsed": false, @@ -856,7 +958,7 @@ "h": 1, "w": 24, "x": 0, - "y": 25 + "y": 31 }, "id": 0, "panels": [] @@ -886,7 +988,7 @@ "h": 6, "w": 4, "x": 0, - "y": 26 + "y": 32 }, "repeatDirection": "h", "options": { @@ -943,7 +1045,7 @@ "h": 6, "w": 4, "x": 4, - "y": 26 + "y": 32 }, "repeatDirection": "h", "options": { @@ -1001,7 +1103,7 @@ "h": 6, "w": 4, "x": 8, - "y": 26 + "y": 32 }, "repeatDirection": "h", "options": { @@ -1059,7 +1161,7 @@ "h": 6, "w": 4, "x": 12, - "y": 26 + "y": 32 }, "repeatDirection": "h", "options": { @@ -1116,7 +1218,7 @@ "h": 6, "w": 4, "x": 16, - "y": 26 + "y": 32 }, "repeatDirection": "h", "options": { @@ -1174,7 +1276,7 @@ "h": 6, "w": 4, "x": 20, - "y": 26 + "y": 32 }, "repeatDirection": "h", "options": { @@ -1232,7 +1334,7 @@ "h": 6, "w": 4, "x": 0, - "y": 32 + "y": 38 }, "repeatDirection": "h", "options": { @@ -1289,7 +1391,7 @@ "h": 6, "w": 4, "x": 4, - "y": 32 + "y": 38 }, "repeatDirection": "h", "options": { @@ -1346,7 +1448,7 @@ "h": 6, "w": 4, "x": 8, - "y": 32 + "y": 38 }, "repeatDirection": "h", "options": { @@ -1403,7 +1505,7 @@ "h": 6, "w": 4, "x": 12, - "y": 32 + "y": 38 }, "repeatDirection": "h", "options": { @@ -1460,7 +1562,7 @@ "h": 6, "w": 4, "x": 16, - "y": 32 + "y": 38 }, "repeatDirection": "h", "options": { @@ -1517,7 +1619,7 @@ "h": 6, "w": 4, "x": 20, - "y": 32 + "y": 38 }, "repeatDirection": "h", "options": { @@ -1574,7 +1676,7 @@ "h": 6, "w": 8, "x": 0, - "y": 38 + "y": 44 }, "repeatDirection": "h", "options": { @@ -1642,7 +1744,7 @@ "h": 6, "w": 8, "x": 8, - "y": 38 + "y": 44 }, "repeatDirection": "h", "options": { @@ -1703,7 +1805,7 @@ "h": 6, "w": 8, "x": 16, - "y": 38 + "y": 44 }, "repeatDirection": "h", "options": { @@ -1764,7 +1866,7 @@ "h": 6, "w": 8, "x": 0, - "y": 44 + "y": 50 }, "repeatDirection": "h", "options": { @@ -1822,7 +1924,7 @@ "h": 6, "w": 8, "x": 8, - "y": 44 + "y": 50 }, "repeatDirection": "h", "options": { @@ -1873,7 +1975,7 @@ "h": 6, "w": 8, "x": 16, - "y": 44 + "y": 50 }, "repeatDirection": "h", "options": { @@ -1924,7 +2026,7 @@ "h": 6, "w": 8, "x": 0, - "y": 50 + "y": 56 }, "repeatDirection": "h", "options": { @@ -1975,7 +2077,7 @@ "h": 6, "w": 8, "x": 8, - "y": 50 + "y": 56 }, "repeatDirection": "h", "options": { @@ -2026,7 +2128,7 @@ "h": 6, "w": 8, "x": 16, - "y": 50 + "y": 56 }, "repeatDirection": "h", "options": { @@ -2097,7 +2199,7 @@ "h": 6, "w": 8, "x": 0, - "y": 56 + "y": 62 }, "repeatDirection": "h", "options": { @@ -2168,7 +2270,7 @@ "h": 6, "w": 8, "x": 8, - "y": 56 + "y": 62 }, "repeatDirection": "h", "options": { @@ -2239,7 +2341,7 @@ "h": 6, "w": 8, "x": 16, - "y": 56 + "y": 62 }, "repeatDirection": "h", "options": { @@ -2290,7 +2392,7 @@ "h": 6, "w": 12, "x": 0, - "y": 62 + "y": 68 }, "repeatDirection": "h", "options": { @@ -2341,7 +2443,7 @@ "h": 6, "w": 12, "x": 12, - "y": 62 + "y": 68 }, "repeatDirection": "h", "options": { @@ -2392,7 +2494,7 @@ "h": 6, "w": 12, "x": 0, - "y": 68 + "y": 74 }, "repeatDirection": "h", "options": { @@ -2443,7 +2545,7 @@ "h": 6, "w": 12, "x": 12, - "y": 68 + "y": 74 }, "repeatDirection": "h", "options": { @@ -2477,7 +2579,7 @@ "h": 1, "w": 24, "x": 0, - "y": 74 + "y": 80 }, "id": 0, "panels": [] @@ -2507,7 +2609,7 @@ "h": 6, "w": 8, "x": 0, - "y": 75 + "y": 81 }, "repeatDirection": "h", "options": { @@ -2558,7 +2660,7 @@ "h": 6, "w": 8, "x": 8, - "y": 75 + "y": 81 }, "repeatDirection": "h", "options": { @@ -2609,7 +2711,7 @@ "h": 6, "w": 8, "x": 16, - "y": 75 + "y": 81 }, "repeatDirection": "h", "options": { From 758706c526a2063ca6b90944425058c30398537f Mon Sep 17 00:00:00 2001 From: Hai He Date: Wed, 2 Sep 2026 12:01:53 +0800 Subject: [PATCH 3/3] feat(grafana): add WAL restore hit ratio and tier tracking Keep WAL restore metrics aligned with the real restore path by retaining the last attempted tier on miss-all-tiers failures and expose a direct prefetch hit-ratio panel in the Grafana dashboard. Assisted-by: Claude Signed-off-by: Hai He --- core/internal/cnpgi/wal.go | 4 +- observability/grafana/client.go | 6 +++ observability/grafana/klio-dashboard.json | 55 ++++++++++++++++++++++- 3 files changed, 62 insertions(+), 3 deletions(-) diff --git a/core/internal/cnpgi/wal.go b/core/internal/cnpgi/wal.go index 7817605c..d167e086 100644 --- a/core/internal/cnpgi/wal.go +++ b/core/internal/cnpgi/wal.go @@ -180,7 +180,9 @@ func (w *walServiceImplementation) restoreWAL( tiers[0], tiers[1] = tiers[1], tiers[0] } + var lastTier tier for _, t := range tiers { + lastTier = t cacheHit, err := w.mgr.restoreWAL(ctx, walRestoreOptions{ configFile: configPath, targetTier: t, @@ -194,7 +196,7 @@ func (w *walServiceImplementation) restoreWAL( } } - return restoreOutcome{}, errWALNotFound + return restoreOutcome{tier: lastTier}, errWALNotFound } // availableTiers returns the tiers the user has opted in to as recovery diff --git a/observability/grafana/client.go b/observability/grafana/client.go index 64843b8b..9f6202cc 100644 --- a/observability/grafana/client.go +++ b/observability/grafana/client.go @@ -149,6 +149,12 @@ func clientPanels() []sizedPanel { "was a local rename; cache_hit=false means PostgreSQL had to wait on a download, so a falling "+ "hit ratio means prefetch is not keeping ahead of replay and the prefetch count may need "+ "raising. The two are orders of magnitude apart, hence the split rather than one pooled line.")), + sized(8, panelHeight, timeseriesPanel("WAL restore hit ratio", "%", + query( + fmt.Sprintf("sum(rate(klio_plugin_wal_restore_duration_nanoseconds_count{%s,cache_hit=\"true\"}[$__rate_interval])) / "+ + "sum(rate(klio_plugin_wal_restore_duration_nanoseconds_count{%s}[$__rate_interval])) * 100", nsMatcher, nsMatcher), + "hit ratio"), + ).Description("Fraction of WAL restores served directly from the prefetch spool instead of waiting on a download. A falling ratio means prefetch is not keeping pace with PostgreSQL replay.")), sized(8, panelHeight, timeseriesPanel("WAL restore rate by outcome", "ops", query( fmt.Sprintf("sum by (outcome) "+ diff --git a/observability/grafana/klio-dashboard.json b/observability/grafana/klio-dashboard.json index a928a878..60c879be 100644 --- a/observability/grafana/klio-dashboard.json +++ b/observability/grafana/klio-dashboard.json @@ -899,6 +899,57 @@ "overrides": [] } }, + { + "type": "timeseries", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "expr": "sum(rate(klio_plugin_wal_restore_duration_nanoseconds_count{k8s_namespace_name=~\"$namespace\",cache_hit=\"true\"}[$__rate_interval])) / sum(rate(klio_plugin_wal_restore_duration_nanoseconds_count{k8s_namespace_name=~\"$namespace\"}[$__rate_interval])) * 100", + "instant": false, + "legendFormat": "hit ratio", + "range": true + } + ], + "title": "WAL restore hit ratio", + "description": "Fraction of WAL restores served directly from the prefetch spool instead of waiting on a download. A falling ratio means prefetch is not keeping pace with PostgreSQL replay.", + "transparent": false, + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "gridPos": { + "h": 6, + "w": 12, + "x": 0, + "y": 25 + }, + "repeatDirection": "h", + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom", + "showLegend": true, + "calcs": [] + }, + "tooltip": { + "mode": "", + "sort": "" + } + }, + "fieldConfig": { + "defaults": { + "unit": "%", + "custom": { + "gradientMode": "opacity", + "fillOpacity": 10 + } + }, + "overrides": [] + } + }, { "type": "timeseries", "targets": [ @@ -922,8 +973,8 @@ }, "gridPos": { "h": 6, - "w": 24, - "x": 0, + "w": 12, + "x": 12, "y": 25 }, "repeatDirection": "h",