diff --git a/src/ServiceControlInstaller.Engine.UnitTests/Validation/PackagedRavenDBTests.cs b/src/ServiceControlInstaller.Engine.UnitTests/Validation/PackagedRavenDBTests.cs index 01de0bd0cb..c32b2b619e 100644 --- a/src/ServiceControlInstaller.Engine.UnitTests/Validation/PackagedRavenDBTests.cs +++ b/src/ServiceControlInstaller.Engine.UnitTests/Validation/PackagedRavenDBTests.cs @@ -77,10 +77,7 @@ await store.Maintenance.SendAsync(new PutIndexesOperation(new IndexDefinition finally { await StopServer(); - if (Directory.Exists(root)) - { - FileUtils.DeleteDirectory(root, recursive: true, contentsOnly: false); - } + await TryDeleteDirectory(root, CancellationToken.None); } async Task StartServer() @@ -143,9 +140,9 @@ async Task StopServer() server.Kill(entireProcessTree: true); } - await server.WaitForExitAsync(CancellationToken.None).WaitAsync(TimeSpan.FromSeconds(15)); - TestContext.Out.WriteLine(await standardOutput.WaitAsync(TimeSpan.FromSeconds(15))); - TestContext.Out.WriteLine(await standardError.WaitAsync(TimeSpan.FromSeconds(15))); + await server.WaitForExitAsync(CancellationToken.None).WaitAsync(TimeSpan.FromSeconds(15), cancellationToken); + await TestContext.Out.WriteLineAsync(await standardOutput.WaitAsync(TimeSpan.FromSeconds(15), cancellationToken)); + await TestContext.Out.WriteLineAsync(await standardError.WaitAsync(TimeSpan.FromSeconds(15), cancellationToken)); } finally { @@ -155,6 +152,30 @@ async Task StopServer() } } + // Cleanup of the temp folder must not fail the test; Directory.Delete keeps the path stable across retries. + static async Task TryDeleteDirectory(string root, CancellationToken cancellationToken, int attempts = 5, TimeSpan? delay = null) + { + var retryDelay = delay ?? TimeSpan.FromMilliseconds(500); + for (var attempt = 1; Directory.Exists(root); attempt++) + { + try + { + Directory.Delete(root, recursive: true); + return; + } + catch (Exception ex) when (ex is IOException or UnauthorizedAccessException) + { + if (attempt == attempts) + { + await TestContext.Out.WriteLineAsync($"WARNING: Could not delete '{root}' after {attempts} attempts: {ex}"); + return; + } + + await Task.Delay(retryDelay * attempt, cancellationToken); + } + } + } + static async Task AssertDocumentAndIndex(IDocumentStore store, CancellationToken cancellationToken) { using var session = store.OpenAsyncSession(); diff --git a/tools/testing-tool/README.md b/tools/testing-tool/README.md index 6e45c6bbed..2ecacf82f7 100644 --- a/tools/testing-tool/README.md +++ b/tools/testing-tool/README.md @@ -4,7 +4,7 @@ A stateless, horizontally-scalable .NET 10 service that generates error load and scenarios against a test ServiceControl instance to validate its error-ingestion performance, with OpenTelemetry observability and a simple web UI for manual scenario control. -The tool targets the error ingestion path only — audit testing is out of scope. +The tool drives both ingestion paths: failed messages into the primary ServiceControl instance, and audited messages and saga snapshots into a ServiceControl.Audit instance. ## What it does @@ -73,6 +73,28 @@ failed message and groups it like the equivalent handler failure. Control via: - `POST /api/bypass/stop` - `GET /api/bypass/status` +### Direct audit-queue writer + +The audit-queue counterpart of the bypass writer. It writes audit envelopes straight to the ServiceControl.Audit instance's audit queue, so audit ingestion can be loaded without any endpoint ever processing a message. Control via: + +- `POST /api/audit-bypass/start` — `{ "rate": 100, "parallelism": 4, "sagaSnapshotRatio": 0.2, "durationSeconds": 60 }` +- `POST /api/audit-bypass/stop` +- `GET /api/audit-bypass/status` + +Most claims emit one processed-message envelope carrying the headers a real endpoint would have stamped: `NServiceBus.ProcessingEndpoint`, the host ids, and `TimeSent`, `ProcessingStarted` and `ProcessingEnded` spread so delivery, processing and critical times are not zero. They rotate across a pool of simulated endpoints, message types and conversations. + +A share of claims instead emit a whole **saga episode**: the three messages that drove one saga through New, Updated and Completed, each followed by the snapshot it produced. The episode is written by a single worker so the phases reach the queue in order, and the pieces are linked the way a real deployment links them: + +- each driving message carries `NServiceBus.InvokedSagas` (`SagaType:SagaId`) and `ServiceControl.SagaStateChange` (`SagaId:New|Updated|Completed`), which is the only thing that lets the audit instance attach saga information to an audited message +- each snapshot's `Initiator.InitiatingMessageId` is the id of the message directly before it, so drill-through from the saga view resolves to a real audited message +- the snapshot body is written in exact CLR casing, because ServiceControl deserializes it with a source-generated, case-sensitive context, and its `EnclosedMessageTypes` header is the bare unqualified type name, which is what the audit instance compares against to route a message to its saga branch + +`sagaSnapshotRatio` is saga snapshots per processed message; the default `0.2` emits two saga snapshots for every ten processed messages. `0` disables saga episodes. + +Audit ingestion needs a ServiceControl.Audit instance on SQL Server or PostgreSQL, so pair `--audit-instances` with `--persistence:PostgreSql` or `--persistence:SqlServer`. Under `--persistence:RavenDb` the audit instance runs on RavenDB and the EF path is not exercised. + +The testing tool's own endpoint also audits its successfully handled load messages to the same queue, which is what makes its installers create the queue before the writer first sends into it. + ### Release-test scenario presets The tool ships with presets mapped from [docs/testing-scenarios.md](/docs/testing-scenarios.md) so @@ -162,11 +184,11 @@ separator) or `--name:value` (colon separator): | Flag | Default | Values | Description | |---|---|---|---| -| `--persistence` | `RavenDb` | `RavenDb`, `SqlServer`, `PostgreSql` | Persistence backend for the ServiceControl error instance | +| `--persistence` | `RavenDb` | `RavenDb`, `SqlServer`, `PostgreSql` | Persistence backend for the ServiceControl error instance and for the audit instances | | `--transport` | `RabbitMq` | `RabbitMq`, `SqlServer` | Transport shared by ServiceControl and the testing tool. `SqlServer` reuses the persistence SQL Server container with a separate `Transport` database | | `--tag` | *(none — uses the current build's image)* | any image tag, e.g. `pr-1234` or `6.3.1` | Override the ServiceControl container image tag (useful for testing PR-based prereleases) | | `--error-ingestion-scale-unit` | `0` | non-negative integer | Number of additional error-ingestion-only scale-out instances to spin up alongside the primary error instance (each runs with `--error-ingestion-only`) | -| `--audit-instances` | `0` | non-negative integer | Number of ServiceControl audit instances to add | +| `--audit-instances` | `1` | non-negative integer | Number of ServiceControl.Audit instances to add. They take the same `--persistence` backend, report custom checks to the primary and export OTel to the collector | The Aspire dashboard provides allocated ports for each service. The testing tool automatically connects to ServiceControl via the platform's transport and REST API URL, and sends its OTLP @@ -242,7 +264,8 @@ All configuration is via environment variables (no files, no database). Settings | `TestingTool__CustomCheckInterval` | `00:00:30` | Default interval for the custom-check-failures job | | `TestingTool__CustomCheckHost` | `ServiceControl` | `Host` field on injected custom-check reports | | `TestingTool__CustomCheckFailureProbability` | `0.4` | Probability (0–1) a given check is reported failed each cycle | -| `TestingTool__AuditQueueName` | `audit` | NServiceBus audit queue for processed messages | +| `TestingTool__AuditQueueName` | `audit` | NServiceBus audit queue the ServiceControl.Audit instance drains | +| `TestingTool__AuditSagaSnapshotRatio` | `0.2` | Default saga snapshots per processed message for the audit-queue writer | | `TestingTool__MonitoringQueueName` | `Particular.Monitoring` | ServiceControl monitoring instance queue that endpoint metrics are sent to | | `TestingTool__ErrorQueueName` | `error` | NServiceBus error queue (ServiceControl monitors this) | | `TestingTool__AutoStartBackgroundNoise` | `false` | Auto-start the background-noise scenario on startup | diff --git a/tools/testing-tool/TestingTool.AppHost/AppHost.cs b/tools/testing-tool/TestingTool.AppHost/AppHost.cs index b3e3fa168b..7c94c661b4 100644 --- a/tools/testing-tool/TestingTool.AppHost/AppHost.cs +++ b/tools/testing-tool/TestingTool.AppHost/AppHost.cs @@ -3,6 +3,7 @@ var options = CliOptions.Parse(args); var persistenceType = options.GetValue("persistence", PersistenceType.RavenDb); +var imageTag = options.GetValueOrDefault("tag"); Console.WriteLine($"Using persistence type: {persistenceType}"); var builder = DistributedApplication.CreateBuilder(args); @@ -51,10 +52,25 @@ platform.AddServicePulse("pulse", primaryErrorInstance, platform.AddServiceControlMonitoringInstance("monitoring")); -for (int i = 0; i < options.GetValue("audit-instances", 0); i++) +// EF persistence lives in the audit instance itself, so audit load needs a real audit container. +// The platform persistence argument stays RavenDB because the Aspire package only knows about +// platform persistences; the SQL wiring is layered on with WithPersistenceType. +var auditInstanceCount = options.GetValue("audit-instances", 0); + +for (int i = 0; i < auditInstanceCount; i++) { + //note: these audit instances are currently independant and are not scale out nodes like the error ingestion ones. platform.AddServiceControlAuditInstance("audit" + i, primaryErrorInstance, raven) - .WithEnvironment("INSTANCENAME", "Audit-" + i); + .WithEnvironment("SERVICECONTROL_AUDIT_INSTANCENAME", "Audit-" + i) + .WithEnvironment("OTEL_EXPORTER_OTLP_ENDPOINT", observability.Collector.GetEndpoint("otlp-grpc")) + // Reports the audit ingestion and failed-import custom checks to the primary, which is the + // only instance ServicePulse asks. Without it a failed audit import is invisible outside + // the audit container's own log. + .WithEnvironment("SERVICECONTROL_AUDIT_SERVICECONTROLQUEUEADDRESS", "Particular.ServiceControl") + .WithAuditQueueName(i == 0 ? "audit" : "audit" + i) + .WithRunMode(PlatformRunMode.SetupAndRun) + .WithIngestionTuning() + .WithPersistenceType(persistenceType); } // --- Testing tool --- @@ -69,6 +85,6 @@ .WaitFor(observability.Collector); // --- Optional: override ServiceControl image tag for prerelease testing --- -builder.UseServiceControlImageTag(options.GetValueOrDefault("tag")); +builder.UseServiceControlImageTag(imageTag); builder.Build().Run(); \ No newline at end of file diff --git a/tools/testing-tool/TestingTool.AppHost/HostBuilderExtensions.cs b/tools/testing-tool/TestingTool.AppHost/HostBuilderExtensions.cs index 6165c3ad3e..740db8ce37 100644 --- a/tools/testing-tool/TestingTool.AppHost/HostBuilderExtensions.cs +++ b/tools/testing-tool/TestingTool.AppHost/HostBuilderExtensions.cs @@ -70,6 +70,52 @@ public static IResourceBuilder WithPersiste + /// + /// Points a ServiceControl.Audit instance at the same SQL Server or PostgreSQL container the + /// primary uses. The keys are namespaced to the audit instance, and there is no body storage: + /// the audit EF persisters store bodies inline in the audit row, capped by MaxBodySizeToStore. + /// + /// + /// The setting root namespace is "ServiceControl.Audit", and the settings reader rewrites both + /// the dot and the slash to an underscore, so ServiceControl.Audit/Database/ConnectionString is + /// SERVICECONTROL_AUDIT_DATABASE_CONNECTIONSTRING. The unprefixed names the audit container's + /// Dockerfile bakes in (PersistenceType, AuditRetentionPeriod) are later in the lookup order, + /// so the namespaced form wins. + /// + public static IResourceBuilder WithPersistenceType( + this IResourceBuilder audit, PersistenceType type) + { + if (type == PersistenceType.RavenDb) + { + return audit; + } + + var db = type switch + { + PersistenceType.SqlServer => audit.ApplicationBuilder.AddSqlServerPersistence("ServiceControl"), + PersistenceType.PostgreSql => audit.ApplicationBuilder.AddPostgresPersistence("servicecontrol"), + _ => throw new ArgumentOutOfRangeException(nameof(type), type, null) + }; + + return audit + .WaitFor(db) + .WithEnvironment("SERVICECONTROL_AUDIT_PERSISTENCETYPE", PersistenceTypeName(type)) + .WithEnvironment("SERVICECONTROL_AUDIT_DATABASE_CONNECTIONSTRING", db); + } + + /// + /// The audit equivalent of the ingestion tuning the primary carries. The setting names are + /// AuditIngestion*, and they are read under the audit namespace, so the primary's + /// SERVICECONTROL_ERRORINGESTION* values have no effect here. + /// + public static IResourceBuilder WithIngestionTuning( + this IResourceBuilder audit) => + audit + .WithEnvironment("SERVICECONTROL_AUDIT_MAXIMUMCONCURRENCYLEVEL", "100") + .WithEnvironment("SERVICECONTROL_AUDIT_AUDITINGESTIONBATCHSIZE", "25") + .WithEnvironment("SERVICECONTROL_AUDIT_AUDITINGESTIONMAXPARALLELWRITERS", "4") + .WithEnvironment("SERVICECONTROL_AUDIT_AUDITINGESTIONBATCHTIMEOUT", "00:00:00.100"); + public static IResourceBuilder AddTransport( this IResourceBuilder platform, TransportType type) { diff --git a/tools/testing-tool/TestingTool.AppHost/TestingTool.AppHost.csproj b/tools/testing-tool/TestingTool.AppHost/TestingTool.AppHost.csproj index b40c89c202..36bc67fa14 100644 --- a/tools/testing-tool/TestingTool.AppHost/TestingTool.AppHost.csproj +++ b/tools/testing-tool/TestingTool.AppHost/TestingTool.AppHost.csproj @@ -4,6 +4,7 @@ enable enable Exe + true @@ -18,4 +19,8 @@ + + + + \ No newline at end of file diff --git a/tools/testing-tool/TestingTool.AppHost/appsettings.json b/tools/testing-tool/TestingTool.AppHost/appsettings.json new file mode 100644 index 0000000000..19779335f2 --- /dev/null +++ b/tools/testing-tool/TestingTool.AppHost/appsettings.json @@ -0,0 +1,11 @@ +{ + "Logging": { + "LogLevel": { + "Default": "Information", + "Aspire": "Information", + "Aspire.Hosting.Dcp": "Warning", + "Microsoft": "Warning", + "Particular": "Information" + } + } +} diff --git a/tools/testing-tool/TestingTool.AppHost/obs/grafana/dashboards/testing-tool.json b/tools/testing-tool/TestingTool.AppHost/obs/grafana/dashboards/testing-tool.json index 8336964639..e7bf4ac347 100644 --- a/tools/testing-tool/TestingTool.AppHost/obs/grafana/dashboards/testing-tool.json +++ b/tools/testing-tool/TestingTool.AppHost/obs/grafana/dashboards/testing-tool.json @@ -1,34 +1,50 @@ { - "title": "Testing Tool — Error Load & Observability", + "title": "Testing Tool: Error and Audit Load & Observability", "uid": "testing-tool", "schemaVersion": 39, - "version": 6, + "version": 8, "timezone": "browser", "refresh": "5s", - "tags": ["testing-tool", "servicecontrol", "otel"], + "tags": [ + "testing-tool", + "servicecontrol", + "otel" + ], "templating": { "list": [ { "name": "scenario", "type": "query", - "datasource": { "type": "prometheus", "uid": "prometheus" }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, "query": "label_values(errors_sent_total, scenario)", "refresh": 2, "includeAll": true, "multi": true, - "current": { "text": "All", "value": "$__all" } + "current": { + "text": "All", + "value": "$__all" + } }, { "name": "sc_instance", "label": "ServiceControl instance", "type": "query", - "datasource": { "type": "prometheus", "uid": "prometheus" }, - "query": "label_values(sc_error_ingestion_message_duration_seconds_count, exported_job)", + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, + "query": "label_values({__name__=~\"sc_(error|audit)_ingestion_message_duration_seconds_count\"}, exported_job)", "refresh": 2, "includeAll": true, "allValue": ".*", "multi": true, - "current": { "text": "All", "value": "$__all" } + "current": { + "text": "All", + "value": "$__all" + } } ] }, @@ -38,20 +54,37 @@ "type": "row", "title": "Load generation (testing tool)", "collapsed": false, - "gridPos": { "h": 1, "w": 24, "x": 0, "y": 0 }, + "gridPos": { + "h": 1, + "w": 24, + "x": 0, + "y": 0 + }, "panels": [] }, { "id": 1, - "title": "Errors Raised / sec — handler path (by scenario)", - "description": "Errors sent to the ServiceControl error queue via the NServiceBus handler path (errors_sent_total), broken down by scenario. The bypass path is emitted separately — see the 'Bypass Errors Written / sec' panel — and combined with this in the 'Errors Raised vs Ingested' panels.", + "title": "Errors Raised / sec \u2014 handler path (by scenario)", + "description": "Errors sent to the ServiceControl error queue via the NServiceBus handler path (errors_sent_total), broken down by scenario. The bypass path is emitted separately \u2014 see the 'Bypass Errors Written / sec' panel \u2014 and combined with this in the 'Errors Raised vs Ingested' panels.", "type": "timeseries", - "gridPos": { "h": 8, "w": 12, "x": 0, "y": 1 }, - "datasource": { "type": "prometheus", "uid": "prometheus" }, + "gridPos": { + "h": 8, + "w": 8, + "x": 0, + "y": 1 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, "fieldConfig": { "defaults": { "unit": "reqps", - "custom": { "drawStyle": "line", "lineInterpolation": "smooth", "fillOpacity": 10 } + "custom": { + "drawStyle": "line", + "lineInterpolation": "smooth", + "fillOpacity": 10 + } } }, "targets": [ @@ -66,12 +99,24 @@ "title": "Bypass Errors Written / sec (by scenario)", "description": "Errors written directly to the ServiceControl error queue by the bypass path (bypass_errors_written_total), bypassing the handler. Emitted separately from errors_sent_total so handler and bypass load can be told apart; the 'Errors Raised vs Ingested' panels combine the two into the total raised.", "type": "timeseries", - "gridPos": { "h": 8, "w": 12, "x": 12, "y": 1 }, - "datasource": { "type": "prometheus", "uid": "prometheus" }, + "gridPos": { + "h": 8, + "w": 8, + "x": 8, + "y": 1 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, "fieldConfig": { "defaults": { "unit": "reqps", - "custom": { "drawStyle": "line", "lineInterpolation": "smooth", "fillOpacity": 10 } + "custom": { + "drawStyle": "line", + "lineInterpolation": "smooth", + "fillOpacity": 10 + } } }, "targets": [ @@ -81,16 +126,61 @@ } ] }, + { + "id": 41, + "title": "Audit Envelopes Written / sec (by kind)", + "description": "Envelopes written directly to the audit queue by the direct audit-queue writer (bypass_audit_messages_written_total), split into processed-message envelopes and saga-update snapshots. Only a ServiceControl.Audit instance drains that queue, so this is zero unless --audit-instances is running.", + "type": "timeseries", + "gridPos": { + "h": 8, + "w": 8, + "x": 16, + "y": 1 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, + "fieldConfig": { + "defaults": { + "unit": "reqps", + "custom": { + "drawStyle": "line", + "lineInterpolation": "smooth", + "fillOpacity": 10 + } + } + }, + "targets": [ + { + "refId": "A", + "expr": "sum by (kind) (rate(bypass_audit_messages_written_total[$__rate_interval]))", + "legendFormat": "audit: {{kind}}" + } + ] + }, { "id": 3, "title": "Errors Replayed / sec", "type": "timeseries", - "gridPos": { "h": 8, "w": 8, "x": 0, "y": 9 }, - "datasource": { "type": "prometheus", "uid": "prometheus" }, + "gridPos": { + "h": 8, + "w": 8, + "x": 0, + "y": 9 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, "fieldConfig": { "defaults": { "unit": "reqps", - "custom": { "drawStyle": "line", "lineInterpolation": "smooth", "fillOpacity": 10 } + "custom": { + "drawStyle": "line", + "lineInterpolation": "smooth", + "fillOpacity": 10 + } } }, "targets": [ @@ -104,12 +194,24 @@ "id": 4, "title": "Errors Archived / sec", "type": "timeseries", - "gridPos": { "h": 8, "w": 8, "x": 8, "y": 9 }, - "datasource": { "type": "prometheus", "uid": "prometheus" }, + "gridPos": { + "h": 8, + "w": 8, + "x": 8, + "y": 9 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, "fieldConfig": { "defaults": { "unit": "reqps", - "custom": { "drawStyle": "line", "lineInterpolation": "smooth", "fillOpacity": 10 } + "custom": { + "drawStyle": "line", + "lineInterpolation": "smooth", + "fillOpacity": 10 + } } }, "targets": [ @@ -123,12 +225,24 @@ "id": 6, "title": "Searches Executed / sec", "type": "timeseries", - "gridPos": { "h": 8, "w": 8, "x": 16, "y": 9 }, - "datasource": { "type": "prometheus", "uid": "prometheus" }, + "gridPos": { + "h": 8, + "w": 8, + "x": 16, + "y": 9 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, "fieldConfig": { "defaults": { "unit": "ops", - "custom": { "drawStyle": "line", "lineInterpolation": "smooth", "fillOpacity": 10 } + "custom": { + "drawStyle": "line", + "lineInterpolation": "smooth", + "fillOpacity": 10 + } } }, "targets": [ @@ -142,12 +256,24 @@ "id": 2, "title": "Search Latency p95 (ms)", "type": "timeseries", - "gridPos": { "h": 8, "w": 12, "x": 0, "y": 17 }, - "datasource": { "type": "prometheus", "uid": "prometheus" }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 17 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, "fieldConfig": { "defaults": { "unit": "ms", - "custom": { "drawStyle": "line", "lineInterpolation": "smooth", "fillOpacity": 10 } + "custom": { + "drawStyle": "line", + "lineInterpolation": "smooth", + "fillOpacity": 10 + } } }, "targets": [ @@ -162,21 +288,41 @@ "title": "Total Errors Raised (cumulative)", "description": "Cumulative errors raised by the testing tool, combining the handler path (errors_sent_total) and the bypass path (bypass_errors_written_total). Compare against 'Errors Ingested' in the 'Errors Raised vs Ingested' panels to gauge ingestion backlog.", "type": "stat", - "gridPos": { "h": 8, "w": 12, "x": 12, "y": 17 }, - "datasource": { "type": "prometheus", "uid": "prometheus" }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 17 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, "fieldConfig": { "defaults": { "unit": "short", - "color": { "mode": "thresholds" }, - "thresholds": { "steps": [ { "color": "green" } ] } + "color": { + "mode": "thresholds" + }, + "thresholds": { + "steps": [ + { + "color": "green" + } + ] + } } }, "options": { - "reduceOptions": { "calcs": ["lastNotNull"] } + "reduceOptions": { + "calcs": [ + "lastNotNull" + ] + } }, "targets": [ { - "expr": "sum(errors_sent_total) + sum(bypass_errors_written_total)", + "expr": "(sum(errors_sent_total) or vector(0)) + (sum(bypass_errors_written_total) or vector(0))", "legendFormat": "raised" } ] @@ -184,69 +330,135 @@ { "id": 101, "type": "row", - "title": "Raised vs ingested", + "title": "Raised vs ingested (error and audit)", "collapsed": false, - "gridPos": { "h": 1, "w": 24, "x": 0, "y": 25 }, + "gridPos": { + "h": 1, + "w": 24, + "x": 0, + "y": 25 + }, "panels": [] }, { "id": 9, - "title": "Errors Raised vs Ingested / sec", - "description": "Compares the rate of errors the testing tool raises — combining the handler path (errors_sent_total) and the bypass path (bypass_errors_written_total) — against the rate ServiceControl successfully ingests (sc.error.ingestion.message_duration_seconds, message_category=failed-message, result=success). The raised series should lead the ingested series; a widening gap indicates an ingestion backlog or failures.", + "title": "Raised vs Ingested / sec (error and audit)", + "description": "Compares the rate of errors the testing tool raises \u2014 combining the handler path (errors_sent_total) and the bypass path (bypass_errors_written_total) \u2014 against the rate ServiceControl successfully ingests (sc.error.ingestion.message_duration_seconds, message_category=failed-message, result=success). The raised series should lead the ingested series; a widening gap indicates an ingestion backlog or failures. The audit pair covers both audit paths: envelopes the direct audit-queue writer sends, and the messages this tool's own endpoint handles successfully, which AuditProcessedMessagesTo forwards to the same queue.", "type": "timeseries", - "gridPos": { "h": 8, "w": 12, "x": 0, "y": 26 }, - "datasource": { "type": "prometheus", "uid": "prometheus" }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 26 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, "fieldConfig": { "defaults": { "unit": "reqps", - "custom": { "drawStyle": "line", "lineInterpolation": "smooth", "fillOpacity": 10 } + "custom": { + "drawStyle": "line", + "lineInterpolation": "smooth", + "fillOpacity": 10 + } } }, "targets": [ { - "expr": "avg_over_time((sum(rate(errors_sent_total[1m])) + sum(rate(bypass_errors_written_total[1m])))[2m:15s])", - "legendFormat": "raised (testing-tool)" + "refId": "A", + "expr": "avg_over_time(((sum(rate(errors_sent_total[1m])) or vector(0)) + (sum(rate(bypass_errors_written_total[1m])) or vector(0)))[2m:15s])", + "legendFormat": "error raised (testing-tool)" }, { + "refId": "B", "expr": "avg_over_time((sum(rate(sc_error_ingestion_message_duration_seconds_count{exported_job=~\"$sc_instance\",message_category=\"failed-message\",result=\"success\"}[1m])))[2m:15s])", - "legendFormat": "ingested (servicecontrol)" + "legendFormat": "error ingested (servicecontrol)" + }, + { + "refId": "C", + "expr": "avg_over_time(((sum(rate(bypass_audit_messages_written_total[1m])) or vector(0)) + (sum(rate(endpoint_audited_messages_total[1m])) or vector(0)))[2m:15s])", + "legendFormat": "audit raised (testing-tool)" + }, + { + "refId": "D", + "expr": "avg_over_time((sum(rate(sc_audit_ingestion_message_duration_seconds_count{exported_job=~\"$sc_instance\",result=\"success\"}[1m])))[2m:15s])", + "legendFormat": "audit ingested (servicecontrol.audit)" } ] }, { "id": 10, - "title": "Errors Raised vs Ingested (cumulative)", + "title": "Raised vs Ingested, cumulative (error and audit)", "description": "Cumulative comparison: total errors raised by the testing tool (handler + bypass) vs total failed messages ServiceControl successfully ingested. A persistent delta is the ingestion backlog (still queued or failed).", "type": "stat", - "gridPos": { "h": 8, "w": 12, "x": 12, "y": 26 }, - "datasource": { "type": "prometheus", "uid": "prometheus" }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 26 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, "fieldConfig": { "defaults": { "unit": "short", - "color": { "mode": "thresholds" }, - "thresholds": { "steps": [ { "color": "green" } ] } + "color": { + "mode": "thresholds" + }, + "thresholds": { + "steps": [ + { + "color": "green" + } + ] + } } }, "options": { - "reduceOptions": { "calcs": ["lastNotNull"] } + "reduceOptions": { + "calcs": [ + "lastNotNull" + ] + } }, "targets": [ { - "expr": "sum(errors_sent_total) + sum(bypass_errors_written_total)", - "legendFormat": "raised" + "refId": "A", + "expr": "(sum(errors_sent_total) or vector(0)) + (sum(bypass_errors_written_total) or vector(0))", + "legendFormat": "error raised" }, { + "refId": "B", "expr": "sum(sc_error_ingestion_message_duration_seconds_count{exported_job=~\"$sc_instance\",message_category=\"failed-message\",result=\"success\"})", - "legendFormat": "ingested" + "legendFormat": "error ingested" + }, + { + "refId": "C", + "expr": "(sum(bypass_audit_messages_written_total) or vector(0)) + (sum(endpoint_audited_messages_total) or vector(0))", + "legendFormat": "audit raised" + }, + { + "refId": "D", + "expr": "sum(sc_audit_ingestion_message_duration_seconds_count{exported_job=~\"$sc_instance\",result=\"success\"})", + "legendFormat": "audit ingested" } ] }, { "id": 102, "type": "row", - "title": "ServiceControl ingestion — rate & throughput", + "title": "ServiceControl ingestion: rate and throughput (error and audit)", "collapsed": false, - "gridPos": { "h": 1, "w": 24, "x": 0, "y": 34 }, + "gridPos": { + "h": 1, + "w": 24, + "x": 0, + "y": 34 + }, "panels": [] }, { @@ -254,25 +466,51 @@ "title": "Ingestion Rate (msg/sec)", "description": "Successful ingestion rate across both message categories, from the count of sc.error.ingestion.message_duration_seconds with result=success. Averaged over a 2-minute window (15s step) to smooth out the stepwise nature of the batched counter, so this is the sustained rate rather than an instantaneous one.", "type": "stat", - "gridPos": { "h": 6, "w": 6, "x": 0, "y": 35 }, - "datasource": { "type": "prometheus", "uid": "prometheus" }, + "gridPos": { + "h": 6, + "w": 6, + "x": 0, + "y": 35 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, "fieldConfig": { "defaults": { "unit": "reqps", "decimals": 1, - "color": { "mode": "thresholds" }, - "thresholds": { "steps": [ { "color": "green" } ] } + "color": { + "mode": "thresholds" + }, + "thresholds": { + "steps": [ + { + "color": "green" + } + ] + } } }, "options": { - "reduceOptions": { "calcs": ["lastNotNull"] }, + "reduceOptions": { + "calcs": [ + "lastNotNull" + ] + }, "graphMode": "area", "colorMode": "value" }, "targets": [ { "expr": "avg_over_time((sum(rate(sc_error_ingestion_message_duration_seconds_count{exported_job=~\"$sc_instance\",result=\"success\"}[1m])))[2m:15s])", - "legendFormat": "ingested" + "legendFormat": "error: ingested", + "refId": "A" + }, + { + "expr": "avg_over_time((sum(rate(sc_audit_ingestion_message_duration_seconds_count{exported_job=~\"$sc_instance\",result=\"success\"}[1m])))[2m:15s])", + "legendFormat": "audit: ingested", + "refId": "B" } ] }, @@ -281,25 +519,51 @@ "title": "Peak Ingestion Rate (selected range)", "description": "Highest sustained ingestion rate seen anywhere in the dashboard's time range. The underlying per-message rate is averaged over a 2-minute window (15s step) before the max is taken, so the value reflects what the instance actually sustained rather than a single batch-flush spike.", "type": "stat", - "gridPos": { "h": 6, "w": 6, "x": 6, "y": 35 }, - "datasource": { "type": "prometheus", "uid": "prometheus" }, + "gridPos": { + "h": 6, + "w": 6, + "x": 6, + "y": 35 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, "fieldConfig": { "defaults": { "unit": "reqps", "decimals": 1, - "color": { "mode": "thresholds" }, - "thresholds": { "steps": [ { "color": "blue" } ] } + "color": { + "mode": "thresholds" + }, + "thresholds": { + "steps": [ + { + "color": "blue" + } + ] + } } }, "options": { - "reduceOptions": { "calcs": ["lastNotNull"] }, + "reduceOptions": { + "calcs": [ + "lastNotNull" + ] + }, "graphMode": "none", "colorMode": "value" }, "targets": [ { "expr": "max_over_time(avg_over_time((sum(rate(sc_error_ingestion_message_duration_seconds_count{exported_job=~\"$sc_instance\",result=\"success\"}[1m])))[2m:15s])[$__range:15s])", - "legendFormat": "peak" + "legendFormat": "error: peak", + "refId": "A" + }, + { + "expr": "max_over_time(avg_over_time((sum(rate(sc_audit_ingestion_message_duration_seconds_count{exported_job=~\"$sc_instance\",result=\"success\"}[1m])))[2m:15s])[$__range:15s])", + "legendFormat": "audit: peak", + "refId": "B" } ] }, @@ -308,56 +572,111 @@ "title": "Total Messages Ingested", "description": "Cumulative count of messages ServiceControl ingested successfully, both failed messages and retry confirmations.", "type": "stat", - "gridPos": { "h": 6, "w": 6, "x": 12, "y": 35 }, - "datasource": { "type": "prometheus", "uid": "prometheus" }, + "gridPos": { + "h": 6, + "w": 6, + "x": 12, + "y": 35 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, "fieldConfig": { "defaults": { "unit": "short", - "color": { "mode": "thresholds" }, - "thresholds": { "steps": [ { "color": "green" } ] } + "color": { + "mode": "thresholds" + }, + "thresholds": { + "steps": [ + { + "color": "green" + } + ] + } } }, "options": { - "reduceOptions": { "calcs": ["lastNotNull"] }, + "reduceOptions": { + "calcs": [ + "lastNotNull" + ] + }, "graphMode": "area" }, "targets": [ { "expr": "sum(sc_error_ingestion_message_duration_seconds_count{exported_job=~\"$sc_instance\",result=\"success\"})", - "legendFormat": "ingested" + "legendFormat": "error: ingested", + "refId": "A" + }, + { + "expr": "sum(sc_audit_ingestion_message_duration_seconds_count{exported_job=~\"$sc_instance\",result=\"success\"})", + "legendFormat": "audit: ingested", + "refId": "B" } ] }, { "id": 23, "title": "Consecutive Batch Failures", - "description": "sc.error.ingestion.consecutive_batch_failures_total — the ingestor's own circuit-breaker signal. It resets to 0 on the first batch that writes, so anything above 0 means storage is currently rejecting batches.", + "description": "sc.error.ingestion.consecutive_batch_failures_total \u2014 the ingestor's own circuit-breaker signal. It resets to 0 on the first batch that writes, so anything above 0 means storage is currently rejecting batches.", "type": "stat", - "gridPos": { "h": 6, "w": 6, "x": 18, "y": 35 }, - "datasource": { "type": "prometheus", "uid": "prometheus" }, + "gridPos": { + "h": 6, + "w": 6, + "x": 18, + "y": 35 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, "fieldConfig": { "defaults": { "unit": "short", - "color": { "mode": "thresholds" }, + "color": { + "mode": "thresholds" + }, "thresholds": { "mode": "absolute", "steps": [ - { "color": "green", "value": null }, - { "color": "orange", "value": 1 }, - { "color": "red", "value": 3 } + { + "color": "green", + "value": null + }, + { + "color": "orange", + "value": 1 + }, + { + "color": "red", + "value": 3 + } ] } } }, "options": { - "reduceOptions": { "calcs": ["lastNotNull"] }, + "reduceOptions": { + "calcs": [ + "lastNotNull" + ] + }, "graphMode": "area", "colorMode": "background" }, "targets": [ { "expr": "max(sc_error_ingestion_consecutive_batch_failures_total{exported_job=~\"$sc_instance\"})", - "legendFormat": "consecutive failures" + "legendFormat": "error: consecutive failures", + "refId": "A" + }, + { + "expr": "max(sc_audit_ingestion_consecutive_batch_failures_total{exported_job=~\"$sc_instance\"})", + "legendFormat": "audit: consecutive failures", + "refId": "B" } ] }, @@ -366,8 +685,16 @@ "title": "Ingestion Rate / sec (by result)", "description": "Every message the ingestor completed, split by outcome: success, skipped (nothing to store) and failed. Stacked, the total is the rate messages left the ingestion pipeline at, regardless of outcome.", "type": "timeseries", - "gridPos": { "h": 8, "w": 12, "x": 0, "y": 41 }, - "datasource": { "type": "prometheus", "uid": "prometheus" }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 41 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, "fieldConfig": { "defaults": { "unit": "reqps", @@ -375,40 +702,81 @@ "drawStyle": "line", "lineInterpolation": "smooth", "fillOpacity": 25, - "stacking": { "mode": "normal", "group": "A" } + "stacking": { + "mode": "normal", + "group": "A" + } } } }, "options": { - "legend": { "displayMode": "table", "placement": "bottom", "calcs": ["mean", "max"] } + "legend": { + "displayMode": "table", + "placement": "bottom", + "calcs": [ + "mean", + "max" + ] + } }, "targets": [ { "expr": "avg_over_time((sum by (result) (rate(sc_error_ingestion_message_duration_seconds_count{exported_job=~\"$sc_instance\"}[1m])))[2m:15s])", - "legendFormat": "{{result}}" + "legendFormat": "error: {{result}}", + "refId": "A" + }, + { + "expr": "avg_over_time((sum by (result) (rate(sc_audit_ingestion_message_duration_seconds_count{exported_job=~\"$sc_instance\"}[1m])))[2m:15s])", + "legendFormat": "audit: {{result}}", + "refId": "B" } ] }, { "id": 25, "title": "Ingestion Rate / sec (by message category and instance)", - "description": "Successful ingestion rate split by message.category — failed-message vs retry-confirmation — and by instance. Retry confirmations are much cheaper than failed messages, so a rate that climbs while latency stays flat is usually a shift in this mix, not extra capacity.", + "description": "Successful ingestion rate split by message.category \u2014 failed-message vs retry-confirmation \u2014 and by instance. Retry confirmations are much cheaper than failed messages, so a rate that climbs while latency stays flat is usually a shift in this mix, not extra capacity.", "type": "timeseries", - "gridPos": { "h": 8, "w": 12, "x": 12, "y": 41 }, - "datasource": { "type": "prometheus", "uid": "prometheus" }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 41 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, "fieldConfig": { "defaults": { "unit": "reqps", - "custom": { "drawStyle": "line", "lineInterpolation": "smooth", "fillOpacity": 10 } + "custom": { + "drawStyle": "line", + "lineInterpolation": "smooth", + "fillOpacity": 10 + } } }, "options": { - "legend": { "displayMode": "table", "placement": "bottom", "calcs": ["mean", "max"] } + "legend": { + "displayMode": "table", + "placement": "bottom", + "calcs": [ + "mean", + "max" + ] + } }, "targets": [ { "expr": "avg_over_time((sum by (exported_job, message_category) (rate(sc_error_ingestion_message_duration_seconds_count{exported_job=~\"$sc_instance\",result=\"success\"}[1m])))[2m:15s])", - "legendFormat": "{{message_category}} - {{exported_job}}" + "legendFormat": "error: {{message_category}} - {{exported_job}}", + "refId": "A" + }, + { + "expr": "avg_over_time((sum by (exported_job, message_category) (rate(sc_audit_ingestion_message_duration_seconds_count{exported_job=~\"$sc_instance\",result=\"success\"}[1m])))[2m:15s])", + "legendFormat": "audit: {{message_category}} - {{exported_job}}", + "refId": "B" } ] }, @@ -417,21 +785,46 @@ "title": "Batches Written / sec (by result)", "description": "Count of sc.error.ingestion.batch_duration_seconds by result. 'full' means the batch hit the maximum size, so the ingestor was never waiting for messages: a sustained run of full batches means the load generator, not ServiceControl, is setting the pace.", "type": "timeseries", - "gridPos": { "h": 8, "w": 12, "x": 0, "y": 49 }, - "datasource": { "type": "prometheus", "uid": "prometheus" }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 49 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, "fieldConfig": { "defaults": { "unit": "reqps", - "custom": { "drawStyle": "line", "lineInterpolation": "smooth", "fillOpacity": 10 } + "custom": { + "drawStyle": "line", + "lineInterpolation": "smooth", + "fillOpacity": 10 + } } }, "options": { - "legend": { "displayMode": "table", "placement": "bottom", "calcs": ["mean", "max"] } + "legend": { + "displayMode": "table", + "placement": "bottom", + "calcs": [ + "mean", + "max" + ] + } }, "targets": [ { "expr": "sum by (result) (rate(sc_error_ingestion_batch_duration_seconds_count{exported_job=~\"$sc_instance\"}[$__rate_interval]))", - "legendFormat": "{{result}}" + "legendFormat": "error: {{result}}", + "refId": "A" + }, + { + "expr": "sum by (result) (rate(sc_audit_ingestion_batch_duration_seconds_count{exported_job=~\"$sc_instance\"}[$__rate_interval]))", + "legendFormat": "audit: {{result}}", + "refId": "B" } ] }, @@ -440,72 +833,145 @@ "title": "Effective Batch Size (messages per batch)", "description": "Message rate divided by batch rate: how many messages each batch actually carried. Sitting at the configured maximum means batching is saturated; a low number under high load points at the ingestor being starved rather than storage being slow.", "type": "timeseries", - "gridPos": { "h": 8, "w": 12, "x": 12, "y": 49 }, - "datasource": { "type": "prometheus", "uid": "prometheus" }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 49 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, "fieldConfig": { "defaults": { "unit": "short", "decimals": 1, - "custom": { "drawStyle": "line", "lineInterpolation": "smooth", "fillOpacity": 10 } + "custom": { + "drawStyle": "line", + "lineInterpolation": "smooth", + "fillOpacity": 10 + } } }, "targets": [ { "expr": "avg_over_time((sum(rate(sc_error_ingestion_message_duration_seconds_count{exported_job=~\"$sc_instance\"}[1m])))[2m:15s]) / avg_over_time((sum(rate(sc_error_ingestion_batch_duration_seconds_count{exported_job=~\"$sc_instance\"}[1m])))[2m:15s])", - "legendFormat": "messages / batch" + "legendFormat": "error: messages / batch", + "refId": "A" + }, + { + "expr": "avg_over_time((sum(rate(sc_audit_ingestion_message_duration_seconds_count{exported_job=~\"$sc_instance\"}[1m])))[2m:15s]) / avg_over_time((sum(rate(sc_audit_ingestion_batch_duration_seconds_count{exported_job=~\"$sc_instance\"}[1m])))[2m:15s])", + "legendFormat": "audit: messages / batch", + "refId": "B" } ] }, { "id": 103, "type": "row", - "title": "ServiceControl ingestion — latency", + "title": "ServiceControl ingestion: latency (error and audit)", "collapsed": false, - "gridPos": { "h": 1, "w": 24, "x": 0, "y": 57 }, + "gridPos": { + "h": 1, + "w": 24, + "x": 0, + "y": 57 + }, "panels": [] }, { "id": 11, "title": "ServiceControl Ingestion Duration (s)", - "description": "ServiceControl OTel histogram sc.error.ingestion.message_duration_seconds for failed messages — p50, p95 and p99 end-to-end duration per ingested error message. Spikes here explain why 'ingested' lags 'raised'. Buckets top out at 5s, so a p99 pinned at 5 means 'at least 5s'.", + "description": "ServiceControl OTel histogram sc.error.ingestion.message_duration_seconds for failed messages \u2014 p50, p95 and p99 end-to-end duration per ingested error message. Spikes here explain why 'ingested' lags 'raised'. Buckets top out at 5s, so a p99 pinned at 5 means 'at least 5s'.", "type": "timeseries", - "gridPos": { "h": 8, "w": 12, "x": 0, "y": 58 }, - "datasource": { "type": "prometheus", "uid": "prometheus" }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 58 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, "fieldConfig": { "defaults": { "unit": "s", - "custom": { "drawStyle": "line", "lineInterpolation": "smooth", "fillOpacity": 10 } + "custom": { + "drawStyle": "line", + "lineInterpolation": "smooth", + "fillOpacity": 10 + } } }, "targets": [ { "expr": "histogram_quantile(0.5, sum by (le, exported_job) (rate(sc_error_ingestion_message_duration_seconds_bucket{exported_job=~\"$sc_instance\",message_category=\"failed-message\"}[$__rate_interval])))", - "legendFormat": "Message P50 - {{exported_job}}" + "legendFormat": "error: Message P50 - {{exported_job}}", + "refId": "A" }, { "expr": "histogram_quantile(0.95, sum by (le, exported_job) (rate(sc_error_ingestion_message_duration_seconds_bucket{exported_job=~\"$sc_instance\",message_category=\"failed-message\"}[$__rate_interval])))", - "legendFormat": "Message P95 - {{exported_job}}" + "legendFormat": "error: Message P95 - {{exported_job}}", + "refId": "B" }, { "expr": "histogram_quantile(0.99, sum by (le, exported_job) (rate(sc_error_ingestion_message_duration_seconds_bucket{exported_job=~\"$sc_instance\",message_category=\"failed-message\"}[$__rate_interval])))", - "legendFormat": "Message P99 - {{exported_job}}" + "legendFormat": "error: Message P99 - {{exported_job}}", + "refId": "C" + }, + { + "expr": "histogram_quantile(0.5, sum by (le, exported_job) (rate(sc_audit_ingestion_message_duration_seconds_bucket{exported_job=~\"$sc_instance\",message_category=~\"audit-message|saga-update\"}[$__rate_interval])))", + "legendFormat": "audit: Message P50 - {{exported_job}}", + "refId": "D" + }, + { + "expr": "histogram_quantile(0.95, sum by (le, exported_job) (rate(sc_audit_ingestion_message_duration_seconds_bucket{exported_job=~\"$sc_instance\",message_category=~\"audit-message|saga-update\"}[$__rate_interval])))", + "legendFormat": "audit: Message P95 - {{exported_job}}", + "refId": "E" + }, + { + "expr": "histogram_quantile(0.99, sum by (le, exported_job) (rate(sc_audit_ingestion_message_duration_seconds_bucket{exported_job=~\"$sc_instance\",message_category=~\"audit-message|saga-update\"}[$__rate_interval])))", + "legendFormat": "audit: Message P99 - {{exported_job}}", + "refId": "F" } ] }, { "id": 28, - "title": "Ingestion Duration Distribution (heatmap)", + "title": "Error Ingestion Duration Distribution (heatmap)", "description": "The whole message_duration_seconds histogram over time rather than a single quantile. Bucket boundaries are 10ms, 50ms, 100ms, 500ms, 1s and 5s. A band drifting upward under load is the shape of the instance running out of headroom.", "type": "heatmap", - "gridPos": { "h": 8, "w": 12, "x": 12, "y": 58 }, - "datasource": { "type": "prometheus", "uid": "prometheus" }, + "gridPos": { + "h": 8, + "w": 6, + "x": 12, + "y": 58 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, "options": { "calculate": false, "cellGap": 1, - "color": { "mode": "scheme", "scheme": "Spectral", "steps": 64, "reverse": false }, - "yAxis": { "unit": "s" }, - "legend": { "show": true }, - "tooltip": { "show": true, "yHistogram": true } + "color": { + "mode": "scheme", + "scheme": "Spectral", + "steps": 64, + "reverse": false + }, + "yAxis": { + "unit": "s" + }, + "legend": { + "show": true + }, + "tooltip": { + "show": true, + "yHistogram": true + } }, "targets": [ { @@ -515,17 +981,73 @@ } ] }, + { + "id": 42, + "title": "Audit Ingestion Duration Distribution (heatmap)", + "description": "Bucket distribution of audit message ingestion duration. Buckets top out at 5s, so a p99 pinned at 5 means the histogram is saturated.", + "type": "heatmap", + "gridPos": { + "h": 8, + "w": 6, + "x": 18, + "y": 58 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, + "options": { + "calculate": false, + "cellGap": 1, + "color": { + "mode": "scheme", + "scheme": "Spectral", + "steps": 64, + "reverse": false + }, + "yAxis": { + "unit": "s" + }, + "legend": { + "show": true + }, + "tooltip": { + "show": true, + "yHistogram": true + } + }, + "targets": [ + { + "refId": "A", + "expr": "sum by (le) (increase(sc_audit_ingestion_message_duration_seconds_bucket{exported_job=~\"$sc_instance\",message_category=~\"audit-message|saga-update\"}[$__rate_interval]))", + "legendFormat": "{{le}}", + "format": "heatmap" + } + ] + }, { "id": 29, - "title": "Storage Write Duration (s)", - "description": "sc.error.ingestion.storage_duration_seconds — the batch's storage write on its own, with announcing and forwarding excluded. Compare against batch duration below to see whether the persister or the rest of the pipeline is the constraint.", + "title": "Storage Write Duration (s), error only", + "description": "sc.error.ingestion.storage_duration_seconds \u2014 the batch's storage write on its own, with announcing and forwarding excluded. Compare against batch duration below to see whether the persister or the rest of the pipeline is the constraint. Error instance only: ServiceControl.Audit emits no sc.audit.ingestion.storage_duration_seconds, so there is no audit equivalent to chart.", "type": "timeseries", - "gridPos": { "h": 8, "w": 12, "x": 0, "y": 66 }, - "datasource": { "type": "prometheus", "uid": "prometheus" }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 66 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, "fieldConfig": { "defaults": { "unit": "s", - "custom": { "drawStyle": "line", "lineInterpolation": "smooth", "fillOpacity": 10 } + "custom": { + "drawStyle": "line", + "lineInterpolation": "smooth", + "fillOpacity": 10 + } } }, "targets": [ @@ -546,72 +1068,139 @@ { "id": 30, "title": "Batch Duration (s)", - "description": "sc.error.ingestion.batch_duration_seconds — the whole batch, storage write plus announcing and forwarding. Batch duration multiplied by concurrent ingestion is what caps the achievable ingestion rate.", + "description": "sc.error.ingestion.batch_duration_seconds \u2014 the whole batch, storage write plus announcing and forwarding. Batch duration multiplied by concurrent ingestion is what caps the achievable ingestion rate.", "type": "timeseries", - "gridPos": { "h": 8, "w": 12, "x": 12, "y": 66 }, - "datasource": { "type": "prometheus", "uid": "prometheus" }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 66 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, "fieldConfig": { "defaults": { "unit": "s", - "custom": { "drawStyle": "line", "lineInterpolation": "smooth", "fillOpacity": 10 } + "custom": { + "drawStyle": "line", + "lineInterpolation": "smooth", + "fillOpacity": 10 + } } }, "targets": [ { "expr": "histogram_quantile(0.5, sum by (le) (rate(sc_error_ingestion_batch_duration_seconds_bucket{exported_job=~\"$sc_instance\"}[$__rate_interval])))", - "legendFormat": "p50" + "legendFormat": "error: p50", + "refId": "A" }, { "expr": "histogram_quantile(0.95, sum by (le) (rate(sc_error_ingestion_batch_duration_seconds_bucket{exported_job=~\"$sc_instance\"}[$__rate_interval])))", - "legendFormat": "p95" + "legendFormat": "error: p95", + "refId": "B" }, { "expr": "histogram_quantile(0.99, sum by (le) (rate(sc_error_ingestion_batch_duration_seconds_bucket{exported_job=~\"$sc_instance\"}[$__rate_interval])))", - "legendFormat": "p99" + "legendFormat": "error: p99", + "refId": "C" + }, + { + "expr": "histogram_quantile(0.5, sum by (le) (rate(sc_audit_ingestion_batch_duration_seconds_bucket{exported_job=~\"$sc_instance\"}[$__rate_interval])))", + "legendFormat": "audit: p50", + "refId": "D" + }, + { + "expr": "histogram_quantile(0.95, sum by (le) (rate(sc_audit_ingestion_batch_duration_seconds_bucket{exported_job=~\"$sc_instance\"}[$__rate_interval])))", + "legendFormat": "audit: p95", + "refId": "E" + }, + { + "expr": "histogram_quantile(0.99, sum by (le) (rate(sc_audit_ingestion_batch_duration_seconds_bucket{exported_job=~\"$sc_instance\"}[$__rate_interval])))", + "legendFormat": "audit: p99", + "refId": "F" } ] }, { "id": 31, - "title": "Mean Duration — message / batch / storage write (s)", + "title": "Mean Duration \u2014 message / batch / storage write (s)", "description": "Histogram sum divided by histogram count, so these are true means rather than quantiles and are not clipped by the 5s top bucket. Mean message duration times the ingestion rate gives the concurrency the instance is sustaining.", "type": "timeseries", - "gridPos": { "h": 8, "w": 12, "x": 0, "y": 74 }, - "datasource": { "type": "prometheus", "uid": "prometheus" }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 74 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, "fieldConfig": { "defaults": { "unit": "s", - "custom": { "drawStyle": "line", "lineInterpolation": "smooth", "fillOpacity": 10 } + "custom": { + "drawStyle": "line", + "lineInterpolation": "smooth", + "fillOpacity": 10 + } } }, "targets": [ { "expr": "avg_over_time((sum(rate(sc_error_ingestion_message_duration_seconds_sum{exported_job=~\"$sc_instance\"}[1m])) / sum(rate(sc_error_ingestion_message_duration_seconds_count{exported_job=~\"$sc_instance\"}[1m])))[2m:15s])", - "legendFormat": "message" + "legendFormat": "error: message", + "refId": "A" }, { "expr": "avg_over_time((sum(rate(sc_error_ingestion_batch_duration_seconds_sum{exported_job=~\"$sc_instance\"}[1m])) / sum(rate(sc_error_ingestion_batch_duration_seconds_count{exported_job=~\"$sc_instance\"}[1m])))[2m:15s])", - "legendFormat": "batch" + "legendFormat": "error: batch", + "refId": "B" }, { "expr": "avg_over_time((sum(rate(sc_error_ingestion_storage_duration_seconds_sum{exported_job=~\"$sc_instance\"}[1m])) / sum(rate(sc_error_ingestion_storage_duration_seconds_count{exported_job=~\"$sc_instance\"}[1m])))[2m:15s])", - "legendFormat": "storage write" + "legendFormat": "error: storage write", + "refId": "C" + }, + { + "expr": "avg_over_time((sum(rate(sc_audit_ingestion_message_duration_seconds_sum{exported_job=~\"$sc_instance\"}[1m])) / sum(rate(sc_audit_ingestion_message_duration_seconds_count{exported_job=~\"$sc_instance\"}[1m])))[2m:15s])", + "legendFormat": "audit: message", + "refId": "D" + }, + { + "expr": "avg_over_time((sum(rate(sc_audit_ingestion_batch_duration_seconds_sum{exported_job=~\"$sc_instance\"}[1m])) / sum(rate(sc_audit_ingestion_batch_duration_seconds_count{exported_job=~\"$sc_instance\"}[1m])))[2m:15s])", + "legendFormat": "audit: batch", + "refId": "E" } ] }, { "id": 32, - "title": "Storage Share of Batch Time", - "description": "Total time spent in storage writes divided by total time spent in batches. Near 100% means the persister owns the batch and tuning belongs there; a low share means announcing and forwarding are where the batch time goes.", + "title": "Storage Share of Batch Time, error only", + "description": "Total time spent in storage writes divided by total time spent in batches. Near 100% means the persister owns the batch and tuning belongs there; a low share means announcing and forwarding are where the batch time goes. Error instance only: ServiceControl.Audit emits no sc.audit.ingestion.storage_duration_seconds, so there is no audit equivalent to chart.", "type": "timeseries", - "gridPos": { "h": 8, "w": 12, "x": 12, "y": 74 }, - "datasource": { "type": "prometheus", "uid": "prometheus" }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 74 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, "fieldConfig": { "defaults": { "unit": "percentunit", "min": 0, "max": 1, - "custom": { "drawStyle": "line", "lineInterpolation": "smooth", "fillOpacity": 20 } + "custom": { + "drawStyle": "line", + "lineInterpolation": "smooth", + "fillOpacity": 20 + } } }, "targets": [ @@ -624,9 +1213,14 @@ { "id": 104, "type": "row", - "title": "ServiceControl ingestion — failures & batch health", + "title": "ServiceControl ingestion: failures and batch health (error and audit)", "collapsed": false, - "gridPos": { "h": 1, "w": 24, "x": 0, "y": 82 }, + "gridPos": { + "h": 1, + "w": 24, + "x": 0, + "y": 82 + }, "panels": [] }, { @@ -634,51 +1228,102 @@ "title": "ServiceControl Ingestion Outcome / sec (by result)", "description": "ServiceControl OTel histogram sc.error.ingestion.message_duration_seconds count, broken down by ingestion result (success / failed / skipped). 'success' is what counts as ingested; 'failed'/'skipped' explain the gap between raised and ingested.", "type": "timeseries", - "gridPos": { "h": 8, "w": 8, "x": 0, "y": 83 }, - "datasource": { "type": "prometheus", "uid": "prometheus" }, + "gridPos": { + "h": 8, + "w": 8, + "x": 0, + "y": 83 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, "fieldConfig": { "defaults": { "unit": "reqps", - "custom": { "drawStyle": "line", "lineInterpolation": "smooth", "fillOpacity": 10 } + "custom": { + "drawStyle": "line", + "lineInterpolation": "smooth", + "fillOpacity": 10 + } } }, "targets": [ { "expr": "avg_over_time((sum by (result) (rate(sc_error_ingestion_message_duration_seconds_count{exported_job=~\"$sc_instance\",message_category=\"failed-message\"}[1m])))[2m:15s])", - "legendFormat": "{{result}}" + "legendFormat": "error: {{result}}", + "refId": "A" + }, + { + "expr": "avg_over_time((sum by (result) (rate(sc_audit_ingestion_message_duration_seconds_count{exported_job=~\"$sc_instance\",message_category=~\"audit-message|saga-update\"}[1m])))[2m:15s])", + "legendFormat": "audit: {{result}}", + "refId": "B" } ] }, { "id": 33, "title": "Ingestion Failures / sec (by disposition)", - "description": "sc.error.ingestion.failures_total — messages the ingestor could not handle, split by what it did with them: 'retry' means the transport will redeliver, 'stored-poison' means it gave up and wrote a failed import. Rising stored-poison is permanent data loss from the load run's point of view.", + "description": "sc.error.ingestion.failures_total \u2014 messages the ingestor could not handle, split by what it did with them: 'retry' means the transport will redeliver, 'stored-poison' means it gave up and wrote a failed import. Rising stored-poison is permanent data loss from the load run's point of view.", "type": "timeseries", - "gridPos": { "h": 8, "w": 8, "x": 8, "y": 83 }, - "datasource": { "type": "prometheus", "uid": "prometheus" }, + "gridPos": { + "h": 8, + "w": 8, + "x": 8, + "y": 83 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, "fieldConfig": { "defaults": { "unit": "reqps", - "custom": { "drawStyle": "line", "lineInterpolation": "smooth", "fillOpacity": 10 } + "custom": { + "drawStyle": "line", + "lineInterpolation": "smooth", + "fillOpacity": 10 + } } }, "options": { - "legend": { "displayMode": "table", "placement": "bottom", "calcs": ["mean", "max"] } + "legend": { + "displayMode": "table", + "placement": "bottom", + "calcs": [ + "mean", + "max" + ] + } }, "targets": [ { "expr": "sum by (result, message_category) (rate(sc_error_ingestion_failures_total{exported_job=~\"$sc_instance\"}[$__rate_interval]))", - "legendFormat": "{{result}} / {{message_category}}" + "legendFormat": "error: {{result}} / {{message_category}}", + "refId": "A" + }, + { + "expr": "sum by (result, message_category) (rate(sc_audit_ingestion_failures_total{exported_job=~\"$sc_instance\"}[$__rate_interval]))", + "legendFormat": "audit: {{result}} / {{message_category}}", + "refId": "B" } ] }, { "id": 34, "title": "Batch Outcome Share", - "description": "Share of batches ending full, partial or failed. A high 'full' share is a healthy saturated pipeline; any sustained 'failed' share is what drives the consecutive-batch-failure circuit breaker.", + "description": "Share of batches ending full, partial or failed. A high 'full' share is a healthy saturated pipeline; any sustained 'failed' share is what drives the consecutive-batch-failure circuit breaker. Error and audit batches stack in separate groups: each share sums to 1 on its own, so stacking them together would overflow the axis.", "type": "timeseries", - "gridPos": { "h": 8, "w": 8, "x": 16, "y": 83 }, - "datasource": { "type": "prometheus", "uid": "prometheus" }, + "gridPos": { + "h": 8, + "w": 8, + "x": 16, + "y": 83 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, "fieldConfig": { "defaults": { "unit": "percentunit", @@ -688,23 +1333,54 @@ "drawStyle": "line", "lineInterpolation": "smooth", "fillOpacity": 30, - "stacking": { "mode": "normal", "group": "A" } + "stacking": { + "mode": "normal", + "group": "A" + } } - } + }, + "overrides": [ + { + "matcher": { + "id": "byFrameRefID", + "options": "B" + }, + "properties": [ + { + "id": "custom.stacking", + "value": { + "mode": "normal", + "group": "audit" + } + } + ] + } + ] }, "targets": [ { "expr": "sum by (result) (rate(sc_error_ingestion_batch_duration_seconds_count{exported_job=~\"$sc_instance\"}[$__rate_interval])) / on() group_left() sum(rate(sc_error_ingestion_batch_duration_seconds_count{exported_job=~\"$sc_instance\"}[$__rate_interval]))", - "legendFormat": "{{result}}" + "legendFormat": "error: {{result}}", + "refId": "A" + }, + { + "expr": "sum by (result) (rate(sc_audit_ingestion_batch_duration_seconds_count{exported_job=~\"$sc_instance\"}[$__rate_interval])) / on() group_left() sum(rate(sc_audit_ingestion_batch_duration_seconds_count{exported_job=~\"$sc_instance\"}[$__rate_interval]))", + "legendFormat": "audit: {{result}}", + "refId": "B" } ] }, { "id": 105, "type": "row", - "title": "ServiceControl recoverability — retry, archive & retention", + "title": "ServiceControl recoverability \u2014 retry, archive & retention", "collapsed": false, - "gridPos": { "h": 1, "w": 24, "x": 0, "y": 91 }, + "gridPos": { + "h": 1, + "w": 24, + "x": 0, + "y": 91 + }, "panels": [] }, { @@ -712,8 +1388,16 @@ "title": "Recoverability Throughput / sec (messages moved)", "description": "Combined view of items moving through the three recoverability pipelines: sc.retry.messages_total (messages staged/forwarded/skipped by retry type), sc.archive.messages_total (messages archived/unarchived), and sc.retention.rows_deleted_total (rows deleted by entity). Stacked, the total is the rate of recoverability work ServiceControl is doing. Compare against the testing tool's own errors_replayed_total and errors_archived_total to see whether ServiceControl is keeping up with the requested work.", "type": "timeseries", - "gridPos": { "h": 8, "w": 12, "x": 0, "y": 92 }, - "datasource": { "type": "prometheus", "uid": "prometheus" }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 92 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, "fieldConfig": { "defaults": { "unit": "reqps", @@ -721,12 +1405,22 @@ "drawStyle": "line", "lineInterpolation": "smooth", "fillOpacity": 25, - "stacking": { "mode": "normal", "group": "A" } + "stacking": { + "mode": "normal", + "group": "A" + } } } }, "options": { - "legend": { "displayMode": "table", "placement": "bottom", "calcs": ["mean", "max"] } + "legend": { + "displayMode": "table", + "placement": "bottom", + "calcs": [ + "mean", + "max" + ] + } }, "targets": [ { @@ -746,18 +1440,36 @@ { "id": 36, "title": "Operations In Progress (retry & archive)", - "description": "sc.retry.operations_in_progress and sc.archive.operations_in_progress — the count of retry and archive operations currently in flight, split by type and state. Retry states are waiting / preparing / forwarding; archive states are started / progressing / finalizing. A non-zero forwarding count under load means retries are actively being dispatched; a stuck preparing count points at a slow store scan.", + "description": "sc.retry.operations_in_progress and sc.archive.operations_in_progress \u2014 the count of retry and archive operations currently in flight, split by type and state. Retry states are waiting / preparing / forwarding; archive states are started / progressing / finalizing. A non-zero forwarding count under load means retries are actively being dispatched; a stuck preparing count points at a slow store scan.", "type": "timeseries", - "gridPos": { "h": 8, "w": 12, "x": 12, "y": 92 }, - "datasource": { "type": "prometheus", "uid": "prometheus" }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 92 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, "fieldConfig": { "defaults": { "unit": "short", - "custom": { "drawStyle": "line", "lineInterpolation": "smooth", "fillOpacity": 10 } + "custom": { + "drawStyle": "line", + "lineInterpolation": "smooth", + "fillOpacity": 10 + } } }, "options": { - "legend": { "displayMode": "table", "placement": "bottom", "calcs": ["lastNotNull"] } + "legend": { + "displayMode": "table", + "placement": "bottom", + "calcs": [ + "lastNotNull" + ] + } }, "targets": [ { @@ -772,19 +1484,38 @@ }, { "id": 37, - "title": "Retry Durations (s) — mean by phase", + "title": "Retry Durations (s) \u2014 mean by phase", "description": "Mean duration for each phase of the retry pipeline: sc.retry.operation_duration_seconds (request to completion), sc.retry.prepare_duration_seconds (store scan + batch creation), sc.retry.stage_duration_seconds (batch staging), and sc.retry.forward_duration_seconds (batch forwarding). Computed as histogram sum / count so these are true means, not clipped by bucket boundaries. Mirrors the ingestion 'Mean Duration' panel (id 31). Operation duration spans the whole retry; the phase breakdown shows where the time goes.", "type": "timeseries", - "gridPos": { "h": 8, "w": 12, "x": 0, "y": 100 }, - "datasource": { "type": "prometheus", "uid": "prometheus" }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 100 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, "fieldConfig": { "defaults": { "unit": "s", - "custom": { "drawStyle": "line", "lineInterpolation": "smooth", "fillOpacity": 10 } + "custom": { + "drawStyle": "line", + "lineInterpolation": "smooth", + "fillOpacity": 10 + } } }, "options": { - "legend": { "displayMode": "table", "placement": "bottom", "calcs": ["mean", "max"] } + "legend": { + "displayMode": "table", + "placement": "bottom", + "calcs": [ + "mean", + "max" + ] + } }, "targets": [ { @@ -807,19 +1538,38 @@ }, { "id": 38, - "title": "Archive & Retention Durations (s) — mean", + "title": "Archive & Retention Durations (s) \u2014 mean", "description": "Mean duration for the remaining recoverability pipelines: sc.archive.operation_duration_seconds (full archive/unarchive operation), sc.archive.batch_duration_seconds (individual archive batch), and sc.retention.cycle_duration_seconds (retention sweep pass). True means (histogram sum / count). Archive operation duration spans the whole group; batch duration is the per-batch write. Retention cycle is the full sweep pass, split by entity in the legend.", "type": "timeseries", - "gridPos": { "h": 8, "w": 12, "x": 12, "y": 100 }, - "datasource": { "type": "prometheus", "uid": "prometheus" }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 100 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, "fieldConfig": { "defaults": { "unit": "s", - "custom": { "drawStyle": "line", "lineInterpolation": "smooth", "fillOpacity": 10 } + "custom": { + "drawStyle": "line", + "lineInterpolation": "smooth", + "fillOpacity": 10 + } } }, "options": { - "legend": { "displayMode": "table", "placement": "bottom", "calcs": ["mean", "max"] } + "legend": { + "displayMode": "table", + "placement": "bottom", + "calcs": [ + "mean", + "max" + ] + } }, "targets": [ { @@ -839,26 +1589,49 @@ { "id": 39, "title": "Consecutive Retention Failures", - "description": "sc.retention.consecutive_failures_total — the retention sweep's own circuit-breaker signal, split by entity (failed_messages, event_log, group_comments). Resets to 0 on the first successful sweep. Anything above 0 means the sweep is currently failing. Mirrors the ingestion 'Consecutive Batch Failures' panel (id 23).", + "description": "sc.retention.consecutive_failures_total \u2014 the retention sweep's own circuit-breaker signal, split by entity (failed_messages, event_log, group_comments). Resets to 0 on the first successful sweep. Anything above 0 means the sweep is currently failing. Mirrors the ingestion 'Consecutive Batch Failures' panel (id 23).", "type": "stat", - "gridPos": { "h": 6, "w": 12, "x": 0, "y": 108 }, - "datasource": { "type": "prometheus", "uid": "prometheus" }, + "gridPos": { + "h": 6, + "w": 12, + "x": 0, + "y": 108 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, "fieldConfig": { "defaults": { "unit": "short", - "color": { "mode": "thresholds" }, + "color": { + "mode": "thresholds" + }, "thresholds": { "mode": "absolute", "steps": [ - { "color": "green", "value": null }, - { "color": "orange", "value": 1 }, - { "color": "red", "value": 3 } + { + "color": "green", + "value": null + }, + { + "color": "orange", + "value": 1 + }, + { + "color": "red", + "value": 3 + } ] } } }, "options": { - "reduceOptions": { "calcs": ["lastNotNull"] }, + "reduceOptions": { + "calcs": [ + "lastNotNull" + ] + }, "graphMode": "area", "colorMode": "background" }, @@ -872,26 +1645,49 @@ { "id": 40, "title": "Pending Bulk Retry Requests", - "description": "sc.retry.pending_bulk_requests — bulk retry requests queued for preparation. A rising value means retry preparation cannot keep up with incoming bulk retry requests; at zero the retry pipeline is idle or keeping pace.", + "description": "sc.retry.pending_bulk_requests \u2014 bulk retry requests queued for preparation. A rising value means retry preparation cannot keep up with incoming bulk retry requests; at zero the retry pipeline is idle or keeping pace.", "type": "stat", - "gridPos": { "h": 6, "w": 12, "x": 12, "y": 108 }, - "datasource": { "type": "prometheus", "uid": "prometheus" }, + "gridPos": { + "h": 6, + "w": 12, + "x": 12, + "y": 108 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, "fieldConfig": { "defaults": { "unit": "short", - "color": { "mode": "thresholds" }, + "color": { + "mode": "thresholds" + }, "thresholds": { "mode": "absolute", "steps": [ - { "color": "green", "value": null }, - { "color": "orange", "value": 1 }, - { "color": "red", "value": 10 } + { + "color": "green", + "value": null + }, + { + "color": "orange", + "value": 1 + }, + { + "color": "red", + "value": 10 + } ] } } }, "options": { - "reduceOptions": { "calcs": ["lastNotNull"] }, + "reduceOptions": { + "calcs": [ + "lastNotNull" + ] + }, "graphMode": "area", "colorMode": "value" }, diff --git a/tools/testing-tool/TestingTool.Contracts/AuditBypassStatus.cs b/tools/testing-tool/TestingTool.Contracts/AuditBypassStatus.cs new file mode 100644 index 0000000000..a30a045ba3 --- /dev/null +++ b/tools/testing-tool/TestingTool.Contracts/AuditBypassStatus.cs @@ -0,0 +1,32 @@ +namespace TestingTool.Contracts; + +/// +/// Live status of the direct audit-queue writer, returned by GET /api/audit-bypass/status +/// and included in the GET /api/status snapshot. +/// +public sealed class AuditBypassStatus +{ + /// Whether the writer is currently emitting audit envelopes. + public bool Running { get; init; } + + /// Target claim rate in claims/second (0 if idle). A saga episode claim emits six envelopes. + public double Rate { get; init; } + + /// Envelopes actually written per second since the run started. + public double AchievedRate { get; init; } + + /// Saga snapshots emitted per processed message for the current run. + public double SagaSnapshotRatio { get; init; } + + /// Total processed-message envelopes written to the audit queue since process start. + public long ProcessedMessagesWritten { get; init; } + + /// Total saga-update snapshots written to the audit queue since process start. + public long SagaSnapshotsWritten { get; init; } + + /// Total audit sends that failed since process start. + public long MessagesFailed { get; init; } + + /// When the current run started (UTC ISO 8601, null if idle). + public string? StartedAt { get; init; } +} diff --git a/tools/testing-tool/TestingTool.Contracts/StartAuditBypassRequest.cs b/tools/testing-tool/TestingTool.Contracts/StartAuditBypassRequest.cs new file mode 100644 index 0000000000..29b1ac370f --- /dev/null +++ b/tools/testing-tool/TestingTool.Contracts/StartAuditBypassRequest.cs @@ -0,0 +1,24 @@ +namespace TestingTool.Contracts; + +/// +/// Request body for POST /api/audit-bypass/start. Starts the direct audit-queue writer, +/// which writes audit envelopes straight to the ServiceControl.Audit instance's audit queue +/// without a real endpoint ever having processed a message. +/// +public sealed class StartAuditBypassRequest +{ + /// Target claim rate in claims/second. Defaults to 100. + public double? Rate { get; init; } + + /// Optional auto-stop duration in seconds. Null/0 = run until explicitly stopped. + public double? DurationSeconds { get; init; } + + /// Parallel worker tasks drawing from the run's shared send budget. Defaults to the processor count. + public int? Parallelism { get; init; } + + /// + /// Saga snapshots to emit per processed message. The default of 0.2 emits two saga snapshots + /// for every ten processed messages. 0 disables saga episodes entirely. + /// + public double? SagaSnapshotRatio { get; init; } +} diff --git a/tools/testing-tool/TestingTool.Contracts/TestingToolStatus.cs b/tools/testing-tool/TestingTool.Contracts/TestingToolStatus.cs index e51d8c73e8..abcd8d3ae2 100644 --- a/tools/testing-tool/TestingTool.Contracts/TestingToolStatus.cs +++ b/tools/testing-tool/TestingTool.Contracts/TestingToolStatus.cs @@ -47,6 +47,15 @@ public sealed class TestingToolStatus /// transport/broker issues on the bypass path — the handler path may be unaffected. public long BypassErrorsFailed { get; init; } + /// Total processed-message envelopes written to the audit queue since process start. + public long AuditMessagesWritten { get; init; } + + /// Total saga-update snapshots written to the audit queue since process start. + public long AuditSagaSnapshotsWritten { get; init; } + + /// Total audit-queue sends that failed since process start. + public long AuditMessagesFailed { get; init; } + /// /// Total internal-looking custom check failures reported to ServiceControl since process start /// (via the custom-check-failures job). Each failure is a ReportCustomCheckResult sent diff --git a/tools/testing-tool/TestingTool/Auditing/AuditEnvelope.cs b/tools/testing-tool/TestingTool/Auditing/AuditEnvelope.cs new file mode 100644 index 0000000000..5e48b4ae8e --- /dev/null +++ b/tools/testing-tool/TestingTool/Auditing/AuditEnvelope.cs @@ -0,0 +1,19 @@ +using NServiceBus; + +namespace TestingTool.Auditing; + +/// +/// Carrier message for the direct audit-queue writer. It is never handled by anything: the whole +/// wire message (headers, body and NServiceBus.EnclosedMessageTypes) is replaced by +/// , because an audit envelope has to look like a message some +/// other endpoint processed rather than one the testing tool sent. +/// +/// +/// It implements because nothing handles it: NServiceBus infers message +/// types from handled types, and refuses to send a type it has no metadata for. +/// +public class AuditEnvelope : IMessage +{ + /// Monotonically increasing sequence number within a generation run. + public long Sequence { get; set; } +} diff --git a/tools/testing-tool/TestingTool/Auditing/AuditMessageFactory.cs b/tools/testing-tool/TestingTool/Auditing/AuditMessageFactory.cs new file mode 100644 index 0000000000..fd7178c868 --- /dev/null +++ b/tools/testing-tool/TestingTool/Auditing/AuditMessageFactory.cs @@ -0,0 +1,263 @@ +using System.Buffers; +using System.Text; +using System.Text.Json; +using NServiceBus; +using TestingTool.Scenarios; + +namespace TestingTool.Auditing; + +/// +/// Builds the wire messages the direct audit-queue writer puts on the ServiceControl.Audit +/// instance's audit queue: processed-message envelopes, and the saga episodes that link a run of +/// those envelopes to a saga's snapshot history. +/// +/// +/// Endpoints and hosts are derived from the shard id alone, so they are stable across runs and +/// KnownEndpoints settles. Message, saga and conversation ids additionally take a per-run salt, +/// because the claim sequence restarts at 1 on every run: without it a restart would replay the +/// same ids and one saga would accumulate repeated New/Updated/Completed cycles. +/// +public sealed class AuditMessageFactory +{ + /// + /// The discriminator ServiceControl matches to route an audit message to its saga branch. + /// AuditPersister compares NServiceBus.EnclosedMessageTypes to this with ==, so an + /// assembly-qualified or multi-type value is silently stored as an ordinary audit row instead, + /// with no error and no saga history. + /// + public const string SagaUpdatedMessageType = "ServiceControl.EndpointPlugin.Messages.SagaState.SagaUpdatedMessage"; + + /// Snapshots per saga: New, then Updated, then Completed. + public const int PhasesPerSaga = 3; + + const string JsonContentType = "application/json"; + + static readonly string[] EndpointNames = + [ + "Sales", "Billing", "Shipping", "Warehouse", "Notifications", "Integration" + ]; + + static readonly string[] MessageTypes = + [ + "Acme.Sales.Messages.PlaceOrder", + "Acme.Sales.Messages.OrderPlaced", + "Acme.Billing.Messages.ChargeCustomer", + "Acme.Billing.Messages.PaymentSettled", + "Acme.Shipping.Messages.DispatchConsignment", + "Acme.Warehouse.Messages.ReserveInventory", + "Acme.Notifications.Messages.SendConfirmation", + "Acme.Integration.Messages.SyncLedgerEntry" + ]; + + static readonly string[] SagaTypes = + [ + "Acme.Sales.Sagas.OrderFulfilmentSaga", + "Acme.Billing.Sagas.SubscriptionRenewalSaga", + "Acme.Shipping.Sagas.ConsignmentTrackingSaga" + ]; + + static readonly string[] PhaseStatus = ["New", "Updated", "Completed"]; + + readonly SimulatedEndpoint[] endpoints; + readonly string shardId; + + public AuditMessageFactory(string shardId) + { + this.shardId = shardId; + endpoints = [.. EndpointNames.Select(name => + SimulatedEndpoint.Create($"{name}.Load-{shardId}", $"host-{shardId}-{name.ToLowerInvariant()}"))]; + } + + /// + /// Builds an envelope for a message an endpoint processed successfully, with no saga involvement. + /// + public RawAuditPayload CreateProcessedMessage(long runSalt, long sequence) => + BuildProcessedMessage(runSalt, sequence, saga: null); + + /// + /// Builds one saga's whole life: for each of New, Updated and Completed, the message that drove + /// the change followed by the snapshot it produced. + /// + /// + /// The episode is built as a unit so a single worker writes the phases in order. Handing phases + /// out from a shared counter lets Completed overtake New once more than one worker is running, + /// and the saga history then reads backwards. + /// + public IReadOnlyList CreateSagaEpisode(long runSalt, long episodeOrdinal, long sequence) + { + // Selecting on the claim number would collapse these pools: only sequences that are a + // multiple of the episode period ever get here, so their residues are not uniform and every + // episode of a run can land on the same endpoint and saga type. + var ordinal = episodeOrdinal; + var endpoint = endpoints[(int)(ordinal % endpoints.Length)]; + var sagaType = SagaTypes[(int)(ordinal % SagaTypes.Length)]; + var sagaId = SimulatedEndpoint.DeterministicGuid($"{shardId}/{runSalt}/{endpoint.Name}/{sagaType}/{ordinal}"); + + // A saga that starts now and advances in milliseconds, not one backdated by minutes: the + // saga history view orders by FinishTime, and the audit tables are partitioned by ingestion + // hour, so timestamps have to stay near wall clock. + var start = DateTimeOffset.UtcNow; + + var episode = new List(PhasesPerSaga * 2); + for (var phase = 0; phase < PhasesPerSaga; phase++) + { + var saga = new SagaParticipation(sagaId, sagaType, endpoint, PhaseStatus[phase], start, phase); + var message = BuildProcessedMessage(runSalt, sequence, saga); + episode.Add(message); + episode.Add(BuildSagaSnapshot(runSalt, saga, message.Headers[Headers.MessageId], sequence + phase)); + } + + return episode; + } + + /// + /// The number of claims between saga episodes that yields the requested snapshot-to-message + /// ratio. A period of p produces p - 1 plain messages plus one episode, and an episode is + /// snapshots alongside the same number of processed messages, so + /// the ratio is P / (p - 1 + P) and the period that solves it is P/r - P + 1. + /// + public static int SagaEpisodePeriod(double sagaSnapshotRatio) => + sagaSnapshotRatio <= 0 ? 0 : Math.Max(1, (int)Math.Round(PhasesPerSaga / sagaSnapshotRatio - PhasesPerSaga + 1, MidpointRounding.AwayFromZero)); + + RawAuditPayload BuildProcessedMessage(long runSalt, long sequence, SagaParticipation? saga) + { + var receiver = saga?.Endpoint ?? endpoints[(int)(sequence % endpoints.Length)]; + var sender = endpoints[(int)(sequence / endpoints.Length % endpoints.Length)]; + var messageType = MessageTypes[(int)(sequence % MessageTypes.Length)]; + + var processingEnded = saga?.PhaseTime ?? DateTimeOffset.UtcNow; + var processingStarted = processingEnded.AddMilliseconds(-(5 + sequence % 120)); + var timeSent = processingStarted.AddMilliseconds(-(10 + sequence % 400)); + + var messageId = SimulatedEndpoint.DeterministicGuid( + $"{shardId}/{runSalt}/msg/{sequence}/{saga?.SagaId.ToString() ?? "none"}/{saga?.Phase ?? 0}").ToString(); + + var headers = CommonHeaders(runSalt, sequence, messageId, receiver, sender, timeSent, processingStarted, processingEnded); + headers[Headers.EnclosedMessageTypes] = messageType; + headers["TestingTool.MessageType"] = messageType; + + if (saga is not null) + { + // These two headers are the only thing that links an audited message to a saga: the EF + // persister rebuilds InvokedSagas from them at query time. The shapes are strict, and a + // malformed value throws inside the per-row projection, which 500s a whole page of + // /api/messages after the rows are already stored. + headers["NServiceBus.InvokedSagas"] = $"{saga.SagaType}:{saga.SagaId}"; + headers["ServiceControl.SagaStateChange"] = $"{saga.SagaId}:{saga.Status}"; + } + + var body = Encoding.UTF8.GetBytes(MessageTextGenerator.GenerateBody(sequence)); + return new RawAuditPayload(headers, body); + } + + RawAuditPayload BuildSagaSnapshot(long runSalt, SagaParticipation saga, string initiatingMessageId, long sequence) + { + var headers = CommonHeaders( + runSalt, + sequence, + SimulatedEndpoint.DeterministicGuid($"{shardId}/{runSalt}/snapshot/{saga.SagaId}/{saga.Phase}").ToString(), + saga.Endpoint, + saga.Endpoint, + saga.PhaseTime.AddMilliseconds(-20), + saga.PhaseTime.AddMilliseconds(-10), + saga.PhaseTime); + + headers[Headers.EnclosedMessageTypes] = SagaUpdatedMessageType; + headers["TestingTool.MessageType"] = SagaUpdatedMessageType; + + var body = WriteSagaUpdatedMessage(saga, initiatingMessageId, sequence); + return new RawAuditPayload(headers, body); + } + + Dictionary CommonHeaders( + long runSalt, + long sequence, + string messageId, + SimulatedEndpoint receiver, + SimulatedEndpoint sender, + DateTimeOffset timeSent, + DateTimeOffset processingStarted, + DateTimeOffset processingEnded) => new() + { + [Headers.MessageId] = messageId, + [Headers.ProcessingEndpoint] = receiver.Name, + [Headers.ProcessingMachine] = receiver.Machine, + [Headers.HostId] = receiver.HostId.ToString("N"), + [Headers.HostDisplayName] = receiver.HostDisplayName, + [Headers.OriginatingEndpoint] = sender.Name, + [Headers.OriginatingMachine] = sender.Machine, + [Headers.OriginatingHostId] = sender.HostId.ToString("N"), + [Headers.TimeSent] = DateTimeOffsetHelper.ToWireFormattedString(timeSent), + [Headers.ProcessingStarted] = DateTimeOffsetHelper.ToWireFormattedString(processingStarted), + [Headers.ProcessingEnded] = DateTimeOffsetHelper.ToWireFormattedString(processingEnded), + [Headers.ContentType] = JsonContentType, + [Headers.MessageIntent] = nameof(MessageIntent.Send), + // Five consecutive messages share a conversation so the message-flow view has a graph + // to draw rather than a run of singletons. + [Headers.ConversationId] = SimulatedEndpoint.DeterministicGuid($"{shardId}/{runSalt}/conversation/{sequence / 5}").ToString(), + ["TestingTool.Bypass"] = "true" + }; + + static byte[] WriteSagaUpdatedMessage(SagaParticipation saga, string initiatingMessageId, long sequence) + { + // Property names are written in the CLR casing on purpose: ServiceControl deserializes with + // a source-generated context that uses the default, case-sensitive naming policy. A + // mis-cased scalar silently becomes a default; a mis-cased Initiator or a null + // ResultingMessages throws inside SagaSnapshotFactory and lands the message in failed imports. + var buffer = new ArrayBufferWriter(1024); + using (var writer = new Utf8JsonWriter(buffer)) + { + writer.WriteStartObject(); + writer.WriteString("SagaId", saga.SagaId); + writer.WriteString("SagaType", saga.SagaType); + writer.WriteString("Endpoint", saga.Endpoint.Name); + writer.WriteBoolean("IsNew", saga.Phase == 0); + writer.WriteBoolean("IsCompleted", saga.Phase == PhasesPerSaga - 1); + writer.WriteString("StartTime", saga.StartTime.UtcDateTime); + writer.WriteString("FinishTime", saga.PhaseTime.UtcDateTime); + writer.WriteString("SagaState", SagaState(saga, sequence)); + + writer.WriteStartObject("Initiator"); + writer.WriteString("InitiatingMessageId", initiatingMessageId); + writer.WriteString("MessageType", MessageTypes[(int)(sequence % MessageTypes.Length)]); + writer.WriteBoolean("IsSagaTimeoutMessage", false); + writer.WriteString("TimeSent", saga.PhaseTime.UtcDateTime); + writer.WriteString("OriginatingMachine", saga.Endpoint.Machine); + writer.WriteString("OriginatingEndpoint", saga.Endpoint.Name); + writer.WriteString("Intent", nameof(MessageIntent.Send)); + writer.WriteEndObject(); + + writer.WriteStartArray("ResultingMessages"); + if (saga.Phase < PhasesPerSaga - 1) + { + writer.WriteStartObject(); + writer.WriteString("MessageType", MessageTypes[(int)((sequence + 1) % MessageTypes.Length)]); + writer.WriteString("TimeSent", saga.PhaseTime.UtcDateTime); + writer.WriteString("Destination", saga.Endpoint.Name); + writer.WriteString("ResultingMessageId", initiatingMessageId); + writer.WriteString("Intent", nameof(MessageIntent.Send)); + writer.WriteEndObject(); + } + writer.WriteEndArray(); + + writer.WriteEndObject(); + } + + return buffer.WrittenSpan.ToArray(); + } + + static string SagaState(SagaParticipation saga, long sequence) => + $$"""{"Id":"{{saga.SagaId}}","Status":"{{saga.Status}}","OrderNumber":"ORD-{{sequence:D8}}","Originator":"load"}"""; + + sealed record SagaParticipation( + Guid SagaId, + string SagaType, + SimulatedEndpoint Endpoint, + string Status, + DateTimeOffset StartTime, + int Phase) + { + /// When this phase finished. Phases are milliseconds apart, as a real saga's are. + public DateTimeOffset PhaseTime { get; } = StartTime.AddMilliseconds(40 * Phase); + } +} diff --git a/tools/testing-tool/TestingTool/Auditing/RawAuditPayload.cs b/tools/testing-tool/TestingTool/Auditing/RawAuditPayload.cs new file mode 100644 index 0000000000..5324e8b186 --- /dev/null +++ b/tools/testing-tool/TestingTool/Auditing/RawAuditPayload.cs @@ -0,0 +1,9 @@ +namespace TestingTool.Auditing; + +/// +/// The wire message the direct audit-queue writer wants on the queue, attached to the outgoing +/// send through the send options context bag and applied by . +/// +/// Headers to set on the outgoing message, overwriting anything the pipeline produced. +/// The serialized body to send in place of the carrier message's body. +public sealed record RawAuditPayload(IReadOnlyDictionary Headers, ReadOnlyMemory Body); diff --git a/tools/testing-tool/TestingTool/Auditing/RawAuditPayloadBehavior.cs b/tools/testing-tool/TestingTool/Auditing/RawAuditPayloadBehavior.cs new file mode 100644 index 0000000000..a91f7dd61a --- /dev/null +++ b/tools/testing-tool/TestingTool/Auditing/RawAuditPayloadBehavior.cs @@ -0,0 +1,33 @@ +using NServiceBus.Pipeline; + +namespace TestingTool.Auditing; + +/// +/// Applies a attached to a send, replacing the serialized body and +/// the headers the outgoing pipeline produced. +/// +/// +/// This runs at the physical stage on purpose. Headers set through SendOptions.SetHeader are +/// applied earlier, so the serializer's NServiceBus.EnclosedMessageTypes and +/// AttachSenderRelatedInfoOnMessageBehavior's NServiceBus.TimeSent would overwrite +/// them. ServiceControl routes an audit message to its saga branch only when +/// EnclosedMessageTypes is exactly SagaUpdatedMessage's unqualified full name, which +/// the serializer never writes. +/// +public sealed class RawAuditPayloadBehavior : Behavior +{ + public override Task Invoke(IOutgoingPhysicalMessageContext context, Func next) + { + if (context.Extensions.TryGet(out var payload)) + { + foreach (var (key, value) in payload.Headers) + { + context.Headers[key] = value; + } + + context.UpdateMessage(payload.Body); + } + + return next(); + } +} diff --git a/tools/testing-tool/TestingTool/Auditing/SimulatedEndpoint.cs b/tools/testing-tool/TestingTool/Auditing/SimulatedEndpoint.cs new file mode 100644 index 0000000000..64d447f09c --- /dev/null +++ b/tools/testing-tool/TestingTool/Auditing/SimulatedEndpoint.cs @@ -0,0 +1,20 @@ +using System.Security.Cryptography; +using System.Text; + +namespace TestingTool.Auditing; + +/// +/// One endpoint instance the audit load pretends to come from. Host ids are derived from the name +/// so a given shard always reports the same instances, which is what lets ServiceControl's +/// KnownEndpoints settle instead of growing without bound. +/// +public sealed record SimulatedEndpoint(string Name, string Machine, Guid HostId) +{ + public string HostDisplayName => Machine; + + public static SimulatedEndpoint Create(string name, string machine) => + new(name, machine, DeterministicGuid(name + "@" + machine)); + + public static Guid DeterministicGuid(string value) => + new(SHA256.HashData(Encoding.UTF8.GetBytes(value)).AsSpan(0, 16)); +} diff --git a/tools/testing-tool/TestingTool/DirectAuditQueueWriter.cs b/tools/testing-tool/TestingTool/DirectAuditQueueWriter.cs new file mode 100644 index 0000000000..15ffd10339 --- /dev/null +++ b/tools/testing-tool/TestingTool/DirectAuditQueueWriter.cs @@ -0,0 +1,264 @@ +using System.Diagnostics; +using System.Diagnostics.Metrics; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using NServiceBus; +using NServiceBus.Extensibility; +using TestingTool.Auditing; +using TestingTool.Contracts; + +namespace TestingTool; + +/// +/// The audit-queue counterpart of : writes audit envelopes +/// straight to the ServiceControl.Audit instance's audit queue over the transport, so audit +/// ingestion can be loaded without any endpoint actually processing messages. +/// +/// Most of what it emits are processed-message envelopes carrying the headers a real endpoint +/// would have stamped. A configurable share of claims instead emit a whole saga episode: three +/// messages that drove a saga through New, Updated and Completed, each followed by the snapshot it +/// produced, with the headers that link the messages to the saga. +/// +public sealed class DirectAuditQueueWriter +{ + private readonly IMessageSession _session; + private readonly TestingToolMetrics _metrics; + private readonly TestingToolOptions _options; + private readonly AuditMessageFactory _factory; + private readonly ILogger _logger; + + private readonly ActivitySource _activitySource = new(TelemetrySetup.Sources.AuditBypass); + private readonly Counter _auditCounter; + + private AuditRun? _run; + private long _messagesWritten; + private long _snapshotsWritten; + private long _messagesFailed; + + public bool IsRunning => Volatile.Read(ref _run) is not null; + public long MessagesWritten => Interlocked.Read(ref _messagesWritten); + public long SnapshotsWritten => Interlocked.Read(ref _snapshotsWritten); + public long MessagesFailed => Interlocked.Read(ref _messagesFailed); + + public DirectAuditQueueWriter( + IMessageSession session, + TestingToolMetrics metrics, + Meter meter, + IOptions options, + AuditMessageFactory factory, + ILogger logger) + { + _session = session; + _metrics = metrics; + _options = options.Value; + _factory = factory; + _logger = logger; + _auditCounter = meter.CreateCounter("bypass_audit_messages_written_total"); + } + + /// Starts writing audit envelopes directly to the audit queue. + /// + /// Saga snapshots per processed message. Null falls back to the configured default. + /// + public bool TryStart(double rate, TimeSpan? duration, int? parallelism, double? sagaSnapshotRatio, out string? error) + { + if (rate <= 0) + { + error = "Rate must be greater than 0"; + return false; + } + + var ratio = sagaSnapshotRatio ?? _options.AuditSagaSnapshotRatio; + if (ratio is < 0 or > 1) + { + error = "Saga snapshot ratio must be between 0 and 1"; + return false; + } + + var workerCount = parallelism is { } p and > 0 ? p : Environment.ProcessorCount; + + var run = new AuditRun(rate, workerCount, duration, ratio); + + // Read the token and build the worker array before publishing the run. Publishing first + // leaves a window where Stop() takes the run, disposes the CTS and returns while this + // method is still about to read Cts.Token, which then throws ObjectDisposedException out + // of the API handler and leaves the writer idle with a bare 500. + var token = run.Cts.Token; + var loops = new Task[workerCount]; + run.Loops = loops; + + if (Interlocked.CompareExchange(ref _run, run, null) is not null) + { + run.Cts.Dispose(); + error = "Audit-queue writer is already running — stop it first"; + return false; + } + + for (var i = 0; i < workerCount; i++) + { + var workerIndex = i; + loops[i] = Task.Run(() => WriteLoop(run, workerIndex, token)); + } + Task.WhenAll(loops).ContinueWith(t => OnRunCompleted(run, t), TaskScheduler.Default); + + _logger.LogInformation( + "Started audit-queue writer at {Rate:F1} claims/s{Duration} across {Workers} workers, saga episode every {Period} claims", + rate, duration is null ? "" : $" for {duration.Value}", workerCount, + run.SagaEpisodePeriod == 0 ? "never" : run.SagaEpisodePeriod.ToString()); + + error = null; + return true; + } + + /// Stops the audit-queue writer. + public void Stop() + { + var run = Interlocked.Exchange(ref _run, null); + if (run is null) return; + + run.Cts.Cancel(); + + try + { + if (run.Loops is { } loops) + Task.WaitAll(loops, TimeSpan.FromSeconds(5)); + } + catch + { + // Loops were cancelled or timed out — expected during stop. + } + + run.Cts.Dispose(); + + _logger.LogInformation("Stopped audit-queue writer: {Messages} messages, {Snapshots} saga snapshots, {Failed} failed in total", + MessagesWritten, SnapshotsWritten, MessagesFailed); + } + + private void OnRunCompleted(AuditRun run, Task workers) + { + if (workers.Exception is { } ex) + { + _logger.LogError(ex.Flatten(), "Audit-queue writer crashed"); + } + + if (Interlocked.CompareExchange(ref _run, null, run) == run) + { + run.Cts.Dispose(); + _logger.LogInformation("Audit-queue writer finished: {Written} written this run, {Failed} failed in total", + run.Written, MessagesFailed); + } + } + + /// Returns the current audit writer status for API consumers. + public AuditBypassStatus GetStatus() + { + var run = Volatile.Read(ref _run); + return new() + { + Running = run is not null, + Rate = run?.Rate ?? 0, + AchievedRate = Math.Round(run?.AchievedRate ?? 0, 1), + SagaSnapshotRatio = run?.SagaSnapshotRatio ?? 0, + ProcessedMessagesWritten = MessagesWritten, + SagaSnapshotsWritten = SnapshotsWritten, + MessagesFailed = MessagesFailed, + StartedAt = run?.StartedAt.ToString("O") + }; + } + + private async Task WriteLoop(AuditRun run, int workerIndex, CancellationToken ct) + { + using var timer = new PeriodicTimer(run.WorkerTick); + + try + { + while (await timer.WaitForNextTickAsync(ct)) + { + while (!ct.IsCancellationRequested && run.TryClaim(out var seq)) + { + var isSagaEpisode = run.SagaEpisodePeriod > 0 && seq % run.SagaEpisodePeriod == 0; + + // A saga episode is written whole by this one worker so its New, Updated and + // Completed phases reach the queue in order. + var payloads = isSagaEpisode + ? _factory.CreateSagaEpisode(run.Salt, seq / run.SagaEpisodePeriod, seq) + : [_factory.CreateProcessedMessage(run.Salt, seq)]; + + foreach (var payload in payloads) + { + if (ct.IsCancellationRequested) break; + await Write(run, payload, seq, workerIndex, ct); + } + } + } + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) { } + } + + private async Task Write(AuditRun run, RawAuditPayload payload, long sequence, int workerIndex, CancellationToken ct) + { + var isSnapshot = payload.Headers[Headers.EnclosedMessageTypes] == AuditMessageFactory.SagaUpdatedMessageType; + var kind = isSnapshot ? SagaSnapshotKind : ProcessedMessageKind; + + var sendOptions = new SendOptions(); + sendOptions.SetDestination(_options.AuditQueueName); + sendOptions.GetExtensions().Set(payload); + + using var activity = _activitySource.StartActivity("audit-write"); + activity?.SetTag("sequence", sequence); + activity?.SetTag("audit.kind", kind); + + try + { + await _session.Send(new AuditEnvelope { Sequence = sequence }, sendOptions, ct); + + run.IncrementWritten(); + if (isSnapshot) + { + Interlocked.Increment(ref _snapshotsWritten); + _metrics.AddAuditSagaSnapshotsWritten(1); + } + else + { + Interlocked.Increment(ref _messagesWritten); + _metrics.AddAuditMessagesWritten(1); + } + _auditCounter.Add(1, new KeyValuePair("kind", kind)); + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + // Our own stop or duration expiry — let it unwind the worker. + throw; + } + catch (Exception ex) + { + // A transport timeout surfaces as TaskCanceledException. Unfiltered, it would be + // mistaken for a stop and silently retire the worker with nothing counted or logged. + Interlocked.Increment(ref _messagesFailed); + _metrics.AddAuditMessagesFailed(1); + _logger.LogWarning(ex, "Audit-queue send failed on worker {Worker} seq {Seq} kind {Kind}", workerIndex, sequence, kind); + } + } + + private const string ProcessedMessageKind = "processed-message"; + private const string SagaSnapshotKind = "saga-snapshot"; +} + +/// Adds the saga mix to the shared run state. +internal sealed class AuditRun : LoadRun +{ + public AuditRun(double rate, int workerCount, TimeSpan? duration, double sagaSnapshotRatio) + : base(rate, workerCount, duration) + { + SagaSnapshotRatio = sagaSnapshotRatio; + SagaEpisodePeriod = AuditMessageFactory.SagaEpisodePeriod(sagaSnapshotRatio); + } + + public double SagaSnapshotRatio { get; } + + /// Distinguishes one run's generated ids from the next, since claims restart at 1. + public long Salt => StartedAt.Ticks; + + /// Claims between saga episodes, or 0 when snapshots are disabled. + public int SagaEpisodePeriod { get; } +} diff --git a/tools/testing-tool/TestingTool/DirectErrorQueueWriter.cs b/tools/testing-tool/TestingTool/DirectErrorQueueWriter.cs index 37d0278394..9a15884619 100644 --- a/tools/testing-tool/TestingTool/DirectErrorQueueWriter.cs +++ b/tools/testing-tool/TestingTool/DirectErrorQueueWriter.cs @@ -32,11 +32,6 @@ public sealed class DirectErrorQueueWriter private readonly ActivitySource _activitySource = new(TelemetrySetup.Sources.Bypass); private readonly Counter _bypassCounter; - // Timer tick bounds for each worker. The floor keeps PeriodicTimer valid (it rejects - // sub-millisecond periods) and avoids spinning; the ceiling keeps low rates responsive. - private static readonly TimeSpan MinTick = TimeSpan.FromMilliseconds(10); - private static readonly TimeSpan MaxTick = TimeSpan.FromSeconds(1); - // The current run, or null when idle. Swapped atomically so Stop() and a run ending on its // own (duration elapsed or crash) can't both tear it down. private BypassRun? _run; @@ -185,8 +180,7 @@ public BypassStatus GetStatus() private async Task WriteLoop(BypassRun run, int workerIndex, CancellationToken ct) { var scenario = run.Scenario; - var tick = TimeSpan.FromSeconds(Math.Clamp(run.WorkerCount / run.Rate, MinTick.TotalSeconds, MaxTick.TotalSeconds)); - using var timer = new PeriodicTimer(tick); + using var timer = new PeriodicTimer(run.WorkerTick); // Pre-compute the failure metadata from the scenario so all messages in this run // share the same exception type and stack trace — ServiceControl groups on type + first @@ -261,64 +255,9 @@ private async Task WriteLoop(BypassRun run, int workerIndex, CancellationToken c catch (OperationCanceledException) { } } } -/// State for one bypass run: target rate, shared send budget, cancellation and workers. -internal sealed class BypassRun +/// Adds the scenario a bypass run is generating failures for to the shared run state. +internal sealed class BypassRun(IScenario scenario, double rate, int workerCount, TimeSpan? duration) + : LoadRun(rate, workerCount, duration) { - private readonly Stopwatch _clock = Stopwatch.StartNew(); - private long _claimed; - private long _written; - - public BypassRun(IScenario scenario, double rate, int workerCount, TimeSpan? duration) - { - Scenario = scenario; - Rate = rate; - WorkerCount = workerCount; - StartedAt = DateTimeOffset.UtcNow; - Cts = duration is { } d - ? new CancellationTokenSource(d) - : new CancellationTokenSource(); - } - - public IScenario Scenario { get; } - public double Rate { get; } - public int WorkerCount { get; } - public DateTimeOffset StartedAt { get; } - public CancellationTokenSource Cts { get; } - public Task[]? Loops { get; set; } - public long Written => Interlocked.Read(ref _written); - - /// Envelopes actually written per second since start, for comparison with . - public double AchievedRate - { - get - { - var elapsed = _clock.Elapsed.TotalSeconds; - return elapsed > 0 ? Written / elapsed : 0; - } - } - - public void IncrementWritten() => Interlocked.Increment(ref _written); - - /// - /// Claims the next sequence number if one is owed (floor(elapsed * rate) not yet - /// reached). Shared by all workers so whichever is free picks up the backlog. - /// - public bool TryClaim(out long sequence) - { - var due = (long)(_clock.Elapsed.TotalSeconds * Rate); - while (true) - { - var claimed = Interlocked.Read(ref _claimed); - if (claimed >= due) - { - sequence = 0; - return false; - } - if (Interlocked.CompareExchange(ref _claimed, claimed + 1, claimed) == claimed) - { - sequence = claimed + 1; - return true; - } - } - } + public IScenario Scenario { get; } = scenario; } diff --git a/tools/testing-tool/TestingTool/FailingMessageHandler.cs b/tools/testing-tool/TestingTool/FailingMessageHandler.cs index 1e550f79d5..c156001824 100644 --- a/tools/testing-tool/TestingTool/FailingMessageHandler.cs +++ b/tools/testing-tool/TestingTool/FailingMessageHandler.cs @@ -1,4 +1,5 @@ using System.Diagnostics; +using System.Diagnostics.Metrics; using System.Runtime.ExceptionServices; using Microsoft.Extensions.Logging; using NServiceBus; @@ -11,9 +12,22 @@ namespace TestingTool; /// ShouldFail returns true, the handler throws — NServiceBus routes the failed message to /// the configured error queue (ServiceControl). The ScenarioName header selects the scenario. /// -public sealed class FailingMessageHandler(IScenarioRegistry registry, ILogger logger) - : IHandleMessages +public sealed class FailingMessageHandler : IHandleMessages { + private readonly IScenarioRegistry registry; + private readonly ILogger logger; + + // Every message this handler completes is forwarded to the audit queue by + // AuditProcessedMessagesTo, so it is audit load the direct writer's counter cannot see. + private readonly Counter endpointAudited; + + public FailingMessageHandler(IScenarioRegistry registry, Meter meter, ILogger logger) + { + this.registry = registry; + this.logger = logger; + endpointAudited = meter.CreateCounter("endpoint_audited_messages_total"); + } + public Task Handle(LoadMessage message, IMessageHandlerContext context) { var scenarioName = context.MessageHeaders.GetValueOrDefault("TestingTool.Scenario") ?? "unknown"; @@ -22,6 +36,7 @@ public Task Handle(LoadMessage message, IMessageHandlerContext context) if (scenario is null) { logger.LogDebug("No scenario '{Scenario}' registered — message {Seq} succeeds", scenarioName, message.Sequence); + endpointAudited.Add(1); return Task.CompletedTask; } @@ -41,6 +56,7 @@ public Task Handle(LoadMessage message, IMessageHandlerContext context) } activity?.SetTag("result", "success"); + endpointAudited.Add(1, new KeyValuePair("scenario", scenario.Name)); return Task.CompletedTask; } } \ No newline at end of file diff --git a/tools/testing-tool/TestingTool/LoadRun.cs b/tools/testing-tool/TestingTool/LoadRun.cs new file mode 100644 index 0000000000..631b7b252b --- /dev/null +++ b/tools/testing-tool/TestingTool/LoadRun.cs @@ -0,0 +1,75 @@ +using System.Diagnostics; + +namespace TestingTool; + +/// +/// State for one direct-write run: target rate, shared send budget, cancellation and workers. +/// Shared by the error and audit queue writers so both pace identically. +/// +internal class LoadRun +{ + private readonly Stopwatch _clock = Stopwatch.StartNew(); + private long _claimed; + private long _written; + + protected LoadRun(double rate, int workerCount, TimeSpan? duration) + { + Rate = rate; + WorkerCount = workerCount; + StartedAt = DateTimeOffset.UtcNow; + Cts = duration is { } d + ? new CancellationTokenSource(d) + : new CancellationTokenSource(); + } + + public double Rate { get; } + public int WorkerCount { get; } + public DateTimeOffset StartedAt { get; } + public CancellationTokenSource Cts { get; } + public Task[]? Loops { get; set; } + public long Written => Interlocked.Read(ref _written); + + /// Envelopes actually written per second since start, for comparison with . + public double AchievedRate + { + get + { + var elapsed = _clock.Elapsed.TotalSeconds; + return elapsed > 0 ? Written / elapsed : 0; + } + } + + public void IncrementWritten() => Interlocked.Increment(ref _written); + + /// + /// Claims the next sequence number if one is owed (floor(elapsed * rate) not yet + /// reached). Shared by all workers so whichever is free picks up the backlog. + /// + public bool TryClaim(out long sequence) + { + var due = (long)(_clock.Elapsed.TotalSeconds * Rate); + while (true) + { + var claimed = Interlocked.Read(ref _claimed); + if (claimed >= due) + { + sequence = 0; + return false; + } + if (Interlocked.CompareExchange(ref _claimed, claimed + 1, claimed) == claimed) + { + sequence = claimed + 1; + return true; + } + } + } + + /// + /// The tick a worker should wait between catch-up passes. The floor keeps PeriodicTimer valid + /// (it rejects sub-millisecond periods) and avoids spinning; the ceiling keeps low rates responsive. + /// + public TimeSpan WorkerTick => TimeSpan.FromSeconds(Math.Clamp(WorkerCount / Rate, MinTick.TotalSeconds, MaxTick.TotalSeconds)); + + private static readonly TimeSpan MinTick = TimeSpan.FromMilliseconds(10); + private static readonly TimeSpan MaxTick = TimeSpan.FromSeconds(1); +} diff --git a/tools/testing-tool/TestingTool/NServiceBusSetup.cs b/tools/testing-tool/TestingTool/NServiceBusSetup.cs index bf36152034..f118655085 100644 --- a/tools/testing-tool/TestingTool/NServiceBusSetup.cs +++ b/tools/testing-tool/TestingTool/NServiceBusSetup.cs @@ -1,6 +1,7 @@ using Microsoft.Data.SqlClient; using Microsoft.Extensions.Options; using NServiceBus; +using TestingTool.Auditing; namespace TestingTool; @@ -62,6 +63,10 @@ public static IServiceCollection AddTestingToolEndpoint(this IServiceCollection // Route failures to the ServiceControl error queue. config.SendFailedMessagesTo(options.ErrorQueueName); config.AuditProcessedMessagesTo(options.AuditQueueName); + + // Lets the direct audit-queue writer replace the wire message after serialization, which is + // the only stage late enough to control NServiceBus.EnclosedMessageTypes and the timestamps. + config.Pipeline.Register(new RawAuditPayloadBehavior(), "Applies raw audit envelopes from the direct audit-queue writer"); config.SendHeartbeatTo(options.ServiceControlInputQueue); // enable metrics diff --git a/tools/testing-tool/TestingTool/Program.cs b/tools/testing-tool/TestingTool/Program.cs index b7fb0e3835..1f93e62640 100644 --- a/tools/testing-tool/TestingTool/Program.cs +++ b/tools/testing-tool/TestingTool/Program.cs @@ -1,5 +1,6 @@ using System.Diagnostics.Metrics; using TestingTool; +using TestingTool.Auditing; using TestingTool.Contracts; using TestingTool.Jobs; using TestingTool.Scenarios; @@ -34,6 +35,8 @@ builder.Services.AddSingleton(); builder.Services.AddSingleton(); builder.Services.AddSingleton(); +builder.Services.AddSingleton(_ => new AuditMessageFactory(shardId)); +builder.Services.AddSingleton(); builder.Services.AddSingleton(); // --- Recoverability/search jobs (Phase 4) --- @@ -84,6 +87,7 @@ runner.StopAll(); jobRunner.StopAll(); app.Services.GetRequiredService().Stop(); + app.Services.GetRequiredService().Stop(); }); // --- Health endpoints (Phase 6) --- @@ -110,6 +114,9 @@ RetentionSweepsNotSupported = metrics.TotalRetentionSweepsNotSupported, BypassErrorsWritten = metrics.TotalBypassErrorsWritten, BypassErrorsFailed = metrics.TotalBypassErrorsFailed, + AuditMessagesWritten = metrics.TotalAuditMessagesWritten, + AuditSagaSnapshotsWritten = metrics.TotalAuditSagaSnapshotsWritten, + AuditMessagesFailed = metrics.TotalAuditMessagesFailed, CustomCheckFailures = metrics.TotalCustomCheckFailures, ShardId = shardId, ActiveScenarios = metrics.ActiveScenarios, @@ -212,6 +219,31 @@ return Results.Ok(writer.GetStatus()); }); +// --- Audit bypass endpoints (direct audit-queue writer) --- +// Writes audit envelopes straight to the ServiceControl.Audit instance's audit queue, including +// whole saga episodes: the messages that drove a saga plus the snapshots they produced. + +app.MapGet("/api/audit-bypass/status", (DirectAuditQueueWriter writer) => Results.Ok(writer.GetStatus())); + +app.MapPost("/api/audit-bypass/start", (StartAuditBypassRequest? request, DirectAuditQueueWriter writer) => +{ + var rate = request?.Rate ?? 100; + var duration = request?.DurationSeconds is { } secs and > 0 + ? TimeSpan.FromSeconds(secs) + : (TimeSpan?)null; + + if (!writer.TryStart(rate, duration, request?.Parallelism, request?.SagaSnapshotRatio, out var error)) + return Results.BadRequest(new { error }); + + return Results.Ok(writer.GetStatus()); +}); + +app.MapPost("/api/audit-bypass/stop", (DirectAuditQueueWriter writer) => +{ + writer.Stop(); + return Results.Ok(writer.GetStatus()); +}); + // --- Release-test scenario endpoints (Phase 5: release-test presets) --- // Lists release-test presets that map to testing-tool scenarios, and allows kicking them off // by release-test name. This satisfies the optional requirement to consider release-test diff --git a/tools/testing-tool/TestingTool/TelemetrySetup.cs b/tools/testing-tool/TestingTool/TelemetrySetup.cs index 292a84c522..9b051d2a9d 100644 --- a/tools/testing-tool/TestingTool/TelemetrySetup.cs +++ b/tools/testing-tool/TestingTool/TelemetrySetup.cs @@ -24,6 +24,7 @@ public static class Sources public const string Replay = "testing-tool.replay"; public const string Search = "testing-tool.search"; public const string Bypass = "testing-tool.bypass"; + public const string AuditBypass = "testing-tool.audit-bypass"; } public static Meter CreateMeter() => new(MeterName, "1.0.0"); @@ -40,6 +41,7 @@ public static OpenTelemetryBuilder AddTestingToolTelemetry(this IServiceCollecti .AddSource(Sources.Replay) .AddSource(Sources.Search) .AddSource(Sources.Bypass) + .AddSource(Sources.AuditBypass) // Also pick up per-scenario activity sources dynamically. .AddSource("testing-tool.*") .AddOtlpExporter()) diff --git a/tools/testing-tool/TestingTool/TestingTool.csproj b/tools/testing-tool/TestingTool/TestingTool.csproj index 27ebd7da6c..5761c4c557 100644 --- a/tools/testing-tool/TestingTool/TestingTool.csproj +++ b/tools/testing-tool/TestingTool/TestingTool.csproj @@ -25,4 +25,8 @@ + + + + \ No newline at end of file diff --git a/tools/testing-tool/TestingTool/TestingToolMetrics.cs b/tools/testing-tool/TestingTool/TestingToolMetrics.cs index acd5310393..0d39ced818 100644 --- a/tools/testing-tool/TestingTool/TestingToolMetrics.cs +++ b/tools/testing-tool/TestingTool/TestingToolMetrics.cs @@ -16,6 +16,9 @@ public sealed class TestingToolMetrics private long _totalRetentionSweepsNotSupported; private long _totalBypassErrorsWritten; private long _totalBypassErrorsFailed; + private long _totalAuditMessagesWritten; + private long _totalAuditSagaSnapshotsWritten; + private long _totalAuditMessagesFailed; private long _totalCustomCheckFailures; private long _activeScenarios; private double _currentRate; @@ -29,6 +32,9 @@ public sealed class TestingToolMetrics public long TotalRetentionSweepsNotSupported => Interlocked.Read(ref _totalRetentionSweepsNotSupported); public long TotalBypassErrorsWritten => Interlocked.Read(ref _totalBypassErrorsWritten); public long TotalBypassErrorsFailed => Interlocked.Read(ref _totalBypassErrorsFailed); + public long TotalAuditMessagesWritten => Interlocked.Read(ref _totalAuditMessagesWritten); + public long TotalAuditSagaSnapshotsWritten => Interlocked.Read(ref _totalAuditSagaSnapshotsWritten); + public long TotalAuditMessagesFailed => Interlocked.Read(ref _totalAuditMessagesFailed); public long TotalCustomCheckFailures => Interlocked.Read(ref _totalCustomCheckFailures); public int ActiveScenarios => (int)Interlocked.Read(ref _activeScenarios); public double CurrentRate => _currentRate; @@ -42,6 +48,9 @@ public sealed class TestingToolMetrics public void AddRetentionSweepsNotSupported(long count) => Interlocked.Add(ref _totalRetentionSweepsNotSupported, count); public void AddBypassErrorsWritten(long count) => Interlocked.Add(ref _totalBypassErrorsWritten, count); public void AddBypassErrorsFailed(long count) => Interlocked.Add(ref _totalBypassErrorsFailed, count); + public void AddAuditMessagesWritten(long count) => Interlocked.Add(ref _totalAuditMessagesWritten, count); + public void AddAuditSagaSnapshotsWritten(long count) => Interlocked.Add(ref _totalAuditSagaSnapshotsWritten, count); + public void AddAuditMessagesFailed(long count) => Interlocked.Add(ref _totalAuditMessagesFailed, count); public void AddCustomCheckFailures(long count) => Interlocked.Add(ref _totalCustomCheckFailures, count); public void SetActiveScenarios(int count) => Interlocked.Exchange(ref _activeScenarios, count); public void SetCurrentRate(double rate) => _currentRate = rate; diff --git a/tools/testing-tool/TestingTool/TestingToolOptions.cs b/tools/testing-tool/TestingTool/TestingToolOptions.cs index b3939ff59e..18402e953b 100644 --- a/tools/testing-tool/TestingTool/TestingToolOptions.cs +++ b/tools/testing-tool/TestingTool/TestingToolOptions.cs @@ -67,8 +67,18 @@ public sealed class TestingToolOptions /// public double CustomCheckFailureProbability { get; set; } = 0.4; + /// + /// Audit queue the ServiceControl.Audit instance drains. EF persistence lives in that instance, + /// not in the primary, so this is the audit container's queue. + /// public string AuditQueueName { get; set; } = "audit"; + /// + /// Default saga snapshots per processed message for the direct audit-queue writer. The default + /// of 0.2 emits two saga snapshots for every ten processed messages. + /// + public double AuditSagaSnapshotRatio { get; set; } = 0.2; + /// ServiceControl monitoring instance input queue that endpoint metrics are sent to. public string MonitoringQueueName { get; set; } = "Particular.Monitoring"; } \ No newline at end of file diff --git a/tools/testing-tool/TestingTool/appsettings.json b/tools/testing-tool/TestingTool/appsettings.json index ec7bf535b9..d405b583de 100644 --- a/tools/testing-tool/TestingTool/appsettings.json +++ b/tools/testing-tool/TestingTool/appsettings.json @@ -19,7 +19,9 @@ "CustomCheckInterval": "00:00:30", "CustomCheckHost": "ServiceControl", "CustomCheckFailureProbability": 0.4, - "AutoStartBackgroundNoise": false + "AutoStartBackgroundNoise": false, + "AuditQueueName": "audit", + "AuditSagaSnapshotRatio": 0.2 }, "OTEL_EXPORTER_OTLP_ENDPOINT": "http://localhost:4317" -} \ No newline at end of file +} diff --git a/tools/testing-tool/TestingTool/wwwroot/index.html b/tools/testing-tool/TestingTool/wwwroot/index.html index 58aa011447..0aa7f9ac50 100644 --- a/tools/testing-tool/TestingTool/wwwroot/index.html +++ b/tools/testing-tool/TestingTool/wwwroot/index.html @@ -194,7 +194,7 @@

🎯 ServiceControl Testing Tool

- +
@@ -255,6 +255,37 @@

Direct Error-Queue Bypass

+
+

Direct Audit-Queue Writer

+
+
+
+ Writes audit envelopes directly to the ServiceControl.Audit instance's audit queue. Most carry the headers a real + endpoint would have stamped, so the audit instance records endpoints, message types, conversations and processing + times. A share of them are whole saga episodes: the three messages that drove a saga through New, Updated and + Completed, each followed by the snapshot it produced, linked by the InvokedSagas headers. Needs an audit instance + on SQL Server or PostgreSQL (--audit-instances with --persistence). +
+
+ Rate (claims/s): + + Workers: + + Saga snapshots per message: + + Duration (sec, optional): + +
+
+ + +
+
+ Status: idle + Messages written: 0 +
+
+