feat(orchestrator): Record batch size, dependency count, and speculated paths per batch as histograms

Author: roychying

platform/metrics/README.md

@@ -91,3 +91,5 @@ There is no default bucket set. The package exports five common sets:
 | `LandLatencyBuckets` | ~1m – 24h | End-to-end request lifecycles, from accepting a land request to its terminal outcome |
 
 Pass one of these sets or a custom `tally.DurationBuckets` to `Begin` or `NamedHistogram`.
+
+For counts recorded with `RecordValue`, such as how many items a unit of work carries or depends on, use `CountBuckets` (0 – 144, Fibonacci-spaced, with zero in its own bucket).

platform/metrics/metrics.go

@@ -181,6 +181,12 @@ var (
 	}
 )
 
+// CountBuckets suits small cardinalities recorded with RecordValue, such as how
+// many items a unit of work carries or depends on. Zero has its own bucket
+// because "none" is a meaningful answer, and the Fibonacci spacing keeps
+// resolution where counts usually fall while still separating outliers.
+var CountBuckets = tally.ValueBuckets{0, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89, 144}
+
 // Op tracks the lifecycle of a named operation. It captures the start time on
 // creation, emits a {name}.start counter, and records the duration and result
 // on a {name}.finish histogram when Complete is called.

platform/metrics/metrics_test.go

@@ -233,6 +233,13 @@ func TestLatencyBuckets_Sorted(t *testing.T) {
 	}
 }
 
+func TestCountBuckets_Sorted(t *testing.T) {
+	for i := 1; i < len(CountBuckets); i++ {
+		assert.Greater(t, CountBuckets[i], CountBuckets[i-1],
+			"CountBuckets[%d] (%v) must be greater than CountBuckets[%d] (%v)", i, CountBuckets[i], i-1, CountBuckets[i-1])
+	}
+}
+
 // counterKeys extracts map keys for error messages.
 func counterKeys(m map[string]tally.CounterSnapshot) []string {
 	keys := make([]string, 0, len(m))

submitqueue/orchestrator/controller/dependencyanalysis/dependencyanalysis.go

@@ -213,6 +213,11 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 			metrics.NamedCounter(c.metricsScope, opName, "batch_store_errors", 1, metrics.TagsFromContext(ctx)...)
 			return fmt.Errorf("failed to mark batch %s created: %w", batch.ID, err)
 		}
+		// Recorded on the Creating→Created transition, which happens once per batch; abandoned batches never reach it.
+		metrics.NamedHistogram(c.metricsScope, opName, "batch_size", metrics.CountBuckets, metrics.TagsFromContext(ctx)...).
+			RecordValue(float64(len(batch.Contains)))
+		metrics.NamedHistogram(c.metricsScope, opName, "batch_dependencies", metrics.CountBuckets, metrics.TagsFromContext(ctx)...).
+			RecordValue(float64(len(batch.Dependencies)))
 		c.logger.Infow("batch created",
 			"batch_id", batch.ID,
 			"queue", batch.Queue,

submitqueue/orchestrator/controller/dependencyanalysis/dependencyanalysis_test.go

@@ -113,6 +113,11 @@ func storageFactoryFor(ctrl *gomock.Controller, store orchstorage.Storage) *orch
 // conflicts.
 func newTestController(t *testing.T, ctrl *gomock.Controller, store orchstorage.Storage, analyzer conflict.Analyzer, publisher *queuemock.MockPublisher) *Controller {
 	t.Helper()
+	return newTestControllerWithScope(t, ctrl, store, analyzer, publisher, tally.NoopScope)
+}
+
+func newTestControllerWithScope(t *testing.T, ctrl *gomock.Controller, store orchstorage.Storage, analyzer conflict.Analyzer, publisher *queuemock.MockPublisher, scope tally.Scope) *Controller {
+	t.Helper()
 
 	if analyzer == nil {
 		analyzer = all.New(analyzerCfg)
@@ -132,7 +137,7 @@ func newTestController(t *testing.T, ctrl *gomock.Controller, store orchstorage.
 	})
 	require.NoError(t, err)
 
-	return NewController(zaptest.NewLogger(t).Sugar(), tally.NoopScope, storageFactoryFor(ctrl, store),
+	return NewController(zaptest.NewLogger(t).Sugar(), scope, storageFactoryFor(ctrl, store),
 		analyzerFactory, registry, topickey.TopicKeyDependencyAnalysis, "orchestrator-dependency-analysis")
 }
 
@@ -248,8 +253,17 @@ func TestController_Process_AnalyzesAndTransitionsToCreated(t *testing.T) {
 	store.EXPECT().GetRequestStore().Return(requestStoreFor(ctrl, liveRequest())).AnyTimes()
 	store.EXPECT().GetRequestBatchStore().Return(noPriorEnrollment(ctrl)).AnyTimes()
 
-	controller := newTestController(t, ctrl, store, nil, nil)
+	scope := tally.NewTestScope("", nil)
+	controller := newTestControllerWithScope(t, ctrl, store, nil, nil, scope)
 	require.NoError(t, controller.Process(context.Background(), newDelivery(t, ctrl, batch.ID, testQueue)))
+
+	histograms := scope.Snapshot().Histograms()
+	size, ok := histograms["dependency_analysis_controller.process.batch_size+"]
+	require.True(t, ok)
+	assert.Equal(t, int64(1), size.Values()[1])
+	dependencies, ok := histograms["dependency_analysis_controller.process.batch_dependencies+"]
+	require.True(t, ok)
+	assert.Equal(t, int64(1), dependencies.Values()[2])
 }
 
 // The analyzer may report one in-flight batch several times when more than one
@@ -325,8 +339,10 @@ func TestController_Process_RedeliveryAfterTransitionOnlyRepublishes(t *testing.
 	publisher.EXPECT().Publish(gomock.Any(), "log", gomock.Any()).Return(nil).AnyTimes()
 	publisher.EXPECT().Publish(gomock.Any(), "speculate", gomock.Any()).Return(nil)
 
-	controller := newTestController(t, ctrl, store, nil, publisher)
+	scope := tally.NewTestScope("", nil)
+	controller := newTestControllerWithScope(t, ctrl, store, nil, publisher, scope)
 	require.NoError(t, controller.Process(context.Background(), newDelivery(t, ctrl, batch.ID, testQueue)))
+	assert.Empty(t, scope.Snapshot().Histograms(), "a batch already created must not be measured again")
 }
 
 // A failure part-way through the index loop leaves the batch in Creating, so

submitqueue/orchestrator/controller/speculate/finalize.go

@@ -275,6 +275,10 @@ func (c *Controller) commitOutcome(ctx context.Context, snap *snapshot, batch en
 	if err != nil {
 		return landed, c.attributed(ctx, err, entity.BatchSubject(batch.ID))
 	}
+	if landed {
+		metrics.NamedHistogram(c.metricsScope, opName, "paths", metrics.CountBuckets, metrics.TagsFromContext(ctx)...).
+			RecordValue(float64(len(snap.pathSets[batch.ID].Paths)))
+	}
 	return landed, nil
 }
 

submitqueue/orchestrator/controller/speculate/run_test.go

@@ -1588,6 +1588,10 @@ func TestRun_BypassesUnsettledDependenciesWithFullCoverage(t *testing.T) {
 	counter, ok := h.metrics.Snapshot().Counters()["test.speculate_controller.process.bypass+"]
 	require.True(t, ok)
 	assert.EqualValues(t, 1, counter.Value())
+
+	paths, ok := h.metrics.Snapshot().Histograms()["test.speculate_controller.process.paths+"]
+	require.True(t, ok, "a finalized head records how many paths it speculated")
+	assert.Equal(t, int64(1), paths.Values()[5])
 }
 
 // The land stage publishes landing as its first act on the dispatch. Both