Report multi-endpoint routing, delivery and storage loss through the event source

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

@@ -102,6 +102,7 @@ private ExportResult ExportMultiTenant(in Batch<LogRecord> batch)
         {
             // A concurrent Export takes a fresh batch rather than sharing the cached one.
             var routeBatch = Interlocked.Exchange(ref _routeBatch, null) ?? new EndpointRouteBatch();
+            routeBatch.BeginExport();
 
             try
             {

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

@@ -104,6 +104,7 @@ 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();
+            routeBatch.BeginExport();
 
             try
             {

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

@@ -150,7 +150,7 @@ internal async Task<HttpMessage> InternalTrackAsync(IEnumerable<TelemetryItem> b
         internal async Task<HttpMessage> InternalTrackAsync(ReadOnlyMemory<byte> body, Uri trackUri, CancellationToken cancellationToken = default)
         {
 #if DEBUG
-            TelemetryDebugWriter.WriteTelemetryFromStorage(body);
+            TelemetryDebugWriter.WriteTelemetryFromStorage(body, trackUri);
 #endif
 
             var message = CreateRequest(RequestContent.Create(body), trackUri);
@@ -183,7 +183,7 @@ internal HttpMessage CreateTrackRequest(IEnumerable<TelemetryItem> body, Uri tra
             }
 
 #if DEBUG
-            TelemetryDebugWriter.WriteTelemetry(content);
+            TelemetryDebugWriter.WriteTelemetry(content, trackUri);
 #endif
 
             return CreateRequest(RequestContent.Create(content.ToBytes()), trackUri);

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

@@ -389,7 +389,7 @@ public ExportResult Track(EndpointRouteBatch routeBatch, TelemetryItemOrigin ori
             // regions. A failing region does not stop the rest.
             for (int i = 0; i < routeBatch.Count; i++)
             {
-                if (SendGroupAsync(routeBatch[i], origin, async: false, cancellationToken).EnsureCompleted() != ExportResult.Success)
+                if (SendGroupAsync(routeBatch[i], routeBatch.Sequence, origin, async: false, cancellationToken).EnsureCompleted() != ExportResult.Success)
                 {
                     result = ExportResult.Failure;
                 }
@@ -402,17 +402,22 @@ public ExportResult Track(EndpointRouteBatch routeBatch, TelemetryItemOrigin ori
         /// A group that cannot be sent is written to its endpoint's own storage partition, so one
         /// region's backlog and back-off never affect another's.
         /// </remarks>
-        private async ValueTask<ExportResult> SendGroupAsync(EndpointRouteBatch.Group group, TelemetryItemOrigin origin, bool async, CancellationToken cancellationToken)
+        private async ValueTask<ExportResult> SendGroupAsync(EndpointRouteBatch.Group group, long exportSequence, TelemetryItemOrigin origin, bool async, CancellationToken cancellationToken)
         {
+            var itemCount = group.TelemetryItems.Count;
             var storage = _multiTenantStorage?.TryGet(group.IngestionEndpoint);
 
             if (storage != null && (IsPersistOnly || storage.TransmissionStateManager.State != TransmissionState.Closed))
             {
-                return SaveGroupForLaterTransmission(group, storage);
+                var deferred = SaveGroupForLaterTransmission(group, storage);
+                ReportDelivery(exportSequence, group, itemCount, deferred == ExportResult.Success ? "persisted" : "dropped");
+
+                return deferred;
             }
 
             var networkSdkStats = _statsbeat?.NetworkSdkStatsManager;
             Uri? trackUri = null;
+            var statusCode = 0;
 
             try
             {
@@ -426,6 +431,8 @@ private async ValueTask<ExportResult> SendGroupAsync(EndpointRouteBatch.Group gr
 
                 stopwatch?.Stop();
 
+                statusCode = httpMessage.HasResponse ? httpMessage.Response.Status : 0;
+
                 var result = HttpPipelineHelper.IsSuccess(httpMessage);
 
                 if (networkSdkStats != null)
@@ -457,13 +464,18 @@ private async ValueTask<ExportResult> SendGroupAsync(EndpointRouteBatch.Group gr
                 {
                     storage?.TransmissionStateManager.ResetConsecutiveErrors();
                     storage?.TransmissionStateManager.CloseTransmission();
+                    ReportDelivery(exportSequence, group, itemCount, "transmitted", itemCount, statusCode);
 
                     return result;
                 }
 
                 storage?.TransmissionStateManager.EnableBackOff(httpMessage.HasResponse ? httpMessage.Response : null);
 
-                return HttpPipelineHelper.ProcessTransmissionResult(httpMessage, storage?.BlobProvider, blob: null, _connectionVars, origin, _isAadEnabled, telemetrySchemaTypeCounter: null, networkSdkStats).ExportResult;
+                var transmission = HttpPipelineHelper.ProcessTransmissionResult(httpMessage, storage?.BlobProvider, blob: null, _connectionVars, origin, _isAadEnabled, telemetrySchemaTypeCounter: null, networkSdkStats);
+                var accepted = AcceptedCount(transmission.ItemsAccepted, itemCount);
+                ReportDelivery(exportSequence, group, itemCount, DescribeDelivery(transmission.ExportResult, accepted, itemCount, statusCode), accepted, statusCode);
+
+                return transmission.ExportResult;
             }
             catch (Exception ex)
             {
@@ -472,10 +484,42 @@ private async ValueTask<ExportResult> SendGroupAsync(EndpointRouteBatch.Group gr
                 networkSdkStats?.TrackException(trackUri?.Host, exceptionType: ex.GetType().FullName);
                 AzureMonitorExporterEventSource.Log.TransmitterFailed(origin, _isAadEnabled, _connectionVars.InstrumentationKey, ex);
 
-                return storage == null ? ExportResult.Failure : SaveGroupForLaterTransmission(group, storage);
+                // An unreachable endpoint arrives here, so this is the outcome most worth reporting.
+                var thrown = storage == null ? ExportResult.Failure : SaveGroupForLaterTransmission(group, storage);
+
+                // Reading the response can throw after ingestion answered, so acceptance is unknown.
+                ReportDelivery(exportSequence, group, itemCount, thrown == ExportResult.Success ? "persisted" : "dropped", accepted: -1, statusCode);
+
+                return thrown;
+            }
+        }
+
+        private static void ReportDelivery(long exportSequence, EndpointRouteBatch.Group group, int itemCount, string outcome, int accepted = -1, int statusCode = 0)
+            => AzureMonitorExporterEventSource.Log.RoutedGroupOutcome(exportSequence, itemCount, group.IngestionEndpoint, outcome, accepted, statusCode);
+
+        /// <summary>
+        /// Only the accepted count is asserted. A 206 settles each item separately, and which of the
+        /// rest were persisted for retry and which were rejected is not knowable from the result.
+        /// </summary>
+        private static string DescribeDelivery(ExportResult result, int accepted, int itemCount, int statusCode)
+        {
+            if (statusCode == ResponseStatusCodes.PartialSuccess)
+            {
+                return "partially accepted";
             }
+
+            if (accepted >= itemCount)
+            {
+                return "transmitted";
+            }
+
+            return result == ExportResult.Success ? "persisted" : "dropped";
         }
 
+        /// <summary>Returns -1 when ingestion reported no count, or one the batch cannot support.</summary>
+        private static int AcceptedCount(int? reported, int itemCount)
+            => reported is int value && value >= 0 && value <= itemCount ? value : -1;
+
         private ExportResult SaveGroupForLaterTransmission(EndpointRouteBatch.Group group, MultiTenantStorage.EndpointStorage storage)
         {
             try

sdk/monitor/Azure.Monitor.OpenTelemetry.Exporter/src/Internals/Diagnostics/AzureMonitorExporterEventSource.cs

@@ -2,6 +2,7 @@
 // Licensed under the MIT License.
 
 using System;
+using System.Diagnostics;
 using System.Diagnostics.CodeAnalysis;
 using System.Diagnostics.Tracing;
 using System.Runtime.CompilerServices;
@@ -584,5 +585,64 @@ public void RoutedTelemetryPersistenceThrew(string ingestionEndpoint, Exception
 
         [Event(66, Message = "Failed to persist routed telemetry for ingestion endpoint '{0}'. This telemetry item will be lost. {1}", Level = EventLevel.Error)]
         public void RoutedTelemetryPersistenceThrew(string ingestionEndpoint, string exceptionMessage) => WriteEvent(66, ingestionEndpoint, exceptionMessage);
+
+        [NonEvent]
+        public void RoutedTelemetryCollected(long exportSequence, string ingestionEndpoint, string instrumentationKey, Activity activity)
+        {
+            if (IsEnabled(EventLevel.Verbose))
+            {
+                RoutedTelemetryCollected(exportSequence, ingestionEndpoint, instrumentationKey, activity.TraceId.ToHexString(), activity.SpanId.ToHexString());
+            }
+        }
+
+        [Event(67, Message = "Export {0}: collected telemetry for ingestion endpoint '{1}'. Instrumentation Key: {2}. Trace Id: {3}. Span Id: {4}", Level = EventLevel.Verbose)]
+        public void RoutedTelemetryCollected(long exportSequence, string ingestionEndpoint, string instrumentationKey, string traceId, string spanId) => WriteEvent(67, exportSequence, ingestionEndpoint, instrumentationKey, traceId, spanId);
+
+        [NonEvent]
+        public void RoutedTelemetryRejected(long exportSequence, MultiTenant.RoutingRejectionReason reason, Activity activity)
+        {
+            if (IsEnabled(EventLevel.Verbose))
+            {
+                RoutedTelemetryRejected(exportSequence, reason.ToString(), activity.TraceId.ToHexString(), activity.SpanId.ToHexString());
+            }
+        }
+
+        [Event(68, Message = "Export {0}: dropped telemetry that could not be routed. Reason: {1}. Trace Id: {2}. Span Id: {3}. The Activity must carry a valid microsoft.instrumentation_key and microsoft.ingestion_endpoint.", Level = EventLevel.Verbose)]
+        public void RoutedTelemetryRejected(long exportSequence, string reason, string traceId, string spanId) => WriteEvent(68, exportSequence, reason, traceId, spanId);
+
+        [Event(69, Message = "No storage partition for ingestion endpoint '{0}': the limit of {1} partitions is already in use. Telemetry for this endpoint is transmitted without an offline storage fallback.", Level = EventLevel.Warning)]
+        public void MultiTenantPartitionCapReached(string ingestionEndpoint, int partitionCount) => WriteEvent(69, ingestionEndpoint, partitionCount);
+
+        [NonEvent]
+        public void RoutedTelemetryEvicted(string evictedOwner, long evictedBytes, string requestingEndpoint)
+        {
+            if (IsEnabled(EventLevel.Warning))
+            {
+                RoutedTelemetryEvicted(evictedOwner, evictedBytes.ToString(System.Globalization.CultureInfo.InvariantCulture), requestingEndpoint);
+            }
+        }
+
+        [Event(70, Message = "Evicted stored telemetry owned by '{0}' ({1} bytes) to make room for a write from ingestion endpoint '{2}'. The owner is an ingestion endpoint, or a storage directory name when the partition was left by an earlier run. The evicted telemetry was never transmitted and is lost.", Level = EventLevel.Warning)]
+        public void RoutedTelemetryEvicted(string evictedOwner, string evictedBytes, string requestingEndpoint) => WriteEvent(70, evictedOwner, evictedBytes, requestingEndpoint);
+
+        // Guarded in the body: neither parameter list matches a typed WriteEvent overload, so the
+        // call allocates an argument array whether or not anything is listening.
+        [Event(71, Message = "Export {0}: collected {1} Activities for {2} ingestion endpoints and dropped {3} that could not be routed.", Level = EventLevel.Informational)]
+        public void RoutedExportSummary(long exportSequence, int collected, int endpointCount, int rejected)
+        {
+            if (IsEnabled(EventLevel.Informational))
+            {
+                WriteEvent(71, exportSequence, collected, endpointCount, rejected);
+            }
+        }
+
+        [Event(72, Message = "Export {0}: {1} telemetry items for ingestion endpoint '{2}' were {3}. Accepted by ingestion: {4}, where -1 means ingestion reported no usable count. Status code: {5}", Level = EventLevel.Informational)]
+        public void RoutedGroupOutcome(long exportSequence, int itemCount, string ingestionEndpoint, string outcome, int itemsAccepted, int statusCode)
+        {
+            if (IsEnabled(EventLevel.Informational))
+            {
+                WriteEvent(72, exportSequence, itemCount, ingestionEndpoint, outcome, itemsAccepted, statusCode);
+            }
+        }
     }
 }

sdk/monitor/Azure.Monitor.OpenTelemetry.Exporter/src/Internals/Diagnostics/TelemetryDebugWriter.cs

@@ -35,12 +35,37 @@ public static void WriteTelemetry(NDJsonWriter content)
             }
         }
 
+        /// <summary>
+        /// Routed sends carry a destination chosen per item, so the payload alone does not say where
+        /// it went.
+        /// </summary>
+        public static void WriteTelemetry(NDJsonWriter content, Uri trackUri)
+        {
+            if (content == null)
+            {
+                return;
+            }
+
+            if (Debugger.IsAttached && Debugger.IsLogging())
+            {
+                Debugger.Log(0, null, $"(TRANSMITTING TELEMETRY TO {trackUri})\n{content}");
+            }
+        }
+
         public static void WriteTelemetryFromStorage(ReadOnlyMemory<byte> content)
         {
             if (Debugger.IsAttached && Debugger.IsLogging())
             {
                 Debugger.Log(0, null, "(TRANSMITTING TELEMETRY FROM STORAGE)\n" + Encoding.UTF8.GetString(content.ToArray()));
             }
         }
+
+        public static void WriteTelemetryFromStorage(ReadOnlyMemory<byte> content, Uri trackUri)
+        {
+            if (Debugger.IsAttached && Debugger.IsLogging())
+            {
+                Debugger.Log(0, null, $"(TRANSMITTING TELEMETRY FROM STORAGE TO {trackUri})\n{Encoding.UTF8.GetString(content.ToArray())}");
+            }
+        }
     }
 }

sdk/monitor/Azure.Monitor.OpenTelemetry.Exporter/src/Internals/MultiTenant/BudgetedBlobProvider.cs

@@ -50,7 +50,7 @@ protected override bool OnTryCreateBlob(byte[] buffer, int leasePeriodMillisecon
 
         private bool TryCreate(byte[] buffer, int leasePeriodMilliseconds, out PersistentBlob blob)
         {
-            var created = _owner.TryCreateBlobWithinBudget(_inner, buffer, leasePeriodMilliseconds, out var inner);
+            var created = _owner.TryCreateBlobWithinBudget(_inner, buffer, leasePeriodMilliseconds, _ingestionEndpoint, out var inner);
             blob = inner!;
 
             if (!created)

sdk/monitor/Azure.Monitor.OpenTelemetry.Exporter/src/Internals/MultiTenant/EndpointRouteBatch.cs

@@ -3,6 +3,7 @@
 
 using System;
 using System.Collections.Generic;
+using System.Threading;
 using Azure.Monitor.OpenTelemetry.Exporter.Models;
 
 namespace Azure.Monitor.OpenTelemetry.Exporter.Internals.MultiTenant
@@ -16,11 +17,18 @@ internal sealed class EndpointRouteBatch
     {
         private const int DefaultCapacity = 4;
 
+        private static long s_exportSequence;
+
         private Group?[] _groups = new Group?[DefaultCapacity];
         private int _count;
 
         internal int Count => _count;
 
+        /// <summary>Stitches one export's routing decisions to the sends they produced.</summary>
+        internal long Sequence { get; private set; }
+
+        internal void BeginExport() => Sequence = Interlocked.Increment(ref s_exportSequence);
+
         internal Group this[int index]
         {
             get

sdk/monitor/Azure.Monitor.OpenTelemetry.Exporter/src/Internals/MultiTenant/MultiTenantStorage.cs

@@ -5,6 +5,7 @@
 using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Diagnostics;
+using System.Diagnostics.Tracing;
 using System.IO;
 using System.Linq;
 using System.Threading;
@@ -118,7 +119,7 @@ internal ExportResult SaveTelemetry(EndpointStorage storage, byte[] content)
         /// eviction cannot help with - a removed directory, a full disk, a denied ACL - and deleting
         /// the backlog for those destroys other tenants' telemetry without saving this batch.
         /// </remarks>
-        internal bool TryCreateBlobWithinBudget(FileBlobProvider inner, byte[] buffer, int leasePeriodMilliseconds, out PersistentBlob? blob)
+        internal bool TryCreateBlobWithinBudget(FileBlobProvider inner, byte[] buffer, int leasePeriodMilliseconds, string requestingEndpoint, out PersistentBlob? blob)
         {
             blob = null;
 
@@ -176,7 +177,7 @@ internal bool TryCreateBlobWithinBudget(FileBlobProvider inner, byte[] buffer, i
 
                     if (!reserved)
                     {
-                        TryEvict(candidates[i]);
+                        TryEvict(candidates[i], requestingEndpoint);
                         reserved = TryReserve(buffer.Length);
                     }
                 }
@@ -396,7 +397,7 @@ private static void Offer(List<EvictionCandidate> candidates, int count, Evictio
             }
         }
 
-        private bool TryEvict(EvictionCandidate candidate)
+        private bool TryEvict(EvictionCandidate candidate, string requestingEndpoint)
         {
             // Measured before deletion because the length is unreadable afterwards.
             var length = FileLength(candidate.Path);
@@ -423,9 +424,34 @@ private bool TryEvict(EvictionCandidate candidate)
 
             Interlocked.Add(ref _currentSizeBytes, -length);
 
+            // Resolving the owner walks the partitions, so it waits until something is listening.
+            if (AzureMonitorExporterEventSource.Log.IsEnabled(EventLevel.Warning, EventKeywords.All))
+            {
+                AzureMonitorExporterEventSource.Log.RoutedTelemetryEvicted(DescribeOwner(candidate.Path), length, requestingEndpoint);
+            }
+
             return true;
         }
 
+        /// <summary>
+        /// Names the endpoint that owned an evicted blob. The directory is a one-way hash, so a
+        /// partition left by an earlier run can only be reported by its directory name.
+        /// </summary>
+        private string DescribeOwner(string path)
+        {
+            var directory = Path.GetDirectoryName(path);
+
+            foreach (var partition in _partitions)
+            {
+                if (string.Equals(partition.Value.Directory, directory, StringComparison.Ordinal))
+                {
+                    return partition.Key;
+                }
+            }
+
+            return Path.GetFileName(directory) ?? path;
+        }
+
         private readonly struct EvictionCandidate
         {
             internal EvictionCandidate(string name, PersistentBlob? blob, string path)
@@ -472,6 +498,11 @@ internal EvictionCandidate(string name, PersistentBlob? blob, string path)
 
                 if (_disposed || _partitions.Count >= MaxEndpointPartitions)
                 {
+                    if (!_disposed)
+                    {
+                        AzureMonitorExporterEventSource.Log.MultiTenantPartitionCapReached(ingestionEndpoint, MaxEndpointPartitions);
+                    }
+
                     return null;
                 }
 

sdk/monitor/Azure.Monitor.OpenTelemetry.Exporter/src/Internals/MultiTenant/RoutingRejectionReason.cs

@@ -0,0 +1,23 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+namespace Azure.Monitor.OpenTelemetry.Exporter.Internals.MultiTenant
+{
+    /// <summary>
+    /// Why an Activity could not be routed. Reported instead of the endpoint itself, which may
+    /// carry the credentials that caused the rejection.
+    /// </summary>
+    internal enum RoutingRejectionReason
+    {
+        None = 0,
+        MissingInstrumentationKey,
+        InstrumentationKeyTooLong,
+        MissingIngestionEndpoint,
+        IngestionEndpointTooLong,
+        IngestionEndpointMalformed,
+        IngestionEndpointNotHttps,
+        IngestionEndpointHasCredentials,
+        IngestionEndpointHasQueryOrFragment,
+        IngestionEndpointHostInvalid,
+    }
+}