Skip to content
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.

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

[Exporter.OpenTelemetryProtocol] Clients for profiling exporter #5759

Closed
wants to merge 3 commits into from
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,12 @@
using System.Runtime.CompilerServices;

#if SIGNED
[assembly: InternalsVisibleTo("OpenTelemetry.AutoInstrumentation, PublicKey=00240000048000009400000006020000002400005253413100040000010001008db7c66f4ebdc6aac4196be5ce1ff4b59b020028e6dbd6e46f15aa40b3215975b92d0a8e45aba5f36114a8cb56241fbfa49f4c017e6c62197857e4e9f62451bc23d3a660e20861f95a57f23e20c77d413ad216ff1bb55f94104d4c501e32b03219d8603fb6fa73401c6ae6808c8daa61b9eaee5d2377d3c23c9ca6016c6582d8")]
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@Kielek - What all things you would need to copy if we don't do this?

[assembly: InternalsVisibleTo("OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests, PublicKey=002400000480000094000000060200000024000052534131000400000100010051c1562a090fb0c9f391012a32198b5e5d9a60e9b80fa2d7b434c9e5ccb7259bd606e66f9660676afc6692b8cdc6793d190904551d2103b7b22fa636dcbb8208839785ba402ea08fc00c8f1500ccef28bbf599aa64ffb1e1d5dc1bf3420a3777badfe697856e9d52070a50c3ea5821c80bef17ca3acffa28f89dd413f096f898")]
[assembly: InternalsVisibleTo("Benchmarks, PublicKey=002400000480000094000000060200000024000052534131000400000100010051c1562a090fb0c9f391012a32198b5e5d9a60e9b80fa2d7b434c9e5ccb7259bd606e66f9660676afc6692b8cdc6793d190904551d2103b7b22fa636dcbb8208839785ba402ea08fc00c8f1500ccef28bbf599aa64ffb1e1d5dc1bf3420a3777badfe697856e9d52070a50c3ea5821c80bef17ca3acffa28f89dd413f096f898")]
[assembly: InternalsVisibleTo("MockOpenTelemetryCollector, PublicKey=002400000480000094000000060200000024000052534131000400000100010051c1562a090fb0c9f391012a32198b5e5d9a60e9b80fa2d7b434c9e5ccb7259bd606e66f9660676afc6692b8cdc6793d190904551d2103b7b22fa636dcbb8208839785ba402ea08fc00c8f1500ccef28bbf599aa64ffb1e1d5dc1bf3420a3777badfe697856e9d52070a50c3ea5821c80bef17ca3acffa28f89dd413f096f898")]
#else
[assembly: InternalsVisibleTo("OpenTelemetry.AutoInstrumentation")]
[assembly: InternalsVisibleTo("OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests")]
[assembly: InternalsVisibleTo("Benchmarks")]
[assembly: InternalsVisibleTo("MockOpenTelemetryCollector")]
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,45 @@
// Copyright The OpenTelemetry Authors
// SPDX-License-Identifier: Apache-2.0

using Grpc.Core;
using OtlpCollector = OpenTelemetry.Proto.Collector.Profiles.V1Experimental;

namespace OpenTelemetry.Exporter.OpenTelemetryProtocol.Implementation.ExportClient;

/// <summary>Class for sending OTLP Logs export request over gRPC.</summary>
internal sealed class OtlpGrpcProfilesExportClient : BaseOtlpGrpcExportClient<OtlpCollector.ExportProfilesServiceRequest>
{
private readonly OtlpCollector.ProfilesService.ProfilesServiceClient profilesClient;

public OtlpGrpcProfilesExportClient(OtlpExporterOptions options, OtlpCollector.ProfilesService.ProfilesServiceClient? profilesServiceClient = null)
: base(options)
{
if (profilesServiceClient != null)
{
this.profilesClient = profilesServiceClient;
}
else
{
this.Channel = options.CreateChannel();
this.profilesClient = new OtlpCollector.ProfilesService.ProfilesServiceClient(this.Channel);
}
}

/// <inheritdoc/>
public override ExportClientResponse SendExportRequest(OtlpCollector.ExportProfilesServiceRequest request, DateTime deadlineUtc, CancellationToken cancellationToken = default)
{
try
{
this.profilesClient.Export(request, headers: this.Headers, deadline: deadlineUtc, cancellationToken: cancellationToken);

// We do not need to return back response and deadline for successful response so using cached value.
return SuccessExportResponse;
}
catch (RpcException ex)
{
OpenTelemetryProtocolExporterEventSource.Log.FailedToReachCollector(this.Endpoint, ex);

return new ExportClientGrpcResponse(success: false, deadlineUtc: deadlineUtc, exception: ex);
}
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,69 @@
// Copyright The OpenTelemetry Authors
// SPDX-License-Identifier: Apache-2.0

using System.Net;
#if NETFRAMEWORK
using System.Net.Http;
#endif
using System.Net.Http.Headers;
using System.Runtime.CompilerServices;
using Google.Protobuf;
using OtlpCollector = OpenTelemetry.Proto.Collector.Profiles.V1Experimental;

namespace OpenTelemetry.Exporter.OpenTelemetryProtocol.Implementation.ExportClient;

/// <summary>Class for sending OTLP log export request over HTTP.</summary>
internal sealed class OtlpHttpProfilesExportClient : BaseOtlpHttpExportClient<OtlpCollector.ExportProfilesServiceRequest>
{
internal const string MediaContentType = "application/x-protobuf";
private const string LogsExportPath = "v1experimental/profiles";

public OtlpHttpProfilesExportClient(OtlpExporterOptions options, HttpClient httpClient)
: base(options, httpClient, LogsExportPath)
{
}

protected override HttpContent CreateHttpContent(OtlpCollector.ExportProfilesServiceRequest exportRequest)
{
return new ExportRequestContent(exportRequest);
}

internal sealed class ExportRequestContent : HttpContent
{
private static readonly MediaTypeHeaderValue ProtobufMediaTypeHeader = new(MediaContentType);

private readonly OtlpCollector.ExportProfilesServiceRequest exportRequest;

public ExportRequestContent(OtlpCollector.ExportProfilesServiceRequest exportRequest)
{
this.exportRequest = exportRequest;
this.Headers.ContentType = ProtobufMediaTypeHeader;
}

#if NET
protected override void SerializeToStream(Stream stream, TransportContext? context, CancellationToken cancellationToken)
{
this.SerializeToStreamInternal(stream);
}
#endif

protected override Task SerializeToStreamAsync(Stream stream, TransportContext? context)
{
this.SerializeToStreamInternal(stream);
return Task.CompletedTask;
}

protected override bool TryComputeLength(out long length)
{
// We can't know the length of the content being pushed to the output stream.
length = -1;
return false;
}

[MethodImpl(MethodImplOptions.AggressiveInlining)]
private void SerializeToStreamInternal(Stream stream)
{
this.exportRequest.WriteTo(stream);
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
using OpenTelemetry.Exporter.OpenTelemetryProtocol.Implementation.Transmission;
using LogOtlpCollector = OpenTelemetry.Proto.Collector.Logs.V1;
using MetricsOtlpCollector = OpenTelemetry.Proto.Collector.Metrics.V1;
using ProfilesOtlpCollector = OpenTelemetry.Proto.Collector.Profiles.V1Experimental;
using TraceOtlpCollector = OpenTelemetry.Proto.Collector.Trace.V1;

namespace OpenTelemetry.Exporter;
Expand Down Expand Up @@ -195,6 +196,37 @@ public static THeaders GetHeaders<THeaders>(this OtlpExporterOptions options, Ac
}
}

public static OtlpExporterTransmissionHandler<ProfilesOtlpCollector.ExportProfilesServiceRequest> GetProfilesExportTransmissionHandler(this OtlpExporterOptions options, ExperimentalOptions experimentalOptions)
{
var exportClient = GetProfilesExportClient(options);
double timeoutMilliseconds = exportClient is OtlpHttpProfilesExportClient httpProfilesExportClient
? httpProfilesExportClient.HttpClient.Timeout.TotalMilliseconds
: options.TimeoutMilliseconds;

if (experimentalOptions.EnableInMemoryRetry)
{
return new OtlpExporterRetryTransmissionHandler<ProfilesOtlpCollector.ExportProfilesServiceRequest>(exportClient, timeoutMilliseconds);
}

if (experimentalOptions.EnableDiskRetry)
{
Debug.Assert(!string.IsNullOrEmpty(experimentalOptions.DiskRetryDirectoryPath), $"{nameof(experimentalOptions.DiskRetryDirectoryPath)} is null or empty");

return new OtlpExporterPersistentStorageTransmissionHandler<ProfilesOtlpCollector.ExportProfilesServiceRequest>(
exportClient,
timeoutMilliseconds,
(byte[] data) =>
{
var request = new ProfilesOtlpCollector.ExportProfilesServiceRequest();
request.MergeFrom(data);
return request;
},
Path.Combine(experimentalOptions.DiskRetryDirectoryPath, "profiles"));
}

return new OtlpExporterTransmissionHandler<ProfilesOtlpCollector.ExportProfilesServiceRequest>(exportClient, timeoutMilliseconds);
}

public static IExportClient<TraceOtlpCollector.ExportTraceServiceRequest> GetTraceExportClient(this OtlpExporterOptions options) =>
options.Protocol switch
{
Expand Down Expand Up @@ -225,6 +257,16 @@ public static THeaders GetHeaders<THeaders>(this OtlpExporterOptions options, Ac
_ => throw new NotSupportedException($"Protocol {options.Protocol} is not supported."),
};

public static IExportClient<ProfilesOtlpCollector.ExportProfilesServiceRequest> GetProfilesExportClient(this OtlpExporterOptions options) =>
options.Protocol switch
{
OtlpExportProtocol.Grpc => new OtlpGrpcProfilesExportClient(options),
OtlpExportProtocol.HttpProtobuf => new OtlpHttpProfilesExportClient(
options,
options.HttpClientFactory?.Invoke() ?? throw new InvalidOperationException("OtlpExporterOptions was missing HttpClientFactory or it returned null.")),
_ => throw new NotSupportedException($"Protocol {options.Protocol} is not supported."),
};

public static void TryEnableIHttpClientFactoryIntegration(this OtlpExporterOptions options, IServiceProvider serviceProvider, string httpClientName)
{
if (serviceProvider != null
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -140,6 +140,9 @@ public void AppendPathIfNotPresent_TracesPath_AppendsCorrectly(string inputUri,
[InlineData(OtlpExportProtocol.Grpc, typeof(OtlpGrpcLogExportClient), false, 10000, null)]
[InlineData(OtlpExportProtocol.HttpProtobuf, typeof(OtlpHttpLogExportClient), false, 10000, null)]
[InlineData(OtlpExportProtocol.HttpProtobuf, typeof(OtlpHttpLogExportClient), true, 8000, null)]
[InlineData(OtlpExportProtocol.Grpc, typeof(OtlpGrpcProfilesExportClient), false, 10000, null)]
[InlineData(OtlpExportProtocol.HttpProtobuf, typeof(OtlpHttpProfilesExportClient), false, 10000, null)]
[InlineData(OtlpExportProtocol.HttpProtobuf, typeof(OtlpHttpProfilesExportClient), true, 8000, null)]
[InlineData(OtlpExportProtocol.Grpc, typeof(OtlpGrpcTraceExportClient), false, 10000, "in_memory")]
[InlineData(OtlpExportProtocol.HttpProtobuf, typeof(OtlpHttpTraceExportClient), false, 10000, "in_memory")]
[InlineData(OtlpExportProtocol.HttpProtobuf, typeof(OtlpHttpTraceExportClient), true, 8000, "in_memory")]
Expand All @@ -149,6 +152,9 @@ public void AppendPathIfNotPresent_TracesPath_AppendsCorrectly(string inputUri,
[InlineData(OtlpExportProtocol.Grpc, typeof(OtlpGrpcLogExportClient), false, 10000, "in_memory")]
[InlineData(OtlpExportProtocol.HttpProtobuf, typeof(OtlpHttpLogExportClient), false, 10000, "in_memory")]
[InlineData(OtlpExportProtocol.HttpProtobuf, typeof(OtlpHttpLogExportClient), true, 8000, "in_memory")]
[InlineData(OtlpExportProtocol.Grpc, typeof(OtlpGrpcProfilesExportClient), false, 10000, "in_memory")]
[InlineData(OtlpExportProtocol.HttpProtobuf, typeof(OtlpHttpProfilesExportClient), false, 10000, "in_memory")]
[InlineData(OtlpExportProtocol.HttpProtobuf, typeof(OtlpHttpProfilesExportClient), true, 8000, "in_memory")]
[InlineData(OtlpExportProtocol.Grpc, typeof(OtlpGrpcTraceExportClient), false, 10000, "disk")]
[InlineData(OtlpExportProtocol.HttpProtobuf, typeof(OtlpHttpTraceExportClient), false, 10000, "disk")]
[InlineData(OtlpExportProtocol.HttpProtobuf, typeof(OtlpHttpTraceExportClient), true, 8000, "disk")]
Expand All @@ -158,9 +164,12 @@ public void AppendPathIfNotPresent_TracesPath_AppendsCorrectly(string inputUri,
[InlineData(OtlpExportProtocol.Grpc, typeof(OtlpGrpcLogExportClient), false, 10000, "disk")]
[InlineData(OtlpExportProtocol.HttpProtobuf, typeof(OtlpHttpLogExportClient), false, 10000, "disk")]
[InlineData(OtlpExportProtocol.HttpProtobuf, typeof(OtlpHttpLogExportClient), true, 8000, "disk")]
[InlineData(OtlpExportProtocol.Grpc, typeof(OtlpGrpcProfilesExportClient), false, 10000, "disk")]
[InlineData(OtlpExportProtocol.HttpProtobuf, typeof(OtlpHttpProfilesExportClient), false, 10000, "disk")]
[InlineData(OtlpExportProtocol.HttpProtobuf, typeof(OtlpHttpProfilesExportClient), true, 8000, "disk")]
public void GetTransmissionHandler_InitializesCorrectHandlerExportClientAndTimeoutValue(OtlpExportProtocol protocol, Type exportClientType, bool customHttpClient, int expectedTimeoutMilliseconds, string? retryStrategy)
{
var exporterOptions = new OtlpExporterOptions() { Protocol = protocol };
var exporterOptions = new OtlpExporterOptions { Protocol = protocol };
if (customHttpClient)
{
exporterOptions.HttpClientFactory = () =>
Expand All @@ -185,10 +194,16 @@ public void GetTransmissionHandler_InitializesCorrectHandlerExportClientAndTimeo

AssertTransmissionHandler(transmissionHandler, exportClientType, expectedTimeoutMilliseconds, retryStrategy);
}
else
else if (exportClientType == typeof(OtlpGrpcLogExportClient) || exportClientType == typeof(OtlpHttpLogExportClient))
{
var transmissionHandler = exporterOptions.GetLogsExportTransmissionHandler(new ExperimentalOptions(configuration));

AssertTransmissionHandler(transmissionHandler, exportClientType, expectedTimeoutMilliseconds, retryStrategy);
}
else
{
var transmissionHandler = exporterOptions.GetProfilesExportTransmissionHandler(new ExperimentalOptions(configuration));

AssertTransmissionHandler(transmissionHandler, exportClientType, expectedTimeoutMilliseconds, retryStrategy);
}
}
Expand Down