Skip to content

Keep active work-item messages in scaling metrics #1428

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
sophiatev merged 3 commits into from
Oct 29, 2026
Merged

Keep active work-item messages in scaling metrics #1428

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
10e00d3
Keep active work-item messages in scaling metrics
Oct 29, 2026
200d500
Keep active work alive in scale recommendations
Oct 29, 2026
a63a4c8
updating the legacy scaling logic
Oct 29, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 32 additions & 16 deletions src/DurableTask.AzureStorage/Monitoring/DisconnectedPerformanceMonitor.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -152,8 +152,8 @@ internal virtual async Task<bool> UpdateQueueMetrics()
this.azureStorageClient,
defaultPartitionCount: AzureStorageOrchestrationServiceSettings.DefaultPartitionCount);

Task<QueueMetric> workItemMetricTask = GetQueueMetricsAsync(workItemQueue);
List<Task<QueueMetric>> controlQueueMetricTasks = controlQueues.Select(GetQueueMetricsAsync).ToList();
Task<QueueMetric> workItemMetricTask = GetQueueMetricsAsync(workItemQueue, includeInvisibleMessages: true);
List<Task<QueueMetric>> controlQueueMetricTasks = controlQueues.Select(queue => GetQueueMetricsAsync(queue, includeInvisibleMessages: false)).ToList();

var tasks = new List<Task>(controlQueueMetricTasks.Count + 1);
tasks.Add(workItemMetricTask);
Expand Down Expand Up @@ -201,7 +201,7 @@ internal virtual async Task<bool> UpdateQueueMetrics()
return true;
}

async Task<QueueMetric> GetQueueMetricsAsync(Queue queue)
async Task<QueueMetric> GetQueueMetricsAsync(Queue queue, bool includeInvisibleMessages)
{
Task<TimeSpan> latencyTask = GetQueueLatencyAsync(queue);
Task<int> lengthTask = GetQueueLengthAsync(queue);
Expand All @@ -212,9 +212,13 @@ async Task<QueueMetric> 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;
Comment thread
sophiatev marked this conversation as resolved.
Resolved
}
}

return new QueueMetric { Latency = latency, Length = length };
Expand Down Expand Up @@ -302,7 +306,7 @@ protected virtual async Task<ControlQueueData> GetAggregateControlQueueLengthAsy
/// <returns>Returns a scale recommendation</returns>
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);
}

/// <summary>
Expand All @@ -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<QueueMetricHistory> controlQueueLatencyHistory = new List<QueueMetricHistory>();

Expand All @@ -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<QueueMetricHistory> 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;
Comment thread
sophiatev marked this conversation as resolved.
Resolved
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");
Expand All @@ -365,19 +374,26 @@ 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(
ScaleAction.AddWorker,
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.
Expand All @@ -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.
Expand Down
193 changes: 189 additions & 4 deletions test/DurableTask.AzureStorage.Tests/AzureStorageScaleTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<QueueMessage> running = await workItems.GetMessagesAsync(activityCount, TimeSpan.FromMinutes(5));
Assert.AreEqual(activityCount, running.Count);
Assert.IsNull(await workItems.PeekMessageAsync());
var heartbeats = new List<PerformanceHeartbeat>();
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()
{
Expand Down Expand Up @@ -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>
{
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<TimeSpan>(),
}).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> { 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()
{
Expand Down Expand Up @@ -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<string, string>
Expand All @@ -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++)
Expand Down
Loading