Skip to content

Port Obvs.Monitoring.ElasticSearch from NEST to Elastic.Clients.Elasticsearch #98

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
christopherread wants to merge 1 commit into from
Closed

Port Obvs.Monitoring.ElasticSearch from NEST to Elastic.Clients.Elasticsearch #98

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion Directory.Packages.props
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@
<PackageVersion Include="Microsoft.Extensions.DependencyModel" Version="10.0.13"/>
<PackageVersion Include="MsgPack.Cli" Version="1.0.1"/>
<PackageVersion Include="NATS.Client" Version="1.1.8"/>
<PackageVersion Include="NEST" Version="7.17.5"/>
<PackageVersion Include="Elastic.Clients.Elasticsearch" Version="9.5.3"/>
<PackageVersion Include="NetJSON" Version="1.4.7"/>
<PackageVersion Include="NetMQ" Version="4.0.4.5"/>
<PackageVersion Include="Newtonsoft.Json" Version="13.0.4"/>
Expand Down
12 changes: 6 additions & 6 deletions Obvs.Monitoring.ElasticSearch/Configuration/ElasticSearchMonitoringConfig.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -57,7 +57,7 @@ public interface ICanSpecifyElasticSearchMonitoringUri<TMessage, TCommand, TEven
{
ICanAddEndpointOrLoggingOrCorrelationOrCreate<TMessage, TCommand, TEvent, TRequest, TResponse> ConnectToServer(string uri);

ICanAddEndpointOrLoggingOrCorrelationOrCreate<TMessage, TCommand, TEvent, TRequest, TResponse> ConnectToServer(IConnectionPool elasticConnectionPool);
ICanAddEndpointOrLoggingOrCorrelationOrCreate<TMessage, TCommand, TEvent, TRequest, TResponse> ConnectToServer(NodePool elasticConnectionPool);
}

internal class ElasticSearchMonitoringConfig<TMessage, TCommand, TEvent, TRequest, TResponse> :
Expand Down Expand Up @@ -94,18 +94,18 @@ public ICanAddEndpointOrLoggingOrCorrelationOrCreate<TMessage, TCommand, TEvent,
var monitorFactory = new ElasticSearchMonitorFactory<TMessage>(
_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<TMessage, TCommand, TEvent, TRequest, TResponse> ConnectToServer(IConnectionPool elasticConnectionPool)
public ICanAddEndpointOrLoggingOrCorrelationOrCreate<TMessage, TCommand, TEvent, TRequest, TResponse> ConnectToServer(NodePool elasticConnectionPool)
{
var monitorFactory = new ElasticSearchMonitorFactory<TMessage>(
_indexPrefix, _types, _instanceName,
_samplePeriod, Scheduler.Default,
new ElasticClient(new ConnectionSettings(elasticConnectionPool)));
new ElasticsearchClient(new ElasticsearchClientSettings(elasticConnectionPool)));

_config.UsingMonitor(monitorFactory);
return _config;
Expand Down
10 changes: 5 additions & 5 deletions Obvs.Monitoring.ElasticSearch/ElasticSearchMonitor.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
{
Expand All @@ -22,7 +22,7 @@ static class Directions
private readonly string _counterName;
private readonly string _indexPrefix;
private readonly IList<Type> _types;
private readonly IElasticClient _client;
private readonly ElasticsearchClient _client;
private readonly Subject<Tuple<TMessage, TimeSpan>> _queueSent = new Subject<Tuple<TMessage, TimeSpan>>();
private readonly Subject<Tuple<TMessage, TimeSpan>> _queueReceived = new Subject<Tuple<TMessage, TimeSpan>>();
private readonly TimeSpan _samplePeriod;
Expand All @@ -34,7 +34,7 @@ static class Directions
private readonly string _processName;
private readonly IList<ObvsCounter> _emptyList = new List<ObvsCounter>();

public ElasticSearchMonitor(string instanceName, string counterName, string indexPrefix, IList<Type> types, TimeSpan samplePeriod, IScheduler scheduler, IElasticClient client)
public ElasticSearchMonitor(string instanceName, string counterName, string indexPrefix, IList<Type> types, TimeSpan samplePeriod, IScheduler scheduler, ElasticsearchClient client)
{
_messageTypeName = typeof(TMessage).Name;
_instanceName = instanceName;
Expand Down Expand Up @@ -69,9 +69,9 @@ private void Save(IList<ObvsCounter> 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);
}
}
}
Expand Down
6 changes: 3 additions & 3 deletions Obvs.Monitoring.ElasticSearch/ElasticSearchMonitorFactory.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
using System;
using System.Collections.Generic;
using System.Reactive.Concurrency;
using Nest;
using Elastic.Clients.Elasticsearch;

namespace Obvs.Monitoring.ElasticSearch
{
Expand All @@ -12,9 +12,9 @@ public class ElasticSearchMonitorFactory<TMessage> : IMonitorFactory<TMessage>
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<Type> types, string instanceName, TimeSpan samplePeriod, IScheduler scheduler, IElasticClient client)
public ElasticSearchMonitorFactory(string indexPrefx, IList<Type> types, string instanceName, TimeSpan samplePeriod, IScheduler scheduler, ElasticsearchClient client)
{
_indexPrefx = indexPrefx;
_types = types;
Expand Down
2 changes: 1 addition & 1 deletion Obvs.Monitoring.ElasticSearch/Obvs.Monitoring.ElasticSearch.csproj
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@
<Description>Monitoring extension for Obvs, using Elastic.</Description>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="NEST" />
<PackageReference Include="Elastic.Clients.Elasticsearch" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\Obvs\Obvs.csproj" />
Expand Down
1 change: 0 additions & 1 deletion Obvs.Monitoring.ElasticSearch/ObvsCounter.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,4 @@
using System;
using Nest;

namespace Obvs.Monitoring.ElasticSearch
{
Expand Down
90 changes: 69 additions & 21 deletions Obvs.Monitoring.Tests/TestElasticSearchMonitoring.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,22 +1,51 @@
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;

namespace Obvs.Monitoring.Tests
{
public class TestElasticSearchMonitoring
{
private readonly List<ApiCallDetails> _calls = new List<ApiCallDetails>();
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<string, IEnumerable<string>> { ["x-elastic-product"] = new[] { "Elasticsearch" } });
var settings = new ElasticsearchClientSettings(invoker)
.DisableDirectStreaming()
.OnRequestCompleted(_calls.Add);
_elasticClient = new ElasticsearchClient(settings);
}

private IEnumerable<ApiCallDetails> 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<IElasticClient>();
TestScheduler testScheduler = new TestScheduler();

IMonitorFactory<TestMessage> factory = new ElasticSearchMonitorFactory<TestMessage>("indexName", new List<Type>(), "instanceName", TimeSpan.FromSeconds(1), testScheduler, elasticClient);
IMonitorFactory<TestMessage> factory = new ElasticSearchMonitorFactory<TestMessage>("indexName", new List<Type>(), "instanceName", TimeSpan.FromSeconds(1), testScheduler, _elasticClient);

IMonitor<TestMessage> monitor = factory.Create("SomeName");

Expand All @@ -26,98 +55,117 @@ public void ShouldBeAbleToCreateMonitor()
[Fact]
public void ShouldNotAttemptToSaveIfNoMessages()
{
IElasticClient elasticClient = A.Fake<IElasticClient>();
TestScheduler testScheduler = new TestScheduler();

IMonitorFactory<TestMessage> factory = new ElasticSearchMonitorFactory<TestMessage>("indexName", new List<Type>(), "instanceName", TimeSpan.FromSeconds(1), testScheduler, elasticClient);
IMonitorFactory<TestMessage> factory = new ElasticSearchMonitorFactory<TestMessage>("indexName", new List<Type>(), "instanceName", TimeSpan.FromSeconds(1), testScheduler, _elasticClient);

IMonitor<TestMessage> monitor = factory.Create("SomeName");

Assert.NotNull(monitor);

testScheduler.AdvanceBy(TimeSpan.FromMinutes(1).Ticks);

A.CallTo(() => elasticClient.Bulk(A<IBulkRequest>._)).MustNotHaveHappened();
Assert.Empty(BulkRequests());
}

[Fact]
public void ShouldAttemptToSaveIfMessagesSent()
{
IElasticClient elasticClient = A.Fake<IElasticClient>();
TestScheduler testScheduler = new TestScheduler();

IMonitorFactory<TestMessage> factory = new ElasticSearchMonitorFactory<TestMessage>("indexName", new List<Type>(), "instanceName", TimeSpan.FromSeconds(1), testScheduler, elasticClient);
IMonitorFactory<TestMessage> factory = new ElasticSearchMonitorFactory<TestMessage>("indexName", new List<Type>(), "instanceName", TimeSpan.FromSeconds(1), testScheduler, _elasticClient);

IMonitor<TestMessage> monitor = factory.Create("SomeName");

monitor.MessageSent(new TestMessage(), TimeSpan.FromMilliseconds(1));

testScheduler.AdvanceBy(TimeSpan.FromMinutes(1).Ticks);

A.CallTo(() => elasticClient.Bulk(A<IBulkRequest>._)).MustHaveHappened(1, Times.Exactly);
Assert.Single(BulkRequests());
}

[Fact]
public void ShouldAttemptToSaveIfMessagesReceived()
{
IElasticClient elasticClient = A.Fake<IElasticClient>();
TestScheduler testScheduler = new TestScheduler();

IMonitorFactory<TestMessage> factory = new ElasticSearchMonitorFactory<TestMessage>("indexName", new List<Type>(), "instanceName", TimeSpan.FromSeconds(1), testScheduler, elasticClient);
IMonitorFactory<TestMessage> factory = new ElasticSearchMonitorFactory<TestMessage>("indexName", new List<Type>(), "instanceName", TimeSpan.FromSeconds(1), testScheduler, _elasticClient);

IMonitor<TestMessage> monitor = factory.Create("SomeName");

monitor.MessageReceived(new TestMessage(), TimeSpan.FromMilliseconds(1));

testScheduler.AdvanceBy(TimeSpan.FromMinutes(1).Ticks);

A.CallTo(() => elasticClient.Bulk(A<IBulkRequest>._)).MustHaveHappened(1, Times.Exactly);
Assert.Single(BulkRequests());
}


[Fact]
public void ShouldAttemptToSaveMultipleCountersIfMessagesSent()
{
IElasticClient elasticClient = A.Fake<IElasticClient>();
TestScheduler testScheduler = new TestScheduler();

var types = new List<Type> { typeof(TestMessage) };
IMonitorFactory<TestMessage> factory = new ElasticSearchMonitorFactory<TestMessage>("indexName", types, "instanceName", TimeSpan.FromSeconds(1), testScheduler, elasticClient);
IMonitorFactory<TestMessage> factory = new ElasticSearchMonitorFactory<TestMessage>("indexName", types, "instanceName", TimeSpan.FromSeconds(1), testScheduler, _elasticClient);

IMonitor<TestMessage> monitor = factory.Create("SomeName");

monitor.MessageSent(new TestMessage(), TimeSpan.FromMilliseconds(1));

testScheduler.AdvanceBy(TimeSpan.FromMinutes(1).Ticks);

A.CallTo(() => elasticClient.Bulk(A<IBulkRequest>.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<IElasticClient>();
TestScheduler testScheduler = new TestScheduler();

var types = new List<Type> {typeof(TestMessage)};
IMonitorFactory<TestMessage> factory = new ElasticSearchMonitorFactory<TestMessage>("indexName", types, "instanceName", TimeSpan.FromSeconds(1), testScheduler, elasticClient);
IMonitorFactory<TestMessage> factory = new ElasticSearchMonitorFactory<TestMessage>("indexName", types, "instanceName", TimeSpan.FromSeconds(1), testScheduler, _elasticClient);

IMonitor<TestMessage> monitor = factory.Create("SomeName");

monitor.MessageReceived(new TestMessage(), TimeSpan.FromMilliseconds(1));

testScheduler.AdvanceBy(TimeSpan.FromMinutes(1).Ticks);

A.CallTo(() => elasticClient.Bulk(A<IBulkRequest>.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<TestMessage> factory = new ElasticSearchMonitorFactory<TestMessage>("indexName", new List<Type>(), "instanceName", TimeSpan.FromSeconds(1), testScheduler, _elasticClient);

IMonitor<TestMessage> 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<IElasticClient>();
TestScheduler testScheduler = new TestScheduler();

IMonitorFactory<TestMessage> factory = new ElasticSearchMonitorFactory<TestMessage>("indexName", new List<Type>(), "instanceName", TimeSpan.FromSeconds(1), testScheduler, elasticClient);
IMonitorFactory<TestMessage> factory = new ElasticSearchMonitorFactory<TestMessage>("indexName", new List<Type>(), "instanceName", TimeSpan.FromSeconds(1), testScheduler, _elasticClient);

IMonitor<TestMessage> monitor = factory.Create("SomeName");

Expand Down
Loading