Add Query Store total-CPU/total-duration compose measures, closes #2732

CHANGELOG.md

@@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Added
+- **Query Store gets total-CPU and total-duration compose measures** ([#2732]) - the compose catalog's Query Store roster could rank by executions, a per-interval peak, or a per-execution average - never by what a query group actually CONSUMED, so "what burned the most CPU over this window" was not expressible against the box's headline dataset (a per-execution average ranks a query that ran once at 30s over one that burned an hour of CPU at 200ms a call). `qs_total_cpu_us` / `qs_total_duration_us` fill the hole as a new weighted-sum measure shape (`MeasureRatioMode.WeightedSum`): the existing Weighted ratio's numerator with no denominator - `SUM(avg * execution_count)` on the raw route, since avg times executions IS each interval's total, and the corrected rollups' pre-multiplied `cpu_us_weighted_sum` / `duration_us_weighted_sum` columns read directly on the CAGG routes. Pure catalog/expression plumbing - no migration, no collector change, no backfill - and time-series, ranked (top query_hash by total CPU), and scalar modes all work for free.
 - **Alert deliveries carry a linked triage page** ([#2710]) - option 2 of the issue's sketch: a purpose-built, read-only `GET /api/triage` endpoint in the Darling web API assembles a per-firing triage page ON READ - the matching `config_alert_log` row(s), the recent collection log, and alert-type-relevant collected data (deadlock/blocking detail for those alerts, CPU trend + top queries for High CPU, the PostgreSQL reads for the PG alert family, a summary fallback for unmapped metrics) - every section running through the SAME `/api/read/{tool}` dispatch the dashboard already serves, so there is zero SQL/projection drift and nothing is provisioned per alert (no `config.custom_views` row, nothing to GC; a link older than retention degrades to honest empties, and a missing server/metric/history row degrades to notes on the page, never a 500). The SPA renders it at `#/triage?server=...&metric=...&at=...&dedup=...` with derive.js-inferred field configs, and the Alert History page links each row to it. All four webhook channels carry the link when the new `web.publicBaseUrl` (darling.json, file-authoritative - deployment plumbing, not a store knob) is set: Slack gets a link button, Teams a `potentialAction` OpenUri, PagerDuty the Events v2 `links` array plus a `custom_details` Triage row, and the generic template a `{{triage_url}}` token. Unset (the default, and always on Lite, which serves no web page) every payload renders byte-identical to before - a link can never be the reason an alert fails to deliver.
 - **Postgres targets get the Poison Wait alert** ([#2711]) - SQL Server's Poison Wait fires on THREADPOOL/RESOURCE_SEMAPHORE waits because they are near-zero in healthy operation, so any sustained accumulation is inherently abnormal. The Postgres analogue now watches the pair with the same trait from the issue's own fleet research - IPC:BtreePage (backends queueing on a hot B-tree index page) and IPC:BufferIo (backends stacked behind one another's in-flight page reads): exactly zero on 4 of 5 production servers checked, and a structural inter-backend contention signal no other alert covers on the fifth. Lock/Relation, the other researched candidate, was rejected - a relation-level lock wait IS blocking, and the Blocking alert already owns that ground. The threshold shape is deliberately NOT the SQL Server avg-ms-per-wait bar: these events average 1-2 ms per wait at six-figure volumes, a shape a per-wait average can never see, so the alert fires on TOTAL accumulated wait time - Warning when a poison event accrues 600 seconds of wait inside 10 minutes (an average of one backend continuously stuck, ~160x the worst healthy baseline observed anywhere on the fleet), Critical at ten backends, judged per event so one event's cooldown never eats the other's. Honors the shared PoisonWaitEnabled switch both engines already use, refuses to re-fire on a store row it has already reported, and seeds its cooldown from alert history so a service restart cannot re-fire on a window the previous process already covered.
 - **PostgreSQL/Aurora targets get a High CPU alert, backed by a new instance-CPU collector** ([#2719]) - PostgreSQL has never had an instance-level CPU signal: `pg_stat_kcache` measures per-QUERY kernel time, not "is this instance saturated right now", and nothing else in core PostgreSQL answers that question. The new `pg_cpu_utilization` collector reads AWS RDS Performance Insights' `os.cpuUtilization.total.avg` OS counter over the RDS/PI API rather than a database connection - chosen over CloudWatch's `AWS/RDS`/`CPUUtilization` after live testing showed CloudWatch's figure reading capacity-relative and roomy on Serverless v2 (~6.8%) against PI's true OS-level utilization on the identical window (~16.8%); PI is universally enabled across the fleet already. Aurora and RDS only - self-hosted PostgreSQL has no equivalent route, so the collector no-ops there rather than fake a signal. The alert reuses the SAME `CpuEnabled`/`CpuThresholdPercent` settings SQL Server's High CPU already reads, so one threshold means the same thing on both engines.
@@ -3067,3 +3068,4 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 [#2710]: https://github.com/erikdarlingdata/PerformanceMonitor/issues/2710
 [#2711]: https://github.com/erikdarlingdata/PerformanceMonitor/issues/2711
 [#2719]: https://github.com/erikdarlingdata/PerformanceMonitor/issues/2719
+[#2732]: https://github.com/erikdarlingdata/PerformanceMonitor/issues/2732

Darling/Darling.Tests/ComposeCaggValueMapperTests.cs

@@ -49,6 +49,8 @@ private static PanelPlan Plan(string measureKey, ComposeAggregate aggregate = Co
     [InlineData("qs_executions", ComposeAggregate.Avg, true)]             /* QS -> sum / sample_count */
     [InlineData("qs_executions", ComposeAggregate.Max, false)]            /* QS kept no per-row max column */
     [InlineData("qs_avg_duration_us", ComposeAggregate.Sum, true)]        /* QS execution-weighted mean */
+    [InlineData("qs_total_duration_us", ComposeAggregate.Sum, true)]      /* QS total -> *_weighted_sum (#2732) */
+    [InlineData("qs_total_cpu_us", ComposeAggregate.Sum, true)]           /* QS total -> *_weighted_sum (#2732) */
     [InlineData("qs_max_duration_us", ComposeAggregate.Max, true)]        /* QS peak -> _max */
     [InlineData("qs_max_duration_us", ComposeAggregate.Avg, false)]       /* only MAX kept */
     [InlineData("wait_time_ms", ComposeAggregate.Sum, false)]             /* no CAGG */
@@ -137,6 +139,24 @@ public void QueryStore_WeightedRatio_Cpu_UsesCpuWeightedSum()
             ComposeCaggValueMapper.BuildCaggNativeExpr(Measure("qs_avg_cpu_us"), Measure("qs_avg_cpu_us").DefaultTimeAgg));
     }
 
+    [Fact]
+    public void QueryStore_WeightedSum_Cpu_ReadsTheWeightedSumColumn_WithNoDenominator()
+    {
+        /* #2732: the total is the CAGG's pre-multiplied product-sum ITSELF — no execution_count_sum divisor,
+           and no NULLIF (there is no division; SUM over zero rows is already NULL). */
+        Assert.Equal(
+            "CAST(SUM(f.cpu_us_weighted_sum) AS double precision)",
+            ComposeCaggValueMapper.BuildCaggNativeExpr(Measure("qs_total_cpu_us"), Measure("qs_total_cpu_us").DefaultTimeAgg));
+    }
+
+    [Fact]
+    public void QueryStore_WeightedSum_Duration_ReadsTheWeightedSumColumn_WithNoDenominator()
+    {
+        Assert.Equal(
+            "CAST(SUM(f.duration_us_weighted_sum) AS double precision)",
+            ComposeCaggValueMapper.BuildCaggNativeExpr(Measure("qs_total_duration_us"), Measure("qs_total_duration_us").DefaultTimeAgg));
+    }
+
     [Fact]
     public void QueryStore_Executions_Sum_ReadsExecutionCountSum()
     {

Darling/Darling.Tests/DarlingComposeTests.cs

@@ -88,9 +88,9 @@ public void EveryScalarMeasureColumn_IsARealPayloadColumn()
     [Fact]
     public void EveryRatioMeasure_ReferencesRealSameSourceScalars()
     {
-        /* Sum/Avg ratios divide two scalar measures; Weighted ratios instead reference RAW source columns
-           (WeightedValueColumn + WeightColumn), pinned separately by WeightedRatio_Columns_AreRealPayloadColumns. */
-        foreach (var ratio in MeasureCatalog.Measures.Where(m => m.Kind == MeasureKind.Ratio && m.RatioMode != MeasureRatioMode.Weighted))
+        /* Sum/Avg ratios divide two scalar measures; Weighted/WeightedSum ratios instead reference RAW source
+           columns (WeightedValueColumn + WeightColumn), pinned separately by WeightedRatio_Columns_AreRealPayloadColumns. */
+        foreach (var ratio in MeasureCatalog.Measures.Where(m => m.Kind == MeasureKind.Ratio && m.RatioMode is MeasureRatioMode.Sum or MeasureRatioMode.Avg))
         {
             var numerator = MeasureCatalog.Measure(ratio.NumeratorKey);
             var denominator = MeasureCatalog.Measure(ratio.DenominatorKey);
@@ -108,13 +108,14 @@ public void WeightedRatio_Columns_AreRealPayloadColumns()
     {
         var payload = PayloadColumnsByTable();
         var weighted = MeasureCatalog.Measures
-            .Where(m => m.Kind == MeasureKind.Ratio && m.RatioMode == MeasureRatioMode.Weighted).ToList();
+            .Where(m => m.Kind == MeasureKind.Ratio && m.RatioMode is MeasureRatioMode.Weighted or MeasureRatioMode.WeightedSum).ToList();
         Assert.NotEmpty(weighted);
         foreach (var ratio in weighted)
         {
             var columns = payload[ratio.SourceTable];
-            /* A Weighted ratio's two operands are raw columns of its own source (the pre-aggregated average and
-               its execution weight), so both must be real payload columns — the compiler emits them directly. */
+            /* A Weighted/WeightedSum ratio's two operands are raw columns of its own source (the pre-aggregated
+               average and its execution weight), so both must be real payload columns — the compiler emits them
+               directly. */
             Assert.True(ratio.WeightedValueColumn is not null && columns.Contains(ratio.WeightedValueColumn),
                 $"weighted ratio '{ratio.Key}' value column '{ratio.WeightedValueColumn}' is not a payload column of '{ratio.SourceTable}'.");
             Assert.True(ratio.WeightColumn is not null && columns.Contains(ratio.WeightColumn),
@@ -522,6 +523,21 @@ window would route to the corrected daily. */
         Assert.Contains("SUM(f.duration_us_weighted_sum) AS double precision) / NULLIF(SUM(f.execution_count_sum), 0)", compiled.Sql, StringComparison.Ordinal);
     }
 
+    [Fact]
+    public void Compile_OldWindow_QueryStore_WeightedSum_RoutesToCagg_ReadsTheWeightedSumColumnAlone()
+    {
+        /* #2732: the total remaps to the reshaped CAGG's pre-multiplied product-sum DIRECTLY — the corrected
+           rollups materialize SUM(avg * execution_count) as cpu_us_weighted_sum, so the total re-aggregates
+           additively with no denominator and no NULLIF. */
+        var (compiled, error) = CompileAged(
+            "{\"source\":\"query_store_stats\",\"ratio\":\"qs_total_cpu_us\",\"timeBucket\":\"hour\",\"viz\":\"line\"}", daysOld: 10);
+        Assert.True(error is null, error);
+        Assert.Contains("FROM collect.query_store_stats_corrected_hourly AS f", compiled!.Sql, StringComparison.Ordinal);
+        Assert.Contains("CAST(SUM(f.cpu_us_weighted_sum) AS double precision)", compiled.Sql, StringComparison.Ordinal);
+        Assert.DoesNotContain("NULLIF", compiled.Sql, StringComparison.Ordinal);
+        Assert.DoesNotContain("avg_cpu_time_us", compiled.Sql, StringComparison.Ordinal); /* not the raw column */
+    }
+
     [Fact]
     public void Compile_VeryOldWindow_QueryStore_RoutesToDailyCagg()
     {
@@ -1158,6 +1174,22 @@ public void Compile_WeightedRatio_IsProductSumOverNullifSumOfWeight(string ratio
         Assert.Contains("/ 1000.0", sql, StringComparison.Ordinal);
     }
 
+    [Theory]
+    [InlineData("qs_total_duration_us", "avg_duration_us")]
+    [InlineData("qs_total_cpu_us", "avg_cpu_time_us")]
+    public void Compile_WeightedSum_IsProductSum_WithNoDenominator(string measureKey, string valueColumn)
+    {
+        /* #2732: the window TOTAL — the Weighted ratios' numerator alone, SUM(avg * execution_count), since
+           avg * execution_count is each interval's total consumption. No NULLIF anywhere: there is no division. */
+        var sql = Compile(ValidPlan($"{{\"source\":\"query_store_stats\",\"ratio\":\"{measureKey}\",\"timeBucket\":\"hour\",\"viz\":\"line\"}}"));
+        Assert.Contains($"CAST(SUM(f.{valueColumn} * f.execution_count) AS double precision)", sql, StringComparison.Ordinal);
+        Assert.DoesNotContain("NULLIF", sql, StringComparison.Ordinal);
+        Assert.Contains("collect.query_store_stats", sql, StringComparison.Ordinal);
+        Assert.DoesNotContain("config.", sql, StringComparison.Ordinal);
+        /* native µs displayed as the default s — a window total across a store runs to seconds-to-hours. */
+        Assert.Contains("/ 1000000.0", sql, StringComparison.Ordinal);
+    }
+
     [Fact]
     public void Compile_QueryStoreRawRoute_DedupsPerIntervalBeforeAggregating()
     {
@@ -1246,6 +1278,35 @@ public void WeightedQsRatios_ValidateAsRatios_AndRejectMeasureReference()
         }
     }
 
+    [Fact]
+    public void WeightedSumQsTotals_ValidateAsRatios_AndRejectMeasureReference()
+    {
+        /* #2732: the totals ride the Ratio kind (their aggregation is the definition, like a ratio's), so the
+           spec/wire seam needs no new special case — referenced as 'ratio', aggregate fixed, default unit 's'. */
+        foreach (var key in new[] { "qs_total_duration_us", "qs_total_cpu_us" })
+        {
+            var plan = ValidPlan($"{{\"source\":\"query_store_stats\",\"ratio\":\"{key}\",\"timeBucket\":\"hour\",\"viz\":\"line\"}}");
+            Assert.Equal(MeasureKind.Ratio, plan.Measure.Kind);
+            Assert.Equal(MeasureRatioMode.WeightedSum, plan.Measure.RatioMode);
+            Assert.Equal("s", plan.Unit);
+
+            var reason = RejectReason($"{{\"source\":\"query_store_stats\",\"measure\":\"{key}\",\"aggregate\":\"sum\",\"viz\":\"table\"}}");
+            Assert.Contains("reference it as 'ratio'", reason, StringComparison.OrdinalIgnoreCase);
+        }
+    }
+
+    [Fact]
+    public void WeightedSumQsTotals_SupportRankedMode_TopQueryHashByTotalCpu()
+    {
+        /* The panel the issue exists for: "top query_hash by total CPU" — ranked mode, ORDER BY value DESC.
+           A per-execution average would rank a one-shot 30s query over one that burned an hour at 200ms/call. */
+        var sql = Compile(ValidPlan(
+            "{\"source\":\"query_store_stats\",\"ratio\":\"qs_total_cpu_us\",\"topN\":10,\"groupBy\":[\"query_hash\"],\"viz\":\"table\"}"));
+        Assert.Contains("CAST(SUM(f.avg_cpu_time_us * f.execution_count) AS double precision)", sql, StringComparison.Ordinal);
+        Assert.Contains("ORDER BY value DESC", sql, StringComparison.Ordinal);
+        Assert.Contains("LIMIT $", sql, StringComparison.Ordinal);
+    }
+
     /* ─────────────────────────── D3: per-panel thresholds (render-only) ─────────────────────────── */
 
     [Fact]
@@ -2208,6 +2269,77 @@ await LiveStoreCleanup.RunAsync(cs!, bodySucceeded, async (cleanup, cleanupCt) =
         }
     }
 
+    [Fact]
+    public async Task ComposedQueryStoreTotalCpuPanel_RunsOnPostgres_AndSumsTheProductOverDedupedIntervals()
+    {
+        /* #2732: the WeightedSum raw route — SUM(avg_cpu_time_us * execution_count) — must ride the same #1841
+           dedup wrapper as everything else on this source. Every row carries avg_cpu_time_us = 500, so the
+           deduped total is 500 * 41 = 20500 µs; an un-deduped product-sum would read 500 * 53 = 26500. */
+        var cs = ConnectionString;
+        Assert.SkipWhen(string.IsNullOrEmpty(cs), "Set DARLING_TEST_PG to a Postgres connection string to run the live compose Query Store test.");
+
+        using var connection = new NpgsqlConnection(cs);
+        await connection.OpenAsync(TestContext.Current.CancellationToken);
+        await PgMigrations.MigrateAsync(connection, TestContext.Current.CancellationToken);
+        await DeleteAsync(connection, TestContext.Current.CancellationToken);
+
+        var end = new DateTime(DateTime.UtcNow.Ticks - (DateTime.UtcNow.Ticks % TimeSpan.TicksPerSecond), DateTimeKind.Utc);
+        var bucket = end.AddHours(-2);
+        var firstExecA = bucket.AddMinutes(1);
+        var firstExecB = bucket.AddMinutes(2);
+
+        var bodySucceeded = false;
+        try
+        {
+            /* The dedup test's fixture: interval A re-collected three times flat at 1, interval B twice while
+               it grew 10 -> 40. Deduped executions = 41. */
+            foreach (var minute in new[] { 5, 10, 15 })
+            {
+                await InsertAsync(connection, bucket.AddMinutes(minute), queryId: 1, planId: 11, firstExecA, execCount: 1);
+            }
+
+            await InsertAsync(connection, bucket.AddMinutes(5), queryId: 2, planId: 22, firstExecB, execCount: 10);
+            await InsertAsync(connection, bucket.AddMinutes(10), queryId: 2, planId: 22, firstExecB, execCount: 40);
+
+            /* unit 'us' (not the 's' default) so the expected value stays integral. */
+            var (plan, parseError) = ComposeSpec.TryParsePanel(
+                (JsonObject)JsonNode.Parse("{\"source\":\"query_store_stats\",\"ratio\":\"qs_total_cpu_us\",\"unit\":\"us\",\"timeBucket\":\"hour\",\"viz\":\"line\"}")!,
+                []);
+            Assert.True(parseError is null, parseError);
+
+            var (compiled, compileError) = ComposeCompiler.Compile(
+                plan!,
+                new ComposeRunContext([ServerName], end.AddHours(-3), end, ComposeRunContext.NoVariables,
+                    RollupAvailability.All, end, RollupCoverage.Unknown));
+            Assert.True(compileError is null, compileError);
+            Assert.False(compiled!.Route.IsCagg, "the window must stay on the raw route for this test to mean anything");
+
+            await using var command = new NpgsqlCommand(compiled.Sql, connection);
+            foreach (var parameter in compiled.Parameters)
+            {
+                command.Parameters.Add(parameter);
+            }
+
+            var total = 0.0;
+            await using (var reader = await command.ExecuteReaderAsync(TestContext.Current.CancellationToken))
+            {
+                while (await reader.ReadAsync(TestContext.Current.CancellationToken))
+                {
+                    total += reader.GetDouble(reader.GetOrdinal("value"));
+                }
+            }
+
+            Assert.Equal(20500.0, total, 3);
+
+            bodySucceeded = true;
+        }
+        finally
+        {
+            await LiveStoreCleanup.RunAsync(cs!, bodySucceeded, async (cleanup, cleanupCt) =>
+                await DeleteAsync(cleanup, cleanupCt));
+        }
+    }
+
     private static async Task DeleteAsync(NpgsqlConnection connection, CancellationToken ct)
     {
         using var command = new NpgsqlCommand("DELETE FROM collect.query_store_stats WHERE server_id = $1", connection);