[Monitor Exporter] Multi-tenant trace export (off by default)

sdk/monitor/Azure.Monitor.OpenTelemetry.Exporter/CHANGELOG.md

@@ -6,6 +6,9 @@
 - Add support for project id attributes propagation
   ([#62052](https://github.com/Azure/azure-sdk-for-net/pull/62052))
 
+- Added multi-tenant support for traces, off by default and enabled with the `Azure.Monitor.OpenTelemetry.EnableMultiTenantExport` AppContext switch. When enabled, an Activity carrying the `microsoft.instrumentation_key` and `microsoft.ingestion_endpoint` attributes is sent to that endpoint instead of the exporter's own; Activities without both attributes are dropped. Live Metrics is disabled while the switch is on.
+  ([#62707](https://github.com/Azure/azure-sdk-for-net/pull/62707))
+
 - Shutting down a provider (including `Dispose()`) now writes pending telemetry to offline storage and uploads it in the background instead of blocking on ingestion. Short-lived applications such as CLI tools previously lost this telemetry, because they exit before a transmission completes; the telemetry is now durable before exit, and delivery is completed by a background drain in this or a subsequent run. `ForceFlush` is unchanged by default and can be opted in with the `Azure.Monitor.OpenTelemetry.Exporter.PersistOnForceFlush` AppContext switch, which applies to traces and logs only: a metric reader cannot distinguish a caller's flush from its periodic collection, so metric `ForceFlush` always transmits. The previous behavior can be restored with the `Azure.Monitor.OpenTelemetry.Exporter.DisablePersistOnShutdown` AppContext switch.
   ([#61818](https://github.com/Azure/azure-sdk-for-net/pull/61818))
 
@@ -16,8 +19,12 @@
 
 ### Bugs Fixed
 
+- The ingestion redirect cache is now keyed by the endpoint it was issued for. A redirect returned by one ingestion endpoint could previously be applied to a request bound for another.
+  ([#62707](https://github.com/Azure/azure-sdk-for-net/pull/62707))
+
 - Telemetry left in offline storage by a process that exited during a transmission is no longer stranded permanently. A leased blob is renamed so that it matches neither the storage provider's blob enumeration nor its retention sweep, and the provider only reclaims those leases on a two minute maintenance timer that a short-lived process never reaches. Expired leases are now reclaimed when storage is drained.
   ([#61818](https://github.com/Azure/azure-sdk-for-net/pull/61818))
+  ([#61818](https://github.com/Azure/azure-sdk-for-net/pull/61818))
 
 - Offline storage is now drained shortly after startup rather than only after the process has been running for two minutes, so telemetry persisted by a previous run is uploaded even when no single run is long-lived.
   ([#61818](https://github.com/Azure/azure-sdk-for-net/pull/61818))

sdk/monitor/Azure.Monitor.OpenTelemetry.Exporter/src/AzureMonitorTraceExporter.cs

@@ -7,6 +7,7 @@
 using Azure.Core.Pipeline;
 using Azure.Monitor.OpenTelemetry.Exporter.Internals;
 using Azure.Monitor.OpenTelemetry.Exporter.Internals.Diagnostics;
+using Azure.Monitor.OpenTelemetry.Exporter.Internals.MultiTenant;
 using OpenTelemetry;
 
 namespace Azure.Monitor.OpenTelemetry.Exporter
@@ -17,9 +18,12 @@ namespace Azure.Monitor.OpenTelemetry.Exporter
     public sealed class AzureMonitorTraceExporter : BaseExporter<Activity>
     {
         private readonly ITransmitter _transmitter;
+        private readonly IMultiTenantTransmitter? _multiTenantTransmitter;
         private readonly string _instrumentationKey;
         private readonly float _sampleRate; // This value is recorded on TelemetryItem.SampleRate.
+        private readonly bool _multiTenantEnabled;
         private AzureMonitorResource? _resource;
+        private EndpointRouteBatch? _routeBatch;
         private bool _disposed;
 
         /// <summary>
@@ -31,10 +35,36 @@ public AzureMonitorTraceExporter(AzureMonitorExporterOptions options) : this(opt
         }
 
         internal AzureMonitorTraceExporter(AzureMonitorExporterOptions options, ITransmitter transmitter)
+            : this(options, transmitter, MultiTenantConfig.Enabled)
+        {
+        }
+
+        /// <remarks>
+        /// The gate is a constructor parameter so a test can exercise either path without mutating
+        /// process-wide state that other tests observe.
+        /// </remarks>
+        internal AzureMonitorTraceExporter(AzureMonitorExporterOptions options, ITransmitter transmitter, bool multiTenantEnabled)
         {
             _sampleRate = (float)Math.Round(options.SamplingRatio * 100);
             _transmitter = transmitter;
             _instrumentationKey = transmitter.InstrumentationKey;
+            _multiTenantEnabled = multiTenantEnabled;
+
+            if (_multiTenantEnabled)
+            {
+                if (transmitter is not IMultiTenantTransmitter multiTenantTransmitter)
+                {
+                    // The caller already took a reference on the shared transmitter, which owns
+                    // storage timers and statsbeat, so it has to be released before unwinding.
+                    transmitter.Dispose();
+
+                    throw new NotSupportedException($"Multi-tenant export requires a transmitter implementing {nameof(IMultiTenantTransmitter)}.");
+                }
+
+                _multiTenantTransmitter = multiTenantTransmitter;
+
+                AzureMonitorExporterEventSource.Log.MultiTenantExportEnabled();
+            }
         }
 
         internal AzureMonitorResource? TraceResource => _resource ??= ParentProvider?.GetResource().CreateAzureMonitorResource(_instrumentationKey);
@@ -47,6 +77,11 @@ public override ExportResult Export(in Batch<Activity> batch)
             // Prevent Azure Monitor's HTTP operations from being instrumented.
             using var scope = SuppressInstrumentationScope.Begin();
 
+            if (_multiTenantEnabled)
+            {
+                return ExportMultiTenant(batch);
+            }
+
             ExportResult exportResult = ExportResult.Failure;
 
             try
@@ -65,6 +100,38 @@ public override ExportResult Export(in Batch<Activity> batch)
             return exportResult;
         }
 
+        private ExportResult ExportMultiTenant(in Batch<Activity> batch)
+        {
+            // A concurrent Export takes a fresh batch rather than sharing the cached one.
+            var routeBatch = Interlocked.Exchange(ref _routeBatch, null) ?? new EndpointRouteBatch();
+
+            try
+            {
+                TraceHelper.OtelToAzureMonitorTraceMultiTenant(batch, TraceResource, _sampleRate, routeBatch);
+
+                if (routeBatch.Count == 0)
+                {
+                    // Most tenants do not enable observability, so a batch carrying no routing tags
+                    // is the normal steady state rather than a failed export.
+                    return ExportResult.Success;
+                }
+
+                // Blocks until every group has been sent, so Reset cannot run under a consumer that
+                // still holds a group's item list.
+                return _multiTenantTransmitter!.Track(routeBatch, TelemetryItemOrigin.AzureMonitorTraceExporter, CancellationToken.None);
+            }
+            catch (Exception ex)
+            {
+                AzureMonitorExporterEventSource.Log.FailedToExport(nameof(AzureMonitorTraceExporter), _instrumentationKey, ex);
+                return ExportResult.Failure;
+            }
+            finally
+            {
+                routeBatch.Reset();
+                Interlocked.Exchange(ref _routeBatch, routeBatch);
+            }
+        }
+
         /// <inheritdoc/>
         protected override void Dispose(bool disposing)
         {

sdk/monitor/Azure.Monitor.OpenTelemetry.Exporter/src/Customizations/ApplicationInsightsRestClient.cs

@@ -17,7 +17,9 @@ namespace Azure.Monitor.OpenTelemetry.Exporter
 {
     internal partial class ApplicationInsightsRestClient
     {
-        private RawRequestUriBuilder? _rawRequestUriBuilder;
+        private const string TrackPath = "v2.1/track";
+
+        private Uri? _trackUri;
 
         /// <summary> Initializes a new instance of ApplicationInsightsRestClient with pre-built pipeline. </summary>
         /// <param name="clientDiagnostics"> The handler for diagnostic messaging in the client. </param>
@@ -117,26 +119,124 @@ internal HttpMessage CreateTrackRequest(ReadOnlyMemory<byte> body)
             return CreateRequest(RequestContent.Create(body));
         }
 
+        /// <summary>
+        /// Builds the absolute track URI for an ingestion endpoint supplied at export time, for
+        /// multi-tenant routing where the destination is not the one this client was built with.
+        /// </summary>
+        internal static Uri CreateTrackUri(string ingestionEndpoint) => new(new Uri(ingestionEndpoint), TrackPath);
+
+        internal async Task<HttpMessage> InternalTrackAsync(IEnumerable<TelemetryItem> body, Uri trackUri, CancellationToken cancellationToken = default)
+        {
+            var message = CreateTrackRequest(body, trackUri);
+
+            try
+            {
+                RedirectPolicy.SetAllowAutoRedirect(message, false);
+                await Pipeline.SendAsync(message, cancellationToken).ConfigureAwait(false);
+            }
+            catch (Exception ex)
+            {
+                AzureMonitorExporterEventSource.Log.FailedToTransmit(ex);
+                if (ex.InnerException?.Source != "System.Net.Http" && ex.InnerException?.Source != "System")
+                {
+                    message?.Dispose();
+                    throw;
+                }
+            }
+
+            return message;
+        }
+
+        internal async Task<HttpMessage> InternalTrackAsync(ReadOnlyMemory<byte> body, Uri trackUri, CancellationToken cancellationToken = default)
+        {
+#if DEBUG
+            TelemetryDebugWriter.WriteTelemetryFromStorage(body);
+#endif
+
+            var message = CreateRequest(RequestContent.Create(body), trackUri);
+
+            try
+            {
+                RedirectPolicy.SetAllowAutoRedirect(message, false);
+                await Pipeline.SendAsync(message, cancellationToken).ConfigureAwait(false);
+            }
+            catch (Exception ex)
+            {
+                AzureMonitorExporterEventSource.Log.FailedToTransmit(ex);
+                if (ex.InnerException?.Source != "System.Net.Http" && ex.InnerException?.Source != "System")
+                {
+                    message?.Dispose();
+                    throw;
+                }
+            }
+
+            return message;
+        }
+
+        internal HttpMessage CreateTrackRequest(IEnumerable<TelemetryItem> body, Uri trackUri)
+        {
+            using var content = new NDJsonWriter();
+            foreach (var item in body)
+            {
+                content.JsonWriter.WriteObjectValue(item);
+                content.WriteNewLine();
+            }
+
+#if DEBUG
+            TelemetryDebugWriter.WriteTelemetry(content);
+#endif
+
+            return CreateRequest(RequestContent.Create(content.ToBytes()), trackUri);
+        }
+
+        private HttpMessage CreateRequest(RequestContent requestContent, Uri trackUri)
+        {
+            var message = Pipeline.CreateMessage();
+            var request = message.Request;
+            request.Method = RequestMethod.Post;
+
+            // A builder per request, because IngestionRedirectPolicy rewrites it and a shared one
+            // would carry one endpoint's redirect onto another endpoint's request.
+            var uri = new RawRequestUriBuilder();
+            uri.Reset(trackUri);
+            request.Uri = uri;
+
+            request.Headers.Add("Content-Type", "application/json");
+            request.Headers.Add("Accept", "application/json");
+            request.Content = requestContent;
+
+            return message;
+        }
+
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         private HttpMessage CreateRequest(RequestContent requestContent)
         {
             var message = Pipeline.CreateMessage();
             var request = message.Request;
             request.Method = RequestMethod.Post;
-            request.Uri = LazyInitializer.EnsureInitialized(ref _rawRequestUriBuilder, () =>
-            {
-                var uri = new RawRequestUriBuilder();
-                uri.Reset(_endpoint);
-                uri.AppendPath("/", false);
-                uri.AppendPath(_apiVersion, true);
-                uri.AppendPath("/track", false);
-                return uri;
-            })!;
+
+            // A builder per request. IngestionRedirectPolicy calls Reset on request.Uri, and a shared
+            // instance would carry that rewrite into every later request permanently.
+            var uri = new RawRequestUriBuilder();
+            uri.Reset(_trackUri ??= BuildTrackUri());
+            request.Uri = uri;
+
             request.Headers.Add("Content-Type", "application/json");
             request.Headers.Add("Accept", "application/json");
             request.Content = requestContent;
 
             return message;
         }
+
+        private Uri BuildTrackUri()
+        {
+            var builder = new RawRequestUriBuilder();
+            builder.Reset(_endpoint);
+            builder.AppendPath("/", false);
+            builder.AppendPath(_apiVersion, true);
+            builder.AppendPath("/track", false);
+
+            return builder.ToUri();
+        }
     }
 }

sdk/monitor/Azure.Monitor.OpenTelemetry.Exporter/src/ExporterRegistrationHostedService.cs

@@ -6,7 +6,9 @@
 using System.Threading;
 using System.Threading.Tasks;
 using Azure.Monitor.OpenTelemetry.Exporter.Internals;
+using Azure.Monitor.OpenTelemetry.Exporter.Internals.Diagnostics;
 using Azure.Monitor.OpenTelemetry.Exporter.Internals.GenAI;
+using Azure.Monitor.OpenTelemetry.Exporter.Internals.MultiTenant;
 using Azure.Monitor.OpenTelemetry.LiveMetrics;
 using Azure.Monitor.OpenTelemetry.LiveMetrics.Internals;
 using Microsoft.Extensions.DependencyInjection;
@@ -60,7 +62,7 @@ private static void Initialize(IServiceProvider serviceProvider)
                 // Add a processor manually to the TracerProvider created by the SDK
                 var exporterOptions = serviceProvider!.GetRequiredService<IOptionsMonitor<AzureMonitorExporterOptions>>().Get(Options.DefaultName);
 
-                if (exporterOptions.EnableLiveMetrics)
+                if (exporterOptions.EnableLiveMetrics && LiveMetricsIsSupported())
                 {
                     var manager = serviceProvider!.GetRequiredService<LiveMetricsClientManager>();
                     tracerProvider.AddProcessor(new LiveMetricsActivityProcessor(manager));
@@ -91,7 +93,7 @@ private static void Initialize(IServiceProvider serviceProvider)
 
                 loggerProvider.AddProcessor(new MainAgentAttributionLogProcessor());
 
-                if (exporterOptions.EnableLiveMetrics)
+                if (exporterOptions.EnableLiveMetrics && LiveMetricsIsSupported())
                 {
                     var manager = serviceProvider!.GetRequiredService<LiveMetricsClientManager>();
 
@@ -107,5 +109,16 @@ private static void Initialize(IServiceProvider serviceProvider)
                 }
             }
         }
+
+        private static bool LiveMetricsIsSupported()
+        {
+            if (!MultiTenantConfig.Enabled)
+            {
+                return true;
+            }
+
+            AzureMonitorExporterEventSource.Log.LiveMetricsDisabledForMultiTenantExport();
+            return false;
+        }
     }
 }