change tags for better data support (#3)

* change tags for better data support * fix assertions * fix mappings * fix mapping * fix message
agoda-com · Dec 22, 2023 · 4052124 · 4052124
1 parent 95a514a
commit 4052124
Show file tree

Hide file tree

Showing 5 changed files with 63 additions and 38 deletions.
diff --git a/src/KafkaFlow.ApplicationInsights.Tests/MainIntegrationTestFixture.cs b/src/KafkaFlow.ApplicationInsights.Tests/MainIntegrationTestFixture.cs
@@ -62,11 +62,11 @@ public async Task WhenSendPostToHealthCheckEndpoint_ShouldReturnOkResponse()
         result.IsSuccessStatusCode.ShouldBeTrue();
         Thread.Sleep(5000);
         dataStore.Count.ShouldBe(7);
-        dataStore.Where(y => y is DependencyTelemetry).Count(x => ((DependencyTelemetry)x).Type == "Kafka-Producer").ShouldBe(3);
-        dataStore.Where(y => y is DependencyTelemetry).Count(x => ((DependencyTelemetry)x).DependencyKind == "Kafka-Producer").ShouldBe(3);
-        dataStore.Where(y => y is DependencyTelemetry).Count(x => ((DependencyTelemetry)x).DependencyTypeName == "Kafka-Producer").ShouldBe(3);
-        dataStore.Where(y => y is DependencyTelemetry).Count(x => ((DependencyTelemetry)x).Type == "Kafka-Consumer").ShouldBe(3);
-        dataStore.Where(y => y is DependencyTelemetry).Count(x => ((DependencyTelemetry)x).DependencyKind == "Kafka-Consumer").ShouldBe(3);
-        dataStore.Where(y => y is DependencyTelemetry).Count(x => ((DependencyTelemetry)x).DependencyTypeName == "Kafka-Consumer").ShouldBe(3);
+        dataStore.Where(y => y is DependencyTelemetry).Count(x => ((DependencyTelemetry)x).Type == "Kafka" && ((DependencyTelemetry)x).Data.StartsWith("Produce")).ShouldBe(3);
+        dataStore.Where(y => y is DependencyTelemetry).Count(x => ((DependencyTelemetry)x).DependencyKind == "Kafka" && ((DependencyTelemetry)x).Data.StartsWith("Produce")).ShouldBe(3);
+        dataStore.Where(y => y is DependencyTelemetry).Count(x => ((DependencyTelemetry)x).DependencyTypeName == "Kafka" && ((DependencyTelemetry)x).Data.StartsWith("Produce")).ShouldBe(3);
+        dataStore.Where(y => y is DependencyTelemetry).Count(x => ((DependencyTelemetry)x).Type == "Kafka" && ((DependencyTelemetry)x).Data.StartsWith("Consume")).ShouldBe(3);
+        dataStore.Where(y => y is DependencyTelemetry).Count(x => ((DependencyTelemetry)x).DependencyKind == "Kafka" && ((DependencyTelemetry)x).Data.StartsWith("Consume")).ShouldBe(3);
+        dataStore.Where(y => y is DependencyTelemetry).Count(x => ((DependencyTelemetry)x).DependencyTypeName == "Kafka" && ((DependencyTelemetry)x).Data.StartsWith("Consume")).ShouldBe(3);
     }
 }
diff --git a/src/KafkaFlow.ApplicationInsights/AppInsightsConsumerEventsHandler.cs b/src/KafkaFlow.ApplicationInsights/AppInsightsConsumerEventsHandler.cs
@@ -7,7 +7,7 @@
 
 public class AppInsightsConsumerEventsHandler
 {
-    public static Task OnConsumeStarted(IMessageContext eventContextMessageContext, TelemetryClient telemetryClient)
+    internal static Task OnConsumeStarted(IMessageContext eventContextMessageContext, TelemetryClient telemetryClient)
     {
         eventContextMessageContext.Items.Add("timer", Stopwatch.StartNew());
         eventContextMessageContext.Items.Add("telemetryClient", telemetryClient);
@@ -22,33 +22,23 @@ public static Task OnConsumeError(IMessageContext eventContextMessageContext, Ex
         {
             {"topic" , eventContextMessageContext.ConsumerContext.Topic},
             {"partition" , eventContextMessageContext.ConsumerContext.Partition.ToString()},
-            {"offset" , eventContextMessageContext.ConsumerContext.Offset.ToString()},
         });
         eventContextMessageContext.Items.TryGetValue("timer", out var timer);
         var theTimer = (Stopwatch)timer;
         theTimer.Stop();
-        telemetryClient.TrackDependency("Kafka-Consumer",
-            eventContextMessageContext.ConsumerContext.Topic,
-            eventContextMessageContext.ConsumerContext.Partition.ToString(),
-            eventContextMessageContext.ConsumerContext.Offset.ToString(),
-            DateTimeOffset.UtcNow,
-            theTimer.Elapsed, "500", false);
+        telemetryClient.TrackKafkaDependency(eventContextMessageContext, "Error", false, theTimer.Elapsed);
         return Task.CompletedTask;
     }
 
-    public static Task OnConsumeCompleted(IMessageContext eventContextMessageContext)
+    internal static Task OnConsumeCompleted(IMessageContext eventContextMessageContext)
     {
         eventContextMessageContext.Items.TryGetValue("telemetryClient", out var telemetryClientOut);
         var telemetryClient = (TelemetryClient)telemetryClientOut;
         eventContextMessageContext.Items.TryGetValue("timer", out var timer);
         var theTimer = (Stopwatch)timer;
         theTimer.Stop();
-        telemetryClient.TrackDependency("Kafka-Consumer",
-            eventContextMessageContext.ConsumerContext.Topic,
-            eventContextMessageContext.ConsumerContext.Partition.ToString(),
-            eventContextMessageContext.ConsumerContext.Offset.ToString(),
-            DateTimeOffset.UtcNow,
-            theTimer.Elapsed, "200", true);
+
+        telemetryClient.TrackKafkaDependency(eventContextMessageContext, "Ok", true, theTimer.Elapsed);
         return Task.CompletedTask;
     }
 }
diff --git a/src/KafkaFlow.ApplicationInsights/AppInsightsProducerEventsHandler.cs b/src/KafkaFlow.ApplicationInsights/AppInsightsProducerEventsHandler.cs
@@ -7,7 +7,7 @@
 
 public class AppInsightsProducerEventsHandler
 {
-    public static Task OnProducerStarted(IMessageContext eventContextMessageContext, TelemetryClient telemetryClient)
+    internal static Task OnProducerStarted(IMessageContext eventContextMessageContext, TelemetryClient telemetryClient)
     {
         eventContextMessageContext.Items.Add("timer", Stopwatch.StartNew());
         eventContextMessageContext.Items.Add("telemetryClient", telemetryClient); 
@@ -22,18 +22,12 @@ public static Task OnProducerError(IMessageContext eventContextMessageContext, E
         {
             {"topic" , eventContextMessageContext.ProducerContext.Topic},
             {"partition" , eventContextMessageContext.ProducerContext.Partition.ToString()},
-            {"offset" , eventContextMessageContext.ProducerContext.Offset.ToString()},
         });
         eventContextMessageContext.Items.TryGetValue("timer", out var timer);
         var theTimer = (Stopwatch)timer;
         theTimer.Stop();
-        var timeComponent = theTimer.ElapsedMilliseconds;
-        telemetryClient.TrackDependency("Kafka-Producer",
-            eventContextMessageContext.ProducerContext.Topic,
-            eventContextMessageContext.ProducerContext.Partition.ToString(),
-            eventContextMessageContext.ProducerContext.Offset.ToString(),
-            DateTimeOffset.UtcNow,
-            theTimer.Elapsed, "500", false);
+
+        telemetryClient.TrackKafkaDependency(eventContextMessageContext, "Error", false, theTimer.Elapsed);
         return Task.CompletedTask;
     }
 
@@ -44,13 +38,8 @@ public static Task OnProducerCompleted(IMessageContext eventContextMessageContex
         eventContextMessageContext.Items.TryGetValue("timer", out var timer);
         var theTimer = (Stopwatch)timer;
         theTimer.Stop();
-        var timeComponent = theTimer.ElapsedMilliseconds;
-        telemetryClient.TrackDependency("Kafka-Producer", 
-            eventContextMessageContext.ProducerContext.Topic, 
-            eventContextMessageContext.ProducerContext.Partition.ToString(),
-            eventContextMessageContext.ProducerContext.Offset.ToString(), 
-            DateTimeOffset.UtcNow, 
-            theTimer.Elapsed, "200", true);
+
+        telemetryClient.TrackKafkaDependency(eventContextMessageContext, "Ok", true, theTimer.Elapsed);
         return Task.CompletedTask;
     }
 }
diff --git a/src/KafkaFlow.ApplicationInsights/Constants.cs b/src/KafkaFlow.ApplicationInsights/Constants.cs
@@ -0,0 +1,6 @@
+internal static class Constants
+{
+    internal const string DependencyType = "Kafka";
+    internal const string ConsumerType = "Consume";
+    internal const string ProducerType = "Produce";
+}
diff --git a/src/KafkaFlow.ApplicationInsights/ExtensionMethods.cs b/src/KafkaFlow.ApplicationInsights/ExtensionMethods.cs
@@ -1,4 +1,5 @@
-using KafkaFlow;
+using System;
+using KafkaFlow;
 using KafkaFlow.Configuration;
 using Microsoft.ApplicationInsights;
 
@@ -31,4 +32,43 @@ public static IKafkaConfigurationBuilder AddAppInsightsInstrumentation(this IKaf
 
         return builder;
     }
+
+    internal static string GetDependencyName(this IMessageContext eventContextMessageContext)
+    {
+        return eventContextMessageContext.ConsumerContext?.Topic ??
+               eventContextMessageContext.ProducerContext?.Topic;
+    }
+
+    internal static string GetData(this IMessageContext eventContextMessageContext)
+    {
+        var type = Constants.ConsumerType;
+        if (eventContextMessageContext.ConsumerContext == null)
+        {
+            type = Constants.ProducerType;
+        }
+
+        var partition = eventContextMessageContext.ConsumerContext?.Partition.ToString() ??
+                        eventContextMessageContext.ProducerContext?.Partition.ToString();
+
+        return $"{type}/{partition}";
+    }
+
+    internal static string GetTarget(this IMessageContext eventContextMessageContext)
+    {
+        return string.Join(",", eventContextMessageContext.Brokers);
+    }
+
+    internal static void TrackKafkaDependency(this TelemetryClient telemetryClient, IMessageContext eventContextMessageContext, string resultCode,
+        bool success, TimeSpan elapsedTime)
+    {
+        telemetryClient.TrackDependency(Constants.DependencyType,
+                                    eventContextMessageContext.GetTarget(),
+                                    eventContextMessageContext.GetDependencyName(),
+                                      eventContextMessageContext.GetData(),
+                                      DateTimeOffset.UtcNow,
+                                      elapsedTime,
+                                      resultCode,
+                                      success);
+    }
 }
+