[Feature]Add support to measure event/stream performance metrics (#166)

* [feature]add support to measure event/stream performance metrics

* remove the parameter when init construct of StreamPressureOptions

Author: Charles-aelf

Directory.Packages.props

@@ -17,7 +17,7 @@
     <PackageVersion Include="Azure.AI.TextAnalytics" Version="5.3.0" />
     <PackageVersion Include="JetBrains.Annotations" Version="2024.2.0" />
     <PackageVersion Include="Microsoft.EntityFrameworkCore.Sqlite.Core" Version="8.0.4" />
-    <PackageVersion Include="Microsoft.Extensions.Options" Version="8.0.2" />
+    <PackageVersion Include="Microsoft.Extensions.Options" Version="9.0.0" />
     <PackageVersion Include="Microsoft.Orleans.Serialization.Abstractions" Version="9.0.1" />
     <PackageVersion Include="Microsoft.SemanticKernel" Version="1.33.0" />
     <PackageVersion Include="Microsoft.SemanticKernel.Core" Version="1.33.0" />
@@ -99,7 +99,8 @@
     <PackageVersion Include="Microsoft.CodeCoverage" Version="17.9.0" />
     <PackageVersion Include="Microsoft.Extensions.AI" Version="9.0.1-preview.1.24570.5" />
     <PackageVersion Include="Microsoft.Extensions.AI.Abstractions" Version="9.0.1-preview.1.24570.5" />
-    <PackageVersion Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="8.0.2" />
+    <PackageVersion Include="Microsoft.Extensions.DependencyInjection" Version="9.0.0" />
+    <PackageVersion Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="9.0.0" />
     <PackageVersion Include="Microsoft.Extensions.FileProviders.Embedded" Version="8.0.4" />
     <PackageVersion Include="Microsoft.Extensions.Http.Polly" Version="9.0.0" />
     <PackageVersion Include="Microsoft.Extensions.Hosting" Version="9.0.0" />

src/Aevatar.Core.Abstractions/Events/EventWrapperBase.cs

@@ -18,6 +18,9 @@ public abstract class EventWrapperBase
     [Id(0)]
     public Dictionary<string, string> ContextMetadata { get; set; } = new Dictionary<string, string>();
 
+    [Id(1)]
+    public DateTime PublishedTimestampUtc { get; set; }
+
     protected EventWrapperBase()
     {
         // Initialize ContextMetadata

src/Aevatar.Core/Aevatar.Core.csproj

@@ -17,6 +17,12 @@
 
     <ItemGroup>
         <PackageReference Include="Microsoft.Orleans.EventSourcing" />
+        <PackageReference Include="Microsoft.Orleans.Streaming" />
+        <PackageReference Include="Orleans.Streams.Kafka" />
+        <PackageReference Include="Microsoft.Extensions.DependencyInjection" />
+        <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" />
+        <PackageReference Include="Microsoft.Extensions.Logging" />
+        <PackageReference Include="Microsoft.Extensions.Options" />
         <PackageReference Include="AElf.OpenTelemetry" />
     </ItemGroup>
 

src/Aevatar.Core/AevatarSyncWorker.cs

@@ -29,7 +29,9 @@ protected sealed override async Task<TResponse> PerformWork(TRequest request,
         {
             var response = await PerformLongRunTask(request);
             _logger.LogInformation($"Performed long run task for request of type {typeof(TRequest).FullName}, response is {response}");
-            await _asyncStream.OnNextAsync(new EventWrapper<TResponse>(response, Guid.NewGuid(), this.GetGrainId()));
+            var eventWrapper = new EventWrapper<TResponse>(response, Guid.NewGuid(), this.GetGrainId());
+            eventWrapper.PublishedTimestampUtc = DateTime.UtcNow;
+            await _asyncStream.OnNextAsync(eventWrapper);
             return response;
         }
         catch (Exception ex)

src/Aevatar.Core/BroadCastGAgentBase.cs

@@ -68,6 +68,7 @@ public async Task BroadCastEventAsync<T>(string streamIdString, T @event) where
     {
         var stream = GenStream<T>(streamIdString);
         var eventWrapper = new EventWrapper<T>(@event, Guid.NewGuid(), this.GetGrainId());
+        eventWrapper.PublishedTimestampUtc = DateTime.UtcNow;
         await stream.OnNextAsync(eventWrapper);
     }
 

src/Aevatar.Core/EventWrapperBaseAsyncObserver.cs

@@ -38,6 +38,17 @@ public static EventWrapperBaseAsyncObserver Create(Func<EventWrapperBase, Task>
 
     public async Task OnNextAsync(EventWrapperBase item, StreamSequenceToken? token = null)
     {
+        if (item == null)
+        {
+            throw new ArgumentNullException(nameof(item));
+        }
+        
+        var eventType = item.GetType().Name;
+        var latency = (DateTime.UtcNow - item.PublishedTimestampUtc).TotalSeconds;
+        _logger?.LogInformation("[OnNextAsync] Consuming event_type={EventType} PublishedTimestampUtc={PublishedTimestampUtc} latency={Latency}s",
+            eventType, item.PublishedTimestampUtc, latency);
+        Observability.EventPublishLatencyMetrics.Record(latency, item, MethodName, ParameterTypeName, null);
+
         await _func(item);
     }
 

src/Aevatar.Core/GAgentAsyncObserver.cs

@@ -2,19 +2,29 @@
 using Orleans.Streams;
 using OrleansCodeGen.Orleans.Runtime;
 using System.Diagnostics;
+using System.Diagnostics.Metrics;
+using Microsoft.Extensions.Logging;
 
 namespace Aevatar.Core;
 
 public class GAgentAsyncObserver : IAsyncObserver<EventWrapperBase>
 {
     private readonly List<EventWrapperBaseAsyncObserver> _observers;
     private readonly string _grainId;
+    private readonly ILogger<GAgentAsyncObserver> _logger;
+    private static readonly Meter Meter = new(OpenTelemetryConstants.AevatarStreamsMeterName);
+    private static readonly Histogram<double> PublishLatencyHistogram = Meter.CreateHistogram<double>(
+        OpenTelemetryConstants.EventPublishLatencyHistogram, "s", "Event publish-to-consume latency");
 
-    public GAgentAsyncObserver(List<EventWrapperBaseAsyncObserver> observers, string grainId)
+    public GAgentAsyncObserver(List<EventWrapperBaseAsyncObserver> observers, string grainId, ILogger<GAgentAsyncObserver> logger)
     {
         _observers = observers;
         _grainId = grainId;
+        _logger = logger;
     }
+
+    public GAgentAsyncObserver(List<EventWrapperBaseAsyncObserver> observers, string grainId)
+        : this(observers, grainId, null) { }
 
     /// <summary>
     /// Finds observers that match the given event type
@@ -65,8 +75,9 @@ private async Task BroadcastToObservers(Func<EventWrapperBaseAsyncObserver, Task
 
     public async Task OnNextAsync(EventWrapperBase item, StreamSequenceToken? token = null)
     {
-        // Extract event and ID from wrapper
         var (eventType, eventId) = EventWrapperHelper.ExtractProperties(item);
+        var latency = (DateTime.UtcNow - item.PublishedTimestampUtc).TotalSeconds;
+        Observability.EventPublishLatencyMetrics.Record(latency, item, null, null, _logger);
 
         // Try to create an activity with parent context if available
         var activity = ActivityHelper.CreateEventActivity(item, eventType, _grainId, eventId, token);

src/Aevatar.Core/GAgentBase.Observers.cs

@@ -103,8 +103,11 @@ private EventWrapperBaseAsyncObserver CreateMethodObserver(
                         ?.GetValue(item)!;
                     var grainId = (GrainId)item.GetType().GetProperty(nameof(EventWrapper<TEvent>.GrainId))
                         ?.GetValue(item)!;
+                    var publishedTimestamp = (DateTime)item.GetType().GetProperty(nameof(EventWrapper<TEvent>.PublishedTimestampUtc))
+                        ?.GetValue(item)!;
 
                     var eventWrapper = new EventWrapper<TEvent>(eventType, eventId, grainId);
+                    eventWrapper.PublishedTimestampUtc = publishedTimestamp;
 
                     Logger.LogInformation("Handling event {EventWrapper} in method {MethodName}", eventWrapper,
                         method.Name);
@@ -260,6 +263,7 @@ private async Task PublishResponse(EventBase result, Guid eventId)
             (TEvent)result,
             eventId,
             this.GetGrainId());
+        responseWrapper.PublishedTimestampUtc = DateTime.UtcNow;
 
         await PublishAsync(responseWrapper);
     }

src/Aevatar.Core/GAgentBase.Publish.cs

@@ -37,6 +37,7 @@ protected async Task<Guid> PublishAsync<T>(T @event) where T : EventBase
         {
             // Create event wrapper with context propagation
             var eventWrapper = new EventWrapper<T>(@event, eventId, this.GetGrainId());
+            eventWrapper.PublishedTimestampUtc = DateTime.UtcNow;
             if (State.Parent == null)
             {
                 Logger.LogInformation(
@@ -66,7 +67,9 @@ private async Task PublishEventUpwardsAsync<T>(T @event, Guid eventId) where T :
     {
         try
         {
-            await SendEventUpwardsAsync(new EventWrapper<T>(@event, eventId, GrainId));
+            var eventWrapper = new EventWrapper<T>(@event, eventId, GrainId);
+            eventWrapper.PublishedTimestampUtc = DateTime.UtcNow;
+            await SendEventUpwardsAsync(eventWrapper);
             Logger.LogDebug("{GrainId} published {Event} to upwards", GrainId.ToString(),
                 JsonConvert.SerializeObject(@event));
         }
@@ -94,7 +97,7 @@ private async Task SendEventUpwardsAsync<T>(EventWrapper<T> eventWrapper) where
         {
             Logger.LogError("{GrainId} failed to send event {EventWrapper} upwards", GrainId.ToString(),
                 eventWrapper);
-            throw new EventPublishingException($"{GrainId.ToString()} failed to event event upwards", ex);
+            throw new EventPublishingException($"{GrainId.ToString()} failed to send event upwards", ex);
         }
     }
 
@@ -111,7 +114,7 @@ private async Task SendEventToSelfAsync<T>(EventWrapper<T> eventWrapper) where T
         {
             Logger.LogError("{GrainId} failed to send event {EventWrapper} to itself", GrainId.ToString(),
                 eventWrapper);
-            throw new EventPublishingException($"{GrainId.ToString()} failed to event event to itself", ex);
+            throw new EventPublishingException($"{GrainId.ToString()} failed to send event to itself", ex);
         }
     }
 
@@ -135,7 +138,7 @@ private async Task SendEventDownwardsAsync<T>(EventWrapper<T> eventWrapper) wher
         {
             Logger.LogError("{GrainId} failed to send event {EventWrapper} downwards", GrainId.ToString(),
                 eventWrapper);
-            throw new EventPublishingException($"{GrainId.ToString()} failed to event event downwards", ex);
+            throw new EventPublishingException($"{GrainId.ToString()} failed to send event downwards", ex);
         }
     }
 }

src/Aevatar.Core/Observability/EventPublishLatencyMetrics.cs

@@ -0,0 +1,25 @@
+using System.Diagnostics.Metrics;
+using Aevatar.Core.Abstractions;
+using Microsoft.Extensions.Logging;
+
+namespace Aevatar.Core.Observability;
+
+public static class EventPublishLatencyMetrics
+{
+    private static readonly Meter Meter = new(OpenTelemetryConstants.AevatarStreamsMeterName);
+    private static readonly Histogram<double> PublishLatencyHistogram = Meter.CreateHistogram<double>(
+        OpenTelemetryConstants.EventPublishLatencyHistogram, "s", "Event publish-to-consume latency");
+
+    public static void Record(double latency, EventWrapperBase item, string? methodName, string? parameterTypeName, ILogger? logger = null)
+    {
+        var eventType = item.GetType().Name;
+        var grainId = item.GetType().GetProperty("GrainId")?.GetValue(item)?.ToString() ?? "unknown";
+        var eventId = item.GetType().GetProperty("EventId")?.GetValue(item)?.ToString() ?? "unknown";
+        PublishLatencyHistogram.Record(latency,
+            new KeyValuePair<string, object?>("grain_id", grainId),
+            new KeyValuePair<string, object?>("method_name", methodName ?? "unknown"),
+            new KeyValuePair<string, object?>("parameter_type", parameterTypeName ?? "unknown"));
+        logger?.LogInformation("[PublishLatency] latency={Latency}s event_type={EventType} grain_id={GrainId} event_id={EventId} method={MethodName} parameter={ParameterType}",
+            latency, eventType, grainId, eventId, methodName ?? "unknown", parameterTypeName ?? "unknown");
+    }
+}