diff --git a/benchmarks/Elastic.Ingest.Elasticsearch.Benchmarks/Benchmarks/BulkRequestCreationForDataStreamBenchmarks.cs b/benchmarks/Elastic.Ingest.Elasticsearch.Benchmarks/Benchmarks/BulkRequestCreationForDataStreamBenchmarks.cs index c44a74a..ef2b2f0 100644 --- a/benchmarks/Elastic.Ingest.Elasticsearch.Benchmarks/Benchmarks/BulkRequestCreationForDataStreamBenchmarks.cs +++ b/benchmarks/Elastic.Ingest.Elasticsearch.Benchmarks/Benchmarks/BulkRequestCreationForDataStreamBenchmarks.cs @@ -4,9 +4,7 @@ using Elastic.Ingest.Elasticsearch.DataStreams; using Elastic.Ingest.Elasticsearch.Serialization; -using Elastic.Transport.Diagnostics; using Performance.Common; -using static Elastic.Transport.HttpMethod; namespace Elastic.Ingest.Elasticsearch.Benchmarks.Benchmarks; @@ -47,10 +45,7 @@ public async Task WriteToStreamAsync() { MemoryStream.Position = 0; var bytes = BulkRequestDataFactory.GetBytes(_data, _options!, _ => _bulkOperationHeader); - var requestData = new RequestData( - POST, "/_bulk", PostData.ReadOnlyMemory(bytes), - _transportConfiguration!, null!, ((ITransportConfiguration)_transportConfiguration!).MemoryStreamFactory, new OpenTelemetryData() - ); - await requestData.PostData.WriteAsync(MemoryStream, _transportConfiguration!, CancellationToken.None); + var postData = PostData.ReadOnlyMemory(bytes); + await postData.WriteAsync(MemoryStream, _transportConfiguration!, false, CancellationToken.None); } } diff --git a/benchmarks/Elastic.Ingest.Elasticsearch.Benchmarks/Benchmarks/BulkRequestCreationWithFixedIndexNameBenchmarks.cs b/benchmarks/Elastic.Ingest.Elasticsearch.Benchmarks/Benchmarks/BulkRequestCreationWithFixedIndexNameBenchmarks.cs index 0d149d4..1b54fab 100644 --- a/benchmarks/Elastic.Ingest.Elasticsearch.Benchmarks/Benchmarks/BulkRequestCreationWithFixedIndexNameBenchmarks.cs +++ b/benchmarks/Elastic.Ingest.Elasticsearch.Benchmarks/Benchmarks/BulkRequestCreationWithFixedIndexNameBenchmarks.cs @@ -3,9 +3,7 @@ // See the LICENSE file in the project root for more information using Elastic.Ingest.Elasticsearch.Serialization; -using Elastic.Transport.Diagnostics; using Performance.Common; -using static Elastic.Transport.HttpMethod; namespace Elastic.Ingest.Elasticsearch.Benchmarks.Benchmarks; @@ -46,10 +44,7 @@ public async Task WriteToStreamAsync() { MemoryStream.Position = 0; var bytes = BulkRequestDataFactory.GetBytes(_data, _options!, e => BulkRequestDataFactory.CreateBulkOperationHeaderForIndex(e, _options!, true)); - var requestData = new RequestData( - POST, "/_bulk", PostData.ReadOnlyMemory(bytes), - _transportConfiguration!, null!, ((ITransportConfiguration)_transportConfiguration!).MemoryStreamFactory, new OpenTelemetryData() - ); - await requestData.PostData.WriteAsync(MemoryStream, _transportConfiguration!, CancellationToken.None); + var postData = PostData.ReadOnlyMemory(bytes); + await postData.WriteAsync(MemoryStream, _transportConfiguration!, false, CancellationToken.None); } } diff --git a/benchmarks/Elastic.Ingest.Elasticsearch.Benchmarks/Benchmarks/BulkRequestCreationWithTemplatedIndexNameBenchmarks.cs b/benchmarks/Elastic.Ingest.Elasticsearch.Benchmarks/Benchmarks/BulkRequestCreationWithTemplatedIndexNameBenchmarks.cs index cb9b80f..e9ea4d1 100644 --- a/benchmarks/Elastic.Ingest.Elasticsearch.Benchmarks/Benchmarks/BulkRequestCreationWithTemplatedIndexNameBenchmarks.cs +++ b/benchmarks/Elastic.Ingest.Elasticsearch.Benchmarks/Benchmarks/BulkRequestCreationWithTemplatedIndexNameBenchmarks.cs @@ -3,9 +3,7 @@ // See the LICENSE file in the project root for more information using Elastic.Ingest.Elasticsearch.Serialization; -using Elastic.Transport.Diagnostics; using Performance.Common; -using static Elastic.Transport.HttpMethod; namespace Elastic.Ingest.Elasticsearch.Benchmarks.Benchmarks; @@ -45,10 +43,7 @@ public async Task DynamicIndexName_WriteToStreamAsync() { MemoryStream.Position = 0; var bytes = BulkRequestDataFactory.GetBytes(_data, _options!, e => BulkRequestDataFactory.CreateBulkOperationHeaderForIndex(e, _options!, false)); - var requestData = new RequestData( - POST, "/_bulk", PostData.ReadOnlyMemory(bytes), - _transportConfiguration!, null!, ((ITransportConfiguration)_transportConfiguration!).MemoryStreamFactory, new OpenTelemetryData() - ); - await requestData.PostData.WriteAsync(MemoryStream, _transportConfiguration!, CancellationToken.None); + var postData = PostData.ReadOnlyMemory(bytes); + await postData.WriteAsync(MemoryStream, _transportConfiguration!, false, CancellationToken.None); } } diff --git a/benchmarks/Elastic.Ingest.Elasticsearch.Benchmarks/Elastic.Ingest.Elasticsearch.Benchmarks.csproj b/benchmarks/Elastic.Ingest.Elasticsearch.Benchmarks/Elastic.Ingest.Elasticsearch.Benchmarks.csproj index ecf0f0b..1633972 100644 --- a/benchmarks/Elastic.Ingest.Elasticsearch.Benchmarks/Elastic.Ingest.Elasticsearch.Benchmarks.csproj +++ b/benchmarks/Elastic.Ingest.Elasticsearch.Benchmarks/Elastic.Ingest.Elasticsearch.Benchmarks.csproj @@ -2,7 +2,7 @@ Exe - net6.0 + net8.0 enable enable pdbonly @@ -10,7 +10,7 @@ - + diff --git a/benchmarks/Elastic.Ingest.Elasticsearch.Benchmarks/Program.cs b/benchmarks/Elastic.Ingest.Elasticsearch.Benchmarks/Program.cs index 2564764..5494e1e 100644 --- a/benchmarks/Elastic.Ingest.Elasticsearch.Benchmarks/Program.cs +++ b/benchmarks/Elastic.Ingest.Elasticsearch.Benchmarks/Program.cs @@ -13,6 +13,7 @@ using BenchmarkDotNet.Running; using System.Globalization; using Elastic.Ingest.Elasticsearch.Benchmarks.Benchmarks; +using Perfolizer.Metrology; #if DEBUG // MANUALLY RUN A BENCHMARKED METHOD DURING DEBUGGING @@ -35,7 +36,7 @@ Console.ReadKey(); #else var config = ManualConfig.Create(DefaultConfig.Instance); -config.SummaryStyle = new SummaryStyle(CultureInfo.CurrentCulture, true, BenchmarkDotNet.Columns.SizeUnit.B, null!, ratioStyle: BenchmarkDotNet.Columns.RatioStyle.Percentage); +config.SummaryStyle = new SummaryStyle(CultureInfo.CurrentCulture, true, SizeUnit.B, null!, ratioStyle: BenchmarkDotNet.Columns.RatioStyle.Percentage); config.AddDiagnoser(MemoryDiagnoser.Default); BenchmarkRunner.Run(config); diff --git a/benchmarks/Elastic.Ingest.Elasticsearch.Profiling/Elastic.Ingest.Elasticsearch.Profiling.csproj b/benchmarks/Elastic.Ingest.Elasticsearch.Profiling/Elastic.Ingest.Elasticsearch.Profiling.csproj index 898d10d..1465c63 100644 --- a/benchmarks/Elastic.Ingest.Elasticsearch.Profiling/Elastic.Ingest.Elasticsearch.Profiling.csproj +++ b/benchmarks/Elastic.Ingest.Elasticsearch.Profiling/Elastic.Ingest.Elasticsearch.Profiling.csproj @@ -2,13 +2,13 @@ Exe - net6.0 + net8.0 enable enable - + diff --git a/benchmarks/Elastic.Ingest.Elasticsearch.Profiling/Program.cs b/benchmarks/Elastic.Ingest.Elasticsearch.Profiling/Program.cs index f740556..7aba134 100644 --- a/benchmarks/Elastic.Ingest.Elasticsearch.Profiling/Program.cs +++ b/benchmarks/Elastic.Ingest.Elasticsearch.Profiling/Program.cs @@ -26,7 +26,7 @@ MemoryProfiler.ForceGc(); MemoryProfiler.CollectAllocations(true); -var configuration = new TransportConfiguration(cloudId, new BasicAuthentication("elastic", elasticPassword)) +var configuration = new TransportConfigurationDescriptor(cloudId, new BasicAuthentication("elastic", elasticPassword)) .ServerCertificateValidationCallback((a,b,c,d) => true); // Trust the local certificate if we're passing through Fiddler with SSL decryption var transport = new DistributedTransport(configuration); diff --git a/benchmarks/Performance.Common/Performance.Common.csproj b/benchmarks/Performance.Common/Performance.Common.csproj index 132c02c..30402ac 100644 --- a/benchmarks/Performance.Common/Performance.Common.csproj +++ b/benchmarks/Performance.Common/Performance.Common.csproj @@ -1,7 +1,7 @@ - net6.0 + net8.0 enable enable diff --git a/build/scripts/CommandLine.fs b/build/scripts/CommandLine.fs index 12a4cc5..545f2b0 100644 --- a/build/scripts/CommandLine.fs +++ b/build/scripts/CommandLine.fs @@ -29,11 +29,11 @@ with interface IArgParserTemplate with member this.Usage = match this with - | Clean _ -> "clean known output locations" - | Build _ -> "Run build" - | Test _ -> "Runs build then tests" - | Release _ -> "runs build, tests, and create and validates the packages shy of publishing them" - | Publish _ -> "Runs the full release" + | Clean -> "clean known output locations" + | Build -> "Run build" + | Test -> "Runs build then tests" + | Release -> "runs build, tests, and create and validates the packages shy of publishing them" + | Publish -> "Runs the full release" | SingleTarget _ -> "Runs the provided sub command without running their dependencies" | Token _ -> "Token to be used to authenticate with github" diff --git a/build/scripts/Targets.fs b/build/scripts/Targets.fs index bd5fd92..698bcea 100644 --- a/build/scripts/Targets.fs +++ b/build/scripts/Targets.fs @@ -49,7 +49,7 @@ let private test (arguments:ParseResults) = let junitOutput = Path.Combine(Paths.Output.FullName, "junit-{assembly}-{framework}-test-results.xml") let loggerPathArgs = sprintf "LogFilePath=%s" junitOutput let loggerArg = sprintf "--logger:\"junit;%s\"" loggerPathArgs - exec "dotnet" ["test"; "-c"; "RELEASE"; loggerArg] |> ignore + exec "dotnet" ["test"; "-c"; "Release"; "--filter"; "FullyQualifiedName!~Elastic.Ingest.Elasticsearch.IntegrationTests"; loggerArg] |> ignore let private generatePackages (arguments:ParseResults) = let output = Paths.RootRelative Paths.Output.FullName diff --git a/build/scripts/scripts.fsproj b/build/scripts/scripts.fsproj index 45890d1..5d884b7 100644 --- a/build/scripts/scripts.fsproj +++ b/build/scripts/scripts.fsproj @@ -2,7 +2,7 @@ Exe - net6.0 + net8.0 false diff --git a/examples/Elastic.Channels.Continuous/Elastic.Channels.Continuous.csproj b/examples/Elastic.Channels.Continuous/Elastic.Channels.Continuous.csproj index 8f82aed..c5f7c2f 100644 --- a/examples/Elastic.Channels.Continuous/Elastic.Channels.Continuous.csproj +++ b/examples/Elastic.Channels.Continuous/Elastic.Channels.Continuous.csproj @@ -2,7 +2,7 @@ Exe - net6.0 + net8.0 enable enable diff --git a/examples/Elastic.Channels.Example/Elastic.Channels.Example.csproj b/examples/Elastic.Channels.Example/Elastic.Channels.Example.csproj index 4a478f1..c250fe3 100644 --- a/examples/Elastic.Channels.Example/Elastic.Channels.Example.csproj +++ b/examples/Elastic.Channels.Example/Elastic.Channels.Example.csproj @@ -2,7 +2,7 @@ Exe - net6.0 + net8.0 enable enable diff --git a/examples/Elastic.Ingest.Apm.Example/Elastic.Ingest.Apm.Example.csproj b/examples/Elastic.Ingest.Apm.Example/Elastic.Ingest.Apm.Example.csproj index 186b39a..387bbac 100644 --- a/examples/Elastic.Ingest.Apm.Example/Elastic.Ingest.Apm.Example.csproj +++ b/examples/Elastic.Ingest.Apm.Example/Elastic.Ingest.Apm.Example.csproj @@ -2,7 +2,7 @@ Exe - net6.0 + net8.0 diff --git a/examples/Elastic.Ingest.Apm.Example/Program.cs b/examples/Elastic.Ingest.Apm.Example/Program.cs index ceb749d..7c58127 100644 --- a/examples/Elastic.Ingest.Apm.Example/Program.cs +++ b/examples/Elastic.Ingest.Apm.Example/Program.cs @@ -29,10 +29,12 @@ private static int Main(string[] args) } var config = new TransportConfiguration(new Uri(args[0])) - .EnableDebugMode() - .Authentication(new ApiKey(args[1])); - //TODO needs - var transport = new DistributedTransport(config); + { + DebugMode = true, + Authentication = new ApiKey(args[1]) + }; + + var transport = new DistributedTransport(config); var numberOfEvents = 800; var maxBufferSize = 200; @@ -48,6 +50,7 @@ private static int Main(string[] args) ExportMaxRetries = 3, ExportBackoffPeriod = times => TimeSpan.FromMilliseconds(1), }; + var channelOptions = new ApmChannelOptions(transport) { BufferOptions = options, @@ -63,9 +66,11 @@ private static int Main(string[] args) ExportRetryCallback = (list) => Interlocked.Increment(ref _retries), ExportExceptionCallback = (e) => _exception = e }; + var channel = new ApmChannel(channelOptions); - string Id() => RandomGenerator.GenerateRandomBytesAsString(8); + static string Id() => RandomGenerator.GenerateRandomBytesAsString(8); + var random = new Random(); for (var i = 0; i < numberOfEvents; i++) { diff --git a/examples/Elastic.Ingest.OpenTelemetry.Example/Elastic.Ingest.OpenTelemetry.Example.csproj b/examples/Elastic.Ingest.OpenTelemetry.Example/Elastic.Ingest.OpenTelemetry.Example.csproj index 93b4b63..970344d 100644 --- a/examples/Elastic.Ingest.OpenTelemetry.Example/Elastic.Ingest.OpenTelemetry.Example.csproj +++ b/examples/Elastic.Ingest.OpenTelemetry.Example/Elastic.Ingest.OpenTelemetry.Example.csproj @@ -2,7 +2,7 @@ Exe - net6.0 + net8.0 diff --git a/examples/playground/playground.csproj b/examples/playground/playground.csproj index 4724cce..028bdd5 100644 --- a/examples/playground/playground.csproj +++ b/examples/playground/playground.csproj @@ -2,7 +2,7 @@ Exe - net6.0 + net8.0 enable enable diff --git a/global.json b/global.json index c317b00..789bff3 100644 --- a/global.json +++ b/global.json @@ -1,6 +1,6 @@ { "sdk": { - "version": "6.0.302", + "version": "8.0.100", "rollForward": "latestFeature", "allowPrerelease": false } diff --git a/src/Elastic.Channels/BufferedChannelBase.cs b/src/Elastic.Channels/BufferedChannelBase.cs index ae2128c..4c038ad 100644 --- a/src/Elastic.Channels/BufferedChannelBase.cs +++ b/src/Elastic.Channels/BufferedChannelBase.cs @@ -405,7 +405,7 @@ async Task FlushBufferAsync() private ValueTask PublishAsync(IOutboundBuffer buffer) { - async Task AsyncSlowPath(IOutboundBuffer b) + async Task AsyncSlowPathAsync(IOutboundBuffer b) { var maxRetries = Options.BufferOptions.ExportMaxRetries; for (var i = 0; i <= maxRetries; i++) @@ -418,7 +418,7 @@ async Task AsyncSlowPath(IOutboundBuffer b) return OutChannel.Writer.TryWrite(buffer) ? new ValueTask(true) - : new ValueTask(AsyncSlowPath(buffer)); + : new ValueTask(AsyncSlowPathAsync(buffer)); } /// > diff --git a/src/Elastic.Ingest.Apm/ApmChannel.cs b/src/Elastic.Ingest.Apm/ApmChannel.cs index caf6077..d9125ba 100644 --- a/src/Elastic.Ingest.Apm/ApmChannel.cs +++ b/src/Elastic.Ingest.Apm/ApmChannel.cs @@ -6,9 +6,7 @@ using System.Collections.Generic; using System.Diagnostics; using System.IO; -using System.Text.Encodings.Web; using System.Text.Json; -using System.Text.Json.Serialization; using System.Threading; using System.Threading.Tasks; using Elastic.Channels; @@ -18,28 +16,13 @@ namespace Elastic.Ingest.Apm; -internal static class ApmChannelStatics -{ - public static readonly byte[] LineFeed = { (byte)'\n' }; - - public static readonly DefaultRequestParameters RequestParams = new() - { - RequestConfiguration = new RequestConfiguration { ContentType = "application/x-ndjson" } - }; - - public static readonly JsonSerializerOptions SerializerOptions = new() - { - DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull, MaxDepth = 64, Encoder = JavaScriptEncoder.UnsafeRelaxedJsonEscaping, - }; -} - /// /// An implementation that sends V2 intake API data /// to APM server. /// public class ApmChannel : TransportChannelBase { - /// +/// public ApmChannel(ApmChannelOptions options) : base(options) { } //retry if APM server returns 429 @@ -64,14 +47,14 @@ public ApmChannel(ApmChannelOptions options) : base(options) { } /// protected override Task ExportAsync(ITransport transport, ArraySegment page, CancellationToken ctx = default) => - transport.RequestAsync(HttpMethod.POST, "/intake/v2/events", + transport.RequestAsync(new EndpointPath(HttpMethod.POST, "/intake/v2/events"), PostData.StreamHandler(page, (_, _) => { /* NOT USED */ }, async (b, stream, ctx) => { await WriteBufferToStreamAsync(b, stream, ctx).ConfigureAwait(false); }) - , ApmChannelStatics.RequestParams, ctx); + , default, ApmChannelStatics.RequestConfig, ctx); private async Task WriteStanzaToStreamAsync(Stream stream, CancellationToken ctx) { @@ -114,7 +97,6 @@ private async Task WriteBufferToStreamAsync(IReadOnlyCollection b }; var dictionary = new Dictionary() { { type, @event } }; - await JsonSerializer.SerializeAsync(stream, dictionary, dictionary.GetType(), ApmChannelStatics.SerializerOptions, ctx) .ConfigureAwait(false); diff --git a/src/Elastic.Ingest.Apm/ApmChannelStatics.cs b/src/Elastic.Ingest.Apm/ApmChannelStatics.cs new file mode 100644 index 0000000..e66f2a9 --- /dev/null +++ b/src/Elastic.Ingest.Apm/ApmChannelStatics.cs @@ -0,0 +1,22 @@ +// Licensed to Elasticsearch B.V under one or more agreements. +// Elasticsearch B.V licenses this file to you under the Apache 2.0 License. +// See the LICENSE file in the project root for more information + +using System.Text.Encodings.Web; +using System.Text.Json; +using System.Text.Json.Serialization; +using Elastic.Transport; + +namespace Elastic.Ingest.Apm; + +internal static class ApmChannelStatics +{ + public static readonly byte[] LineFeed = [(byte)'\n']; + + public static readonly RequestConfiguration RequestConfig = new() { ContentType = "application/x-ndjson" }; + + public static readonly JsonSerializerOptions SerializerOptions = new() + { + DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull, MaxDepth = 64, Encoder = JavaScriptEncoder.UnsafeRelaxedJsonEscaping, + }; +} diff --git a/src/Elastic.Ingest.Elasticsearch/DataStreams/DataStreamChannel.cs b/src/Elastic.Ingest.Elasticsearch/DataStreams/DataStreamChannel.cs index 44cd75b..b485876 100644 --- a/src/Elastic.Ingest.Elasticsearch/DataStreams/DataStreamChannel.cs +++ b/src/Elastic.Ingest.Elasticsearch/DataStreams/DataStreamChannel.cs @@ -24,7 +24,7 @@ public DataStreamChannel(DataStreamChannelOptions options, ICollection options, ICollection protected override string TemplateWildcard => Options.DataStream.GetNamespaceWildcard(); - /// - protected override string BulkUrl => _url; + /// + protected override string BulkPathAndQuery => _url; /// /// Gets a default index template for the current diff --git a/src/Elastic.Ingest.Elasticsearch/ElasticsearchChannelBase.cs b/src/Elastic.Ingest.Elasticsearch/ElasticsearchChannelBase.cs index d391a0f..7dafc43 100644 --- a/src/Elastic.Ingest.Elasticsearch/ElasticsearchChannelBase.cs +++ b/src/Elastic.Ingest.Elasticsearch/ElasticsearchChannelBase.cs @@ -47,7 +47,7 @@ protected override bool Retry(BulkResponse response) /// /// The URL for the bulk request. /// - protected virtual string BulkUrl => "_bulk"; + protected virtual string BulkPathAndQuery => "_bulk?filter_path=error,items.*.status,items.*.error"; /// protected override bool RetryAllItems(BulkResponse response) => response.ApiCallDetails.HttpStatusCode == 429; @@ -75,18 +75,18 @@ protected override Task ExportAsync(ITransport transport, ArraySeg #pragma warning restore CS0618 { var bytes = BulkRequestDataFactory.GetBytes(page, Options, CreateBulkOperationHeader); - return transport.RequestAsync(HttpMethod.POST, BulkUrl, PostData.ReadOnlyMemory(bytes), RequestParams, ctx); + return transport.RequestAsync(HttpMethod.POST, BulkPathAndQuery, PostData.ReadOnlyMemory(bytes), ctx); } #endif #pragma warning disable IDE0022 // Use expression body for method - return transport.RequestAsync(HttpMethod.POST, BulkUrl, + return transport.RequestAsync(new (HttpMethod.POST, BulkPathAndQuery), PostData.StreamHandler(page, (_, _) => { /* NOT USED */ }, async (b, stream, ctx) => { await BulkRequestDataFactory.WriteBufferToStreamAsync(b, stream, Options, CreateBulkOperationHeader, ctx).ConfigureAwait(false); }) - , RequestParams, ctx); + , ctx); #pragma warning restore IDE0022 // Use expression body for method } diff --git a/src/Elastic.Ingest.Elasticsearch/ElasticsearchChannelStatics.cs b/src/Elastic.Ingest.Elasticsearch/ElasticsearchChannelStatics.cs index 13a431a..b0bb302 100644 --- a/src/Elastic.Ingest.Elasticsearch/ElasticsearchChannelStatics.cs +++ b/src/Elastic.Ingest.Elasticsearch/ElasticsearchChannelStatics.cs @@ -7,11 +7,9 @@ using System.Text.Encodings.Web; using System.Text.Json; using System.Text.Json.Serialization; -using Elastic.Transport; namespace Elastic.Ingest.Elasticsearch; -internal class ElasticsearchRequestParameters : RequestParameters { } internal static class ElasticsearchChannelStatics { public static readonly byte[] LineFeed = { (byte)'\n' }; @@ -19,9 +17,6 @@ internal static class ElasticsearchChannelStatics public static readonly byte[] DocUpdateHeaderStart = Encoding.UTF8.GetBytes("{\"doc_as_upsert\": true, \"doc\": "); public static readonly byte[] DocUpdateHeaderEnd = Encoding.UTF8.GetBytes(" }"); - public static readonly ElasticsearchRequestParameters RequestParams = - new() { QueryString = { { "filter_path", "error, items.*.status,items.*.error" } } }; - public static readonly HashSet RetryStatusCodes = new(new[] { 502, 503, 504, 429 }); public static readonly JsonSerializerOptions SerializerOptions = new () diff --git a/src/Elastic.Ingest.Elasticsearch/Indices/IndexChannel.cs b/src/Elastic.Ingest.Elasticsearch/Indices/IndexChannel.cs index 0e471ed..49b0789 100644 --- a/src/Elastic.Ingest.Elasticsearch/Indices/IndexChannel.cs +++ b/src/Elastic.Ingest.Elasticsearch/Indices/IndexChannel.cs @@ -24,13 +24,13 @@ public IndexChannel(IndexChannelOptions options) : this(options, null) { /// public IndexChannel(IndexChannelOptions options, ICollection>? callbackListeners) : base(options, callbackListeners) { - _url = base.BulkUrl; + _url = base.BulkPathAndQuery; // When the configured index format represents a fixed index name, we can optimize by providing a URL with the target index specified. // We can later avoid the overhead of calculating and adding the index name to the operation headers. if (string.Format(Options.IndexFormat, DateTimeOffset.Now).Equals(Options.IndexFormat, StringComparison.Ordinal)) { - _url = $"{Options.IndexFormat}/{base.BulkUrl}"; + _url = $"{Options.IndexFormat}/{base.BulkPathAndQuery}"; _skipIndexNameOnOperations = true; } @@ -38,8 +38,8 @@ public IndexChannel(IndexChannelOptions options, ICollection - protected override string BulkUrl => _url; + /// + protected override string BulkPathAndQuery => _url; /// protected override BulkOperationHeader CreateBulkOperationHeader(TEvent @event) => BulkRequestDataFactory.CreateBulkOperationHeaderForIndex(@event, Options, _skipIndexNameOnOperations); diff --git a/src/Elastic.Ingest.OpenTelemetry/CustomOtlpTraceExporter.cs b/src/Elastic.Ingest.OpenTelemetry/CustomOtlpTraceExporter.cs index 03d8538..2f5070d 100644 --- a/src/Elastic.Ingest.OpenTelemetry/CustomOtlpTraceExporter.cs +++ b/src/Elastic.Ingest.OpenTelemetry/CustomOtlpTraceExporter.cs @@ -37,7 +37,6 @@ public CustomOtlpTraceExporter(OtlpExporterOptions options, TraceChannelOptions } } - /// public class TraceChannelOptions : ChannelOptionsBase { @@ -67,7 +66,7 @@ public TraceChannel(TraceChannelOptions options, ICollection)batchConstructor.Invoke(new[] {buffer, options.BufferOptions.OutboundBufferMaxSize }); return batch; }; - } private Func, Batch> BatchCreator { get; } diff --git a/src/Elastic.Ingest.Transport/Elastic.Ingest.Transport.csproj b/src/Elastic.Ingest.Transport/Elastic.Ingest.Transport.csproj index 4c0bb12..9ef1518 100644 --- a/src/Elastic.Ingest.Transport/Elastic.Ingest.Transport.csproj +++ b/src/Elastic.Ingest.Transport/Elastic.Ingest.Transport.csproj @@ -11,7 +11,7 @@ - + diff --git a/src/Elastic.Ingest.Transport/LibraryVersion.cs b/src/Elastic.Ingest.Transport/LibraryVersion.cs index eb1a026..e8b94ae 100644 --- a/src/Elastic.Ingest.Transport/LibraryVersion.cs +++ b/src/Elastic.Ingest.Transport/LibraryVersion.cs @@ -2,10 +2,6 @@ // Elasticsearch B.V licenses this file to you under the Apache 2.0 License. // See the LICENSE file in the project root for more information -using System; -using System.Diagnostics; -using System.Reflection; -using System.Text.RegularExpressions; using Elastic.Transport; namespace Elastic.Ingest.Transport; diff --git a/tests/Elastic.Channels.Tests/Elastic.Channels.Tests.csproj b/tests/Elastic.Channels.Tests/Elastic.Channels.Tests.csproj index a6bb990..ad81de3 100644 --- a/tests/Elastic.Channels.Tests/Elastic.Channels.Tests.csproj +++ b/tests/Elastic.Channels.Tests/Elastic.Channels.Tests.csproj @@ -1,7 +1,7 @@ - net6.0 + net8.0 diff --git a/tests/Elastic.Ingest.Elasticsearch.IntegrationTests/Elastic.Ingest.Elasticsearch.IntegrationTests.csproj b/tests/Elastic.Ingest.Elasticsearch.IntegrationTests/Elastic.Ingest.Elasticsearch.IntegrationTests.csproj index 80f436f..fca89cc 100644 --- a/tests/Elastic.Ingest.Elasticsearch.IntegrationTests/Elastic.Ingest.Elasticsearch.IntegrationTests.csproj +++ b/tests/Elastic.Ingest.Elasticsearch.IntegrationTests/Elastic.Ingest.Elasticsearch.IntegrationTests.csproj @@ -1,7 +1,7 @@ - net6.0 + net8.0 false enable diff --git a/tests/Elastic.Ingest.Elasticsearch.IntegrationTests/IngestionCluster.cs b/tests/Elastic.Ingest.Elasticsearch.IntegrationTests/IngestionCluster.cs index d0b62fe..eb0f047 100644 --- a/tests/Elastic.Ingest.Elasticsearch.IntegrationTests/IngestionCluster.cs +++ b/tests/Elastic.Ingest.Elasticsearch.IntegrationTests/IngestionCluster.cs @@ -31,37 +31,37 @@ public ElasticsearchClient CreateClient(ITestOutputHelper output, string? hostna var isCi = !string.IsNullOrEmpty(Environment.GetEnvironmentVariable("CI")); var nodes = NodesUris(hostname); var connectionPool = new StaticNodePool(nodes); - var settings = new ElasticsearchClientSettings(connectionPool) - .RequestTimeout(TimeSpan.FromSeconds(5)) - .ServerCertificateValidationCallback(CertificateValidations.AllowAll) - .OnRequestCompleted(d => - { - // ON CI only logged failed requests - // Locally we just log everything for ease of development - try - { - if (isCi) - { - if (!d.HasSuccessfulStatusCode) - output.WriteLine(d.DebugInformation); - } - else output.WriteLine(d.DebugInformation); - } - catch - { - // ignored - } - }) - .EnableDebugMode() - //do not request server stack traces on CI, too noisy - .IncludeServerStackTraceOnError(!isCi); - if (cluster.DetectedProxy != None) - { - var proxyUrl = cluster.DetectedProxy == Fiddler ? "ipv4.fiddler" : "localhost"; - settings = settings.Proxy(new Uri($"http://{proxyUrl}:8080"), null!, null!); - } + //var settings = new ElasticsearchClientSettings(connectionPool) + // .RequestTimeout(TimeSpan.FromSeconds(5)) + // .ServerCertificateValidationCallback(CertificateValidations.AllowAll) + // .OnRequestCompleted(d => + // { + // // ON CI only logged failed requests + // // Locally we just log everything for ease of development + // try + // { + // if (isCi) + // { + // if (!d.HasSuccessfulStatusCode) + // output.WriteLine(d.DebugInformation); + // } + // else output.WriteLine(d.DebugInformation); + // } + // catch + // { + // // ignored + // } + // }) + // .EnableDebugMode() + // //do not request server stack traces on CI, too noisy + // .IncludeServerStackTraceOnError(!isCi); + //if (cluster.DetectedProxy != None) + //{ + // var proxyUrl = cluster.DetectedProxy == Fiddler ? "ipv4.fiddler" : "localhost"; + // settings = settings.Proxy(new Uri($"http://{proxyUrl}:8080"), null!, null!); + //} - return new ElasticsearchClient(settings); + return new ElasticsearchClient(); }); } diff --git a/tests/Elastic.Ingest.Elasticsearch.Tests/ChannelTestWithSingleDocResponseBase.cs b/tests/Elastic.Ingest.Elasticsearch.Tests/ChannelTestWithSingleDocResponseBase.cs index edb4c02..33ec3cd 100644 --- a/tests/Elastic.Ingest.Elasticsearch.Tests/ChannelTestWithSingleDocResponseBase.cs +++ b/tests/Elastic.Ingest.Elasticsearch.Tests/ChannelTestWithSingleDocResponseBase.cs @@ -14,8 +14,7 @@ protected ChannelTestWithSingleDocResponseBase(string url = "https://localhost:9 Transport = new DistributedTransport( new TransportConfiguration(new SingleNodePool(new Uri(url)), new InMemoryRequestInvoker(Encoding.UTF8.GetBytes("{\"items\":[{\"create\":{\"status\":201}}]}"))) - .DisablePing() - .EnableDebugMode()); + { DisablePings = true, DebugMode = true }); protected ITransport Transport { get; } } diff --git a/tests/Elastic.Ingest.Elasticsearch.Tests/Elastic.Ingest.Elasticsearch.Tests.csproj b/tests/Elastic.Ingest.Elasticsearch.Tests/Elastic.Ingest.Elasticsearch.Tests.csproj index 230063d..0747052 100644 --- a/tests/Elastic.Ingest.Elasticsearch.Tests/Elastic.Ingest.Elasticsearch.Tests.csproj +++ b/tests/Elastic.Ingest.Elasticsearch.Tests/Elastic.Ingest.Elasticsearch.Tests.csproj @@ -1,11 +1,11 @@ - net6.0 + net8.0 - + diff --git a/tests/Elastic.Ingest.Elasticsearch.Tests/Setup.cs b/tests/Elastic.Ingest.Elasticsearch.Tests/Setup.cs index 1a9cb12..9272497 100644 --- a/tests/Elastic.Ingest.Elasticsearch.Tests/Setup.cs +++ b/tests/Elastic.Ingest.Elasticsearch.Tests/Setup.cs @@ -16,7 +16,7 @@ namespace Elastic.Ingest.Elasticsearch.Tests; public static class TestSetup { - public static DistributedTransport CreateClient(Func setup) + public static DistributedTransport CreateClient(Func setup) { var cluster = Virtual.Elasticsearch.Bootstrap(numberOfNodes: 1).Ping(c=>c.SucceedAlways()); var virtualSettings = setup(cluster) @@ -26,10 +26,11 @@ public static DistributedTransport CreateClient(Func virtualSettings); //audit.VisualizeCalls(cluster.ClientCallRules.Count); - var settings = new TransportConfiguration(virtualSettings.ConnectionPool, virtualSettings.Connection) + var settings = new TransportConfigurationDescriptor(virtualSettings.ConnectionPool, virtualSettings.Connection) .DisablePing() .EnableDebugMode(); - return new DistributedTransport(settings); + + return new DistributedTransport(settings); } public static ClientCallRule BulkResponse(this ClientCallRule rule, params int[] statusCodes) => @@ -43,7 +44,7 @@ public class TestSession : IDisposable private int _retries; private int _maxRetriesExceeded; - public TestSession(DistributedTransport transport) + public TestSession(DistributedTransport transport) { Transport = transport; BufferOptions = new BufferOptions @@ -73,7 +74,7 @@ public TestSession(DistributedTransport transport) public IndexChannel Channel { get; } - public DistributedTransport Transport { get; } + public DistributedTransport Transport { get; } public IndexChannelOptions ChannelOptions { get; } @@ -101,7 +102,7 @@ public void Dispose() } } - public static TestSession CreateTestSession(DistributedTransport transport) => new(transport); + public static TestSession CreateTestSession(DistributedTransport transport) => new(transport); public static void WriteAndWait(this TestSession session, int events = 1) {