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

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)

core/internal/cnpgi/metrics_test.go

@@ -52,6 +52,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,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.

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

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,32 +136,42 @@ 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)
 	}
 	if err != nil {
 		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.
@@ -154,21 +180,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 restoreOutcome{tier: t, cacheHit: cacheHit}, nil
 		}
 		if !errors.Is(err, errWALNotFound) {
-			return err
+			return restoreOutcome{tier: t}, err
 		}
 	}
 
-	return errWALNotFound
+	return restoreOutcome{tier: lastTier}, errWALNotFound
 }
 
 // availableTiers returns the tiers the user has opted in to as recovery
@@ -297,15 +325,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)