Add support for Filter and Enrich for OpenTelemetry activities (#859)

* Add support for `Filter` and `Enrich` for OpenTelemetry activities

* Make `internal` methods in `internal Telemetry` `public`

* Fix package versions and whatnot

* Remove `TracerProviderBuilderExtensions`

* Include `Deserialize` in the receive activity

* Revert back accidental change

* Add `ParentContext` to `NatsInstrumentationContext`

* Make `GetActivityContext` public to provide the ability to get context activity context

* Make preprocessor directive more accurate

* Revert .csproj formatting

* Move public artifacts out of the `Internal` namespace/folder

* Fix build script

---------

Co-authored-by: Ziya Suzen <[email protected]>

Author: aradalvand

.github/workflows/signoffs.yml

@@ -16,7 +16,7 @@ jobs:
         if: github.event_name == 'pull_request'
         run: |
           for commit in $(git rev-list "origin/${{ github.base_ref }}".."${{ github.event.pull_request.head.sha }}"); do
-            if ! git verify-commit --raw $commit 2>&1 | grep -q SIG; then
+            if ! git verify-commit --raw $commit 2>&1 | grep -iq SIG; then
               echo "--------------------------------------------------------------"
               echo "Error: Commit $commit is not signed using GPG, SSH, or S/MIME"
               echo "https://docs.github.com/en/authentication/managing-commit-signature-verification/signing-commits"

src/NATS.Client.Core/Internal/Telemetry.cs

@@ -6,14 +6,13 @@ namespace NATS.Client.Core.Internal;
 // https://opentelemetry.io/docs/specs/semconv/messaging/messaging-spans/#messaging-attributes
 internal static class Telemetry
 {
-    internal static readonly ActivitySource NatsActivities = new(name: NatsActivitySource);
-
-    private const string NatsActivitySource = "NATS.Net";
+    public const string NatsActivitySource = "NATS.Net";
+    public static readonly ActivitySource NatsActivities = new(name: NatsActivitySource);
     private static readonly object BoxedTrue = true;
 
-    internal static bool HasListeners() => NatsActivities.HasListeners();
+    public static bool HasListeners() => NatsActivities.HasListeners();
 
-    internal static Activity? StartSendActivity(
+    public static Activity? StartSendActivity(
         string name,
         INatsConnection? connection,
         string subject,
@@ -23,6 +22,19 @@ internal static class Telemetry
         if (!NatsActivities.HasListeners())
             return null;
 
+        var instrumentationContext = new NatsInstrumentationContext(
+            Subject: subject,
+            Headers: null,
+            ReplyTo: replyTo,
+            QueueGroup: null,
+            BodySize: null,
+            Size: null,
+            Connection: connection,
+            ParentContext: parentContext);
+
+        if (NatsInstrumentationOptions.Default.Filter is { } filter && !filter(instrumentationContext))
+            return null;
+
         KeyValuePair<string, object?>[] tags;
         if (connection is NatsConnection { ServerInfo: not null } conn)
         {
@@ -63,14 +75,19 @@ internal static class Telemetry
                 tags[3] = new KeyValuePair<string, object?>(Constants.ReplyTo, replyTo);
         }
 
-        return NatsActivities.StartActivity(
+        var activity = NatsActivities.StartActivity(
             name,
             kind: ActivityKind.Producer,
             parentContext: parentContext ?? default,
             tags: tags);
+
+        if (activity is not null)
+            NatsInstrumentationOptions.Default.Enrich?.Invoke(activity, instrumentationContext);
+
+        return activity;
     }
 
-    internal static void AddTraceContextHeaders(Activity? activity, ref NatsHeaders? headers)
+    public static void AddTraceContextHeaders(Activity? activity, ref NatsHeaders? headers)
     {
         if (activity is null)
             return;
@@ -87,15 +104,15 @@ internal static void AddTraceContextHeaders(Activity? activity, ref NatsHeaders?
                     return;
                 }
 
-                // There are cases where headers reused internally (e.g. JetStream publish retry)
+                // There are cases where headers reused publicly (e.g. JetStream publish retry)
                 // there may also be cases where application can reuse the same header
                 // in which case we should still be able to overwrite headers with telemetry fields
                 // even though headers would be set to readonly before being passed down in publish methods.
                 headers.SetOverrideReadOnly(fieldName, fieldValue);
             });
     }
 
-    internal static Activity? StartReceiveActivity(
+    public static Activity? StartReceiveActivity(
         INatsConnection? connection,
         string name,
         string subscriptionSubject,
@@ -109,6 +126,22 @@ internal static void AddTraceContextHeaders(Activity? activity, ref NatsHeaders?
         if (!NatsActivities.HasListeners())
             return null;
 
+        if (headers is null || !TryParseTraceContext(headers, out var context))
+            context = default;
+
+        var instrumentationContext = new NatsInstrumentationContext(
+            Subject: subject,
+            Headers: headers,
+            ReplyTo: replyTo,
+            QueueGroup: queueGroup,
+            BodySize: bodySize,
+            Size: size,
+            Connection: connection,
+            ParentContext: context);
+
+        if (NatsInstrumentationOptions.Default.Filter is { } filter && !filter(instrumentationContext))
+            return null;
+
         KeyValuePair<string, object?>[] tags;
         if (connection is NatsConnection { ServerInfo: not null } conn)
         {
@@ -162,17 +195,19 @@ internal static void AddTraceContextHeaders(Activity? activity, ref NatsHeaders?
                 tags[9] = new KeyValuePair<string, object?>(Constants.ReplyTo, replyTo);
         }
 
-        if (headers is null || !TryParseTraceContext(headers, out var context))
-            context = default;
-
-        return NatsActivities.StartActivity(
+        var activity = NatsActivities.StartActivity(
             name,
             kind: ActivityKind.Consumer,
             parentContext: context,
             tags: tags);
+
+        if (activity is not null)
+            NatsInstrumentationOptions.Default.Enrich?.Invoke(activity, instrumentationContext);
+
+        return activity;
     }
 
-    internal static void SetException(Activity? activity, Exception exception)
+    public static void SetException(Activity? activity, Exception exception)
     {
         if (activity is null)
             return;
@@ -251,11 +286,14 @@ private static bool TryParseTraceContext(NatsHeaders headers, out ActivityContex
             },
             out var traceParent,
             out var traceState);
-
+#if NETSTANDARD2_0_OR_GREATER || NET7_0_OR_GREATER
+        return ActivityContext.TryParse(traceParent, traceState, isRemote: true, out context);
+#else
         return ActivityContext.TryParse(traceParent, traceState, out context);
+#endif
     }
 
-    internal class Constants
+    public class Constants
     {
         public const string True = "true";
         public const string False = "false";