AWS Lambda #721
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -3,6 +3,7 @@ | |
| <ManagePackageVersionsCentrally>true</ManagePackageVersionsCentrally> | ||
| </PropertyGroup> | ||
| <ItemGroup> | ||
| <PackageVersion Include="Amazon.Lambda.Core" Version="3.1.0" /> | ||
| <PackageVersion Include="Google.Protobuf" Version="3.26.1" /> | ||
| <PackageVersion Include="MartinCostello.Logging.XUnit" Version="0.7.1" /> | ||
| <PackageVersion Include="Microsoft.Bcl.HashCode" Version="6.0.0" /> | ||
|
|
@@ -15,8 +16,11 @@ | |
| <PackageVersion Include="Microsoft.SourceLink.GitHub" Version="8.0.0" /> | ||
| <PackageVersion Include="Microsoft.VisualStudio.Threading.Analyzers" Version="18.7.23" /> | ||
| <PackageVersion Include="NexusRpc" Version="0.3.0" /> | ||
| <PackageVersion Include="OpenTelemetry" Version="1.15.3" /> | ||
|
Contributor
There was a problem hiding this comment. Can you add the new |
||
| <PackageVersion Include="OpenTelemetry.Api" Version="1.15.3" /> | ||
| <PackageVersion Include="OpenTelemetry.Exporter.InMemory" Version="1.15.3" /> | ||
| <PackageVersion Include="OpenTelemetry.Exporter.OpenTelemetryProtocol" Version="1.15.3" /> | ||
| <PackageVersion Include="OpenTelemetry.Extensions.AWS" Version="1.15.1" /> | ||
| <PackageVersion Include="StyleCop.Analyzers" Version="1.2.0-beta.435" /> | ||
| <PackageVersion Include="System.CommandLine" Version="2.0.8" /> | ||
| <PackageVersion Include="System.Diagnostics.DiagnosticSource" Version="7.0.0" /> | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,214 @@ | ||
| using System; | ||
| using System.Collections.Generic; | ||
| using System.Linq; | ||
| using System.Threading; | ||
| using System.Threading.Tasks; | ||
| using OpenTelemetry; | ||
| using OpenTelemetry.Exporter; | ||
| using OpenTelemetry.Resources; | ||
| using OpenTelemetry.Trace; | ||
| using Temporalio.Client.Interceptors; | ||
| using Temporalio.Runtime; | ||
| using TemporalOpenTelemetry = Temporalio.Extensions.OpenTelemetry; | ||
|
|
||
| namespace Temporalio.Extensions.Aws.Lambda.OpenTelemetry | ||
| { | ||
| /// <summary> | ||
| /// OpenTelemetry helpers for Temporal workers running inside AWS Lambda. | ||
| /// </summary> | ||
| public static class LambdaWorkerOpenTelemetry | ||
| { | ||
| private const string DefaultCollectorEndpoint = "http://localhost:4317"; | ||
| private const string DefaultServiceName = "temporal-lambda-worker"; | ||
| private const string OTelExporterOtlpEndpointEnvironmentVariable = | ||
| "OTEL_EXPORTER_OTLP_ENDPOINT"; | ||
|
|
||
| private const string OTelServiceNameEnvironmentVariable = "OTEL_SERVICE_NAME"; | ||
| private const string LambdaFunctionNameEnvironmentVariable = "AWS_LAMBDA_FUNCTION_NAME"; | ||
| private const string ServiceNameResourceAttribute = "service.name"; | ||
|
|
||
| /// <summary> | ||
| /// Configure OpenTelemetry metrics and tracing with AWS Lambda defaults. | ||
| /// </summary> | ||
| /// <param name="config">Lambda worker configuration to mutate.</param> | ||
| /// <param name="options">Optional OpenTelemetry configuration.</param> | ||
| /// <remarks> | ||
| /// This creates an OTLP trace exporter and tracer provider, configures Core SDK metrics | ||
| /// through a Temporal runtime, adds the Temporal tracing interceptor, and registers a | ||
| /// per-invocation shutdown hook to force-flush traces and dispose the tracer provider | ||
| /// before the Lambda invocation ends. | ||
| /// Any existing <see cref="Temporalio.Client.TemporalConnectionOptions.Runtime" /> is | ||
| /// replaced. | ||
| /// </remarks> | ||
| public static void ApplyDefaults( | ||
|
jmaeagle99 marked this conversation as resolved.
|
||
| TemporalLambdaWorkerOptions config, | ||
| LambdaWorkerOpenTelemetryOptions? options = null) | ||
| { | ||
| if (config == null) | ||
| { | ||
| throw new ArgumentNullException(nameof(config)); | ||
| } | ||
|
|
||
| var resolvedOptions = ResolveOptions(options); | ||
| #pragma warning disable CA2000 // The per-invocation shutdown hook owns provider disposal. | ||
| var tracerProvider = CreateTracerProvider(resolvedOptions); | ||
| #pragma warning restore CA2000 | ||
|
|
||
| config.ClientOptions.Interceptors = AddTracingInterceptor( | ||
| config.ClientOptions.Interceptors); | ||
| config.ClientOptions.Runtime = CreateRuntime(resolvedOptions); | ||
| config.AddShutdownHook( | ||
| async cancellationToken => | ||
| { | ||
| // CreateHandler runs configuration once per invocation, so this provider is | ||
| // invocation-scoped rather than warm-container-scoped. ForceFlush is the only | ||
| // bounded part of provider shutdown: Dispose is synchronous and has no | ||
| // cancellation-aware API. Run disposal after the flush attempt so exporting gets | ||
| // first use of the remaining Lambda deadline, while still releasing provider | ||
| // resources before the next warm invocation can accumulate another provider. | ||
| try | ||
| { | ||
| await ForceFlushAsync( | ||
| tracerProvider, | ||
| config.ShutdownDeadlineBuffer, | ||
| cancellationToken).ConfigureAwait(false); | ||
| } | ||
| finally | ||
| { | ||
| tracerProvider.Dispose(); | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Resolve options using process environment variables. | ||
| /// </summary> | ||
| /// <param name="options">Options to resolve.</param> | ||
| /// <returns>Resolved options.</returns> | ||
| internal static ResolvedLambdaWorkerOpenTelemetryOptions ResolveOptions( | ||
| LambdaWorkerOpenTelemetryOptions? options = null) | ||
| { | ||
| options ??= new LambdaWorkerOpenTelemetryOptions(); | ||
| if (options.MetricsExportInterval <= TimeSpan.Zero) | ||
| { | ||
| throw new ArgumentOutOfRangeException( | ||
| nameof(options), | ||
| "MetricsExportInterval must be greater than zero"); | ||
| } | ||
|
|
||
| var serviceName = FirstNonEmpty( | ||
| options.ServiceName, | ||
| Environment.GetEnvironmentVariable(OTelServiceNameEnvironmentVariable), | ||
| Environment.GetEnvironmentVariable(LambdaFunctionNameEnvironmentVariable), | ||
| DefaultServiceName); | ||
| var collectorEndpoint = FirstNonEmpty( | ||
| options.CollectorEndpoint, | ||
| Environment.GetEnvironmentVariable(OTelExporterOtlpEndpointEnvironmentVariable), | ||
| DefaultCollectorEndpoint); | ||
|
|
||
| return new ResolvedLambdaWorkerOpenTelemetryOptions( | ||
| new Uri(collectorEndpoint), | ||
| serviceName, | ||
| options.MetricsExportInterval); | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Force-flush the tracer provider asynchronously. | ||
| /// </summary> | ||
| /// <param name="tracerProvider">Tracer provider to flush.</param> | ||
| /// <param name="shutdownDeadlineBuffer">Maximum time to wait for the flush.</param> | ||
| /// <param name="cancellationToken">Cancellation token.</param> | ||
| /// <returns>A task for the flush.</returns> | ||
| internal static async Task ForceFlushAsync( | ||
| TracerProvider tracerProvider, | ||
| TimeSpan shutdownDeadlineBuffer, | ||
| CancellationToken cancellationToken) | ||
| { | ||
| if (cancellationToken.IsCancellationRequested) | ||
| { | ||
| return; | ||
| } | ||
|
|
||
| var flushTask = Task.Run( | ||
| () => tracerProvider.ForceFlush(ToTimeoutMilliseconds(shutdownDeadlineBuffer))); | ||
| try | ||
| { | ||
| await flushTask.WaitAsync(cancellationToken).ConfigureAwait(false); | ||
| } | ||
| catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) | ||
| { | ||
| flushTask.Forget(); | ||
| } | ||
| } | ||
|
|
||
| private static string FirstNonEmpty(params string?[] values) => | ||
| values.First(value => !string.IsNullOrEmpty(value))!; | ||
|
|
||
| private static TracerProvider CreateTracerProvider( | ||
| ResolvedLambdaWorkerOpenTelemetryOptions options) => | ||
| Sdk.CreateTracerProviderBuilder(). | ||
| AddXRayTraceId(). | ||
| SetResourceBuilder( | ||
| ResourceBuilder.CreateDefault().AddService(options.ServiceName)). | ||
| AddSource( | ||
| TemporalOpenTelemetry.TracingInterceptor.ClientSource.Name, | ||
| TemporalOpenTelemetry.TracingInterceptor.WorkflowsSource.Name, | ||
| TemporalOpenTelemetry.TracingInterceptor.ActivitiesSource.Name, | ||
| TemporalOpenTelemetry.TracingInterceptor.NexusSource.Name). | ||
| AddOtlpExporter(exporterOptions => | ||
| { | ||
| exporterOptions.Endpoint = options.CollectorEndpoint; | ||
| #pragma warning disable CS0618 // ADOT Lambda parity uses OTLP gRPC on localhost:4317. | ||
| exporterOptions.Protocol = OtlpExportProtocol.Grpc; | ||
| #pragma warning restore CS0618 | ||
| }). | ||
| Build(); | ||
|
|
||
| private static List<IClientInterceptor> AddTracingInterceptor( | ||
| IReadOnlyCollection<IClientInterceptor>? interceptors) | ||
| { | ||
| var newInterceptors = interceptors?.ToList() ?? new List<IClientInterceptor>(); | ||
| newInterceptors.Add(new TemporalOpenTelemetry.TracingInterceptor()); | ||
| return newInterceptors; | ||
| } | ||
|
|
||
| private static TemporalRuntime CreateRuntime( | ||
| ResolvedLambdaWorkerOpenTelemetryOptions options) | ||
| { | ||
| var openTelemetryOptions = new Temporalio.Runtime.OpenTelemetryOptions( | ||
| options.CollectorEndpoint) | ||
| { | ||
| MetricsExportInterval = options.MetricsExportInterval, | ||
| Protocol = OpenTelemetryProtocol.Grpc, | ||
| }; | ||
| return new TemporalRuntime(new TemporalRuntimeOptions | ||
| { | ||
| Telemetry = new TelemetryOptions | ||
| { | ||
| Metrics = new MetricsOptions(openTelemetryOptions) | ||
| { | ||
| GlobalTags = new[] | ||
| { | ||
| new KeyValuePair<string, string>( | ||
| ServiceNameResourceAttribute, | ||
| options.ServiceName), | ||
| }, | ||
| }, | ||
| }, | ||
| }); | ||
| } | ||
|
|
||
| private static int ToTimeoutMilliseconds(TimeSpan timeout) | ||
| { | ||
| if (timeout <= TimeSpan.Zero) | ||
| { | ||
| return 0; | ||
| } | ||
| if (timeout.TotalMilliseconds >= int.MaxValue) | ||
| { | ||
| return int.MaxValue; | ||
| } | ||
| return (int)timeout.TotalMilliseconds; | ||
| } | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,27 @@ | ||
| using System; | ||
|
|
||
| namespace Temporalio.Extensions.Aws.Lambda.OpenTelemetry | ||
| { | ||
| /// <summary> | ||
| /// Options for <see cref="LambdaWorkerOpenTelemetry.ApplyDefaults"/>. | ||
| /// </summary> | ||
| public class LambdaWorkerOpenTelemetryOptions | ||
| { | ||
| /// <summary> | ||
| /// Gets or sets how often the Core SDK exports metrics to the collector. | ||
| /// </summary> | ||
| public TimeSpan MetricsExportInterval { get; set; } = TimeSpan.FromSeconds(10); | ||
|
|
||
| /// <summary> | ||
| /// Gets or sets the OpenTelemetry service name. If unset, this falls back to | ||
| /// OTEL_SERVICE_NAME, then AWS_LAMBDA_FUNCTION_NAME, then "temporal-lambda-worker". | ||
| /// </summary> | ||
| public string? ServiceName { get; set; } | ||
|
|
||
| /// <summary> | ||
| /// Gets or sets the OTLP collector endpoint. If unset, this falls back to | ||
| /// OTEL_EXPORTER_OTLP_ENDPOINT, then "http://localhost:4317". | ||
| /// </summary> | ||
| public string? CollectorEndpoint { get; set; } | ||
| } | ||
| } |
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.