From fc530c118478e81ac1a416a2e7a992d7bceca9db Mon Sep 17 00:00:00 2001 From: Brian VanLoo Date: Fri, 21 Aug 2026 12:10:54 -0700 Subject: [PATCH] Add observability for multicursor queue-state resolution loss (#11695) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## What changed? Adds metrics for how often and by how much queue slices fail to narrow their predicate, and how large persisted queue state actually is. - `queue_slice_pending_keys` — histogram, recorded on every narrowing attempt (declined or succeeded). This is the distribution `queueShrinkPredicateMaxPendingKeys` should be sized against. - `shard_info_size` / `queue_state_size` — histograms recorded when a shard record is actually written, giving whole-record and per-category size. - `queue_state_size_total` / `queue_slice_count_total` — counters paired with the histograms above (and with the existing `queue_slice_count`), so an exact bytes-per-slice ratio is possible. - `queue_slice_count` gains a `task_category` tag (previously untagged beyond `operation`). These are only metrics changes - no behavior changes. ## Why? A slice only narrows its predicate below `queueShrinkPredicateMaxPendingKeys` (10) pending namespaces; above that it stays universal and re-reads the whole range every time. Raising that threshold safely requires knowing the pending-key distribution and the persisted size. This PR is the baseline for evaluating a follow-on encoding. There are two counters because this server's tally-backed Prometheus reporter doesn't preserve the true recorded value when a histogram flushes — it replays each sample as its bucket's upper bound, so a histogram's `_sum` has no more precision than its buckets. ## How did you test it? - [x] built - [x] covered by existing tests - [x] added new unit test(s) - [x] run locally and tested manually (`queue_predicate_resolution_loss` confirmed live against a local server under forced narrowing-decline conditions) ## Potential risks --- common/metrics/metric_defs.go | 5 ++ service/history/queues/queue_base.go | 7 +- service/history/queues/queue_base_test.go | 97 +++++++++++++++++++++++ service/history/queues/slice.go | 1 + service/history/queues/slice_test.go | 75 ++++++++++++++++++ service/history/shard/context_impl.go | 15 ++++ service/history/shard/context_test.go | 96 ++++++++++++++++++++++ 7 files changed, 295 insertions(+), 1 deletion(-) diff --git a/common/metrics/metric_defs.go b/common/metrics/metric_defs.go index 74a39eb568..61eab914af 100644 --- a/common/metrics/metric_defs.go +++ b/common/metrics/metric_defs.go @@ -724,6 +724,9 @@ var ( HistoryCount = NewDimensionlessHistogramDef("history_count") TasksCompletedPerShardInfoUpdate = NewDimensionlessHistogramDef("tasks_per_shardinfo_update") TimeBetweenShardInfoUpdates = NewTimerDef("time_between_shardinfo_update") + ShardInfoSize = NewBytesHistogramDef("shard_info_size") + QueueStateSize = NewBytesHistogramDef("queue_state_size") + QueueStateSizeTotal = NewCounterDef("queue_state_size_total") SearchAttributesSize = NewBytesHistogramDef("search_attributes_size") MemoSize = NewBytesHistogramDef("memo_size") TooManyPendingChildWorkflows = NewCounterDef( @@ -953,6 +956,8 @@ var ( QueueScheduleLatency = NewTimerDef("queue_latency_schedule") // latency for scheduling 100 tasks in one task channel QueueReaderCountHistogram = NewDimensionlessHistogramDef("queue_reader_count") QueueSliceCountHistogram = NewDimensionlessHistogramDef("queue_slice_count") + QueueSliceCountTotal = NewCounterDef("queue_slice_count_total") + QueueSlicePendingKeys = NewDimensionlessHistogramDef("queue_slice_pending_keys") QueueActionCounter = NewCounterDef("queue_actions") QueueAlertShadowCounter = NewCounterDef("queue_alert_shadow") QueuePredicateResolutionLoss = NewCounterDef( diff --git a/service/history/queues/queue_base.go b/service/history/queues/queue_base.go index 974e66e474..16beb86992 100644 --- a/service/history/queues/queue_base.go +++ b/service/history/queues/queue_base.go @@ -329,7 +329,12 @@ func (p *queueBase) checkpoint() { } } metrics.QueueReaderCountHistogram.With(p.metricsHandler).Record(int64(len(readerScopes))) - metrics.QueueSliceCountHistogram.With(p.metricsHandler).Record(int64(p.monitor.GetTotalSliceCount())) + sliceCount := int64(p.monitor.GetTotalSliceCount()) + categoryTag := metrics.TaskCategoryTag(p.category.Name()) + // The counter is a true accumulator; the histogram's _sum is not, since tally's Prometheus + // reporter replays each sample as its bucket's upper bound, not the recorded value. + metrics.QueueSliceCountHistogram.With(p.metricsHandler).Record(sliceCount, categoryTag) + metrics.QueueSliceCountTotal.With(p.metricsHandler).Record(sliceCount, categoryTag) metrics.PendingTasksCounter.With(p.metricsHandler).Record(int64(p.monitor.GetTotalPendingTaskCount())) // NOTE: Must range-complete task first. diff --git a/service/history/queues/queue_base_test.go b/service/history/queues/queue_base_test.go index 313120987e..a7c0891f80 100644 --- a/service/history/queues/queue_base_test.go +++ b/service/history/queues/queue_base_test.go @@ -17,6 +17,7 @@ import ( "go.temporal.io/server/common/dynamicconfig" "go.temporal.io/server/common/log" "go.temporal.io/server/common/metrics" + "go.temporal.io/server/common/metrics/metricstest" "go.temporal.io/server/common/persistence" "go.temporal.io/server/common/persistence/serialization" "go.temporal.io/server/common/predicates" @@ -456,6 +457,102 @@ func (s *queueBaseSuite) TestCheckPoint_NoPendingTasks() { s.True(exclusiveReaderHighWatermark.CompareTo(base.exclusiveDeletionHighWatermark) == 0) } +func (s *queueBaseSuite) TestCheckPoint_RecordsSliceCountWithTaskCategoryTag() { + numSlices := 3 + scopes := NewRandomScopes(numSlices) + queueState := &queueState{ + readerScopes: map[int64][]Scope{ + DefaultReaderId: scopes, + }, + exclusiveReaderHighWatermark: tasks.MaximumKey, + } + persistenceState := ToPersistenceQueueState(queueState) + + mockShard := shard.NewTestContext( + s.controller, + &persistencespb.ShardInfo{ + ShardId: 0, + RangeId: 10, + QueueStates: map[int32]*persistencespb.QueueState{ + int32(tasks.CategoryIDTimer): persistenceState, + }, + }, + s.config, + ) + mockShard.Resource.ClusterMetadata.EXPECT().GetCurrentClusterName().Return(cluster.TestCurrentClusterName).AnyTimes() + mockShard.Resource.ClusterMetadata.EXPECT().GetAllClusterInfo().Return(cluster.TestAllClusterInfo).AnyTimes() + + captureHandler := metricstest.NewCaptureHandler() + capture := captureHandler.StartCapture() + defer captureHandler.StopCapture(capture) + s.metricsHandler = captureHandler + + base := s.newQueueBase(mockShard, tasks.CategoryTimer, nil) + base.checkpointTimer = time.NewTimer(s.options.CheckpointInterval()) + + // set to a smaller value so that delete will be triggered, matching TestCheckPoint_SlicePredicateAction + base.exclusiveDeletionHighWatermark = tasks.MinimumKey + + mockShard.Resource.ExecutionMgr.EXPECT().RangeCompleteHistoryTasks(gomock.Any(), gomock.Any()).Return(nil).Times(1) + mockShard.Resource.ShardMgr.EXPECT().UpdateShard(gomock.Any(), gomock.Any()).Return(nil).Times(1) + + base.checkpoint() + + snapshot := capture.Snapshot() + recordings := snapshot[metrics.QueueSliceCountHistogram.Name()] + s.Require().Len(recordings, 1) + s.Equal(int64(numSlices), recordings[0].Value) + s.Equal(tasks.CategoryTimer.Name(), recordings[0].Tags["task_category"]) +} + +func (s *queueBaseSuite) TestCheckPoint_RecordsSliceCountTotal() { + numSlices := 3 + scopes := NewRandomScopes(numSlices) + queueState := &queueState{ + readerScopes: map[int64][]Scope{ + DefaultReaderId: scopes, + }, + exclusiveReaderHighWatermark: tasks.MaximumKey, + } + persistenceState := ToPersistenceQueueState(queueState) + + mockShard := shard.NewTestContext( + s.controller, + &persistencespb.ShardInfo{ + ShardId: 0, + RangeId: 10, + QueueStates: map[int32]*persistencespb.QueueState{ + int32(tasks.CategoryIDTimer): persistenceState, + }, + }, + s.config, + ) + mockShard.Resource.ClusterMetadata.EXPECT().GetCurrentClusterName().Return(cluster.TestCurrentClusterName).AnyTimes() + mockShard.Resource.ClusterMetadata.EXPECT().GetAllClusterInfo().Return(cluster.TestAllClusterInfo).AnyTimes() + + captureHandler := metricstest.NewCaptureHandler() + capture := captureHandler.StartCapture() + defer captureHandler.StopCapture(capture) + s.metricsHandler = captureHandler + + base := s.newQueueBase(mockShard, tasks.CategoryTimer, nil) + base.checkpointTimer = time.NewTimer(s.options.CheckpointInterval()) + + // set to a smaller value so that delete will be triggered, matching TestCheckPoint_SlicePredicateAction + base.exclusiveDeletionHighWatermark = tasks.MinimumKey + + mockShard.Resource.ExecutionMgr.EXPECT().RangeCompleteHistoryTasks(gomock.Any(), gomock.Any()).Return(nil).Times(1) + mockShard.Resource.ShardMgr.EXPECT().UpdateShard(gomock.Any(), gomock.Any()).Return(nil).Times(1) + + base.checkpoint() + + snapshot := capture.Snapshot() + recordings := snapshot[metrics.QueueSliceCountTotal.Name()] + s.Require().Len(recordings, 1) + s.Equal(int64(numSlices), recordings[0].Value) + s.Equal(tasks.CategoryTimer.Name(), recordings[0].Tags["task_category"]) +} + func (s *queueBaseSuite) TestCheckPoint_SlicePredicateAction() { exclusiveReaderHighWatermark := tasks.MaximumKey scopes := NewRandomScopes(3) diff --git a/service/history/queues/slice.go b/service/history/queues/slice.go index e07b9c238a..bd1bfa83df 100644 --- a/service/history/queues/slice.go +++ b/service/history/queues/slice.go @@ -348,6 +348,7 @@ func (s *SliceImpl) shrinkPredicate() { // TODO: this should be generic enough to shrink any predicate type, probably doesn't belong here. pendingPerKey := s.pendingPerKey + metrics.QueueSlicePendingKeys.With(s.metricsHandler).Record(int64(len(pendingPerKey))) if len(pendingPerKey) > s.maxPendingKeysFn() { // only shrink predicate if there're few keys left metrics.QueuePredicateResolutionLoss.With(s.metricsHandler).Record( diff --git a/service/history/queues/slice_test.go b/service/history/queues/slice_test.go index 15f4f68fa5..3a47a4e5b9 100644 --- a/service/history/queues/slice_test.go +++ b/service/history/queues/slice_test.go @@ -18,6 +18,7 @@ import ( "go.temporal.io/server/common/dynamicconfig" "go.temporal.io/server/common/log" "go.temporal.io/server/common/metrics" + "go.temporal.io/server/common/metrics/metricstest" "go.temporal.io/server/common/namespace" "go.temporal.io/server/common/predicates" ctasks "go.temporal.io/server/common/tasks" @@ -450,6 +451,80 @@ func (s *sliceSuite) TestShrinkScope_ShrinkPredicate() { } } +func (s *sliceSuite) TestShrinkScope_RecordsPendingKeysHistogram() { + testCases := []struct { + name string + numPendingNamespaces int + leaveIteratorsOpen bool + expectDeclined bool + }{ + { + name: "narrowing succeeds", + numPendingNamespaces: 3, + }, + { + name: "narrowing declines", + numPendingNamespaces: 12, + expectDeclined: true, + }, + { + name: "slice still reading its range: no sample recorded", + numPendingNamespaces: 3, + leaveIteratorsOpen: true, + }, + } + + for _, tc := range testCases { + s.Run(tc.name, func() { + r := NewRandomRange() + predicate := predicates.Universal[tasks.Task]() + + handler := metricstest.NewCaptureHandler() + capture := handler.StartCapture() + defer handler.StopCapture(capture) + + slice := NewSlice(nil, s.executableFactory, s.monitor, NewScope(r, predicate), GrouperNamespaceID{}, noPredicateSizeLimit, defaultMaxPendingKeys, handler) + if !tc.leaveIteratorsOpen { + slice.iterators = []Iterator{} // manually set iterators to be empty to trigger predicate update + } + + // One pending executable per namespace, each with its own distinct namespace ID, + // so the pending-key count is exact rather than a coincidence of random assignment + // across a shared, smaller pool of namespace IDs. + executables := s.randomExecutablesInRange(r, tc.numPendingNamespaces) + for _, executable := range executables { + mockExecutable := executable.(*MockExecutable) + mockExecutable.EXPECT().GetTask().Return(mockExecutable).AnyTimes() + mockExecutable.EXPECT().GetNamespaceID().Return(uuid.NewString()).AnyTimes() + mockExecutable.EXPECT().State().Return(ctasks.TaskStatePending).MaxTimes(1) + slice.add(executable) + } + + slice.ShrinkScope() + s.validateSliceState(slice) + + snapshot := capture.Snapshot() + + if tc.leaveIteratorsOpen { + s.Empty(snapshot[metrics.QueueSlicePendingKeys.Name()]) + s.Empty(snapshot[metrics.QueuePredicateResolutionLoss.Name()]) + return + } + + pendingKeysRecordings := snapshot[metrics.QueueSlicePendingKeys.Name()] + s.Require().Len(pendingKeysRecordings, 1) + s.Equal(int64(tc.numPendingNamespaces), pendingKeysRecordings[0].Value) + + lossRecordings := snapshot[metrics.QueuePredicateResolutionLoss.Name()] + if tc.expectDeclined { + s.Len(lossRecordings, 1) + } else { + s.Empty(lossRecordings) + } + }) + } +} + func (s *sliceSuite) TestSelectTasks_NoError() { r := NewRandomRange() namespaceIDs := []string{uuid.NewString(), uuid.NewString(), uuid.NewString(), uuid.NewString()} diff --git a/service/history/shard/context_impl.go b/service/history/shard/context_impl.go index f1369261f8..b55b29c23e 100644 --- a/service/history/shard/context_impl.go +++ b/service/history/shard/context_impl.go @@ -1257,6 +1257,21 @@ func (s *ContextImpl) updateShardInfo( s.tasksCompletedSinceLastUpdate = 0 updatedShardInfo := trimShardInfo(s.config, s.clusterMetadata.GetAllClusterInfo(), s.copyShardInfo(s.shardInfo)) + + metrics.ShardInfoSize.With(s.metricsHandler).Record(int64(updatedShardInfo.Size())) + for categoryID, queueState := range updatedShardInfo.QueueStates { + category, ok := s.taskCategoryRegistry.GetCategoryByID(int(categoryID)) + if !ok { + continue + } + sizeBytes := int64(queueState.Size()) + categoryTag := metrics.TaskCategoryTag(category.Name()) + // The counter is a true accumulator; the histogram's _sum is not, since tally's Prometheus + // reporter replays each sample as its bucket's upper bound, not the recorded value. + metrics.QueueStateSize.With(s.metricsHandler).Record(sizeBytes, categoryTag) + metrics.QueueStateSizeTotal.With(s.metricsHandler).Record(sizeBytes, categoryTag) + } + request := &persistence.UpdateShardRequest{ ShardInfo: updatedShardInfo, PreviousRangeID: s.shardInfo.GetRangeId(), diff --git a/service/history/shard/context_test.go b/service/history/shard/context_test.go index 8b369b9992..abaef132d9 100644 --- a/service/history/shard/context_test.go +++ b/service/history/shard/context_test.go @@ -994,6 +994,102 @@ func (s *contextSuite) TestUpdateShardInfo_FirstUpdate() { s.Equal(0, s.mockShard.tasksCompletedSinceLastUpdate) } +func (s *contextSuite) TestUpdateShardInfo_RecordsSizeMetrics() { + s.mockShard.state = contextStateAcquired + s.setImmediateAckLevels(map[int32]int64{ + int32(tasks.CategoryIDTransfer): 100, + int32(tasks.CategoryIDTimer): 200, + }) + + expectedTransferSize := int64(s.mockShard.shardInfo.QueueStates[int32(tasks.CategoryIDTransfer)].Size()) + expectedTimerSize := int64(s.mockShard.shardInfo.QueueStates[int32(tasks.CategoryIDTimer)].Size()) + + s.mockShardManager.EXPECT().UpdateShard(gomock.Any(), gomock.Any()).Return(nil).Times(1) + + captureHandler := metricstest.NewCaptureHandler() + s.mockShard.SetMetricsHandler(captureHandler) + capture := captureHandler.StartCapture() + defer captureHandler.StopCapture(capture) + + err := s.mockShard.updateShardInfo(0, func() {}) + s.NoError(err) + + snapshot := capture.Snapshot() + + shardInfoSizeRecordings := snapshot[metrics.ShardInfoSize.Name()] + s.Require().Len(shardInfoSizeRecordings, 1) + s.GreaterOrEqual(shardInfoSizeRecordings[0].Value.(int64), expectedTransferSize) + s.GreaterOrEqual(shardInfoSizeRecordings[0].Value.(int64), expectedTimerSize) + + queueStateSizeRecordings := snapshot[metrics.QueueStateSize.Name()] + s.Require().Len(queueStateSizeRecordings, 2) + + sizeByCategory := make(map[string]int64, len(queueStateSizeRecordings)) + for _, recording := range queueStateSizeRecordings { + sizeByCategory[recording.Tags["task_category"]] = recording.Value.(int64) + } + s.Equal(map[string]int64{ + tasks.CategoryTransfer.Name(): expectedTransferSize, + tasks.CategoryTimer.Name(): expectedTimerSize, + }, sizeByCategory) +} + +func (s *contextSuite) TestUpdateShardInfo_RecordsQueueStateSizeTotal() { + s.mockShard.state = contextStateAcquired + s.setImmediateAckLevels(map[int32]int64{ + int32(tasks.CategoryIDTransfer): 100, + int32(tasks.CategoryIDTimer): 200, + }) + + expectedTransferSize := int64(s.mockShard.shardInfo.QueueStates[int32(tasks.CategoryIDTransfer)].Size()) + expectedTimerSize := int64(s.mockShard.shardInfo.QueueStates[int32(tasks.CategoryIDTimer)].Size()) + + s.mockShardManager.EXPECT().UpdateShard(gomock.Any(), gomock.Any()).Return(nil).Times(1) + + captureHandler := metricstest.NewCaptureHandler() + s.mockShard.SetMetricsHandler(captureHandler) + capture := captureHandler.StartCapture() + defer captureHandler.StopCapture(capture) + + err := s.mockShard.updateShardInfo(0, func() {}) + s.NoError(err) + + snapshot := capture.Snapshot() + recordings := snapshot[metrics.QueueStateSizeTotal.Name()] + s.Require().Len(recordings, 2) + + sizeByCategory := make(map[string]int64, len(recordings)) + for _, recording := range recordings { + sizeByCategory[recording.Tags["task_category"]] = recording.Value.(int64) + } + s.Equal(map[string]int64{ + tasks.CategoryTransfer.Name(): expectedTransferSize, + tasks.CategoryTimer.Name(): expectedTimerSize, + }, sizeByCategory) +} + +func (s *contextSuite) TestUpdateShardInfo_DoesNotRecordSizeMetrics_WhenThrottled() { + s.mockShard.state = contextStateAcquired + + // First call always persists, establishing lastUpdated. + s.mockShardManager.EXPECT().UpdateShard(gomock.Any(), gomock.Any()).Return(nil).Times(1) + s.NoError(s.mockShard.updateShardInfo(0, func() {})) + + captureHandler := metricstest.NewCaptureHandler() + s.mockShard.SetMetricsHandler(captureHandler) + capture := captureHandler.StartCapture() + defer captureHandler.StopCapture(capture) + + // No time has passed and too few tasks completed: shouldn't persist, and shouldn't record size. + s.mockShardManager.EXPECT().UpdateShard(gomock.Any(), gomock.Any()).Times(0) + s.NoError(s.mockShard.updateShardInfo(0, func() {})) + + snapshot := capture.Snapshot() + s.Empty(snapshot[metrics.ShardInfoSize.Name()]) + s.Empty(snapshot[metrics.QueueStateSize.Name()]) + s.Empty(snapshot[metrics.QueueStateSizeTotal.Name()]) +} + // setImmediateAckLevels replaces the shard's queue states so each given immediate category has its // ack level at the given task id, i.e. a backlog of everything above it. func (s *contextSuite) setImmediateAckLevels(ackLevelByCategoryID map[int32]int64) {