[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, and sampling defaults to fixed-rate rather than rate-limited because a per-process rate limit would be shared across every tenant the process carries. The switch cannot be combined with Microsoft Entra ID authentication, because the credential is scoped to the exporter's own audience and would be sent to endpoints supplied by telemetry.
+  ([#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,6 +19,9 @@
 
 ### 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))
 

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

@@ -8,6 +8,7 @@
 using Azure.Core;
 using Azure.Monitor.OpenTelemetry.Exporter.Internals;
 using Azure.Monitor.OpenTelemetry.Exporter.Internals.Diagnostics;
+using Azure.Monitor.OpenTelemetry.Exporter.Internals.MultiTenant;
 using Azure.Monitor.OpenTelemetry.Exporter.Internals.GenAI;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.DependencyInjection.Extensions;
@@ -84,9 +85,7 @@ public static TracerProviderBuilder AddAzureMonitorTraceExporter(
                     AzureMonitorExporterEventSource.Log.LiveMetricsNotSupported(methodName: nameof(AddAzureMonitorTraceExporter));
                 }
 
-                builder.SetSampler(exporterOptions.TracesPerSecond != null ?
-                    new RateLimitedSampler(exporterOptions.TracesPerSecond.Value) :
-                    new ApplicationInsightsSampler(exporterOptions.SamplingRatio));
+                builder.SetSampler(SamplerFactory.Create(exporterOptions, MultiTenantConfig.Enabled));
 
                 if (credential != null)
                 {

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,23 @@ private static void Initialize(IServiceProvider serviceProvider)
                 }
             }
         }
+
+        private static int s_liveMetricsSuppressionReported;
+
+        private static bool LiveMetricsIsSupported()
+        {
+            if (!MultiTenantConfig.Enabled)
+            {
+                return true;
+            }
+
+            // Asked once per signal, but the answer is a property of the process.
+            if (Interlocked.Exchange(ref s_liveMetricsSuppressionReported, 1) == 0)
+            {
+                AzureMonitorExporterEventSource.Log.LiveMetricsDisabledForMultiTenantExport();
+            }
+
+            return false;
+        }
     }
 }

sdk/monitor/Azure.Monitor.OpenTelemetry.Exporter/src/Internals/ActivityTagsProcessor.cs

@@ -10,6 +10,7 @@ namespace Azure.Monitor.OpenTelemetry.Exporter.Internals
     internal struct ActivityTagsProcessor
     {
         private readonly bool _includeUnmappedTags;
+        private readonly bool _recognizeRoutingTags;
 
         public AzMonList MappedTags;
         public AzMonList UnMappedTags;
@@ -37,6 +38,7 @@ internal struct ActivityTagsProcessor
         public ActivityTagsProcessor()
         {
             _includeUnmappedTags = true;
+            _recognizeRoutingTags = false;
             MappedTags = AzMonList.InitializeForMappedTags();
             UnMappedTags = AzMonList.Initialize();
         }
@@ -45,9 +47,16 @@ public ActivityTagsProcessor()
         /// Callers that only read <see cref="MappedTags"/> can skip collecting the unmapped
         /// tags, which avoids a pooled buffer and the string conversion of array-valued tags.
         /// </summary>
-        public ActivityTagsProcessor(bool includeUnmappedTags)
+        /// <param name="includeUnmappedTags">Whether to collect tags that match no semantic slot.</param>
+        /// <param name="recognizeRoutingTags">
+        /// Only the multi-tenant conversion consumes the routing slots. Claiming them anywhere else
+        /// would take those attributes out of custom dimensions with nothing to emit them instead,
+        /// silently dropping them from telemetry the feature is not even involved in.
+        /// </param>
+        public ActivityTagsProcessor(bool includeUnmappedTags, bool recognizeRoutingTags = false)
         {
             _includeUnmappedTags = includeUnmappedTags;
+            _recognizeRoutingTags = recognizeRoutingTags;
             MappedTags = AzMonList.InitializeForMappedTags();
             UnMappedTags = includeUnmappedTags ? AzMonList.Initialize() : default;
         }
@@ -62,7 +71,7 @@ public void CategorizeTags(Activity activity)
                     continue;
                 }
 
-                if (SemanticSlotMap.TryGetSlot(tag.Key, out var slot))
+                if (SemanticSlotMap.TryGetSlot(tag.Key, out var slot) && (_recognizeRoutingTags || !IsRoutingSlot(slot)))
                 {
                     switch (slot)
                     {
@@ -138,5 +147,8 @@ public void Return()
             MappedTags.Return();
             UnMappedTags.Return();
         }
+
+        private static bool IsRoutingSlot(SemanticSlot slot)
+            => slot == SemanticSlot.MicrosoftInstrumentationKey || slot == SemanticSlot.MicrosoftIngestionEndpoint;
     }
 }