backend: keep slow storage successes and count CAS conflicts as a metric (audit F10)

The dependency filter dropped every successful storage call and the expected
412/409 outcomes, so storage-side latency and ETag contention were invisible
until a call failed outright (seven days of App Insights held zero storage
dependency rows).

- StorageDependencyFilterProcessor keeps a successful storage call at or above
  2 s and stamps SlowStorage=true; the operation name is on the row, the table
  name is not (the Azure SDK span carries neither URL nor table).
- StorageMetrics wraps TelemetryClient.GetMetric("StorageCasConflict",
  operation, table, outcome): pre-aggregated per minute, bypasses the worker's
  adaptive sampling, null-tolerant for tests.
- The CAS loops report conflicts: session event count, session counter
  reconcile, session store, tenant stats, MCP user/tenant usage.
- TableStorageService keeps its two-argument constructors (Moq matches by
  exact arity); DI resolves the three-argument overload.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Oliver Kieselbach
2026-09-05 22:09:08 +02:00
parent d0cb7e66f4
commit fc9e9d13e6
9 changed files with 176 additions and 3 deletions
@@ -1,3 +1,4 @@
using System;
using AutopilotMonitor.Functions.Telemetry;
using Microsoft.ApplicationInsights.Channel;
using Microsoft.ApplicationInsights.DataContracts;
@@ -39,6 +40,47 @@ public class StorageDependencyFilterProcessorTests
Assert.Empty(next.Received);
}
[Fact]
public void SlowSuccessfulStorageDependency_IsKeptAndStamped()
{
var (processor, next) = Build();
var dep = Dep("", success: true, type: "InProc | Microsoft.Tables");
dep.Name = "TableClient.SubmitTransaction";
dep.Duration = StorageDependencyFilterProcessor.SlowStorageThreshold;
processor.Process(dep);
var kept = Assert.Single(next.Received);
Assert.Equal("true", ((DependencyTelemetry)kept).Properties[StorageDependencyFilterProcessor.SlowStoragePropertyKey]);
}
[Fact]
public void FastSuccessfulStorageDependency_JustUnderThreshold_IsDropped()
{
var (processor, next) = Build();
var dep = Dep("myacct.table.core.windows.net", success: true);
dep.Duration = StorageDependencyFilterProcessor.SlowStorageThreshold - TimeSpan.FromMilliseconds(1);
processor.Process(dep);
Assert.Empty(next.Received);
Assert.False(dep.Properties.ContainsKey(StorageDependencyFilterProcessor.SlowStoragePropertyKey));
}
[Fact]
public void SlowExpectedFailure_IsStillDropped()
{
// 412/409/404 are control flow whatever their duration; only successes get the slow pass.
var (processor, next) = Build();
var dep = Dep("myacct.table.core.windows.net", success: false);
dep.ResultCode = "412";
dep.Duration = TimeSpan.FromSeconds(10);
processor.Process(dep);
Assert.Empty(next.Received);
}
[Fact]
public void SuccessfulStorageDependency_WithNullSuccess_IsDropped()
{
@@ -0,0 +1,33 @@
using AutopilotMonitor.Functions.Telemetry;
using Microsoft.ApplicationInsights;
using Microsoft.ApplicationInsights.Extensibility;
using Xunit;
namespace AutopilotMonitor.Functions.Tests;
public class StorageMetricsTests
{
[Fact]
public void Without_a_client_every_call_is_a_no_op()
{
var metrics = new StorageMetrics(null);
metrics.CasConflict("op", "Sessions", CasOutcome.Retried);
metrics.CasConflict("op", "Sessions", CasOutcome.Exhausted);
}
[Fact]
public void With_a_client_the_metric_is_registered_once_and_accepts_values()
{
using var config = new TelemetryConfiguration { DisableTelemetry = true };
var client = new TelemetryClient(config);
var metrics = new StorageMetrics(client);
metrics.CasConflict("IncrementSessionEventCount", "Sessions", CasOutcome.Retried);
metrics.CasConflict("IncrementSessionEventCount", "Sessions", CasOutcome.Exhausted);
// GetMetric returns the same pre-aggregator for the same identifier; a second wrapper
// over the same client must not fail on re-registration.
var again = new StorageMetrics(client);
again.CasConflict("MutateTenantStat", "PlatformStats", CasOutcome.Retried);
}
}
@@ -1,6 +1,7 @@
using AutopilotMonitor.Functions.Helpers;
using AutopilotMonitor.Functions.Security;
using AutopilotMonitor.Functions.Services;
using AutopilotMonitor.Functions.Telemetry;
using AutopilotMonitor.Shared;
using AutopilotMonitor.Shared.DataAccess;
using Azure;
@@ -20,12 +21,15 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage
private readonly TableClient _tableClient;
private readonly TableClient _tenantTableClient;
private readonly ILogger<TableUserUsageRepository> _logger;
private readonly StorageMetrics? _metrics;
public TableUserUsageRepository(
TableStorageService storage,
ILogger<TableUserUsageRepository> logger)
ILogger<TableUserUsageRepository> logger,
StorageMetrics? metrics = null)
{
_logger = logger;
_metrics = metrics;
_tableClient = storage.GetTableClient(Constants.TableNames.UserUsageLog);
_tenantTableClient = storage.GetTableClient(Constants.TableNames.McpTenantUsage);
}
@@ -67,15 +71,18 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage
}
catch (RequestFailedException addEx) when (addEx.Status == 409)
{
_metrics?.CasConflict("IncrementUserUsage", Constants.TableNames.UserUsageLog, CasOutcome.Retried);
continue;
}
}
catch (RequestFailedException ex) when (ex.Status == 412)
{
_metrics?.CasConflict("IncrementUserUsage", Constants.TableNames.UserUsageLog, CasOutcome.Retried);
continue;
}
}
_metrics?.CasConflict("IncrementUserUsage", Constants.TableNames.UserUsageLog, CasOutcome.Exhausted);
_logger.LogWarning("Failed to increment user usage after {MaxRetries} retries: user={UserId}, endpoint={Endpoint}",
maxRetries, LogSanitizer.Clean(userId), LogSanitizer.Clean(endpoint));
}
@@ -253,15 +260,18 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage
}
catch (RequestFailedException addEx) when (addEx.Status == 409)
{
_metrics?.CasConflict("IncrementTenantUsage", Constants.TableNames.McpTenantUsage, CasOutcome.Retried);
continue;
}
}
catch (RequestFailedException ex) when (ex.Status == 412)
{
_metrics?.CasConflict("IncrementTenantUsage", Constants.TableNames.McpTenantUsage, CasOutcome.Retried);
continue;
}
}
_metrics?.CasConflict("IncrementTenantUsage", Constants.TableNames.McpTenantUsage, CasOutcome.Exhausted);
_logger.LogWarning("Failed to increment tenant usage after {MaxRetries} retries: tenant={TenantId}, user={UserId}",
maxRetries, LogSanitizer.Clean(tenantId), LogSanitizer.Clean(userId));
}
@@ -75,10 +75,16 @@ builder.Services
// Drop successful Azure Storage dependencies (Table/Queue/Blob) from the worker telemetry
// pipeline to curb AppDependencies ingestion cost — this backend is storage-I/O heavy and those
// rows are high-volume, low-value. Failed storage calls and all non-storage dependencies
// (HTTP/Graph/SQL/SignalR) are preserved. See StorageDependencyFilterProcessor for the contract.
// rows are high-volume, low-value. Failed storage calls, slow successes (>= 2 s, stamped
// SlowStorage=true) and all non-storage dependencies (HTTP/Graph/SQL/SignalR) are preserved.
// See StorageDependencyFilterProcessor for the contract.
builder.Services.AddApplicationInsightsTelemetryProcessor<AutopilotMonitor.Functions.Telemetry.StorageDependencyFilterProcessor>();
// Pre-aggregated storage metrics (StorageCasConflict by operation/table/outcome): metrics bypass
// the worker's adaptive sampling and the dependency filter, so ETag conflicts stay visible as a
// trend although their 412/409 dependency rows are dropped.
builder.Services.AddSingleton<AutopilotMonitor.Functions.Telemetry.StorageMetrics>();
// Configure JWT Authentication for Multi-Tenant Azure AD
builder.Services.AddAuthentication(JwtBearerDefaults.AuthenticationScheme)
.AddMicrosoftIdentityWebApi(options =>
@@ -1,6 +1,7 @@
using Azure;
using Azure.Data.Tables;
using AutopilotMonitor.Functions.Helpers;
using AutopilotMonitor.Functions.Telemetry;
using AutopilotMonitor.Functions.Security;
using AutopilotMonitor.Functions.Services.Caching;
using AutopilotMonitor.Shared;
@@ -1046,6 +1047,8 @@ namespace AutopilotMonitor.Functions.Services
}
catch (RequestFailedException ex) when (ex.Status == 412 || ex.Status == 409)
{
_metrics?.CasConflict("MutateTenantStat", Constants.TableNames.PlatformStats,
attempt == TenantStatsCasRetries ? CasOutcome.Exhausted : CasOutcome.Retried);
if (attempt == TenantStatsCasRetries)
{
_logger.LogWarning(
@@ -2,6 +2,7 @@ using Azure;
using AutopilotMonitor.Functions.DataAccess.TableStorage;
using Azure.Data.Tables;
using AutopilotMonitor.Functions.Helpers;
using AutopilotMonitor.Functions.Telemetry;
using AutopilotMonitor.Functions.Pagination;
using AutopilotMonitor.Functions.Security;
using AutopilotMonitor.Functions.Services.Caching;
@@ -818,6 +819,7 @@ namespace AutopilotMonitor.Functions.Services
}
catch (RequestFailedException ex) when ((ex.Status == 412 || ex.Status == 409) && casAttempt < MaxStoreSessionCasAttempts - 1)
{
_metrics?.CasConflict("StoreSession", Constants.TableNames.Sessions, CasOutcome.Retried);
_logger.LogDebug(
"StoreSessionAsync ETag CAS conflict (status={Status}, attempt={Attempt}); retrying for tenant={TenantId} session={SessionId}",
ex.Status, casAttempt + 1, registration.TenantId, registration.SessionId);
@@ -2608,6 +2610,8 @@ namespace AutopilotMonitor.Functions.Services
catch (Azure.RequestFailedException ex) when (ex.Status == 412)
{
retryCount++;
_metrics?.CasConflict("IncrementSessionEventCount", Constants.TableNames.Sessions,
retryCount >= maxRetries ? CasOutcome.Exhausted : CasOutcome.Retried);
if (retryCount >= maxRetries)
{
_logger.LogWarning($"Failed to increment event count for session {sessionId} after {maxRetries} retries due to ETag conflicts");
@@ -2697,6 +2701,8 @@ namespace AutopilotMonitor.Functions.Services
catch (Azure.RequestFailedException ex) when (ex.Status == 412)
{
retryCount++;
_metrics?.CasConflict("ReconcileSessionCounters", Constants.TableNames.Sessions,
retryCount >= maxRetries ? CasOutcome.Exhausted : CasOutcome.Retried);
if (retryCount >= maxRetries)
{
_logger.LogWarning($"Failed to reconcile session counters for session {sessionId} after {maxRetries} retries due to ETag conflicts");
@@ -4,6 +4,7 @@ using Azure.Data.Tables;
using Azure.Identity;
using AutopilotMonitor.Functions.Security;
using AutopilotMonitor.Functions.Services.Caching;
using AutopilotMonitor.Functions.Telemetry;
using AutopilotMonitor.Shared;
using AutopilotMonitor.Shared.DataAccess;
using AutopilotMonitor.Shared.Models;
@@ -27,6 +28,7 @@ namespace AutopilotMonitor.Functions.Services
{
private readonly TableServiceClient _tableServiceClient;
private readonly ILogger<TableStorageService> _logger;
private readonly StorageMetrics? _metrics;
private bool _tablesInitialized = false;
private readonly object _initLock = new object();
@@ -80,8 +82,14 @@ namespace AutopilotMonitor.Functions.Services
}
public TableStorageService(IConfiguration configuration, ILogger<TableStorageService> logger)
: this(configuration, logger, null)
{
}
public TableStorageService(IConfiguration configuration, ILogger<TableStorageService> logger, StorageMetrics? metrics)
{
_logger = logger;
_metrics = metrics;
var connectionString = configuration["AzureTableStorageConnectionString"];
var storageAccountName = configuration["AzureStorageAccountName"];
@@ -114,7 +122,13 @@ namespace AutopilotMonitor.Functions.Services
/// even via InternalsVisibleTo.
/// </summary>
public TableStorageService(TableServiceClient tableServiceClient, ILogger<TableStorageService> logger)
: this(tableServiceClient, logger, null)
{
}
public TableStorageService(TableServiceClient tableServiceClient, ILogger<TableStorageService> logger, StorageMetrics? metrics)
{
_metrics = metrics;
_tableServiceClient = tableServiceClient;
_logger = logger;
}
@@ -1,3 +1,4 @@
using System;
using Microsoft.ApplicationInsights.Channel;
using Microsoft.ApplicationInsights.DataContracts;
using Microsoft.ApplicationInsights.Extensibility;
@@ -58,6 +59,18 @@ public sealed class StorageDependencyFilterProcessor : ITelemetryProcessor
// Expected storage outcomes that are normal control flow, not failures worth billing.
private static readonly string[] ExpectedStorageStatusCodes = { "404", "412", "409" };
/// <summary>
/// A successful storage call at or above this duration is kept and stamped
/// <see cref="SlowStoragePropertyKey"/>: dropping every success left storage-side latency
/// invisible until a call failed outright (audit 2026-09-05 F10). Slow successes are rare
/// by definition, so keeping them costs little; the operation name (<c>Name</c>) is on the
/// row, the table name is not — the Azure SDK span does not carry it.
/// </summary>
public static readonly TimeSpan SlowStorageThreshold = TimeSpan.FromMilliseconds(2000);
/// <summary>Property stamped on kept slow-success rows so KQL can select them directly.</summary>
public const string SlowStoragePropertyKey = "SlowStorage";
private readonly ITelemetryProcessor _next;
public StorageDependencyFilterProcessor(ITelemetryProcessor next) => _next = next;
@@ -86,6 +99,13 @@ public sealed class StorageDependencyFilterProcessor : ITelemetryProcessor
if (isStorage)
{
// A slow success is signal: keep it, mark it, let it through.
if (dependency.Success != false && dependency.Duration >= SlowStorageThreshold)
{
dependency.Properties[SlowStoragePropertyKey] = "true";
return false;
}
// Successful storage chatter is noise; so are expected 404/412/409 outcomes.
// Every other failure (429, 5xx, auth, timeout) is kept.
return dependency.Success != false || IsExpectedStorageOutcome(dependency);
@@ -0,0 +1,39 @@
using Microsoft.ApplicationInsights;
using Microsoft.ApplicationInsights.Metrics;
namespace AutopilotMonitor.Functions.Telemetry;
/// <summary>Whether a compare-and-swap loop retried after a conflict or gave up.</summary>
public enum CasOutcome
{
Retried,
Exhausted,
}
/// <summary>
/// Low-cardinality storage metrics that survive the worker's adaptive sampling and the
/// dependency filter. <see cref="StorageDependencyFilterProcessor"/> drops the expected 412/409
/// outcomes to curb cost, which made ETag conflicts invisible; the CAS loops report them here
/// instead, pre-aggregated per minute by the SDK (<c>GetMetric</c>), so a rising conflict rate
/// on one table shows up as a trend without one row per conflict (audit 2026-09-05 F10).
/// <para>
/// Dimensions: <c>operation</c> (the CAS helper name, ~10 values), <c>table</c> (the table
/// constant) and <c>outcome</c> (<c>retried</c> / <c>exhausted</c>). Never a tenant, session
/// or user id. Null-tolerant: without a <see cref="TelemetryClient"/> every call is a no-op.
/// </para>
/// </summary>
public sealed class StorageMetrics
{
public const string CasConflictMetricName = "StorageCasConflict";
private readonly Metric? _casConflicts;
public StorageMetrics(TelemetryClient? telemetryClient = null)
{
_casConflicts = telemetryClient?.GetMetric(CasConflictMetricName, "operation", "table", "outcome");
}
/// <summary>Counts one ETag / insert conflict inside a CAS loop.</summary>
public void CasConflict(string operation, string table, CasOutcome outcome)
=> _casConflicts?.TrackValue(1, operation, table, outcome == CasOutcome.Retried ? "retried" : "exhausted");
}