kvserver: test metrics in TestBaseQueueCallback* and TestReplicateQueueDecommissionScannerDisabled

his commit extends TestBaseQueueCallback* and
TestReplicateQueueDecommissionScannerDisabled to also verify metric updates.

Author: wenyihu6

pkg/kv/kvserver/queue_test.go

@@ -110,6 +110,8 @@ func makeTestBaseQueue(name string, impl queueImpl, store *Store, cfg queueConfi
 	cfg.pending = metric.NewGauge(metric.Metadata{Name: "pending"})
 	cfg.processingNanos = metric.NewCounter(metric.Metadata{Name: "processingnanos"})
 	cfg.purgatory = metric.NewGauge(metric.Metadata{Name: "purgatory"})
+	cfg.enqueueAdd = metric.NewCounter(metric.Metadata{Name: "enqueueadd"})
+	cfg.enqueueUnexpectedError = metric.NewCounter(metric.Metadata{Name: "enqueueunexpectederror"})
 	cfg.disabledConfig = testQueueEnabled
 	return newBaseQueue(name, impl, store, cfg)
 }
@@ -1326,6 +1328,7 @@ func TestBaseQueueCallbackOnEnqueueResult(t *testing.T) {
 				t.Fatal("unexpected call to onProcessResult")
 			},
 		})
+		require.Equal(t, bq.enqueueAdd.Count(), int64(1))
 		require.True(t, queued)
 	})
 
@@ -1352,6 +1355,7 @@ func TestBaseQueueCallbackOnEnqueueResult(t *testing.T) {
 					t.Fatal("unexpected call to onProcessResult")
 				},
 			})
+			require.Equal(t, int64(i+1), bq.enqueueAdd.Count())
 			require.True(t, queued)
 		}
 		// Set range id back to 1.
@@ -1372,6 +1376,8 @@ func TestBaseQueueCallbackOnEnqueueResult(t *testing.T) {
 				t.Fatal("unexpected call to onProcessResult")
 			},
 		})
+		require.Equal(t, int64(0), bq.enqueueAdd.Count())
+		require.Equal(t, int64(1), bq.enqueueUnexpectedError.Count())
 		require.False(t, queued)
 	})
 	t.Run("stopped", func(t *testing.T) {
@@ -1390,6 +1396,8 @@ func TestBaseQueueCallbackOnEnqueueResult(t *testing.T) {
 			},
 		})
 		require.False(t, queued)
+		require.Equal(t, int64(0), bq.enqueueAdd.Count())
+		require.Equal(t, int64(1), bq.enqueueUnexpectedError.Count())
 	})
 
 	t.Run("alreadyqueued", func(t *testing.T) {
@@ -1407,6 +1415,8 @@ func TestBaseQueueCallbackOnEnqueueResult(t *testing.T) {
 			},
 		})
 		require.True(t, queued)
+		require.Equal(t, int64(1), bq.enqueueAdd.Count())
+		require.Equal(t, int64(0), bq.enqueueUnexpectedError.Count())
 
 		// Inserting again on the same range id should fail.
 		queued, _ = bq.testingAddWithCallback(ctx, r, 1.0, processCallback{
@@ -1419,6 +1429,8 @@ func TestBaseQueueCallbackOnEnqueueResult(t *testing.T) {
 			},
 		})
 		require.False(t, queued)
+		require.Equal(t, int64(1), bq.enqueueAdd.Count())
+		require.Equal(t, int64(0), bq.enqueueUnexpectedError.Count())
 	})
 
 	t.Run("purgatory", func(t *testing.T) {
@@ -1442,6 +1454,8 @@ func TestBaseQueueCallbackOnEnqueueResult(t *testing.T) {
 			},
 		})
 		require.False(t, queued)
+		require.Equal(t, int64(0), bq.enqueueAdd.Count())
+		require.Equal(t, int64(0), bq.enqueueUnexpectedError.Count())
 	})
 
 	t.Run("processing", func(t *testing.T) {
@@ -1453,7 +1467,7 @@ func TestBaseQueueCallbackOnEnqueueResult(t *testing.T) {
 		item.setProcessing()
 		bq.addLocked(item)
 		// Inserting a range that is already being processed should not enqueue again.
-		requeued, _ := bq.testingAddWithCallback(ctx, r, 1.0, processCallback{
+		markedAsRequeued, _ := bq.testingAddWithCallback(ctx, r, 1.0, processCallback{
 			onEnqueueResult: func(indexOnHeap int, err error) {
 				require.Equal(t, -1, indexOnHeap)
 				require.ErrorIs(t, err, errReplicaAlreadyProcessing)
@@ -1462,7 +1476,9 @@ func TestBaseQueueCallbackOnEnqueueResult(t *testing.T) {
 				t.Fatal("unexpected call to onProcessResult")
 			},
 		})
-		require.True(t, requeued)
+		require.True(t, markedAsRequeued)
+		require.Equal(t, int64(0), bq.enqueueAdd.Count())
+		require.Equal(t, int64(0), bq.enqueueUnexpectedError.Count())
 	})
 	t.Run("fullqueue", func(t *testing.T) {
 		testQueue := &testQueueImpl{}
@@ -1482,6 +1498,8 @@ func TestBaseQueueCallbackOnEnqueueResult(t *testing.T) {
 			},
 		})
 		require.True(t, queued)
+		require.Equal(t, int64(1), bq.enqueueAdd.Count())
+		require.Equal(t, int64(0), bq.enqueueUnexpectedError.Count())
 	})
 }
 

pkg/kv/kvserver/replicate_queue_test.go

@@ -2484,8 +2484,10 @@ func TestReplicateQueueDecommissionScannerDisabled(t *testing.T) {
 				value = store.Metrics().DecommissioningRangeCount.Value()
 			case "enqueue":
 				value = store.Metrics().DecommissioningNudgerEnqueue.Count()
-			case "not_leaseholder_or_invalid_lease":
-				value = store.Metrics().DecommissioningNudgerNotLeaseholderOrInvalidLease.Count()
+			case "enqueue_success":
+				value = store.Metrics().DecommissioningNudgerEnqueueSuccess.Count()
+			case "process_success":
+				value = store.Metrics().DecommissioningNudgerProcessSuccess.Count()
 			default:
 				t.Fatalf("unknown metric type: %s", metricType)
 			}
@@ -2506,9 +2508,10 @@ func TestReplicateQueueDecommissionScannerDisabled(t *testing.T) {
 
 	// Wait for the enqueue logic to trigger and validate metrics were updated.
 	testutils.SucceedsSoon(t, func() error {
+		// TODO(wenyihu6): is there a race condition here where we might not observe
+		// decommissioning_ranges increasing?
 		afterDecommissioningRanges := getDecommissioningNudgerMetricValue(t, tc, "decommissioning_ranges")
 		afterEnqueued := getDecommissioningNudgerMetricValue(t, tc, "enqueue")
-
 		if afterDecommissioningRanges <= initialDecommissioningRanges {
 			return errors.New("expected DecommissioningRangeCount to increase")
 		}
@@ -2531,4 +2534,8 @@ func TestReplicateQueueDecommissionScannerDisabled(t *testing.T) {
 		}
 		return nil
 	})
+	afterEnqueueSuccess := getDecommissioningNudgerMetricValue(t, tc, "enqueue_success")
+	require.Greater(t, afterEnqueueSuccess, int64(0))
+	afterProcessSuccess := getDecommissioningNudgerMetricValue(t, tc, "process_success")
+	require.Greater(t, afterProcessSuccess, int64(0))
 }