diff --git a/src/DurableTask.AzureStorage/Monitoring/DisconnectedPerformanceMonitor.cs b/src/DurableTask.AzureStorage/Monitoring/DisconnectedPerformanceMonitor.cs index 9b26db831..f3385ef6b 100644 --- a/src/DurableTask.AzureStorage/Monitoring/DisconnectedPerformanceMonitor.cs +++ b/src/DurableTask.AzureStorage/Monitoring/DisconnectedPerformanceMonitor.cs @@ -152,8 +152,8 @@ internal virtual async Task UpdateQueueMetrics() this.azureStorageClient, defaultPartitionCount: AzureStorageOrchestrationServiceSettings.DefaultPartitionCount); - Task workItemMetricTask = GetQueueMetricsAsync(workItemQueue); - List> controlQueueMetricTasks = controlQueues.Select(GetQueueMetricsAsync).ToList(); + Task workItemMetricTask = GetQueueMetricsAsync(workItemQueue, includeInvisibleMessages: true); + List> controlQueueMetricTasks = controlQueues.Select(queue => GetQueueMetricsAsync(queue, includeInvisibleMessages: false)).ToList(); var tasks = new List(controlQueueMetricTasks.Count + 1); tasks.Add(workItemMetricTask); @@ -201,7 +201,7 @@ internal virtual async Task UpdateQueueMetrics() return true; } - async Task GetQueueMetricsAsync(Queue queue) + async Task GetQueueMetricsAsync(Queue queue, bool includeInvisibleMessages) { Task latencyTask = GetQueueLatencyAsync(queue); Task lengthTask = GetQueueLengthAsync(queue); @@ -212,9 +212,13 @@ async Task GetQueueMetricsAsync(Queue queue) if (latency == TimeSpan.MinValue) { - // No available queue messages (peek returned null) + // No visible messages means no queue latency, but activities may still be running. latency = TimeSpan.Zero; - length = 0; + if (!includeInvisibleMessages) + { + // Control queues can contain timers that are not due yet. + length = 0; + } } return new QueueMetric { Latency = latency, Length = length }; @@ -302,7 +306,7 @@ protected virtual async Task GetAggregateControlQueueLengthAsy /// Returns a scale recommendation public virtual ScaleRecommendation MakeScaleRecommendation(int workerCount) { - return MakeScaleRecommendation(workerCount, this.PartitionCount, this.WorkItemQueueLatencies, this.ControlQueueLatencies); + return MakeScaleRecommendation(workerCount, this.PartitionCount, this.currentWorkItemQueueLength, this.WorkItemQueueLatencies, this.ControlQueueLatencies); } /// @@ -318,7 +322,8 @@ public virtual ScaleRecommendation MakeScaleRecommendation(int workerCount, Perf return new ScaleRecommendation(ScaleAction.None, keepWorkersAlive: true, reason: "No heartbeat metrics"); } - int partitionCount = performanceHeartbeats.Last().PartitionCount; + PerformanceHeartbeat latestHeartbeat = performanceHeartbeats.Last(); + int partitionCount = latestHeartbeat.PartitionCount; QueueMetricHistory workItemQueueLatencyHistory = new QueueMetricHistory(QueueLengthSampleSize); List controlQueueLatencyHistory = new List(); @@ -336,17 +341,21 @@ public virtual ScaleRecommendation MakeScaleRecommendation(int workerCount, Perf } } - return MakeScaleRecommendation(workerCount, partitionCount, workItemQueueLatencyHistory, controlQueueLatencyHistory); + return MakeScaleRecommendation(workerCount, partitionCount, latestHeartbeat.WorkItemQueueLength, workItemQueueLatencyHistory, controlQueueLatencyHistory); } internal ScaleRecommendation MakeScaleRecommendation( int workerCount, int partitionCount, + int workItemQueueLength, QueueMetricHistory workItemQueueLatencyHistory, List controlQueueLatencyHistory) { - // REVIEW: Is zero latency a reliable indicator of idle? - bool taskHubIsIdle = IsIdle(workItemQueueLatencyHistory) && controlQueueLatencyHistory.TrueForAll(IsIdle); + // Invisible activity messages still need worker capacity. + double requiredActivityWorkers = Math.Ceiling(workItemQueueLength / (double)this.settings.MaxConcurrentTaskActivityWorkItems); + bool hasSpareActivityWorkers = workerCount > requiredActivityWorkers; + bool workItemsAreIdle = workItemQueueLength == 0 && IsIdle(workItemQueueLatencyHistory); + bool taskHubIsIdle = workItemsAreIdle && controlQueueLatencyHistory.TrueForAll(IsIdle); if (workerCount == 0 && !taskHubIsIdle) { return new ScaleRecommendation(ScaleAction.AddWorker, keepWorkersAlive: true, reason: "First worker"); @@ -365,6 +374,13 @@ internal ScaleRecommendation MakeScaleRecommendation( keepWorkersAlive: false, reason: "Task hub is idle"); } + else if (workerCount < requiredActivityWorkers) + { + return new ScaleRecommendation( + ScaleAction.AddWorker, + keepWorkersAlive: true, + reason: $"Activity workers needed: {requiredActivityWorkers} > {workerCount}"); + } else if (this.IsHighLatency(workItemQueueLatencyHistory)) { return new ScaleRecommendation( @@ -372,12 +388,12 @@ internal ScaleRecommendation MakeScaleRecommendation( keepWorkersAlive: true, reason: $"Work-item queue latency: {workItemQueueLatencyHistory.Latest} > {this.highLatencyThreshold}"); } - else if (workerCount > partitionCount && IsIdle(workItemQueueLatencyHistory)) + else if (workerCount > partitionCount && hasSpareActivityWorkers && IsIdle(workItemQueueLatencyHistory)) { return new ScaleRecommendation( ScaleAction.RemoveWorker, keepWorkersAlive: true, - reason: $"Work-items idle, #workers > partitions ({workerCount} > {partitionCount})"); + reason: $"Activity capacity available, #workers > partitions ({workerCount} > {partitionCount})"); } // Control queues are partitioned; only scale-out if there are more partitions than workers. @@ -390,15 +406,15 @@ internal ScaleRecommendation MakeScaleRecommendation( keepWorkersAlive: true, reason: $"High control queue latency: {metric.Latest} > {this.highLatencyThreshold}"); } - else if (workerCount > controlQueueLatencyHistory.Count(h => !IsIdle(h)) && IsIdle(workItemQueueLatencyHistory)) + else if (workerCount > controlQueueLatencyHistory.Count(h => !IsIdle(h)) && hasSpareActivityWorkers && IsIdle(workItemQueueLatencyHistory)) { - // If the work item queues are idle, scale down to the number of non-idle control queues. + // Keep enough workers for activities and non-idle control queues. return new ScaleRecommendation( ScaleAction.RemoveWorker, - keepWorkersAlive: controlQueueLatencyHistory.Any(IsIdle), + keepWorkersAlive: workItemQueueLength > 0 || controlQueueLatencyHistory.Any(IsIdle), reason: $"One or more control queues idle"); } - else if (workerCount > 1) + else if (workerCount > 1 && hasSpareActivityWorkers) { // If all queues are operating efficiently, it can be hard to know if we need to reduce the worker count. // We want to avoid the case where a constant trickle of load after a big scale-out prevents scaling back in. diff --git a/test/DurableTask.AzureStorage.Tests/AzureStorageScaleTests.cs b/test/DurableTask.AzureStorage.Tests/AzureStorageScaleTests.cs index 7d7ac8a78..f42ac0683 100644 --- a/test/DurableTask.AzureStorage.Tests/AzureStorageScaleTests.cs +++ b/test/DurableTask.AzureStorage.Tests/AzureStorageScaleTests.cs @@ -700,6 +700,101 @@ public async Task MonitorIdleTaskHubDisconnected() Assert.IsNotNull(recommendation.Reason); } + [TestMethod] + public async Task MonitorInFlightWorkItemsDisconnected() + { + string connectionString = TestHelpers.GetTestStorageAccountConnectionString(); + var settings = new AzureStorageOrchestrationServiceSettings + { + PartitionCount = 1, + MaxConcurrentTaskActivityWorkItems = 6, + StorageAccountClientProvider = new StorageAccountClientProvider(connectionString), + TaskHubName = nameof(MonitorInFlightWorkItemsDisconnected), + UseAppLease = false, + }; + var service = new AzureStorageOrchestrationService(settings); + var monitor = new DisconnectedPerformanceMonitor(settings); + await service.CreateAsync(recreateInstanceStore: true); + + try + { + // 1. Queue activities and a timer that is not due yet. + const int activityCount = 18; + Queue workItems = service.WorkItemQueue.InnerQueue; + Queue control = service.AllControlQueues.Single().InnerQueue; + for (int i = 0; i < activityCount; i++) + { + await workItems.AddMessageAsync("activity", visibilityDelay: null); + } + await control.AddMessageAsync("timer", visibilityDelay: TimeSpan.FromHours(1)); + + PerformanceHeartbeat heartbeat = await monitor.PulseAsync(); + Assert.AreEqual(activityCount, heartbeat.WorkItemQueueLength); + CollectionAssert.AreEqual(new[] { 0 }, heartbeat.ControlQueueLengths.ToArray()); + + // 2. Running activities keep their messages invisible. + IReadOnlyCollection running = await workItems.GetMessagesAsync(activityCount, TimeSpan.FromMinutes(5)); + Assert.AreEqual(activityCount, running.Count); + Assert.IsNull(await workItems.PeekMessageAsync()); + var heartbeats = new List(); + for (int i = 0; i < DisconnectedPerformanceMonitor.QueueLengthSampleSize; i++) + { + heartbeat = await monitor.PulseAsync(currentWorkerCount: 3); + heartbeats.Add(heartbeat); + Assert.AreEqual(activityCount, heartbeat.WorkItemQueueLength); + Assert.AreEqual(TimeSpan.Zero, heartbeat.WorkItemQueueLatency); + CollectionAssert.AreEqual(new[] { 0 }, heartbeat.ControlQueueLengths.ToArray()); + Assert.AreEqual(ScaleAction.None, heartbeat.ScaleRecommendation.Action); + Assert.IsTrue(heartbeat.ScaleRecommendation.KeepWorkersAlive); + } + foreach (int workerCount in new[] { 0, 1, 2, 3, 4 }) + { + ScaleAction expected = workerCount < 3 ? ScaleAction.AddWorker : workerCount == 3 ? ScaleAction.None : ScaleAction.RemoveWorker; + foreach (ScaleRecommendation recommendation in new[] + { + monitor.MakeScaleRecommendation(workerCount), + monitor.MakeScaleRecommendation(workerCount, heartbeats.ToArray()), + }) + { + Assert.AreEqual(expected, recommendation.Action); + Assert.IsTrue(recommendation.KeepWorkersAlive); + } + } + + // 3. New work adds to the running work, so scale-out is still possible. + await workItems.AddMessageAsync("next activity", visibilityDelay: null); + heartbeat = await monitor.PulseAsync(currentWorkerCount: 3); + Assert.AreEqual(activityCount + 1, heartbeat.WorkItemQueueLength); + Assert.AreEqual(ScaleAction.AddWorker, heartbeat.ScaleRecommendation.Action); + QueueMessage next = await workItems.GetMessageAsync(TimeSpan.FromMinutes(5)); + Assert.IsNotNull(next); + + // 4. Completed activities no longer keep workers alive. + foreach (QueueMessage message in running.Append(next)) + { + await workItems.DeleteMessageAsync(message); + } + heartbeats.Clear(); + for (int i = 0; i < DisconnectedPerformanceMonitor.QueueLengthSampleSize; i++) + { + heartbeat = await monitor.PulseAsync(currentWorkerCount: 1); + heartbeats.Add(heartbeat); + Assert.AreEqual(0, heartbeat.WorkItemQueueLength); + Assert.AreEqual(TimeSpan.Zero, heartbeat.WorkItemQueueLatency); + CollectionAssert.AreEqual(new[] { 0 }, heartbeat.ControlQueueLengths.ToArray()); + } + Assert.AreEqual(ScaleAction.RemoveWorker, heartbeat.ScaleRecommendation.Action); + Assert.IsFalse(heartbeat.ScaleRecommendation.KeepWorkersAlive); + ScaleRecommendation completed = monitor.MakeScaleRecommendation(1, heartbeats.ToArray()); + Assert.AreEqual(ScaleAction.RemoveWorker, completed.Action); + Assert.IsFalse(completed.KeepWorkersAlive); + } + finally + { + await service.DeleteAsync(); + } + } + [TestMethod] public async Task UpdateTaskHubJsonWithNewPartitionCount() { @@ -1044,6 +1139,90 @@ public async Task ScaleDecision_WorkItemLatency_NotMaxPollingDelay() Assert.IsTrue(recommendation.KeepWorkersAlive); } + [DataTestMethod] + [DataRow(0, 1, 0, 0, ScaleAction.AddWorker)] + [DataRow(1, 1, 0, 0, ScaleAction.None)] + [DataRow(5, 1, 0, 0, ScaleAction.RemoveWorker)] + [DataRow(3, 1, 0, 100, ScaleAction.RemoveWorker)] + [DataRow(2, 4, 100, 100, ScaleAction.None)] + [DataRow(1, 1, 1000, 0, ScaleAction.AddWorker)] + [DataRow(1, 1, 0, 1000, ScaleAction.AddWorker)] + [DataRow(1, 4, 0, 0, ScaleAction.AddWorker)] + [DataRow(2, 6, 0, 0, ScaleAction.None)] + [DataRow(2, 7, 0, 0, ScaleAction.AddWorker)] + [DataRow(3, 7, 0, 0, ScaleAction.None)] + [DataRow(4, 7, 0, 0, ScaleAction.RemoveWorker)] + [DataRow(2, 1, 0, 500, ScaleAction.None)] + [DataRow(5, 13, 0, 0, ScaleAction.None)] + public void ScaleDecision_AdHoc_RetainedWorkItems( + int workerCount, int activityCount, int activityLatency, int controlLatency, ScaleAction expected) + { + // 1. Allow three activities per worker and fill the sampling window. + var monitor = GetFakePerformanceMonitor(maxConcurrentActivities: 3); + monitor.EnableRandomScaleDownOnLowLatency = true; + PerformanceHeartbeat[] heartbeats = Enumerable.Range(0, DisconnectedPerformanceMonitor.QueueLengthSampleSize) + .Select(_ => new PerformanceHeartbeat + { + PartitionCount = 4, + WorkItemQueueLength = activityCount, + WorkItemQueueLatency = TimeSpan.FromMilliseconds(activityLatency), + ControlQueueLatencies = new List + { + TimeSpan.FromMilliseconds(controlLatency), TimeSpan.FromMilliseconds(controlLatency), + TimeSpan.Zero, TimeSpan.Zero, + }, + }).ToArray(); + + // 2. Keep enough activity capacity and respect control queue demand. + ScaleRecommendation recommendation = monitor.MakeScaleRecommendation(workerCount, heartbeats); + Assert.AreEqual(expected, recommendation.Action); + Assert.IsTrue(recommendation.KeepWorkersAlive); + } + + [TestMethod] + public void ScaleDecision_AdHoc_RetainedWorkItemsWithoutControlQueues() + { + // 1. Keep one activity while no control queues are reported. + var monitor = GetFakePerformanceMonitor(maxConcurrentActivities: 3); + PerformanceHeartbeat[] heartbeats = Enumerable.Range(0, DisconnectedPerformanceMonitor.QueueLengthSampleSize) + .Select(_ => new PerformanceHeartbeat + { + PartitionCount = 4, + WorkItemQueueLength = 1, + ControlQueueLatencies = new List(), + }).ToArray(); + + // 2. Remove the spare worker, but keep the remaining worker alive. + ScaleRecommendation recommendation = monitor.MakeScaleRecommendation(2, heartbeats); + Assert.AreEqual(ScaleAction.RemoveWorker, recommendation.Action); + Assert.IsTrue(recommendation.KeepWorkersAlive); + } + + [DataTestMethod] + [DataRow(18, 0, 1, ScaleAction.RemoveWorker, false)] + [DataRow(18, 4, 2, ScaleAction.None, true)] + [DataRow(18, 4, 3, ScaleAction.RemoveWorker, true)] + [DataRow(1, 7, 2, ScaleAction.AddWorker, true)] + public void ScaleDecision_AdHoc_LatestWorkItemCount( + int previousCount, int latestCount, int workerCount, ScaleAction expected, bool keepWorkersAlive) + { + // 1. Change the activity count in the newest sample. + var monitor = GetFakePerformanceMonitor(maxConcurrentActivities: 3); + PerformanceHeartbeat[] heartbeats = Enumerable.Range(0, DisconnectedPerformanceMonitor.QueueLengthSampleSize) + .Select(_ => new PerformanceHeartbeat + { + PartitionCount = 4, + WorkItemQueueLength = previousCount, + ControlQueueLatencies = new List { TimeSpan.Zero, TimeSpan.Zero, TimeSpan.Zero, TimeSpan.Zero }, + }).ToArray(); + heartbeats.Last().WorkItemQueueLength = latestCount; + + // 2. Only the newest count determines activity capacity. + ScaleRecommendation recommendation = monitor.MakeScaleRecommendation(workerCount, heartbeats); + Assert.AreEqual(expected, recommendation.Action); + Assert.AreEqual(keepWorkersAlive, recommendation.KeepWorkersAlive); + } + [TestMethod] public void ScaleDecision_AdHoc_WorkItemLatency_High() { @@ -1997,9 +2176,9 @@ public void ScaleDecision_AdHoc_ControlQueueLatency_NotMaxPollingDelay() } #endregion - static FakePerformanceMonitor GetFakePerformanceMonitor() + static FakePerformanceMonitor GetFakePerformanceMonitor(int maxConcurrentActivities = 10) { - return new FakePerformanceMonitor(TestHelpers.GetTestStorageAccountConnectionString(), "taskHub"); + return new FakePerformanceMonitor(TestHelpers.GetTestStorageAccountConnectionString(), "taskHub", maxConcurrentActivities: maxConcurrentActivities); } class NoOpOrchestration : TaskOrchestration @@ -2015,8 +2194,14 @@ class FakePerformanceMonitor : DisconnectedPerformanceMonitor public FakePerformanceMonitor( string storageConnectionString, string taskHub, - int partitionCount = AzureStorageOrchestrationServiceSettings.DefaultPartitionCount) - : base(storageConnectionString, taskHub) + int partitionCount = AzureStorageOrchestrationServiceSettings.DefaultPartitionCount, + int maxConcurrentActivities = 10) + : base(new AzureStorageOrchestrationServiceSettings + { + StorageAccountClientProvider = new StorageAccountClientProvider(storageConnectionString), + TaskHubName = taskHub, + MaxConcurrentTaskActivityWorkItems = maxConcurrentActivities, + }) { this.PartitionCount = partitionCount; for (int i = 0; i < partitionCount; i++)