feat: create telemetry workflow

samples/KafkaFlow.Sample.Dashboard/KafkaFlow.Sample.Dashboard.csproj

@@ -16,7 +16,6 @@
   </ItemGroup>  
 
   <Target Name="DebugNpmInstall" BeforeTargets="Build">
-    <!-- Ensure Node.js is installed -->
     <Exec Command="node --version" ContinueOnError="true">
       <Output TaskParameter="ExitCode" PropertyName="ErrorCode" />
     </Exec>
@@ -26,11 +25,15 @@
   </Target>
 
   <Target Name="PublishRunWebpack" AfterTargets="ComputeFilesToPublish">
-    <!-- As part of publishing, ensure the JS resources are freshly built in production mode -->
     <Exec WorkingDirectory="$(OutDir)$(SpaRoot)" Command="npm run build -- --prod" />
-    <Exec WorkingDirectory="$(OutDir)$(SpaRoot)" Command="if not exist $(PublishDir)$(SpaRoot)dist mkdir $(PublishDir)$(SpaRoot)dist"/>
-    <Exec WorkingDirectory="$(OutDir)$(SpaRoot)" Command="del $(PublishDir)$(SpaRoot)dist\*.* /s /q"/>
-    <Exec WorkingDirectory="$(OutDir)$(SpaRoot)" Command="move dist\dashboard\* $(PublishDir)$(SpaRoot)dist"/>    
+
+    <Exec WorkingDirectory="$(OutDir)$(SpaRoot)" Command="if not exist $(PublishDir)$(SpaRoot)dist mkdir $(PublishDir)$(SpaRoot)dist" Condition=" '$(OS)' == 'Windows_NT' "/>
+    <Exec WorkingDirectory="$(OutDir)$(SpaRoot)" Command="del $(PublishDir)$(SpaRoot)dist\*.* /s /q" Condition=" '$(OS)' == 'Windows_NT' "/>
+    <Exec WorkingDirectory="$(OutDir)$(SpaRoot)" Command="move dist\dashboard\* $(PublishDir)$(SpaRoot)dist" Condition=" '$(OS)' == 'Windows_NT' "/>
+
+    <Exec WorkingDirectory="$(OutDir)$(SpaRoot)" Command="mkdir $(PublishDir)$(SpaRoot)dist -p" Condition=" '$(OS)' == 'Unix' "/>
+    <Exec WorkingDirectory="$(OutDir)$(SpaRoot)" Command="rm $(PublishDir)$(SpaRoot)dist/* -rf" Condition=" '$(OS)' == 'Unix' "/>
+    <Exec WorkingDirectory="$(OutDir)$(SpaRoot)" Command="mv dist/dashboard/* $(PublishDir)$(SpaRoot)dist" Condition=" '$(OS)' == 'Unix' "/>
   </Target>
 
 </Project>

src/KafkaFlow.Admin.WebApi/Adapters/ConsumerResponseAdapter.cs

@@ -41,6 +41,7 @@ private static IEnumerable<PartitionAssignment> GetLocalInfo(IMessageConsumer co
                     Topic = c.Key,
                     HostName = Environment.MachineName,
                     PausedPartitions = c.Select(x => x.Partition.Value).ToList(),
+                    LastUpdate = DateTime.Now,
                 })
                 .Union(consumer.RunningPartitions
                     .GroupBy(c => c.Topic)
@@ -49,6 +50,7 @@ private static IEnumerable<PartitionAssignment> GetLocalInfo(IMessageConsumer co
                         Topic = c.Key,
                         HostName = Environment.MachineName,
                         RunningPartitions = c.Select(x => x.Partition.Value).ToList(),
+                        LastUpdate = DateTime.Now,
                     }));
         }
 

src/KafkaFlow.Admin/Extensions/ClusterConfigurationBuilderExtensions.cs

@@ -33,7 +33,7 @@ public static IClusterConfigurationBuilder EnableAdminMessages(
             cluster.DependencyConfigurator.AddSingleton<IAdminProducer, AdminProducer>();
 
             cluster.DependencyConfigurator.AddSingleton<IMemoryCache, MemoryCache>();
-            cluster.DependencyConfigurator.AddSingleton<ITelemetryCache, TelemetryCache>();
+            cluster.DependencyConfigurator.AddSingleton<ITelemetryCache, InMemoryTelemetryCache>();
 
             return cluster
                 .AddProducer<AdminProducer>(
@@ -81,7 +81,7 @@ public static IClusterConfigurationBuilder EnableTelemetry(
             string consumerGroup)
         {
             cluster.DependencyConfigurator.AddSingleton<IMemoryCache, MemoryCache>();
-            cluster.DependencyConfigurator.AddSingleton<ITelemetryCache, TelemetryCache>();
+            cluster.DependencyConfigurator.AddSingleton<ITelemetryCache, InMemoryTelemetryCache>();
 
             var groupId =
                 $"{consumerGroup}-{Environment.MachineName}-{Convert.ToBase64String(Guid.NewGuid().ToByteArray())}";

src/KafkaFlow.Admin/TelemetryCache.cs → src/KafkaFlow.Admin/InMemoryTelemetryCache.cs

@@ -7,15 +7,15 @@ namespace KafkaFlow.Admin
     /// <summary>
     /// Provide cache operations related to telemetry data
     /// </summary>
-    public class TelemetryCache : ITelemetryCache
+    public class InMemoryTelemetryCache : ITelemetryCache
     {
         private readonly IMemoryCache cache;
 
         /// <summary>
-        /// Initializes a new instance of the <see cref="TelemetryCache"/> class.
+        /// Initializes a new instance of the <see cref="InMemoryTelemetryCache"/> class.
         /// </summary>
-        /// <param name="cache">The memory cache interface to get metric data</param>
-        public TelemetryCache(IMemoryCache cache) => this.cache = cache;
+        /// <param name="cache">The memory cache class to manage telemetry data</param>
+        public InMemoryTelemetryCache(IMemoryCache cache) => this.cache = cache;
 
         /// <inheritdoc />
         public List<ConsumerMetric> Get(string groupId, string consumerName)

src/KafkaFlow.Admin/TelemetryScheduler.cs

@@ -6,21 +6,21 @@ namespace KafkaFlow.Admin
 
     internal class TelemetryScheduler
     {
-        private static readonly Lazy<Dictionary<string, Timer>> timers = new (() => new Dictionary<string, Timer>());
+        private static readonly Lazy<Dictionary<string, Timer>> Timers = new (() => new Dictionary<string, Timer>());
 
         public static void Set(string key, TimerCallback callback, TimeSpan dueTime, TimeSpan period)
         {
             Unset(key);
 
-            timers.Value[key] = new Timer(callback, null, dueTime, period);
+            Timers.Value[key] = new Timer(callback, null, dueTime, period);
         }
 
         public static void Unset(string key)
         {
-            if (timers.Value.TryGetValue(key, out var timer))
+            if (Timers.Value.TryGetValue(key, out var timer))
             {
                 timer.Dispose();
-                timers.Value.Remove(key);
+                Timers.Value.Remove(key);
             }
         }
     }

src/KafkaFlow.UnitTests/Consumer/ConsumerManagerTests.cs

@@ -36,6 +36,10 @@ public void Setup()
             this.logHandlerMock = new Mock<ILogHandler>(MockBehavior.Strict);
             this.dependencyResolver = new Mock<IDependencyResolver>();
 
+            this.consumerMock
+                .Setup(x => x.FlowManager)
+                .Returns(new Mock<IConsumerFlowManager>().Object);
+
             this.consumerMock
                 .Setup(x => x.OnPartitionsAssigned(It.IsAny<Action<IDependencyResolver, IConsumer<byte[], byte[]>, List<TopicPartition>>>()))
                 .Callback((Action<IDependencyResolver, IConsumer<byte[], byte[]>, List<TopicPartition>> value) => this.onPartitionAssignedHandler = value);

src/KafkaFlow/Configuration/ClusterConfiguration.cs

@@ -7,7 +7,7 @@ namespace KafkaFlow.Configuration
     internal class ClusterConfiguration
     {
         private readonly Func<SecurityInformation> securityInformationHandler;
-        private Action<IDependencyResolver> onStopHandler = _ => { };
+        private readonly Action<IDependencyResolver> onStopHandler = _ => { };
         private readonly List<IProducerConfiguration> producers = new();
         private readonly List<IConsumerConfiguration> consumers = new();
 
@@ -31,12 +31,12 @@ public ClusterConfiguration(
 
         public IReadOnlyCollection<IConsumerConfiguration> Consumers => this.consumers.AsReadOnly();
 
+        public Action<IDependencyResolver> OnStopHandler => this.onStopHandler;
+
         public void AddConsumers(IEnumerable<IConsumerConfiguration> configurations) => this.consumers.AddRange(configurations);
 
         public void AddProducers(IEnumerable<IProducerConfiguration> configurations) => this.producers.AddRange(configurations);
 
         public SecurityInformation GetSecurityInformation() => this.securityInformationHandler?.Invoke();
-
-        public Action<IDependencyResolver> OnStopHandler => this.onStopHandler;
     }
 }

src/KafkaFlow/Configuration/ConsumerConfigurationBuilder.cs

@@ -7,7 +7,7 @@ namespace KafkaFlow.Configuration
     using Confluent.Kafka;
     using KafkaFlow.Consumers.DistributionStrategies;
 
-    internal class ConsumerConfigurationBuilder : IConsumerConfigurationBuilder
+    internal sealed class ConsumerConfigurationBuilder : IConsumerConfigurationBuilder
     {
         private readonly List<string> topics = new();
         private readonly List<Action<string>> statisticsHandlers = new();

src/KafkaFlow/Configuration/IConsumerConfiguration.cs

@@ -30,7 +30,7 @@ public interface IConsumerConfiguration
         string ConsumerName { get; }
 
         /// <summary>
-        /// Gets the consumer readonly flag
+        /// Gets a value indicating whether the consumer is readonly or not
         /// </summary>
         bool IsReadonly { get; }
 

src/KafkaFlow/Configuration/KafkaConfiguration.cs

@@ -1,14 +1,11 @@
 namespace KafkaFlow.Configuration
 {
-    using System;
     using System.Collections.Generic;
 
     internal class KafkaConfiguration
     {
         private readonly List<ClusterConfiguration> clusters = new();
 
-        public Action<IDependencyResolver> OnStopHandler = _ => { };
-
         public IReadOnlyCollection<ClusterConfiguration> Clusters => this.clusters;
 
         public void AddClusters(IEnumerable<ClusterConfiguration> configurations) => this.clusters.AddRange(configurations);

src/KafkaFlow/Consumers/IConsumerFlowManager.cs

@@ -31,6 +31,10 @@ public interface IConsumerFlowManager : IDisposable
         /// <param name="topicPartitions">A list of partitions</param>
         void Resume(IReadOnlyCollection<TopicPartition> topicPartitions);
 
+        /// <summary>
+        /// Removes the running partitions from the list of paused partitions
+        /// </summary>
+        /// <param name="partitionsRunning">A list of partitions that are running</param>
         void UpdatePausedPartitions(IEnumerable<TopicPartition> partitionsRunning);
     }
 }

src/KafkaFlow/Consumers/IMessageConsumer.cs

@@ -16,7 +16,7 @@ public interface IMessageConsumer
         string ConsumerName { get; }
 
         /// <summary>
-        /// Gets the readonly flag defined in the configuration
+        /// Gets a value indicating whether the consumer is readonly or not
         /// </summary>
         bool IsReadonly { get; }