feat: integrate ActivitySource for improved observability in resource watchers and operator builder (#895)

Add telemetry options to the resource watcher 

```csharp
builder
    .Services
    .AddKubernetesOperator(settings => settings.Name = OperatorName)

....

builder.Services
    .AddOpenTelemetry()
    .WithTracing(tracingProviderBuilder
        => tracingProviderBuilder
            .SetResourceBuilder(
                ResourceBuilder.CreateDefault()
                    .AddService(serviceName: OperatorName, serviceVersion: "1.0.0"))
            .AddSource(OperatorName))
    .UseAzureMonitor();
```

thanks 👋🏼 
m.

Author: kimpenhaus

src/KubeOps.Operator/Builder/OperatorBuilder.cs

@@ -1,4 +1,6 @@
-using k8s;
+using System.Diagnostics;
+
+using k8s;
 using k8s.Models;
 
 using KubeOps.Abstractions.Builder;
@@ -19,7 +21,7 @@
 
 namespace KubeOps.Operator.Builder;
 
-internal class OperatorBuilder : IOperatorBuilder
+internal sealed class OperatorBuilder : IOperatorBuilder
 {
     private readonly OperatorSettings _settings;
 
@@ -96,6 +98,7 @@ public IOperatorBuilder AddFinalizer<TImplementation, TEntity>(string identifier
     private void AddOperatorBase()
     {
         Services.AddSingleton(_settings);
+        Services.AddSingleton(new ActivitySource(_settings.Name));
 
         // Add the default configuration and the client separately. This allows external users to override either
         // just the config (e.g. for integration tests) or to replace the whole client, e.g. with a mock.

src/KubeOps.Operator/Watcher/LeaderAwareResourceWatcher{TEntity}.cs

@@ -1,3 +1,5 @@
+using System.Diagnostics;
+
 using k8s;
 using k8s.LeaderElection;
 using k8s.Models;
@@ -13,6 +15,7 @@
 namespace KubeOps.Operator.Watcher;
 
 internal sealed class LeaderAwareResourceWatcher<TEntity>(
+    ActivitySource activitySource,
     ILogger<LeaderAwareResourceWatcher<TEntity>> logger,
     IServiceProvider provider,
     TimedEntityQueue<TEntity> queue,
@@ -21,7 +24,14 @@ internal sealed class LeaderAwareResourceWatcher<TEntity>(
     IKubernetesClient client,
     IHostApplicationLifetime hostApplicationLifetime,
     LeaderElector elector)
-    : ResourceWatcher<TEntity>(logger, provider, queue, settings, labelSelector, client)
+    : ResourceWatcher<TEntity>(
+        activitySource,
+        logger,
+        provider,
+        queue,
+        settings,
+        labelSelector,
+        client)
     where TEntity : IKubernetesObject<V1ObjectMeta>
 {
     private CancellationTokenSource _cts = new();

src/KubeOps.Operator/Watcher/ResourceWatcher{TEntity}.cs

@@ -1,4 +1,5 @@
 using System.Collections.Concurrent;
+using System.Diagnostics;
 using System.Net;
 using System.Runtime.Serialization;
 using System.Text.Json;
@@ -21,6 +22,7 @@
 namespace KubeOps.Operator.Watcher;
 
 public class ResourceWatcher<TEntity>(
+    ActivitySource activitySource,
     ILogger<ResourceWatcher<TEntity>> logger,
     IServiceProvider provider,
     TimedEntityQueue<TEntity> requeue,
@@ -201,6 +203,7 @@ private async Task WatchClientEventsAsync(CancellationToken stoppingToken)
                                    cancellationToken: stoppingToken))
                 {
 #pragma warning disable SA1312
+                    using var __ = activitySource.StartActivity($"""processing "{type}" event""", ActivityKind.Consumer);
                     using var _ = logger.BeginScope(EntityLoggingScope.CreateFor(type, entity));
                     logger.LogInformation(
                         """Received watch event "{EventType}" for "{Kind}/{Name}", last observed resource version: {ResourceVersion}.""",

test/KubeOps.Operator.Test/Watcher/ResourceWatcher{TEntity}.Test.cs

@@ -1,4 +1,5 @@
-using System.Runtime.CompilerServices;
+using System.Diagnostics;
+using System.Runtime.CompilerServices;
 
 using k8s;
 using k8s.Models;
@@ -21,18 +22,19 @@ public sealed class ResourceWatcherTest
     public async Task Restarting_Watcher_Should_Trigger_New_Watch()
     {
         // Arrange.
+        var activitySource = new ActivitySource("unit-test");
         var logger = Mock.Of<ILogger<ResourceWatcher<V1Pod>>>();
         var serviceProvider = Mock.Of<IServiceProvider>();
         var timedEntityQueue = new TimedEntityQueue<V1Pod>();
-        var operatorSettings = new OperatorSettings() { Namespace = "unit-test" };
+        var operatorSettings = new OperatorSettings { Namespace = "unit-test" };
         var kubernetesClient = Mock.Of<IKubernetesClient>();
         var labelSelector = new DefaultEntityLabelSelector<V1Pod>();
 
         Mock.Get(kubernetesClient)
             .Setup(client => client.WatchAsync<V1Pod>("unit-test", null, null, true, It.IsAny<CancellationToken>()))
             .Returns<string?, string?, string?, bool?, CancellationToken>((_, _, _, _, cancellationToken) => WaitForCancellationAsync<(WatchEventType, V1Pod)>(cancellationToken));
 
-        var resourceWatcher = new ResourceWatcher<V1Pod>(logger, serviceProvider, timedEntityQueue, operatorSettings, labelSelector, kubernetesClient);
+        var resourceWatcher = new ResourceWatcher<V1Pod>(activitySource, logger, serviceProvider, timedEntityQueue, operatorSettings, labelSelector, kubernetesClient);
 
         // Act.
         // Start and stop the watcher.