HAFailover.cs

Go to the documentation of this file.

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text.RegularExpressions;
using System.Threading;

namespace kinetica;

    public enum HAFailoverOrder
    {
        Random,

        Sequential
    }

    public class ClusterAddressInfo
    {
        public Uri ActiveHeadNodeUrl { get; set; }

        public IDictionary<string, string> SystemProperties { get; set; }

        public IList<Uri> WorkerRankUrls { get; set; }

        public HashSet<string> HostNames { get; set; }

        public Uri HostManagerUrl { get; set; }

        public bool IsPrimaryCluster { get; set; }

        public ClusterAddressInfo(
            Uri activeHeadNodeUrl,
            IDictionary<string, string> systemProperties,
            IList<Uri> workerRankUrls,
            HashSet<string> hostNames,
            Uri hostManagerUrl)
        {
            ActiveHeadNodeUrl = activeHeadNodeUrl;
            SystemProperties = systemProperties ?? new Dictionary<string, string>();
            WorkerRankUrls = workerRankUrls ?? new List<Uri>();
            HostNames = hostNames ?? new HashSet<string>();
            HostManagerUrl = hostManagerUrl;
            IsPrimaryCluster = false;

            // Ensure that all the known ranks' hostnames are also accounted for
            UpdateHostnamesBasedOnRankUrls();
        }

        public ClusterAddressInfo(Uri activeHeadNodeUrl, int hostManagerPort)
        {
            ActiveHeadNodeUrl = activeHeadNodeUrl;
            SystemProperties = new Dictionary<string, string>();
            WorkerRankUrls = new List<Uri>();
            HostNames = new HashSet<string>();
            IsPrimaryCluster = false;

            // Create host manager URL
            var builder = new UriBuilder(activeHeadNodeUrl)
            {
                Port = hostManagerPort,
                Path = string.Empty
            };
            HostManagerUrl = builder.Uri;

            // Ensure that all the known ranks' hostnames are also accounted for
            UpdateHostnamesBasedOnRankUrls();
        }

        private void UpdateHostnamesBasedOnRankUrls()
        {
            // Put the head rank's hostname in the saved hostnames
            if (ActiveHeadNodeUrl != null && !string.IsNullOrEmpty(ActiveHeadNodeUrl.Host))
            {
                if (!DoesClusterContainNode(ActiveHeadNodeUrl.Host))
                {
                    HostNames.Add(ActiveHeadNodeUrl.Host);
                }
            }

            // Put each worker rank's hostname in the saved hostnames
            if (WorkerRankUrls != null)
            {
                foreach (var workerUrl in WorkerRankUrls)
                {
                    if (workerUrl != null && !string.IsNullOrEmpty(workerUrl.Host))
                    {
                        if (!DoesClusterContainNode(workerUrl.Host))
                        {
                            HostNames.Add(workerUrl.Host);
                        }
                    }
                }
            }
        }

        public bool DoesClusterContainNode(string hostName)
        {
            return HostNames.Contains(hostName);
        }

        public override string ToString()
        {
            var workerUrls = string.Join(", ", WorkerRankUrls.Select(u => u.ToString()));
            var hostnames = string.Join(", ", HostNames);
            return $"{{ activeHeadNodeUrl: {ActiveHeadNodeUrl}, workerRankUrls: [{workerUrls}], hostNames: [{hostnames}], hostManagerUrl: {HostManagerUrl}, isPrimaryCluster: {IsPrimaryCluster} }}";
        }

        public override bool Equals(object? obj)
        {
            if (obj is ClusterAddressInfo other)
            {
                return ActiveHeadNodeUrl?.ToString() == other.ActiveHeadNodeUrl?.ToString();
            }
            return false;
        }

        public override int GetHashCode()
        {
            return ActiveHeadNodeUrl?.ToString().GetHashCode() ?? 0;
        }
    }

    public class HAFailoverManager
    {
        // Default port for host manager URLs
        public const int DefaultHostManagerPort = 9300;

        // System properties response keys
        private const string SYSTEM_PROP_ENABLE_HTTPD = "conf.enable_httpd_proxy";
        private const string SYSTEM_PROP_ENABLE_MH = "conf.enable_worker_http_servers";
        private const string SYSTEM_PROP_NUM_HOSTS = "conf.number_of_hosts";
        private const string SYSTEM_PROP_HEAD_NODE_URLS = "conf.ha_ring_head_nodes_full";
        private const string SYSTEM_PROP_SERVER_URLS = "conf.worker_http_server_urls";
        private const string SYSTEM_PROP_ENABLE_HA = "conf.enable_ha";
        private const string SYSTEM_PROP_TRUE = "TRUE";

        private readonly List<ClusterAddressInfo> _hostAddresses;
        private readonly List<int> _haUrlIndices;
        private string _primaryUrlHostname;
        private int _currentClusterIndexPointer;
        private int _numClusterSwitches;
        private readonly object _lock = new object();

        public bool DisableFailover { get; set; }

        public bool DisableAutoDiscovery { get; set; }

        public int HostManagerPort { get; set; }

        public HAFailoverOrder FailoverOrder { get; set; }

        public Regex? HostnameRegex { get; set; }

        public HAFailoverManager()
        {
            _hostAddresses = new List<ClusterAddressInfo>();
            _haUrlIndices = new List<int>();
            _primaryUrlHostname = string.Empty;
            _currentClusterIndexPointer = 0;
            _numClusterSwitches = 0;
            DisableFailover = false;
            DisableAutoDiscovery = false;
            HostManagerPort = DefaultHostManagerPort;
            FailoverOrder = HAFailoverOrder.Random;
        }

        public int HARingSize
        {
            get
            {
                lock (_lock)
                {
                    return _hostAddresses.Count;
                }
            }
        }

        public int NumClusterSwitches
        {
            get
            {
                lock (_lock)
                {
                    return _numClusterSwitches;
                }
            }
        }

        public IList<ClusterAddressInfo> GetHostAddresses()
        {
            lock (_lock)
            {
                return new List<ClusterAddressInfo>(_hostAddresses);
            }
        }

        public IList<Uri> GetUrls()
        {
            lock (_lock)
            {
                return _hostAddresses.Select(h => h.ActiveHeadNodeUrl).ToList();
            }
        }

        public IList<Uri> GetFailoverUrls()
        {
            lock (_lock)
            {
                return _haUrlIndices
                    .Where(i => i < _hostAddresses.Count)
                    .Select(i => _hostAddresses[i].ActiveHeadNodeUrl)
                    .ToList();
            }
        }

        public ClusterAddressInfo? GetClusterInfo()
        {
            lock (_lock)
            {
                if (_hostAddresses.Count == 0)
                    return null;

                if (_hostAddresses.Count == 1)
                    return _hostAddresses[0];

                if (_currentClusterIndexPointer < _haUrlIndices.Count)
                {
                    int index = _haUrlIndices[_currentClusterIndexPointer];
                    if (index < _hostAddresses.Count)
                        return _hostAddresses[index];
                }

                return null;
            }
        }

        public Uri? GetUrl()
        {
            return GetClusterInfo()?.ActiveHeadNodeUrl;
        }

        public void AddCluster(ClusterAddressInfo clusterInfo)
        {
            lock (_lock)
            {
                _hostAddresses.Add(clusterInfo);
            }
        }

        public void Initialize(IList<Uri> urls, Kinetica? kinetica = null)
        {
            lock (_lock)
            {
                _hostAddresses.Clear();
                _haUrlIndices.Clear();
                _currentClusterIndexPointer = 0;
                _numClusterSwitches = 0;

                var urlQueue = new Queue<Uri>(urls);
                int numUserGivenUrls = urlQueue.Count;
                int numProcessedUrls = 0;
                var clusterIndicesOfUserGivenUrls = new List<int>();

                while (urlQueue.Count > 0)
                {
                    var url = urlQueue.Dequeue();
                    bool isUserGivenUrl = numProcessedUrls < numUserGivenUrls;
                    numProcessedUrls++;

                    // Check if this hostname is already in a known cluster
                    int? existingIndex = GetIndexOfClusterContainingNode(url.Host);
                    if (existingIndex.HasValue)
                    {
                        if (isUserGivenUrl)
                            clusterIndicesOfUserGivenUrls.Add(existingIndex.Value);
                        continue;
                    }

                    // Create minimal cluster info
                    ClusterAddressInfo? clusterInfo = null;

                    if (!DisableAutoDiscovery && kinetica != null)
                    {
                        try
                        {
                            // Try to get system properties from the server
                            var systemProps = kinetica.showSystemProperties().property_map;
                            clusterInfo = CreateClusterAddressInfo(url, systemProps);

                            // Get HA ring URLs and add them to the queue
                            var haRingUrls = GetHARingHeadNodeUrls(systemProps);
                            foreach (var haUrl in haRingUrls)
                            {
                                if (!GetIndexOfClusterContainingNode(haUrl.Host).HasValue &&
                                    !urlQueue.Contains(haUrl))
                                {
                                    urlQueue.Enqueue(haUrl);
                                }
                            }
                        }
                        catch
                        {
                            // Failed to get system properties, use minimal info
                            clusterInfo = new ClusterAddressInfo(url, HostManagerPort);
                        }
                    }
                    else
                    {
                        clusterInfo = new ClusterAddressInfo(url, HostManagerPort);
                    }

                    if (isUserGivenUrl)
                        clusterIndicesOfUserGivenUrls.Add(_hostAddresses.Count);

                    _hostAddresses.Add(clusterInfo);
                }

                if (_hostAddresses.Count == 0)
                    throw new KineticaException("Could not connect to any working Kinetica server");

                // Set the primary cluster
                if (_hostAddresses.Count == 1)
                {
                    _hostAddresses[0].IsPrimaryCluster = true;
                    _primaryUrlHostname = _hostAddresses[0].ActiveHeadNodeUrl.Host;
                }
                else
                {
                    // If all user-given URLs belong to the same cluster, make that the primary
                    if (string.IsNullOrEmpty(_primaryUrlHostname))
                    {
                        var uniqueIndices = new HashSet<int>(clusterIndicesOfUserGivenUrls);
                        if (uniqueIndices.Count == 1 && clusterIndicesOfUserGivenUrls.Count > 0)
                        {
                            int primaryIndex = clusterIndicesOfUserGivenUrls[0];
                            _primaryUrlHostname = _hostAddresses[primaryIndex].ActiveHeadNodeUrl.Host;
                        }
                    }
                }

                // Flag the primary cluster and move it to the front
                if (!string.IsNullOrEmpty(_primaryUrlHostname))
                {
                    int? primaryIndex = GetIndexOfClusterContainingNode(_primaryUrlHostname);
                    if (primaryIndex.HasValue)
                    {
                        _hostAddresses[primaryIndex.Value].IsPrimaryCluster = true;
                        if (primaryIndex.Value > 0)
                        {
                            var primary = _hostAddresses[primaryIndex.Value];
                            _hostAddresses.RemoveAt(primaryIndex.Value);
                            _hostAddresses.Insert(0, primary);
                        }
                    }
                }

                // Randomize URLs for failover
                RandomizeUrls();
            }
        }

        private int? GetIndexOfClusterContainingNode(string hostname)
        {
            for (int i = 0; i < _hostAddresses.Count; i++)
            {
                if (_hostAddresses[i].DoesClusterContainNode(hostname))
                    return i;
            }
            return null;
        }

        private ClusterAddressInfo CreateClusterAddressInfo(Uri url, IDictionary<string, string> systemProperties)
        {
            bool isHttpdEnabled = GetSystemPropertyBool(systemProperties, SYSTEM_PROP_ENABLE_HTTPD);
            bool isMultiHeadEnabled = GetSystemPropertyBool(systemProperties, SYSTEM_PROP_ENABLE_MH);

            // Get rank URLs
            var rankUrls = new List<Uri>();
            if (isMultiHeadEnabled && systemProperties.TryGetValue(SYSTEM_PROP_SERVER_URLS, out var serverUrlsStr))
            {
                rankUrls = ParseRankUrls(serverUrlsStr);
            }

            // Get head node URL (first rank URL, or fall back to given URL)
            Uri activeHeadNodeUrl;
            if (rankUrls.Count > 0)
            {
                activeHeadNodeUrl = rankUrls[0];
                rankUrls.RemoveAt(0);
            }
            else
            {
                activeHeadNodeUrl = url;
            }

            // Get hostnames
            var hostNames = GetHostnamesFromSystemProperties(systemProperties);

            // Create host manager URL
            Uri hostManagerUrl;
            if (isHttpdEnabled && !string.IsNullOrEmpty(activeHeadNodeUrl.AbsolutePath) && activeHeadNodeUrl.AbsolutePath != "/")
            {
                var builder = new UriBuilder(activeHeadNodeUrl)
                {
                    Path = "/gpudb-host-manager"
                };
                hostManagerUrl = builder.Uri;
            }
            else
            {
                var builder = new UriBuilder(activeHeadNodeUrl)
                {
                    Port = HostManagerPort,
                    Path = string.Empty
                };
                hostManagerUrl = builder.Uri;
            }

            return new ClusterAddressInfo(activeHeadNodeUrl, systemProperties, rankUrls, hostNames, hostManagerUrl);
        }

        private List<Uri> ParseRankUrls(string serverUrlsStr)
        {
            var result = new List<Uri>();
            var rankEntries = serverUrlsStr.Split(';');

            foreach (var entry in rankEntries)
            {
                if (string.IsNullOrEmpty(entry))
                    continue;

                var urls = entry.Split(',');
                foreach (var urlStr in urls)
                {
                    if (Uri.TryCreate(urlStr.Trim(), UriKind.Absolute, out var parsedUrl))
                    {
                        bool shouldAdd = HostnameRegex == null || HostnameRegex.IsMatch(parsedUrl.Host);
                        if (shouldAdd)
                        {
                            result.Add(parsedUrl);
                            break;
                        }
                    }
                }
            }

            return result;
        }

        private HashSet<string> GetHostnamesFromSystemProperties(IDictionary<string, string> systemProperties)
        {
            var hostnames = new HashSet<string>();

            if (!systemProperties.TryGetValue(SYSTEM_PROP_NUM_HOSTS, out var numHostsStr) ||
                !int.TryParse(numHostsStr, out int numHosts))
            {
                return hostnames;
            }

            for (int i = 0; i < numHosts; i++)
            {
                string key = $"conf.host{i}_public_urls";
                if (systemProperties.TryGetValue(key, out var hostnameStr))
                {
                    var hostUrls = hostnameStr.Split(',');
                    foreach (var hostname in hostUrls)
                    {
                        var host = hostname.Trim();
                        // Strip protocol if present
                        int idx = host.IndexOf("://");
                        if (idx >= 0)
                            host = host.Substring(idx + 3);

                        bool shouldAdd = HostnameRegex == null || HostnameRegex.IsMatch(host);
                        if (shouldAdd)
                        {
                            hostnames.Add(host);
                            break;
                        }
                    }
                }
            }

            return hostnames;
        }

        private List<Uri> GetHARingHeadNodeUrls(IDictionary<string, string> systemProperties)
        {
            var haUrls = new List<Uri>();

            if (!GetSystemPropertyBool(systemProperties, SYSTEM_PROP_ENABLE_HA))
                return haUrls;

            if (!systemProperties.TryGetValue(SYSTEM_PROP_HEAD_NODE_URLS, out var haRingStr) ||
                string.IsNullOrEmpty(haRingStr))
            {
                return haUrls;
            }

            var clusterEntries = haRingStr.Split(';');
            foreach (var entry in clusterEntries)
            {
                if (string.IsNullOrEmpty(entry))
                    continue;

                var urls = entry.Split(',');
                foreach (var urlStr in urls)
                {
                    if (Uri.TryCreate(urlStr.Trim(), UriKind.Absolute, out var parsedUrl))
                    {
                        bool shouldAdd = HostnameRegex == null || HostnameRegex.IsMatch(parsedUrl.Host);
                        if (shouldAdd)
                        {
                            haUrls.Add(parsedUrl);
                            break;
                        }
                    }
                }
            }

            return haUrls;
        }

        private static bool GetSystemPropertyBool(IDictionary<string, string> properties, string key)
        {
            return properties.TryGetValue(key, out var value) &&
                   value.Equals(SYSTEM_PROP_TRUE, StringComparison.OrdinalIgnoreCase);
        }

        private void RandomizeUrls()
        {
            _haUrlIndices.Clear();
            for (int i = 0; i < _hostAddresses.Count; i++)
            {
                _haUrlIndices.Add(i);
            }

            if (FailoverOrder == HAFailoverOrder.Random && _haUrlIndices.Count > 1)
            {
                // Keep primary (index 0) first, shuffle the rest
                var rng = new Random();
                int n = _haUrlIndices.Count;
                for (int i = n - 1; i > 1; i--)
                {
                    int j = rng.Next(1, i + 1);
                    (_haUrlIndices[i], _haUrlIndices[j]) = (_haUrlIndices[j], _haUrlIndices[i]);
                }
            }
        }

        private void SelectNextCluster()
        {
            _currentClusterIndexPointer = (_currentClusterIndexPointer + 1) % _hostAddresses.Count;
            Interlocked.Increment(ref _numClusterSwitches);
        }

        public Uri SwitchUrl(Uri oldUrl, int oldNumClusterSwitches, Func<Uri, bool>? isKineticaRunning = null)
        {
            lock (_lock)
            {
                if (DisableFailover)
                {
                    throw new KineticaException("Failover is disabled!");
                }

                // If there is only one URL, then we can't switch URLs
                if (_hostAddresses.Count == 1)
                {
                    throw new KineticaException("Only one cluster in ring; HA failover unavailable");
                }

                // Get how many more times other threads have switched clusters
                int countSwitchesSinceInvocation = _numClusterSwitches - oldNumClusterSwitches;

                // Check if another thread has tried all the clusters in the HA ring
                if (countSwitchesSinceInvocation >= _hostAddresses.Count)
                {
                    throw new KineticaException($"Fail-over attempted as many times as clusters in the ring; URLs attempted: {string.Join(", ", GetUrls())}");
                }

                // Check if another thread beat us to switching the URL
                var currentUrl = GetUrl();
                if (currentUrl != null && !currentUrl.Equals(oldUrl) && countSwitchesSinceInvocation > 0)
                {
                    return currentUrl;
                }

                // This thread is the first one here--select the next cluster to use
                while (true)
                {
                    SelectNextCluster();

                    currentUrl = GetUrl();
                    if (currentUrl == null)
                    {
                        throw new KineticaException("No current URL available");
                    }

                    if (currentUrl.Equals(oldUrl))
                    {
                        // Re-shuffle and throw exception
                        RandomizeUrls();
                        throw new KineticaException($"Circled back to original URL; no clusters available for fail-over among these: {string.Join(", ", GetUrls())}");
                    }

                    // Check if the new cluster is running
                    bool isRunning = isKineticaRunning?.Invoke(currentUrl) ?? true;
                    if (isRunning)
                    {
                        return currentUrl;
                    }
                }
            }
        }
    }