Extension diagnostics

examples/AspNetGrains/Node1/Program.cs

@@ -29,7 +29,7 @@
 
 app.MapGet("/diagnostics", (ActorSystem system) =>
 {
-    var entries = system.Diagnostics.Get();
+    var entries = system.Diagnostics.GetDiagnostics();
     return entries;
 });
 

examples/AspNetGrains/Node2/Program.cs

@@ -21,7 +21,7 @@
 
 app.MapGet("/diagnostics", (ActorSystem system) =>
 {
-    var entries = system.Diagnostics.Get();
+    var entries = system.Diagnostics.GetDiagnostics();
     return entries;
 });
 

src/Proto.Actor/Diagnostics/DiagnosticsStore.cs

@@ -1,6 +1,9 @@
 using System;
+using System.Collections.Generic;
+using System.Linq;
 using System.Text.Json.Serialization;
 using System.Threading;
+using System.Threading.Tasks;
 using Microsoft.Extensions.Logging;
 using Proto.Utils;
 
@@ -9,12 +12,14 @@ namespace Proto.Diagnostics;
 
 public class DiagnosticsStore
 {
+    private readonly ActorSystem _system;
     private readonly ILogger _logger = Log.CreateLogger<DiagnosticsStore>();
     private readonly ConcurrentSet<DiagnosticsEntry> _entries = new();
     private readonly LogLevel _logLevel;
 
     public DiagnosticsStore(ActorSystem system)
     {
+        _system = system;
         _logLevel = system.Config.DiagnosticsLogLevel;
         RegisterEnvironmentSettings();
     }
@@ -62,9 +67,19 @@ public void RegisterObject(string module, string key, object data)
         }
     }
 
-    public DiagnosticsEntry[] Get()
+    public async Task<DiagnosticsEntry[]> GetDiagnostics()
     {
-        return _entries.ToArray();
+        var entries = new List<DiagnosticsEntry>();
+        var extensions = _system.Extensions.GetAll().ToArray();
+
+        foreach (var e in extensions)
+        {
+            var res = await e.GetDiagnostics();
+            entries.AddRange(res);
+        }
+        entries.AddRange(_entries.ToArray());
+
+        return entries.ToArray();
     }
 }
 

src/Proto.Actor/Extensions/ActorSystemExtensions.cs

@@ -5,6 +5,7 @@
 // -----------------------------------------------------------------------
 
 using System;
+using System.Collections.Generic;
 
 namespace Proto.Extensions;
 
@@ -77,4 +78,15 @@ public void Register<T>(IActorSystemExtension<T> extension) where T : IActorSyst
             _actorSystem.Diagnostics.RegisterEvent("ActorSystem", $"Extension enabled {typeof(T).Name}");
         }
     }
+
+    public IEnumerable<IActorSystemExtension> GetAll()
+    {
+        foreach (var e in _extensions)
+        {
+            if (e == null)
+                continue;
+
+            yield return e;
+        }
+    }
 }

src/Proto.Actor/Extensions/IActorSystemExtension.cs

@@ -4,7 +4,10 @@
 // </copyright>
 // -----------------------------------------------------------------------
 
+using System;
 using System.Threading;
+using System.Threading.Tasks;
+using Proto.Diagnostics;
 
 namespace Proto.Extensions;
 
@@ -16,6 +19,7 @@ public interface IActorSystemExtension
     private static int _nextId;
 
     internal static int GetNextId() => Interlocked.Increment(ref _nextId);
+    Task<DiagnosticsEntry[]> GetDiagnostics() => Task.FromResult(Array.Empty<DiagnosticsEntry>());
 }
 
 /// <summary>

src/Proto.Cluster/Cluster.cs

@@ -18,6 +18,7 @@
 using Proto.Cluster.Metrics;
 using Proto.Cluster.PubSub;
 using Proto.Cluster.Seed;
+using Proto.Diagnostics;
 using Proto.Extensions;
 using Proto.Remote;
 using Proto.Utils;
@@ -35,6 +36,17 @@ public class Cluster : IActorSystemExtension<Cluster>
     private Func<IEnumerable<Measurement<long>>>? _clusterMembersObserver;
     private readonly TaskCompletionSource<bool> _shutdownCompletedTcs = new();
 
+    public async Task<DiagnosticsEntry[]> GetDiagnostics()
+    {
+        var blocked = new DiagnosticsEntry("Cluster", "Blocked", System.Remote().BlockList.BlockedMembers.ToArray());
+
+        var t = await Gossip.GetState<ClusterTopology>(GossipKeys.Topology);
+
+        var topology = new DiagnosticsEntry("Cluster", "Topology", t);
+
+        return new[] { blocked, topology };
+    }
+
     public Cluster(ActorSystem system, ClusterConfig config)
     {
         System = system;

src/Proto.Cluster/Gossip/Gossiper.cs

@@ -184,7 +184,13 @@ internal Task StartAsync()
             _cluster.Config.GossipMaxSend));
 
         _pid = _context.SpawnNamedSystem(props, GossipActorName);
-        _cluster.System.EventStream.Subscribe<ClusterTopology>(topology => _context.Send(_pid, topology));
+        _cluster.System.EventStream.Subscribe<ClusterTopology>(topology =>
+        {
+            var tmp = topology.Clone();
+            tmp.Joined.Clear();
+            tmp.Left.Clear();
+            _context.Send(_pid, tmp);
+        });
         Logger.LogInformation("Started Cluster Gossip");
         _ = SafeTask.Run(GossipLoop);
 

src/Proto.Cluster/Member/MemberList.cs

@@ -305,7 +305,6 @@ private void SelfBlocked()
     private void BroadcastTopologyChanges(ClusterTopology topology)
     {
         _system.Logger()?.LogDebug("MemberList sending state");
-        _cluster.Gossip.SetState(GossipKeys.Topology, topology);
         _eventStream.Publish(topology);
 
         //Console.WriteLine($"{_system.Id} Broadcasting {topology.TopologyHash} - {topology.Members.Count}");