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

@@ -111,6 +111,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)
 }
@@ -1321,6 +1323,7 @@ func TestBaseQueueCallbackOnEnqueueResult(t *testing.T) {
 				t.Fatal("unexpected call to onProcessResult")
 			},
 		})
+		require.Equal(t, bq.enqueueAdd.Count(), int64(1))
 		require.True(t, queued)
 	})
 
@@ -1347,6 +1350,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.
@@ -1367,6 +1371,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) {
@@ -1385,6 +1391,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) {
@@ -1402,6 +1410,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{
@@ -1414,6 +1424,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) {
@@ -1437,6 +1449,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) {
@@ -1448,7 +1462,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)
@@ -1457,7 +1471,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{}
@@ -1477,6 +1493,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

@@ -2485,8 +2485,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)
 			}
@@ -2507,9 +2509,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")
 		}
@@ -2532,4 +2535,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))
 }