feat: add klio.plugin.wal.restore_duration end-to-end metric

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)

core/internal/cnpgi/metrics_test.go

@@ -57,6 +57,7 @@ func setupTestMeter(t *testing.T) *sdkmetric.ManualReader {
 	})
 
 	opentelemetry.InitPluginBackupMetrics()
+	opentelemetry.InitPluginWalMetrics()
 
 	return reader
 }

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 "<name>.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.

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)
+		})
+	}
+}