From 8ec35a5e5ee8c1eb91115fa9c591df1ed2fc87f4 Mon Sep 17 00:00:00 2001 From: Christopher Read <1907605+christopherread@users.noreply.github.com> Date: Sat, 24 Oct 2026 23:50:38 -0700 Subject: [PATCH] Port Obvs.Monitoring.ElasticSearch from NEST to Elastic.Clients.Elasticsearch NEST is deprecated and can't talk to Elasticsearch 9. Use Elastic.Clients.Elasticsearch 9.5.3: ElasticsearchClient replaces IElasticClient and NodePool replaces IConnectionPool in the public API. Counters are still bulk-indexed into -yyyy.MM.dd with camelCase fields, so existing indexes keep working. The tests now run the real client against Elastic's in-memory transport and assert on the bulk requests it sends, as the new client has no interface to fake. Fixes #77 Co-Authored-By: Claude Opus 5.5 --- Directory.Packages.props | 2 +- .../ElasticSearchMonitoringConfig.cs | 12 +-- .../ElasticSearchMonitor.cs | 10 +-- .../ElasticSearchMonitorFactory.cs | 6 +- .../Obvs.Monitoring.ElasticSearch.csproj | 2 +- Obvs.Monitoring.ElasticSearch/ObvsCounter.cs | 1 - .../TestElasticSearchMonitoring.cs | 90 ++++++++++++++----- 7 files changed, 85 insertions(+), 38 deletions(-) diff --git a/Directory.Packages.props b/Directory.Packages.props index ec3fc538..a9df3916 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -11,7 +11,7 @@ - + diff --git a/Obvs.Monitoring.ElasticSearch/Configuration/ElasticSearchMonitoringConfig.cs b/Obvs.Monitoring.ElasticSearch/Configuration/ElasticSearchMonitoringConfig.cs index 9c25cab0..3c17c37d 100644 --- a/Obvs.Monitoring.ElasticSearch/Configuration/ElasticSearchMonitoringConfig.cs +++ b/Obvs.Monitoring.ElasticSearch/Configuration/ElasticSearchMonitoringConfig.cs @@ -1,8 +1,8 @@ using System; using System.Collections.Generic; using System.Reactive.Concurrency; -using Elasticsearch.Net; -using Nest; +using Elastic.Clients.Elasticsearch; +using Elastic.Transport; using Obvs.Configuration; namespace Obvs.Monitoring.ElasticSearch.Configuration @@ -57,7 +57,7 @@ public interface ICanSpecifyElasticSearchMonitoringUri ConnectToServer(string uri); - ICanAddEndpointOrLoggingOrCorrelationOrCreate ConnectToServer(IConnectionPool elasticConnectionPool); + ICanAddEndpointOrLoggingOrCorrelationOrCreate ConnectToServer(NodePool elasticConnectionPool); } internal class ElasticSearchMonitoringConfig : @@ -94,18 +94,18 @@ public ICanAddEndpointOrLoggingOrCorrelationOrCreate( _indexPrefix, _types, _instanceName, _samplePeriod, Scheduler.Default, - new ElasticClient(new ConnectionSettings(new Uri(uri)))); + new ElasticsearchClient(new ElasticsearchClientSettings(new Uri(uri)))); _config.UsingMonitor(monitorFactory); return _config; } - public ICanAddEndpointOrLoggingOrCorrelationOrCreate ConnectToServer(IConnectionPool elasticConnectionPool) + public ICanAddEndpointOrLoggingOrCorrelationOrCreate ConnectToServer(NodePool elasticConnectionPool) { var monitorFactory = new ElasticSearchMonitorFactory( _indexPrefix, _types, _instanceName, _samplePeriod, Scheduler.Default, - new ElasticClient(new ConnectionSettings(elasticConnectionPool))); + new ElasticsearchClient(new ElasticsearchClientSettings(elasticConnectionPool))); _config.UsingMonitor(monitorFactory); return _config; diff --git a/Obvs.Monitoring.ElasticSearch/ElasticSearchMonitor.cs b/Obvs.Monitoring.ElasticSearch/ElasticSearchMonitor.cs index 5d2d18df..b94da2f2 100644 --- a/Obvs.Monitoring.ElasticSearch/ElasticSearchMonitor.cs +++ b/Obvs.Monitoring.ElasticSearch/ElasticSearchMonitor.cs @@ -6,7 +6,7 @@ using System.Reactive.Concurrency; using System.Reactive.Linq; using System.Reactive.Subjects; -using Nest; +using Elastic.Clients.Elasticsearch; namespace Obvs.Monitoring.ElasticSearch { @@ -22,7 +22,7 @@ static class Directions private readonly string _counterName; private readonly string _indexPrefix; private readonly IList _types; - private readonly IElasticClient _client; + private readonly ElasticsearchClient _client; private readonly Subject> _queueSent = new Subject>(); private readonly Subject> _queueReceived = new Subject>(); private readonly TimeSpan _samplePeriod; @@ -34,7 +34,7 @@ static class Directions private readonly string _processName; private readonly IList _emptyList = new List(); - public ElasticSearchMonitor(string instanceName, string counterName, string indexPrefix, IList types, TimeSpan samplePeriod, IScheduler scheduler, IElasticClient client) + public ElasticSearchMonitor(string instanceName, string counterName, string indexPrefix, IList types, TimeSpan samplePeriod, IScheduler scheduler, ElasticsearchClient client) { _messageTypeName = typeof(TMessage).Name; _instanceName = instanceName; @@ -69,9 +69,9 @@ private void Save(IList items) { string indexName = string.Format("{0}-{1}", _indexPrefix, DateTime.Today.ToString("yyyy.MM.dd")); var response = _client.IndexMany(items, indexName); - if (!response.IsValid && response.ServerError != null) + if (!response.IsValidResponse && response.ElasticsearchServerError != null) { - Console.WriteLine(response.ServerError.Error); + Console.WriteLine(response.ElasticsearchServerError.Error); } } } diff --git a/Obvs.Monitoring.ElasticSearch/ElasticSearchMonitorFactory.cs b/Obvs.Monitoring.ElasticSearch/ElasticSearchMonitorFactory.cs index cd61666e..ffc09d89 100644 --- a/Obvs.Monitoring.ElasticSearch/ElasticSearchMonitorFactory.cs +++ b/Obvs.Monitoring.ElasticSearch/ElasticSearchMonitorFactory.cs @@ -1,7 +1,7 @@ using System; using System.Collections.Generic; using System.Reactive.Concurrency; -using Nest; +using Elastic.Clients.Elasticsearch; namespace Obvs.Monitoring.ElasticSearch { @@ -12,9 +12,9 @@ public class ElasticSearchMonitorFactory : IMonitorFactory private readonly string _instanceName; private readonly TimeSpan _samplePeriod; private readonly IScheduler _scheduler; - private readonly IElasticClient _client; + private readonly ElasticsearchClient _client; - public ElasticSearchMonitorFactory(string indexPrefx, IList types, string instanceName, TimeSpan samplePeriod, IScheduler scheduler, IElasticClient client) + public ElasticSearchMonitorFactory(string indexPrefx, IList types, string instanceName, TimeSpan samplePeriod, IScheduler scheduler, ElasticsearchClient client) { _indexPrefx = indexPrefx; _types = types; diff --git a/Obvs.Monitoring.ElasticSearch/Obvs.Monitoring.ElasticSearch.csproj b/Obvs.Monitoring.ElasticSearch/Obvs.Monitoring.ElasticSearch.csproj index cc9be8ab..ab011d8e 100644 --- a/Obvs.Monitoring.ElasticSearch/Obvs.Monitoring.ElasticSearch.csproj +++ b/Obvs.Monitoring.ElasticSearch/Obvs.Monitoring.ElasticSearch.csproj @@ -11,7 +11,7 @@ Monitoring extension for Obvs, using Elastic. - + diff --git a/Obvs.Monitoring.ElasticSearch/ObvsCounter.cs b/Obvs.Monitoring.ElasticSearch/ObvsCounter.cs index 88cc7f3f..0996de6b 100644 --- a/Obvs.Monitoring.ElasticSearch/ObvsCounter.cs +++ b/Obvs.Monitoring.ElasticSearch/ObvsCounter.cs @@ -1,5 +1,4 @@ using System; -using Nest; namespace Obvs.Monitoring.ElasticSearch { diff --git a/Obvs.Monitoring.Tests/TestElasticSearchMonitoring.cs b/Obvs.Monitoring.Tests/TestElasticSearchMonitoring.cs index 3835d8b8..31231d89 100644 --- a/Obvs.Monitoring.Tests/TestElasticSearchMonitoring.cs +++ b/Obvs.Monitoring.Tests/TestElasticSearchMonitoring.cs @@ -1,8 +1,10 @@ using System; using System.Collections.Generic; -using FakeItEasy; +using System.Linq; +using System.Text; +using Elastic.Clients.Elasticsearch; +using Elastic.Transport; using Microsoft.Reactive.Testing; -using Nest; using Obvs.Monitoring.ElasticSearch; using Xunit; @@ -10,13 +12,40 @@ namespace Obvs.Monitoring.Tests { public class TestElasticSearchMonitoring { + private readonly List _calls = new List(); + private readonly ElasticsearchClient _elasticClient; + + public TestElasticSearchMonitoring() + { + // Send requests to an in-memory transport that answers like a real server, and record them + var invoker = new InMemoryRequestInvoker( + Encoding.UTF8.GetBytes("{\"took\":1,\"errors\":false,\"items\":[]}"), 200, null, "application/json", + new Dictionary> { ["x-elastic-product"] = new[] { "Elasticsearch" } }); + var settings = new ElasticsearchClientSettings(invoker) + .DisableDirectStreaming() + .OnRequestCompleted(_calls.Add); + _elasticClient = new ElasticsearchClient(settings); + } + + private IEnumerable BulkRequests() + { + return _calls.Where(call => call.Uri.AbsolutePath.EndsWith("/_bulk")); + } + + // A bulk request body has an action line then a document line for each indexed counter + private static int IndexOperations(ApiCallDetails call) + { + return Encoding.UTF8.GetString(call.RequestBodyInBytes) + .Split('\n') + .Count(line => line.StartsWith("{\"index\"")); + } + [Fact] public void ShouldBeAbleToCreateMonitor() { - IElasticClient elasticClient = A.Fake(); TestScheduler testScheduler = new TestScheduler(); - IMonitorFactory factory = new ElasticSearchMonitorFactory("indexName", new List(), "instanceName", TimeSpan.FromSeconds(1), testScheduler, elasticClient); + IMonitorFactory factory = new ElasticSearchMonitorFactory("indexName", new List(), "instanceName", TimeSpan.FromSeconds(1), testScheduler, _elasticClient); IMonitor monitor = factory.Create("SomeName"); @@ -26,10 +55,9 @@ public void ShouldBeAbleToCreateMonitor() [Fact] public void ShouldNotAttemptToSaveIfNoMessages() { - IElasticClient elasticClient = A.Fake(); TestScheduler testScheduler = new TestScheduler(); - IMonitorFactory factory = new ElasticSearchMonitorFactory("indexName", new List(), "instanceName", TimeSpan.FromSeconds(1), testScheduler, elasticClient); + IMonitorFactory factory = new ElasticSearchMonitorFactory("indexName", new List(), "instanceName", TimeSpan.FromSeconds(1), testScheduler, _elasticClient); IMonitor monitor = factory.Create("SomeName"); @@ -37,16 +65,15 @@ public void ShouldNotAttemptToSaveIfNoMessages() testScheduler.AdvanceBy(TimeSpan.FromMinutes(1).Ticks); - A.CallTo(() => elasticClient.Bulk(A._)).MustNotHaveHappened(); + Assert.Empty(BulkRequests()); } [Fact] public void ShouldAttemptToSaveIfMessagesSent() { - IElasticClient elasticClient = A.Fake(); TestScheduler testScheduler = new TestScheduler(); - IMonitorFactory factory = new ElasticSearchMonitorFactory("indexName", new List(), "instanceName", TimeSpan.FromSeconds(1), testScheduler, elasticClient); + IMonitorFactory factory = new ElasticSearchMonitorFactory("indexName", new List(), "instanceName", TimeSpan.FromSeconds(1), testScheduler, _elasticClient); IMonitor monitor = factory.Create("SomeName"); @@ -54,16 +81,15 @@ public void ShouldAttemptToSaveIfMessagesSent() testScheduler.AdvanceBy(TimeSpan.FromMinutes(1).Ticks); - A.CallTo(() => elasticClient.Bulk(A._)).MustHaveHappened(1, Times.Exactly); + Assert.Single(BulkRequests()); } [Fact] public void ShouldAttemptToSaveIfMessagesReceived() { - IElasticClient elasticClient = A.Fake(); TestScheduler testScheduler = new TestScheduler(); - IMonitorFactory factory = new ElasticSearchMonitorFactory("indexName", new List(), "instanceName", TimeSpan.FromSeconds(1), testScheduler, elasticClient); + IMonitorFactory factory = new ElasticSearchMonitorFactory("indexName", new List(), "instanceName", TimeSpan.FromSeconds(1), testScheduler, _elasticClient); IMonitor monitor = factory.Create("SomeName"); @@ -71,18 +97,17 @@ public void ShouldAttemptToSaveIfMessagesReceived() testScheduler.AdvanceBy(TimeSpan.FromMinutes(1).Ticks); - A.CallTo(() => elasticClient.Bulk(A._)).MustHaveHappened(1, Times.Exactly); + Assert.Single(BulkRequests()); } [Fact] public void ShouldAttemptToSaveMultipleCountersIfMessagesSent() { - IElasticClient elasticClient = A.Fake(); TestScheduler testScheduler = new TestScheduler(); var types = new List { typeof(TestMessage) }; - IMonitorFactory factory = new ElasticSearchMonitorFactory("indexName", types, "instanceName", TimeSpan.FromSeconds(1), testScheduler, elasticClient); + IMonitorFactory factory = new ElasticSearchMonitorFactory("indexName", types, "instanceName", TimeSpan.FromSeconds(1), testScheduler, _elasticClient); IMonitor monitor = factory.Create("SomeName"); @@ -90,17 +115,17 @@ public void ShouldAttemptToSaveMultipleCountersIfMessagesSent() testScheduler.AdvanceBy(TimeSpan.FromMinutes(1).Ticks); - A.CallTo(() => elasticClient.Bulk(A.That.Matches(request => request.Operations.Count == 1 + types.Count))).MustHaveHappened(1, Times.Exactly); + var bulk = Assert.Single(BulkRequests()); + Assert.Equal(1 + types.Count, IndexOperations(bulk)); } [Fact] public void ShouldAttemptToSaveMultipleCountersIfMessagesReceived() { - IElasticClient elasticClient = A.Fake(); TestScheduler testScheduler = new TestScheduler(); var types = new List {typeof(TestMessage)}; - IMonitorFactory factory = new ElasticSearchMonitorFactory("indexName", types, "instanceName", TimeSpan.FromSeconds(1), testScheduler, elasticClient); + IMonitorFactory factory = new ElasticSearchMonitorFactory("indexName", types, "instanceName", TimeSpan.FromSeconds(1), testScheduler, _elasticClient); IMonitor monitor = factory.Create("SomeName"); @@ -108,16 +133,39 @@ public void ShouldAttemptToSaveMultipleCountersIfMessagesReceived() testScheduler.AdvanceBy(TimeSpan.FromMinutes(1).Ticks); - A.CallTo(() => elasticClient.Bulk(A.That.Matches(request => request.Operations.Count == 1 + types.Count))).MustHaveHappened(1, Times.Exactly); + var bulk = Assert.Single(BulkRequests()); + Assert.Equal(1 + types.Count, IndexOperations(bulk)); } + [Fact] + public void ShouldIndexCountersInDailyIndexWithCamelCaseFields() + { + TestScheduler testScheduler = new TestScheduler(); + + IMonitorFactory factory = new ElasticSearchMonitorFactory("indexName", new List(), "instanceName", TimeSpan.FromSeconds(1), testScheduler, _elasticClient); + + IMonitor monitor = factory.Create("SomeName"); + + monitor.MessageSent(new TestMessage(), TimeSpan.FromMilliseconds(1)); + + testScheduler.AdvanceBy(TimeSpan.FromMinutes(1).Ticks); + + // Same index naming and document shape as the NEST-based versions, so existing indexes keep working + var bulk = Assert.Single(BulkRequests()); + Assert.Equal($"/indexName-{DateTime.Today:yyyy.MM.dd}/_bulk", bulk.Uri.AbsolutePath); + Assert.True(bulk.HasSuccessfulStatusCode); + var body = Encoding.UTF8.GetString(bulk.RequestBodyInBytes); + Assert.Contains("\"counterName\":\"SomeName\"", body); + Assert.Contains("\"instanceName\":\"instanceName\"", body); + Assert.Contains("\"direction\":\"send\"", body); + } + [Fact] public void ShouldDisposeCleanly() { - IElasticClient elasticClient = A.Fake(); TestScheduler testScheduler = new TestScheduler(); - IMonitorFactory factory = new ElasticSearchMonitorFactory("indexName", new List(), "instanceName", TimeSpan.FromSeconds(1), testScheduler, elasticClient); + IMonitorFactory factory = new ElasticSearchMonitorFactory("indexName", new List(), "instanceName", TimeSpan.FromSeconds(1), testScheduler, _elasticClient); IMonitor monitor = factory.Create("SomeName");