Add a basic metric for metastore DLQ recovery (#4563)

* Add a basic metric for metastore DLQ recovery

* Add debug log for blocks successfully added to the DLQ

* Cover new metric in existing test

Author: aleks-p

pkg/metastore/index/dlq/metrics.go

@@ -0,0 +1,29 @@
+package dlq
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+type metrics struct {
+	recoveryAttempts *prometheus.CounterVec
+}
+
+func newMetrics(reg prometheus.Registerer) *metrics {
+	m := &metrics{
+		recoveryAttempts: prometheus.NewCounterVec(
+			prometheus.CounterOpts{
+				Namespace: "pyroscope",
+				Subsystem: "metastore",
+				Name:      "dlq_recovery_attempts_total",
+				Help:      "Total number of DLQ block recovery attempts by status.",
+			},
+			[]string{"status"},
+		),
+	}
+
+	if reg != nil {
+		reg.MustRegister(m.recoveryAttempts)
+	}
+
+	return m
+}

pkg/metastore/index/dlq/recovery.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/thanos-io/objstore"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -36,18 +37,20 @@ type Recovery struct {
 	logger    log.Logger
 	metastore Metastore
 	bucket    objstore.Bucket
+	metrics   *metrics
 
 	started bool
 	cancel  func()
 	m       sync.Mutex
 }
 
-func NewRecovery(logger log.Logger, config Config, metastore Metastore, bucket objstore.Bucket) *Recovery {
+func NewRecovery(logger log.Logger, config Config, metastore Metastore, bucket objstore.Bucket, reg prometheus.Registerer) *Recovery {
 	return &Recovery{
 		config:    config,
 		logger:    logger,
 		metastore: metastore,
 		bucket:    bucket,
+		metrics:   newMetrics(reg),
 	}
 }
 
@@ -121,37 +124,46 @@ func (r *Recovery) recover(ctx context.Context, path string) (err error) {
 	switch {
 	case err == nil:
 	case errors.Is(err, context.Canceled):
+		r.metrics.recoveryAttempts.WithLabelValues("canceled").Inc()
 		return err
 	case r.bucket.IsObjNotFoundErr(err):
 		// This is somewhat opportunistic: the error is likely caused by a competing recovery
 		// process that has already recovered the block, before we've discovered that the
 		// leadership has changed.
+		r.metrics.recoveryAttempts.WithLabelValues("not_found").Inc()
 		level.Warn(r.logger).Log("msg", "block metadata not found; skipping", "path", path)
 		return nil
 	default:
 		// This is somewhat opportunistic, as we don't know if the error is transient or not.
 		// we should consider an explicit retry mechanism with backoff and a limit on the
 		// number of attempts.
+		r.metrics.recoveryAttempts.WithLabelValues("read_error").Inc()
 		level.Warn(r.logger).Log("msg", "failed to read block metadata; to be retried", "err", err, "path", path)
 		return err
 	}
 
 	var meta metastorev1.BlockMeta
 	if err = meta.UnmarshalVT(b); err != nil {
-		level.Error(r.logger).Log("msg", "invalid block metadata; skipping", "err", err, "path", path)
+		r.metrics.recoveryAttempts.WithLabelValues("unmarshal_error").Inc()
+		level.Error(r.logger).Log("msg", "failed to unmarshal block metadata; skipping", "err", err, "path", path)
 		return nil
 	}
 
 	switch _, err = r.metastore.AddRecoveredBlock(ctx, &metastorev1.AddBlockRequest{Block: &meta}); {
 	case err == nil:
+		r.metrics.recoveryAttempts.WithLabelValues("success").Inc()
+		level.Debug(r.logger).Log("msg", "successfully recovered block from DLQ", "block_id", meta.Id, "path", path)
 		return nil
 	case status.Code(err) == codes.InvalidArgument:
-		level.Error(r.logger).Log("msg", "invalid block metadata", "err", err, "path", path)
+		r.metrics.recoveryAttempts.WithLabelValues("invalid_metadata").Inc()
+		level.Error(r.logger).Log("msg", "block metadata rejected by metastore; skipping", "err", err, "block_id", meta.Id, "path", path)
 		return nil
 	case raftnode.IsRaftLeadershipError(err):
+		r.metrics.recoveryAttempts.WithLabelValues("leadership_change").Inc()
 		level.Warn(r.logger).Log("msg", "leadership change; recovery interrupted", "err", err, "path", path)
 		return err
 	default:
+		r.metrics.recoveryAttempts.WithLabelValues("metastore_error").Inc()
 		level.Error(r.logger).Log("msg", "failed to add block metadata; to be retried", "err", err, "path", path)
 		return err
 	}

pkg/metastore/index/dlq/recovery_test.go

@@ -6,6 +6,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
@@ -51,7 +53,7 @@ func TestRecoverTick(t *testing.T) {
 		addMeta(bucket, meta)
 	}
 
-	r := NewRecovery(test.NewTestingLogger(t), Config{}, srv, bucket)
+	r := NewRecovery(test.NewTestingLogger(t), Config{}, srv, bucket, prometheus.NewRegistry())
 	r.recoverTick(context.Background())
 
 	expected := []*metastorev1.BlockMeta{
@@ -65,6 +67,10 @@ func TestRecoverTick(t *testing.T) {
 		require.Equal(t, actual[i].Id, expected[i].Id)
 		require.Equal(t, actual[i].Shard, expected[i].Shard)
 	}
+
+	assert.Equal(t, 3.0, testutil.ToFloat64(r.metrics.recoveryAttempts.WithLabelValues("success")))
+	assert.Equal(t, 0.0, testutil.ToFloat64(r.metrics.recoveryAttempts.WithLabelValues("unmarshal_error")))
+	assert.Equal(t, 0.0, testutil.ToFloat64(r.metrics.recoveryAttempts.WithLabelValues("invalid_metadata")))
 }
 
 func TestNotRaftLeader(t *testing.T) {
@@ -89,10 +95,13 @@ func TestNotRaftLeader(t *testing.T) {
 		addMeta(bucket, meta)
 	}
 
-	r := NewRecovery(test.NewTestingLogger(t), Config{}, srv, bucket)
+	r := NewRecovery(test.NewTestingLogger(t), Config{}, srv, bucket, prometheus.NewRegistry())
 	r.recoverTick(context.Background())
 
 	assert.Equal(t, 1, len(bucket.Objects()))
+
+	assert.Equal(t, 1.0, testutil.ToFloat64(r.metrics.recoveryAttempts.WithLabelValues("metastore_error")))
+	assert.Equal(t, 0.0, testutil.ToFloat64(r.metrics.recoveryAttempts.WithLabelValues("success")))
 }
 
 func TestStartStop(t *testing.T) {
@@ -129,7 +138,7 @@ func TestStartStop(t *testing.T) {
 		addMeta(bucket, meta)
 	}
 
-	r := NewRecovery(test.NewTestingLogger(t), Config{CheckInterval: time.Millisecond * 10}, srv, bucket)
+	r := NewRecovery(test.NewTestingLogger(t), Config{CheckInterval: time.Millisecond * 10}, srv, bucket, prometheus.NewRegistry())
 	r.Start()
 	defer r.Stop()
 
@@ -150,6 +159,8 @@ func TestStartStop(t *testing.T) {
 		require.Equal(t, actual[i].Id, expected[i].Id)
 		require.Equal(t, actual[i].Shard, expected[i].Shard)
 	}
+
+	assert.Equal(t, 3.0, testutil.ToFloat64(r.metrics.recoveryAttempts.WithLabelValues("success")))
 }
 
 func addMeta(bucket *memory.InMemBucket, meta *metastorev1.BlockMeta) {

pkg/metastore/metastore.go

@@ -175,7 +175,7 @@ func New(
 	m.indexService = NewIndexService(m.logger, m.raft, m.leaderRead, m.index, m.placement)
 	m.tenantService = NewTenantService(m.logger, m.followerRead, m.index)
 	m.queryService = NewQueryService(m.logger, m.followerRead, m.index)
-	m.recovery = dlq.NewRecovery(logger, config.Index.Recovery, m.indexService, bucket)
+	m.recovery = dlq.NewRecovery(logger, config.Index.Recovery, m.indexService, bucket, m.reg)
 	m.cleaner = cleaner.NewCleaner(m.logger, m.overrides, config.Index.Cleaner, m.indexService)
 
 	// These are the services that only run on the raft leader.

pkg/segmentwriter/segment.go

@@ -676,6 +676,7 @@ func (sw *segmentsWriter) storeMetadata(ctx context.Context, meta *metastorev1.B
 	}()
 
 	if err = s.sw.storeMetadataDLQ(ctx, meta); err == nil {
+		level.Debug(s.logger).Log("msg", "successfully wrote block metadata to DLQ", "block_id", meta.Id)
 		return nil
 	}
 

pkg/segmentwriter/segment_test.go

@@ -392,7 +392,7 @@ func TestDLQRecoveryMock(t *testing.T) {
 		Return(&metastorev1.AddBlockResponse{}, nil)
 	recovery := dlq.NewRecovery(test.NewTestingLogger(t), dlq.Config{
 		CheckInterval: 100 * time.Millisecond,
-	}, srv, sw.bucket)
+	}, srv, sw.bucket, nil)
 	recovery.Start()
 	defer recovery.Stop()