Singleton class and settings based on current Akka Typed implementati…

…on (#6050)

Co-authored-by: Aaron Stannard <[email protected]>

docs/articles/clustering/cluster-singleton.md

@@ -4,6 +4,8 @@ title: Cluster Singleton
 ---
 # Cluster Singleton
 
+## Introduction
+
 For some use cases it is convenient and sometimes also mandatory to ensure that you have exactly one actor of a certain type running somewhere in the cluster.
 
 Some examples:
@@ -15,16 +17,22 @@ Some examples:
 
 Using a singleton should not be the first design choice. It has several drawbacks, such as single-point of bottleneck. Single-point of failure is also a relevant concern, but for some cases this feature takes care of that by making sure that another singleton instance will eventually be started.
 
+### Singleton Manager
+
 The cluster singleton pattern is implemented by `Akka.Cluster.Tools.Singleton.ClusterSingletonManager`. It manages one singleton actor instance among all cluster nodes or a group of nodes tagged with a specific role. `ClusterSingletonManager` is an actor that is supposed to be started on all nodes, or all nodes with specified role, in the cluster. The actual singleton actor is started by the `ClusterSingletonManager` on the oldest node by creating a child actor from supplied Props. `ClusterSingletonManager` makes sure that at most one singleton instance is running at any point in time.
 
 The singleton actor is always running on the oldest member with specified role. The oldest member is determined by `Akka.Cluster.Member#IsOlderThan`. This can change when removing that member from the cluster. Be aware that there is a short time period when there is no active singleton during the hand-over process.
 
 The cluster failure detector will notice when oldest node becomes unreachable due to things like CLR crash, hard shut down, or network failure. Then a new oldest node will take over and a new singleton actor is created. For these failure scenarios there will not be a graceful hand-over, but more than one active singletons is prevented by all reasonable means. Some corner cases are eventually resolved by configurable timeouts.
 
+### Singleton Proxy
+
 You can access the singleton actor by using the provided `Akka.Cluster.Tools.Singleton.ClusterSingletonProxy`, which will route all messages to the current instance of the singleton. The proxy will keep track of the oldest node in the cluster and resolve the singleton's `IActorRef` by explicitly sending the singleton's `ActorSelection` the `Akka.Actor.Identify` message and waiting for it to reply. This is performed periodically if the singleton doesn't reply within a certain (configurable) time. Given the implementation, there might be periods of time during which the `IActorRef` is unavailable, e.g., when a node leaves the cluster. In these cases, the proxy will buffer the messages sent to the singleton and then deliver them when the singleton is finally available. If the buffer is full the `ClusterSingletonProxy` will drop old messages when new messages are sent via the proxy. The size of the buffer is configurable and it can be disabled by using a buffer size of 0.
 
 It's worth noting that messages can always be lost because of the distributed nature of these actors. As always, additional logic should be implemented in the singleton (acknowledgement) and in the client (retry) actors to ensure at-least-once message delivery.
 
+The singleton instance will not run on members with status `WeaklyUp`.
+
 ## Potential Problems to Be Aware Of
 
 This pattern may seem to be very tempting to use at first, but it has several drawbacks, some of them are listed below:
@@ -35,6 +43,65 @@ This pattern may seem to be very tempting to use at first, but it has several dr
 
 Especially the last point is something you should be aware of — in general when using the Cluster Singleton pattern you should take care of downing nodes yourself and not rely on the timing based auto-down feature.
 
+## An Example (Simplified API)
+
+Any `Actor` can be run as a singleton. E.g. a basic counter:
+
+```csharp
+public class Counter : UntypedActor
+{
+    public sealed class Increment
+    {
+        public static Increment Instance => new Increment();
+        private Increment() { }
+    }
+
+    public sealed class GetValue
+    {
+        public IActorRef ReplyTo { get; }
+        public GetValue(IActorRef replyTo) => ReplyTo = replyTo;
+    }
+
+    public sealed class GoodByeCounter
+    {
+        public static GoodByeCounter Instance => new GoodByeCounter();
+        private GoodByeCounter() { }
+    }
+
+    private int _value;
+
+    public static Props Props => Props.Create<Counter>();
+
+    protected override void OnReceive(object message)
+    {
+        switch (message)
+        {
+            case Increment _:
+                _value++;
+                break;
+            case GetValue msg:
+                msg.ReplyTo.Tell(_value);
+                break;
+            case GoodByeCounter _:
+                Context.Stop(Self);
+                break;
+            default:
+                base.Unhandled(message);
+                break;
+        }
+    }
+}
+```
+
+Then on every node in the cluster, or every node with a given role, use the `ClusterSingleton` extension to spawn the singleton:
+
+```csharp
+var singleton = ClusterSingleton.Get(system);
+// start if needed and provide a proxy to a named singleton
+var proxy = singleton.Init(SingletonActor.Create(Counter.Props, "GlobalCounter"));
+proxy.Tell(Counter.Increment);
+```
+
 ## An Example
 
 Assume that we need one single entry point to an external system. An actor that receives messages from a JMS queue with the strict requirement that only one JMS consumer must exist to be make sure that the messages are processed in order. That is perhaps not how one would like to design things, but a typical real-world scenario when integrating with external systems.

src/contrib/cluster/Akka.Cluster.Tools.Tests/Singleton/ClusterSingletonApiSpec.cs

@@ -0,0 +1,133 @@
+//-----------------------------------------------------------------------
+// <copyright file="ClusterSingletonConfigSpec.cs" company="Akka.NET Project">
+//     Copyright (C) 2009-2021 Lightbend Inc. <http://www.lightbend.com>
+//     Copyright (C) 2013-2021 .NET Foundation <https://github.com/akkadotnet/akka.net>
+// </copyright>
+//-----------------------------------------------------------------------
+
+using System;
+using System.Threading.Tasks;
+using Akka.Actor;
+using Akka.Cluster.Tools.Singleton;
+using Akka.Configuration;
+using Akka.TestKit;
+using Akka.TestKit.Configs;
+using Akka.TestKit.Extensions;
+using Xunit;
+using Xunit.Abstractions;
+
+namespace Akka.Cluster.Tools.Tests.Singleton
+{
+    public class ClusterSingletonApiSpec : AkkaSpec
+    {
+        #region Internal
+
+        public sealed class Pong
+        {
+            public static Pong Instance => new Pong();
+            private Pong() { }
+        }
+
+        public sealed class Ping
+        {
+            public IActorRef RespondTo { get; }
+            public Ping(IActorRef respondTo) => RespondTo = respondTo;
+        }
+
+        public sealed class Perish
+        {
+            public static Perish Instance => new Perish();
+            private Perish() { }
+        }
+
+        public class PingPong : UntypedActor
+        {
+            protected override void OnReceive(object message)
+            {
+                switch (message)
+                {
+                    case Ping ping:
+                        ping.RespondTo.Tell(Pong.Instance);
+                        break;
+                    case Perish _:
+                        Context.Stop(Self);
+                        break;
+                }
+            }
+        }
+
+        #endregion
+
+        private readonly Cluster _clusterNode1;
+        private readonly Cluster _clusterNode2;
+        private readonly ActorSystem _system2;
+
+        public static Config GetConfig() => ConfigurationFactory.ParseString(@"
+            akka.loglevel = DEBUG
+            akka.actor.provider = ""cluster""
+            akka.cluster.roles = [""singleton""]
+            akka.remote {
+                dot-netty.tcp {
+                    hostname = ""127.0.0.1""
+                    port = 0
+                }
+            }").WithFallback(TestConfigs.DefaultConfig);
+
+        public ClusterSingletonApiSpec(ITestOutputHelper testOutput)
+            : base(GetConfig(), testOutput)
+        {
+            _clusterNode1 = Cluster.Get(Sys);
+
+            _system2 = ActorSystem.Create(
+                Sys.Name,
+                ConfigurationFactory.ParseString("akka.cluster.roles = [\"singleton\"]").WithFallback(Sys.Settings.Config));
+
+            _clusterNode2 = Cluster.Get(_system2);
+        }
+
+        [Fact]
+        public void A_cluster_singleton_must_be_accessible_from_two_nodes_in_a_cluster()
+        {
+            var node1UpProbe = CreateTestProbe(Sys);
+            var node2UpProbe = CreateTestProbe(Sys);
+
+            _clusterNode1.Join(_clusterNode1.SelfAddress);
+            node1UpProbe.AwaitAssert(() => _clusterNode1.SelfMember.Status.ShouldBe(MemberStatus.Up), TimeSpan.FromSeconds(3));
+
+            _clusterNode2.Join(_clusterNode2.SelfAddress);
+            node2UpProbe.AwaitAssert(() => _clusterNode2.SelfMember.Status.ShouldBe(MemberStatus.Up), TimeSpan.FromSeconds(3));
+
+            var cs1 = ClusterSingleton.Get(Sys);
+            var cs2 = ClusterSingleton.Get(_system2);
+
+            var settings = ClusterSingletonSettings.Create(Sys).WithRole("singleton");
+            var node1ref = cs1.Init(SingletonActor.Create(Props.Create<PingPong>(), "ping-pong").WithStopMessage(Perish.Instance).WithSettings(settings));
+            var node2ref = cs2.Init(SingletonActor.Create(Props.Create<PingPong>(), "ping-pong").WithStopMessage(Perish.Instance).WithSettings(settings));
+
+            // subsequent spawning returns the same refs
+            cs1.Init(SingletonActor.Create(Props.Create<PingPong>(), "ping-pong").WithStopMessage(Perish.Instance).WithSettings(settings)).ShouldBe(node1ref);
+            cs2.Init(SingletonActor.Create(Props.Create<PingPong>(), "ping-pong").WithStopMessage(Perish.Instance).WithSettings(settings)).ShouldBe(node2ref);
+
+            var node1PongProbe = CreateTestProbe(Sys);
+            var node2PongProbe = CreateTestProbe(_system2);
+
+            node1PongProbe.AwaitAssert(() =>
+            {
+                node1ref.Tell(new Ping(node1PongProbe.Ref));
+                node1PongProbe.ExpectMsg<Pong>();
+            }, TimeSpan.FromSeconds(3));
+
+            node2PongProbe.AwaitAssert(() =>
+            {
+                node2ref.Tell(new Ping(node2PongProbe.Ref));
+                node2PongProbe.ExpectMsg<Pong>();
+            }, TimeSpan.FromSeconds(3));
+        }
+
+        protected override async Task AfterAllAsync()
+        {
+            await base.AfterAllAsync();
+            await _system2.Terminate().AwaitWithTimeout(TimeSpan.FromSeconds(3));
+        }
+    }
+}

src/contrib/cluster/Akka.Cluster.Tools/Singleton/ClusterSingleton.cs

@@ -0,0 +1,131 @@
+//-----------------------------------------------------------------------
+// <copyright file="ClusterSingletonManager.cs" company="Akka.NET Project">
+//     Copyright (C) 2009-2021 Lightbend Inc. <http://www.lightbend.com>
+//     Copyright (C) 2013-2021 .NET Foundation <https://github.com/akkadotnet/akka.net>
+// </copyright>
+//-----------------------------------------------------------------------
+
+using System;
+using System.Collections.Concurrent;
+using Akka.Actor;
+using Akka.Annotations;
+using Akka.Util;
+
+namespace Akka.Cluster.Tools.Singleton
+{
+    /// <summary>
+    /// This class is not intended for user extension other than for test purposes (e.g. stub implementation). 
+    /// More methods may be added in the future and that may break such implementations.
+    /// </summary>
+    [DoNotInherit]
+    public class ClusterSingleton : IExtension
+    {
+        private readonly ActorSystem _system;
+        private readonly Lazy<Cluster> _cluster;
+        private readonly ConcurrentDictionary<string, IActorRef> _proxies = new ConcurrentDictionary<string, IActorRef>();
+
+        public static ClusterSingleton Get(ActorSystem system) =>
+            system.WithExtension<ClusterSingleton, ClusterSingletonProvider>();
+
+        public ClusterSingleton(ExtendedActorSystem system)
+        {
+            _system = system;
+            _cluster = new Lazy<Cluster>(() => Cluster.Get(system));
+        }
+
+        /// <summary>
+        /// Start if needed and provide a proxy to a named singleton.
+        /// 
+        /// <para>If there already is a manager running for the given `singletonName` on this node, no additional manager is started.</para>
+        /// <para>If there already is a proxy running for the given `singletonName` on this node, an <see cref="IActorRef"/> to that is returned.</para>
+        /// </summary>
+        /// <returns>A proxy actor that can be used to communicate with the singleton in the cluster</returns>
+        public IActorRef Init(SingletonActor singleton)
+        {
+            var settings = singleton.Settings.GetOrElse(ClusterSingletonSettings.Create(_system));
+            if (settings.ShouldRunManager(_cluster.Value))
+            {
+                var managerName = ManagerNameFor(singleton.Name);
+                try
+                {
+                    _system.ActorOf(ClusterSingletonManager.Props(
+                        singletonProps: singleton.Props,
+                        terminationMessage: singleton.StopMessage.GetOrElse(PoisonPill.Instance),
+                        settings: settings.ToManagerSettings(singleton.Name)),
+                        managerName);
+                }
+                catch (InvalidActorNameException ex) when (ex.Message.EndsWith("is not unique!"))
+                {
+                    // This is fine. We just wanted to make sure it is running and it already is
+                }
+            }
+
+            return GetProxy(singleton.Name, settings);
+        }
+
+        private IActorRef GetProxy(string name, ClusterSingletonSettings settings)
+        {
+            IActorRef ProxyCreator()
+            {
+                var proxyName = $"singletonProxy{name}";
+                return _system.ActorOf(ClusterSingletonProxy.Props(
+                    singletonManagerPath: $"/user/{ManagerNameFor(name)}",
+                    settings: settings.ToProxySettings(name)),
+                    proxyName);
+            }
+
+            return _proxies.GetOrAdd(name, _ => ProxyCreator());
+        }
+
+
+        private string ManagerNameFor(string singletonName) => $"singletonManager{singletonName}";
+    }
+
+    public class ClusterSingletonProvider : ExtensionIdProvider<ClusterSingleton>
+    {
+        public override ClusterSingleton CreateExtension(ExtendedActorSystem system) => new ClusterSingleton(system);
+    }
+
+    public class SingletonActor
+    {
+        public string Name { get; }
+
+        public Props Props { get; }
+
+        public Option<object> StopMessage { get; }
+
+        public Option<ClusterSingletonSettings> Settings { get; }
+
+        public static SingletonActor Create(Props props, string name) =>
+            new SingletonActor(name, props, Option<object>.None, Option<ClusterSingletonSettings>.None);
+
+        private SingletonActor(string name, Props props, Option<object> stopMessage, Option<ClusterSingletonSettings> settings)
+        {
+            Name = name;
+            Props = props;
+            StopMessage = stopMessage;
+            Settings = settings;
+        }
+
+        /// <summary>
+        /// <see cref="Props"/> of the singleton actor, such as dispatcher settings.
+        /// </summary>
+        public SingletonActor WithProps(Props props) => Copy(props: props);
+
+        /// <summary>
+        /// Message sent to the singleton to tell it to stop, e.g. when being migrated. 
+        /// If this is not defined, a <see cref="PoisonPill"/> will be used instead. 
+        /// It can be useful to define a custom stop message if the singleton needs to 
+        /// perform some asynchronous cleanup or interactions before stopping.
+        /// </summary>
+        public SingletonActor WithStopMessage(object stopMessage) => Copy(stopMessage: stopMessage);
+
+        /// <summary>
+        /// Additional settings, typically loaded from configuration.
+        /// </summary>
+        public SingletonActor WithSettings(ClusterSingletonSettings settings) => Copy(settings: settings);
+
+        private SingletonActor Copy(string name = null, Props props = null, Option<object> stopMessage = default, Option<ClusterSingletonSettings> settings = default) =>
+            new SingletonActor(name ?? Name, props ?? Props, stopMessage.HasValue ? stopMessage : StopMessage, settings.HasValue ? settings : Settings);
+    }
+}