feat: OTEL redis support

App/Config/settings.yaml

@@ -4,7 +4,7 @@ AllowedHosts: "*"
 Kestrel:
   Endpoints:
     Http:
-      Url: http://+:9001
+      Url: http://+:9002
 Logging:
   LogLevel:
     Default: Information

App/Modules/Bookings/Data/BookingCdcModel.cs

@@ -0,0 +1,8 @@
+namespace App.Modules.Bookings.Data;
+
+public class BookingCdcModel
+{
+
+}
+
+

App/Modules/Bookings/Data/BookingCdcRepository.cs

@@ -0,0 +1,14 @@
+namespace App.Modules.Bookings.Data;
+
+
+
+
+public interface IBookingCdcRepository
+{
+  Task Add(string type);
+}
+
+public class BookingCdcRepository
+{
+
+}

App/Modules/Common/OtelRedis.cs

@@ -0,0 +1,142 @@
+using System.Diagnostics;
+using System.Text.Json;
+using OpenTelemetry;
+using OpenTelemetry.Context.Propagation;
+using StackExchange.Redis;
+using StackExchange.Redis.Extensions.Core.Abstractions;
+
+namespace App.Modules.Common;
+
+public record OtelRedisMessage<T>(Dictionary<string, string> Context, T Message);
+
+public class OtelRedisDatabase(IRedisDatabase redis)
+{
+
+  private static readonly TextMapPropagator Propagator = new TraceContextPropagator();
+
+  private static readonly JsonSerializerOptions SerializeOptions = new()
+  {
+    PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
+  };
+
+  private static readonly ActivitySource ActivitySource = new("AutoTrace.Redis.Shim");
+
+
+  public async Task<long> PublishAsync<T>(RedisChannel channel, T message, CommandFlags flag = CommandFlags.None)
+  {
+    var carrier = new Dictionary<string, string>();
+
+    using var activity = ActivitySource.StartActivity();
+
+    if (activity != null) Propagator.Inject(new PropagationContext(activity.Context, Baggage.Current), carrier, (c, k, v) => c[k] = v);
+
+    var otelMessage = new OtelRedisMessage<T>(carrier, message);
+    return await redis.PublishAsync(channel, otelMessage, flag);
+  }
+
+  public async Task SubscribeAsync<T>(RedisChannel channel, Func<T?, Task> handler,
+    CommandFlags flag = CommandFlags.None)
+  {
+    await redis.SubscribeAsync<OtelRedisMessage<T>>(channel, async m =>
+    {
+      var carrier = m?.Context ?? [];
+
+      var parentContext = Propagator.Extract(default, carrier,
+        (c, k) => c.TryGetValue(k, out var v) ? new[] { v } : Enumerable.Empty<string>());
+      Baggage.Current = parentContext.Baggage;
+
+      using var activity = ActivitySource.StartActivity();
+      var message = m == null ? default : m.Message;
+      await handler(message);
+    }, flag);
+  }
+
+  public RedisValue StreamAdd<T>(RedisKey key, T message, RedisValue? messageId = null, int? maxLength = null,
+    bool useApproximateMaxLength = false, CommandFlags flags = CommandFlags.None)
+  {
+    var carrier = new Dictionary<string, string>();
+
+    using var activity = ActivitySource.StartActivity();
+    if (activity != null)
+      Propagator.Inject(new PropagationContext(activity.Context, Baggage.Current), carrier, (c, k, v) => c[k] = v);
+
+    var contextJson = JsonSerializer.Serialize(carrier, SerializeOptions);
+    var messageJson = JsonSerializer.Serialize(message, SerializeOptions);
+
+    NameValueEntry[] nameValueEntry = [
+      new NameValueEntry("context", contextJson),
+      new NameValueEntry("message", messageJson),
+    ];
+
+    return redis.Database.StreamAdd(key, nameValueEntry,
+      messageId, maxLength, useApproximateMaxLength, flags);
+  }
+
+  public async Task StreamRead<T>(RedisKey key, RedisValue position, Func<T?, Task> handler, int? count = null,
+    CommandFlags flags = CommandFlags.None)
+  {
+    var entries = redis.Database.StreamRead(key, position, count, flags);
+    var messages = entries.Select(x => this.FromNameValueEntry<T>(x.Values));
+
+    foreach (var otelRedisMessage in messages)
+    {
+
+      var carrier = otelRedisMessage?.Context ?? [];
+      var parentContext = Propagator.Extract(default, carrier,
+        (c, k) => c.TryGetValue(k, out var v) ? new[] { v } : Enumerable.Empty<string>());
+      Baggage.Current = parentContext.Baggage;
+
+      using var activity = ActivitySource.StartActivity();
+
+      var m = otelRedisMessage == null ? default : otelRedisMessage.Message;
+      await handler(m);
+    }
+  }
+
+  public async Task StreamReadGroup<T>(RedisKey key, RedisValue groupName, RedisValue consumerName,
+    Func<T?, Task> handler, RedisValue? position = null, int? count = null, bool noAck = false,
+    CommandFlags flags = CommandFlags.None)
+  {
+    var entries = redis.Database.StreamReadGroup(key, groupName, consumerName, position, count, flags);
+    var messages = entries.Select(x => this.FromNameValueEntry<T>(x.Values));
+
+    foreach (var otelRedisMessage in messages)
+    {
+      var carrier = otelRedisMessage?.Context ?? [];
+      var parentContext = Propagator.Extract(default, carrier,
+        (c, k) => c.TryGetValue(k, out var v) ? new[] { v } : Enumerable.Empty<string>());
+      Baggage.Current = parentContext.Baggage;
+
+      using var activity = ActivitySource.StartActivity();
+      var m = otelRedisMessage == null ? default : otelRedisMessage.Message;
+      await handler(m);
+    }
+  }
+
+
+  private NameValueEntry[] ToNameValueEntry<T>(T message)
+  {
+    var carrier = new Dictionary<string, string>();
+    var contextJson = JsonSerializer.Serialize(carrier, SerializeOptions);
+    var messageJson = JsonSerializer.Serialize(message, SerializeOptions);
+    NameValueEntry[] nameValueEntry =
+    [
+      new NameValueEntry("context", contextJson),
+      new NameValueEntry("message", messageJson),
+    ];
+    return nameValueEntry;
+  }
+
+  private OtelRedisMessage<T>? FromNameValueEntry<T>(NameValueEntry[] nameValueEntry)
+  {
+    string? contextJson = nameValueEntry.FirstOrDefault(x => x.Name == "context").Value;
+    string? messageJson = nameValueEntry.FirstOrDefault(x => x.Name == "message").Value;
+    if (contextJson == null || messageJson == null) return null;
+
+    var context = JsonSerializer.Deserialize<Dictionary<string, string>>(contextJson, SerializeOptions);
+    var message = JsonSerializer.Deserialize<T>(messageJson, SerializeOptions);
+    if (context == null || message == null) return null;
+
+    return new OtelRedisMessage<T>(context, message);
+  }
+}

App/Modules/System/SystemController.cs

@@ -4,13 +4,15 @@
 using Asp.Versioning;
 using Microsoft.AspNetCore.Mvc;
 using Microsoft.Extensions.Options;
+using StackExchange.Redis.Extensions.Core.Abstractions;
 
 namespace App.Modules.System;
 
 [ApiVersionNeutral]
 [ApiController]
 [Route("/")]
-public class SystemController(IOptionsSnapshot<AppOption> app, IAuthHelper h) : AtomiControllerBase(h)
+public class SystemController(IRedisClientFactory factory, IOptionsSnapshot<AppOption> app, IAuthHelper h)
+  : AtomiControllerBase(h)
 {
   [HttpGet]
   public ActionResult<object> SystemInfo()

App/Properties/launchSettings.json

@@ -8,8 +8,7 @@
       "launchUrl": "swagger",
       "hotReloadEnabled": true,
       "environmentVariables": {
-        "ASPNETCORE_ENVIRONMENT": "Development",
-        "LANDSCAPE": "lapras"
+        "ASPNETCORE_ENVIRONMENT": "Development"
       }
     }
   }

Taskfile.yml

@@ -17,6 +17,12 @@ tasks:
       LANDSCAPE: lapras
     cmds:
       - dotnet ef migrations --project ./App add {{.CLI_ARGS}}
+  run:
+    desc: Runs local .NET Server
+    env:
+      LANDSCAPE: corsola
+    cmds:
+      - dotnet run --project App
   setup:
     desc: Setups dotnet
     cmds:

infra/api_chart/app/settings.yaml

@@ -4,7 +4,7 @@ AllowedHosts: "*"
 Kestrel:
   Endpoints:
     Http:
-      Url: http://+:9001
+      Url: http://+:9002
 Logging:
   LogLevel:
     Default: Information

infra/migrate.Dockerfile

@@ -10,5 +10,5 @@ WORKDIR /app
 RUN dotnet tool install --global dotnet-ef
 ENV PATH="$PATH:/home/dotnet/.dotnet/tools"
 ENV LANDSCAPE=lapras
-RUN dotnet-ef migrations bundle --project ./App 
+RUN dotnet-ef migrations bundle --project ./App
 CMD [ "./efbundle" ]

infra/migration_chart/app/settings.yaml

@@ -4,7 +4,7 @@ AllowedHosts: "*"
 Kestrel:
   Endpoints:
     Http:
-      Url: http://+:9001
+      Url: http://+:9002
 Logging:
   LogLevel:
     Default: Information