diff --git a/.claude/commands b/.claude/commands index e8b8c08c..a10cfa5a 160000 --- a/.claude/commands +++ b/.claude/commands @@ -1 +1 @@ -Subproject commit e8b8c08c1b8b8aad79a0ca3045ca792fef51ac58 +Subproject commit a10cfa5a12bdf83392c631b0e54b67b4290a1cca diff --git a/internal b/internal index 5308b812..7cb44729 160000 --- a/internal +++ b/internal @@ -1 +1 @@ -Subproject commit 5308b812a7e0d1c10f61b2ecf10205103e186c5d +Subproject commit 7cb447298ebedcc3fc77290715656e5108d3fd45 diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/AgentEfficiencyMetricsServiceTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/AgentEfficiencyMetricsServiceTests.cs index 754a7c27..087d8265 100644 --- a/src/Backend/AutopilotMonitor.Functions.Tests/AgentEfficiencyMetricsServiceTests.cs +++ b/src/Backend/AutopilotMonitor.Functions.Tests/AgentEfficiencyMetricsServiceTests.cs @@ -4,6 +4,7 @@ using AutopilotMonitor.Shared.Models; using AutopilotMonitor.Shared.Pagination; using Microsoft.Extensions.Logging.Abstractions; using Moq; +using System.Threading; namespace AutopilotMonitor.Functions.Tests; @@ -21,7 +22,7 @@ public class AgentEfficiencyMetricsServiceTests { var repo = new Mock(); repo - .Setup(r => r.GetAllSessionsPageAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny?>(), It.IsAny?>())) + .Setup(r => r.GetAllSessionsPageAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny?>(), It.IsAny?>(), It.IsAny())) .ReturnsAsync(new RawPage(new List(), null)); repo .Setup(r => r.GetSessionEventsByTypesAsync(It.IsAny(), It.IsAny(), It.IsAny>(), It.IsAny?>(), It.IsAny())) @@ -67,7 +68,7 @@ public class AgentEfficiencyMetricsServiceTests private static void SetupSessions(Mock repo, List sessions) { repo - .Setup(r => r.GetAllSessionsPageAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny?>(), It.IsAny?>())) + .Setup(r => r.GetAllSessionsPageAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny?>(), It.IsAny?>(), It.IsAny())) .ReturnsAsync(new RawPage(sessions, null)); } @@ -201,8 +202,8 @@ public class AgentEfficiencyMetricsServiceTests string? capturedTenant = null; IEnumerable? capturedSelect = null; repo - .Setup(r => r.GetAllSessionsPageAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny?>(), It.IsAny?>())) - .Callback?, IEnumerable?>((t, _, _, _, _, sel) => { capturedTenant = t; capturedSelect = sel; }) + .Setup(r => r.GetAllSessionsPageAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny?>(), It.IsAny?>(), It.IsAny())) + .Callback?, IEnumerable?, CancellationToken>((t, _, _, _, _, sel, _) => { capturedTenant = t; capturedSelect = sel; }) .ReturnsAsync(new RawPage(new List(), null)); var result = await service.ComputeAsync(days: 45, limit: 100, tenantId: Tenant); @@ -220,7 +221,7 @@ public class AgentEfficiencyMetricsServiceTests var (service, repo) = CreateService(); int callCount = 0; repo - .Setup(r => r.GetAllSessionsPageAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny?>(), It.IsAny?>())) + .Setup(r => r.GetAllSessionsPageAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny?>(), It.IsAny?>(), It.IsAny())) .Callback(() => callCount++) .ReturnsAsync(new RawPage(new List(), null)); diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/IngestStorageFailureClassificationTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/IngestStorageFailureClassificationTests.cs index aefdaf43..a19fabfb 100644 --- a/src/Backend/AutopilotMonitor.Functions.Tests/IngestStorageFailureClassificationTests.cs +++ b/src/Backend/AutopilotMonitor.Functions.Tests/IngestStorageFailureClassificationTests.cs @@ -51,4 +51,14 @@ public class IngestStorageFailureClassificationTests Assert.Equal(HttpStatusCode.InternalServerError, status); Assert.Null(retryAfter); } + + [Fact] + public void A_connection_failure_without_status_is_503_with_retry_after() + { + var (status, _, retryAfter) = IngestTelemetryFunction.ClassifyStorageFailure( + new RequestFailedException(0, "connection reset before a status arrived")); + + Assert.Equal(HttpStatusCode.ServiceUnavailable, status); + Assert.Equal(IngestTelemetryFunction.StorageRetryAfterSeconds, retryAfter); + } } diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/PlatformMetricsServiceWindowTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/PlatformMetricsServiceWindowTests.cs index 4c811ed6..58fbc8bf 100644 --- a/src/Backend/AutopilotMonitor.Functions.Tests/PlatformMetricsServiceWindowTests.cs +++ b/src/Backend/AutopilotMonitor.Functions.Tests/PlatformMetricsServiceWindowTests.cs @@ -4,6 +4,7 @@ using AutopilotMonitor.Shared.Models; using AutopilotMonitor.Shared.Pagination; using Microsoft.Extensions.Logging.Abstractions; using Moq; +using System.Threading; namespace AutopilotMonitor.Functions.Tests; @@ -23,7 +24,7 @@ public class PlatformMetricsServiceWindowTests { var sessionRepo = new Mock(); sessionRepo - .Setup(r => r.GetAllSessionsPageAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny?>(), It.IsAny?>())) + .Setup(r => r.GetAllSessionsPageAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny?>(), It.IsAny?>(), It.IsAny())) .ReturnsAsync(EmptyPage()); sessionRepo .Setup(r => r.GetSessionEventsAsync(It.IsAny(), It.IsAny(), It.IsAny())) @@ -64,8 +65,8 @@ public class PlatformMetricsServiceWindowTests int? capturedDays = null; repo - .Setup(r => r.GetAllSessionsPageAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny?>(), It.IsAny?>())) - .Callback?, IEnumerable?>((_, d, _, _, _, _) => capturedDays = d) + .Setup(r => r.GetAllSessionsPageAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny?>(), It.IsAny?>(), It.IsAny())) + .Callback?, IEnumerable?, CancellationToken>((_, d, _, _, _, _, _) => capturedDays = d) .ReturnsAsync(EmptyPage()); var result = await service.ComputePlatformMetricsAsync(days, limit: 50); @@ -89,8 +90,8 @@ public class PlatformMetricsServiceWindowTests int? capturedPageSize = null; repo - .Setup(r => r.GetAllSessionsPageAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny?>(), It.IsAny?>())) - .Callback?, IEnumerable?>((_, _, ps, _, _, _) => capturedPageSize = ps) + .Setup(r => r.GetAllSessionsPageAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny?>(), It.IsAny?>(), It.IsAny())) + .Callback?, IEnumerable?, CancellationToken>((_, _, ps, _, _, _, _) => capturedPageSize = ps) .ReturnsAsync(EmptyPage()); var result = await service.ComputePlatformMetricsAsync(days: days, limit: limit); @@ -106,8 +107,8 @@ public class PlatformMetricsServiceWindowTests int? capturedDays = null; repo - .Setup(r => r.GetAllSessionsPageAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny?>(), It.IsAny?>())) - .Callback?, IEnumerable?>((_, d, _, _, _, _) => capturedDays = d) + .Setup(r => r.GetAllSessionsPageAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny?>(), It.IsAny?>(), It.IsAny())) + .Callback?, IEnumerable?, CancellationToken>((_, d, _, _, _, _, _) => capturedDays = d) .ReturnsAsync(EmptyPage()); var result = await service.ComputePlatformMetricsAsync(0); @@ -140,7 +141,7 @@ public class PlatformMetricsServiceWindowTests var (service, repo) = CreateService(); int callCount = 0; repo - .Setup(r => r.GetAllSessionsPageAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny?>(), It.IsAny?>())) + .Setup(r => r.GetAllSessionsPageAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny?>(), It.IsAny?>(), It.IsAny())) .Callback(() => callCount++) .ReturnsAsync(EmptyPage()); @@ -160,7 +161,7 @@ public class PlatformMetricsServiceWindowTests var (service, repo) = CreateService(); var sessions = Enumerable.Range(1, 7).Select(MakeSession).ToList(); repo - .Setup(r => r.GetAllSessionsPageAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny?>(), It.IsAny?>())) + .Setup(r => r.GetAllSessionsPageAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny?>(), It.IsAny?>(), It.IsAny())) .ReturnsAsync(new RawPage(sessions, null)); var result = await service.ComputePlatformMetricsAsync(days: 77, limit: 50); @@ -177,7 +178,7 @@ public class PlatformMetricsServiceWindowTests var (service, repo) = CreateService(); var sessions = Enumerable.Range(1, 25).Select(MakeSession).ToList(); repo - .Setup(r => r.GetAllSessionsPageAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny?>(), It.IsAny?>())) + .Setup(r => r.GetAllSessionsPageAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny?>(), It.IsAny?>(), It.IsAny())) .ReturnsAsync(new RawPage(sessions, null)); var fullFetchSessions = new List(); @@ -216,7 +217,7 @@ public class PlatformMetricsServiceWindowTests var (service, repo) = CreateService(); var sessions = Enumerable.Range(1, 21).Select(MakeSession).ToList(); repo - .Setup(r => r.GetAllSessionsPageAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny?>(), It.IsAny?>())) + .Setup(r => r.GetAllSessionsPageAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny?>(), It.IsAny?>(), It.IsAny())) .ReturnsAsync(new RawPage(sessions, null)); var session21 = sessions[20].SessionId; diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/ProcessStallClassifierTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/ProcessStallClassifierTests.cs new file mode 100644 index 00000000..6b93e54d --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions.Tests/ProcessStallClassifierTests.cs @@ -0,0 +1,84 @@ +using System; +using AutopilotMonitor.Functions.Telemetry; +using Xunit; + +namespace AutopilotMonitor.Functions.Tests; + +/// +/// The stall classifier turns two runtime snapshots around a late timer tick into a cause. +/// GC wins when its pause clock covers at least half the drift; queued work or thread injection +/// means starvation; otherwise the drift stays unexplained rather than guessed. +/// +public class ProcessStallClassifierTests +{ + private static ProcessSnapshot Snapshot( + double gcPauseMs = 0, int gen2 = 0, int threads = 8, long pending = 0, long completed = 0) + => new( + GcPause: TimeSpan.FromMilliseconds(gcPauseMs), + Gen0: 0, Gen1: 0, Gen2: gen2, + ThreadPoolThreads: threads, + PendingWorkItems: pending, + CompletedWorkItems: completed, + WorkingSetBytes: 0, HeapSizeBytes: 0, CommittedBytes: 0, + LastGcGeneration: 0, LastGcPause: TimeSpan.Zero, LastGcCompacted: false); + + [Fact] + public void Gc_pause_covering_half_the_drift_is_a_gc_stall() + { + var before = Snapshot(gcPauseMs: 1000, gen2: 3); + var after = Snapshot(gcPauseMs: 6200, gen2: 4, pending: 12); + + var report = ProcessStallClassifier.Classify(TimeSpan.FromSeconds(10), before, after); + + Assert.Equal(StallCause.Gc, report.Cause); + Assert.Equal(TimeSpan.FromMilliseconds(5200), report.GcPauseDelta); + Assert.Equal(1, report.Gen2Delta); + } + + [Fact] + public void Queued_work_without_gc_is_thread_pool_starvation() + { + var before = Snapshot(gcPauseMs: 1000); + var after = Snapshot(gcPauseMs: 1100, pending: 40); + + var report = ProcessStallClassifier.Classify(TimeSpan.FromSeconds(8), before, after); + + Assert.Equal(StallCause.ThreadPool, report.Cause); + } + + [Fact] + public void Thread_injection_without_gc_is_thread_pool_starvation() + { + var before = Snapshot(threads: 8); + var after = Snapshot(threads: 11); + + var report = ProcessStallClassifier.Classify(TimeSpan.FromSeconds(3), before, after); + + Assert.Equal(StallCause.ThreadPool, report.Cause); + Assert.Equal(3, report.ThreadPoolThreadsDelta); + } + + [Fact] + public void Drift_neither_clock_explains_stays_unknown() + { + var before = Snapshot(gcPauseMs: 500, threads: 8, completed: 100); + var after = Snapshot(gcPauseMs: 900, threads: 8, completed: 130); + + var report = ProcessStallClassifier.Classify(TimeSpan.FromSeconds(5), before, after); + + Assert.Equal(StallCause.Unknown, report.Cause); + Assert.Equal(30, report.CompletedWorkItemsDelta); + } + + [Fact] + public void A_wrapped_pause_clock_never_yields_a_negative_delta() + { + var before = Snapshot(gcPauseMs: 900); + var after = Snapshot(gcPauseMs: 100); + + var report = ProcessStallClassifier.Classify(TimeSpan.FromSeconds(2), before, after); + + Assert.Equal(TimeSpan.Zero, report.GcPauseDelta); + Assert.Equal(StallCause.Unknown, report.Cause); + } +} diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/SingleFlightCacheTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/SingleFlightCacheTests.cs index bd346250..b049dea6 100644 --- a/src/Backend/AutopilotMonitor.Functions.Tests/SingleFlightCacheTests.cs +++ b/src/Backend/AutopilotMonitor.Functions.Tests/SingleFlightCacheTests.cs @@ -99,4 +99,47 @@ public class SingleFlightCacheTests // w1 is gone: its factory runs again. Assert.Equal("one-again", await cache.GetOrAddAsync("w1", Ttl, () => Task.FromResult("one-again"))); } + + [Fact] + public async Task Factory_exceeding_the_budget_faults_every_waiter_and_is_evicted() + { + var cache = new SingleFlightCache(() => DateTime.UtcNow, factoryBudget: TimeSpan.FromMilliseconds(50)); + var runs = 0; + + async Task Hanging(CancellationToken ct) + { + Interlocked.Increment(ref runs); + await Task.Delay(Timeout.Infinite, ct); + return 1; + } + + var first = cache.GetOrAddAsync("k", Ttl, Hanging); + var second = cache.GetOrAddAsync("k", Ttl, Hanging); + + await Assert.ThrowsAnyAsync(() => first); + await Assert.ThrowsAnyAsync(() => second); + Assert.Equal(1, runs); + Assert.Equal(0, cache.Count); + + // The next caller starts a fresh run instead of inheriting the faulted one. + Assert.Equal(7, await cache.GetOrAddAsync("k", Ttl, _ => Task.FromResult(7))); + } + + [Fact] + public async Task Factory_finishing_inside_the_budget_is_cached() + { + var cache = new SingleFlightCache(() => DateTime.UtcNow, factoryBudget: TimeSpan.FromSeconds(5)); + var runs = 0; + + Task Quick(CancellationToken ct) + { + Assert.True(ct.CanBeCanceled); + Interlocked.Increment(ref runs); + return Task.FromResult(3); + } + + Assert.Equal(3, await cache.GetOrAddAsync("k", Ttl, Quick)); + Assert.Equal(3, await cache.GetOrAddAsync("k", Ttl, Quick)); + Assert.Equal(1, runs); + } } diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/StorageClientOptionsTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/StorageClientOptionsTests.cs new file mode 100644 index 00000000..a7cd57c7 --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions.Tests/StorageClientOptionsTests.cs @@ -0,0 +1,70 @@ +using System; +using Azure.Core; +using AutopilotMonitor.Functions.Helpers; +using Xunit; + +namespace AutopilotMonitor.Functions.Tests; + +/// +/// The storage SDK budgets are sized so one hot-path operation — every attempt into its +/// timeout, every backoff at its cap — finishes before the agent's 30 s upload timeout. A +/// change to either side that breaks that relation turns a storage brownout into zombie +/// requests plus agent replays (audit 2026-09-05 F09). +/// +public class StorageClientOptionsTests +{ + [Fact] + public void Table_worst_case_stays_under_the_agent_upload_timeout() + { + var worst = StorageClientOptions.WorstCase(StorageClientOptions.Table().Retry); + Assert.True(worst < StorageClientOptions.AgentUploadTimeout, $"worst case {worst} must stay under {StorageClientOptions.AgentUploadTimeout}"); + } + + [Fact] + public void Queue_worst_case_stays_under_the_agent_upload_timeout() + { + var worst = StorageClientOptions.WorstCase(StorageClientOptions.Queue().Retry); + Assert.True(worst < StorageClientOptions.AgentUploadTimeout, $"worst case {worst} must stay under {StorageClientOptions.AgentUploadTimeout}"); + } + + [Fact] + public void Blob_worst_case_is_bounded() + { + var worst = StorageClientOptions.WorstCase(StorageClientOptions.Blob().Retry); + Assert.True(worst < TimeSpan.FromMinutes(5), $"worst case {worst} must stay under five minutes"); + } + + [Fact] + public void Worst_case_counts_every_attempt_and_every_capped_backoff() + { + // 3 attempts × 8 s + backoffs 0.5 s·1.2 and 1 s·1.2 (both under the 4 s cap) = 25.8 s. + var retry = StorageClientOptions.Table().Retry; + Assert.Equal(RetryMode.Exponential, retry.Mode); + Assert.Equal(StorageClientOptions.HotPathMaxRetries, retry.MaxRetries); + Assert.Equal(StorageClientOptions.HotPathNetworkTimeout, retry.NetworkTimeout); + + var expected = TimeSpan.FromTicks(StorageClientOptions.HotPathNetworkTimeout.Ticks * 3) + + TimeSpan.FromTicks((long)(StorageClientOptions.HotPathRetryDelay.Ticks * 1.2)) + + TimeSpan.FromTicks((long)(StorageClientOptions.HotPathRetryDelay.Ticks * 2 * 1.2)); + + Assert.Equal(expected, StorageClientOptions.WorstCase(retry)); + } + + [Fact] + public void Blob_worst_case_counts_four_attempts_and_three_backoffs() + { + var blob = StorageClientOptions.Blob().Retry; + // 4 attempts × 60 s + backoffs 0.6 s, 1.2 s, 2.4 s (all under the 4 s cap). + var expected = TimeSpan.FromTicks(StorageClientOptions.BlobNetworkTimeout.Ticks * 4) + + TimeSpan.FromMilliseconds(600) + TimeSpan.FromMilliseconds(1200) + TimeSpan.FromMilliseconds(2400); + Assert.Equal(expected, StorageClientOptions.WorstCase(blob)); + } + + [Fact] + public void Queue_options_carry_the_requested_encoding() + { + Assert.Equal(Azure.Storage.Queues.QueueMessageEncoding.Base64, + StorageClientOptions.Queue(Azure.Storage.Queues.QueueMessageEncoding.Base64).MessageEncoding); + Assert.Equal(Azure.Storage.Queues.QueueMessageEncoding.None, StorageClientOptions.Queue().MessageEncoding); + } +} diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/StorageErrorsTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/StorageErrorsTests.cs index b5e5f6ba..bc257238 100644 --- a/src/Backend/AutopilotMonitor.Functions.Tests/StorageErrorsTests.cs +++ b/src/Backend/AutopilotMonitor.Functions.Tests/StorageErrorsTests.cs @@ -1,6 +1,9 @@ using Azure; using AutopilotMonitor.Functions.Helpers; using Xunit; +using System; +using System.IO; +using System.Net.Http; namespace AutopilotMonitor.Functions.Tests; @@ -59,4 +62,40 @@ public class StorageErrorsTests Assert.False(StorageErrors.IsPayloadTooLarge(null!)); Assert.False(StorageErrors.IsAlreadyExists(null!)); } + + [Fact] + public void No_response_at_all_is_transient() + => Assert.True(StorageErrors.IsTransient(Failure(0))); + + // ---- Exception overload: what an exhausted SDK budget looks like on a path without a caller token ---- + + [Fact] + public void Sdk_network_timeout_is_transient() + => Assert.True(StorageErrors.IsTransient((Exception)new TaskCanceledException("exceeded the configured timeout"))); + + [Fact] + public void Socket_failures_are_transient() + { + Assert.True(StorageErrors.IsTransient(new IOException("connection reset"))); + Assert.True(StorageErrors.IsTransient(new HttpRequestException("name resolution failed"))); + } + + [Fact] + public void Aggregate_of_retried_attempts_is_transient_only_when_every_attempt_was() + { + var allTransient = new AggregateException(new TaskCanceledException(), Failure(503), new IOException()); + var mixed = new AggregateException(new TaskCanceledException(), Failure(404)); + + Assert.True(StorageErrors.IsTransient(allTransient)); + Assert.False(StorageErrors.IsTransient(mixed)); + Assert.False(StorageErrors.IsTransient(new AggregateException())); + } + + [Fact] + public void Ordinary_exceptions_are_not_transient() + { + Assert.False(StorageErrors.IsTransient(new InvalidOperationException())); + Assert.False(StorageErrors.IsTransient((Exception)Failure(412))); + Assert.False(StorageErrors.IsTransient((Exception?)null!)); + } } diff --git a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSessionRepository.cs b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSessionRepository.cs index f7151245..a80cf4e3 100644 --- a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSessionRepository.cs +++ b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSessionRepository.cs @@ -57,8 +57,8 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage public Task> GetSessionsPageAsync(string tenantId, int? days, int pageSize, string? continuation) => _storage.GetSessionsPageAsync(tenantId, days, pageSize, continuation); - public Task> GetAllSessionsPageAsync(string? tenantIdFilter, int? days, int pageSize, string? continuation, IReadOnlyCollection? allowedTenantIds = null, IEnumerable? select = null) - => _storage.GetAllSessionsPageAsync(tenantIdFilter, days, pageSize, continuation, allowedTenantIds, select); + public Task> GetAllSessionsPageAsync(string? tenantIdFilter, int? days, int pageSize, string? continuation, IReadOnlyCollection? allowedTenantIds = null, IEnumerable? select = null, CancellationToken cancellationToken = default) + => _storage.GetAllSessionsPageAsync(tenantIdFilter, days, pageSize, continuation, allowedTenantIds, select, cancellationToken); public Task GetSessionStatsAsync(string tenantId, int days) => _storage.GetSessionStatsAsync(tenantId, days); diff --git a/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestTelemetryFunction.cs b/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestTelemetryFunction.cs index 77d0d627..6e8e9fa0 100644 --- a/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestTelemetryFunction.cs +++ b/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestTelemetryFunction.cs @@ -335,6 +335,17 @@ namespace AutopilotMonitor.Functions.Functions.Ingest _logger.LogWarning(ex, "IngestTelemetry: storage request failed with status {Status} ({Code}) — answering {Http}", ex.Status, ex.ErrorCode, (int)status); return AsOutput(await WriteErrorAsync(req, status, message, retryAfter)); } + catch (Exception ex) when (StorageErrors.IsTransient(ex)) + { + // The SDK ran out of its budget (StorageClientOptions: per-attempt timeout × + // retries, sized under the agent's 30 s) or the connection failed. A later replay + // can succeed, so 503 + Retry-After keeps the agent on its retry ladder while it + // is still listening; the generic 500 below would be replayed too, but without + // the wait hint and logged as a bug. + _logger.LogWarning(ex, "IngestTelemetry: storage call exhausted its budget ({Type}) — answering 503", ex.GetType().Name); + return AsOutput(await WriteErrorAsync(req, HttpStatusCode.ServiceUnavailable, + "Storage temporarily unavailable; retry later", StorageRetryAfterSeconds)); + } catch (Exception ex) { _logger.LogError(ex, "IngestTelemetry: unhandled exception"); diff --git a/src/Backend/AutopilotMonitor.Functions/Functions/Sessions/GetAllSessionsFunction.cs b/src/Backend/AutopilotMonitor.Functions/Functions/Sessions/GetAllSessionsFunction.cs index b8ed4e5f..bada80d9 100644 --- a/src/Backend/AutopilotMonitor.Functions/Functions/Sessions/GetAllSessionsFunction.cs +++ b/src/Backend/AutopilotMonitor.Functions/Functions/Sessions/GetAllSessionsFunction.cs @@ -72,8 +72,11 @@ namespace AutopilotMonitor.Functions.Functions.Sessions } } + // The request token ends the per-tenant fan-out when the browser tab closes or + // the caller gives up — a cross-tenant page nobody reads must not keep 32 queries alive. var page = await _sessionRepo.GetAllSessionsPageAsync( - parsed.FilterTenantId, parsed.Days, parsed.PageSize, azureToken, allowedTenantIds); + parsed.FilterTenantId, parsed.Days, parsed.PageSize, azureToken, allowedTenantIds, + cancellationToken: req.FunctionContext.CancellationToken); string? nextLink = null; if (!string.IsNullOrEmpty(page.NextRawToken)) diff --git a/src/Backend/AutopilotMonitor.Functions/Helpers/StorageClientOptions.cs b/src/Backend/AutopilotMonitor.Functions/Helpers/StorageClientOptions.cs new file mode 100644 index 00000000..46f894a0 --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions/Helpers/StorageClientOptions.cs @@ -0,0 +1,116 @@ +using System; +using Azure.Core; +using Azure.Data.Tables; +using Azure.Storage.Blobs; +using Azure.Storage.Queues; + +namespace AutopilotMonitor.Functions.Helpers; + +/// +/// One place that sizes the Azure Storage SDK retry and timeout budgets. The SDK defaults +/// (100 s network timeout, 3 retries, exponential backoff from 0.8 s) let one hanging Table call +/// hold a request for minutes, while the agent gives up on its upload after +/// and replays the batch — under a storage brownout that fills +/// every worker slot with requests nobody is waiting for and doubles the load (audit 2026-09-05 +/// F09). The hot-path budget therefore keeps the WORST CASE of one SDK operation (every attempt +/// running into its timeout, every backoff at its cap) under the agent's timeout, so a request +/// that cannot finish fails as 503 + Retry-After while the agent is still listening and its own +/// retry ladder takes over. Pinned by StorageClientOptionsTests. +/// +/// Table and Queue share the hot-path budget: the ingest writes tables and enqueues follow-up +/// work on every batch. Blob is not on the ingest path (diagnostics uploads go straight to the +/// SAS) and carries single-shot uploads of tens of MB (IME MSI archive, report ZIPs), so it gets +/// a bounded but generous budget. No extra retry layer (Polly) on top of the SDK: that would +/// multiply retries instead of bounding them. +/// +/// +/// Known gap: a service Retry-After header on a 429/503 can stretch one backoff beyond +/// the computed cap — Azure.Core honours the larger of the two. Storage answers with small +/// values; the bound below is the code-side guarantee, not a hard ceiling on service hints. +/// +/// +public static class StorageClientOptions +{ + /// + /// The agent's upload HttpClient timeout (MtlsHttpClientFactory.DefaultTimeout on the + /// agent side). Every hot-path storage call chain must stay under it. + /// + public static readonly TimeSpan AgentUploadTimeout = TimeSpan.FromSeconds(30); + + /// Per-attempt network timeout for Table and Queue calls. + public static readonly TimeSpan HotPathNetworkTimeout = TimeSpan.FromSeconds(8); + + /// Retries after the first attempt for Table and Queue calls. + public const int HotPathMaxRetries = 2; + + public static readonly TimeSpan HotPathRetryDelay = TimeSpan.FromMilliseconds(500); + public static readonly TimeSpan HotPathMaxRetryDelay = TimeSpan.FromSeconds(4); + + /// Per-attempt network timeout for Blob calls (single-shot uploads, streamed reads). + public static readonly TimeSpan BlobNetworkTimeout = TimeSpan.FromSeconds(60); + public const int BlobMaxRetries = 3; + + /// + /// Azure.Core adds up to 20 % jitter on top of the exponential delay; the worst-case bound + /// accounts for it. + /// + internal const double BackoffJitterFactor = 1.2; + + public static TableClientOptions Table() + { + var options = new TableClientOptions(); + ApplyHotPath(options.Retry); + return options; + } + + public static QueueClientOptions Queue(QueueMessageEncoding encoding = QueueMessageEncoding.None) + { + var options = new QueueClientOptions { MessageEncoding = encoding }; + ApplyHotPath(options.Retry); + return options; + } + + public static BlobClientOptions Blob() + { + var options = new BlobClientOptions(); + options.Retry.Mode = RetryMode.Exponential; + options.Retry.MaxRetries = BlobMaxRetries; + options.Retry.Delay = HotPathRetryDelay; + options.Retry.MaxDelay = HotPathMaxRetryDelay; + options.Retry.NetworkTimeout = BlobNetworkTimeout; + return options; + } + + private static void ApplyHotPath(RetryOptions retry) + { + retry.Mode = RetryMode.Exponential; + retry.MaxRetries = HotPathMaxRetries; + retry.Delay = HotPathRetryDelay; + retry.MaxDelay = HotPathMaxRetryDelay; + retry.NetworkTimeout = HotPathNetworkTimeout; + } + + /// + /// Upper bound on the wall-clock one SDK operation can take under these options: every + /// attempt runs into , every backoff is the + /// exponential delay (with jitter) capped at . + /// + public static TimeSpan WorstCase(RetryOptions retry) + { + if (retry == null) throw new ArgumentNullException(nameof(retry)); + + var attempts = retry.MaxRetries + 1; + var total = TimeSpan.FromTicks(retry.NetworkTimeout.Ticks * attempts); + + for (var attempt = 0; attempt < retry.MaxRetries; attempt++) + { + var backoff = retry.Mode == RetryMode.Exponential + ? TimeSpan.FromTicks((long)(retry.Delay.Ticks * Math.Pow(2, attempt) * BackoffJitterFactor)) + : retry.Delay; + if (backoff > retry.MaxDelay) backoff = retry.MaxDelay; + total += backoff; + } + + return total; + } +} diff --git a/src/Backend/AutopilotMonitor.Functions/Helpers/StorageErrors.cs b/src/Backend/AutopilotMonitor.Functions/Helpers/StorageErrors.cs index 96bd828c..40530780 100644 --- a/src/Backend/AutopilotMonitor.Functions/Helpers/StorageErrors.cs +++ b/src/Backend/AutopilotMonitor.Functions/Helpers/StorageErrors.cs @@ -1,4 +1,7 @@ using System; +using System.IO; +using System.Linq; +using System.Net.Http; using Azure; namespace AutopilotMonitor.Functions.Helpers; @@ -21,9 +24,31 @@ public static class StorageErrors "PropertyValueTooLarge", ]; - /// Throttling, timeouts and service-side failures — a later retry can succeed. + /// + /// Throttling, timeouts and service-side failures — a later retry can succeed. Status 0 is + /// the SDK's "no response" (connection refused/reset before any status arrived). + /// public static bool IsTransient(RequestFailedException ex) - => ex != null && Array.IndexOf(TransientStatuses, ex.Status) >= 0; + => ex != null && (ex.Status == 0 || Array.IndexOf(TransientStatuses, ex.Status) >= 0); + + /// + /// The exception shapes an exhausted SDK budget surfaces as (see StorageClientOptions): + /// the per-attempt network timeout is an , a socket + /// failure an / , and after + /// several attempts Azure.Core wraps them in an . Only for + /// call paths WITHOUT a caller cancellation token (the ingest has none): there every + /// cancellation is the SDK's own timeout, never the caller giving up. + /// + public static bool IsTransient(Exception ex) => ex switch + { + null => false, + RequestFailedException failed => IsTransient(failed), + OperationCanceledException => true, + IOException => true, + HttpRequestException => true, + AggregateException aggregate => aggregate.InnerExceptions.Count > 0 && aggregate.InnerExceptions.All(IsTransient), + _ => false, + }; /// /// The transaction or entity exceeds a size limit. HTTP 413 in production; some backends diff --git a/src/Backend/AutopilotMonitor.Functions/Middleware/RequestTelemetryMiddleware.cs b/src/Backend/AutopilotMonitor.Functions/Middleware/RequestTelemetryMiddleware.cs index d2b438b9..3632e096 100644 --- a/src/Backend/AutopilotMonitor.Functions/Middleware/RequestTelemetryMiddleware.cs +++ b/src/Backend/AutopilotMonitor.Functions/Middleware/RequestTelemetryMiddleware.cs @@ -18,6 +18,9 @@ namespace AutopilotMonitor.Functions.Middleware; /// public class RequestTelemetryMiddleware : IFunctionsWorkerMiddleware { + /// GC pause during the request below this is not worth a column on the row. + internal static readonly TimeSpan GcPauseStampThreshold = TimeSpan.FromMilliseconds(100); + private readonly TelemetryClient _telemetryClient; // Exact allowlist of device endpoints that READ + validate the X-Tenant-Id header (cert auth @@ -52,6 +55,7 @@ public class RequestTelemetryMiddleware : IFunctionsWorkerMiddleware var sw = Stopwatch.StartNew(); var startTime = DateTimeOffset.UtcNow; + var gcPauseAtStart = GC.GetTotalPauseDuration(); Exception? caughtException = null; try @@ -170,6 +174,14 @@ public class RequestTelemetryMiddleware : IFunctionsWorkerMiddleware if (caughtException != null) requestTelemetry.Properties["ExceptionType"] = caughtException.GetType().Name; + // Process-wide GC pause time that elapsed while this request was in flight — not the + // request's own allocation cost. The point is the opposite: a slow request whose + // duration is mostly GC pause was stalled by the process, not by its own work (the + // ProcessStall event says which kind of stall). Stamped only when worth a look. + var gcPause = GC.GetTotalPauseDuration() - gcPauseAtStart; + if (gcPause >= GcPauseStampThreshold) + requestTelemetry.Properties["GcPauseMs"] = Math.Round(gcPause.TotalMilliseconds).ToString(System.Globalization.CultureInfo.InvariantCulture); + // Bypass the worker's adaptive sampling for this item. The SDK sampling processor // passes through any item whose SamplingPercentage is already set, so this is the // per-item opt-out that host.json "excludedTypes" (host process only) cannot provide. diff --git a/src/Backend/AutopilotMonitor.Functions/Program.cs b/src/Backend/AutopilotMonitor.Functions/Program.cs index 752e0552..17f3614d 100644 --- a/src/Backend/AutopilotMonitor.Functions/Program.cs +++ b/src/Backend/AutopilotMonitor.Functions/Program.cs @@ -235,6 +235,7 @@ builder.Services.AddHostedService(); // Initialize all tables at startup builder.Services.AddSingleton(startupClock); builder.Services.AddHostedService(); // BackendStartupMs / BackendTableInitMs metrics on ApplicationStarted +builder.Services.AddHostedService(); // ProcessStall event + metric: process-wide pauses attributed to GC / thread-pool starvation // Data Access Layer — repository interfaces backed by Table Storage. // To switch to Cosmos DB: replace AddTableStorageDataAccess() with AddCosmosDataAccess(). diff --git a/src/Backend/AutopilotMonitor.Functions/Services/BlobStorageService.cs b/src/Backend/AutopilotMonitor.Functions/Services/BlobStorageService.cs index 95766712..e803b47a 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/BlobStorageService.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/BlobStorageService.cs @@ -10,6 +10,7 @@ using Azure.Storage.Sas; using AutopilotMonitor.Shared.Models.Deletion; using Microsoft.Extensions.Configuration; using Microsoft.Extensions.Logging; +using AutopilotMonitor.Functions.Helpers; namespace AutopilotMonitor.Functions.Services { @@ -40,13 +41,13 @@ namespace AutopilotMonitor.Functions.Services if (!string.IsNullOrEmpty(storageAccountName)) { var blobUri = new Uri($"https://{storageAccountName}.blob.core.windows.net"); - _blobServiceClient = new BlobServiceClient(blobUri, new DefaultAzureCredential()); + _blobServiceClient = new BlobServiceClient(blobUri, new DefaultAzureCredential(), StorageClientOptions.Blob()); _usesManagedIdentity = true; _logger.LogInformation("Blob Storage initialized with Managed Identity (account: {Account})", storageAccountName); } else if (!string.IsNullOrEmpty(connectionString)) { - _blobServiceClient = new BlobServiceClient(connectionString); + _blobServiceClient = new BlobServiceClient(connectionString, StorageClientOptions.Blob()); _usesManagedIdentity = false; _logger.LogInformation("Blob Storage initialized with connection string"); } diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Caching/SingleFlightCache.cs b/src/Backend/AutopilotMonitor.Functions/Services/Caching/SingleFlightCache.cs index 0a0e35cc..1b2ca1f5 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/Caching/SingleFlightCache.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/Caching/SingleFlightCache.cs @@ -20,10 +20,22 @@ namespace AutopilotMonitor.Functions.Services.Caching /// /// The factory runs WITHOUT any caller's cancellation token on purpose: the task is shared, /// so one caller's abort must not fault every other waiter — and a finished result is exactly - /// what the aborted caller's retry wants to find. + /// what the aborted caller's retry wants to find. What it runs on instead is the cache's own + /// : a factory that hangs (a cross-tenant fan-out on a slow + /// storage account) would otherwise hold every waiter until the SDK's worst case; past the + /// budget the shared task faults with , the entry is + /// evicted, and the next caller starts fresh. Factories that ignore the token they are handed + /// are not bounded — pass it into the SDK calls. /// internal sealed class SingleFlightCache { + /// + /// Wall-clock one factory run may take. Sized for the cross-tenant fan-outs (32 parallel + /// partition queries, each under the hot-path SDK budget) — a single-flight aggregate that + /// needs longer is a scan that belongs in a job, not behind a portal request. + /// + internal static readonly TimeSpan DefaultFactoryBudget = TimeSpan.FromSeconds(30); + private sealed class Entry { public Entry(Lazy> task, DateTime expiresUtc) @@ -39,6 +51,7 @@ namespace AutopilotMonitor.Functions.Services.Caching private readonly ConcurrentDictionary _entries = new(StringComparer.Ordinal); private readonly Func _utcNow; private readonly int _maxEntries; + private readonly TimeSpan _factoryBudget; /// /// Upper bound on live entries. When a NEW key is added beyond it, every other key is @@ -47,20 +60,33 @@ namespace AutopilotMonitor.Functions.Services.Caching /// public SingleFlightCache(int maxEntries = int.MaxValue) : this(() => DateTime.UtcNow, maxEntries) { } - /// Clock seam for tests. - internal SingleFlightCache(Func utcNow, int maxEntries = int.MaxValue) + /// Clock and budget seams for tests. + internal SingleFlightCache(Func utcNow, int maxEntries = int.MaxValue, TimeSpan? factoryBudget = null) { if (maxEntries < 1) throw new ArgumentOutOfRangeException(nameof(maxEntries)); _utcNow = utcNow; _maxEntries = maxEntries; + _factoryBudget = factoryBudget ?? DefaultFactoryBudget; + } + + /// + /// for a + /// factory with nothing to cancel (in-memory work, or SDK calls that take no token). + /// + public Task GetOrAddAsync(string key, TimeSpan ttl, Func> factory) + { + if (factory == null) throw new ArgumentNullException(nameof(factory)); + return GetOrAddAsync(key, ttl, _ => factory()); } /// /// Returns the cached value for , or runs /// exactly once for all concurrent callers and caches its result for . - /// A factory failure propagates to every waiter and leaves nothing behind. + /// The token handed to the factory is the cache's budget, never a caller's. A factory + /// failure (including the budget running out) propagates to every waiter and leaves + /// nothing behind. /// - public async Task GetOrAddAsync(string key, TimeSpan ttl, Func> factory) + public async Task GetOrAddAsync(string key, TimeSpan ttl, Func> factory) { if (key == null) throw new ArgumentNullException(nameof(key)); if (factory == null) throw new ArgumentNullException(nameof(factory)); @@ -131,12 +157,18 @@ namespace AutopilotMonitor.Functions.Services.Caching } } - private static Entry CreateEntry(Func> factory, DateTime now, TimeSpan ttl) + private Entry CreateEntry(Func> factory, DateTime now, TimeSpan ttl) { // Lazy defers the factory to the first awaiter, so it runs on the caller's context // and any synchronous prefix stays attributable to that request. - var lazy = new Lazy>(factory, LazyThreadSafetyMode.ExecutionAndPublication); + var lazy = new Lazy>(() => RunWithinBudgetAsync(factory), LazyThreadSafetyMode.ExecutionAndPublication); return new Entry(lazy, now.Add(ttl)); } + + private async Task RunWithinBudgetAsync(Func> factory) + { + using var budget = new CancellationTokenSource(_factoryBudget); + return await factory(budget.Token).ConfigureAwait(false); + } } } diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Diagnostics/HostedDiagnosticsBlobService.cs b/src/Backend/AutopilotMonitor.Functions/Services/Diagnostics/HostedDiagnosticsBlobService.cs index cb0c7fe3..956146fe 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/Diagnostics/HostedDiagnosticsBlobService.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/Diagnostics/HostedDiagnosticsBlobService.cs @@ -13,6 +13,7 @@ using Azure.Storage.Blobs.Models; using Azure.Storage.Sas; using Microsoft.Extensions.Configuration; using Microsoft.Extensions.Logging; +using AutopilotMonitor.Functions.Helpers; namespace AutopilotMonitor.Functions.Services.Diagnostics { @@ -66,7 +67,7 @@ namespace AutopilotMonitor.Functions.Services.Diagnostics if (!string.IsNullOrEmpty(storageAccountName)) { var blobUri = new Uri($"https://{storageAccountName}.blob.core.windows.net"); - _blobServiceClient = new BlobServiceClient(blobUri, new DefaultAzureCredential()); + _blobServiceClient = new BlobServiceClient(blobUri, new DefaultAzureCredential(), StorageClientOptions.Blob()); _usesManagedIdentity = true; _logger.LogInformation( "HostedDiagnosticsBlobService initialized with Managed Identity (account: {Account}, container: {Container})", @@ -74,7 +75,7 @@ namespace AutopilotMonitor.Functions.Services.Diagnostics } else if (!string.IsNullOrEmpty(connectionString)) { - _blobServiceClient = new BlobServiceClient(connectionString); + _blobServiceClient = new BlobServiceClient(connectionString, StorageClientOptions.Blob()); _usesManagedIdentity = false; _logger.LogInformation( "HostedDiagnosticsBlobService initialized with connection string (container: {Container})", diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Queueing/QueueClientFactory.cs b/src/Backend/AutopilotMonitor.Functions/Services/Queueing/QueueClientFactory.cs index 3eb7ee7c..2c70a095 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/Queueing/QueueClientFactory.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/Queueing/QueueClientFactory.cs @@ -2,6 +2,7 @@ using System; using Azure.Identity; using Azure.Storage.Queues; using Microsoft.Extensions.Configuration; +using AutopilotMonitor.Functions.Helpers; namespace AutopilotMonitor.Functions.Services.Queueing { @@ -48,12 +49,12 @@ namespace AutopilotMonitor.Functions.Services.Queueing $"https://{storageAccountName}.queue.core.windows.net/{queueName}"); return new QueueClient(uri, credential, options); }; - _serviceBuilder = () => new QueueServiceClient(serviceUri, credential); + _serviceBuilder = () => new QueueServiceClient(serviceUri, credential, StorageClientOptions.Queue()); } else if (!string.IsNullOrEmpty(connectionString)) { _builder = (queueName, options) => new QueueClient(connectionString, queueName, options); - _serviceBuilder = () => new QueueServiceClient(connectionString); + _serviceBuilder = () => new QueueServiceClient(connectionString, StorageClientOptions.Queue()); } else { @@ -82,9 +83,8 @@ namespace AutopilotMonitor.Functions.Services.Queueing if (string.IsNullOrWhiteSpace(queueName)) throw new ArgumentException("Queue name must not be empty.", nameof(queueName)); - var options = base64 - ? new QueueClientOptions { MessageEncoding = QueueMessageEncoding.Base64 } - : new QueueClientOptions(); + // Hot-path budget (StorageClientOptions): producers enqueue on every ingest batch. + var options = StorageClientOptions.Queue(base64 ? QueueMessageEncoding.Base64 : QueueMessageEncoding.None); return _builder(queueName, options); } diff --git a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Metrics.cs b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Metrics.cs index fdaba36e..b0668850 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Metrics.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Metrics.cs @@ -610,11 +610,11 @@ namespace AutopilotMonitor.Functions.Services { var bucketStart = new DateTime(sinceUtc.Ticks - sinceUtc.Ticks % TimeSpan.TicksPerMinute, DateTimeKind.Utc); var snapshot = await _appSummarySnapshotCache.GetOrAddAsync( - bucketStart.ToString("yyyyMMddHHmm"), AppSummarySnapshotTtl, () => ScanCrossTenantAppSummariesAsync(bucketStart)); + bucketStart.ToString("yyyyMMddHHmm"), AppSummarySnapshotTtl, ct => ScanCrossTenantAppSummariesAsync(bucketStart, ct)); return new List(snapshot); } - private async Task> ScanCrossTenantAppSummariesAsync(DateTime sinceUtc) + private async Task> ScanCrossTenantAppSummariesAsync(DateTime sinceUtc, CancellationToken cancellationToken) { var tableClient = _tableServiceClient.GetTableClient(Constants.TableNames.AppInstallSummaries); var windowFilter = AppSummaryWindowFilter(sinceUtc); @@ -624,12 +624,12 @@ namespace AutopilotMonitor.Functions.Services if (tenantIds.Count == 0) { // Empty config table (fresh install): the legacy cross-partition scan is the safety net. - return await ScanAppInstallSummariesAsync(tableClient, windowFilter, select, CancellationToken.None); + return await ScanAppInstallSummariesAsync(tableClient, windowFilter, select, cancellationToken); } return await BoundedFanOut.RunAsync(tenantIds, BoundedFanOut.CrossTenantConcurrency, (tenantId, ct) => ScanAppInstallSummariesAsync(tableClient, $"PartitionKey eq '{tenantId}' and {windowFilter}", select, ct), - CancellationToken.None); + cancellationToken); } /// diff --git a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Sessions.cs b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Sessions.cs index 87d1ce82..fd6e783c 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Sessions.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Sessions.cs @@ -1162,7 +1162,8 @@ namespace AutopilotMonitor.Functions.Services public async Task> GetAllSessionsPageAsync( string? tenantIdFilter, int? days, int pageSize, string? continuation, IReadOnlyCollection? allowedTenantIds = null, - IEnumerable? select = null) + IEnumerable? select = null, + CancellationToken cancellationToken = default) { if (pageSize < 1) throw new ArgumentOutOfRangeException(nameof(pageSize)); @@ -1178,7 +1179,7 @@ namespace AutopilotMonitor.Functions.Services return new RawPage(tenantPage.Sessions, tenantPage.HasMore ? tenantPage.NextCursor : null); } - var page = await FetchAllSessionsPageInternalAsync(maxResults: pageSize, cursor: continuation, days: days, allowedTenantIds: allowedTenantIds, select: select); + var page = await FetchAllSessionsPageInternalAsync(maxResults: pageSize, cursor: continuation, days: days, allowedTenantIds: allowedTenantIds, select: select, cancellationToken: cancellationToken); return new RawPage(page.Sessions, page.HasMore ? page.NextCursor : null); } @@ -1278,10 +1279,10 @@ namespace AutopilotMonitor.Functions.Services var cacheKey = $"{days}|{DescribeTenantBound(allowedTenantIds)}"; return await _sessionStatsCache.GetOrAddAsync(cacheKey, SessionStatsCacheTtl, - () => ComputeAllSessionStatsAsync(days, allowedTenantIds)); + ct => ComputeAllSessionStatsAsync(days, allowedTenantIds, ct)); } - private async Task ComputeAllSessionStatsAsync(int days, IReadOnlyCollection? allowedTenantIds) + private async Task ComputeAllSessionStatsAsync(int days, IReadOnlyCollection? allowedTenantIds, CancellationToken cancellationToken) { var tenantIds = ApplyTenantBound(await GetTenantIdsCachedAsync(), allowedTenantIds); if (tenantIds.Count == 0) @@ -1296,7 +1297,7 @@ namespace AutopilotMonitor.Functions.Services } var rows = await DrainSessionsIndexWindowAsync( - tenantIds, startUtc: DateTime.UtcNow.AddDays(-days), endUtc: null, SessionStatsProjection, CancellationToken.None); + tenantIds, startUtc: DateTime.UtcNow.AddDays(-days), endUtc: null, SessionStatsProjection, cancellationToken); return AggregateSessionStats(rows, days); } @@ -1461,7 +1462,7 @@ namespace AutopilotMonitor.Functions.Services /// private async Task<(List Sessions, bool HasMore, string? NextCursor)> FetchAllSessionsPageInternalAsync( int maxResults, string? cursor, int? days, IReadOnlyCollection? allowedTenantIds = null, - IEnumerable? select = null) + IEnumerable? select = null, CancellationToken cancellationToken = default) { try { @@ -1531,7 +1532,7 @@ namespace AutopilotMonitor.Functions.Services if (tenantSessions.Count >= fetchPerTenant) break; } return tenantSessions; - }, CancellationToken.None); + }, cancellationToken); // Step 3: Merge-sort across tenants by StartedAt descending allSessions = allSessions.OrderByDescending(s => s.StartedAt).ToList(); diff --git a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.cs b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.cs index ea4ad09d..f8dc36e7 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.cs @@ -12,6 +12,7 @@ using Microsoft.Extensions.Configuration; using Microsoft.Extensions.Logging; using Newtonsoft.Json; using Newtonsoft.Json.Linq; +using AutopilotMonitor.Functions.Helpers; namespace AutopilotMonitor.Functions.Services { @@ -50,7 +51,7 @@ namespace AutopilotMonitor.Functions.Services /// storage failure so callers keep their own fallback semantics (empty page vs. legacy scan). /// internal Task> GetTenantIdsCachedAsync() - => _tenantIdCache.GetOrAddAsync(TenantIdCacheKey, TenantIdCacheTtl, () => QueryTenantIdsAsync(CancellationToken.None)); + => _tenantIdCache.GetOrAddAsync(TenantIdCacheKey, TenantIdCacheTtl, ct => QueryTenantIdsAsync(ct)); private async Task> QueryTenantIdsAsync(CancellationToken cancellationToken) { @@ -98,13 +99,13 @@ namespace AutopilotMonitor.Functions.Services { // Managed Identity: use DefaultAzureCredential with storage account name var tableUri = new Uri($"https://{storageAccountName}.table.core.windows.net"); - _tableServiceClient = new TableServiceClient(tableUri, new DefaultAzureCredential()); + _tableServiceClient = new TableServiceClient(tableUri, new DefaultAzureCredential(), StorageClientOptions.Table()); _logger.LogInformation("Table Storage initialized with Managed Identity (account: {Account})", storageAccountName); } else if (!string.IsNullOrEmpty(connectionString)) { // Fallback: connection string (local dev, legacy) - _tableServiceClient = new TableServiceClient(connectionString); + _tableServiceClient = new TableServiceClient(connectionString, StorageClientOptions.Table()); _logger.LogInformation("Table Storage initialized with connection string"); } else diff --git a/src/Backend/AutopilotMonitor.Functions/Telemetry/ProcessStallMonitor.cs b/src/Backend/AutopilotMonitor.Functions/Telemetry/ProcessStallMonitor.cs new file mode 100644 index 00000000..3cb28306 --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions/Telemetry/ProcessStallMonitor.cs @@ -0,0 +1,241 @@ +using System; +using System.Collections.Generic; +using System.Diagnostics; +using System.Globalization; +using System.Threading; +using System.Threading.Tasks; +using AutopilotMonitor.Functions.Services; +using Microsoft.ApplicationInsights; +using Microsoft.ApplicationInsights.Channel; +using Microsoft.ApplicationInsights.DataContracts; +using Microsoft.ApplicationInsights.Metrics; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; + +namespace AutopilotMonitor.Functions.Telemetry; + +/// +/// Point-in-time reading of the runtime counters a process-wide pause leaves traces in. Two +/// consecutive snapshots around a late timer tick say WHAT paused the process: the GC pause +/// clock (, exact) or the thread pool (queued work, +/// thread injection). +/// +public readonly record struct ProcessSnapshot( + TimeSpan GcPause, + int Gen0, + int Gen1, + int Gen2, + int ThreadPoolThreads, + long PendingWorkItems, + long CompletedWorkItems, + long WorkingSetBytes, + long HeapSizeBytes, + long CommittedBytes, + int LastGcGeneration, + TimeSpan LastGcPause, + bool LastGcCompacted) +{ + public static ProcessSnapshot Capture() + { + var info = GC.GetGCMemoryInfo(); + var lastPause = TimeSpan.Zero; + foreach (var pause in info.PauseDurations) + lastPause += pause; + + return new ProcessSnapshot( + GcPause: GC.GetTotalPauseDuration(), + Gen0: GC.CollectionCount(0), + Gen1: GC.CollectionCount(1), + Gen2: GC.CollectionCount(2), + ThreadPoolThreads: ThreadPool.ThreadCount, + PendingWorkItems: ThreadPool.PendingWorkItemCount, + CompletedWorkItems: ThreadPool.CompletedWorkItemCount, + WorkingSetBytes: Environment.WorkingSet, + HeapSizeBytes: info.HeapSizeBytes, + CommittedBytes: info.TotalCommittedBytes, + LastGcGeneration: info.Generation, + LastGcPause: lastPause, + LastGcCompacted: info.Compacted); + } +} + +/// What a stall report says caused the pause. +public static class StallCause +{ + /// At least half of the drift was GC pause time. + public const string Gc = "gc"; + + /// Work was queued or the pool injected threads: starvation, not GC. + public const string ThreadPool = "threadpool"; + + /// Neither clock explains the drift (CPU steal, host pause, page-in). + public const string Unknown = "unknown"; +} + +/// One classified stall: the drift plus the counter deltas that explain it. +public sealed record StallReport( + TimeSpan Drift, + string Cause, + TimeSpan GcPauseDelta, + int Gen0Delta, + int Gen1Delta, + int Gen2Delta, + int ThreadPoolThreadsDelta, + long CompletedWorkItemsDelta, + ProcessSnapshot After); + +/// +/// Pure classification so the rule is unit-testable without a live process pause. +/// +public static class ProcessStallClassifier +{ + /// Fraction of the drift the GC pause clock must cover to blame the GC. + internal const double GcShareThreshold = 0.5; + + public static StallReport Classify(TimeSpan drift, ProcessSnapshot before, ProcessSnapshot after) + { + var gcPauseDelta = after.GcPause - before.GcPause; + if (gcPauseDelta < TimeSpan.Zero) gcPauseDelta = TimeSpan.Zero; + + var threadsDelta = after.ThreadPoolThreads - before.ThreadPoolThreads; + + string cause; + if (drift > TimeSpan.Zero && gcPauseDelta.Ticks >= drift.Ticks * GcShareThreshold) + cause = StallCause.Gc; + else if (after.PendingWorkItems > 0 || threadsDelta > 0) + cause = StallCause.ThreadPool; + else + cause = StallCause.Unknown; + + return new StallReport( + Drift: drift, + Cause: cause, + GcPauseDelta: gcPauseDelta, + Gen0Delta: after.Gen0 - before.Gen0, + Gen1Delta: after.Gen1 - before.Gen1, + Gen2Delta: after.Gen2 - before.Gen2, + ThreadPoolThreadsDelta: threadsDelta, + CompletedWorkItemsDelta: after.CompletedWorkItems - before.CompletedWorkItems, + After: after); + } +} + +/// +/// Detects process-wide pauses in the worker and reports what caused them. Production showed +/// episodes where every in-flight request on ONE instance finished at the same instant after +/// 5–10 s, with no slow storage call, no warning, no cold start and idle CPU (2026-09-03 12:15 +/// and 19:00) — a pause of the whole process, invisible to per-minute performance counters. +/// +/// Mechanism: a 1 s timer tick measures its own lateness. A tick that arrives +/// or more late means the process did not run continuations for +/// that long. Two s around the tick attribute the drift: the GC +/// pause clock is exact, so a drift the GC covers is a GC pause; queued work or thread +/// injection means thread-pool starvation (the tick's own continuation waits in that queue); +/// anything else is . Cost: one delay per second per instance. +/// +/// +/// Output: customEvents with the deltas (sampling bypassed — +/// the events are rare and every one is wanted) and the pre-aggregated metric +/// (value = drift ms, dimension cause) for the trend. Pairs with +/// the GcPauseMs request dimension stamped by RequestTelemetryMiddleware, which says how +/// much of one slow request was GC pause. Query: +/// customEvents | where name == 'ProcessStall' | project timestamp, cloud_RoleInstance, customDimensions. +/// +/// +public sealed class ProcessStallMonitor : BackgroundService +{ + public const string EventName = "ProcessStall"; + public const string MetricName = "ProcessStall"; + + internal static readonly TimeSpan Tick = TimeSpan.FromSeconds(1); + + /// Lateness of one tick that counts as a stall. + internal static readonly TimeSpan StallThreshold = TimeSpan.FromSeconds(1); + + private readonly TelemetryClient? _telemetry; + private readonly Metric? _metric; + private readonly BackendBuildInfo _buildInfo; + private readonly ILogger _logger; + + public ProcessStallMonitor(BackendBuildInfo buildInfo, ILogger logger, TelemetryClient? telemetry = null) + { + _buildInfo = buildInfo; + _logger = logger; + _telemetry = telemetry; + _metric = telemetry?.GetMetric(MetricName, "cause"); + } + + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + var previous = ProcessSnapshot.Capture(); + var stopwatch = new Stopwatch(); + + while (!stoppingToken.IsCancellationRequested) + { + stopwatch.Restart(); + try + { + await Task.Delay(Tick, stoppingToken).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + break; + } + + var drift = stopwatch.Elapsed - Tick; + var current = ProcessSnapshot.Capture(); + + if (drift >= StallThreshold) + { + try + { + Report(ProcessStallClassifier.Classify(drift, previous, current)); + } + catch (Exception ex) + { + _logger.LogWarning(ex, "ProcessStallMonitor: failed to report a stall"); + } + } + + previous = current; + } + } + + private void Report(StallReport report) + { + var driftMs = Math.Round(report.Drift.TotalMilliseconds); + _metric?.TrackValue(driftMs, report.Cause); + + if (_telemetry == null) return; + + var after = report.After; + var evt = new EventTelemetry(EventName); + var props = evt.Properties; + props["driftMs"] = Format(driftMs); + props["cause"] = report.Cause; + props["gcPauseMs"] = Format(Math.Round(report.GcPauseDelta.TotalMilliseconds)); + props["gen0"] = Format(report.Gen0Delta); + props["gen1"] = Format(report.Gen1Delta); + props["gen2"] = Format(report.Gen2Delta); + props["lastGcGeneration"] = Format(after.LastGcGeneration); + props["lastGcPauseMs"] = Format(Math.Round(after.LastGcPause.TotalMilliseconds)); + props["lastGcCompacted"] = after.LastGcCompacted ? "true" : "false"; + props["heapMb"] = Format(after.HeapSizeBytes >> 20); + props["committedMb"] = Format(after.CommittedBytes >> 20); + props["workingSetMb"] = Format(after.WorkingSetBytes >> 20); + props["threadPoolThreads"] = Format(after.ThreadPoolThreads); + props["threadPoolThreadsDelta"] = Format(report.ThreadPoolThreadsDelta); + props["pendingWorkItems"] = Format(after.PendingWorkItems); + props["completedWorkItemsDelta"] = Format(report.CompletedWorkItemsDelta); + props["instance"] = Environment.MachineName; + props["version"] = _buildInfo.Version; + + // Rare by construction, and every one is evidence: opt out of the worker's adaptive + // sampling the same way the canonical request record does. + ((ISupportSampling)evt).SamplingPercentage = 100; + _telemetry.TrackEvent(evt); + } + + private static string Format(double value) => value.ToString(CultureInfo.InvariantCulture); + private static string Format(long value) => value.ToString(CultureInfo.InvariantCulture); +} diff --git a/src/Shared/AutopilotMonitor.Shared/DataAccess/ISessionRepository.cs b/src/Shared/AutopilotMonitor.Shared/DataAccess/ISessionRepository.cs index 0e3cf5c2..a61029ad 100644 --- a/src/Shared/AutopilotMonitor.Shared/DataAccess/ISessionRepository.cs +++ b/src/Shared/AutopilotMonitor.Shared/DataAccess/ISessionRepository.cs @@ -66,7 +66,8 @@ namespace AutopilotMonitor.Shared.DataAccess Task> GetAllSessionsPageAsync( string? tenantIdFilter, int? days, int pageSize, string? continuation, IReadOnlyCollection? allowedTenantIds = null, - IEnumerable? select = null); + IEnumerable? select = null, + CancellationToken cancellationToken = default); /// /// Server-side aggregation for the dashboard stats cards (per-tenant scope).