Upgrading Elasticsearch to OTel version 1.0.2 (open-telemetry#83)

* Upgrading Elasticsearch to OTel version to 1.0.1

* Update OTel to make use of AddLegacyActivity

* Add shared project for instrumentation helper source

* Added an Elasticsearch sampler tests (currently fails)

* Got tests passing and add tests to show that sampling is working

* Move constants into ElasticsearchRequestPipelineDiagnosticListener class. Fix formatting issues.

* Adding Enrich option for Elasticsearch instrumentation

* Skip shared project test coverage

* Add tests to verify downstream suppression. Change DiagnosticSourceListener to use it's own RuntimeContextSlot.

* Update to latest OTel

Co-authored-by: Cijo Thomas <[email protected]>

.github/codecov.yml

@@ -22,3 +22,4 @@ comment:
 ignore:
   - "test/**/*"     # ignore test folder
   - "**.md"          # ignore md files
+  - "src/OpenTelemetry.Contrib.Shared"  # copied from main OTel project and has code coverage there

NuGet.config

@@ -3,6 +3,7 @@
   <packageSources>
     <clear />
     <add key="NuGet" value="https://api.nuget.org/v3/index.json" />
+    <add key="MyGet" value="https://www.myget.org/F/opentelemetry/api/v3/index.json" />
   </packageSources>
   <disabledPackageSources />
 </configuration>

build/Common.props

@@ -21,7 +21,7 @@
       Please sort alphabetically.
       Refer to https://docs.microsoft.com/en-us/nuget/concepts/package-versioning for semver syntax.
     -->
-    <MinVerPkgVer>[2.3.0,3.0)</MinVerPkgVer>
+    <MinVerPkgVer>[2.4.0,3.0)</MinVerPkgVer>
     <MicrosoftCodeAnalysisFxCopAnalyzersPkgVer>[3.3.0]</MicrosoftCodeAnalysisFxCopAnalyzersPkgVer>
     <MicrosoftCodeCoveragePkgVer>[16.7.1]</MicrosoftCodeCoveragePkgVer>
     <MicrosoftExtensionsHostingAbstractionsPkgVer>[2.1.0,5.0)</MicrosoftExtensionsHostingAbstractionsPkgVer>
@@ -44,4 +44,30 @@
       <PrivateAssets>all</PrivateAssets>
     </PackageReference>-->
   </ItemGroup>
+
+  <ItemGroup Condition="'$(IncludeSharedInstrumentationSource)'=='true'">
+    <Compile Include="$(RepoRoot)\src\OpenTelemetry.Contrib.Shared\DiagnosticSourceInstrumentation\ActivityInstrumentationHelper.cs" Link="Includes\ActivityInstrumentationHelper.cs" />
+    <Compile Include="$(RepoRoot)\src\OpenTelemetry.Contrib.Shared\DiagnosticSourceInstrumentation\DiagnosticSourceListener.cs" Link="Includes\DiagnosticSourceListener.cs" />
+    <Compile Include="$(RepoRoot)\src\OpenTelemetry.Contrib.Shared\DiagnosticSourceInstrumentation\DiagnosticSourceSubscriber.cs" Link="Includes\DiagnosticSourceSubscriber.cs" />
+    <Compile Include="$(RepoRoot)\src\OpenTelemetry.Contrib.Shared\DiagnosticSourceInstrumentation\InstrumentationEventSource.cs" Link="Includes\InstrumentationEventSource.cs" />
+    <Compile Include="$(RepoRoot)\src\OpenTelemetry.Contrib.Shared\DiagnosticSourceInstrumentation\ListenerHandler.cs" Link="Includes\ListenerHandler.cs" />
+    <Compile Include="$(RepoRoot)\src\OpenTelemetry.Contrib.Shared\DiagnosticSourceInstrumentation\MultiTypePropertyFetcher.cs" Link="Includes\MultiTypePropertyFetcher.cs" />
+    <Compile Include="$(RepoRoot)\src\OpenTelemetry.Contrib.Shared\DiagnosticSourceInstrumentation\PropertyFetcher.cs" Link="Includes\PropertyFetcher.cs" />
+    <Compile Include="$(RepoRoot)\src\OpenTelemetry.Contrib.Shared\Api\ActivityHelperExtensions.cs" Link="Includes\ActivityHelperExtensions.cs" />
+    <Compile Include="$(RepoRoot)\src\OpenTelemetry.Contrib.Shared\Api\EnumerationHelper.cs" Link="Includes\EnumerationHelper.cs" />
+    <Compile Include="$(RepoRoot)\src\OpenTelemetry.Contrib.Shared\Api\ExceptionExtensions.cs" Link="Includes\ExceptionExtensions.cs" />
+    <Compile Include="$(RepoRoot)\src\OpenTelemetry.Contrib.Shared\Api\IActivityEnumerator.cs" Link="Includes\IActivityEnumerator.cs" />
+    <Compile Include="$(RepoRoot)\src\OpenTelemetry.Contrib.Shared\Api\SemanticConventions.cs" Link="Includes\SemanticConventions.cs" />
+    <Compile Include="$(RepoRoot)\src\OpenTelemetry.Contrib.Shared\Api\SpanAttributeConstants.cs" Link="Includes\SpanAttributeConstants.cs" />
+    <Compile Include="$(RepoRoot)\src\OpenTelemetry.Contrib.Shared\Api\SpanHelper.cs" Link="Includes\SpanHelper.cs" />
+    <Compile Include="$(RepoRoot)\src\OpenTelemetry.Contrib.Shared\Api\StatusHelper.cs" Link="Includes\StatusHelper.cs" />
+  </ItemGroup>
+
+  <ItemGroup Condition="'$(IncludeSharedTestSource)'=='true'">
+    <Compile Include="$(RepoRoot)\src\OpenTelemetry.Contrib.Shared\Test\TestExporter.cs" Link="Includes\TestExporter.cs" />
+    <Compile Include="$(RepoRoot)\src\OpenTelemetry.Contrib.Shared\Test\TestSampler.cs" Link="Includes\TestSampler.cs" />
+    <Compile Include="$(RepoRoot)\src\OpenTelemetry.Contrib.Shared\Test\TestActivityExportProcessor.cs" Link="Includes\TestActivityExportProcessor.cs" />
+    <Compile Include="$(RepoRoot)\src\OpenTelemetry.Contrib.Shared\Test\TestActivityProcessor.cs" Link="Includes\TestActivityProcessor.cs" />
+  </ItemGroup>
+
 </Project>

opentelemetry-dotnet-contrib.sln

@@ -99,7 +99,9 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "OpenTelemetry.Contrib.Exten
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "OpenTelemetry.Contrib.Instrumentation.AWS", "src\OpenTelemetry.Contrib.Instrumentation.AWS\OpenTelemetry.Contrib.Instrumentation.AWS.csproj", "{970673DA-F308-4960-A58D-ECCEA44CEF6B}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "OpenTelemetry.Contrib.Instrumentation.AWS.Tests", "test\OpenTelemetry.Contrib.Instrumentation.AWS.Tests\OpenTelemetry.Contrib.Instrumentation.AWS.Tests.csproj", "{CAB66B50-DAB6-49B8-83F9-6CCF520C4A36}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "OpenTelemetry.Contrib.Instrumentation.AWS.Tests", "test\OpenTelemetry.Contrib.Instrumentation.AWS.Tests\OpenTelemetry.Contrib.Instrumentation.AWS.Tests.csproj", "{CAB66B50-DAB6-49B8-83F9-6CCF520C4A36}"
+EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "OpenTelemetry.Contrib.Shared", "src\OpenTelemetry.Contrib.Shared\OpenTelemetry.Contrib.Shared.csproj", "{F52B9D81-2155-433A-B6F2-4CD7CBBEC7E6}"
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "OpenTelemetry.Contrib.Instrumentation.Wcf.Tests", "test\OpenTelemetry.Contrib.Instrumentation.Wcf.Tests\OpenTelemetry.Contrib.Instrumentation.Wcf.Tests.csproj", "{76BAB24F-85DB-4FCE-89D0-EFB4185004C9}"
 EndProject
@@ -177,6 +179,10 @@ Global
 		{CAB66B50-DAB6-49B8-83F9-6CCF520C4A36}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{CAB66B50-DAB6-49B8-83F9-6CCF520C4A36}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{CAB66B50-DAB6-49B8-83F9-6CCF520C4A36}.Release|Any CPU.Build.0 = Release|Any CPU
+		{F52B9D81-2155-433A-B6F2-4CD7CBBEC7E6}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{F52B9D81-2155-433A-B6F2-4CD7CBBEC7E6}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{F52B9D81-2155-433A-B6F2-4CD7CBBEC7E6}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{F52B9D81-2155-433A-B6F2-4CD7CBBEC7E6}.Release|Any CPU.Build.0 = Release|Any CPU
 		{76BAB24F-85DB-4FCE-89D0-EFB4185004C9}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
 		{76BAB24F-85DB-4FCE-89D0-EFB4185004C9}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{76BAB24F-85DB-4FCE-89D0-EFB4185004C9}.Release|Any CPU.ActiveCfg = Release|Any CPU
@@ -207,6 +213,7 @@ Global
 		{9CE513AC-CFC5-4DD1-9F16-8719EDCE9BF9} = {2097345F-4DD3-477D-BC54-A922F9B2B402}
 		{970673DA-F308-4960-A58D-ECCEA44CEF6B} = {22DF5DC0-1290-4E83-A9D8-6BB7DE3B3E63}
 		{CAB66B50-DAB6-49B8-83F9-6CCF520C4A36} = {2097345F-4DD3-477D-BC54-A922F9B2B402}
+		{F52B9D81-2155-433A-B6F2-4CD7CBBEC7E6} = {22DF5DC0-1290-4E83-A9D8-6BB7DE3B3E63}
 		{76BAB24F-85DB-4FCE-89D0-EFB4185004C9} = {2097345F-4DD3-477D-BC54-A922F9B2B402}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution

src/OpenTelemetry.Contrib.Instrumentation.Elasticsearch/ElasticsearchClientInstrumentation.cs

@@ -16,7 +16,6 @@
 using System;
 using OpenTelemetry.Contrib.Instrumentation.ElasticsearchClient.Implementation;
 using OpenTelemetry.Instrumentation;
-using OpenTelemetry.Trace;
 
 namespace OpenTelemetry.Contrib.Instrumentation.ElasticsearchClient
 {
@@ -30,16 +29,10 @@ internal class ElasticsearchClientInstrumentation : IDisposable
         /// <summary>
         /// Initializes a new instance of the <see cref="ElasticsearchClientInstrumentation"/> class.
         /// </summary>
-        /// <param name="activitySource">ActivitySource adapter instance.</param>
         /// <param name="options">Configuration options for Elasticsearch client instrumentation.</param>
-        public ElasticsearchClientInstrumentation(ActivitySourceAdapter activitySource, ElasticsearchClientInstrumentationOptions options)
+        public ElasticsearchClientInstrumentation(ElasticsearchClientInstrumentationOptions options)
         {
-            if (activitySource == null)
-            {
-                throw new ArgumentNullException(nameof(activitySource));
-            }
-
-            this.diagnosticSourceSubscriber = new DiagnosticSourceSubscriber(new ElasticsearchRequestPipelineDiagnosticListener(activitySource, options), null);
+            this.diagnosticSourceSubscriber = new DiagnosticSourceSubscriber(new ElasticsearchRequestPipelineDiagnosticListener(options), null);
             this.diagnosticSourceSubscriber.Subscribe();
         }
 

src/OpenTelemetry.Contrib.Instrumentation.Elasticsearch/ElasticsearchClientInstrumentationOptions.cs

@@ -14,6 +14,9 @@
 // limitations under the License.
 // </copyright>
 
+using System;
+using System.Diagnostics;
+
 namespace OpenTelemetry.Contrib.Instrumentation.ElasticsearchClient
 {
     /// <summary>
@@ -30,5 +33,16 @@ public class ElasticsearchClientInstrumentationOptions
         /// Gets or sets a value indicating whether the JSON request body should be parsed out of the request debug information and formatted as indented JSON.
         /// </summary>
         public bool ParseAndFormatRequest { get; set; } = false;
+
+        /// <summary>
+        /// Gets or sets an action to enrich an Activity.
+        /// </summary>
+        /// <remarks>
+        /// <para><see cref="Activity"/>: the activity being enriched.</para>
+        /// <para>string: the name of the event.</para>
+        /// <para>object: the raw object from which additional information can be extracted to enrich the activity.
+        /// The type of this object depends on the event, which is given by the above parameter.</para>
+        /// </remarks>
+        public Action<Activity, string, object> Enrich { get; set; }
     }
 }

src/OpenTelemetry.Contrib.Instrumentation.Elasticsearch/Implementation/ElasticsearchInstrumentationEventSource.cs

@@ -14,7 +14,9 @@
 // limitations under the License.
 // </copyright>
 
+using System;
 using System.Diagnostics.Tracing;
+using OpenTelemetry.Internal;
 
 namespace OpenTelemetry.Contrib.Instrumentation.ElasticsearchClient.Implementation
 {
@@ -31,5 +33,20 @@ public void NullPayload(string handlerName, string eventName)
         {
             this.WriteEvent(1, handlerName, eventName);
         }
+
+        [NonEvent]
+        public void EnrichmentException(Exception ex)
+        {
+            if (this.IsEnabled(EventLevel.Error, (EventKeywords)(-1)))
+            {
+                this.EnrichmentException(ex.ToInvariantString());
+            }
+        }
+
+        [Event(2, Message = "Enrichment threw exception. Exception {0}.", Level = EventLevel.Error)]
+        public void EnrichmentException(string exception)
+        {
+            this.WriteEvent(2, exception);
+        }
     }
 }

src/OpenTelemetry.Contrib.Instrumentation.Elasticsearch/Implementation/ElasticsearchRequestPipelineDiagnosticListener.cs

@@ -20,6 +20,7 @@
 using System.Linq;
 using System.Net.Http;
 using System.Net.Sockets;
+using System.Reflection;
 using System.Text;
 using System.Text.Json;
 using System.Text.RegularExpressions;
@@ -30,10 +31,18 @@ namespace OpenTelemetry.Contrib.Instrumentation.ElasticsearchClient.Implementati
 {
     internal class ElasticsearchRequestPipelineDiagnosticListener : ListenerHandler
     {
+        internal const string DatabaseSystemName = "elasticsearch";
+        internal const string ExceptionCustomPropertyName = "OTel.Elasticsearch.Exception";
+        internal const string AttributeDbMethod = "db.method";
+
+        internal static readonly AssemblyName AssemblyName = typeof(ElasticsearchRequestPipelineDiagnosticListener).Assembly.GetName();
+        internal static readonly string ActivitySourceName = AssemblyName.Name;
+        internal static readonly Version Version = AssemblyName.Version;
+        internal static readonly ActivitySource ActivitySource = new ActivitySource(ActivitySourceName, Version.ToString());
+
         private static readonly Regex ParseRequest = new Regex(@"\n# Request:\r?\n(\{.*)\n# Response", RegexOptions.Compiled | RegexOptions.Singleline);
         private static readonly ConcurrentDictionary<object, string> MethodNameCache = new ConcurrentDictionary<object, string>();
 
-        private readonly ActivitySourceAdapter activitySource;
         private readonly ElasticsearchClientInstrumentationOptions options;
         private readonly MultiTypePropertyFetcher<Uri> uriFetcher = new MultiTypePropertyFetcher<Uri>("Uri");
         private readonly MultiTypePropertyFetcher<object> methodFetcher = new MultiTypePropertyFetcher<object>("Method");
@@ -43,15 +52,9 @@ internal class ElasticsearchRequestPipelineDiagnosticListener : ListenerHandler
         private readonly MultiTypePropertyFetcher<object> failureReasonFetcher = new MultiTypePropertyFetcher<object>("FailureReason");
         private readonly MultiTypePropertyFetcher<byte[]> responseBodyFetcher = new MultiTypePropertyFetcher<byte[]>("ResponseBodyInBytes");
 
-        public ElasticsearchRequestPipelineDiagnosticListener(ActivitySourceAdapter activitySource, ElasticsearchClientInstrumentationOptions options)
+        public ElasticsearchRequestPipelineDiagnosticListener(ElasticsearchClientInstrumentationOptions options)
             : base("Elasticsearch.Net.RequestPipeline")
         {
-            if (activitySource == null)
-            {
-                throw new ArgumentNullException(nameof(activitySource));
-            }
-
-            this.activitySource = activitySource;
             this.options = options;
         }
 
@@ -65,11 +68,12 @@ public override void OnStartActivity(Activity activity, object payload)
                 return;
             }
 
+            ActivityInstrumentationHelper.SetActivitySourceProperty(activity, ActivitySource);
+            ActivityInstrumentationHelper.SetKindProperty(activity, ActivityKind.Client);
+
             var method = this.methodFetcher.Fetch(payload);
             activity.DisplayName = this.GetDisplayName(activity, method);
 
-            this.activitySource.Start(activity, ActivityKind.Client);
-
             if (this.options.SuppressDownstreamInstrumentation)
             {
                 SuppressInstrumentationScope.Enter();
@@ -82,61 +86,67 @@ public override void OnStartActivity(Activity activity, object payload)
 
             var elasticIndex = this.GetElasticIndex(uri);
             activity.DisplayName = this.GetDisplayName(activity, method, elasticIndex);
-            activity.SetTag(Constants.AttributeDbSystem, "elasticsearch");
+            activity.SetTag(SemanticConventions.AttributeDbSystem, DatabaseSystemName);
 
             if (elasticIndex != null)
             {
-                activity.SetTag(Constants.AttributeDbName, elasticIndex);
+                activity.SetTag(SemanticConventions.AttributeDbName, elasticIndex);
             }
 
             var uriHostNameType = Uri.CheckHostName(uri.Host);
             if (uriHostNameType == UriHostNameType.IPv4 || uriHostNameType == UriHostNameType.IPv6)
             {
-                activity.SetTag(Constants.AttributeNetPeerIp, uri.Host);
+                activity.SetTag(SemanticConventions.AttributeNetPeerIp, uri.Host);
             }
             else
             {
-                activity.SetTag(Constants.AttributeNetPeerName, uri.Host);
+                activity.SetTag(SemanticConventions.AttributeNetPeerName, uri.Host);
             }
 
             if (uri.Port > 0)
             {
-                activity.SetTag(Constants.AttributeNetPeerPort, uri.Port);
+                activity.SetTag(SemanticConventions.AttributeNetPeerPort, uri.Port);
             }
 
             if (method != null)
             {
-                activity.SetTag(Constants.AttributeDbMethod, method.ToString());
+                activity.SetTag(AttributeDbMethod, method.ToString());
             }
 
-            activity.SetTag(Constants.AttributeDbUrl, uri.OriginalString);
+            activity.SetTag(SemanticConventions.AttributeDbUrl, uri.OriginalString);
+
+            try
+            {
+                this.options.Enrich?.Invoke(activity, "OnStartActivity", payload);
+            }
+            catch (Exception ex)
+            {
+                ElasticsearchInstrumentationEventSource.Log.EnrichmentException(ex);
+            }
         }
 
         public override void OnStopActivity(Activity activity, object payload)
         {
             if (activity.IsAllDataRequested)
             {
                 var statusCode = this.httpStatusFetcher.Fetch(payload);
+                activity.SetStatus(SpanHelper.ResolveSpanStatusForHttpStatusCode(statusCode.GetValueOrDefault()));
 
-                var status = Status.Error;
-
-                if (statusCode >= 100 && statusCode <= 399)
+                if (statusCode.HasValue)
                 {
-                    status = Status.Unset;
+                    activity.SetTag(SemanticConventions.AttributeHttpStatusCode, (int)statusCode);
                 }
 
-                activity.SetStatus(status);
-
                 var debugInformation = this.debugInformationFetcher.Fetch(payload);
                 if (debugInformation != null)
                 {
-                    activity.SetTag(Constants.AttributeDbStatement, this.ParseAndFormatRequest(activity, debugInformation));
+                    activity.SetTag(SemanticConventions.AttributeDbStatement, this.ParseAndFormatRequest(activity, debugInformation));
                 }
 
                 var originalException = this.originalExceptionFetcher.Fetch(payload);
                 if (originalException != null)
                 {
-                    activity.SetCustomProperty(Constants.ExceptionCustomPropertyName, originalException);
+                    activity.SetCustomProperty(ExceptionCustomPropertyName, originalException);
 
                     var failureReason = this.failureReasonFetcher.Fetch(originalException);
                     if (failureReason != null)
@@ -169,9 +179,16 @@ public override void OnStopActivity(Activity activity, object payload)
                         }
                     }
                 }
-            }
 
-            this.activitySource.Stop(activity);
+                try
+                {
+                    this.options.Enrich?.Invoke(activity, "OnStopActivity", payload);
+                }
+                catch (Exception ex)
+                {
+                    ElasticsearchInstrumentationEventSource.Log.EnrichmentException(ex);
+                }
+            }
         }
 
         private string GetDisplayName(Activity activity, object method, string elasticType = null)
@@ -233,8 +250,8 @@ private string ParseAndFormatRequest(Activity activity, string debugInformation)
                 return debugInformation;
             }
 
-            string method = activity.GetTagValue(Constants.AttributeDbMethod).ToString();
-            string url = activity.GetTagValue(Constants.AttributeDbUrl).ToString();
+            string method = activity.GetTagValue(AttributeDbMethod).ToString();
+            string url = activity.GetTagValue(SemanticConventions.AttributeDbUrl).ToString();
 
             if (method == "GET")
             {