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

core/cmd/cnpgi/restore.go

@@ -28,6 +28,7 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 
 	"github.com/cloudnative-pg/klio/core/internal/cnpgi"
+	"github.com/cloudnative-pg/klio/core/internal/opentelemetry"
 	"github.com/cloudnative-pg/klio/core/pkg/config"
 )
 
@@ -40,6 +41,9 @@ var restoreJobCmd = &cobra.Command{
 	Hidden: true,
 	Args:   cobra.ExactArgs(1),
 	RunE: func(cmd *cobra.Command, args []string) error {
+		shutdownOtel := opentelemetry.Init(cmd.Context())
+		defer shutdownOtel()
+
 		configFile, _ := cmd.Root().PersistentFlags().GetString("config")
 		pluginPath, _ := cmd.Flags().GetString("plugin-path")
 		clusterName, _ := cmd.Flags().GetString("cluster-name")

core/internal/cnpgi/metrics.go

@@ -53,6 +53,38 @@ 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,
+	info restoreOutcome,
+	clusterName string,
+) {
+	restoreTier := info.tier
+	if restoreTier == "" {
+		restoreTier = tierUnknown
+	}
+	if clusterName == "" {
+		clusterName = unknownAttributeValue
+	}
+
+	opentelemetry.PluginWal.RestoreDuration.Record(ctx, duration.Nanoseconds(),
+		metric.WithAttributes(
+			info.result.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,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

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

@@ -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"
 )
 
@@ -48,6 +49,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 +101,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 +125,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 +139,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 +155,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 tierUnknown; 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, tierUnknown, 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, tierUnknown, errors.New("no WAL tier configured")
 	}
 
 	// Try the previously-successful tier first, when both are available.
@@ -154,21 +208,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
@@ -233,6 +289,9 @@ func (mgr *grpcClientManager) getClient(ctx context.Context, opts walRestoreOpti
 		address = configuration.Client.Wal.Address
 	case tier2:
 		address = configuration.Client.Wal.Tier2Address
+	case tierUnknown:
+		// Only ever a metric attribute value, never a tier to connect to.
+		fallthrough
 	default:
 		return nil, fmt.Errorf("unknown tier %q", opts.targetTier)
 	}
@@ -297,15 +356,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)