Add APM server ingestion channel

Added `ApmChannel` class that extends the `TransportChannelBase` to provide event ingestion to APM server, which sends V2 intake API data.

Also updated the implementation of `IMetricSet`, `ITransaction`, `IError` and `ISpan` to include an `IIntakeRoot` interface which is used to define the incoming root JSON structure to APM server.

Updated several dependency versions in some projects. Also added reference to `Elastic.Ingest.Transport` in `Elastic.Apm.csproj`, enabling data transport using our shared infrastructure.

Added new files `ApmChannelOptions` and `IIntakeRoot` to manage channel options and declare the interface for incoming JSON data structure respectively. `IntakeErrorItem` and `EventIntakeResponse` are created to handle responses from ingestion processes.

Author: Mpdreamz

build/scripts/Build.fs

@@ -21,6 +21,7 @@ open Tooling
 
 module Build =
 
+    //TODO remove oldDiagnosticSourceVersion
     let private oldDiagnosticSourceVersion = SemVer.parse "4.6.0"
     let private diagnosticSourceVersion6 = SemVer.parse "6.0.0"
 

src/Elastic.Apm/Api/IError.cs

@@ -12,7 +12,7 @@ namespace Elastic.Apm.Api
 	/// Represents an error which was captured by the agent.
 	/// </summary>
 	[Specification("error.json")]
-	public interface IError
+	public interface IError : IIntakeRoot
 	{
 		/// <summary>
 		/// The culprit that caused this error.

src/Elastic.Apm/Api/IIntakeRoot.cs

@@ -0,0 +1,8 @@
+// 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
+
+namespace Elastic.Apm.Api;
+
+public interface IIntakeRoot { }

src/Elastic.Apm/Api/IMetricSet.cs

@@ -12,7 +12,7 @@ namespace Elastic.Apm.Api
 	/// Data captured by the agent representing a metric occurring in a monitored service
 	/// </summary>
 	[Specification("metricset.json")]
-	public interface IMetricSet
+	public interface IMetricSet : IIntakeRoot
 	{
 		/// <summary>
 		/// List of captured metrics as key - value pairs

src/Elastic.Apm/Api/ISpan.cs

@@ -10,7 +10,7 @@ namespace Elastic.Apm.Api
 	/// An event captured by an agent occurring in a monitored service
 	/// </summary>
 	[Specification("span.json")]
-	public interface ISpan : IExecutionSegment
+	public interface ISpan : IExecutionSegment, IIntakeRoot
 	{
 		/// <summary>
 		/// The action of the span.

src/Elastic.Apm/Api/ITransaction.cs

@@ -19,7 +19,7 @@ namespace Elastic.Apm.Api
 	/// provide different transaction implementations.
 	/// </remarks>
 	[Specification("transaction.json")]
-	public interface ITransaction : IExecutionSegment
+	public interface ITransaction : IExecutionSegment, IIntakeRoot
 	{
 		/// <summary>
 		/// Contains data related to FaaS (Function as a Service) events.

src/Elastic.Apm/Elastic.Apm.csproj

@@ -68,7 +68,8 @@
   </ItemGroup>
 
   <ItemGroup Condition="'$(DiagnosticSourceVersion)' == ''">
-    <PackageReference Include="System.Diagnostics.DiagnosticSource" Version="5.0.0" />
+    <!--- TODO BUMP Elastic.Transport down to 6.0.0 -->
+    <PackageReference Include="System.Diagnostics.DiagnosticSource" Version="6.0.1" />
   </ItemGroup>
   <!-- DiagnosticSourceVersion MsBuild property can be used to compile the agent against a specific version of
        System.Diagnostics.DiagnosticSource. Used when creating the ElasticApmAgentStartupHook zip file to
@@ -84,6 +85,7 @@
     <PackageReference Include="System.Diagnostics.PerformanceCounter" Version="6.0.1" />
   </ItemGroup>
   <ItemGroup>
+    <PackageReference Include="Elastic.Ingest.Transport" Version="0.5.5" />
     <PackageReference Include="System.Threading.Tasks.Dataflow" Version="4.9.0" />
     <!-- Used by Ben.Demystifier -->
     <PackageReference Include="System.Reflection.Metadata" Version="5.0.0" />

src/Elastic.Apm/Ingest/ApmChannel.cs

@@ -0,0 +1,137 @@
+// 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;
+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.Apm.Api;
+using Elastic.Apm.Report;
+using Elastic.Channels;
+using Elastic.Ingest.Transport;
+using Elastic.Transport;
+
+namespace Elastic.Apm.Ingest;
+
+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,
+	};
+}
+
+/// <summary>
+/// An <see cref="TransportChannelBase{TChannelOptions,TEvent,TResponse,TBulkResponseItem}"/> implementation that sends V2 intake API data
+/// to APM server.
+/// </summary>
+public class ApmChannel
+	: TransportChannelBase<ApmChannelOptions, IIntakeRoot, EventIntakeResponse, IntakeErrorItem>
+	, IPayloadSender
+{
+	/// <inheritdoc cref="ApmChannel"/>
+	public ApmChannel(ApmChannelOptions options) : base(options) { }
+
+	void IPayloadSender.QueueError(IError error) => TryWrite(error);
+
+	void IPayloadSender.QueueMetrics(IMetricSet metrics) => TryWrite(metrics);
+
+	void IPayloadSender.QueueSpan(ISpan span) => TryWrite(span);
+
+	void IPayloadSender.QueueTransaction(ITransaction transaction) => TryWrite(transaction);
+
+	//retry if APM server returns 429
+	/// <inheritdoc cref="ResponseItemsBufferedChannelBase{TChannelOptions,TEvent,TResponse,TBulkResponseItem}.Retry"/>
+	protected override bool Retry(EventIntakeResponse response) => response.ApiCallDetails.HttpStatusCode == 429;
+
+	/// <inheritdoc cref="ResponseItemsBufferedChannelBase{TChannelOptions,TEvent,TResponse,TBulkResponseItem}.RetryAllItems"/>
+	protected override bool RetryAllItems(EventIntakeResponse response) => response.ApiCallDetails.HttpStatusCode == 429;
+
+	//APM does not return the status for all events sent. Therefor we always return an empty set for individual items to retry
+	/// <inheritdoc cref="ResponseItemsBufferedChannelBase{TChannelOptions,TEvent,TResponse,TBulkResponseItem}.Zip"/>
+	protected override List<(IIntakeRoot, IntakeErrorItem)> Zip(EventIntakeResponse response, IReadOnlyCollection<IIntakeRoot> page) =>
+		_emptyZip;
+
+	private readonly List<(IIntakeRoot, IntakeErrorItem)> _emptyZip = new();
+
+	/// <inheritdoc cref="ResponseItemsBufferedChannelBase{TChannelOptions,TEvent,TResponse,TBulkResponseItem}.RetryEvent"/>
+	protected override bool RetryEvent((IIntakeRoot, IntakeErrorItem) @event) => false;
+
+	/// <inheritdoc cref="ResponseItemsBufferedChannelBase{TChannelOptions,TEvent,TResponse,TBulkResponseItem}.RejectEvent"/>
+	protected override bool RejectEvent((IIntakeRoot, IntakeErrorItem) @event) => false;
+
+	/// <inheritdoc cref="BufferedChannelBase{TChannelOptions,TEvent,TResponse}.ExportAsync"/>
+	protected override Task<EventIntakeResponse> ExportAsync(HttpTransport transport, ArraySegment<IIntakeRoot> page, CancellationToken ctx = default) =>
+		transport.RequestAsync<EventIntakeResponse>(HttpMethod.POST, "/intake/v2/events",
+			PostData.StreamHandler(page,
+				(_, _) =>
+				{
+					/* NOT USED */
+				},
+				async (b, stream, ctx) => { await WriteBufferToStreamAsync(b, stream, ctx).ConfigureAwait(false); })
+			, ApmChannelStatics.RequestParams, ctx);
+
+	private async Task WriteStanzaToStreamAsync(Stream stream, CancellationToken ctx)
+	{
+		// {"metadata":{"process":{"pid":1234,"title":"/usr/lib/jvm/java-10-openjdk-amd64/bin/java","ppid":1,"argv":["-v"]},
+		// "system":{"architecture":"amd64","detected_hostname":"8ec7ceb99074","configured_hostname":"host1","platform":"Linux","container":{"id":"8ec7ceb990749e79b37f6dc6cd3628633618d6ce412553a552a0fa6b69419ad4"},
+		// "kubernetes":{"namespace":"default","pod":{"uid":"b17f231da0ad128dc6c6c0b2e82f6f303d3893e3","name":"instrumented-java-service"},"node":{"name":"node-name"}}},
+		// "service":{"name":"1234_service-12a3","version":"4.3.0","node":{"configured_name":"8ec7ceb990749e79b37f6dc6cd3628633618d6ce412553a552a0fa6b69419ad4"},"environment":"production","language":{"name":"Java","version":"10.0.2"},
+		// "agent":{"version":"1.10.0","name":"java","ephemeral_id":"e71be9ac-93b0-44b9-a997-5638f6ccfc36"},"framework":{"name":"spring","version":"5.0.0"},"runtime":{"name":"Java","version":"10.0.2"}},"labels":{"group":"experimental","ab_testing":true,"segment":5}}}
+		// TODO cache
+		var p = Process.GetCurrentProcess();
+		var metadata = new
+		{
+			metadata = new
+			{
+				process = new { pid = p.Id, title = p.ProcessName },
+				service = new
+				{
+					name = System.Text.RegularExpressions.Regex.Replace(p.ProcessName, "[^a-zA-Z0-9 _-]", "_"),
+					version = "1.0.0",
+					agent = new { name = "dotnet", version = "0.0.1" }
+				}
+			}
+		};
+		await JsonSerializer.SerializeAsync(stream, metadata, metadata.GetType(), ApmChannelStatics.SerializerOptions, ctx)
+			.ConfigureAwait(false);
+		await stream.WriteAsync(ApmChannelStatics.LineFeed, 0, 1, ctx).ConfigureAwait(false);
+	}
+
+	private async Task WriteBufferToStreamAsync(IReadOnlyCollection<IIntakeRoot> b, Stream stream, CancellationToken ctx)
+	{
+		await WriteStanzaToStreamAsync(stream, ctx).ConfigureAwait(false);
+		foreach (var @event in b)
+		{
+			if (@event == null) continue;
+
+			var type = @event switch
+			{
+				ITransaction => "transaction",
+				ISpan => "span",
+				IError => "error",
+				IMetricSet => "metricset",
+				_ => "unknown"
+			};
+			var dictionary = new Dictionary<string, object>() { { type, @event } };
+
+			await JsonSerializer.SerializeAsync(stream, dictionary, dictionary.GetType(), ApmChannelStatics.SerializerOptions, ctx)
+				.ConfigureAwait(false);
+
+			await stream.WriteAsync(ApmChannelStatics.LineFeed, 0, 1, ctx).ConfigureAwait(false);
+		}
+	}
+}

src/Elastic.Apm/Ingest/ApmChannelOptions.cs

@@ -0,0 +1,18 @@
+// 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 Elastic.Apm.Api;
+using Elastic.Ingest.Transport;
+using Elastic.Transport;
+
+namespace Elastic.Apm.Ingest;
+
+/// <summary>
+/// Channel options for <see cref="ApmChannel"/>
+/// </summary>
+public class ApmChannelOptions : TransportChannelOptionsBase<IIntakeRoot, EventIntakeResponse, IntakeErrorItem>
+{
+	/// <inheritdoc cref="ApmChannelOptions"/>
+	public ApmChannelOptions(HttpTransport transport) : base(transport) { }
+}

src/Elastic.Apm/Ingest/IngestResponse.cs

@@ -0,0 +1,34 @@
+// 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.Collections.Generic;
+using System.Text.Json.Serialization;
+using Elastic.Transport;
+
+namespace Elastic.Apm.Ingest;
+
+/// <summary> </summary>
+public class EventIntakeResponse : TransportResponse
+{
+	/// <summary> </summary>
+	[JsonPropertyName("accepted")]
+	public long Accepted { get; set; }
+
+	/// <summary> </summary>
+	[JsonPropertyName("errors")]
+	//[JsonConverter(typeof(ResponseItemsConverter))]
+	public IReadOnlyCollection<IntakeErrorItem> Errors { get; set; } = null!;
+}
+
+/// <summary> </summary>
+public class IntakeErrorItem
+{
+	/// <summary> </summary>
+	[JsonPropertyName("message")]
+	public string Message { get; set; } = null!;
+
+	/// <summary> </summary>
+	[JsonPropertyName("document")]
+	public string Document { get; set; } = null!;
+}