diff --git a/core/internal/cnpgi/metrics.go b/core/internal/cnpgi/metrics.go index 908ff563e..e70948764 100644 --- a/core/internal/cnpgi/metrics.go +++ b/core/internal/cnpgi/metrics.go @@ -29,6 +29,10 @@ import ( "github.com/cloudnative-pg/klio/core/internal/opentelemetry" ) +// unknownAttributeValue tags an attribute whose real value was not yet known +// when the metric was recorded. +const unknownAttributeValue = "unknown" + // clusterAttr returns the `cluster_name` attribute every plugin backup metric // carries, so panels can attribute backup activity to a specific PostgreSQL // cluster even when several clusters share one namespace. @@ -69,6 +73,34 @@ func recordBackupSuccess(ctx context.Context, clusterName string, duration time. metric.WithAttributes(cluster, 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, + info restoreOutcome, + clusterName string, +) { + restoreTier := string(info.tier) + if restoreTier == "" { + restoreTier = unknownAttributeValue + } + if clusterName == "" { + clusterName = unknownAttributeValue + } + + opentelemetry.PluginWal.RestoreDuration.Record(ctx, duration.Nanoseconds(), + metric.WithAttributes( + info.result.Attribute(), + opentelemetry.CacheHitOf(info.cacheHit).Attribute(), + opentelemetry.AttributeKeyTier.Of(restoreTier), + opentelemetry.AttributeKeyClusterName.Of(clusterName), + )) +} + // recordBackupFailure records a failed backup. func recordBackupFailure(ctx context.Context, clusterName string, duration time.Duration, err error) { cluster := clusterAttr(clusterName) diff --git a/core/internal/cnpgi/metrics_test.go b/core/internal/cnpgi/metrics_test.go index 650ba22d0..9bfe64541 100644 --- a/core/internal/cnpgi/metrics_test.go +++ b/core/internal/cnpgi/metrics_test.go @@ -57,6 +57,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 0f7105a56..86955ceca 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,11 @@ func (p *walPrefetcher) Close() error { return p.downloadPool.Wait() } -// getCompleteWAL retrieves a complete WAL file from cache or downloads it. -// -//nolint:cyclop // complexity is slightly over limit but refactoring would hurt readability -func (p *walPrefetcher) getCompleteWAL(ctx context.Context, walName, targetPath string) error { +// 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. +func (p *walPrefetcher) getCompleteWAL(ctx context.Context, walName, targetPath string) (bool, error) { contextLogger := log.FromContext(ctx).WithValues("walName", walName) p.mu.Lock() @@ -190,7 +203,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 +219,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 +240,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 5807f83ac..8c4758469 100644 --- a/core/internal/cnpgi/prefetcher_test.go +++ b/core/internal/cnpgi/prefetcher_test.go @@ -21,7 +21,9 @@ package cnpgi import ( "context" + "errors" "os" + "path/filepath" "testing" "github.com/stretchr/testify/assert" @@ -389,6 +391,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)) @@ -405,3 +435,67 @@ func TestCanHavePartial(t *testing.T) { assert.False(t, canHavePartial("000000010000000000000001.00000028.backup")) assert.False(t, canHavePartial("000000010000000000000001.partial")) } + +// TestGetCompleteWALCacheHit checks the cache_hit result of getCompleteWAL for +// entries already tracked by the prefetcher. Only a prefetch that was complete +// when the WAL was requested is a hit: one still downloading makes the caller +// wait, so it is a miss even once its download completes. +func TestGetCompleteWALCacheHit(t *testing.T) { + const walName = "000000010000000000000001" + errDownload := errors.New("download failed") + + tests := []struct { + name string + isPrefetch bool + inFlight bool + err error + wantHit bool + }{ + {name: "ready prefetch is a hit", isPrefetch: true, wantHit: true}, + {name: "in-flight prefetch is a miss", isPrefetch: true, inFlight: true}, + {name: "ready direct download is a miss"}, + {name: "failed download is a miss", isPrefetch: true, err: errDownload}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + spoolDir := t.TempDir() + p := newWALPrefetcher(spoolDir, nil, 2, 2) + defer func() { _ = p.Close() }() + + spoolPath := filepath.Join(spoolDir, walName) + require.NoError(t, os.WriteFile(spoolPath, []byte("wal"), 0o600)) + + entry := &walEntry{ + state: walStateReady, + spoolPath: spoolPath, + done: make(chan struct{}), + isPrefetch: tt.isPrefetch, + } + if tt.err != nil { + entry.err = tt.err + } + // The hit is decided from the state seen at lookup. An in-flight + // entry keeps its downloading state, while its done channel is + // closed so the request does not block waiting for it. + if tt.inFlight { + entry.state = walStateDownloading + } + close(entry.done) + + p.mu.Lock() + p.entries[walName] = entry + p.mu.Unlock() + + targetPath := filepath.Join(t.TempDir(), walName) + hit, err := p.getCompleteWAL(context.Background(), walName, targetPath) + if tt.err != nil { + require.ErrorIs(t, err, tt.err) + } else { + require.NoError(t, err) + assert.FileExists(t, targetPath) + } + assert.Equal(t, tt.wantHit, hit) + }) + } +} diff --git a/core/internal/cnpgi/wal.go b/core/internal/cnpgi/wal.go index f1191d713..94d110de1 100644 --- a/core/internal/cnpgi/wal.go +++ b/core/internal/cnpgi/wal.go @@ -38,6 +38,7 @@ import ( "google.golang.org/grpc/status" "github.com/cloudnative-pg/klio/core/internal/client/klioclient/grpcclient" + "github.com/cloudnative-pg/klio/core/internal/opentelemetry" "github.com/cloudnative-pg/klio/core/pkg/config" ) @@ -97,6 +98,20 @@ 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 outcome/clusterName, so failures — + // including the fast validation bail-outs below — are measured too. The + // result starts as a failure so a bail-out that never reaches restoreWAL + // is counted as one. + var ( + outcome restoreOutcome + clusterName string + ) + outcome.result = opentelemetry.OutcomeFailure + defer func() { + recordWalRestore(ctx, time.Since(startCall), outcome, 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,8 +136,11 @@ func (w *walServiceImplementation) Restore( return nil, errors.New("no WAL repository found for the cluster") } - err = w.restoreWAL(ctx, walName, destinationPath, confPath) - if errors.Is(err, errWALNotFound) { + cacheHit, restoreTier, err := w.restoreWAL(ctx, walName, destinationPath, confPath) + outcome.cacheHit = cacheHit + outcome.tier = restoreTier + outcome.result = restoreResult(err) + if outcome.result == opentelemetry.OutcomeNotFound { return &wal.WALRestoreResult{}, status.Errorf(codes.NotFound, "WAL file not found: %q", walName) } if err != nil { @@ -133,19 +152,51 @@ func (w *walServiceImplementation) Restore( return &wal.WALRestoreResult{}, nil } +// restoreResult classifies a restoreWAL error for the `outcome` attribute of +// the restore duration metric. Callers branch on the returned Outcome rather +// than re-testing the error, so this stays the only definition of "not found" +// and nothing derived from it can disagree. +// +// `not_found` is deliberately not a failure. PostgreSQL routinely asks for a +// segment or a timeline history file that was never archived, and that absence +// is how it learns it has reached the end of the archive. CloudNativePG +// collects every plugin error the same way and never inspects the gRPC code, so +// this attribute is the only place the distinction survives; folding it into +// `failure` would report a constant failure rate on a healthy cluster. +func restoreResult(err error) opentelemetry.Outcome { + switch { + case err == nil: + return opentelemetry.OutcomeSuccess + case errors.Is(err, errWALNotFound): + return opentelemetry.OutcomeNotFound + default: + return opentelemetry.OutcomeFailure + } +} + +// restoreOutcome carries what Restore tags its end-to-end duration metric +// with: the result it classifies from the restore error, plus the facts only +// known deep in the restore path. On failure the latter hold whatever was known +// so far (tier is the last one attempted, or empty; cacheHit is false). +type restoreOutcome struct { + tier tier + cacheHit bool + result opentelemetry.Outcome +} + func (w *walServiceImplementation) restoreWAL( ctx context.Context, walName, destinationPath string, configPath string, -) error { +) (bool, tier, error) { cfg, err := config.NewFromFile(afero.NewOsFs(), configPath) if err != nil { - return fmt.Errorf("while loading configuration from file %q: %w", configPath, err) + return false, "", 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 false, "", errors.New("no WAL tier configured") } // Try the previously-successful tier first, when both are available. @@ -154,21 +205,23 @@ func (w *walServiceImplementation) restoreWAL( tiers[0], tiers[1] = tiers[1], tiers[0] } + var lastTier tier for _, t := range tiers { - err := w.mgr.restoreWAL(ctx, walRestoreOptions{ + lastTier = t + cacheHit, err := w.mgr.restoreWAL(ctx, walRestoreOptions{ configFile: configPath, targetTier: t, }, walName, destinationPath) if err == nil { w.currentTier.Store(t) - return nil + return cacheHit, t, nil } if !errors.Is(err, errWALNotFound) { - return err + return false, t, err } } - return errWALNotFound + return false, lastTier, errWALNotFound } // availableTiers returns the tiers the user has opted in to as recovery @@ -297,15 +350,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 f14d78c2d..4bed68282 100644 --- a/core/internal/cnpgi/wal_test.go +++ b/core/internal/cnpgi/wal_test.go @@ -21,12 +21,21 @@ package cnpgi import ( "context" + "errors" + "fmt" "os" "path/filepath" "reflect" "strings" "testing" + "time" + "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 +135,157 @@ 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, unknownAttributeValue, 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()) +} + +// TestRestoreRecordsRestoreFailure checks that a restore that fails inside +// restoreWAL is still tagged `failure`, and that the tier and cluster name it +// could not determine are reported as "unknown" rather than as an empty +// attribute value, which would show up as a nameless series on every per-tier +// panel. +func TestRestoreRecordsRestoreFailure(t *testing.T) { + reader := setupTestMeter(t) + t.Setenv("POD_NAME", "cluster-a-1") + + w := newWalServiceImplementation(newGRPCClientManager(), WALCapabilityOptions{}) + + // A primary with no promotion token resolves to the default archive config + // path, which does not exist here, so restoreWAL fails loading it. + _, err := w.Restore(context.Background(), &wal.WALRestoreRequest{ + SourceWalName: "000000010000000000000001", + DestinationFileName: filepath.Join(t.TempDir(), "000000010000000000000001"), + ClusterDefinition: []byte( + `{"metadata":{"name":"cluster-a"},"status":{"currentPrimary":"cluster-a-1"}}`), + }) + require.Error(t, err) + + dps := findInt64HistogramDataPoints( + collectOTelMetrics(t, reader), opentelemetry.PluginWalRestoreDurationMetric) + require.Len(t, dps, 1) + + 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, unknownAttributeValue, tier.AsString(), "no tier was reached") + + cluster, ok := dps[0].Attributes.Value("cluster_name") + require.True(t, ok, "data point must carry a cluster_name attribute") + assert.Equal(t, "cluster-a", cluster.AsString()) +} + +// TestRestoreResult checks how a restore error is classified for the `outcome` +// attribute. A served restore needs a real Klio server, so the success case is +// asserted here too: it is exactly the case a per-return-path assignment lost. +func TestRestoreResult(t *testing.T) { + tests := []struct { + name string + err error + want opentelemetry.Outcome + }{ + {name: "served", err: nil, want: opentelemetry.OutcomeSuccess}, + {name: "absent from the archive", err: errWALNotFound, want: opentelemetry.OutcomeNotFound}, + { + name: "absent from the archive, wrapped", + err: fmt.Errorf("tier2: %w", errWALNotFound), + want: opentelemetry.OutcomeNotFound, + }, + {name: "restore broke", err: errors.New("connection refused"), want: opentelemetry.OutcomeFailure}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, restoreResult(tt.err)) + }) + } +} + +// TestRecordWalRestoreOutcomes checks that a WAL the archive simply does not +// hold is recorded as `not_found`, kept apart from a real `failure`. PostgreSQL +// asking for a segment or timeline history file that was never archived is the +// routine end-of-archive signal, and CloudNativePG cannot tell the two apart — +// it collects every plugin error identically without inspecting the gRPC code — +// so this attribute is the only place the distinction survives. Pooling them +// would report a constant failure rate on a healthy cluster and would drag down +// the prefetch hit ratio, whose denominator counts successes only. +func TestRecordWalRestoreOutcomes(t *testing.T) { + tests := []struct { + name string + outcome opentelemetry.Outcome + want string + }{ + {name: "served", outcome: opentelemetry.OutcomeSuccess, want: "success"}, + {name: "absent from the archive", outcome: opentelemetry.OutcomeNotFound, want: "not_found"}, + {name: "restore broke", outcome: opentelemetry.OutcomeFailure, want: "failure"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + reader := setupTestMeter(t) + + recordWalRestore(context.Background(), time.Second, + restoreOutcome{tier: tier1, cacheHit: false, result: tt.outcome}, "cluster-a") + + dps := findInt64HistogramDataPoints( + collectOTelMetrics(t, reader), opentelemetry.PluginWalRestoreDurationMetric) + require.Len(t, dps, 1) + + outcome, ok := dps[0].Attributes.Value("outcome") + require.True(t, ok, "data point must carry an outcome attribute") + assert.Equal(t, tt.want, outcome.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 8052385a9..9a50f0c4d 100644 --- a/core/internal/opentelemetry/attributes.go +++ b/core/internal/opentelemetry/attributes.go @@ -51,6 +51,9 @@ const ( OutcomeSuccess Outcome = "success" // OutcomeFailure marks the failure flavor of an operation counter. OutcomeFailure Outcome = "failure" + // OutcomeNotFound marks an operation that completed normally but found + // nothing to act on. + OutcomeNotFound Outcome = "not_found" ) // Attribute returns the `outcome` attribute for this outcome. @@ -58,6 +61,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 +171,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 f7ea0f46c..777537e00 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" @@ -231,6 +233,19 @@ 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` / `not_found`), 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 @@ -239,6 +254,7 @@ var ( ServerBackup ServerBackupMetrics ServerWal ServerWalMetrics ClientWal ClientWalMetrics + PluginWal PluginWalMetrics ) // All metric instruments are created when this package is loaded, in the @@ -252,6 +268,7 @@ func init() { InitServerBackupMetrics() InitServerWalMetrics() InitClientWalMetrics() + InitPluginWalMetrics() } // InitPluginBackupMetrics creates OTel instruments for backup lifecycle tracking. @@ -501,3 +518,22 @@ 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`, "+ + "`failure`, or `not_found` — the last being PostgreSQL asking for a segment or timeline "+ + "history file that was never archived, which is the routine end-of-archive signal and not "+ + "an error; `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 64757a119..9c3095a6a 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 f7ccd594c..263d6ff7e 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) { diff --git a/documentation/web/docs/developer/running-e2e-tests.md b/documentation/web/docs/developer/running-e2e-tests.md index 172d8eef9..1f5de90b7 100644 --- a/documentation/web/docs/developer/running-e2e-tests.md +++ b/documentation/web/docs/developer/running-e2e-tests.md @@ -154,7 +154,9 @@ The E2E tests are located in `operator/test/e2e/` and include: backing `/klio` (`PVCResize`) - **`otel_test.go`** - OpenTelemetry metrics and traces export: deploys an OTEL Collector and verifies that backup lifecycle metrics and - traces are correctly exported via OTLP. After the success-path + traces are correctly exported via OTLP. It also runs the + `restore_command` in the primary to check that WAL restores are recorded + in `klio.plugin.wal.restore_duration`. After the success-path assertions it deletes the Klio server and triggers a failing backup to verify the `failure_category=repository_error` attribute on `klio.plugin.backup.runs` (`OTELMetricsAndTraces`) diff --git a/documentation/web/docs/user/grafana-dashboards.md b/documentation/web/docs/user/grafana-dashboards.md index acdd67130..360300375 100644 --- a/documentation/web/docs/user/grafana-dashboards.md +++ b/documentation/web/docs/user/grafana-dashboards.md @@ -25,7 +25,11 @@ 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/p90/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 p50/p90/p99 end-to-end restore duration split by + prefetch cache hit (as both a rolling-window and a since-restart total + variant), the restore rate and total by tier and outcome, and the prefetch + hit ratio. ![Klio client and plugin metrics](images/klio_client_and_plugin_metrics.png) @@ -65,6 +69,18 @@ from the alerting guidance in [OpenTelemetry](opentelemetry.md): staleness panel, it tells a slow pipeline (timestamps advancing, gap growing) apart from a stalled one (timestamps and LSN both frozen). +One more is a statistical caveat rather than an alerting signal: + +- **WAL restore duration percentiles** split 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 **WAL restore prefetch hit ratio** falling over time means + prefetch is no longer keeping up with replay. + ## Prerequisites The dashboard reads the Prometheus names of Klio's metrics (for example diff --git a/documentation/web/docs/user/images/klio_client_and_plugin_metrics.png b/documentation/web/docs/user/images/klio_client_and_plugin_metrics.png index 5b54b9d0d..690751164 100644 Binary files a/documentation/web/docs/user/images/klio_client_and_plugin_metrics.png and b/documentation/web/docs/user/images/klio_client_and_plugin_metrics.png differ diff --git a/documentation/web/docs/user/opentelemetry.md b/documentation/web/docs/user/opentelemetry.md index a5544fda4..02ab34c02 100644 --- a/documentation/web/docs/user/opentelemetry.md +++ b/documentation/web/docs/user/opentelemetry.md @@ -160,9 +160,10 @@ attribute key. | Attribute | Values | Applies to | |---|---|---| -| `tier` | `tier1` (local disk on the Klio server), `tier2` (remote object store) | All `klio.server.wal.*` and `klio.server.backup.*` instruments. | -| `cluster_name` | Name of the PostgreSQL cluster the recording belongs to | All `klio.plugin.backup.*` instruments, all `klio.server.wal.*` instruments (counters, gauges, and the WAL duration histograms), `klio.client.wal.*`, the `klio.server.backup.*` PostgreSQL backup gauges (`backups`, `latest_backup_*`, `oldest_backup_*`) and the `klio.server.backup.relay` / `klio.server.backup.maintenance` counters. | -| `outcome` | `success`, `failure` | `klio.plugin.backup.runs`, `klio.server.backup.relay`, `klio.server.backup.maintenance`, `klio.server.backup.verifications`, and all WAL duration histograms (`klio.server.wal.*_duration`, `klio.client.wal.block_duration`). | +| `tier` | `tier1` (local disk on the Klio server), `tier2` (remote object store); `unknown` on `klio.plugin.wal.restore_duration` when the restore failed before any tier was tried | All `klio.server.wal.*` and `klio.server.backup.*` instruments, `klio.plugin.wal.restore_duration`. | +| `cluster_name` | Name of the PostgreSQL cluster the recording belongs to | All `klio.plugin.backup.*` instruments, `klio.plugin.wal.restore_duration`, all `klio.server.wal.*` instruments (counters, gauges, and the WAL duration histograms), `klio.client.wal.*`, the `klio.server.backup.*` PostgreSQL backup gauges (`backups`, `latest_backup_*`, `oldest_backup_*`) and the `klio.server.backup.relay` / `klio.server.backup.maintenance` counters. | +| `outcome` | `success`, `failure`; `not_found` on `klio.plugin.wal.restore_duration` only | `klio.plugin.backup.runs`, `klio.plugin.wal.restore_duration`, `klio.server.backup.relay`, `klio.server.backup.maintenance`, `klio.server.backup.verifications`, and all WAL duration histograms (`klio.server.wal.*_duration`, `klio.client.wal.block_duration`). | +| `cache_hit` | `true` (served from the prefetch spool), `false` (downloaded) | `klio.plugin.wal.restore_duration`. | | `failure_category` | `repository_error`, `source_error`, `verification`, `timeout`, `canceled`, `unknown` | `klio.plugin.backup.runs` failure data points only. | | `path` | `put` (WAL ingest), `get` (WAL serve) | `klio.server.wal.block_duration`, `klio.client.wal.block_duration`. | | `stage` | put: `wrap`, `write`, `flush`, `send` (client); get: `read`, `unwrap`, `send` | `klio.server.wal.block_duration`, `klio.client.wal.block_duration`. | @@ -294,6 +295,21 @@ to the server-side `klio.server.wal.latest_written_timeline` gauge |---|---|---|---| | `klio.client.wal.timeline` | Gauge | - | Timeline ID the WAL streaming client is currently streaming. Carries `cluster_name` | +### WAL restore duration (plugin sidecar) + +The plugin sidecar times every WAL file it restores for PostgreSQL +(`restore_command`), from the request to the file being in place. This +is the restore latency PostgreSQL sees, including prefetch hits that +never reach the Klio server and so are missing from +`klio.server.wal.get_duration`. + +| Metric Name | Type | Unit | Description | +|---|---|---|---| +| `klio.plugin.wal.restore_duration` | Histogram | ns | Per-file end-to-end duration of a WAL restore, split by `outcome` (`success`, `failure`, or `not_found` when PostgreSQL asks for a WAL file that was never archived, which is normal at the end of the archive). Carries `cache_hit`, `tier` and `cluster_name` | + +It uses the same buckets as the per-file server histograms. A +prefetch hit is usually faster than the smallest bucket. + ### Backup verification metrics (server) As part of processing each backup the server verifies it. Verification diff --git a/observability/grafana/client.go b/observability/grafana/client.go index ddf5087b1..f70814254 100644 --- a/observability/grafana/client.go +++ b/observability/grafana/client.go @@ -30,14 +30,16 @@ import ( // emitted by the Klio plugin sidecar that runs in each PostgreSQL pod: the // 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_*`). Both -// families carry a cluster_name, so every panel groups by cluster_name and is -// scoped by $namespace and $cluster; nothing folds several clusters that share -// a namespace into a single value. +// child process (`klio.client.wal.*`, exported as `klio_client_wal_*`), and +// the WAL restores it serves to PostgreSQL (`klio.plugin.wal.*`, exported as +// `klio_plugin_wal_*`). All three families carry a cluster_name, so every +// panel groups by cluster_name and is scoped by $namespace and $cluster; +// nothing folds several clusters that share a namespace into a single value. func clientPanels() []sizedPanel { return []sizedPanel{ sized(gridWidth, descriptionPanelHeight, descriptionPanel( - "Backup lifecycle and WAL streaming as seen by the plugin sidecar running in each PostgreSQL pod.")), + "Backup lifecycle, WAL streaming and WAL restore as seen by the plugin sidecar running in each "+ + "PostgreSQL pod.")), // Backup activity. // Current backup state, grouped by cluster so a namespace hosting // several clusters shows one series each instead of a folded total. @@ -149,5 +151,55 @@ func clientPanels() []sizedPanel { clientMatcher, "{{cluster_name}}")..., ).Description("50th/90th/99th-percentile duration of the client's gRPC send of a WAL block to the "+ "server, per cluster. Reflects all activity since the server last restarted.")), + + // WAL restore. Every RESTORE_WAL request the plugin serves to PostgreSQL + // records klio.plugin.wal.restore_duration. The percentiles keep only + // successful restores and split on cache_hit: a prefetch hit is a local + // rename and a miss waits on a download, so pooled together the + // percentiles would drift with the hit ratio instead of describing + // either case. + sized(largestPanelWidth, largePanelHeight, + timeseriesPanel("WAL restore duration percentiles (rate)", units.Nanoseconds, + quantileTargets("klio_plugin_wal_restore_duration_nanoseconds_bucket", "le, cluster_name, cache_hit", + "outcome=\"success\","+clientMatcher, "{{cluster_name}} / cache_hit={{cache_hit}}")..., + ).Description("50th/90th/99th-percentile end-to-end duration of successful WAL restores served by the "+ + "plugin to PostgreSQL, per cluster, split by whether the WAL was already in the prefetch spool "+ + "(cache_hit=true) or had to be downloaded (cache_hit=false). Reflects recent activity, over the "+ + "rate interval window.")), + + sized(largestPanelWidth, largePanelHeight, + timeseriesPanel("WAL restore duration percentiles (total)", units.Nanoseconds, + quantileTargetsAbsolute("klio_plugin_wal_restore_duration_nanoseconds_bucket", + "le, cluster_name, cache_hit", "outcome=\"success\","+clientMatcher, + "{{cluster_name}} / cache_hit={{cache_hit}}")..., + ).Description("50th/90th/99th-percentile end-to-end duration of successful WAL restores served by the "+ + "plugin to PostgreSQL, per cluster, split by whether the WAL was already in the prefetch spool "+ + "(cache_hit=true) or had to be downloaded (cache_hit=false). Reflects all activity since the "+ + "plugin sidecar last restarted.")), + + sized(largePanelWidth, mediumPanelHeight, timeseriesPanel("WAL restores (rate)", units.OpsPerSecond, + query(fmt.Sprintf("sum by (cluster_name, tier, outcome) "+ + "(rate(klio_plugin_wal_restore_duration_nanoseconds_count{%s}[$__rate_interval]))", clientMatcher), + "{{cluster_name}} {{tier}} / {{outcome}}"), + ).Description("Rate of WAL restores served by the plugin to PostgreSQL, per cluster, split by the tier "+ + "that served them and by outcome. not_found is PostgreSQL asking for a WAL that was never archived, "+ + "the routine end-of-archive signal rather than an error.")), + + sized(largePanelWidth, mediumPanelHeight, timeseriesPanel("WAL restores (total)", units.Number, + query(fmt.Sprintf("sum by (cluster_name, tier, outcome) "+ + "(klio_plugin_wal_restore_duration_nanoseconds_count{%s})", clientMatcher), + "{{cluster_name}} {{tier}} / {{outcome}}"), + ).Description("Total WAL restores served by the plugin to PostgreSQL since the plugin sidecar last "+ + "restarted, per cluster, split by the tier that served them and by outcome.")), + + sized(largePanelWidth, mediumPanelHeight, timeseriesPanel("WAL restore prefetch hit ratio", units.PercentUnit, + query(fmt.Sprintf("sum by (cluster_name) (rate(klio_plugin_wal_restore_duration_nanoseconds_count"+ + "{outcome=\"success\",cache_hit=\"true\",%s}[$__rate_interval])) / "+ + "sum by (cluster_name) (rate(klio_plugin_wal_restore_duration_nanoseconds_count"+ + "{outcome=\"success\",%s}[$__rate_interval]))", clientMatcher, clientMatcher), + "{{cluster_name}}"), + ).Description("Fraction of successful WAL restores served straight from the prefetch spool instead of "+ + "waiting on a download, per cluster. A falling ratio means prefetch is not keeping pace with "+ + "PostgreSQL replay.")), } } diff --git a/observability/grafana/klio-dashboard.json b/observability/grafana/klio-dashboard.json index 330ada6eb..ebcf0d0ca 100644 --- a/observability/grafana/klio-dashboard.json +++ b/observability/grafana/klio-dashboard.json @@ -43,7 +43,7 @@ "repeatDirection": "h", "options": { "mode": "markdown", - "content": "Backup lifecycle and WAL streaming as seen by the plugin sidecar running in each PostgreSQL pod." + "content": "Backup lifecycle, WAL streaming and WAL restore as seen by the plugin sidecar running in each PostgreSQL pod." } }, { @@ -874,6 +874,301 @@ "overrides": [] } }, + { + "type": "timeseries", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "expr": "histogram_quantile(0.50, sum by (le, cluster_name, cache_hit) (rate(klio_plugin_wal_restore_duration_nanoseconds_bucket{outcome=\"success\",k8s_namespace_name=~\"$namespace\",cluster_name=~\"$cluster\"}[$__rate_interval])))", + "instant": false, + "legendFormat": "{{cluster_name}} / cache_hit={{cache_hit}} p50", + "range": true + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "expr": "histogram_quantile(0.90, sum by (le, cluster_name, cache_hit) (rate(klio_plugin_wal_restore_duration_nanoseconds_bucket{outcome=\"success\",k8s_namespace_name=~\"$namespace\",cluster_name=~\"$cluster\"}[$__rate_interval])))", + "instant": false, + "legendFormat": "{{cluster_name}} / cache_hit={{cache_hit}} p90", + "range": true + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "expr": "histogram_quantile(0.99, sum by (le, cluster_name, cache_hit) (rate(klio_plugin_wal_restore_duration_nanoseconds_bucket{outcome=\"success\",k8s_namespace_name=~\"$namespace\",cluster_name=~\"$cluster\"}[$__rate_interval])))", + "instant": false, + "legendFormat": "{{cluster_name}} / cache_hit={{cache_hit}} p99", + "range": true + } + ], + "title": "WAL restore duration percentiles (rate)", + "description": "50th/90th/99th-percentile end-to-end duration of successful WAL restores served by the plugin to PostgreSQL, per cluster, split by whether the WAL was already in the prefetch spool (cache_hit=true) or had to be downloaded (cache_hit=false). Reflects recent activity, over the rate interval window.", + "transparent": false, + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 31 + }, + "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": "histogram_quantile(0.50, sum by (le, cluster_name, cache_hit) (klio_plugin_wal_restore_duration_nanoseconds_bucket{outcome=\"success\",k8s_namespace_name=~\"$namespace\",cluster_name=~\"$cluster\"}))", + "instant": false, + "legendFormat": "{{cluster_name}} / cache_hit={{cache_hit}} p50", + "range": true + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "expr": "histogram_quantile(0.90, sum by (le, cluster_name, cache_hit) (klio_plugin_wal_restore_duration_nanoseconds_bucket{outcome=\"success\",k8s_namespace_name=~\"$namespace\",cluster_name=~\"$cluster\"}))", + "instant": false, + "legendFormat": "{{cluster_name}} / cache_hit={{cache_hit}} p90", + "range": true + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "expr": "histogram_quantile(0.99, sum by (le, cluster_name, cache_hit) (klio_plugin_wal_restore_duration_nanoseconds_bucket{outcome=\"success\",k8s_namespace_name=~\"$namespace\",cluster_name=~\"$cluster\"}))", + "instant": false, + "legendFormat": "{{cluster_name}} / cache_hit={{cache_hit}} p99", + "range": true + } + ], + "title": "WAL restore duration percentiles (total)", + "description": "50th/90th/99th-percentile end-to-end duration of successful WAL restores served by the plugin to PostgreSQL, per cluster, split by whether the WAL was already in the prefetch spool (cache_hit=true) or had to be downloaded (cache_hit=false). Reflects all activity since the plugin sidecar last restarted.", + "transparent": false, + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 31 + }, + "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 (cluster_name, tier, outcome) (rate(klio_plugin_wal_restore_duration_nanoseconds_count{k8s_namespace_name=~\"$namespace\",cluster_name=~\"$cluster\"}[$__rate_interval]))", + "instant": false, + "legendFormat": "{{cluster_name}} {{tier}} / {{outcome}}", + "range": true + } + ], + "title": "WAL restores (rate)", + "description": "Rate of WAL restores served by the plugin to PostgreSQL, per cluster, split by the tier that served them and by outcome. not_found is PostgreSQL asking for a WAL that was never archived, the routine end-of-archive signal rather than an error.", + "transparent": false, + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "gridPos": { + "h": 6, + "w": 8, + "x": 0, + "y": 39 + }, + "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": "timeseries", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "expr": "sum by (cluster_name, tier, outcome) (klio_plugin_wal_restore_duration_nanoseconds_count{k8s_namespace_name=~\"$namespace\",cluster_name=~\"$cluster\"})", + "instant": false, + "legendFormat": "{{cluster_name}} {{tier}} / {{outcome}}", + "range": true + } + ], + "title": "WAL restores (total)", + "description": "Total WAL restores served by the plugin to PostgreSQL since the plugin sidecar last restarted, per cluster, split by the tier that served them and by outcome.", + "transparent": false, + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "gridPos": { + "h": 6, + "w": 8, + "x": 8, + "y": 39 + }, + "repeatDirection": "h", + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom", + "showLegend": true, + "calcs": [] + }, + "tooltip": { + "mode": "", + "sort": "" + } + }, + "fieldConfig": { + "defaults": { + "unit": "none", + "custom": { + "gradientMode": "opacity", + "fillOpacity": 10 + } + }, + "overrides": [] + } + }, + { + "type": "timeseries", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "expr": "sum by (cluster_name) (rate(klio_plugin_wal_restore_duration_nanoseconds_count{outcome=\"success\",cache_hit=\"true\",k8s_namespace_name=~\"$namespace\",cluster_name=~\"$cluster\"}[$__rate_interval])) / sum by (cluster_name) (rate(klio_plugin_wal_restore_duration_nanoseconds_count{outcome=\"success\",k8s_namespace_name=~\"$namespace\",cluster_name=~\"$cluster\"}[$__rate_interval]))", + "instant": false, + "legendFormat": "{{cluster_name}}", + "range": true + } + ], + "title": "WAL restore prefetch hit ratio", + "description": "Fraction of successful WAL restores served straight from the prefetch spool instead of waiting on a download, per cluster. A falling ratio means prefetch is not keeping pace with PostgreSQL replay.", + "transparent": false, + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "gridPos": { + "h": 6, + "w": 8, + "x": 16, + "y": 39 + }, + "repeatDirection": "h", + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom", + "showLegend": true, + "calcs": [] + }, + "tooltip": { + "mode": "", + "sort": "" + } + }, + "fieldConfig": { + "defaults": { + "unit": "percentunit", + "custom": { + "gradientMode": "opacity", + "fillOpacity": 10 + } + }, + "overrides": [] + } + }, { "type": "row", "collapsed": false, @@ -882,7 +1177,7 @@ "h": 1, "w": 24, "x": 0, - "y": 31 + "y": 45 }, "id": 0, "panels": [] @@ -894,7 +1189,7 @@ "h": 2, "w": 24, "x": 0, - "y": 32 + "y": 46 }, "repeatDirection": "h", "options": { @@ -927,7 +1222,7 @@ "h": 6, "w": 8, "x": 0, - "y": 34 + "y": 48 }, "repeatDirection": "h", "options": { @@ -984,7 +1279,7 @@ "h": 6, "w": 8, "x": 8, - "y": 34 + "y": 48 }, "repeatDirection": "h", "options": { @@ -1042,7 +1337,7 @@ "h": 6, "w": 8, "x": 16, - "y": 34 + "y": 48 }, "repeatDirection": "h", "options": { @@ -1099,7 +1394,7 @@ "h": 6, "w": 8, "x": 0, - "y": 40 + "y": 54 }, "repeatDirection": "h", "options": { @@ -1156,7 +1451,7 @@ "h": 6, "w": 8, "x": 8, - "y": 40 + "y": 54 }, "repeatDirection": "h", "options": { @@ -1213,7 +1508,7 @@ "h": 6, "w": 8, "x": 16, - "y": 40 + "y": 54 }, "repeatDirection": "h", "options": { @@ -1319,7 +1614,7 @@ "h": 6, "w": 24, "x": 0, - "y": 46 + "y": 60 }, "repeatDirection": "h", "options": { @@ -1368,7 +1663,7 @@ "h": 6, "w": 8, "x": 0, - "y": 52 + "y": 66 }, "repeatDirection": "h", "options": { @@ -1425,7 +1720,7 @@ "h": 6, "w": 8, "x": 8, - "y": 52 + "y": 66 }, "repeatDirection": "h", "options": { @@ -1482,7 +1777,7 @@ "h": 6, "w": 8, "x": 16, - "y": 52 + "y": 66 }, "repeatDirection": "h", "options": { @@ -1588,7 +1883,7 @@ "h": 6, "w": 24, "x": 0, - "y": 58 + "y": 72 }, "repeatDirection": "h", "options": { @@ -1637,7 +1932,7 @@ "h": 6, "w": 4, "x": 0, - "y": 64 + "y": 78 }, "repeatDirection": "h", "options": { @@ -1695,7 +1990,7 @@ "h": 6, "w": 4, "x": 4, - "y": 64 + "y": 78 }, "repeatDirection": "h", "options": { @@ -1752,7 +2047,7 @@ "h": 6, "w": 4, "x": 8, - "y": 64 + "y": 78 }, "repeatDirection": "h", "options": { @@ -1810,7 +2105,7 @@ "h": 6, "w": 4, "x": 12, - "y": 64 + "y": 78 }, "repeatDirection": "h", "options": { @@ -1868,7 +2163,7 @@ "h": 6, "w": 4, "x": 16, - "y": 64 + "y": 78 }, "repeatDirection": "h", "options": { @@ -1925,7 +2220,7 @@ "h": 6, "w": 4, "x": 20, - "y": 64 + "y": 78 }, "repeatDirection": "h", "options": { @@ -1982,7 +2277,7 @@ "h": 6, "w": 8, "x": 0, - "y": 70 + "y": 84 }, "repeatDirection": "h", "options": { @@ -2036,7 +2331,7 @@ "h": 6, "w": 8, "x": 8, - "y": 70 + "y": 84 }, "repeatDirection": "h", "options": { @@ -2111,7 +2406,7 @@ "h": 6, "w": 8, "x": 16, - "y": 70 + "y": 84 }, "repeatDirection": "h", "options": { @@ -2160,7 +2455,7 @@ "h": 6, "w": 12, "x": 0, - "y": 76 + "y": 90 }, "repeatDirection": "h", "options": { @@ -2211,7 +2506,7 @@ "h": 6, "w": 12, "x": 12, - "y": 76 + "y": 90 }, "repeatDirection": "h", "options": { @@ -2262,7 +2557,7 @@ "h": 6, "w": 12, "x": 0, - "y": 82 + "y": 96 }, "repeatDirection": "h", "options": { @@ -2313,7 +2608,7 @@ "h": 6, "w": 12, "x": 12, - "y": 82 + "y": 96 }, "repeatDirection": "h", "options": { @@ -2384,7 +2679,7 @@ "h": 8, "w": 12, "x": 0, - "y": 88 + "y": 102 }, "repeatDirection": "h", "options": { @@ -2455,7 +2750,7 @@ "h": 8, "w": 12, "x": 12, - "y": 88 + "y": 102 }, "repeatDirection": "h", "options": { @@ -2526,7 +2821,7 @@ "h": 8, "w": 12, "x": 0, - "y": 96 + "y": 110 }, "repeatDirection": "h", "options": { @@ -2597,7 +2892,7 @@ "h": 8, "w": 12, "x": 12, - "y": 96 + "y": 110 }, "repeatDirection": "h", "options": { @@ -2668,7 +2963,7 @@ "h": 8, "w": 12, "x": 0, - "y": 104 + "y": 118 }, "repeatDirection": "h", "options": { @@ -2739,7 +3034,7 @@ "h": 8, "w": 12, "x": 12, - "y": 104 + "y": 118 }, "repeatDirection": "h", "options": { @@ -2790,7 +3085,7 @@ "h": 6, "w": 8, "x": 0, - "y": 112 + "y": 126 }, "repeatDirection": "h", "options": { @@ -2841,7 +3136,7 @@ "h": 6, "w": 8, "x": 8, - "y": 112 + "y": 126 }, "repeatDirection": "h", "options": { @@ -2892,7 +3187,7 @@ "h": 6, "w": 8, "x": 16, - "y": 112 + "y": 126 }, "repeatDirection": "h", "options": { @@ -2943,7 +3238,7 @@ "h": 6, "w": 12, "x": 0, - "y": 118 + "y": 132 }, "repeatDirection": "h", "options": { @@ -2994,7 +3289,7 @@ "h": 6, "w": 12, "x": 12, - "y": 118 + "y": 132 }, "repeatDirection": "h", "options": { @@ -3028,7 +3323,7 @@ "h": 1, "w": 24, "x": 0, - "y": 124 + "y": 138 }, "id": 0, "panels": [] @@ -3040,7 +3335,7 @@ "h": 2, "w": 24, "x": 0, - "y": 125 + "y": 139 }, "repeatDirection": "h", "options": { @@ -3083,7 +3378,7 @@ "h": 6, "w": 8, "x": 0, - "y": 127 + "y": 141 }, "repeatDirection": "h", "options": { @@ -3144,7 +3439,7 @@ "h": 6, "w": 8, "x": 8, - "y": 127 + "y": 141 }, "repeatDirection": "h", "options": { @@ -3195,7 +3490,7 @@ "h": 6, "w": 8, "x": 16, - "y": 127 + "y": 141 }, "repeatDirection": "h", "options": { diff --git a/operator/test/e2e/otel_test.go b/operator/test/e2e/otel_test.go index e976e91cd..677de6bcd 100644 --- a/operator/test/e2e/otel_test.go +++ b/operator/test/e2e/otel_test.go @@ -49,6 +49,7 @@ import ( machineryFeatures "github.com/cloudnative-pg/klio/operator/test/machinery/pkg/features" "github.com/cloudnative-pg/klio/operator/test/machinery/pkg/namespaces" "github.com/cloudnative-pg/klio/operator/test/machinery/pkg/pods" + machineryPostgres "github.com/cloudnative-pg/klio/operator/test/machinery/pkg/postgres" "github.com/cloudnative-pg/klio/operator/test/utils/conditions" "github.com/cloudnative-pg/klio/operator/test/utils/metrics" "github.com/cloudnative-pg/klio/operator/test/utils/templates/certificates" @@ -621,6 +622,9 @@ func assertOTELMetricsReceived( // Verify WAL metrics (Tier 1 and Tier 2) assertOTELWALMetrics(ctx, t, cfg, scenario, promMetrics, collectorPod) + // Verify the plugin WAL restore metric + assertOTELWALRestoreMetrics(ctx, t, cfg, scenario, collectorPod) + // Verify server-side backup processing metrics (relay + maintenance) assertOTELServerBackupProcessingMetrics(t, cfg, scenario, collectorPod) @@ -879,6 +883,111 @@ func assertOTELWALMetrics( } } +// assertOTELWALRestoreMetrics verifies klio.plugin.wal.restore_duration. The +// scenario has a single instance, so PostgreSQL never restores a WAL on its +// own: the test runs CNPG's wal-restore (the restore_command) in the primary, +// once for a WAL Klio holds and once for one it never archived, and checks +// that both a success and a not_found restore are recorded. +func assertOTELWALRestoreMetrics( + ctx context.Context, + t *testing.T, + cfg *envconf.Config, + scenario *otelMetricsScenario, + collectorPod *corev1.Pod, +) { + t.Helper() + + r, err := resources.New(cfg.Client().RESTConfig()) + require.NoError(t, err, "failed to create resources client") + + var cluster cnpgv1.Cluster + require.NoError(t, r.Get(ctx, scenario.cnpgCluster.Name, scenario.namespace.Name, &cluster), + "failed to get CNPG cluster") + var primaryPod corev1.Pod + require.NoError(t, r.Get(ctx, cluster.Status.CurrentPrimary, scenario.namespace.Name, &primaryPod), + "failed to get primary pod") + + // pg_switch_wal returns the end of the segment it closed, so this is the + // name of a complete WAL file the streaming client has sent to Klio. The + // checkpoint runs on its own: psql would print its command tag too. + _, err = machineryPostgres.ExecPostgresQuery(ctx, r, &primaryPod, "postgres", "CHECKPOINT") + require.NoError(t, err, "failed to run a checkpoint") + walName, err := machineryPostgres.ExecPostgresQuery(ctx, r, &primaryPod, "postgres", + "SELECT pg_walfile_name(pg_switch_wal())") + require.NoError(t, err, "failed to switch WAL") + + // restoreWAL runs the restore_command for walName into a scratch file in + // PGDATA, which the plugin sidecar shares, and removes it afterwards. + restoreWAL := func(walName string) error { + var stdout, stderr bytes.Buffer + cmd := []string{ + "sh", "-c", + `/controller/manager wal-restore "$1" "$PGDATA/$1.e2e" && rm -f "$PGDATA/$1.e2e"`, + "sh", walName, + } + + return r.ExecInPod(ctx, primaryPod.Namespace, primaryPod.Name, "postgres", cmd, &stdout, &stderr) + } + + t.Logf("Restoring WAL %s through the plugin", walName) + err = wait.For( + func(context.Context) (bool, error) { + // The WAL may not be on the Klio server yet: retry until it is. + return restoreWAL(walName) == nil, nil + }, + wait.WithTimeout(90*time.Second), + wait.WithInterval(5*time.Second), + ) + require.NoError(t, err, "WAL %s could not be restored through the plugin", walName) + + // A WAL far past the end of the archive, which Klio never received. + require.Error(t, restoreWAL("00000001000000FF000000FF"), + "restoring a never-archived WAL must fail") + + successLabels := map[string]string{ + "outcome": "success", "tier": "tier1", "cache_hit": "false", + "cluster_name": scenario.cnpgCluster.Name, + } + notFoundLabels := map[string]string{ + "outcome": "not_found", "cluster_name": scenario.cnpgCluster.Name, + } + + t.Log("Waiting for the WAL restore duration metric") + err = wait.For( + func(ctx context.Context) (bool, error) { + freshMetrics, fetchErr := fetchCollectorMetrics(ctx, cfg.Client().RESTConfig(), + scenario.namespace.Name, collectorPod.Name) + if fetchErr != nil { + return false, nil //nolint:nilerr // retry on transient errors + } + + success, found := freshMetrics.GetHistogram("klio_plugin_wal_restore_duration_nanoseconds", + successLabels) + if !found { + return false, nil + } + notFound, found := freshMetrics.GetHistogram("klio_plugin_wal_restore_duration_nanoseconds", + notFoundLabels) + if !found { + return false, nil + } + + assert.GreaterOrEqual(t, success.SampleCount, uint64(1), + "should have at least 1 successful WAL restore") + assert.Greater(t, success.BucketCount, 1, + "restore duration histogram should expose explicit buckets, not just +Inf") + assert.GreaterOrEqual(t, notFound.SampleCount, uint64(1), + "should have at least 1 not_found WAL restore") + + return true, nil + }, + wait.WithTimeout(90*time.Second), + wait.WithInterval(10*time.Second), + ) + require.NoError(t, err, "klio_plugin_wal_restore_duration_nanoseconds success and not_found "+ + "data points not found within timeout") +} + // assertOTELServerBackupMetrics verifies the server-side PostgreSQL backup // metrics (klio.server.backup.*) derived from the snapshotted backup metadata. // The SnapshotMetricsCollector emits these on both tiers on its poll interval,