Support for Multiple Endpoints (#1083)

* expose command line parameter for endpoints

* return list of endpoints from hostname value

* change endpoint to endpoints in Options.cs

* switch back to string and do explicit split during address validation

* create multiple server tcp instances to support multiple endpoints

* add test for multi endpoints

* fix testUtils regression

* addloopback test

* fix typo

* logger exception message instead of stack trace

* allow unix and tcp socket use

* calculate ip to advertise from endpoint list

* remove ip address logging

* check that at least one endpoint is TCP for cluster

* first round of comments

* remove empty entries

* add multiple socket test

* nit: rename

* add better comments

* handle dash in addresses from bind optio

* add test for checking parsing of multiple addresses

* remove unused code

* Fix cluster BDN and add ClusterAnnounce options

* sigh! add comma separator

* remove trim

* log config ip

---------

Co-authored-by: Tal Zaccai <[email protected]>

Author: vazois

benchmark/BDN.benchmark/Cluster/ClusterContext.cs

@@ -35,8 +35,9 @@ public void SetupSingleInstance(bool disableSlotVerification = false)
             {
                 QuietMode = true,
                 EnableCluster = !disableSlotVerification,
-                EndPoint = new IPEndPoint(IPAddress.Loopback, port),
+                EndPoints = [new IPEndPoint(IPAddress.Loopback, port)],
                 CleanClusterConfig = true,
+                ClusterAnnounceEndpoint = new IPEndPoint(IPAddress.Loopback, port)
             };
             if (RuntimeInformation.IsOSPlatform(OSPlatform.Linux))
                 opt.CheckpointDir = "/tmp";

benchmark/BDN.benchmark/Embedded/EmbeddedRespServer.cs

@@ -21,7 +21,7 @@ internal sealed class EmbeddedRespServer : GarnetServer
         /// <param name="opts">Server options to configure the base GarnetServer instance</param>
         /// <param name="loggerFactory">Logger factory to configure the base GarnetServer instance</param>
         /// <param name="server">Server network</param>
-        public EmbeddedRespServer(GarnetServerOptions opts, ILoggerFactory loggerFactory = null, GarnetServerEmbedded server = null) : base(opts, loggerFactory, server)
+        public EmbeddedRespServer(GarnetServerOptions opts, ILoggerFactory loggerFactory = null, GarnetServerEmbedded server = null) : base(opts, loggerFactory, server == null ? null : [server])
         {
             this.garnetServerEmbedded = server;
             this.subscribeBroker = opts.DisablePubSub ? null :

libs/cluster/Server/ClusterManager.cs

@@ -50,8 +50,18 @@ public unsafe ClusterManager(ClusterProvider clusterProvider, ILogger logger = n
             clusterConfigDevice = deviceFactory.Get(new FileDescriptor(directoryName: "", fileName: "nodes.conf"));
             pool = new(1, (int)clusterConfigDevice.SectorSize);
 
-            if (opts.EndPoint is not IPEndPoint endpoint)
-                throw new NotImplementedException("Cluster mode for unix domain sockets has not been implemented.");
+            IPEndPoint endpoint = null;
+            foreach (var endPoint in opts.EndPoints)
+            {
+                if (endPoint is IPEndPoint _endpoint)
+                {
+                    endpoint = _endpoint;
+                    break;
+                }
+            }
+
+            if (endpoint == null)
+                throw new GarnetException("No valid IPEndPoint found in endPoint list");
 
             var address = clusterProvider.storeWrapper.GetIp();
 

libs/cluster/Server/ClusterProvider.cs

@@ -439,23 +439,25 @@ internal ReplicationLogCheckpointManager GetReplicationLogCheckpointManager(Stor
         /// <returns></returns>
         internal bool BumpAndWaitForEpochTransition()
         {
-            var server = storeWrapper.TcpServer;
             BumpCurrentEpoch();
-            while (true)
+            foreach (var server in storeWrapper.TcpServer)
             {
-            retry:
-                Thread.Yield();
-                // Acquire latest bumped epoch
-                var currentEpoch = GarnetCurrentEpoch;
-                var sessions = server.ActiveClusterSessions();
-                foreach (var s in sessions)
+                while (true)
                 {
-                    var entryEpoch = s.LocalCurrentEpoch;
-                    // Retry if at least one session has not yet caught up to the current epoch.
-                    if (entryEpoch != 0 && entryEpoch < currentEpoch)
-                        goto retry;
+                retry:
+                    Thread.Yield();
+                    // Acquire latest bumped epoch
+                    var currentEpoch = GarnetCurrentEpoch;
+                    var sessions = ((GarnetServerTcp)server).ActiveClusterSessions();
+                    foreach (var s in sessions)
+                    {
+                        var entryEpoch = s.LocalCurrentEpoch;
+                        // Retry if at least one session has not yet caught up to the current epoch.
+                        if (entryEpoch != 0 && entryEpoch < currentEpoch)
+                            goto retry;
+                    }
+                    break;
                 }
-                break;
             }
             return true;
         }

libs/cluster/Server/Gossip.cs

@@ -161,12 +161,12 @@ public async Task TryMeetAsync(string address, int port, bool acquireLock = true
 
                 if (gsn == null)
                 {
-                    var endpoint = await Format.TryCreateEndpoint(address, port, useForBind: true, logger: logger);
-                    if (endpoint == null)
+                    var endpoints = await Format.TryCreateEndpoint(address, port, tryConnect: true, logger: logger);
+                    if (endpoints == null)
                     {
-                        logger?.LogError("Could not parse endpoint {address} {port}", address, port);
+                        logger?.LogError("Invalid CLUSTER MEET endpoint!");
                     }
-                    gsn = new GarnetServerNode(clusterProvider, endpoint, tlsOptions?.TlsClientOptions, logger: logger);
+                    gsn = new GarnetServerNode(clusterProvider, endpoints[0], tlsOptions?.TlsClientOptions, logger: logger);
                     created = true;
                 }
 

libs/cluster/Session/RespClusterBasicCommands.cs

@@ -168,7 +168,7 @@ private bool NetworkClusterMeet(out bool invalidParameters)
                 return true;
             }
 
-            logger?.LogTrace("CLUSTER MEET {ipaddressStr} {port}", ipAddress, port);
+            logger?.LogTrace("CLUSTER MEET");
             clusterProvider.clusterManager.RunMeetTask(ipAddress, port);
             while (!RespWriteUtils.TryWriteDirect(CmdStrings.RESP_OK, ref dcurr, dend))
                 SendAndReset();

libs/common/Format.cs

@@ -2,7 +2,9 @@
 // Licensed under the MIT license.
 
 using System;
+using System.Collections.Generic;
 using System.Diagnostics.CodeAnalysis;
+using System.Linq;
 using System.Net;
 using System.Net.Sockets;
 using System.Threading.Tasks;
@@ -31,116 +33,109 @@ internal static bool IsNullOrWhiteSpace([NotNullWhen(false)] this string s) =>
 #pragma warning disable format
     public static class Format
     {
+        /// <summary>
+        /// Parse address list string containing address separated by whitespace
+        /// </summary>
+        /// <param name="addressList">Space separated string of IP addresses</param>
+        /// <param name="port">Endpoint Port</param>
+        /// <param name="endpoints">List of endpoints generated from the input IPs</param>
+        /// <param name="errorHostnameOrAddress">Output error if any</param>
+        /// <param name="logger">Logger</param>
+        /// <returns>True if parse and address validation was successful, otherwise false</returns>
+        public static bool TryParseAddressList(string addressList, int port, out EndPoint[] endpoints, out string errorHostnameOrAddress, ILogger logger = null)
+        {
+            endpoints = null;
+            errorHostnameOrAddress = null;
+            // Check if input null or empty
+            if (string.IsNullOrEmpty(addressList) || string.IsNullOrWhiteSpace(addressList))
+            {
+                endpoints = [new IPEndPoint(IPAddress.Any, port)];
+                return true;
+            }
+
+            var addresses = addressList.Split([',',' '], StringSplitOptions.TrimEntries | StringSplitOptions.RemoveEmptyEntries);
+            var endpointList = new List<EndPoint>();
+            // Validate addresses and create endpoints
+            foreach (var singleAddressOrHostname in addresses)
+            {
+                var e = TryCreateEndpoint(singleAddressOrHostname, port, tryConnect: false, logger).Result;
+                if(e == null)
+                {
+                    endpoints = null;
+                    errorHostnameOrAddress = singleAddressOrHostname;
+                    return false;
+                }
+                endpointList.AddRange(e);
+            }
+            endpoints = [.. endpointList];
+
+            return true;
+        }
+
         /// <summary>
         /// Try to create an endpoint from address and port
         /// </summary>
-        /// <param name="addressOrHostname">This could be an address or a hostname that the method tries to resolve</param>
-        /// <param name="port"></param>
-        /// <param name="useForBind">Binding does not poll connection because is supposed to be called from the server side</param>
-        /// <param name="logger"></param>
+        /// <param name="singleAddressOrHostname">This could be an address or a hostname that the method tries to resolve</param>
+        /// <param name="port">Port number to use for the endpoints</param>
+        /// <param name="tryConnect">Whether to try to connect to the created endpoints to ensure that it is reachable</param>
+        /// <param name="logger">Logger</param>
         /// <returns></returns>
-        public static async Task<EndPoint> TryCreateEndpoint(string addressOrHostname, int port, bool useForBind = false, ILogger logger = null)
+        public static async Task<EndPoint[]> TryCreateEndpoint(string singleAddressOrHostname, int port, bool tryConnect = false, ILogger logger = null)
         {
-            IPEndPoint endpoint = null;
-            if (string.IsNullOrEmpty(addressOrHostname) || string.IsNullOrWhiteSpace(addressOrHostname))
-                return new IPEndPoint(IPAddress.Any, port);
+            if (string.IsNullOrEmpty(singleAddressOrHostname) || string.IsNullOrWhiteSpace(singleAddressOrHostname))
+                return [new IPEndPoint(IPAddress.Any, port)];
+
+            if (singleAddressOrHostname[0] == '-')
+                singleAddressOrHostname = singleAddressOrHostname.Substring(1);
 
-            if (IPAddress.TryParse(addressOrHostname, out var ipAddress))
-                return new IPEndPoint(ipAddress, port);
+            if (singleAddressOrHostname.Equals("localhost", StringComparison.CurrentCultureIgnoreCase))
+                return [new IPEndPoint(IPAddress.Loopback, port)];
+
+            if (IPAddress.TryParse(singleAddressOrHostname, out var ipAddress))
+                return [new IPEndPoint(ipAddress, port)];
 
             // Sanity check, there should be at least one ip address available
             try
             {
-                var ipAddresses = Dns.GetHostAddresses(addressOrHostname);
+                var ipAddresses = Dns.GetHostAddresses(singleAddressOrHostname);
                 if (ipAddresses.Length == 0)
                 {
-                    logger?.LogError("No IP address found for hostname:{hostname}", addressOrHostname);
+                    logger?.LogError("No IP address found for hostname:{hostname}", singleAddressOrHostname);
                     return null;
                 }
 
-                if (useForBind)
+                if (tryConnect)
                 {
                     foreach (var entry in ipAddresses)
                     {
-                        endpoint = new IPEndPoint(entry, port);
-                        var IsListening = await IsReachable(endpoint);
-                        if (IsListening) break;
+                        var endpoint = new IPEndPoint(entry, port);
+                        var IsListening = await TryConnect(endpoint);
+                        if (IsListening) return [endpoint];
                     }
                 }
                 else
                 {
                     var machineHostname = GetHostName();
 
-                    // Hostname does match the one acquired from machine name
-                    if (!addressOrHostname.Equals(machineHostname, StringComparison.OrdinalIgnoreCase))
+                    // User-provided hostname does not match the machine hostname
+                    if (!singleAddressOrHostname.Equals(machineHostname, StringComparison.OrdinalIgnoreCase))
                     {
-                        logger?.LogError("Provided hostname does not much acquired machine name {addressOrHostname} {machineHostname}!", addressOrHostname, machineHostname);
-                        return null;
-                    }
-
-                    if (ipAddresses.Length > 1) {
-                        logger?.LogError("Error hostname resolved to multiple endpoints. Garnet does not support multiple endpoints!");
+                        logger?.LogError("Provided hostname does not much acquired machine name {addressOrHostname} {machineHostname}!", singleAddressOrHostname, machineHostname);
                         return null;
                     }
 
-                    return new IPEndPoint(ipAddresses[0], port);
+                    return ipAddresses.Select(ip => new IPEndPoint(ip, port)).ToArray();
                 }
-                logger?.LogError("No reachable IP address found for hostname:{hostname}", addressOrHostname);
+                logger?.LogError("No reachable IP address found for hostname:{hostname}", singleAddressOrHostname);
             }
             catch (Exception ex)
             {
-                logger?.LogError(ex, "Error while trying to resolve hostname:{hostname}", addressOrHostname);
-            }
-
-            return endpoint;
-
-            async Task<bool> IsReachable(IPEndPoint endpoint)
-            {
-                using (var tcpClient = new TcpClient())
-                {
-                    try
-                    {
-                        await tcpClient.ConnectAsync(endpoint.Address, endpoint.Port);
-                        logger?.LogTrace("Reachable {ip} {port}", endpoint.Address, endpoint.Port);
-                        return true;
-                    }
-                    catch
-                    {
-                        logger?.LogTrace("Unreachable {ip} {port}", endpoint.Address, endpoint.Port);
-                        return false;
-                    }
-                }
+                logger?.LogError("Error while trying to resolve hostname: {exMessage} [{hostname}]", ex.Message, singleAddressOrHostname);
             }
-        }
 
-        /// <summary>
-        /// Try to
-        /// </summary>
-        /// <param name="address"></param>
-        /// <param name="port"></param>
-        /// <param name="logger"></param>
-        /// <returns></returns>
-        public static async Task<IPEndPoint> TryValidateAndConnectAddress2(string address, int port, ILogger logger = null)
-        {
-            IPEndPoint endpoint = null;
-            if (!IPAddress.TryParse(address, out var ipAddress))
-            {
-                // Try to identify reachable IP address from hostname
-                var hostEntry = Dns.GetHostEntry(address);
-                foreach (var entry in hostEntry.AddressList)
-                {
-                    endpoint = new IPEndPoint(entry, port);
-                    var IsListening = await IsReachable(endpoint);
-                    if (IsListening) break;
-                }
-            }
-            else
-            {
-                // If address is valid create endpoint
-                endpoint = new IPEndPoint(ipAddress, port);
-            }
+            return null;
 
-            async Task<bool> IsReachable(IPEndPoint endpoint)
+            async Task<bool> TryConnect(IPEndPoint endpoint)
             {
                 using (var tcpClient = new TcpClient())
                 {
@@ -157,8 +152,6 @@ async Task<bool> IsReachable(IPEndPoint endpoint)
                     }
                 }
             }
-
-            return endpoint;
         }
 
         /// <summary>

libs/host/Configuration/Options.cs

@@ -40,9 +40,17 @@ internal sealed class Options
         public int Port { get; set; }
 
         [IpAddressValidation(false)]
-        [Option("bind", Required = false, HelpText = "IP address to bind server to (default: any)")]
+        [Option("bind", Required = false, HelpText = "Whitespace or comma separated string of IP addresses to bind server to (default: any)")]
         public string Address { get; set; }
 
+        [IntRangeValidation(0, 65535)]
+        [Option("cluster-announce-port", Required = false, HelpText = "Port that this node advertises to other nodes to connect to for gossiping.")]
+        public int ClusterAnnouncePort { get; set; }
+
+        [IpAddressValidation(false)]
+        [Option("cluster-announce-ip", Required = false, HelpText = "IP address that this node advertises to other nodes to connect to for gossiping.")]
+        public string ClusterAnnounceIp { get; set; }
+
         [MemorySizeValidation]
         [Option('m', "memory", Required = false, HelpText = "Total log memory used in bytes (rounds down to power of 2)")]
         public string MemorySize { get; set; }
@@ -617,9 +625,6 @@ public bool IsValid(out List<string> invalidOptions, ILogger logger = null)
             this.runtimeLogger = logger;
             foreach (var prop in typeof(Options).GetProperties())
             {
-                if (prop.Name.Equals("runtimeLogger"))
-                    continue;
-
                 // Ignore if property is not decorated with the OptionsAttribute or the ValidationAttribute
                 var validationAttr = prop.GetCustomAttributes(typeof(ValidationAttribute)).FirstOrDefault();
                 if (!Attribute.IsDefined(prop, typeof(OptionAttribute)) || validationAttr == null)
@@ -659,18 +664,21 @@ public GarnetServerOptions GetServerOptions(ILogger logger = null)
             var checkpointDir = CheckpointDir;
             if (!useAzureStorage) checkpointDir = new DirectoryInfo(string.IsNullOrEmpty(checkpointDir) ? (string.IsNullOrEmpty(logDir) ? "." : logDir) : checkpointDir).FullName;
 
-            EndPoint endpoint;
-            if (!string.IsNullOrEmpty(UnixSocketPath))
-            {
-                endpoint = new UnixDomainSocketEndPoint(UnixSocketPath);
-            }
-            else
+            if (!Format.TryParseAddressList(Address, Port, out var endpoints, out _) || endpoints.Length == 0)
+                throw new GarnetException($"Invalid endpoint format {Address} {Port}.");
+
+            EndPoint[] clusterAnnounceEndpoint = null;
+            if (ClusterAnnounceIp != null)
             {
-                endpoint = Format.TryCreateEndpoint(Address, Port, useForBind: false).Result;
-                if (endpoint == null)
-                    throw new GarnetException($"Invalid endpoint format {Address} {Port}.");
+                ClusterAnnouncePort = ClusterAnnouncePort == 0 ? Port : ClusterAnnouncePort;
+                clusterAnnounceEndpoint = Format.TryCreateEndpoint(ClusterAnnounceIp, ClusterAnnouncePort, tryConnect: false, logger: logger).GetAwaiter().GetResult();
+                if (clusterAnnounceEndpoint == null || !endpoints.Any(endpoint => endpoint.Equals(clusterAnnounceEndpoint[0])))
+                    throw new GarnetException("Cluster announce endpoint does not match list of listen endpoints provided!");
             }
 
+            if (!string.IsNullOrEmpty(UnixSocketPath))
+                endpoints = [.. endpoints, new UnixDomainSocketEndPoint(UnixSocketPath)];
+
             // Unix file permission octal to UnixFileMode
             var unixSocketPermissions = (UnixFileMode)Convert.ToInt32(UnixSocketPermission.ToString(), 8);
 
@@ -723,7 +731,8 @@ public GarnetServerOptions GetServerOptions(ILogger logger = null)
             }
             return new GarnetServerOptions(logger)
             {
-                EndPoint = endpoint,
+                EndPoints = endpoints,
+                ClusterAnnounceEndpoint = clusterAnnounceEndpoint?[0],
                 MemorySize = MemorySize,
                 PageSize = PageSize,
                 SegmentSize = SegmentSize,