diff --git a/CHANGELOG.md b/CHANGELOG.md index c1a4020..6f4016b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,14 @@ # Changelog +## Unreleased + +## 1.0.1 + +- `NexusTimeSession` catalog/ops extras (ISP: not on `ITimeSession`): `ListSeries`, + `GetCardinality`, `ApplyRetention` over Wire.V1 RPCs with DTOs + `TimeSeriesCatalogEntry`, `TimeCardinalitySnapshot`, `TimeMetricCardinality`. +- Suite-layout ProjectReference resolution for `Mishima.Nexus.Wire.V1` (alongside sibling monorepo path). + ## 1.0.0 - Initial `NexusClient`: connects via `nexus+graph|time|document|relational://` URIs (explicit port diff --git a/Directory.Build.props b/Directory.Build.props index 0024e5c..460a8b0 100644 --- a/Directory.Build.props +++ b/Directory.Build.props @@ -39,6 +39,11 @@ $([System.IO.Path]::GetFullPath('$(MSBuildThisFileDirectory)..\robotico-graph-csharp\src\Robotico.Graph\Robotico.Graph.csproj')) $([System.IO.Path]::GetFullPath('$(MSBuildThisFileDirectory)..\robotico-time-csharp\src\Robotico.Time\Robotico.Time.csproj')) - $([System.IO.Path]::GetFullPath('$(MSBuildThisFileDirectory)..\..\mishima-suite\mishima-nexus\src\Mishima.Nexus.Wire.V1\Mishima.Nexus.Wire.V1.csproj')) + + <_MishimaNexusWireV1Suite>$([System.IO.Path]::GetFullPath('$(MSBuildThisFileDirectory)..\..\mishima-nexus\src\Mishima.Nexus.Wire.V1\Mishima.Nexus.Wire.V1.csproj')) + + <_MishimaNexusWireV1Sibling>$([System.IO.Path]::GetFullPath('$(MSBuildThisFileDirectory)..\..\mishima-suite\mishima-nexus\src\Mishima.Nexus.Wire.V1\Mishima.Nexus.Wire.V1.csproj')) + $(_MishimaNexusWireV1Suite) + $(_MishimaNexusWireV1Sibling) diff --git a/src/Robotico.Nexus.Client/Time/NexusTimeSession.cs b/src/Robotico.Nexus.Client/Time/NexusTimeSession.cs index 974da8e..055f4e4 100644 --- a/src/Robotico.Nexus.Client/Time/NexusTimeSession.cs +++ b/src/Robotico.Nexus.Client/Time/NexusTimeSession.cs @@ -322,6 +322,108 @@ public sealed class NexusTimeSession : ITimeSession }); } + /// + /// Lists registered series for explorers (Nexus wire extra; not on ). + /// + public Robotico.Result.Result> ListSeries(int limit = 1000) + { + Robotico.Result.Result disposed = EnsureNotDisposed(); + if (disposed.IsError(out IError? disposedError)) + { + return Robotico.Result.Result.Error>(disposedError!); + } + + int normalizedLimit = limit < 1 ? 1000 : limit; + return InvokeGeneric(() => + { + TimeListSeriesResponse response = _client.ListSeries( + new TimeListSeriesRequest { Limit = normalizedLimit }, + Metadata()); + TimeSeriesCatalogEntry[] entries = new TimeSeriesCatalogEntry[response.Series.Count]; + for (int i = 0; i < response.Series.Count; i++) + { + TimeSeriesCatalogEntryWire wire = response.Series[i]; + Dictionary labels = new(StringComparer.Ordinal); + foreach (TimeLabelPair pair in wire.Labels) + { + labels[pair.Key] = pair.Value; + } + + entries[i] = new TimeSeriesCatalogEntry( + wire.SeriesId, + wire.MetricName, + labels, + wire.SampleCount); + } + + return NexusWireStatus.ToResult(response.Status, (IReadOnlyList)entries); + }); + } + + /// + /// Cardinality inspection (Nexus wire extra; not on ). + /// + public Robotico.Result.Result GetCardinality(int topN = 20) + { + Robotico.Result.Result disposed = EnsureNotDisposed(); + if (disposed.IsError(out IError? disposedError)) + { + return Robotico.Result.Result.Error(disposedError!); + } + + int normalizedTopN = topN < 1 ? 20 : topN; + return InvokeGeneric(() => + { + TimeGetCardinalityResponse response = _client.GetCardinality( + new TimeGetCardinalityRequest { TopN = normalizedTopN }, + Metadata()); + TimeMetricCardinality[] top = + [ + .. response.TopMetricsBySeriesCount.Select(static m => + new TimeMetricCardinality(m.MetricName, m.SeriesCount)), + ]; + TimeCardinalitySnapshot snap = new( + response.ActiveSeriesCount, + response.TotalSampleCount, + top); + return NexusWireStatus.ToResult(response.Status, snap); + }); + } + + /// + /// Applies retention cutoff (Nexus wire extra; not on ). + /// Drops samples/events with timestamp strictly before + /// . + /// + /// Removed sample/event count reported by the engine. + public Robotico.Result.Result ApplyRetention(long retainAfterExclusiveUnixNanos) + { + Robotico.Result.Result disposed = EnsureNotDisposed(); + if (disposed.IsError(out IError? disposedError)) + { + return Robotico.Result.Result.Error(disposedError!); + } + + if (retainAfterExclusiveUnixNanos <= 0) + { + return Robotico.Result.Result.Error( + TimeErrors.Error( + "retainAfterExclusiveUnixNanos must be positive.", + TimeErrors.InvalidArgument)); + } + + return InvokeGeneric(() => + { + TimeApplyRetentionResponse response = _client.ApplyRetention( + new TimeApplyRetentionRequest + { + RetainAfterExclusiveUnixNanos = retainAfterExclusiveUnixNanos, + }, + Metadata()); + return NexusWireStatus.ToResult(response.Status, response.RemovedCount); + }); + } + /// public void Dispose() => _disposed = true; diff --git a/src/Robotico.Nexus.Client/Time/TimeCardinalitySnapshot.cs b/src/Robotico.Nexus.Client/Time/TimeCardinalitySnapshot.cs new file mode 100644 index 0000000..5cd69bc --- /dev/null +++ b/src/Robotico.Nexus.Client/Time/TimeCardinalitySnapshot.cs @@ -0,0 +1,13 @@ +namespace Robotico.Nexus.Client.Time; + +/// +/// Cardinality snapshot from . +/// Not on ITimeSession (ISP): ops inspection surface for explorers. +/// +/// Active series in the logical database. +/// Total samples across all series. +/// Top metrics by series fan-out. +public sealed record TimeCardinalitySnapshot( + long ActiveSeriesCount, + long TotalSampleCount, + IReadOnlyList TopMetricsBySeriesCount); diff --git a/src/Robotico.Nexus.Client/Time/TimeMetricCardinality.cs b/src/Robotico.Nexus.Client/Time/TimeMetricCardinality.cs new file mode 100644 index 0000000..6a98e5a --- /dev/null +++ b/src/Robotico.Nexus.Client/Time/TimeMetricCardinality.cs @@ -0,0 +1,6 @@ +namespace Robotico.Nexus.Client.Time; + +/// Per-metric series count for cardinality inspection. +/// Metric name. +/// Number of series with this metric name. +public sealed record TimeMetricCardinality(string MetricName, long SeriesCount); diff --git a/src/Robotico.Nexus.Client/Time/TimeSeriesCatalogEntry.cs b/src/Robotico.Nexus.Client/Time/TimeSeriesCatalogEntry.cs new file mode 100644 index 0000000..510b4ea --- /dev/null +++ b/src/Robotico.Nexus.Client/Time/TimeSeriesCatalogEntry.cs @@ -0,0 +1,15 @@ +namespace Robotico.Nexus.Client.Time; + +/// +/// One registered series from . +/// Not on ITimeSession (ISP): catalog ops are Nexus wire extras for explorers/ops. +/// +/// Engine-assigned series id. +/// Metric name (Prometheus-style). +/// Series labels (sorted as returned by the engine). +/// Current sample count for the series. +public sealed record TimeSeriesCatalogEntry( + ulong SeriesId, + string MetricName, + IReadOnlyDictionary Labels, + int SampleCount); diff --git a/tests/Robotico.Nexus.Client.Tests/TestSupport/FakeTimeEngineService.cs b/tests/Robotico.Nexus.Client.Tests/TestSupport/FakeTimeEngineService.cs index 7f4c465..f0607c9 100644 --- a/tests/Robotico.Nexus.Client.Tests/TestSupport/FakeTimeEngineService.cs +++ b/tests/Robotico.Nexus.Client.Tests/TestSupport/FakeTimeEngineService.cs @@ -7,11 +7,13 @@ namespace Robotico.Nexus.Client.Tests.TestSupport; public sealed class FakeTimeEngineService : NexusTimeEngineService.NexusTimeEngineServiceBase { private readonly Dictionary> _samplesBySeries = []; + private readonly Dictionary Labels)> _seriesMeta = []; private readonly Dictionary> _eventsByTable = []; private readonly Dictionary> _fieldsByTable = []; private ulong _nextSeriesId = 1; private ulong _nextTableId = 1; private ulong _nextEventId = 1; + private long _retentionRemoved; /// public override Task Ping(TimePingRequest request, ServerCallContext context) @@ -40,9 +42,76 @@ public sealed class FakeTimeEngineService : NexusTimeEngineService.NexusTimeEngi { ulong id = _nextSeriesId++; _samplesBySeries[id] = []; + _seriesMeta[id] = (request.MetricName, [.. request.Labels]); return Task.FromResult(new TimeRegisterSeriesResponse { Status = NexusStatuses.Ok, SeriesId = id }); } + /// + public override Task ListSeries(TimeListSeriesRequest request, ServerCallContext context) + { + TimeListSeriesResponse response = new() { Status = NexusStatuses.Ok }; + int limit = request.Limit is 0 ? int.MaxValue : request.Limit; + foreach ((ulong id, (string metric, IReadOnlyList labels)) in _seriesMeta.Take(limit)) + { + TimeSeriesCatalogEntryWire entry = new() + { + SeriesId = id, + MetricName = metric, + SampleCount = _samplesBySeries.TryGetValue(id, out List? samples) ? samples.Count : 0, + }; + entry.Labels.AddRange(labels); + response.Series.Add(entry); + } + + return Task.FromResult(response); + } + + /// + public override Task GetCardinality(TimeGetCardinalityRequest request, ServerCallContext context) + { + Dictionary byMetric = new(StringComparer.Ordinal); + foreach ((string metric, _) in _seriesMeta.Values) + { + byMetric[metric] = byMetric.GetValueOrDefault(metric) + 1; + } + + int topN = request.TopN is 0 ? 20 : request.TopN; + TimeGetCardinalityResponse response = new() + { + Status = NexusStatuses.Ok, + ActiveSeriesCount = _seriesMeta.Count, + TotalSampleCount = _samplesBySeries.Values.Sum(list => list.Count), + }; + foreach ((string metric, int count) in byMetric.OrderByDescending(static p => p.Value).ThenBy(static p => p.Key, StringComparer.Ordinal).Take(topN)) + { + response.TopMetricsBySeriesCount.Add(new TimeMetricCardinalityWire { MetricName = metric, SeriesCount = count }); + } + + return Task.FromResult(response); + } + + /// + public override Task ApplyRetention(TimeApplyRetentionRequest request, ServerCallContext context) + { + long removed = 0; + foreach (List samples in _samplesBySeries.Values) + { + int before = samples.Count; + samples.RemoveAll(s => s.TimestampUnixNanos < request.RetainAfterExclusiveUnixNanos); + removed += before - samples.Count; + } + + foreach (List events in _eventsByTable.Values) + { + int before = events.Count; + events.RemoveAll(e => e.TimestampUnixNanos < request.RetainAfterExclusiveUnixNanos); + removed += before - events.Count; + } + + _retentionRemoved = removed; + return Task.FromResult(new TimeApplyRetentionResponse { Status = NexusStatuses.Ok, RemovedCount = removed }); + } + /// public override Task AppendSample(TimeAppendSampleRequest request, ServerCallContext context) { diff --git a/tests/Robotico.Nexus.Client.Tests/Time/NexusTimeSessionTests.cs b/tests/Robotico.Nexus.Client.Tests/Time/NexusTimeSessionTests.cs index b9b792c..9851526 100644 --- a/tests/Robotico.Nexus.Client.Tests/Time/NexusTimeSessionTests.cs +++ b/tests/Robotico.Nexus.Client.Tests/Time/NexusTimeSessionTests.cs @@ -1,4 +1,5 @@ using Robotico.Nexus.Client.Tests.TestSupport; +using Robotico.Nexus.Client.Time; using Robotico.Result.Errors; using Robotico.Time; using Xunit; @@ -230,9 +231,36 @@ public class NexusTimeSessionTests : IAsyncLifetime Assert.Single(hits!); } + [Fact] + public void ListSeries_GetCardinality_ApplyRetention_CatalogOps() + { + NexusTimeSession nexus = Assert.IsType(_session); + + ulong s1 = RegisterSeries("cpu.load"); + _session.AppendSample(s1, 100, 1.0); + _session.AppendSample(s1, 200, 2.0); + ulong s2 = RegisterSeries("cpu.load"); + RegisterSeries("mem.used"); + + Assert.True(nexus.ListSeries(10).IsSuccess(out IReadOnlyList? listed, out _)); + Assert.Equal(3, listed!.Count); + Assert.Contains(listed, e => e is { SeriesId: var id, MetricName: "cpu.load", SampleCount: 2 } && id == s1); + + Assert.True(nexus.GetCardinality(10).IsSuccess(out TimeCardinalitySnapshot? card, out _)); + Assert.Equal(3, card!.ActiveSeriesCount); + Assert.Equal(2, card.TotalSampleCount); + Assert.Contains(card.TopMetricsBySeriesCount, m => m is { MetricName: "cpu.load", SeriesCount: 2 }); + + Assert.True(nexus.ApplyRetention(150).IsSuccess(out long removed, out _)); + Assert.True(removed >= 1); + Assert.True(nexus.ApplyRetention(0).IsError(out IError? retErr)); + Assert.Equal(TimeErrors.InvalidArgument, retErr!.Code); + } + [Fact] public void AllOperations_Fail_AfterDispose() { + NexusTimeSession nexus = Assert.IsType(_session); _session.Dispose(); Assert.True(_session.Doctor().IsError(out IError? e1)); @@ -267,6 +295,13 @@ public class NexusTimeSessionTests : IAsyncLifetime Assert.True(_session.SearchFullText(1, 0, "x", 10).IsError(out IError? e11)); Assert.Equal(NexusErrors.Disposed, e11!.Code); + + Assert.True(nexus.ListSeries().IsError(out IError? e12)); + Assert.Equal(NexusErrors.Disposed, e12!.Code); + Assert.True(nexus.GetCardinality().IsError(out IError? e13)); + Assert.Equal(NexusErrors.Disposed, e13!.Code); + Assert.True(nexus.ApplyRetention(1).IsError(out IError? e14)); + Assert.Equal(NexusErrors.Disposed, e14!.Code); } private ulong RegisterSeries(string metricName)