WorkerList.cs

Go to the documentation of this file.

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

namespace kinetica.Utils;

    public sealed class WorkerList : IReadOnlyList<Uri?>
    {
        private readonly List<Uri?> _urls;

        public bool IsMultiHeadEnabled { get; private set; }

        public Regex? IpRegex { get; private set; }

        public bool IsQueriedUrlList { get; private set; }

        public int Count => _urls.Count;

        public Uri? this[int index] => index >= 0 && index < _urls.Count ? _urls[index] : null;

        public WorkerList()
        {
            _urls = new List<Uri?>();
            IsMultiHeadEnabled = false;
            IpRegex = null;
            IsQueriedUrlList = false;
        }

        public WorkerList(Kinetica db) : this(db, (Regex?)null)
        {
        }

        public WorkerList(Kinetica db, Regex? ip_regex)
        {
            _urls = new List<Uri?>();
            IpRegex = ip_regex;
            IsQueriedUrlList = true;
            IsMultiHeadEnabled = false;

            if (db == null)
                throw new ArgumentNullException(nameof(db));

            // Check if auto-discovery is enabled on the connection
            // If HAManager exists and has auto-discovery disabled, return empty list
            if (db.HAManager?.DisableAutoDiscovery == true)
            {
                return;
            }

            // Get the system properties from the database server
            IDictionary<string, string> system_properties = db.showSystemProperties().property_map;

            // Find out if multi-head ingest is turned on or not
            if (!system_properties.TryGetValue(ShowSystemPropertiesResponse.PropertyMap.CONF_ENABLE_WORKER_HTTP_SERVERS, out string? multi_head_ingestion_param))
                throw new KineticaException("Missing value for " + ShowSystemPropertiesResponse.PropertyMap.CONF_ENABLE_WORKER_HTTP_SERVERS);

            bool is_multi_head_ingest_enabled = multi_head_ingestion_param?.Equals(ShowSystemPropertiesResponse.PropertyMap.TRUE, StringComparison.OrdinalIgnoreCase) ?? false;

            // Nothing to do if multi-head ingestion is disabled
            if (!is_multi_head_ingest_enabled)
            {
                return;
            }

            IsMultiHeadEnabled = true;

            // Multi-head ingestion IS enabled; find the worker URLs
            if (system_properties.TryGetValue("conf.worker_http_server_urls", out string? worker_urls_str) && !string.IsNullOrEmpty(worker_urls_str))
            {
                ParseWorkerUrls(worker_urls_str);
            }
            else
            {
                // Fall back to IPs and ports
                ParseWorkerIpsAndPorts(system_properties, db);
            }

            // Check that this list is not empty (but only if no removed ranks exist)
            if (_urls.Count == 0)
                throw new KineticaException("No worker HTTP servers found.");
        }

        public static WorkerList FromUrls(IEnumerable<Uri> urls)
        {
            var workerList = new WorkerList();
            foreach (var url in urls)
            {
                workerList._urls.Add(url);
            }
            return workerList;
        }

        public static WorkerList WithIpPrefix(Kinetica db, string? ip_prefix)
        {
            if (string.IsNullOrEmpty(ip_prefix))
            {
                return new WorkerList(db);
            }
            var regex = new Regex("^" + Regex.Escape(ip_prefix) + ".*");
            return new WorkerList(db, regex);
        }

        private void ParseWorkerUrls(string worker_urls_str)
        {
            string[] worker_url_lists = worker_urls_str.Split(';');

            // Skip rank 0 (index 0), start from rank 1
            for (int i = 1; i < worker_url_lists.Length; ++i)
            {
                string url_list = worker_url_lists[i];

                // Handle removed ranks (empty string)
                if (string.IsNullOrEmpty(url_list))
                {
                    _urls.Add(null);
                    continue;
                }

                // Need to split each of the URL lists on a comma
                string[] urls = url_list.Split(',');

                bool matching_url_found = false;

                // Find at least one URL to work with
                foreach (string url_str in urls)
                {
                    try
                    {
                        // If a regular expression is given, then see if this one is a match
                        if (IpRegex != null)
                        {
                            // Extract host from URL for matching
                            var tempUri = new Uri(url_str);
                            matching_url_found = IpRegex.IsMatch(tempUri.Host);
                        }
                        else
                        {
                            // No regex given, so take the first URL encountered for this worker
                            matching_url_found = true;
                        }

                        if (matching_url_found)
                        {
                            Uri url = new Uri(url_str);
                            _urls.Add(url);
                            break;
                        }
                    }
                    catch (Exception ex)
                    {
                        throw new KineticaException($"Invalid URL '{url_str}': {ex.Message}");
                    }
                }

                if (!matching_url_found)
                    throw new KineticaException($"No matching URL found for worker #{i}.");
            }
        }

        private void ParseWorkerIpsAndPorts(IDictionary<string, string> system_properties, Kinetica db)
        {
            if (!system_properties.TryGetValue(ShowSystemPropertiesResponse.PropertyMap.CONF_WORKER_HTTP_SERVER_IPS, out string? worker_ips_str) || string.IsNullOrEmpty(worker_ips_str))
                throw new KineticaException("Missing value for " + ShowSystemPropertiesResponse.PropertyMap.CONF_WORKER_HTTP_SERVER_IPS);

            if (!system_properties.TryGetValue(ShowSystemPropertiesResponse.PropertyMap.CONF_WORKER_HTTP_SERVER_PORTS, out string? worker_ports_str) || string.IsNullOrEmpty(worker_ports_str))
                throw new KineticaException("Missing value for " + ShowSystemPropertiesResponse.PropertyMap.CONF_WORKER_HTTP_SERVER_PORTS);

            // Split the strings
            string[] worker_ip_lists = worker_ips_str.Split(';');
            string[] worker_ports = worker_ports_str.Split(';');

            // Check that there are the same number of IPs and ports supplied
            if (worker_ip_lists.Length != worker_ports.Length)
                throw new KineticaException("Inconsistent number of values for "
                                             + ShowSystemPropertiesResponse.PropertyMap.CONF_WORKER_HTTP_SERVER_IPS
                                             + " and "
                                             + ShowSystemPropertiesResponse.PropertyMap.CONF_WORKER_HTTP_SERVER_PORTS);

            // Get the protocol from the database URL
            string scheme = db.URL.Scheme;

            // Skip rank 0 (index 0), start from rank 1
            for (int i = 1; i < worker_ip_lists.Length; ++i)
            {
                string ip_list = worker_ip_lists[i];

                // Handle removed ranks (empty string)
                if (string.IsNullOrEmpty(ip_list))
                {
                    _urls.Add(null);
                    continue;
                }

                // Need to split each of the IP lists on a comma
                string[] ips = ip_list.Split(',');

                bool matching_ip_found = false;

                // Find at least one IP to work with
                foreach (string ip in ips)
                {
                    try
                    {
                        // If a regular expression is given, then see if this one is a match
                        if (IpRegex != null)
                            matching_ip_found = IpRegex.IsMatch(ip);
                        else
                            matching_ip_found = true;

                        if (matching_ip_found)
                        {
                            UriBuilder uri_builder = new UriBuilder(scheme, ip, int.Parse(worker_ports[i]));
                            Uri url = uri_builder.Uri;
                            _urls.Add(url);
                            break;
                        }
                    }
                    catch (Exception ex)
                    {
                        throw new KineticaException($"Error creating URL for worker #{i}: {ex.Message}");
                    }
                }

                if (!matching_ip_found)
                    throw new KineticaException($"No matching IP found for worker #{i}.");
            }
        }

        // ==================== Mutation Methods ====================

        public void Add(Uri url)
        {
            _urls.Add(url);
        }

        public void AddOrNull(Uri? url)
        {
            _urls.Add(url);
        }

        public void Clear()
        {
            _urls.Clear();
        }

        // ==================== Query Methods ====================

        public bool IsEmpty => _urls.Count == 0;

        public IEnumerable<Uri> GetActiveUrls()
        {
            return _urls.Where(u => u != null).Cast<Uri>();
        }

        public int ActiveCount => _urls.Count(u => u != null);

        public List<string> ToUrlStrings()
        {
            return _urls
                .Where(u => u != null)
                .Select(u => u!.ToString())
                .ToList();
        }

        public Uri? Get(int index)
        {
            return index >= 0 && index < _urls.Count ? _urls[index] : null;
        }

        // ==================== IReadOnlyList Implementation ====================

        public IEnumerator<Uri?> GetEnumerator()
        {
            return _urls.GetEnumerator();
        }

        IEnumerator IEnumerable.GetEnumerator()
        {
            return GetEnumerator();
        }

        // ==================== Backward Compatibility ====================

        public IEnumerable<Uri> GetActiveUrlsEnumerator()
        {
            return GetActiveUrls();
        }
    }