diff --git a/src/KurrentDB.Core/ClusterVNodeStartup.cs b/src/KurrentDB.Core/ClusterVNodeStartup.cs index a4676a82735..f3a369fdb6b 100644 --- a/src/KurrentDB.Core/ClusterVNodeStartup.cs +++ b/src/KurrentDB.Core/ClusterVNodeStartup.cs @@ -314,7 +314,7 @@ public void ConfigureServices(IServiceCollection services) { services.AddGrpcReflection(); #endif - 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 new file mode 100644 index 00000000000..6213ff9dd4e --- /dev/null +++ b/src/KurrentDB.Core/DuckDB/DuckDBCpuMetrics.cs @@ -0,0 +1,70 @@ +// 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; + +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; + + // 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); + + // 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 (metrics._currentCpuNanoseconds is null) + return; // per-thread CPU time unavailable on this platform: no-op + + _metrics = metrics; + _activity = activity; + _startNanoseconds = metrics._currentCpuNanoseconds(); + } + + public void Dispose() { + if (_metrics is null) + return; + + var elapsedNanoseconds = _metrics._currentCpuNanoseconds!() - _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..1dbedc0ef2c 100644 --- a/src/KurrentDB.Core/DuckDB/InjectionExtensions.cs +++ b/src/KurrentDB.Core/DuckDB/InjectionExtensions.cs @@ -2,6 +2,7 @@ // 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.DuckDB; @@ -15,12 +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); + // 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.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..e7f2bb03419 --- /dev/null +++ b/src/KurrentDB.SecondaryIndexing.Tests/Diagnostics/DuckDBCpuMetricsTests.cs @@ -0,0 +1,98 @@ +// 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"); + } + + // 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) => { + 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.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 c5357584cc3..10efe7b09c7 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 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, cpuMetrics); + } - 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" ],