Refactor pvcnoded to reduce Daemon.py size

This branch commit refactors the pvcnoded component to better adhere to
good programming practices. The previous Daemon.py was a massive file
which contained almost 2000 lines of direct, root-level code which was
directly imported. Not only was this poor practice, but this resulted
in a nigh-unmaintainable file which was hard even for me to understand.

This refactoring splits a large section of the code from Daemon.py into
separate small modules and functions in the `util/` directory. This will
hopefully make most of the functionality easy to find and modify without
having to dig through a single large file.

Further the existing subcomponents have been moved to the `objects/`
directory which clearly separates them.

Finally, the Daemon.py code has mostly been moved into a function,
`entrypoint()`, which is then called from the `pvcnoded.py` stub.

An additional item is that most format strings have been replaced by
f-strings to make use of the Python 3.6 features in Daemon.py and the
utility files.

node-daemon/pvcnoded/util/config.py

@@ -0,0 +1,384 @@
+#!/usr/bin/env python3
+
+# config.py - Utility functions for pvcnoded configuration parsing
+# Part of the Parallel Virtual Cluster (PVC) system
+#
+#    Copyright (C) 2018-2021 Joshua M. Boniface <joshua@boniface.me>
+#
+#    This program is free software: you can redistribute it and/or modify
+#    it under the terms of the GNU General Public License as published by
+#    the Free Software Foundation, version 3.
+#
+#    This program is distributed in the hope that it will be useful,
+#    but WITHOUT ANY WARRANTY; without even the implied warranty of
+#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#    GNU General Public License for more details.
+#
+#    You should have received a copy of the GNU General Public License
+#    along with this program.  If not, see <https://www.gnu.org/licenses/>.
+#
+###############################################################################
+
+import os
+import subprocess
+import yaml
+from socket import gethostname
+from re import findall
+from psutil import cpu_count
+from ipaddress import ip_address, ip_network
+
+
+class MalformedConfigurationError(Exception):
+    """
+    An except when parsing the PVC Node daemon configuration file
+    """
+    def __init__(self, error=None):
+        self.msg = f'ERROR: Configuration file is malformed: {error}'
+
+    def __str__(self):
+        return str(self.msg)
+
+
+def get_static_data():
+    """
+    Data that is obtained once at node startup for use later
+    """
+    staticdata = list()
+    staticdata.append(str(cpu_count()))  # CPU count
+    staticdata.append(
+        subprocess.run(
+            ['uname', '-r'], stdout=subprocess.PIPE
+        ).stdout.decode('ascii').strip()
+    )
+    staticdata.append(
+        subprocess.run(
+            ['uname', '-o'], stdout=subprocess.PIPE
+        ).stdout.decode('ascii').strip()
+    )
+    staticdata.append(
+        subprocess.run(
+            ['uname', '-m'], stdout=subprocess.PIPE
+        ).stdout.decode('ascii').strip()
+    )
+
+    return staticdata
+
+
+def get_configuration_path():
+    try:
+        return os.environ['PVCD_CONFIG_FILE']
+    except KeyError:
+        print('ERROR: The "PVCD_CONFIG_FILE" environment variable must be set.')
+        os._exit(1)
+
+
+def get_hostname():
+    node_fqdn = gethostname()
+    node_hostname = node_fqdn.split('.', 1)[0]
+    node_domain = ''.join(node_fqdn.split('.', 1)[1:])
+    try:
+        node_id = findall(r'\d+', node_hostname)[-1]
+    except IndexError:
+        node_id = 0
+
+    return node_fqdn, node_hostname, node_domain, node_id
+
+
+def validate_floating_ip(config, network):
+    if network not in ['cluster', 'storage', 'upstream']:
+        return False, f'Specified network type "{network}" is not valid'
+
+    floating_key = f'{network}_floating_ip'
+    network_key = f'{network}_network'
+
+    # Verify the network provided is valid
+    try:
+        network = ip_network(config[network_key])
+    except Exception:
+        return False, f'Network address {config[network_key]} for {network_key} is not valid'
+
+    # Verify that the floating IP is valid (and in the network)
+    try:
+        floating_address = ip_address(config[floating_key].split('/')[0])
+        if floating_address not in list(network.hosts()):
+            raise
+    except Exception:
+        return False, f'Floating address {config[floating_key]} for {floating_key} is not valid'
+
+    return True, ''
+
+
+def get_configuration():
+    """
+    Parse the configuration of the node daemon.
+    """
+    pvcnoded_config_file = get_configuration_path()
+
+    print('Loading configuration from file "{}"'.format(pvcnoded_config_file))
+
+    with open(pvcnoded_config_file, 'r') as cfgfile:
+        try:
+            o_config = yaml.load(cfgfile, Loader=yaml.SafeLoader)
+        except Exception as e:
+            print('ERROR: Failed to parse configuration file: {}'.format(e))
+            os._exit(1)
+
+    node_fqdn, node_hostname, node_domain, node_id = get_hostname()
+
+    # Create the configuration dictionary
+    config = dict()
+
+    # Get the initial base configuration
+    try:
+        o_base = o_config['pvc']
+        o_cluster = o_config['pvc']['cluster']
+    except Exception as e:
+        raise MalformedConfigurationError(e)
+
+    config_general = {
+        'node':          o_base.get('node', node_hostname),
+        'node_hostname': node_hostname,
+        'node_fqdn':     node_fqdn,
+        'node_domain':   node_domain,
+        'node_id':       node_id,
+        'coordinators':  o_cluster.get('coordinators', list()),
+        'debug':         o_base.get('debug', False),
+    }
+
+    config = {**config, **config_general}
+
+    # Get the functions configuration
+    try:
+        o_functions = o_config['pvc']['functions']
+    except Exception as e:
+        raise MalformedConfigurationError(e)
+
+    config_functions = {
+        'enable_hypervisor': o_functions.get('enable_hypervisor', False),
+        'enable_networking': o_functions.get('enable_networking', False),
+        'enable_storage':    o_functions.get('enable_storage', False),
+        'enable_api':        o_functions.get('enable_api', False),
+    }
+
+    config = {**config, **config_functions}
+
+    # Get the directory configuration
+    try:
+        o_directories = o_config['pvc']['system']['configuration']['directories']
+    except Exception as e:
+        raise MalformedConfigurationError(e)
+
+    config_directories = {
+        'dynamic_directory':     o_directories.get('dynamic_directory', None),
+        'log_directory':         o_directories.get('log_directory', None),
+        'console_log_directory': o_directories.get('console_log_directory', None),
+    }
+
+    # Define our dynamic directory schema
+    config_directories['dnsmasq_dynamic_directory'] = config_directories['dynamic_directory'] + '/dnsmasq'
+    config_directories['pdns_dynamic_directory'] = config_directories['dynamic_directory'] + '/pdns'
+    config_directories['nft_dynamic_directory'] = config_directories['dynamic_directory'] + '/nft'
+
+    # Define our log directory schema
+    config_directories['dnsmasq_log_directory'] = config_directories['log_directory'] + '/dnsmasq'
+    config_directories['pdns_log_directory'] = config_directories['log_directory'] + '/pdns'
+    config_directories['nft_log_directory'] = config_directories['log_directory'] + '/nft'
+
+    config = {**config, **config_directories}
+
+    # Get the logging configuration
+    try:
+        o_logging = o_config['pvc']['system']['configuration']['logging']
+    except Exception as e:
+        raise MalformedConfigurationError(e)
+
+    config_logging = {
+        'file_logging':                  o_logging.get('file_logging', False),
+        'stdout_logging':                o_logging.get('stdout_logging', False),
+        'zookeeper_logging':             o_logging.get('zookeeper_logging', False),
+        'log_colours':                   o_logging.get('log_colours', False),
+        'log_dates':                     o_logging.get('log_dates', False),
+        'log_keepalives':                o_logging.get('log_keepalives', False),
+        'log_keepalive_cluster_details': o_logging.get('log_keepalive_cluster_details', False),
+        'log_keepalive_storage_details': o_logging.get('log_keepalive_storage_details', False),
+        'console_log_lines':             o_logging.get('console_log_lines', False),
+        'node_log_lines':                o_logging.get('node_log_lines', False),
+    }
+
+    config = {**config, **config_logging}
+
+    # Get the interval configuration
+    try:
+        o_intervals = o_config['pvc']['system']['intervals']
+    except Exception as e:
+        raise MalformedConfigurationError(e)
+
+    config_intervals = {
+        'vm_shutdown_timeout':  int(o_intervals.get('vm_shutdown_timeout', 60)),
+        'keepalive_interval':   int(o_intervals.get('keepalive_interval', 5)),
+        'fence_intervals':      int(o_intervals.get('fence_intervals', 6)),
+        'suicide_intervals':    int(o_intervals.get('suicide_interval', 0)),
+    }
+
+    config = {**config, **config_intervals}
+
+    # Get the fencing configuration
+    try:
+        o_fencing = o_config['pvc']['system']['fencing']
+        o_fencing_actions = o_fencing['actions']
+        o_fencing_ipmi = o_fencing['ipmi']
+    except Exception as e:
+        raise MalformedConfigurationError(e)
+
+    config_fencing = {
+        'successful_fence': o_fencing_actions.get('successful_fence', None),
+        'failed_fence':     o_fencing_actions.get('failed_fence', None),
+        'ipmi_hostname':    o_fencing_ipmi.get('host', f'{node_hostname}-lom.{node_domain}'),
+        'ipmi_username':    o_fencing_ipmi.get('user', 'null'),
+        'ipmi_password':    o_fencing_ipmi.get('pass', 'null'),
+    }
+
+    config = {**config, **config_fencing}
+
+    # Get the migration configuration
+    try:
+        o_migration = o_config['pvc']['system']['migration']
+    except Exception as e:
+        raise MalformedConfigurationError(e)
+
+    config_migration = {
+        'migration_target_selector': o_migration.get('target_selector', 'mem'),
+    }
+
+    config = {**config, **config_migration}
+
+    if config['enable_networking']:
+        # Get the node networks configuration
+        try:
+            o_networks = o_config['pvc']['cluster']['networks']
+            o_network_cluster = o_networks['cluster']
+            o_network_storage = o_networks['storage']
+            o_network_upstream = o_networks['upstream']
+            o_sysnetworks = o_config['pvc']['system']['configuration']['networking']
+            o_sysnetwork_cluster = o_sysnetworks['cluster']
+            o_sysnetwork_storage = o_sysnetworks['storage']
+            o_sysnetwork_upstream = o_sysnetworks['upstream']
+        except Exception as e:
+            raise MalformedConfigurationError(e)
+
+        config_networks = {
+            'cluster_domain':       o_network_cluster.get('domain', None),
+            'cluster_network':      o_network_cluster.get('network', None),
+            'cluster_floating_ip':  o_network_cluster.get('floating_ip', None),
+            'cluster_dev':          o_sysnetwork_cluster.get('device', None),
+            'cluster_mtu':          o_sysnetwork_cluster.get('mtu', None),
+            'cluster_dev_ip':       o_sysnetwork_cluster.get('address', None),
+            'storage_domain':       o_network_storage.get('domain', None),
+            'storage_network':      o_network_storage.get('network', None),
+            'storage_floating_ip':  o_network_storage.get('floating_ip', None),
+            'storage_dev':          o_sysnetwork_storage.get('device', None),
+            'storage_mtu':          o_sysnetwork_storage.get('mtu', None),
+            'storage_dev_ip':       o_sysnetwork_storage.get('address', None),
+            'upstream_domain':      o_network_upstream.get('domain', None),
+            'upstream_network':     o_network_upstream.get('network', None),
+            'upstream_floating_ip': o_network_upstream.get('floating_ip', None),
+            'upstream_gateway':     o_network_upstream.get('gateway', None),
+            'upstream_dev':         o_sysnetwork_upstream.get('device', None),
+            'upstream_mtu':         o_sysnetwork_upstream.get('mtu', None),
+            'upstream_dev_ip':      o_sysnetwork_upstream.get('address', None),
+            'bridge_dev':           o_sysnetworks.get('bridge_device', None),
+            'enable_sriov':         o_sysnetworks.get('sriov_enable', False),
+            'sriov_device':         o_sysnetworks.get('sriov_device', list())
+        }
+
+        config = {**config, **config_networks}
+
+        for network_type in ['cluster', 'storage', 'upstream']:
+            result, msg = validate_floating_ip(config, network_type)
+            if not result:
+                raise MalformedConfigurationError(msg)
+
+            address_key = '{}_dev_ip'.format(network_type)
+            network_key = f'{network_type}_network'
+            network = ip_network(config[network_key])
+            # With autoselection of addresses, construct an IP from the relevant network
+            if config[address_key] == 'by-id':
+                # The NodeID starts at 1, but indexes start at 0
+                address_id = int(config['node_id']) - 1
+                # Grab the nth address from the network
+                config[address_key] = '{}/{}'.format(list(network.hosts())[address_id], network.prefixlen)
+            # Validate the provided IP instead
+            else:
+                try:
+                    address = ip_address(config[address_key].split('/')[0])
+                    if address not in list(network.hosts()):
+                        raise
+                except Exception:
+                    raise MalformedConfigurationError(
+                        f'IP address {config[address_key]} for {address_key} is not valid'
+                    )
+
+        # Get the PowerDNS aggregator database configuration
+        try:
+            o_pdnsdb = o_config['pvc']['coordinator']['dns']['database']
+        except Exception as e:
+            raise MalformedConfigurationError(e)
+
+        config_pdnsdb = {
+            'pdns_postgresql_host':     o_pdnsdb.get('host', None),
+            'pdns_postgresql_port':     o_pdnsdb.get('port', None),
+            'pdns_postgresql_dbname':   o_pdnsdb.get('name', None),
+            'pdns_postgresql_user':     o_pdnsdb.get('user', None),
+            'pdns_postgresql_password': o_pdnsdb.get('pass', None),
+        }
+
+        config = {**config, **config_pdnsdb}
+
+        # Get the Cloud-Init Metadata database configuration
+        try:
+            o_metadatadb = o_config['pvc']['coordinator']['metadata']['database']
+        except Exception as e:
+            raise MalformedConfigurationError(e)
+
+        config_metadatadb = {
+            'metadata_postgresql_host':     o_metadatadb.get('host', None),
+            'metadata_postgresql_port':     o_metadatadb.get('port', None),
+            'metadata_postgresql_dbname':   o_metadatadb.get('name', None),
+            'metadata_postgresql_user':     o_metadatadb.get('user', None),
+            'metadata_postgresql_password': o_metadatadb.get('pass', None),
+        }
+
+        config = {**config, **config_metadatadb}
+
+    if config['enable_storage']:
+        # Get the storage configuration
+        try:
+            o_storage = o_config['pvc']['system']['configuration']['storage']
+        except Exception as e:
+            raise MalformedConfigurationError(e)
+
+        config_storage = {
+            'ceph_config_file':   o_storage.get('ceph_config_file', None),
+            'ceph_admin_keyring': o_storage.get('ceph_admin_keyring', None),
+        }
+
+        config = {**config, **config_storage}
+
+        # Add our node static data to the config
+        config['static_data'] = get_static_data()
+
+    return config
+
+
+def validate_directories(config):
+    if not os.path.exists(config['dynamic_directory']):
+        os.makedirs(config['dynamic_directory'])
+        os.makedirs(config['dnsmasq_dynamic_directory'])
+        os.makedirs(config['pdns_dynamic_directory'])
+        os.makedirs(config['nft_dynamic_directory'])
+
+    if not os.path.exists(config['log_directory']):
+        os.makedirs(config['log_directory'])
+        os.makedirs(config['dnsmasq_log_directory'])
+        os.makedirs(config['pdns_log_directory'])
+        os.makedirs(config['nft_log_directory'])

node-daemon/pvcnoded/util/fencing.py

@@ -0,0 +1,187 @@
+#!/usr/bin/env python3
+
+# fencing.py - Utility functions for pvcnoded fencing
+# Part of the Parallel Virtual Cluster (PVC) system
+#
+#    Copyright (C) 2018-2021 Joshua M. Boniface <joshua@boniface.me>
+#
+#    This program is free software: you can redistribute it and/or modify
+#    it under the terms of the GNU General Public License as published by
+#    the Free Software Foundation, version 3.
+#
+#    This program is distributed in the hope that it will be useful,
+#    but WITHOUT ANY WARRANTY; without even the implied warranty of
+#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#    GNU General Public License for more details.
+#
+#    You should have received a copy of the GNU General Public License
+#    along with this program.  If not, see <https://www.gnu.org/licenses/>.
+#
+###############################################################################
+
+import time
+
+import daemon_lib.common as common
+
+from pvcnoded.objects.VMInstance import VMInstance
+
+
+#
+# Fence thread entry function
+#
+def fence_node(node_name, zkhandler, config, logger):
+    # We allow exactly 6 saving throws (30 seconds) for the host to come back online or we kill it
+    failcount_limit = 6
+    failcount = 0
+    while failcount < failcount_limit:
+        # Wait 5 seconds
+        time.sleep(config['keepalive_interval'])
+        # Get the state
+        node_daemon_state = zkhandler.read(('node.state.daemon', node_name))
+        # Is it still 'dead'
+        if node_daemon_state == 'dead':
+            failcount += 1
+            logger.out('Node "{}" failed {}/{} saving throws'.format(node_name, failcount, failcount_limit), state='s')
+        # It changed back to something else so it must be alive
+        else:
+            logger.out('Node "{}" passed a saving throw; canceling fence'.format(node_name), state='o')
+            return
+
+    logger.out('Fencing node "{}" via IPMI reboot signal'.format(node_name), state='s')
+
+    # Get IPMI information
+    ipmi_hostname = zkhandler.read(('node.ipmi.hostname', node_name))
+    ipmi_username = zkhandler.read(('node.ipmi.username', node_name))
+    ipmi_password = zkhandler.read(('node.ipmi.password', node_name))
+
+    # Shoot it in the head
+    fence_status = reboot_via_ipmi(ipmi_hostname, ipmi_username, ipmi_password, logger)
+    # Hold to ensure the fence takes effect and system stabilizes
+    time.sleep(config['keepalive_interval'] * 2)
+
+    # Force into secondary network state if needed
+    if node_name in config['coordinators']:
+        logger.out('Forcing secondary status for node "{}"'.format(node_name), state='i')
+        zkhandler.write([
+            (('node.state.router', node_name), 'secondary')
+        ])
+        if zkhandler.read('base.config.primary_node') == node_name:
+            zkhandler.write([
+                ('base.config.primary_node', 'none')
+            ])
+
+    # If the fence succeeded and successful_fence is migrate
+    if fence_status and config['successful_fence'] == 'migrate':
+        migrateFromFencedNode(zkhandler, node_name, config, logger)
+
+    # If the fence failed and failed_fence is migrate
+    if not fence_status and config['failed_fence'] == 'migrate' and config['suicide_intervals'] != '0':
+        migrateFromFencedNode(zkhandler, node_name, config, logger)
+
+
+# Migrate hosts away from a fenced node
+def migrateFromFencedNode(zkhandler, node_name, config, logger):
+    logger.out('Migrating VMs from dead node "{}" to new hosts'.format(node_name), state='i')
+
+    # Get the list of VMs
+    dead_node_running_domains = zkhandler.read(('node.running_domains', node_name)).split()
+
+    # Set the node to a custom domainstate so we know what's happening
+    zkhandler.write([
+        (('node.state.domain', node_name), 'fence-flush')
+    ])
+
+    # Migrate a VM after a flush
+    def fence_migrate_vm(dom_uuid):
+        VMInstance.flush_locks(zkhandler, logger, dom_uuid)
+
+        target_node = common.findTargetNode(zkhandler, dom_uuid)
+
+        if target_node is not None:
+            logger.out('Migrating VM "{}" to node "{}"'.format(dom_uuid, target_node), state='i')
+            zkhandler.write([
+                (('domain.state', dom_uuid), 'start'),
+                (('domain.node', dom_uuid), target_node),
+                (('domain.last_node', dom_uuid), node_name),
+            ])
+        else:
+            logger.out('No target node found for VM "{}"; VM will autostart on next unflush/ready of current node'.format(dom_uuid), state='i')
+            zkhandler.write({
+                (('domain.state', dom_uuid), 'stopped'),
+                (('domain.meta.autostart', dom_uuid), 'True'),
+            })
+
+    # Loop through the VMs
+    for dom_uuid in dead_node_running_domains:
+        fence_migrate_vm(dom_uuid)
+
+    # Set node in flushed state for easy remigrating when it comes back
+    zkhandler.write([
+        (('node.state.domain', node_name), 'flushed')
+    ])
+
+
+#
+# Perform an IPMI fence
+#
+def reboot_via_ipmi(ipmi_hostname, ipmi_user, ipmi_password, logger):
+    # Forcibly reboot the node
+    ipmi_command_reset = '/usr/bin/ipmitool -I lanplus -H {} -U {} -P {} chassis power reset'.format(
+        ipmi_hostname, ipmi_user, ipmi_password
+    )
+    ipmi_reset_retcode, ipmi_reset_stdout, ipmi_reset_stderr = common.run_os_command(ipmi_command_reset)
+
+    if ipmi_reset_retcode != 0:
+        logger.out(f'Failed to reboot dead node: {ipmi_reset_stderr}', state='e')
+
+    time.sleep(1)
+
+    # Power on the node (just in case it is offline)
+    ipmi_command_start = '/usr/bin/ipmitool -I lanplus -H {} -U {} -P {} chassis power on'.format(
+        ipmi_hostname, ipmi_user, ipmi_password
+    )
+    ipmi_start_retcode, ipmi_start_stdout, ipmi_start_stderr = common.run_os_command(ipmi_command_start)
+
+    time.sleep(2)
+
+    # Check the chassis power state
+    logger.out('Checking power state of dead node', state='i')
+    ipmi_command_status = '/usr/bin/ipmitool -I lanplus -H {} -U {} -P {} chassis power status'.format(
+        ipmi_hostname, ipmi_user, ipmi_password
+    )
+    ipmi_status_retcode, ipmi_status_stdout, ipmi_status_stderr = common.run_os_command(ipmi_command_status)
+
+    if ipmi_reset_retcode == 0:
+        if ipmi_status_stdout == "Chassis Power is on":
+            # We successfully rebooted the node and it is powered on; this is a succeessful fence
+            logger.out('Successfully rebooted dead node', state='o')
+            return True
+        elif ipmi_status_stdout == "Chassis Power is off":
+            # We successfully rebooted the node but it is powered off; this might be expected or not, but the node is confirmed off so we can call it a successful fence
+            logger.out('Chassis power is in confirmed off state after successfuly IPMI reboot; proceeding with fence-flush', state='o')
+            return True
+        else:
+            # We successfully rebooted the node but it is in some unknown power state; since this might indicate a silent failure, we must call it a failed fence
+            logger.out('Chassis power is in an unknown state after successful IPMI reboot; not performing fence-flush', state='e')
+            return False
+    else:
+        if ipmi_status_stdout == "Chassis Power is off":
+            # We failed to reboot the node but it is powered off; it has probably suffered a serious hardware failure, but the node is confirmed off so we can call it a successful fence
+            logger.out('Chassis power is in confirmed off state after failed IPMI reboot; proceeding with fence-flush', state='o')
+            return True
+        else:
+            # We failed to reboot the node but it is in some unknown power state (including "on"); since this might indicate a silent failure, we must call it a failed fence
+            logger.out('Chassis power is not in confirmed off state after failed IPMI reboot; not performing fence-flush', state='e')
+            return False
+
+
+#
+# Verify that IPMI connectivity to this host exists (used during node init)
+#
+def verify_ipmi(ipmi_hostname, ipmi_user, ipmi_password):
+    ipmi_command = f'/usr/bin/ipmitool -I lanplus -H {ipmi_hostname} -U {ipmi_user} -P {ipmi_password} chassis power status'
+    retcode, stdout, stderr = common.run_os_command(ipmi_command, timeout=2)
+    if retcode == 0 and stdout != "Chassis Power is on":
+        return True
+    else:
+        return False

node-daemon/pvcnoded/util/keepalive.py

@@ -0,0 +1,718 @@
+#!/usr/bin/env python3
+
+# keepalive.py - Utility functions for pvcnoded Keepalives
+# Part of the Parallel Virtual Cluster (PVC) system
+#
+#    Copyright (C) 2018-2021 Joshua M. Boniface <joshua@boniface.me>
+#
+#    This program is free software: you can redistribute it and/or modify
+#    it under the terms of the GNU General Public License as published by
+#    the Free Software Foundation, version 3.
+#
+#    This program is distributed in the hope that it will be useful,
+#    but WITHOUT ANY WARRANTY; without even the implied warranty of
+#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#    GNU General Public License for more details.
+#
+#    You should have received a copy of the GNU General Public License
+#    along with this program.  If not, see <https://www.gnu.org/licenses/>.
+#
+###############################################################################
+
+import pvcnoded.util.fencing
+
+import daemon_lib.common as common
+
+from apscheduler.schedulers.background import BackgroundScheduler
+from rados import Rados
+from xml.etree import ElementTree
+from queue import Queue
+from threading import Thread
+from datetime import datetime
+
+import json
+import re
+import libvirt
+import psutil
+import os
+import time
+
+
+# State table for pretty stats
+libvirt_vm_states = {
+    0: "NOSTATE",
+    1: "RUNNING",
+    2: "BLOCKED",
+    3: "PAUSED",
+    4: "SHUTDOWN",
+    5: "SHUTOFF",
+    6: "CRASHED",
+    7: "PMSUSPENDED"
+}
+
+
+def start_keepalive_timer(logger, config, zkhandler, this_node):
+    keepalive_interval = config['keepalive_interval']
+    logger.out(f'Starting keepalive timer ({keepalive_interval} second interval)', state='s')
+    keepalive_timer = BackgroundScheduler()
+    keepalive_timer.add_job(
+        node_keepalive,
+        args=(logger, config, zkhandler, this_node),
+        trigger='interval',
+        seconds=keepalive_interval)
+    keepalive_timer.start()
+    return keepalive_timer
+
+
+def stop_keepalive_timer(logger, keepalive_timer):
+    try:
+        keepalive_timer.shutdown()
+        logger.out('Stopping keepalive timer', state='s')
+    except Exception:
+        logger.out('Failed to stop keepalive timer', state='w')
+
+
+# Ceph stats update function
+def collect_ceph_stats(logger, config, zkhandler, this_node, queue):
+    pool_list = zkhandler.children('base.pool')
+    osd_list = zkhandler.children('base.osd')
+
+    debug = config['debug']
+    if debug:
+        logger.out("Thread starting", state='d', prefix='ceph-thread')
+
+    # Connect to the Ceph cluster
+    try:
+        ceph_conn = Rados(conffile=config['ceph_config_file'], conf=dict(keyring=config['ceph_admin_keyring']))
+        if debug:
+            logger.out("Connecting to cluster", state='d', prefix='ceph-thread')
+        ceph_conn.connect(timeout=1)
+    except Exception as e:
+        logger.out('Failed to open connection to Ceph cluster: {}'.format(e), state='e')
+        return
+
+    if debug:
+        logger.out("Getting health stats from monitor", state='d', prefix='ceph-thread')
+
+    # Get Ceph cluster health for local status output
+    command = {"prefix": "health", "format": "json"}
+    try:
+        health_status = json.loads(ceph_conn.mon_command(json.dumps(command), b'', timeout=1)[1])
+        ceph_health = health_status['status']
+    except Exception as e:
+        logger.out('Failed to obtain Ceph health data: {}'.format(e), state='e')
+        ceph_health = 'HEALTH_UNKN'
+
+    if ceph_health in ['HEALTH_OK']:
+        ceph_health_colour = logger.fmt_green
+    elif ceph_health in ['HEALTH_UNKN']:
+        ceph_health_colour = logger.fmt_cyan
+    elif ceph_health in ['HEALTH_WARN']:
+        ceph_health_colour = logger.fmt_yellow
+    else:
+        ceph_health_colour = logger.fmt_red
+
+    # Primary-only functions
+    if this_node.router_state == 'primary':
+        if debug:
+            logger.out("Set ceph health information in zookeeper (primary only)", state='d', prefix='ceph-thread')
+
+        command = {"prefix": "status", "format": "pretty"}
+        ceph_status = ceph_conn.mon_command(json.dumps(command), b'', timeout=1)[1].decode('ascii')
+        try:
+            zkhandler.write([
+                ('base.storage', str(ceph_status))
+            ])
+        except Exception as e:
+            logger.out('Failed to set Ceph status data: {}'.format(e), state='e')
+
+        if debug:
+            logger.out("Set ceph rados df information in zookeeper (primary only)", state='d', prefix='ceph-thread')
+
+        # Get rados df info
+        command = {"prefix": "df", "format": "pretty"}
+        ceph_df = ceph_conn.mon_command(json.dumps(command), b'', timeout=1)[1].decode('ascii')
+        try:
+            zkhandler.write([
+                ('base.storage.util', str(ceph_df))
+            ])
+        except Exception as e:
+            logger.out('Failed to set Ceph utilization data: {}'.format(e), state='e')
+
+        if debug:
+            logger.out("Set pool information in zookeeper (primary only)", state='d', prefix='ceph-thread')
+
+        # Get pool info
+        command = {"prefix": "df", "format": "json"}
+        ceph_df_output = ceph_conn.mon_command(json.dumps(command), b'', timeout=1)[1].decode('ascii')
+        try:
+            ceph_pool_df_raw = json.loads(ceph_df_output)['pools']
+        except Exception as e:
+            logger.out('Failed to obtain Pool data (ceph df): {}'.format(e), state='w')
+            ceph_pool_df_raw = []
+
+        retcode, stdout, stderr = common.run_os_command('rados df --format json', timeout=1)
+        try:
+            rados_pool_df_raw = json.loads(stdout)['pools']
+        except Exception as e:
+            logger.out('Failed to obtain Pool data (rados df): {}'.format(e), state='w')
+            rados_pool_df_raw = []
+
+        pool_count = len(ceph_pool_df_raw)
+        if debug:
+            logger.out("Getting info for {} pools".format(pool_count), state='d', prefix='ceph-thread')
+        for pool_idx in range(0, pool_count):
+            try:
+                # Combine all the data for this pool
+                ceph_pool_df = ceph_pool_df_raw[pool_idx]
+                rados_pool_df = rados_pool_df_raw[pool_idx]
+                pool = ceph_pool_df
+                pool.update(rados_pool_df)
+
+                # Ignore any pools that aren't in our pool list
+                if pool['name'] not in pool_list:
+                    if debug:
+                        logger.out("Pool {} not in pool list {}".format(pool['name'], pool_list), state='d', prefix='ceph-thread')
+                    continue
+                else:
+                    if debug:
+                        logger.out("Parsing data for pool {}".format(pool['name']), state='d', prefix='ceph-thread')
+
+                # Assemble a useful data structure
+                pool_df = {
+                    'id': pool['id'],
+                    'stored_bytes': pool['stats']['stored'],
+                    'free_bytes': pool['stats']['max_avail'],
+                    'used_bytes': pool['stats']['bytes_used'],
+                    'used_percent': pool['stats']['percent_used'],
+                    'num_objects': pool['stats']['objects'],
+                    'num_object_clones': pool['num_object_clones'],
+                    'num_object_copies': pool['num_object_copies'],
+                    'num_objects_missing_on_primary': pool['num_objects_missing_on_primary'],
+                    'num_objects_unfound': pool['num_objects_unfound'],
+                    'num_objects_degraded': pool['num_objects_degraded'],
+                    'read_ops': pool['read_ops'],
+                    'read_bytes': pool['read_bytes'],
+                    'write_ops': pool['write_ops'],
+                    'write_bytes': pool['write_bytes']
+                }
+
+                # Write the pool data to Zookeeper
+                zkhandler.write([
+                    (('pool.stats', pool['name']), str(json.dumps(pool_df)))
+                ])
+            except Exception as e:
+                # One or more of the status commands timed out, just continue
+                logger.out('Failed to format and send pool data: {}'.format(e), state='w')
+                pass
+
+    # Only grab OSD stats if there are OSDs to grab (otherwise `ceph osd df` hangs)
+    osds_this_node = 0
+    if len(osd_list) > 0:
+        # Get data from Ceph OSDs
+        if debug:
+            logger.out("Get data from Ceph OSDs", state='d', prefix='ceph-thread')
+
+        # Parse the dump data
+        osd_dump = dict()
+
+        command = {"prefix": "osd dump", "format": "json"}
+        osd_dump_output = ceph_conn.mon_command(json.dumps(command), b'', timeout=1)[1].decode('ascii')
+        try:
+            osd_dump_raw = json.loads(osd_dump_output)['osds']
+        except Exception as e:
+            logger.out('Failed to obtain OSD data: {}'.format(e), state='w')
+            osd_dump_raw = []
+
+        if debug:
+            logger.out("Loop through OSD dump", state='d', prefix='ceph-thread')
+        for osd in osd_dump_raw:
+            osd_dump.update({
+                str(osd['osd']): {
+                    'uuid': osd['uuid'],
+                    'up': osd['up'],
+                    'in': osd['in'],
+                    'primary_affinity': osd['primary_affinity']
+                }
+            })
+
+        # Parse the df data
+        if debug:
+            logger.out("Parse the OSD df data", state='d', prefix='ceph-thread')
+
+        osd_df = dict()
+
+        command = {"prefix": "osd df", "format": "json"}
+        try:
+            osd_df_raw = json.loads(ceph_conn.mon_command(json.dumps(command), b'', timeout=1)[1])['nodes']
+        except Exception as e:
+            logger.out('Failed to obtain OSD data: {}'.format(e), state='w')
+            osd_df_raw = []
+
+        if debug:
+            logger.out("Loop through OSD df", state='d', prefix='ceph-thread')
+        for osd in osd_df_raw:
+            osd_df.update({
+                str(osd['id']): {
+                    'utilization': osd['utilization'],
+                    'var': osd['var'],
+                    'pgs': osd['pgs'],
+                    'kb': osd['kb'],
+                    'weight': osd['crush_weight'],
+                    'reweight': osd['reweight'],
+                }
+            })
+
+        # Parse the status data
+        if debug:
+            logger.out("Parse the OSD status data", state='d', prefix='ceph-thread')
+
+        osd_status = dict()
+
+        command = {"prefix": "osd status", "format": "pretty"}
+        try:
+            osd_status_raw = ceph_conn.mon_command(json.dumps(command), b'', timeout=1)[1].decode('ascii')
+        except Exception as e:
+            logger.out('Failed to obtain OSD status data: {}'.format(e), state='w')
+            osd_status_raw = []
+
+        if debug:
+            logger.out("Loop through OSD status data", state='d', prefix='ceph-thread')
+
+        for line in osd_status_raw.split('\n'):
+            # Strip off colour
+            line = re.sub(r'\x1b(\[.*?[@-~]|\].*?(\x07|\x1b\\))', '', line)
+            # Split it for parsing
+            line = line.split()
+            if len(line) > 1 and line[1].isdigit():
+                # This is an OSD line so parse it
+                osd_id = line[1]
+                node = line[3].split('.')[0]
+                used = line[5]
+                avail = line[7]
+                wr_ops = line[9]
+                wr_data = line[11]
+                rd_ops = line[13]
+                rd_data = line[15]
+                state = line[17]
+                osd_status.update({
+                    str(osd_id): {
+                        'node': node,
+                        'used': used,
+                        'avail': avail,
+                        'wr_ops': wr_ops,
+                        'wr_data': wr_data,
+                        'rd_ops': rd_ops,
+                        'rd_data': rd_data,
+                        'state': state
+                    }
+                })
+
+        # Merge them together into a single meaningful dict
+        if debug:
+            logger.out("Merge OSD data together", state='d', prefix='ceph-thread')
+
+        osd_stats = dict()
+
+        for osd in osd_list:
+            if zkhandler.read(('osd.node', osd)) == config['node_hostname']:
+                osds_this_node += 1
+            try:
+                this_dump = osd_dump[osd]
+                this_dump.update(osd_df[osd])
+                this_dump.update(osd_status[osd])
+                osd_stats[osd] = this_dump
+            except KeyError as e:
+                # One or more of the status commands timed out, just continue
+                logger.out('Failed to parse OSD stats into dictionary: {}'.format(e), state='w')
+
+        # Upload OSD data for the cluster (primary-only)
+        if this_node.router_state == 'primary':
+            if debug:
+                logger.out("Trigger updates for each OSD", state='d', prefix='ceph-thread')
+
+            for osd in osd_list:
+                try:
+                    stats = json.dumps(osd_stats[osd])
+                    zkhandler.write([
+                        (('osd.stats', osd), str(stats))
+                    ])
+                except KeyError as e:
+                    # One or more of the status commands timed out, just continue
+                    logger.out('Failed to upload OSD stats from dictionary: {}'.format(e), state='w')
+
+    ceph_conn.shutdown()
+
+    queue.put(ceph_health_colour)
+    queue.put(ceph_health)
+    queue.put(osds_this_node)
+
+    if debug:
+        logger.out("Thread finished", state='d', prefix='ceph-thread')
+
+
+# VM stats update function
+def collect_vm_stats(logger, config, zkhandler, this_node, queue):
+    debug = config['debug']
+    if debug:
+        logger.out("Thread starting", state='d', prefix='vm-thread')
+
+    # Connect to libvirt
+    libvirt_name = "qemu:///system"
+    if debug:
+        logger.out("Connecting to libvirt", state='d', prefix='vm-thread')
+    lv_conn = libvirt.open(libvirt_name)
+    if lv_conn is None:
+        logger.out('Failed to open connection to "{}"'.format(libvirt_name), state='e')
+
+    memalloc = 0
+    memprov = 0
+    vcpualloc = 0
+    # Toggle state management of dead VMs to restart them
+    if debug:
+        logger.out("Toggle state management of dead VMs to restart them", state='d', prefix='vm-thread')
+    # Make a copy of the d_domain; if not, and it changes in flight, this can fail
+    fixed_d_domain = this_node.d_domain.copy()
+    for domain, instance in fixed_d_domain.items():
+        if domain in this_node.domain_list:
+            # Add the allocated memory to our memalloc value
+            memalloc += instance.getmemory()
+            memprov += instance.getmemory()
+            vcpualloc += instance.getvcpus()
+            if instance.getstate() == 'start' and instance.getnode() == this_node.name:
+                if instance.getdom() is not None:
+                    try:
+                        if instance.getdom().state()[0] != libvirt.VIR_DOMAIN_RUNNING:
+                            logger.out("VM {} has failed".format(instance.domname), state='w', prefix='vm-thread')
+                            raise
+                    except Exception:
+                        # Toggle a state "change"
+                        logger.out("Resetting state to {} for VM {}".format(instance.getstate(), instance.domname), state='i', prefix='vm-thread')
+                        zkhandler.write([
+                            (('domain.state', domain), instance.getstate())
+                        ])
+        elif instance.getnode() == this_node.name:
+            memprov += instance.getmemory()
+
+    # Get list of running domains from Libvirt
+    running_domains = lv_conn.listAllDomains(libvirt.VIR_CONNECT_LIST_DOMAINS_ACTIVE)
+
+    # Get statistics from any running VMs
+    for domain in running_domains:
+        try:
+            # Get basic information about the VM
+            tree = ElementTree.fromstring(domain.XMLDesc())
+            domain_uuid = domain.UUIDString()
+            domain_name = domain.name()
+
+            # Get all the raw information about the VM
+            if debug:
+                logger.out("Getting general statistics for VM {}".format(domain_name), state='d', prefix='vm-thread')
+            domain_state, domain_maxmem, domain_mem, domain_vcpus, domain_cputime = domain.info()
+            # We can't properly gather stats from a non-running VMs so continue
+            if domain_state != libvirt.VIR_DOMAIN_RUNNING:
+                continue
+            domain_memory_stats = domain.memoryStats()
+            domain_cpu_stats = domain.getCPUStats(True)[0]
+        except Exception as e:
+            if debug:
+                try:
+                    logger.out("Failed getting VM information for {}: {}".format(domain.name(), e), state='d', prefix='vm-thread')
+                except Exception:
+                    pass
+            continue
+
+        # Ensure VM is present in the domain_list
+        if domain_uuid not in this_node.domain_list:
+            this_node.domain_list.append(domain_uuid)
+
+        if debug:
+            logger.out("Getting disk statistics for VM {}".format(domain_name), state='d', prefix='vm-thread')
+        domain_disk_stats = []
+        for disk in tree.findall('devices/disk'):
+            disk_name = disk.find('source').get('name')
+            if not disk_name:
+                disk_name = disk.find('source').get('file')
+            disk_stats = domain.blockStats(disk.find('target').get('dev'))
+            domain_disk_stats.append({
+                "name": disk_name,
+                "rd_req": disk_stats[0],
+                "rd_bytes": disk_stats[1],
+                "wr_req": disk_stats[2],
+                "wr_bytes": disk_stats[3],
+                "err": disk_stats[4]
+            })
+
+        if debug:
+            logger.out("Getting network statistics for VM {}".format(domain_name), state='d', prefix='vm-thread')
+        domain_network_stats = []
+        for interface in tree.findall('devices/interface'):
+            interface_type = interface.get('type')
+            if interface_type not in ['bridge']:
+                continue
+            interface_name = interface.find('target').get('dev')
+            interface_bridge = interface.find('source').get('bridge')
+            interface_stats = domain.interfaceStats(interface_name)
+            domain_network_stats.append({
+                "name": interface_name,
+                "bridge": interface_bridge,
+                "rd_bytes": interface_stats[0],
+                "rd_packets": interface_stats[1],
+                "rd_errors": interface_stats[2],
+                "rd_drops": interface_stats[3],
+                "wr_bytes": interface_stats[4],
+                "wr_packets": interface_stats[5],
+                "wr_errors": interface_stats[6],
+                "wr_drops": interface_stats[7]
+            })
+
+        # Create the final dictionary
+        domain_stats = {
+            "state": libvirt_vm_states[domain_state],
+            "maxmem": domain_maxmem,
+            "livemem": domain_mem,
+            "cpus": domain_vcpus,
+            "cputime": domain_cputime,
+            "mem_stats": domain_memory_stats,
+            "cpu_stats": domain_cpu_stats,
+            "disk_stats": domain_disk_stats,
+            "net_stats": domain_network_stats
+        }
+
+        if debug:
+            logger.out("Writing statistics for VM {} to Zookeeper".format(domain_name), state='d', prefix='vm-thread')
+
+        try:
+            zkhandler.write([
+                (('domain.stats', domain_uuid), str(json.dumps(domain_stats)))
+            ])
+        except Exception as e:
+            if debug:
+                logger.out("{}".format(e), state='d', prefix='vm-thread')
+
+    # Close the Libvirt connection
+    lv_conn.close()
+
+    queue.put(len(running_domains))
+    queue.put(memalloc)
+    queue.put(memprov)
+    queue.put(vcpualloc)
+
+    if debug:
+        logger.out("Thread finished", state='d', prefix='vm-thread')
+
+
+# Keepalive update function
+def node_keepalive(logger, config, zkhandler, this_node):
+    debug = config['debug']
+    if debug:
+        logger.out("Keepalive starting", state='d', prefix='main-thread')
+
+    # Set the migration selector in Zookeeper for clients to read
+    if config['enable_hypervisor']:
+        if this_node.router_state == 'primary':
+            try:
+                if zkhandler.read('base.config.migration_target_selector') != config['migration_target_selector']:
+                    raise
+            except Exception:
+                zkhandler.write([
+                    ('base.config.migration_target_selector', config['migration_target_selector'])
+                ])
+
+    # Set the upstream IP in Zookeeper for clients to read
+    if config['enable_networking']:
+        if this_node.router_state == 'primary':
+            try:
+                if zkhandler.read('base.config.upstream_ip') != config['upstream_floating_ip']:
+                    raise
+            except Exception:
+                zkhandler.write([
+                    ('base.config.upstream_ip', config['upstream_floating_ip'])
+                ])
+
+    # Get past state and update if needed
+    if debug:
+        logger.out("Get past state and update if needed", state='d', prefix='main-thread')
+
+    past_state = zkhandler.read(('node.state.daemon', this_node.name))
+    if past_state != 'run' and past_state != 'shutdown':
+        this_node.daemon_state = 'run'
+        zkhandler.write([
+            (('node.state.daemon', this_node.name), 'run')
+        ])
+    else:
+        this_node.daemon_state = 'run'
+
+    # Ensure the primary key is properly set
+    if debug:
+        logger.out("Ensure the primary key is properly set", state='d', prefix='main-thread')
+    if this_node.router_state == 'primary':
+        if zkhandler.read('base.config.primary_node') != this_node.name:
+            zkhandler.write([
+                ('base.config.primary_node', this_node.name)
+            ])
+
+    # Run VM statistics collection in separate thread for parallelization
+    if config['enable_hypervisor']:
+        vm_thread_queue = Queue()
+        vm_stats_thread = Thread(target=collect_vm_stats, args=(logger, config, zkhandler, this_node, vm_thread_queue), kwargs={})
+        vm_stats_thread.start()
+
+    # Run Ceph status collection in separate thread for parallelization
+    if config['enable_storage']:
+        ceph_thread_queue = Queue()
+        ceph_stats_thread = Thread(target=collect_ceph_stats, args=(logger, config, zkhandler, this_node, ceph_thread_queue), kwargs={})
+        ceph_stats_thread.start()
+
+    # Get node performance statistics
+    this_node.memtotal = int(psutil.virtual_memory().total / 1024 / 1024)
+    this_node.memused = int(psutil.virtual_memory().used / 1024 / 1024)
+    this_node.memfree = int(psutil.virtual_memory().free / 1024 / 1024)
+    this_node.cpuload = os.getloadavg()[0]
+
+    # Join against running threads
+    if config['enable_hypervisor']:
+        vm_stats_thread.join(timeout=4.0)
+        if vm_stats_thread.is_alive():
+            logger.out('VM stats gathering exceeded 4s timeout, continuing', state='w')
+    if config['enable_storage']:
+        ceph_stats_thread.join(timeout=4.0)
+        if ceph_stats_thread.is_alive():
+            logger.out('Ceph stats gathering exceeded 4s timeout, continuing', state='w')
+
+    # Get information from thread queues
+    if config['enable_hypervisor']:
+        try:
+            this_node.domains_count = vm_thread_queue.get()
+            this_node.memalloc = vm_thread_queue.get()
+            this_node.memprov = vm_thread_queue.get()
+            this_node.vcpualloc = vm_thread_queue.get()
+        except Exception:
+            pass
+    else:
+        this_node.domains_count = 0
+        this_node.memalloc = 0
+        this_node.memprov = 0
+        this_node.vcpualloc = 0
+
+    if config['enable_storage']:
+        try:
+            ceph_health_colour = ceph_thread_queue.get()
+            ceph_health = ceph_thread_queue.get()
+            osds_this_node = ceph_thread_queue.get()
+        except Exception:
+            ceph_health_colour = logger.fmt_cyan
+            ceph_health = 'UNKNOWN'
+            osds_this_node = '?'
+
+    # Set our information in zookeeper
+    keepalive_time = int(time.time())
+    if debug:
+        logger.out("Set our information in zookeeper", state='d', prefix='main-thread')
+    try:
+        zkhandler.write([
+            (('node.memory.total', this_node.name), str(this_node.memtotal)),
+            (('node.memory.used', this_node.name), str(this_node.memused)),
+            (('node.memory.free', this_node.name), str(this_node.memfree)),
+            (('node.memory.allocated', this_node.name), str(this_node.memalloc)),
+            (('node.memory.provisioned', this_node.name), str(this_node.memprov)),
+            (('node.vcpu.allocated', this_node.name), str(this_node.vcpualloc)),
+            (('node.cpu.load', this_node.name), str(this_node.cpuload)),
+            (('node.count.provisioned_domains', this_node.name), str(this_node.domains_count)),
+            (('node.running_domains', this_node.name), ' '.join(this_node.domain_list)),
+            (('node.keepalive', this_node.name), str(keepalive_time)),
+        ])
+    except Exception:
+        logger.out('Failed to set keepalive data', state='e')
+
+    # Display node information to the terminal
+    if config['log_keepalives']:
+        if this_node.router_state == 'primary':
+            cst_colour = logger.fmt_green
+        elif this_node.router_state == 'secondary':
+            cst_colour = logger.fmt_blue
+        else:
+            cst_colour = logger.fmt_cyan
+        logger.out(
+            '{}{} keepalive @ {}{} [{}{}{}]'.format(
+                logger.fmt_purple,
+                config['node_hostname'],
+                datetime.now(),
+                logger.fmt_end,
+                logger.fmt_bold + cst_colour,
+                this_node.router_state,
+                logger.fmt_end
+            ),
+            state='t'
+        )
+        if config['log_keepalive_cluster_details']:
+            logger.out(
+                '{bold}Maintenance:{nofmt} {maint}  '
+                '{bold}Active VMs:{nofmt} {domcount}  '
+                '{bold}Networks:{nofmt} {netcount}  '
+                '{bold}Load:{nofmt} {load}  '
+                '{bold}Memory [MiB]: VMs:{nofmt} {allocmem}  '
+                '{bold}Used:{nofmt} {usedmem}  '
+                '{bold}Free:{nofmt} {freemem}'.format(
+                    bold=logger.fmt_bold,
+                    nofmt=logger.fmt_end,
+                    maint=this_node.maintenance,
+                    domcount=this_node.domains_count,
+                    netcount=len(zkhandler.children('base.network')),
+                    load=this_node.cpuload,
+                    freemem=this_node.memfree,
+                    usedmem=this_node.memused,
+                    allocmem=this_node.memalloc,
+                ),
+                state='t'
+            )
+        if config['enable_storage'] and config['log_keepalive_storage_details']:
+            logger.out(
+                '{bold}Ceph cluster status:{nofmt} {health_colour}{health}{nofmt}  '
+                '{bold}Total OSDs:{nofmt} {total_osds}  '
+                '{bold}Node OSDs:{nofmt} {node_osds}  '
+                '{bold}Pools:{nofmt} {total_pools}  '.format(
+                    bold=logger.fmt_bold,
+                    health_colour=ceph_health_colour,
+                    nofmt=logger.fmt_end,
+                    health=ceph_health,
+                    total_osds=len(zkhandler.children('base.osd')),
+                    node_osds=osds_this_node,
+                    total_pools=len(zkhandler.children('base.pool'))
+                ),
+                state='t'
+            )
+
+    # Look for dead nodes and fence them
+    if not this_node.maintenance:
+        if debug:
+            logger.out("Look for dead nodes and fence them", state='d', prefix='main-thread')
+        if config['daemon_mode'] == 'coordinator':
+            for node_name in zkhandler.children('base.node'):
+                try:
+                    node_daemon_state = zkhandler.read(('node.state.daemon', node_name))
+                    node_keepalive = int(zkhandler.read(('node.keepalive', node_name)))
+                except Exception:
+                    node_daemon_state = 'unknown'
+                    node_keepalive = 0
+
+                # Handle deadtime and fencng if needed
+                # (A node is considered dead when its keepalive timer is >6*keepalive_interval seconds
+                # out-of-date while in 'start' state)
+                node_deadtime = int(time.time()) - (int(config['keepalive_interval']) * int(config['fence_intervals']))
+                if node_keepalive < node_deadtime and node_daemon_state == 'run':
+                    logger.out('Node {} seems dead - starting monitor for fencing'.format(node_name), state='w')
+                    zk_lock = zkhandler.writelock(('node.state.daemon', node_name))
+                    with zk_lock:
+                        # Ensures that, if we lost the lock race and come out of waiting,
+                        # we won't try to trigger our own fence thread.
+                        if zkhandler.read(('node.state.daemon', node_name)) != 'dead':
+                            fence_thread = Thread(target=pvcnoded.util.fencing.fence_node, args=(node_name, zkhandler, config, logger), kwargs={})
+                            fence_thread.start()
+                            # Write the updated data after we start the fence thread
+                            zkhandler.write([
+                                (('node.state.daemon', node_name), 'dead')
+                            ])
+
+    if debug:
+        logger.out("Keepalive finished", state='d', prefix='main-thread')

node-daemon/pvcnoded/util/libvirt.py

@@ -0,0 +1,36 @@
+#!/usr/bin/env python3
+
+# libvirt.py - Utility functions for pvcnoded libvirt
+# Part of the Parallel Virtual Cluster (PVC) system
+#
+#    Copyright (C) 2018-2021 Joshua M. Boniface <joshua@boniface.me>
+#
+#    This program is free software: you can redistribute it and/or modify
+#    it under the terms of the GNU General Public License as published by
+#    the Free Software Foundation, version 3.
+#
+#    This program is distributed in the hope that it will be useful,
+#    but WITHOUT ANY WARRANTY; without even the implied warranty of
+#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#    GNU General Public License for more details.
+#
+#    You should have received a copy of the GNU General Public License
+#    along with this program.  If not, see <https://www.gnu.org/licenses/>.
+#
+###############################################################################
+
+import libvirt
+
+
+def validate_libvirtd(logger, config):
+    if config['enable_hypervisor']:
+        libvirt_check_name = f'qemu+tcp://{config["node_hostname"]}/system'
+        logger.out(f'Connecting to Libvirt daemon at {libvirt_check_name}', state='i')
+        try:
+            lv_conn = libvirt.open(libvirt_check_name)
+            lv_conn.close()
+        except Exception as e:
+            logger.out(f'Failed to connect to Libvirt daemon: {e}', state='e')
+            return False
+
+    return True

node-daemon/pvcnoded/util/networking.py

@@ -0,0 +1,181 @@
+#!/usr/bin/env python3
+
+# networking.py - Utility functions for pvcnoded networking
+# Part of the Parallel Virtual Cluster (PVC) system
+#
+#    Copyright (C) 2018-2021 Joshua M. Boniface <joshua@boniface.me>
+#
+#    This program is free software: you can redistribute it and/or modify
+#    it under the terms of the GNU General Public License as published by
+#    the Free Software Foundation, version 3.
+#
+#    This program is distributed in the hope that it will be useful,
+#    but WITHOUT ANY WARRANTY; without even the implied warranty of
+#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#    GNU General Public License for more details.
+#
+#    You should have received a copy of the GNU General Public License
+#    along with this program.  If not, see <https://www.gnu.org/licenses/>.
+#
+###############################################################################
+
+import daemon_lib.common as common
+
+from time import sleep
+from os import makedirs
+
+
+def setup_sriov(logger, config):
+    logger.out('Setting up SR-IOV device support', state='i')
+
+    # Enable unsafe interrupts for the vfio_iommu_type1 kernel module
+    try:
+        common.run_os_command('modprobe vfio_iommu_type1 allow_unsafe_interrupts=1')
+        with open('/sys/module/vfio_iommu_type1/parameters/allow_unsafe_interrupts', 'w') as mfh:
+            mfh.write('Y')
+    except Exception:
+        logger.out('Failed to enable vfio_iommu_type1 kernel module; SR-IOV may fail', state='w')
+
+    # Loop through our SR-IOV NICs and enable the numvfs for each
+    for device in config['sriov_device']:
+        logger.out(f'Preparing SR-IOV PF {device["phy"]} with {device["vfcount"]} VFs', state='i')
+        try:
+            with open(f'/sys/class/net/{device["phy"]}/device/sriov_numvfs', 'r') as vfh:
+                current_vf_count = vfh.read().strip()
+            with open(f'/sys/class/net/{device["phy"]}/device/sriov_numvfs', 'w') as vfh:
+                vfh.write(str(device['vfcount']))
+        except FileNotFoundError:
+            logger.out(f'Failed to open SR-IOV configuration for PF {device["phy"]}; device may not support SR-IOV', state='w')
+        except OSError:
+            logger.out(f'Failed to set SR-IOV VF count for PF {device["phy"]} to {device["vfcount"]}; already set to {current_vf_count}', state='w')
+
+        if device.get('mtu', None) is not None:
+            logger.out(f'Setting SR-IOV PF {device["phy"]} to MTU {device["mtu"]}', state='i')
+            common.run_os_command(f'ip link set {device["phy"]} mtu {device["mtu"]} up')
+
+
+def setup_interfaces(logger, config):
+    # Set up the Cluster interface
+    cluster_dev = config['cluster_dev']
+    cluster_mtu = config['cluster_mtu']
+    cluster_dev_ip = config['cluster_dev_ip']
+
+    logger.out(f'Setting up Cluster network interface {cluster_dev} with MTU {cluster_mtu}', state='i')
+
+    common.run_os_command(f'ip link set {cluster_dev} mtu {cluster_mtu} up')
+
+    logger.out(f'Setting up Cluster network bridge on interface {cluster_dev} with IP {cluster_dev_ip}', state='i')
+
+    common.run_os_command(f'brctl addbr brcluster')
+    common.run_os_command(f'brctl addif brcluster {cluster_dev}')
+    common.run_os_command(f'ip link set brcluster mtu {cluster_mtu} up')
+    common.run_os_command(f'ip address add {cluster_dev_ip} dev brcluster')
+
+    # Set up the Storage interface
+    storage_dev = config['storage_dev']
+    storage_mtu = config['storage_mtu']
+    storage_dev_ip = config['storage_dev_ip']
+
+    logger.out(f'Setting up Storage network interface {storage_dev} with MTU {storage_mtu}', state='i')
+
+    common.run_os_command(f'ip link set {storage_dev} mtu {storage_mtu} up')
+
+    if storage_dev == cluster_dev:
+        if storage_dev_ip != cluster_dev_ip:
+            logger.out(f'Setting up Storage network on Cluster network bridge with IP {storage_dev_ip}', state='i')
+
+            common.run_os_command(f'ip address add {storage_dev_ip} dev brcluster')
+    else:
+        logger.out(f'Setting up Storage network bridge on interface {storage_dev} with IP {storage_dev_ip}', state='i')
+
+        common.run_os_command(f'brctl addbr brstorage')
+        common.run_os_command(f'brctl addif brstorage {storage_dev}')
+        common.run_os_command(f'ip link set brstorage mtu {storage_mtu} up')
+        common.run_os_command(f'ip address add {storage_dev_ip} dev brstorage')
+
+    # Set up the Upstream interface
+    upstream_dev = config['upstream_dev']
+    upstream_mtu = config['upstream_mtu']
+    upstream_dev_ip = config['upstream_dev_ip']
+
+    logger.out(f'Setting up Upstream network interface {upstream_dev} with MTU {upstream_mtu}', state='i')
+
+    if upstream_dev == cluster_dev:
+        if upstream_dev_ip != cluster_dev_ip:
+            logger.out(f'Setting up Upstream network on Cluster network bridge with IP {upstream_dev_ip}', state='i')
+
+            common.run_os_command(f'ip address add {upstream_dev_ip} dev brcluster')
+    else:
+        logger.out(f'Setting up Upstream network bridge on interface {upstream_dev} with IP {upstream_dev_ip}', state='i')
+
+        common.run_os_command(f'brctl addbr brupstream')
+        common.run_os_command(f'brctl addif brupstream {upstream_dev}')
+        common.run_os_command(f'ip link set brupstream mtu {upstream_mtu} up')
+        common.run_os_command(f'ip address add {upstream_dev_ip} dev brupstream')
+
+    upstream_gateway = config['upstream_gateway']
+    if upstream_gateway is not None:
+        logger.out(f'Setting up Upstream networok default gateway IP {upstream_gateway}', state='i')
+        if upstream_dev == cluster_dev:
+            common.run_os_command(f'ip route add default via {upstream_gateway} dev brcluster')
+        else:
+            common.run_os_command(f'ip route add default via {upstream_gateway} dev brupstream')
+
+    # Set up sysctl tweaks to optimize networking
+    # Enable routing functions
+    common.run_os_command('sysctl net.ipv4.ip_forward=1')
+    common.run_os_command('sysctl net.ipv6.ip_forward=1')
+    # Enable send redirects
+    common.run_os_command('sysctl net.ipv4.conf.all.send_redirects=1')
+    common.run_os_command('sysctl net.ipv4.conf.default.send_redirects=1')
+    common.run_os_command('sysctl net.ipv6.conf.all.send_redirects=1')
+    common.run_os_command('sysctl net.ipv6.conf.default.send_redirects=1')
+    # Accept source routes
+    common.run_os_command('sysctl net.ipv4.conf.all.accept_source_route=1')
+    common.run_os_command('sysctl net.ipv4.conf.default.accept_source_route=1')
+    common.run_os_command('sysctl net.ipv6.conf.all.accept_source_route=1')
+    common.run_os_command('sysctl net.ipv6.conf.default.accept_source_route=1')
+    # Disable RP filtering on Cluster and Upstream interfaces (to allow traffic pivoting)
+    common.run_os_command(f'sysctl net.ipv4.conf.{cluster_dev}.rp_filter=0')
+    common.run_os_command(f'sysctl net.ipv4.conf.brcluster.rp_filter=0')
+    common.run_os_command(f'sysctl net.ipv4.conf.{upstream_dev}.rp_filter=0')
+    common.run_os_command(f'sysctl net.ipv4.conf.brupstream.rp_filter=0')
+    common.run_os_command(f'sysctl net.ipv6.conf.{cluster_dev}.rp_filter=0')
+    common.run_os_command(f'sysctl net.ipv6.conf.brcluster.rp_filter=0')
+    common.run_os_command(f'sysctl net.ipv6.conf.{upstream_dev}.rp_filter=0')
+    common.run_os_command(f'sysctl net.ipv6.conf.brupstream.rp_filter=0')
+
+    # Stop DNSMasq if it is running
+    common.run_os_command('systemctl stop dnsmasq.service')
+
+    logger.out('Waiting 3 seconds for networking to come up', state='s')
+    sleep(3)
+
+
+def create_nft_configuration(logger, config):
+    if config['enable_networking']:
+        logger.out('Creating NFT firewall configuration', state='i')
+
+        dynamic_directory = config['nft_dynamic_directory']
+
+        # Create directories
+        makedirs(f'{dynamic_directory}/networks', exist_ok=True)
+        makedirs(f'{dynamic_directory}/static', exist_ok=True)
+
+        # Set up the base rules
+        nftables_base_rules = f"""# Base rules
+        flush ruleset
+        # Add the filter table and chains
+        add table inet filter
+        add chain inet filter forward {{ type filter hook forward priority 0; }}
+        add chain inet filter input {{ type filter hook input priority 0; }}
+        # Include static rules and network rules
+        include "{dynamic_directory}/static/*"
+        include "{dynamic_directory}/networks/*"
+        """
+
+        # Write the base firewall config
+        nftables_base_filename = f'{dynamic_directory}/base.nft'
+        with open(nftables_base_filename, 'w') as nftfh:
+            nftfh.write(nftables_base_rules)
+        common.reload_firewall_rules(nftables_base_filename, logger)

node-daemon/pvcnoded/util/services.py

@@ -0,0 +1,77 @@
+#!/usr/bin/env python3
+
+# services.py - Utility functions for pvcnoded external services
+# Part of the Parallel Virtual Cluster (PVC) system
+#
+#    Copyright (C) 2018-2021 Joshua M. Boniface <joshua@boniface.me>
+#
+#    This program is free software: you can redistribute it and/or modify
+#    it under the terms of the GNU General Public License as published by
+#    the Free Software Foundation, version 3.
+#
+#    This program is distributed in the hope that it will be useful,
+#    but WITHOUT ANY WARRANTY; without even the implied warranty of
+#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#    GNU General Public License for more details.
+#
+#    You should have received a copy of the GNU General Public License
+#    along with this program.  If not, see <https://www.gnu.org/licenses/>.
+#
+###############################################################################
+
+import daemon_lib.common as common
+from time import sleep
+
+
+def start_zookeeper(logger, config):
+    if config['daemon_mode'] == 'coordinator':
+        logger.out('Starting Zookeeper daemon', state='i')
+        # TODO: Move our handling out of Systemd and integrate it directly as a subprocess?
+        common.run_os_command('systemctl start zookeeper.service')
+
+
+def start_libvirtd(logger, config):
+    if config['enable_hypervisor']:
+        logger.out('Starting Libvirt daemon', state='i')
+        # TODO: Move our handling out of Systemd and integrate it directly as a subprocess?
+        common.run_os_command('systemctl start libvirtd.service')
+
+
+def start_patroni(logger, config):
+    if config['enable_networking'] and config['daemon_mode'] == 'coordinator':
+        logger.out('Starting Patroni daemon', state='i')
+        # TODO: Move our handling out of Systemd and integrate it directly as a subprocess?
+        common.run_os_command('systemctl start patroni.service')
+
+
+def start_frrouting(logger, config):
+    if config['enable_networking'] and config['daemon_mode'] == 'coordinator':
+        logger.out('Starting FRRouting daemon', state='i')
+        # TODO: Move our handling out of Systemd and integrate it directly as a subprocess?
+        common.run_os_command('systemctl start frr.service')
+
+
+def start_ceph_mon(logger, config):
+    if config['enable_storage'] and config['daemon_mode'] == 'coordinator':
+        logger.out('Starting Ceph Monitor daemon', state='i')
+        # TODO: Move our handling out of Systemd and integrate it directly as a subprocess?
+        common.run_os_command(f'systemctl start ceph-mon@{config["node_hostname"]}.service')
+
+
+def start_ceph_mgr(logger, config):
+    if config['enable_storage'] and config['daemon_mode'] == 'coordinator':
+        logger.out('Starting Ceph Manager daemon', state='i')
+        # TODO: Move our handling out of Systemd and integrate it directly as a subprocess?
+        common.run_os_command(f'systemctl start ceph-mgr@{config["node_hostname"]}.service')
+
+
+def start_system_services(logger, config):
+    start_zookeeper(logger, config)
+    start_libvirtd(logger, config)
+    start_patroni(logger, config)
+    start_frrouting(logger, config)
+    start_ceph_mon(logger, config)
+    start_ceph_mgr(logger, config)
+
+    logger.out('Waiting 3 seconds for daemons to start', state='s')
+    sleep(3)

node-daemon/pvcnoded/util/zookeeper.py

@@ -0,0 +1,132 @@
+#!/usr/bin/env python3
+
+# <Filename> - <Description>
+# zookeeper.py - Utility functions for pvcnoded Zookeeper connections
+# Part of the Parallel Virtual Cluster (PVC) system
+#
+#    Copyright (C) 2018-2021 Joshua M. Boniface <joshua@boniface.me>
+#
+#    This program is free software: you can redistribute it and/or modify
+#    it under the terms of the GNU General Public License as published by
+#    the Free Software Foundation, version 3.
+#
+#    This program is distributed in the hope that it will be useful,
+#    but WITHOUT ANY WARRANTY; without even the implied warranty of
+#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#    GNU General Public License for more details.
+#
+#    You should have received a copy of the GNU General Public License
+#    along with this program.  If not, see <https://www.gnu.org/licenses/>.
+#
+##############################################################################
+
+from daemon_lib.zkhandler import ZKHandler
+
+import os
+import time
+
+
+def connect(logger, config):
+    # Create an instance of the handler
+    zkhandler = ZKHandler(config, logger)
+
+    try:
+        logger.out('Connecting to Zookeeper on coordinator nodes {}'.format(config['coordinators']), state='i')
+        # Start connection
+        zkhandler.connect(persistent=True)
+    except Exception as e:
+        logger.out('ERROR: Failed to connect to Zookeeper cluster: {}'.format(e), state='e')
+        os._exit(1)
+
+    logger.out('Validating Zookeeper schema', state='i')
+
+    try:
+        node_schema_version = int(zkhandler.read(('node.data.active_schema', config['node_hostname'])))
+    except Exception:
+        node_schema_version = int(zkhandler.read('base.schema.version'))
+        zkhandler.write([
+            (('node.data.active_schema', config['node_hostname']), node_schema_version)
+        ])
+
+    # Load in the current node schema version
+    zkhandler.schema.load(node_schema_version)
+
+    # Record the latest intalled schema version
+    latest_schema_version = zkhandler.schema.find_latest()
+    logger.out('Latest installed schema is {}'.format(latest_schema_version), state='i')
+    zkhandler.write([
+        (('node.data.latest_schema', config['node_hostname']), latest_schema_version)
+    ])
+
+    # If we are the last node to get a schema update, fire the master update
+    if latest_schema_version > node_schema_version:
+        node_latest_schema_version = list()
+        for node in zkhandler.children('base.node'):
+            node_latest_schema_version.append(int(zkhandler.read(('node.data.latest_schema', node))))
+
+        # This is true if all elements of the latest schema version are identical to the latest version,
+        # i.e. they have all had the latest schema installed and ready to load.
+        if node_latest_schema_version.count(latest_schema_version) == len(node_latest_schema_version):
+            zkhandler.write([
+                ('base.schema.version', latest_schema_version)
+            ])
+
+    return zkhandler, node_schema_version
+
+
+def validate_schema(logger, zkhandler):
+    # Validate our schema against the active version
+    if not zkhandler.schema.validate(zkhandler, logger):
+        logger.out('Found schema violations, applying', state='i')
+        zkhandler.schema.apply(zkhandler)
+    else:
+        logger.out('Schema successfully validated', state='o')
+
+
+def setup_node(logger, config, zkhandler):
+    # Check if our node exists in Zookeeper, and create it if not
+    if config['daemon_mode'] == 'coordinator':
+        init_routerstate = 'secondary'
+    else:
+        init_routerstate = 'client'
+
+    if zkhandler.exists(('node', config['node_hostname'])):
+        logger.out(f'Node is {logger.fmt_green}present{logger.fmt_end} in Zookeeper', state='i')
+        # Update static data just in case it's changed
+        zkhandler.write([
+            (('node', config['node_hostname']), config['daemon_mode']),
+            (('node.mode', config['node_hostname']), config['daemon_mode']),
+            (('node.state.daemon', config['node_hostname']), 'init'),
+            (('node.state.router', config['node_hostname']), init_routerstate),
+            (('node.data.static', config['node_hostname']), ' '.join(config['static_data'])),
+            (('node.data.pvc_version', config['node_hostname']), config['pvcnoded_version']),
+            (('node.ipmi.hostname', config['node_hostname']), config['ipmi_hostname']),
+            (('node.ipmi.username', config['node_hostname']), config['ipmi_username']),
+            (('node.ipmi.password', config['node_hostname']), config['ipmi_password']),
+        ])
+    else:
+        logger.out(f'Node is {logger.fmt_red}absent{logger.fmt_end} in Zookeeper; adding new node', state='i')
+        keepalive_time = int(time.time())
+        zkhandler.write([
+            (('node', config['node_hostname']), config['daemon_mode']),
+            (('node.keepalive', config['node_hostname']), str(keepalive_time)),
+            (('node.mode', config['node_hostname']), config['daemon_mode']),
+            (('node.state.daemon', config['node_hostname']), 'init'),
+            (('node.state.domain', config['node_hostname']), 'flushed'),
+            (('node.state.router', config['node_hostname']), init_routerstate),
+            (('node.data.static', config['node_hostname']), ' '.join(config['static_data'])),
+            (('node.data.pvc_version', config['node_hostname']), config['pvcnoded_version']),
+            (('node.ipmi.hostname', config['node_hostname']), config['ipmi_hostname']),
+            (('node.ipmi.username', config['node_hostname']), config['ipmi_username']),
+            (('node.ipmi.password', config['node_hostname']), config['ipmi_password']),
+            (('node.memory.total', config['node_hostname']), '0'),
+            (('node.memory.used', config['node_hostname']), '0'),
+            (('node.memory.free', config['node_hostname']), '0'),
+            (('node.memory.allocated', config['node_hostname']), '0'),
+            (('node.memory.provisioned', config['node_hostname']), '0'),
+            (('node.vcpu.allocated', config['node_hostname']), '0'),
+            (('node.cpu.load', config['node_hostname']), '0.0'),
+            (('node.running_domains', config['node_hostname']), '0'),
+            (('node.count.provisioned_domains', config['node_hostname']), '0'),
+            (('node.count.networks', config['node_hostname']), '0'),
+        ])