using Akka.Actor; using Akka.Cluster; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using ZB.MOM.WW.OtOpcUa.Commons.Interfaces; using CommonsNodeId = ZB.MOM.WW.OtOpcUa.Commons.Types.NodeId; namespace ZB.MOM.WW.OtOpcUa.Cluster; /// /// Thread-safe live view of cluster membership and role topology. Subscribes to /// , , and /// through an internal subscriber actor and keeps /// a snapshot of role-to-members + role-to-leader. The CLR-facing event surface is /// . /// public sealed class ClusterRoleInfo : IClusterRoleInfo, IDisposable { private readonly Akka.Cluster.Cluster _cluster; private readonly ILogger _logger; private readonly CommonsNodeId _localNode; private readonly HashSet _localRoles; private readonly string? _clusterRole; private readonly string? _clusterId; private readonly object _lock = new(); private readonly Dictionary _roleLeaders = new(StringComparer.Ordinal); private readonly Dictionary> _membersByRole = new(StringComparer.Ordinal); private IActorRef? _subscriber; /// Initializes a new instance of the ClusterRoleInfo class. /// The Akka actor system. /// The cluster configuration options. /// The logger instance. public ClusterRoleInfo(ActorSystem system, IOptions options, ILogger logger) { _cluster = Akka.Cluster.Cluster.Get(system); _logger = logger; // NodeId encodes host:port so cluster members on shared hosts (test loopback, dev VMs // sharing a bind IP) stay distinct. Production hosts have unique DNS names so the port // suffix is harmless redundancy. _localNode = CommonsNodeId.Parse($"{options.Value.PublicHostname}:{options.Value.Port}"); // OrdinalIgnoreCase (Gitea #532): Cluster:Roles binds verbatim, so a node authored with 'Driver' // used to answer HasRole("driver") false while gossiping a role Akka would still match against // its own ordinal comparison — the node was a driver to the cluster and not to itself. _localRoles = new HashSet(options.Value.Roles, StringComparer.OrdinalIgnoreCase); // Derived from the node's OWN configured roles (not live Cluster.State) so this identity is // available at host-build time, before the cluster forms. Iterate the ORIGINAL array — not // _localRoles — because first-wins requires configuration order, which HashSet does not // guarantee. The OTOPCUA_ROLES fallback is already applied to the bound options by // AddOtOpcUaCluster's PostConfigure, so nothing is re-resolved here. var clusterRoles = RoleParser.ClusterRolesIn(options.Value.Roles); if (clusterRoles.Count > 0) { _clusterRole = clusterRoles[0]; _clusterId = RoleParser.ClusterIdFromRole(_clusterRole); if (clusterRoles.Count > 1) { _logger.LogWarning( "Node configured with {Count} cluster-scoped roles ({Roles}); using the first ({ClusterRole})", clusterRoles.Count, string.Join(", ", clusterRoles), _clusterRole); } } SeedFromCurrentState(); _subscriber = system.ActorOf(Props.Create(() => new SubscriberActor(this)), "clusterroleinfo-subscriber"); } /// public CommonsNodeId LocalNode => _localNode; /// public IReadOnlySet LocalRoles => _localRoles; /// public string? ClusterRole => _clusterRole; /// public string? ClusterId => _clusterId; /// public bool HasRole(string role) => _localRoles.Contains(role); /// public IReadOnlyList MembersWithRole(string role) { lock (_lock) { if (!_membersByRole.TryGetValue(role, out var members)) return Array.Empty(); return members .Select(m => ToNodeId(m.Address)) .ToArray(); } } /// public CommonsNodeId? RoleLeader(string role) { lock (_lock) { return _roleLeaders.TryGetValue(role, out var leader) && leader is not null ? ToNodeId(leader.Address) : (CommonsNodeId?)null; } } /// Occurs when the leader for a role changes. public event EventHandler? RoleLeaderChanged; private void SeedFromCurrentState() { var snapshot = _cluster.State; lock (_lock) { foreach (var member in snapshot.Members) foreach (var role in member.Roles) { if (!_membersByRole.TryGetValue(role, out var set)) _membersByRole[role] = set = new HashSet(); set.Add(member); } foreach (var role in snapshot.Members.SelectMany(m => m.Roles).Distinct()) { var leaderAddr = snapshot.RoleLeader(role); _roleLeaders[role] = leaderAddr is not null ? snapshot.Members.FirstOrDefault(m => m.Address == leaderAddr) : null; } } } /// Handles a cluster member event (member up/removed). /// The member event from the cluster. internal void HandleMemberEvent(ClusterEvent.IMemberEvent evt) { lock (_lock) { switch (evt) { case ClusterEvent.MemberUp up: foreach (var role in up.Member.Roles) { if (!_membersByRole.TryGetValue(role, out var set)) _membersByRole[role] = set = new HashSet(); set.Add(up.Member); } break; case ClusterEvent.MemberRemoved removed: foreach (var role in removed.Member.Roles) if (_membersByRole.TryGetValue(role, out var set)) set.Remove(removed.Member); break; } } } /// Handles a role leader change event. /// The role leader changed event from the cluster. internal void HandleRoleLeaderChanged(ClusterEvent.RoleLeaderChanged evt) { CommonsNodeId? previous = null; CommonsNodeId? next = null; var raise = false; lock (_lock) { _roleLeaders.TryGetValue(evt.Role, out var prevMember); if (prevMember is not null) previous = ToNodeId(prevMember.Address); var nextMember = evt.Leader is null ? null : _cluster.State.Members.FirstOrDefault(m => m.Address == evt.Leader); _roleLeaders[evt.Role] = nextMember; if (nextMember is not null) next = ToNodeId(nextMember.Address); raise = !Nullable.Equals(previous, next); } if (!raise) return; try { RoleLeaderChanged?.Invoke(this, new RoleLeaderChangedEventArgs { Role = evt.Role, PreviousLeader = previous, NewLeader = next, }); } catch (Exception ex) { _logger.LogWarning(ex, "RoleLeaderChanged subscriber threw for role {Role}", evt.Role); } } private static CommonsNodeId ToNodeId(Akka.Actor.Address address) => CommonsNodeId.Parse($"{address.Host ?? string.Empty}:{address.Port ?? 0}"); /// Disposes the ClusterRoleInfo and stops the subscriber actor. public void Dispose() { _subscriber?.Tell(PoisonPill.Instance); _subscriber = null; } private sealed class SubscriberActor : ReceiveActor { /// Initializes a new instance of the SubscriberActor class. /// The ClusterRoleInfo instance to forward events to. public SubscriberActor(ClusterRoleInfo owner) { Receive(e => owner.HandleMemberEvent(e)); Receive(e => owner.HandleRoleLeaderChanged(e)); // LeaderChanged is intentionally a no-op here: ServiceLevel lives in RedundancyStateActor // (per-cluster mesh Phase 6: a cluster-{ClusterId}-scoped singleton spawned on every // driver node, no longer the admin-role singleton it once was) and has no consumer on this // ClusterRoleInfo path by design. Receive(_ => { }); Receive(_ => { /* seeded from initial snapshot */ }); } /// protected override void PreStart() { Akka.Cluster.Cluster.Get(Context.System).Subscribe( Self, ClusterEvent.InitialStateAsEvents, typeof(ClusterEvent.IMemberEvent), typeof(ClusterEvent.LeaderChanged), typeof(ClusterEvent.RoleLeaderChanged)); } /// protected override void PostStop() => Akka.Cluster.Cluster.Get(Context.System).Unsubscribe(Self); } }