From 2af62e3e65b6366faf9362b544fb12239889c7eb Mon Sep 17 00:00:00 2001 From: realtonyyoung <6655045+realtonyyoung@users.noreply.github.com> Date: Fri, 12 Jun 2026 12:58:37 -0400 Subject: [PATCH 1/3] Add caller-side DuckDB CPU attribution metric Adds a KurrentDB.DuckDB meter with a kurrentdb.duckdb.cpu.seconds counter (tags: activity=query|read|commit|checkpoint, source=caller) measuring CPU consumed by DuckDB operations executed on KurrentDB threads, so node CPU can be decomposed into KurrentDB vs DuckDB. ThreadCpuTime reads the calling thread's cumulative CPU time via clock_gettime(CLOCK_THREAD_CPUTIME_ID) on Linux/macOS and GetThreadTimes on Windows. Measurement scopes are ref structs so they cannot span an await, which would invalidate per-thread CPU deltas. Instrumented synchronous DuckDB sections: index commits, user index checkpoints, the index readers, and QueryEngine's prepare/execute section. CPU burned on DuckDB's internal worker threads and during streaming chunk fetches is not visible from the caller side; that share lands with the planned Quack external_threads work as source=workers. Co-Authored-By: Claude Fable 5 --- src/KurrentDB.Core/DuckDB/DuckDBCpuMetrics.cs | 62 ++++++++++++++++++ .../DuckDB/InjectionExtensions.cs | 6 ++ src/KurrentDB.Core/DuckDB/ThreadCpuTime.cs | 63 +++++++++++++++++++ .../Diagnostics/DuckDBCpuMetricsTests.cs | 58 +++++++++++++++++ .../Indexes/DefaultIndexProcessorTests.cs | 4 +- .../DefaultIndexReaderTests/IndexTestBase.cs | 6 +- .../Indexes/Category/CategoryIndexReader.cs | 6 +- .../Indexes/Default/DefaultIndexProcessor.cs | 5 ++ .../Indexes/Default/DefaultIndexReader.cs | 6 +- .../Indexes/EventType/EventTypeIndexReader.cs | 6 +- .../Indexes/SecondaryIndexReaderBase.cs | 31 +++++---- .../Indexes/User/UserIndexEngine.cs | 4 +- .../User/UserIndexEngineSubscription.cs | 7 ++- .../Indexes/User/UserIndexProcessor.cs | 6 ++ .../Indexes/User/UserIndexReader.cs | 10 +-- .../Query/QueryEngine.cs | 26 +++++--- src/KurrentDB/metricsconfig.json | 1 + 17 files changed, 271 insertions(+), 36 deletions(-) create mode 100644 src/KurrentDB.Core/DuckDB/DuckDBCpuMetrics.cs create mode 100644 src/KurrentDB.Core/DuckDB/ThreadCpuTime.cs create mode 100644 src/KurrentDB.SecondaryIndexing.Tests/Diagnostics/DuckDBCpuMetricsTests.cs diff --git a/src/KurrentDB.Core/DuckDB/DuckDBCpuMetrics.cs b/src/KurrentDB.Core/DuckDB/DuckDBCpuMetrics.cs new file mode 100644 index 00000000000..08cc4bd98b9 --- /dev/null +++ b/src/KurrentDB.Core/DuckDB/DuckDBCpuMetrics.cs @@ -0,0 +1,62 @@ +// Copyright (c) Kurrent, Inc and/or licensed to Kurrent, Inc under one or more agreements. +// Kurrent, Inc licenses this file to you under the Kurrent License v1 (see LICENSE.md). + +using System.Collections.Generic; +using System.Diagnostics.Metrics; + +namespace KurrentDB.Core.DuckDB; + +// Attributes CPU consumed by DuckDB operations executed on KurrentDB threads. +// DuckDB also schedules work onto its own native worker threads; that share is not visible +// to caller-side measurement and is reported separately once KurrentDB owns those workers. +public class DuckDBCpuMetrics { + public const string MeterName = "KurrentDB.DuckDB"; + + public static class Activities { + public const string Query = "query"; + public const string Read = "read"; + public const string Commit = "commit"; + public const string Checkpoint = "checkpoint"; + } + + private static readonly KeyValuePair SourceTag = new("source", "caller"); + + private readonly Counter _cpuSeconds; + + public DuckDBCpuMetrics(Meter meter, string serviceName) { + _cpuSeconds = meter.CreateCounter( + $"{serviceName}.duckdb.cpu.seconds", + description: "CPU time consumed by DuckDB operations on KurrentDB threads, in seconds"); + } + + public CpuScope Measure(string activity) => new(this, activity); + + // A ref struct so it cannot live across an await: the CPU delta is only valid when start + // and stop are read on the same thread. + public readonly ref struct CpuScope { + private readonly DuckDBCpuMetrics _metrics; + private readonly string _activity; + private readonly long _startNanoseconds; + + internal CpuScope(DuckDBCpuMetrics metrics, string activity) { + if (!ThreadCpuTime.IsSupported) + return; + + _metrics = metrics; + _activity = activity; + _startNanoseconds = ThreadCpuTime.CurrentNanoseconds; + } + + public void Dispose() { + if (_metrics is null) + return; + + var elapsedNanoseconds = ThreadCpuTime.CurrentNanoseconds - _startNanoseconds; + if (elapsedNanoseconds > 0) + _metrics._cpuSeconds.Add( + elapsedNanoseconds / 1e9, + new KeyValuePair("activity", _activity), + SourceTag); + } + } +} diff --git a/src/KurrentDB.Core/DuckDB/InjectionExtensions.cs b/src/KurrentDB.Core/DuckDB/InjectionExtensions.cs index 2e9228ff705..f9aa6a28d3a 100644 --- a/src/KurrentDB.Core/DuckDB/InjectionExtensions.cs +++ b/src/KurrentDB.Core/DuckDB/InjectionExtensions.cs @@ -2,8 +2,10 @@ // Kurrent, Inc licenses this file to you under the Kurrent License v1 (see LICENSE.md). using System; +using System.Diagnostics.Metrics; using System.Threading.Tasks; using Kurrent.Quack.ConnectionPool; +using KurrentDB.Common.Configuration; using KurrentDB.DuckDB; using Microsoft.AspNetCore.Builder; using Microsoft.AspNetCore.Connections; @@ -21,6 +23,10 @@ public static IServiceCollection AddDuckDb(this IServiceCollection services) { services.AddSingleton(sp => sp.GetRequiredService().Shared); services.AddSingleton(); services.AddSingleton(CreatePoolPerConnectionInterceptor); + services.AddSingleton(static sp => { + var serviceName = sp.GetService()?.ServiceName ?? "kurrentdb"; + return new DuckDBCpuMetrics(new Meter(DuckDBCpuMetrics.MeterName, "1.0.0"), serviceName); + }); return services; } diff --git a/src/KurrentDB.Core/DuckDB/ThreadCpuTime.cs b/src/KurrentDB.Core/DuckDB/ThreadCpuTime.cs new file mode 100644 index 00000000000..d8f07395234 --- /dev/null +++ b/src/KurrentDB.Core/DuckDB/ThreadCpuTime.cs @@ -0,0 +1,63 @@ +// Copyright (c) Kurrent, Inc and/or licensed to Kurrent, Inc under one or more agreements. +// Kurrent, Inc licenses this file to you under the Kurrent License v1 (see LICENSE.md). + +using System; +using System.Runtime.InteropServices; + +namespace KurrentDB.Core.DuckDB; + +// Reads the cumulative CPU time consumed by the calling thread. +// A delta between two readings is only meaningful when both are taken on the same thread. +internal static class ThreadCpuTime { + // CLOCK_THREAD_CPUTIME_ID differs per libc: bits/time.h on Linux, sys/_types/_clockid_t on macOS + private const int ClockThreadCpuTimeIdLinux = 3; + private const int ClockThreadCpuTimeIdMacOS = 16; + + public static readonly bool IsSupported = Detect(); + + public static long CurrentNanoseconds => OperatingSystem.IsWindows() + ? GetWindowsThreadCpuNanoseconds() + : GetPosixThreadCpuNanoseconds(); + + private static long GetPosixThreadCpuNanoseconds() { + var clockId = OperatingSystem.IsLinux() ? ClockThreadCpuTimeIdLinux : ClockThreadCpuTimeIdMacOS; + return clock_gettime(clockId, out var ts) == 0 + ? ts.Seconds * 1_000_000_000L + ts.Nanoseconds + : 0; + } + + private static long GetWindowsThreadCpuNanoseconds() { + // GetThreadTimes reports in 100ns units, accrued at scheduler-quantum granularity. + // Cumulative totals are statistically accurate; very short individual deltas are not. + return GetThreadTimes(GetCurrentThread(), out _, out _, out var kernelTime, out var userTime) + ? (kernelTime + userTime) * 100 + : 0; + } + + private static bool Detect() { + if (!OperatingSystem.IsWindows() && !OperatingSystem.IsLinux() && !OperatingSystem.IsMacOS()) + return false; + + try { + _ = CurrentNanoseconds; + return true; + } catch (Exception e) when (e is DllNotFoundException or EntryPointNotFoundException) { + return false; + } + } + + [StructLayout(LayoutKind.Sequential)] + private struct Timespec { + public nint Seconds; + public nint Nanoseconds; + } + + [DllImport("libc")] + private static extern int clock_gettime(int clockId, out Timespec ts); + + [DllImport("kernel32.dll", ExactSpelling = true)] + private static extern nint GetCurrentThread(); + + [DllImport("kernel32.dll", ExactSpelling = true)] + private static extern bool GetThreadTimes(nint thread, out long creationTime, out long exitTime, out long kernelTime, out long userTime); +} diff --git a/src/KurrentDB.SecondaryIndexing.Tests/Diagnostics/DuckDBCpuMetricsTests.cs b/src/KurrentDB.SecondaryIndexing.Tests/Diagnostics/DuckDBCpuMetricsTests.cs new file mode 100644 index 00000000000..861074d36ef --- /dev/null +++ b/src/KurrentDB.SecondaryIndexing.Tests/Diagnostics/DuckDBCpuMetricsTests.cs @@ -0,0 +1,58 @@ +// Copyright (c) Kurrent, Inc and/or licensed to Kurrent, Inc under one or more agreements. +// Kurrent, Inc licenses this file to you under the Kurrent License v1 (see LICENSE.md). + +using System.Diagnostics; +using System.Diagnostics.Metrics; +using KurrentDB.Core.DuckDB; + +namespace KurrentDB.SecondaryIndexing.Tests.Diagnostics; + +public class DuckDBCpuMetricsTests { + [Fact] + public void busy_scope_records_cpu_seconds_with_activity_and_source_tags() { + using var meter = new Meter("test"); + var metrics = new DuckDBCpuMetrics(meter, "kurrentdb"); + + List<(double Value, KeyValuePair[] Tags)> measurements = []; + using var listener = Listen(meter, measurements); + + using (metrics.Measure(DuckDBCpuMetrics.Activities.Commit)) { + // busy spin long enough to accrue CPU past Windows' quantum-granular thread accounting + var stopwatch = Stopwatch.StartNew(); + while (stopwatch.ElapsedMilliseconds < 100) { + } + } + + var measurement = Assert.Single(measurements); + Assert.InRange(measurement.Value, 0.001, 30); + Assert.Contains(measurement.Tags, t => t is { Key: "activity", Value: "commit" }); + Assert.Contains(measurement.Tags, t => t is { Key: "source", Value: "caller" }); + } + + [Fact] + public void idle_scope_records_at_most_negligible_cpu() { + using var meter = new Meter("test"); + var metrics = new DuckDBCpuMetrics(meter, "kurrentdb"); + + List<(double Value, KeyValuePair[] Tags)> measurements = []; + using var listener = Listen(meter, measurements); + + using (metrics.Measure(DuckDBCpuMetrics.Activities.Read)) { + Thread.Sleep(100); + } + + Assert.True(measurements.Sum(m => m.Value) < 0.05, $"expected near-zero CPU, got {measurements.Sum(m => m.Value)}s"); + } + + private static MeterListener Listen(Meter meter, List<(double, KeyValuePair[])> measurements) { + var listener = new MeterListener(); + listener.InstrumentPublished = (instrument, l) => { + if (instrument.Meter == meter && instrument.Name == "kurrentdb.duckdb.cpu.seconds") + l.EnableMeasurementEvents(instrument); + }; + listener.SetMeasurementEventCallback( + (_, value, tags, _) => measurements.Add((value, tags.ToArray()))); + listener.Start(); + return listener; + } +} diff --git a/src/KurrentDB.SecondaryIndexing.Tests/Indexes/DefaultIndexProcessorTests.cs b/src/KurrentDB.SecondaryIndexing.Tests/Indexes/DefaultIndexProcessorTests.cs index 2d4a5f37561..b5d37a1cb95 100644 --- a/src/KurrentDB.SecondaryIndexing.Tests/Indexes/DefaultIndexProcessorTests.cs +++ b/src/KurrentDB.SecondaryIndexing.Tests/Indexes/DefaultIndexProcessorTests.cs @@ -4,6 +4,7 @@ using DotNext; using Kurrent.Quack; using KurrentDB.Core.Data; +using KurrentDB.Core.DuckDB; using KurrentDB.Core.Index.Hashes; using KurrentDB.Core.Tests.Fakes; using KurrentDB.SecondaryIndexing.Indexes.Default; @@ -211,7 +212,8 @@ public DefaultIndexProcessorTests() { var publisher = new FakePublisher(); - _processor = new(DuckDb, publisher, hasher, new("test"), NullLoggerFactory.Instance); + _processor = new(DuckDb, publisher, hasher, new("test"), NullLoggerFactory.Instance, + new DuckDBCpuMetrics(new("test"), "kurrentdb")); } public override ValueTask DisposeAsync() { diff --git a/src/KurrentDB.SecondaryIndexing.Tests/Indexes/DefaultIndexReaderTests/IndexTestBase.cs b/src/KurrentDB.SecondaryIndexing.Tests/Indexes/DefaultIndexReaderTests/IndexTestBase.cs index 52737ff2b5e..82e365d3a4c 100644 --- a/src/KurrentDB.SecondaryIndexing.Tests/Indexes/DefaultIndexReaderTests/IndexTestBase.cs +++ b/src/KurrentDB.SecondaryIndexing.Tests/Indexes/DefaultIndexReaderTests/IndexTestBase.cs @@ -2,6 +2,7 @@ // Kurrent, Inc licenses this file to you under the Kurrent License v1 (see LICENSE.md). using KurrentDB.Core.Data; +using KurrentDB.Core.DuckDB; using KurrentDB.Core.Index.Hashes; using KurrentDB.Core.Tests.Fakes; using KurrentDB.SecondaryIndexing.Indexes.Default; @@ -22,9 +23,10 @@ protected IndexTestBase() { var hasher = new CompositeHasher(new XXHashUnsafe(), new Murmur3AUnsafe()); var publisher = new FakePublisher(); - _processor = new(DuckDb, publisher, hasher, new("test"), NullLoggerFactory.Instance); + var cpuMetrics = new DuckDBCpuMetrics(new("test"), "kurrentdb"); + _processor = new(DuckDb, publisher, hasher, new("test"), NullLoggerFactory.Instance, cpuMetrics); - Sut = new(DuckDb, _processor, _readIndexStub.ReadIndex); + Sut = new(DuckDb, _processor, _readIndexStub.ReadIndex, cpuMetrics); } protected void IndexEvents(ResolvedEvent[] events, bool shouldCommit) { diff --git a/src/KurrentDB.SecondaryIndexing/Indexes/Category/CategoryIndexReader.cs b/src/KurrentDB.SecondaryIndexing/Indexes/Category/CategoryIndexReader.cs index 4c163873e59..9480d887970 100644 --- a/src/KurrentDB.SecondaryIndexing/Indexes/Category/CategoryIndexReader.cs +++ b/src/KurrentDB.SecondaryIndexing/Indexes/Category/CategoryIndexReader.cs @@ -4,6 +4,7 @@ using Kurrent.Quack; using Kurrent.Quack.ConnectionPool; using KurrentDB.Core.Data; +using KurrentDB.Core.DuckDB; using KurrentDB.Core.Services.Storage.ReaderIndex; using KurrentDB.SecondaryIndexing.Indexes.Default; using KurrentDB.SecondaryIndexing.Storage; @@ -16,8 +17,9 @@ namespace KurrentDB.SecondaryIndexing.Indexes.Category; internal class CategoryIndexReader( DuckDBConnectionPool sharedPool, DefaultIndexProcessor processor, - IReadIndex index) - : SecondaryIndexReaderBase(sharedPool, index) { + IReadIndex index, + DuckDBCpuMetrics cpuMetrics) + : SecondaryIndexReaderBase(sharedPool, index, cpuMetrics) { protected override string GetId(string indexName) => CategoryIndex.TryParseCategoryName(indexName, out var categoryName) ? categoryName diff --git a/src/KurrentDB.SecondaryIndexing/Indexes/Default/DefaultIndexProcessor.cs b/src/KurrentDB.SecondaryIndexing/Indexes/Default/DefaultIndexProcessor.cs index 5c6ff43cd3a..c5457ee966c 100644 --- a/src/KurrentDB.SecondaryIndexing/Indexes/Default/DefaultIndexProcessor.cs +++ b/src/KurrentDB.SecondaryIndexing/Indexes/Default/DefaultIndexProcessor.cs @@ -14,6 +14,7 @@ using KurrentDB.Common.Configuration; using KurrentDB.Core.Bus; using KurrentDB.Core.Data; +using KurrentDB.Core.DuckDB; using KurrentDB.Core.Index.Hashes; using KurrentDB.Core.Messages; using KurrentDB.Core.Services; @@ -33,6 +34,7 @@ internal class DefaultIndexProcessor : Disposable, ISecondaryIndexProcessor { private readonly ILongHasher _hasher; private readonly ILogger _log; private readonly BufferedView _appender; + private readonly DuckDBCpuMetrics _cpuMetrics; private Atomic _lastPosition; public TFPos LastIndexedPosition { @@ -47,10 +49,12 @@ public DefaultIndexProcessor( [FromKeyedServices(SecondaryIndexingConstants.InjectionKey)] Meter meter, ILoggerFactory loggerFactory, + DuckDBCpuMetrics cpuMetrics, MetricsConfiguration? metricsConfiguration = null, TimeProvider? clock = null ) { _connection = db.Open(); + _cpuMetrics = cpuMetrics; _log = loggerFactory.CreateLogger(); _appender = new(_connection, "idx_all", "log_position", DefaultIndexViewName); var serviceName = metricsConfiguration?.ServiceName ?? "kurrentdb"; @@ -153,6 +157,7 @@ public void Commit() { try { using var duration = Tracker.StartCommitDuration(); + using var cpu = _cpuMetrics.Measure(DuckDBCpuMetrics.Activities.Commit); _appender.Flush(); } catch (Exception e) { _log.LogError(e, "Failed to commit records to index at log position {LogPosition}", LastIndexedPosition); diff --git a/src/KurrentDB.SecondaryIndexing/Indexes/Default/DefaultIndexReader.cs b/src/KurrentDB.SecondaryIndexing/Indexes/Default/DefaultIndexReader.cs index 4751c589027..a3481980227 100644 --- a/src/KurrentDB.SecondaryIndexing/Indexes/Default/DefaultIndexReader.cs +++ b/src/KurrentDB.SecondaryIndexing/Indexes/Default/DefaultIndexReader.cs @@ -4,6 +4,7 @@ using Kurrent.Quack; using Kurrent.Quack.ConnectionPool; using KurrentDB.Core.Data; +using KurrentDB.Core.DuckDB; using KurrentDB.Core.Services; using KurrentDB.Core.Services.Storage.ReaderIndex; using KurrentDB.SecondaryIndexing.Storage; @@ -14,8 +15,9 @@ namespace KurrentDB.SecondaryIndexing.Indexes.Default; internal class DefaultIndexReader( DuckDBConnectionPool sharedPool, DefaultIndexProcessor processor, - IReadIndex index -) : SecondaryIndexReaderBase(sharedPool, index) { + IReadIndex index, + DuckDBCpuMetrics cpuMetrics +) : SecondaryIndexReaderBase(sharedPool, index, cpuMetrics) { protected override string GetId(string indexName) => string.Empty; protected override List GetDbRecordsForwards(DuckDBConnectionPool db, diff --git a/src/KurrentDB.SecondaryIndexing/Indexes/EventType/EventTypeIndexReader.cs b/src/KurrentDB.SecondaryIndexing/Indexes/EventType/EventTypeIndexReader.cs index 0dc58df9bda..d743a373854 100644 --- a/src/KurrentDB.SecondaryIndexing/Indexes/EventType/EventTypeIndexReader.cs +++ b/src/KurrentDB.SecondaryIndexing/Indexes/EventType/EventTypeIndexReader.cs @@ -4,6 +4,7 @@ using Kurrent.Quack; using Kurrent.Quack.ConnectionPool; using KurrentDB.Core.Data; +using KurrentDB.Core.DuckDB; using KurrentDB.Core.Services.Storage.ReaderIndex; using KurrentDB.SecondaryIndexing.Indexes.Default; using KurrentDB.SecondaryIndexing.Storage; @@ -14,8 +15,9 @@ namespace KurrentDB.SecondaryIndexing.Indexes.EventType; internal class EventTypeIndexReader( DuckDBConnectionPool sharedPool, DefaultIndexProcessor processor, - IReadIndex index) - : SecondaryIndexReaderBase(sharedPool, index) { + IReadIndex index, + DuckDBCpuMetrics cpuMetrics) + : SecondaryIndexReaderBase(sharedPool, index, cpuMetrics) { protected override string GetId(string streamName) => EventTypeIndex.TryParseEventType(streamName, out var eventTypeName) ? eventTypeName : string.Empty; diff --git a/src/KurrentDB.SecondaryIndexing/Indexes/SecondaryIndexReaderBase.cs b/src/KurrentDB.SecondaryIndexing/Indexes/SecondaryIndexReaderBase.cs index 0a92762af00..9a83539262a 100644 --- a/src/KurrentDB.SecondaryIndexing/Indexes/SecondaryIndexReaderBase.cs +++ b/src/KurrentDB.SecondaryIndexing/Indexes/SecondaryIndexReaderBase.cs @@ -3,6 +3,7 @@ using Kurrent.Quack.ConnectionPool; using KurrentDB.Core.Data; +using KurrentDB.Core.DuckDB; using KurrentDB.Core.Services.Storage; using KurrentDB.Core.Services.Storage.ReaderIndex; using KurrentDB.SecondaryIndexing.Storage; @@ -10,7 +11,7 @@ namespace KurrentDB.SecondaryIndexing.Indexes; -public abstract class SecondaryIndexReaderBase(DuckDBConnectionPool sharedPool, IReadIndex index) : ISecondaryIndexReader { +public abstract class SecondaryIndexReaderBase(DuckDBConnectionPool sharedPool, IReadIndex index, DuckDBCpuMetrics cpuMetrics) : ISecondaryIndexReader { protected abstract string? GetId(string indexName); protected abstract List GetDbRecordsForwards(DuckDBConnectionPool pool, string? id, long startPosition, int maxCount, bool excludeFirst); @@ -57,12 +58,15 @@ ReadIndexEventsForwardCompleted NoData(ReadIndexResult result, bool endOfStream, => new(result, ResolvedEvent.EmptyArray, pos, lastIndexedPosition, endOfStream, error); async ValueTask<(long, IReadOnlyList)> GetEventsForwards(long startPosition) { - var indexPrepares = GetDbRecordsForwards( - GetPool(msg.Pool), - id, - startPosition, - msg.MaxCount, - msg.ExcludeStart); + List indexPrepares; + using (cpuMetrics.Measure(DuckDBCpuMetrics.Activities.Read)) { + indexPrepares = GetDbRecordsForwards( + GetPool(msg.Pool), + id, + startPosition, + msg.MaxCount, + msg.ExcludeStart); + } var events = await reader.ReadRecords(indexPrepares, true, token); return (indexPrepares.Count, events); @@ -101,11 +105,14 @@ ReadIndexEventsBackwardCompleted NoData(ReadIndexResult result, string? error = => new(result, ResolvedEvent.EmptyArray, pos, lastIndexedPosition, false, error); async ValueTask<(long, IReadOnlyList)> GetEventsBackwards(TFPos startPosition) { - var indexPrepares = GetDbRecordsBackwards(GetPool(msg.Pool), - id, - startPosition.PreparePosition, - msg.MaxCount, - msg.ExcludeStart); + List indexPrepares; + using (cpuMetrics.Measure(DuckDBCpuMetrics.Activities.Read)) { + indexPrepares = GetDbRecordsBackwards(GetPool(msg.Pool), + id, + startPosition.PreparePosition, + msg.MaxCount, + msg.ExcludeStart); + } var events = await reader.ReadRecords(indexPrepares, false, token); return (indexPrepares.Count, events); diff --git a/src/KurrentDB.SecondaryIndexing/Indexes/User/UserIndexEngine.cs b/src/KurrentDB.SecondaryIndexing/Indexes/User/UserIndexEngine.cs index 8957d2a9365..a82f3546c2c 100644 --- a/src/KurrentDB.SecondaryIndexing/Indexes/User/UserIndexEngine.cs +++ b/src/KurrentDB.SecondaryIndexing/Indexes/User/UserIndexEngine.cs @@ -9,6 +9,7 @@ using KurrentDB.Core; using KurrentDB.Core.Bus; using KurrentDB.Core.Data; +using KurrentDB.Core.DuckDB; using KurrentDB.Core.Messages; using KurrentDB.Core.Services.Storage; using KurrentDB.Core.Services.Storage.ReaderIndex; @@ -49,11 +50,12 @@ public UserIndexEngine( TFChunkDbConfig chunkDbConfig, [FromKeyedServices(SecondaryIndexingConstants.InjectionKey)] Meter meter, + DuckDBCpuMetrics cpuMetrics, ILoggerFactory loggerFactory) { _client = client; _log = loggerFactory.CreateLogger(); _writerCheckpoint = chunkDbConfig.WriterCheckpoint; - _subscription = new(client, publisher, serializer, options, db, index, meter, GetLastAppendedRecord, loggerFactory, _cts!.Token); + _subscription = new(client, publisher, serializer, options, db, index, meter, cpuMetrics, GetLastAppendedRecord, loggerFactory, _cts!.Token); subscriber.Subscribe(this); subscriber.Subscribe(this); diff --git a/src/KurrentDB.SecondaryIndexing/Indexes/User/UserIndexEngineSubscription.cs b/src/KurrentDB.SecondaryIndexing/Indexes/User/UserIndexEngineSubscription.cs index c851e4db14b..10ca049e5ba 100644 --- a/src/KurrentDB.SecondaryIndexing/Indexes/User/UserIndexEngineSubscription.cs +++ b/src/KurrentDB.SecondaryIndexing/Indexes/User/UserIndexEngineSubscription.cs @@ -12,6 +12,7 @@ using KurrentDB.Core; using KurrentDB.Core.Bus; using KurrentDB.Core.Data; +using KurrentDB.Core.DuckDB; using KurrentDB.Core.Messages; using KurrentDB.Core.Resilience; using KurrentDB.Core.Services.Storage; @@ -32,6 +33,7 @@ public partial class UserIndexEngineSubscription( DuckDBConnectionPool db, IReadIndex readIndex, Meter meter, + DuckDBCpuMetrics cpuMetrics, Func<(long, DateTime)> getLastAppendedRecord, ILoggerFactory logFactory, CancellationToken token) @@ -226,10 +228,11 @@ createdEvent.Fields.Count is 0 publisher: publisher, meter: meter, getLastAppendedRecord: getLastAppendedRecord, - loggerFactory: logFactory + loggerFactory: logFactory, + cpuMetrics: cpuMetrics ); - var reader = new UserIndexReader(sharedPool: db, processor, readIndex); + var reader = new UserIndexReader(sharedPool: db, processor, readIndex, cpuMetrics); UserIndexSubscription subscription = new UserIndexSubscription( publisher: publisher, diff --git a/src/KurrentDB.SecondaryIndexing/Indexes/User/UserIndexProcessor.cs b/src/KurrentDB.SecondaryIndexing/Indexes/User/UserIndexProcessor.cs index 3cdb4d36dbd..dcb30d82075 100644 --- a/src/KurrentDB.SecondaryIndexing/Indexes/User/UserIndexProcessor.cs +++ b/src/KurrentDB.SecondaryIndexing/Indexes/User/UserIndexProcessor.cs @@ -15,6 +15,7 @@ using KurrentDB.Common.Configuration; using KurrentDB.Core.Bus; using KurrentDB.Core.Data; +using KurrentDB.Core.DuckDB; using KurrentDB.Core.Messages; using KurrentDB.Scripting; using KurrentDB.SecondaryIndexing.Diagnostics; @@ -44,6 +45,7 @@ internal class UserIndexProcessor : UserIndexProcessor private readonly UserIndexSql _sql; private readonly object _skip; private readonly ILogger _log; + private readonly DuckDBCpuMetrics _cpuMetrics; private ulong _sequenceId; private Atomic _lastPosition; @@ -66,10 +68,12 @@ public UserIndexProcessor( Meter meter, Func<(long, DateTime)> getLastAppendedRecord, ILoggerFactory loggerFactory, + DuckDBCpuMetrics cpuMetrics, MetricsConfiguration? metricsConfiguration = null, TimeProvider? clock = null) { IndexName = indexName; _sql = sql; + _cpuMetrics = cpuMetrics; _log = loggerFactory.CreateLogger(); _queryStreamName = UserIndexHelpers.GetQueryStreamName(IndexName); @@ -221,6 +225,7 @@ public void Checkpoint(TFPos position, DateTime timestamp) { Created = new DateTimeOffset(timestamp).ToUnixTimeMilliseconds() }; + using var cpu = _cpuMetrics.Measure(DuckDBCpuMetrics.Activities.Checkpoint); UserIndexSql.SetCheckpoint(_connection, checkpointArgs); } @@ -233,6 +238,7 @@ public override void Commit() { try { using var duration = Tracker.StartCommitDuration(); + using var cpu = _cpuMetrics.Measure(DuckDBCpuMetrics.Activities.Commit); _appender.Flush(); _log.LogUserIndexCommitted(IndexName); } catch (Exception ex) { diff --git a/src/KurrentDB.SecondaryIndexing/Indexes/User/UserIndexReader.cs b/src/KurrentDB.SecondaryIndexing/Indexes/User/UserIndexReader.cs index 3ec3d9c1e61..17cd829ee08 100644 --- a/src/KurrentDB.SecondaryIndexing/Indexes/User/UserIndexReader.cs +++ b/src/KurrentDB.SecondaryIndexing/Indexes/User/UserIndexReader.cs @@ -5,21 +5,23 @@ using Kurrent.Quack.ConnectionPool; using Kurrent.Quack.Threading; using KurrentDB.Core.Data; +using KurrentDB.Core.DuckDB; using KurrentDB.Core.Services.Storage.ReaderIndex; using KurrentDB.SecondaryIndexing.Storage; namespace KurrentDB.SecondaryIndexing.Indexes.User; -internal abstract class UserIndexReader(DuckDBConnectionPool sharedPool, IReadIndex index) - : SecondaryIndexReaderBase(sharedPool, index) { +internal abstract class UserIndexReader(DuckDBConnectionPool sharedPool, IReadIndex index, DuckDBCpuMetrics cpuMetrics) + : SecondaryIndexReaderBase(sharedPool, index, cpuMetrics) { internal abstract BufferedView.Snapshot CaptureSnapshot(DuckDBAdvancedConnection connection); } internal class UserIndexReader( DuckDBConnectionPool sharedPool, UserIndexProcessor processor, - IReadIndex index -) : UserIndexReader(sharedPool, index) where TField : IField { + IReadIndex index, + DuckDBCpuMetrics cpuMetrics +) : UserIndexReader(sharedPool, index, cpuMetrics) where TField : IField { protected override string? GetId(string indexStream) { // the field is used as the ID. null when there is no field diff --git a/src/KurrentDB.SecondaryIndexing/Query/QueryEngine.cs b/src/KurrentDB.SecondaryIndexing/Query/QueryEngine.cs index c5357584cc3..c8418f2a07f 100644 --- a/src/KurrentDB.SecondaryIndexing/Query/QueryEngine.cs +++ b/src/KurrentDB.SecondaryIndexing/Query/QueryEngine.cs @@ -12,6 +12,7 @@ using Kurrent.Quack.Arrow; using Kurrent.Quack.ConnectionPool; using Kurrent.Quack.Threading; +using KurrentDB.Core.DuckDB; using KurrentDB.SecondaryIndexing.Indexes.Default; using KurrentDB.SecondaryIndexing.Indexes.User; @@ -23,9 +24,11 @@ namespace KurrentDB.SecondaryIndexing.Query; /// /// /// +/// internal sealed partial class QueryEngine(DefaultIndexProcessor defaultIndex, UserIndexEngine userIndex, - DuckDBConnectionPool sharedPool) : IQueryEngine { + DuckDBConnectionPool sharedPool, + DuckDBCpuMetrics cpuMetrics) : IQueryEngine { // 32 bytes key is aligned with HMAC SHA-3 256 hash length private readonly ReadOnlyMemory _signatureKey = RandomNumberGenerator.GetBytes(32); @@ -52,11 +55,16 @@ public async ValueTask ExecuteAsync(ReadOnlyMemory preparedQuer var reader = default(QueryResultReader); var cancellation = connection.InterruptQueryOnCancellation(token); try { - CaptureSnapshots(in parsedQuery, connection, snapshots, token); - statement = new(connection, parsedQuery.Query); - consumer.Bind(new QueryBinder(in statement)); + // caller-side CPU attribution covers the synchronous part only: snapshot capture, + // parse/plan and, for materialized results, execution. CPU burned inside DuckDB + // during streaming chunk fetches is not visible from here. + using (cpuMetrics.Measure(DuckDBCpuMetrics.Activities.Query)) { + CaptureSnapshots(in parsedQuery, connection, snapshots, token); + statement = new(connection, parsedQuery.Query); + consumer.Bind(new QueryBinder(in statement)); + reader = new(in statement, consumer.UseStreaming); + } - reader = new(in statement, consumer.UseStreaming); await consumer.ConsumeAsync(reader, token); reader.ThrowOnError(); // to handle query interruption from Quack (see catch block) } catch (DuckDBException e) when (e.ErrorType is DuckDBErrorType.Interrupt) { @@ -110,9 +118,11 @@ private TResult GetArrowSchema(ReadOnlySpan preparedQ var options = connection.GetArrowOptions(); var statement = default(PreparedStatement); try { - CaptureSnapshots(in parsedQuery, connection, snapshots, CancellationToken.None); - statement = new(connection, parsedQuery.Query); - return TReflector.Reflect(statement, options); + using (cpuMetrics.Measure(DuckDBCpuMetrics.Activities.Query)) { + CaptureSnapshots(in parsedQuery, connection, snapshots, CancellationToken.None); + statement = new(connection, parsedQuery.Query); + return TReflector.Reflect(statement, options); + } } finally { statement.Dispose(); options.Dispose(); diff --git a/src/KurrentDB/metricsconfig.json b/src/KurrentDB/metricsconfig.json index fda6ad698d3..3eb04bfaff6 100644 --- a/src/KurrentDB/metricsconfig.json +++ b/src/KurrentDB/metricsconfig.json @@ -9,6 +9,7 @@ "Kurrent.Connectors", "Kurrent.Connectors.Sinks", "KurrentDB.SecondaryIndexes", + "KurrentDB.DuckDB", "KurrentDB.Kontext" ], From ae5a03ec94689a5b6d3da184f6e40beaebb6b492 Mon Sep 17 00:00:00 2001 From: realtonyyoung <6655045+realtonyyoung@users.noreply.github.com> Date: Thu, 18 Jun 2026 21:58:28 -0400 Subject: [PATCH 2/3] Attribute streaming result-fetch CPU in DuckDB query metric The query CPU scope in QueryEngine.ExecuteAsync covered only the synchronous setup (snapshot capture, parse/plan, ExecuteQuery). In streaming mode DuckDB produces result chunks later, inside QueryResultReader.TryRead -> _result.TryFetch, called from ConsumeAsync after that scope closes; FinalizeEnumeration (via Dispose) could also drain chunks unattributed. Large streamed queries therefore under-reported query CPU. Thread DuckDBCpuMetrics into QueryResultReader and measure the synchronous TryFetch calls in TryRead and FinalizeEnumeration under the same query activity. Adds an integration test asserting the streaming read path emits caller-attributed query CPU (tagging/wiring; a per-fetch count assertion is avoided as it would be flaky under the Windows GetThreadTimes quantum). Co-Authored-By: Claude Opus 4.8 --- .../IntegrationTests/ReadTests.cs | 46 +++++++++++++++++++ .../Query/QueryEngine.QueryResult.cs | 9 +++- .../Query/QueryEngine.cs | 8 ++-- 3 files changed, 58 insertions(+), 5 deletions(-) diff --git a/src/KurrentDB.SecondaryIndexing.Tests/IntegrationTests/ReadTests.cs b/src/KurrentDB.SecondaryIndexing.Tests/IntegrationTests/ReadTests.cs index c3b3f1f08df..eab4697bbdb 100644 --- a/src/KurrentDB.SecondaryIndexing.Tests/IntegrationTests/ReadTests.cs +++ b/src/KurrentDB.SecondaryIndexing.Tests/IntegrationTests/ReadTests.cs @@ -1,8 +1,10 @@ // Copyright (c) Kurrent, Inc and/or licensed to Kurrent, Inc under one or more agreements. // Kurrent, Inc licenses this file to you under the Kurrent License v1 (see LICENSE.md). +using System.Diagnostics.Metrics; using KurrentDB.Core.ClientPublisher; using KurrentDB.Core.Data; +using KurrentDB.Core.DuckDB; using KurrentDB.Core.Services; using KurrentDB.Core.Services.Storage.ReaderIndex; using KurrentDB.Core.Services.Transport.Common; @@ -44,6 +46,50 @@ await engine.ExecuteAsync( Assert.True(consumer.RowCount > 0); } + // Guards that the streaming query path (QueryResultReader.TryRead/FinalizeEnumeration, where DuckDB + // produces result chunks) emits caller-attributed CPU under the query activity. We assert tagging and + // wiring rather than a per-fetch measurement count: a single sub-vector-size chunk fetch can complete + // below the Windows GetThreadTimes quantum and record zero, so a count assertion would be flaky there. + [Fact] + public async Task QueryEngineStreamingReadAttributesQueryCpuToCaller() { + var engine = Fixture.NodeServices.GetRequiredService(); + using var preparedSql = engine.PrepareQuery( + "SELECT metadata FROM kdb.records"u8, + new() { UseDigitalSignature = true }); + + object gate = new(); + List[]> queryMeasurements = []; + using var listener = new MeterListener(); + listener.InstrumentPublished = (instrument, l) => { + if (instrument.Meter.Name == DuckDBCpuMetrics.MeterName && instrument.Name == "kurrentdb.duckdb.cpu.seconds") + l.EnableMeasurementEvents(instrument); + }; + listener.SetMeasurementEventCallback((_, _, tags, _) => { + var copy = tags.ToArray(); + // Ignore background index commit/checkpoint activity on the shared meter. + if (copy.Any(t => t is { Key: "activity", Value: "query" })) { + lock (gate) + queryMeasurements.Add(copy); + } + }); + listener.Start(); + + var consumer = new RowCountReader(); + await engine.ExecuteAsync( + preparedSql.Memory, + consumer, + new() { CheckIntegrity = true }, + TestContext.Current.CancellationToken); + + listener.Dispose(); // stop callbacks before asserting + + Assert.True(consumer.RowCount > 0); // the streaming fetch loop actually ran + lock (gate) { + Assert.NotEmpty(queryMeasurements); + Assert.All(queryMeasurements, tags => Assert.Contains(tags, t => t is { Key: "source", Value: "caller" })); + } + } + [Theory] [InlineData(true)] [InlineData(false)] diff --git a/src/KurrentDB.SecondaryIndexing/Query/QueryEngine.QueryResult.cs b/src/KurrentDB.SecondaryIndexing/Query/QueryEngine.QueryResult.cs index 5cbbbb8e294..9e012d20866 100644 --- a/src/KurrentDB.SecondaryIndexing/Query/QueryEngine.QueryResult.cs +++ b/src/KurrentDB.SecondaryIndexing/Query/QueryEngine.QueryResult.cs @@ -5,16 +5,19 @@ using DotNext; using Kurrent.Quack; using Kurrent.Quack.Arrow; +using KurrentDB.Core.DuckDB; using QuackQueryResult = Kurrent.Quack.QueryResult; namespace KurrentDB.SecondaryIndexing.Query; partial class QueryEngine { private sealed class QueryResultReader : Disposable, IQueryResultReader { + private readonly DuckDBCpuMetrics _cpuMetrics; private QuackQueryResult _result; private DataChunk _chunk; - public QueryResultReader(in PreparedStatement statement, bool useStreaming) { + public QueryResultReader(in PreparedStatement statement, bool useStreaming, DuckDBCpuMetrics cpuMetrics) { + _cpuMetrics = cpuMetrics; _result = statement.ExecuteQuery(useStreaming); } @@ -26,12 +29,16 @@ public QueryResultReader(in PreparedStatement statement, bool useStreaming) { public bool TryRead() { _chunk.Dispose(); + // In streaming mode this fetch is where DuckDB produces the next result chunk, + // so its CPU is attributed to the query here rather than in QueryEngine.ExecuteAsync. + using var cpu = _cpuMetrics.Measure(DuckDBCpuMetrics.Activities.Query); return _result.TryFetch(out _chunk); } public ref readonly DataChunk Chunk => ref _chunk; private void FinalizeEnumeration() { + using var cpu = _cpuMetrics.Measure(DuckDBCpuMetrics.Activities.Query); while (_result.TryFetch(out _chunk)) { _chunk.Dispose(); } diff --git a/src/KurrentDB.SecondaryIndexing/Query/QueryEngine.cs b/src/KurrentDB.SecondaryIndexing/Query/QueryEngine.cs index c8418f2a07f..10efe7b09c7 100644 --- a/src/KurrentDB.SecondaryIndexing/Query/QueryEngine.cs +++ b/src/KurrentDB.SecondaryIndexing/Query/QueryEngine.cs @@ -55,14 +55,14 @@ public async ValueTask ExecuteAsync(ReadOnlyMemory preparedQuer var reader = default(QueryResultReader); var cancellation = connection.InterruptQueryOnCancellation(token); try { - // caller-side CPU attribution covers the synchronous part only: snapshot capture, - // parse/plan and, for materialized results, execution. CPU burned inside DuckDB - // during streaming chunk fetches is not visible from here. + // caller-side CPU attribution for the synchronous setup: snapshot capture, + // parse/plan and, for materialized results, execution. Streaming chunk fetches + // happen later inside ConsumeAsync and are attributed by QueryResultReader itself. using (cpuMetrics.Measure(DuckDBCpuMetrics.Activities.Query)) { CaptureSnapshots(in parsedQuery, connection, snapshots, token); statement = new(connection, parsedQuery.Query); consumer.Bind(new QueryBinder(in statement)); - reader = new(in statement, consumer.UseStreaming); + reader = new(in statement, consumer.UseStreaming, cpuMetrics); } await consumer.ConsumeAsync(reader, token); From 650cfe18e08d6314239661946871d4f70a6849e8 Mon Sep 17 00:00:00 2001 From: realtonyyoung <6655045+realtonyyoung@users.noreply.github.com> Date: Thu, 18 Jun 2026 22:37:10 -0400 Subject: [PATCH 3/3] Fix DuckDB CPU meter service name; make CPU measurement deterministically testable Review fixes for PR #5642: - AddDuckDb resolved the meter's service name via sp.GetService(), which is not registered in the collection AddDuckDb runs from, so it always fell back to "kurrentdb" and mis-named the meter under legacy "eventstore" naming. Pass the configured ServiceName in from ClusterVNodeStartup instead, dropping the dead DI lookup. - DuckDBCpuMetrics now takes an optional CPU-time source (defaulting to ThreadCpuTime), so tests can inject a deterministic clock. Replace the timing-dependent streaming-read integration assertion (which could flake when a small query's caller CPU rounded to zero ticks, notably under Windows' quantum-granular GetThreadTimes) with deterministic unit tests that verify the exact recorded delta, tags, and the zero-delta no-op gate. Co-Authored-By: Claude Opus 4.8 --- src/KurrentDB.Core/ClusterVNodeStartup.cs | 2 +- src/KurrentDB.Core/DuckDB/DuckDBCpuMetrics.cs | 18 ++++++-- .../DuckDB/InjectionExtensions.cs | 11 ++--- .../Diagnostics/DuckDBCpuMetricsTests.cs | 40 ++++++++++++++++ .../IntegrationTests/ReadTests.cs | 46 ------------------- 5 files changed, 59 insertions(+), 58 deletions(-) diff --git a/src/KurrentDB.Core/ClusterVNodeStartup.cs b/src/KurrentDB.Core/ClusterVNodeStartup.cs index d6b795ce72e..b13d4c312ae 100644 --- a/src/KurrentDB.Core/ClusterVNodeStartup.cs +++ b/src/KurrentDB.Core/ClusterVNodeStartup.cs @@ -291,7 +291,7 @@ public void ConfigureServices(IServiceCollection services) { }) .AddServiceOptions>(options => options.MaxReceiveMessageSize = TFConsts.EffectiveMaxLogRecordSize); - services.AddDuckDb(); + services.AddDuckDb(_metricsConfiguration.ServiceName); // Ask the node itself to add DI registrations _configureNodeServices(services); diff --git a/src/KurrentDB.Core/DuckDB/DuckDBCpuMetrics.cs b/src/KurrentDB.Core/DuckDB/DuckDBCpuMetrics.cs index 08cc4bd98b9..6213ff9dd4e 100644 --- a/src/KurrentDB.Core/DuckDB/DuckDBCpuMetrics.cs +++ b/src/KurrentDB.Core/DuckDB/DuckDBCpuMetrics.cs @@ -1,6 +1,7 @@ // Copyright (c) Kurrent, Inc and/or licensed to Kurrent, Inc under one or more agreements. // Kurrent, Inc licenses this file to you under the Kurrent License v1 (see LICENSE.md). +using System; using System.Collections.Generic; using System.Diagnostics.Metrics; @@ -23,10 +24,17 @@ public static class Activities { private readonly Counter _cpuSeconds; - public DuckDBCpuMetrics(Meter meter, string serviceName) { + // Reads the calling thread's cumulative CPU time in nanoseconds. Null when per-thread CPU + // time is unavailable on this platform, in which case measurement is a no-op. Injectable so + // tests can supply a deterministic source instead of relying on real per-thread CPU accounting. + private readonly Func _currentCpuNanoseconds; + + public DuckDBCpuMetrics(Meter meter, string serviceName, Func currentCpuNanoseconds = null) { _cpuSeconds = meter.CreateCounter( $"{serviceName}.duckdb.cpu.seconds", description: "CPU time consumed by DuckDB operations on KurrentDB threads, in seconds"); + _currentCpuNanoseconds = currentCpuNanoseconds + ?? (ThreadCpuTime.IsSupported ? () => ThreadCpuTime.CurrentNanoseconds : (Func)null); } public CpuScope Measure(string activity) => new(this, activity); @@ -39,19 +47,19 @@ public readonly ref struct CpuScope { private readonly long _startNanoseconds; internal CpuScope(DuckDBCpuMetrics metrics, string activity) { - if (!ThreadCpuTime.IsSupported) - return; + if (metrics._currentCpuNanoseconds is null) + return; // per-thread CPU time unavailable on this platform: no-op _metrics = metrics; _activity = activity; - _startNanoseconds = ThreadCpuTime.CurrentNanoseconds; + _startNanoseconds = metrics._currentCpuNanoseconds(); } public void Dispose() { if (_metrics is null) return; - var elapsedNanoseconds = ThreadCpuTime.CurrentNanoseconds - _startNanoseconds; + var elapsedNanoseconds = _metrics._currentCpuNanoseconds!() - _startNanoseconds; if (elapsedNanoseconds > 0) _metrics._cpuSeconds.Add( elapsedNanoseconds / 1e9, diff --git a/src/KurrentDB.Core/DuckDB/InjectionExtensions.cs b/src/KurrentDB.Core/DuckDB/InjectionExtensions.cs index f9aa6a28d3a..1dbedc0ef2c 100644 --- a/src/KurrentDB.Core/DuckDB/InjectionExtensions.cs +++ b/src/KurrentDB.Core/DuckDB/InjectionExtensions.cs @@ -5,7 +5,6 @@ using System.Diagnostics.Metrics; using System.Threading.Tasks; using Kurrent.Quack.ConnectionPool; -using KurrentDB.Common.Configuration; using KurrentDB.DuckDB; using Microsoft.AspNetCore.Builder; using Microsoft.AspNetCore.Connections; @@ -17,16 +16,16 @@ namespace KurrentDB.Core.DuckDB; public static class InjectionExtensions { - public static IServiceCollection AddDuckDb(this IServiceCollection services) { + public static IServiceCollection AddDuckDb(this IServiceCollection services, string serviceName) { services.AddSingleton(); services.AddHostedService(sp => sp.GetRequiredService()); services.AddSingleton(sp => sp.GetRequiredService().Shared); services.AddSingleton(); services.AddSingleton(CreatePoolPerConnectionInterceptor); - services.AddSingleton(static sp => { - var serviceName = sp.GetService()?.ServiceName ?? "kurrentdb"; - return new DuckDBCpuMetrics(new Meter(DuckDBCpuMetrics.MeterName, "1.0.0"), serviceName); - }); + // serviceName is passed in (not resolved from DI) because MetricsConfiguration is not + // available in every collection AddDuckDb is used from; this keeps the meter named + // consistently with the rest of the metrics pipeline, including legacy "eventstore" naming. + services.AddSingleton(new DuckDBCpuMetrics(new Meter(DuckDBCpuMetrics.MeterName, "1.0.0"), serviceName)); return services; } diff --git a/src/KurrentDB.SecondaryIndexing.Tests/Diagnostics/DuckDBCpuMetricsTests.cs b/src/KurrentDB.SecondaryIndexing.Tests/Diagnostics/DuckDBCpuMetricsTests.cs index 861074d36ef..e7f2bb03419 100644 --- a/src/KurrentDB.SecondaryIndexing.Tests/Diagnostics/DuckDBCpuMetricsTests.cs +++ b/src/KurrentDB.SecondaryIndexing.Tests/Diagnostics/DuckDBCpuMetricsTests.cs @@ -44,6 +44,46 @@ public void idle_scope_records_at_most_negligible_cpu() { Assert.True(measurements.Sum(m => m.Value) < 0.05, $"expected near-zero CPU, got {measurements.Sum(m => m.Value)}s"); } + // Deterministic: a fake CPU-time source replaces real per-thread accounting, so the assertion does + // not depend on the scope accruing measurable CPU (which can round to zero on small/fast work, + // especially under Windows' scheduler-quantum-granular GetThreadTimes). + [Theory] + [InlineData(DuckDBCpuMetrics.Activities.Query)] + [InlineData(DuckDBCpuMetrics.Activities.Commit)] + public void measure_records_exact_caller_cpu_delta_with_tags(string activity) { + using var meter = new Meter("test"); + var cpuNanoseconds = 1_000_000_000L; + var metrics = new DuckDBCpuMetrics(meter, "kurrentdb", () => cpuNanoseconds); + + List<(double Value, KeyValuePair[] Tags)> measurements = []; + using var listener = Listen(meter, measurements); + + using (metrics.Measure(activity)) { + cpuNanoseconds += 250_000_000; // +0.25s of caller CPU + } + + var measurement = Assert.Single(measurements); + Assert.Equal(0.25, measurement.Value, precision: 9); + Assert.Contains(measurement.Tags, t => t.Key == "activity" && (string?)t.Value == activity); + Assert.Contains(measurement.Tags, t => t is { Key: "source", Value: "caller" }); + } + + [Fact] + public void measure_with_zero_cpu_delta_records_nothing() { + using var meter = new Meter("test"); + var cpuNanoseconds = 5_000_000_000L; + var metrics = new DuckDBCpuMetrics(meter, "kurrentdb", () => cpuNanoseconds); + + List<(double Value, KeyValuePair[] Tags)> measurements = []; + using var listener = Listen(meter, measurements); + + using (metrics.Measure(DuckDBCpuMetrics.Activities.Query)) { + // no CPU accrues during the scope + } + + Assert.Empty(measurements); + } + private static MeterListener Listen(Meter meter, List<(double, KeyValuePair[])> measurements) { var listener = new MeterListener(); listener.InstrumentPublished = (instrument, l) => { diff --git a/src/KurrentDB.SecondaryIndexing.Tests/IntegrationTests/ReadTests.cs b/src/KurrentDB.SecondaryIndexing.Tests/IntegrationTests/ReadTests.cs index eab4697bbdb..c3b3f1f08df 100644 --- a/src/KurrentDB.SecondaryIndexing.Tests/IntegrationTests/ReadTests.cs +++ b/src/KurrentDB.SecondaryIndexing.Tests/IntegrationTests/ReadTests.cs @@ -1,10 +1,8 @@ // Copyright (c) Kurrent, Inc and/or licensed to Kurrent, Inc under one or more agreements. // Kurrent, Inc licenses this file to you under the Kurrent License v1 (see LICENSE.md). -using System.Diagnostics.Metrics; using KurrentDB.Core.ClientPublisher; using KurrentDB.Core.Data; -using KurrentDB.Core.DuckDB; using KurrentDB.Core.Services; using KurrentDB.Core.Services.Storage.ReaderIndex; using KurrentDB.Core.Services.Transport.Common; @@ -46,50 +44,6 @@ await engine.ExecuteAsync( Assert.True(consumer.RowCount > 0); } - // Guards that the streaming query path (QueryResultReader.TryRead/FinalizeEnumeration, where DuckDB - // produces result chunks) emits caller-attributed CPU under the query activity. We assert tagging and - // wiring rather than a per-fetch measurement count: a single sub-vector-size chunk fetch can complete - // below the Windows GetThreadTimes quantum and record zero, so a count assertion would be flaky there. - [Fact] - public async Task QueryEngineStreamingReadAttributesQueryCpuToCaller() { - var engine = Fixture.NodeServices.GetRequiredService(); - using var preparedSql = engine.PrepareQuery( - "SELECT metadata FROM kdb.records"u8, - new() { UseDigitalSignature = true }); - - object gate = new(); - List[]> queryMeasurements = []; - using var listener = new MeterListener(); - listener.InstrumentPublished = (instrument, l) => { - if (instrument.Meter.Name == DuckDBCpuMetrics.MeterName && instrument.Name == "kurrentdb.duckdb.cpu.seconds") - l.EnableMeasurementEvents(instrument); - }; - listener.SetMeasurementEventCallback((_, _, tags, _) => { - var copy = tags.ToArray(); - // Ignore background index commit/checkpoint activity on the shared meter. - if (copy.Any(t => t is { Key: "activity", Value: "query" })) { - lock (gate) - queryMeasurements.Add(copy); - } - }); - listener.Start(); - - var consumer = new RowCountReader(); - await engine.ExecuteAsync( - preparedSql.Memory, - consumer, - new() { CheckIntegrity = true }, - TestContext.Current.CancellationToken); - - listener.Dispose(); // stop callbacks before asserting - - Assert.True(consumer.RowCount > 0); // the streaming fetch loop actually ran - lock (gate) { - Assert.NotEmpty(queryMeasurements); - Assert.All(queryMeasurements, tags => Assert.Contains(tags, t => t is { Key: "source", Value: "caller" })); - } - } - [Theory] [InlineData(true)] [InlineData(false)]