Distributed tracing: Producer

src/Confluent.Kafka/AssemblyInfo.cs

@@ -0,0 +1,3 @@
+using System.Runtime.CompilerServices;
+
+[assembly: InternalsVisibleTo("Confluent.Kafka.IntegrationTests, PublicKey=0024000004800000940000000602000000240000525341310004000001000100cd1e243be8c66588df8c1cec2cd2e81a0df139c996688f657475ea2105783efa6dcb659fff876dc51bc16b68ae3d48a0465e81b6b72e2d3d71e50f680fc2547bc3aa8014b9a862e5e7aa24e0fdc282220c5c8db6084cfb7cb4a503f76768020197cd7d93a215506ad203b7c94a2557ba6140c65935af3c3de1fe1846cac8e6ba")]

src/Confluent.Kafka/Confluent.Kafka.csproj

@@ -24,7 +24,8 @@
     <PackageReference Include="librdkafka.redist" Version="1.4.2">
         <PrivateAssets Condition="'$(TargetFrameworkIdentifier)' == '.NETFramework'">None</PrivateAssets>
     </PackageReference>
-    <PackageReference Include="System.Memory" Version="4.5.0" />
+    <PackageReference Include="System.Memory" Version="4.5.3" />
+    <PackageReference Include="System.Diagnostics.DiagnosticSource" Version="4.6.0" />
   </ItemGroup>
 
   <ItemGroup Condition=" '$(TargetFramework)' == 'netstandard1.3' ">

src/Confluent.Kafka/Diagnostics.cs

@@ -0,0 +1,118 @@
+// Copyright 2016-2018 Confluent Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// Refer to LICENSE for more information.
+
+using System;
+using System.Diagnostics;
+using System.Linq;
+using System.Runtime.CompilerServices;
+using System.Text;
+
+namespace Confluent.Kafka
+{
+    internal static class Diagnostics
+    {
+        public const string DiagnosticListenerName = "Confluent.Kafka.Listener";
+        public const string TraceParentHeaderName = "traceparent";
+        public const string TraceStateHeaderName = "tracestate";
+
+        public static DiagnosticListener DiagnosticSource { get; } = new DiagnosticListener(DiagnosticListenerName);
+
+        public static class Producer
+        {
+            public const string ActivityName = DiagnosticListenerName + ".MessageProduced";
+            public const string StartName = ActivityName + ".Start";
+            public const string StopName = ActivityName + ".Stop";
+            public const string ExceptionName = ActivityName + ".Exception";
+
+            [MethodImpl(MethodImplOptions.AggressiveInlining)]
+            public static Activity Start<TKey, TValue>(string topic, Message<TKey, TValue> message)
+            {
+                if (DiagnosticSource.IsEnabled(ActivityName, topic))
+                {
+                    Activity activity = new Activity(ActivityName);
+                    activity.Start();
+
+                    AddActivityStateToHeaders(activity, message.Headers);
+
+                    if (DiagnosticSource.IsEnabled(StartName))
+                    {
+                        DiagnosticSource.Write(StartName, new { Topic = topic, Message = message });
+                    }
+
+                    return activity;
+                }
+
+                return null;
+            }
+
+            [MethodImpl(MethodImplOptions.AggressiveInlining)]
+            public static void Stop<TKey, TValue>(string topic, DeliveryResult<TKey, TValue> deliveryResult)
+            {
+                if (DiagnosticSource.IsEnabled(StopName))
+                {
+                    DiagnosticSource.Write(StopName, new { Topic = topic, DeliveryResult = deliveryResult });
+                }
+            }
+
+            [MethodImpl(MethodImplOptions.AggressiveInlining)]
+            public static void Exception<TKey, TValue>(string topic, Message<TKey, TValue> message, Exception exception)
+            {
+                if (DiagnosticSource.IsEnabled(ExceptionName))
+                {
+                    DiagnosticSource.Write(ExceptionName, new { Topic = topic, Message = message, Exception = exception });
+                }
+            }
+
+            [MethodImpl(MethodImplOptions.AggressiveInlining)]
+            public static void StopOrException<TKey, TValue>(string topic, DeliveryReport<TKey, TValue> deliveryReport)
+            {
+                if (deliveryReport.Error.IsError)
+                {
+                    if (DiagnosticSource.IsEnabled(ExceptionName))
+                    {
+                        DiagnosticSource.Write(
+                            ExceptionName,
+                            new
+                            {
+                                Topic = topic,
+                                deliveryReport.Message,
+                                Exception = new ProduceException<TKey, TValue>(deliveryReport.Error, deliveryReport)
+                            });
+                    }
+                }
+                else if (DiagnosticSource.IsEnabled(StopName))
+                {
+                    DiagnosticSource.Write(StopName, new { Topic = topic, DeliveryResult = deliveryReport });
+                }
+            }
+
+            [MethodImpl(MethodImplOptions.AggressiveInlining)]
+            private static void AddActivityStateToHeaders(Activity activity, Headers headers)
+            {
+                if (headers?.Any(h => h.Key == TraceParentHeaderName) != true)
+                {
+                    headers.Add(TraceParentHeaderName, Encoding.UTF8.GetBytes(activity.Id));
+
+                    string traceState = activity.TraceStateString;
+                    if (traceState != null)
+                    {
+                        headers.Add(TraceStateHeaderName, Encoding.UTF8.GetBytes(traceState));
+                    }
+                }
+            }
+        }
+    }
+}