[OneCollectorExporter] Failed transmission callback

src/OpenTelemetry.Exporter.OneCollector/.publicApi/net462/PublicAPI.Unshipped.txt

@@ -0,0 +1,2 @@
+OpenTelemetry.Exporter.OneCollector.OneCollectorExporter<T>.RegisterPayloadTransmittedCallback(OpenTelemetry.Exporter.OneCollector.OneCollectorExporterPayloadTransmittedCallbackAction! callback, bool includeFailures) -> System.IDisposable?
+OpenTelemetry.Exporter.OneCollector.OneCollectorExporterPayloadTransmittedCallbackArguments.Succeeded.get -> bool

src/OpenTelemetry.Exporter.OneCollector/.publicApi/net6.0/PublicAPI.Unshipped.txt

@@ -0,0 +1,2 @@
+OpenTelemetry.Exporter.OneCollector.OneCollectorExporter<T>.RegisterPayloadTransmittedCallback(OpenTelemetry.Exporter.OneCollector.OneCollectorExporterPayloadTransmittedCallbackAction! callback, bool includeFailures) -> System.IDisposable?
+OpenTelemetry.Exporter.OneCollector.OneCollectorExporterPayloadTransmittedCallbackArguments.Succeeded.get -> bool

src/OpenTelemetry.Exporter.OneCollector/.publicApi/net7.0/PublicAPI.Unshipped.txt

@@ -0,0 +1,2 @@
+OpenTelemetry.Exporter.OneCollector.OneCollectorExporter<T>.RegisterPayloadTransmittedCallback(OpenTelemetry.Exporter.OneCollector.OneCollectorExporterPayloadTransmittedCallbackAction! callback, bool includeFailures) -> System.IDisposable?
+OpenTelemetry.Exporter.OneCollector.OneCollectorExporterPayloadTransmittedCallbackArguments.Succeeded.get -> bool

src/OpenTelemetry.Exporter.OneCollector/.publicApi/netstandard2.0/PublicAPI.Unshipped.txt

@@ -0,0 +1,2 @@
+OpenTelemetry.Exporter.OneCollector.OneCollectorExporter<T>.RegisterPayloadTransmittedCallback(OpenTelemetry.Exporter.OneCollector.OneCollectorExporterPayloadTransmittedCallbackAction! callback, bool includeFailures) -> System.IDisposable?
+OpenTelemetry.Exporter.OneCollector.OneCollectorExporterPayloadTransmittedCallbackArguments.Succeeded.get -> bool

src/OpenTelemetry.Exporter.OneCollector/.publicApi/netstandard2.1/PublicAPI.Unshipped.txt

@@ -0,0 +1,2 @@
+OpenTelemetry.Exporter.OneCollector.OneCollectorExporter<T>.RegisterPayloadTransmittedCallback(OpenTelemetry.Exporter.OneCollector.OneCollectorExporterPayloadTransmittedCallbackAction! callback, bool includeFailures) -> System.IDisposable?
+OpenTelemetry.Exporter.OneCollector.OneCollectorExporterPayloadTransmittedCallbackArguments.Succeeded.get -> bool

src/OpenTelemetry.Exporter.OneCollector/CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## Unreleased
 
+* Added support for receiving tranmission failures via the
+  `RegisterPayloadTransmittedCallback` API.
+  ([#1309](https://github.com/open-telemetry/opentelemetry-dotnet-contrib/pull/1309))
+
 ## 1.5.1
 
 Released 2023-Aug-07

src/OpenTelemetry.Exporter.OneCollector/Internal/Transports/HttpJsonPostTransport.cs

@@ -27,7 +27,8 @@ internal sealed class HttpJsonPostTransport : ITransport, IDisposable
     private static readonly string SdkVersion = $"OTel-{Environment.OSVersion.Platform}-.net-{typeof(OneCollectorExporter<>).Assembly.GetName()?.Version?.ToString() ?? "0.0.0"}";
     private static readonly string UserAgent = $".NET/{Environment.Version} HttpClient";
 
-    private readonly CallbackManager<OneCollectorExporterPayloadTransmittedCallbackAction> payloadTransmittedCallbacks = new();
+    private readonly CallbackManager<OneCollectorExporterPayloadTransmittedCallbackAction> payloadTransmittedSuccessCallbacks = new();
+    private readonly CallbackManager<OneCollectorExporterPayloadTransmittedCallbackAction> payloadTransmittedFailureCallbacks = new();
     private readonly Uri endpoint;
     private readonly string instrumentationKey;
     private readonly OneCollectorExporterHttpTransportCompressionType compressionType;
@@ -56,15 +57,25 @@ public HttpJsonPostTransport(
 
     public void Dispose()
     {
-        this.payloadTransmittedCallbacks.Dispose();
+        this.payloadTransmittedSuccessCallbacks.Dispose();
+        this.payloadTransmittedFailureCallbacks.Dispose();
         this.buffer?.Dispose();
     }
 
-    public IDisposable RegisterPayloadTransmittedCallback(OneCollectorExporterPayloadTransmittedCallbackAction callback)
+    public IDisposable RegisterPayloadTransmittedCallback(OneCollectorExporterPayloadTransmittedCallbackAction callback, bool includeFailures)
     {
         Guard.ThrowIfNull(callback);
 
-        return this.payloadTransmittedCallbacks.Add(callback);
+        var successRegistration = this.payloadTransmittedSuccessCallbacks.Add(callback);
+
+        if (!includeFailures)
+        {
+            return successRegistration;
+        }
+
+        var failureRegistration = this.payloadTransmittedFailureCallbacks.Add(callback);
+
+        return new TranmissionCallbackWrapper(successRegistration, failureRegistration);
     }
 
     public bool Send(in TransportSendRequest sendRequest)
@@ -112,13 +123,14 @@ public bool Send(in TransportSendRequest sendRequest)
 
                 OneCollectorExporterEventSource.Log.WriteTransportDataSentEventIfEnabled(sendRequest.ItemType, sendRequest.NumberOfItems, this.Description);
 
-                var root = this.payloadTransmittedCallbacks.Root;
+                var root = this.payloadTransmittedSuccessCallbacks.Root;
                 if (root != null)
                 {
                     this.InvokePayloadTransmittedCallbacks(
                         root,
                         streamStartingPosition,
-                        in sendRequest);
+                        in sendRequest,
+                        succeeded: true);
                 }
 
                 return true;
@@ -135,6 +147,16 @@ public bool Send(in TransportSendRequest sendRequest)
                     collectorErrors,
                     errorDetails);
 
+                var root = this.payloadTransmittedFailureCallbacks.Root;
+                if (root != null)
+                {
+                    this.InvokePayloadTransmittedCallbacks(
+                        root,
+                        streamStartingPosition,
+                        in sendRequest,
+                        succeeded: false);
+                }
+
                 return false;
             }
         }
@@ -185,7 +207,8 @@ private HttpContent BuildRequestContent(Stream stream)
     private void InvokePayloadTransmittedCallbacks(
         OneCollectorExporterPayloadTransmittedCallbackAction callback,
         long streamStartingPosition,
-        in TransportSendRequest sendRequest)
+        in TransportSendRequest sendRequest,
+        bool succeeded)
     {
         var stream = sendRequest.ItemStream;
 
@@ -200,7 +223,8 @@ private void InvokePayloadTransmittedCallbacks(
                     sendRequest.ItemSerializationFormat,
                     stream,
                     OneCollectorExporterTransportProtocolType.HttpJsonPost,
-                    this.endpoint));
+                    this.endpoint,
+                    succeeded));
         }
         catch (Exception ex)
         {
@@ -212,6 +236,26 @@ private void InvokePayloadTransmittedCallbacks(
         }
     }
 
+    private sealed class TranmissionCallbackWrapper : IDisposable
+    {
+        private readonly IDisposable successRegistration;
+        private readonly IDisposable failureRegistration;
+
+        public TranmissionCallbackWrapper(
+            IDisposable successRegistration,
+            IDisposable failureRegistration)
+        {
+            this.successRegistration = successRegistration;
+            this.failureRegistration = failureRegistration;
+        }
+
+        public void Dispose()
+        {
+            this.successRegistration.Dispose();
+            this.failureRegistration.Dispose();
+        }
+    }
+
     private sealed class NonDisposingStreamContent : HttpContent
     {
 #pragma warning disable CA2213 // Disposable fields should be disposed

src/OpenTelemetry.Exporter.OneCollector/Internal/Transports/ITransport.cs

@@ -22,5 +22,7 @@ internal interface ITransport
 
     bool Send(in TransportSendRequest sendRequest);
 
-    IDisposable RegisterPayloadTransmittedCallback(OneCollectorExporterPayloadTransmittedCallbackAction callback);
+    IDisposable RegisterPayloadTransmittedCallback(
+        OneCollectorExporterPayloadTransmittedCallbackAction callback,
+        bool includeFailures);
 }

src/OpenTelemetry.Exporter.OneCollector/OneCollectorExporter.cs

@@ -75,7 +75,22 @@ public sealed override ExportResult Export(in Batch<T> batch)
     /// Call <see cref="IDisposable.Dispose"/> on the returned instance to
     /// cancel the registration.</returns>
     public IDisposable? RegisterPayloadTransmittedCallback(OneCollectorExporterPayloadTransmittedCallbackAction callback)
-        => this.sink.Transport?.RegisterPayloadTransmittedCallback(callback);
+        => this.RegisterPayloadTransmittedCallback(callback, includeFailures: false);
+
+    /// <summary>
+    /// Register a callback action that will be triggered any time a payload is
+    /// transmitted by the exporter.
+    /// </summary>
+    /// <param name="callback"><see
+    /// cref="OneCollectorExporterPayloadTransmittedCallbackAction"/>.</param>
+    /// <param name="includeFailures">Specify <see langword="true"/> to receive
+    /// callbacks when transmission fails.</param>
+    /// <returns><see langword="null"/> if no transport is tied to the exporter
+    /// or an <see cref="IDisposable"/> representing the registered callback.
+    /// Call <see cref="IDisposable.Dispose"/> on the returned instance to
+    /// cancel the registration.</returns>
+    public IDisposable? RegisterPayloadTransmittedCallback(OneCollectorExporterPayloadTransmittedCallbackAction callback, bool includeFailures)
+        => this.sink.Transport?.RegisterPayloadTransmittedCallback(callback, includeFailures);
 
     /// <inheritdoc/>
     protected override void Dispose(bool disposing)

src/OpenTelemetry.Exporter.OneCollector/OneCollectorExporterPayloadTransmittedCallbackArguments.cs

@@ -39,7 +39,8 @@ internal OneCollectorExporterPayloadTransmittedCallbackArguments(
         OneCollectorExporterSerializationFormatType payloadSerializationFormat,
         Stream payloadStream,
         OneCollectorExporterTransportProtocolType transportProtocol,
-        Uri transportEndpoint)
+        Uri transportEndpoint,
+        bool succeeded)
     {
         Debug.Assert(payloadStream != null, "payload stream was null");
         Debug.Assert(payloadStream!.CanSeek, "payload stream was not seekable");
@@ -49,6 +50,7 @@ internal OneCollectorExporterPayloadTransmittedCallbackArguments(
         this.payloadStream = payloadStream;
         this.TransportProtocol = transportProtocol;
         this.TransportEndpoint = transportEndpoint!;
+        this.Succeeded = succeeded;
     }
 
     /// <summary>
@@ -61,6 +63,11 @@ internal OneCollectorExporterPayloadTransmittedCallbackArguments(
     /// </summary>
     public Uri TransportEndpoint { get; }
 
+    /// <summary>
+    /// Gets a value indicating whether or not the payload transmission was successful or not.
+    /// </summary>
+    public bool Succeeded { get; }
+
     /// <summary>
     /// Gets the payload serialization format.
     /// </summary>

test/OpenTelemetry.Exporter.OneCollector.Tests/HttpJsonPostTransportTests.cs

@@ -102,6 +102,8 @@ public void RegisterPayloadTransmittedCallbackTest()
          * 1) Exisiting callback fires again and is verified. Then we remove the callback.
          * 2) Verifies callback is NOT attached and NOT fired.
          * 3) Callback is attached again and verified to fire. Then we remove the callback.
+         * 4) Tests the callback on a failed message with includeFailures: false.
+         * 5) Tests the callback on a failed message with includeFailures: true.
          */
 
         RunHttpServerTest(
@@ -122,21 +124,30 @@ public void RegisterPayloadTransmittedCallbackTest()
                 Assert.True(string.IsNullOrWhiteSpace(req.Headers["Content-Encoding"]));
                 Assert.Equal(request, Encoding.ASCII.GetString(body.ToArray()));
             },
+            shouldTestFailFunc: (iteration) => iteration == 4 || iteration == 5,
             testStartingAction: (iteration, transport) =>
             {
                 switch (iteration)
                 {
                     case 0:
                     case 3:
                         Assert.Null(callbackRegistration);
-                        callbackRegistration = transport.RegisterPayloadTransmittedCallback(OnPayloadTransmitted);
+                        callbackRegistration = transport.RegisterPayloadTransmittedCallback(OnPayloadTransmitted, includeFailures: false);
                         break;
                     case 1:
                         Assert.NotNull(callbackRegistration);
                         break;
                     case 2:
                         Assert.Null(callbackRegistration);
                         break;
+                    case 4:
+                        Assert.Null(callbackRegistration);
+                        callbackRegistration = transport.RegisterPayloadTransmittedCallback(OnPayloadTransmitted, includeFailures: false);
+                        break;
+                    case 5:
+                        Assert.Null(callbackRegistration);
+                        callbackRegistration = transport.RegisterPayloadTransmittedCallback(OnPayloadTransmitted, includeFailures: true);
+                        break;
                 }
             },
             testFinishedAction: (iteration, transport) =>
@@ -149,8 +160,18 @@ public void RegisterPayloadTransmittedCallbackTest()
                         break;
                     case 1:
                     case 3:
+                    case 4:
+                    case 5:
                         Assert.NotNull(callbackRegistration);
-                        Assert.True(callbackFired);
+                        if (iteration == 4)
+                        {
+                            Assert.False(callbackFired);
+                        }
+                        else
+                        {
+                            Assert.True(callbackFired);
+                        }
+
                         callbackRegistration.Dispose();
                         callbackRegistration = null;
                         break;
@@ -163,9 +184,9 @@ public void RegisterPayloadTransmittedCallbackTest()
                 callbackFired = false;
                 lastCompletedIteration = iteration;
             },
-            testIterations: 4);
+            testIterations: 6);
 
-        Assert.Equal(3, lastCompletedIteration);
+        Assert.Equal(5, lastCompletedIteration);
 
         void OnPayloadTransmitted(in OneCollectorExporterPayloadTransmittedCallbackArguments arguments)
         {
@@ -178,6 +199,14 @@ void OnPayloadTransmitted(in OneCollectorExporterPayloadTransmittedCallbackArgum
             Assert.Equal(OneCollectorExporterSerializationFormatType.CommonSchemaV4JsonStream, arguments.PayloadSerializationFormat);
             Assert.Equal(OneCollectorExporterTransportProtocolType.HttpJsonPost, arguments.TransportProtocol);
             Assert.NotNull(arguments.TransportEndpoint);
+            if (lastCompletedIteration == 4)
+            {
+                Assert.False(arguments.Succeeded);
+            }
+            else
+            {
+                Assert.True(arguments.Succeeded);
+            }
         }
     }
 
@@ -197,16 +226,19 @@ private static void RunHttpServerTest(
         Action<HttpListenerRequest, MemoryStream> assertRequestAction,
         int numberOfItemsInRequestBody = 1,
         int testIterations = 1,
+        Func<int, bool>? shouldTestFailFunc = null,
         Action<int, ITransport>? testStartingAction = null,
         Action<int, ITransport>? testFinishedAction = null)
     {
+        shouldTestFailFunc ??= static iteration => false;
+        bool failTest = false;
         bool requestReceivedAndAsserted = false;
         Exception? testException = null;
 
         using var testServer = TestHttpServer.RunServer(
             context =>
             {
-                context.Response.StatusCode = 200;
+                context.Response.StatusCode = failTest ? 400 : 200;
 
                 using MemoryStream requestBody = new MemoryStream();
 
@@ -241,6 +273,8 @@ private static void RunHttpServerTest(
 
             for (int i = 0; i < testIterations; i++)
             {
+                failTest = shouldTestFailFunc(i);
+
                 testStartingAction?.Invoke(i, transport);
 
                 using var requestBodyStream = new MemoryStream(requestBodyBytes);
@@ -259,7 +293,7 @@ private static void RunHttpServerTest(
                     throw testException;
                 }
 
-                Assert.True(result);
+                Assert.NotEqual(failTest, result);
                 Assert.True(requestReceivedAndAsserted);
 
                 testFinishedAction?.Invoke(i, transport);

test/OpenTelemetry.Exporter.OneCollector.Tests/WriteDirectlyToTransportSinkTests.cs

@@ -165,7 +165,7 @@ private sealed class TestTransport : ITransport
 
         public List<byte[]> ExportedData { get; } = new();
 
-        public IDisposable RegisterPayloadTransmittedCallback(OneCollectorExporterPayloadTransmittedCallbackAction callback)
+        public IDisposable RegisterPayloadTransmittedCallback(OneCollectorExporterPayloadTransmittedCallbackAction callback, bool includeFailures)
         {
             throw new NotImplementedException();
         }