Debianize the packaging

pvcd/NodeInstance.py

@@ -0,0 +1,364 @@
+#!/usr/bin/env python3
+
+# NodeInstance.py - Class implementing a PVC node and run by pvcd
+# Part of the Parallel Virtual Cluster (PVC) system
+#
+#    Copyright (C) 2018  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, either version 3 of the License, or
+#    (at your option) any later version.
+#
+#    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, sys, psutil, socket, time, libvirt, kazoo.client, threading, subprocess
+import pvcd.ansiiprint as ansiiprint
+
+class NodeInstance():
+    # Initialization function
+    def __init__(self, this_node, name, t_node, s_domain, zk, config):
+        # Passed-in variables on creation
+        self.zk = zk
+        self.config = config
+        self.this_node = this_node
+        self.name = name
+        self.daemon_state = 'stop'
+        self.domain_state = 'ready'
+        self.t_node = t_node
+        self.active_node_list = []
+        self.flushed_node_list = []
+        self.inactive_node_list = []
+        self.s_domain = s_domain
+        self.domain_list = []
+        self.ipmi_hostname = self.config['ipmi_hostname']
+        self.domains_count = 0
+        self.memused = 0
+        self.memfree = 0
+        self.inflush = False
+
+        # Zookeeper handlers for changed states
+        @zk.DataWatch('/nodes/{}/daemonstate'.format(self.name))
+        def watch_hypervisor_daemonstate(data, stat, event=""):
+            try:
+                self.daemon_state = data.decode('ascii')
+            except AttributeError:
+                self.daemon_state = 'stop'
+
+        @zk.DataWatch('/nodes/{}/domainstate'.format(self.name))
+        def watch_hypervisor_domainstate(data, stat, event=""):
+            try:
+                self.domain_state = data.decode('ascii')
+            except AttributeError:
+                self.domain_state = 'unknown'
+
+            # toggle state management of this node
+            if self.name == self.this_node:
+                if self.domain_state == 'flush' and self.inflush == False:
+                    self.flush()
+                if self.domain_state == 'unflush' and self.inflush == False:
+                    self.unflush()
+
+
+        @zk.DataWatch('/nodes/{}/memfree'.format(self.name))
+        def watch_hypervisor_memfree(data, stat, event=""):
+            try:
+                self.memfree = data.decode('ascii')
+            except AttributeError:
+                self.memfree = 0
+    
+        @zk.DataWatch('/nodes/{}/memused'.format(self.name))
+        def watch_hypervisor_memused(data, stat, event=""):
+            try:
+                self.memused = data.decode('ascii')
+            except AttributeError:
+                self.memused = 0
+    
+        @zk.DataWatch('/nodes/{}/runningdomains'.format(self.name))
+        def watch_hypervisor_runningdomains(data, stat, event=""):
+            try:
+                self.domain_list = data.decode('ascii').split()
+            except AttributeError:
+                self.domain_list = []
+
+        @zk.DataWatch('/nodes/{}/domainscount'.format(self.name))
+        def watch_hypervisor_domainscount(data, stat, event=""):
+            try:
+                self.domains_count = data.decode('ascii')
+            except AttributeError:
+                self.domains_count = 0
+    
+    # Get value functions
+    def getfreemem(self):
+        return self.memfree
+
+    def getcpuload(self):
+        return self.cpuload
+
+    def getname(self):
+        return self.name
+
+    def getdaemonstate(self):
+        return self.daemon_state
+
+    def getdomainstate(self):
+        return self.domain_state
+
+    def getdomainlist(self):
+        return self.domain_list
+
+    # Update value functions
+    def updatenodelist(self, t_node):
+        self.t_node = t_node
+
+    def updatedomainlist(self, s_domain):
+        self.s_domain = s_domain
+
+    # Flush all VMs on the host
+    def flush(self):
+        self.inflush = True
+        ansiiprint.echo('Flushing node "{}" of running VMs'.format(self.name), '', 'i')
+        ansiiprint.echo('Domain list: {}'.format(', '.join(self.domain_list)), '', 'c')
+        for dom_uuid in self.domain_list:
+            most_memfree = 0
+            target_hypervisor = None
+            hypervisor_list = self.zk.get_children('/nodes')
+            current_hypervisor = self.zk.get('/domains/{}/hypervisor'.format(dom_uuid))[0].decode('ascii')
+            if current_hypervisor != self.this_node:
+                continue
+
+            for hypervisor in hypervisor_list:
+                daemon_state = self.zk.get('/nodes/{}/daemonstate'.format(hypervisor))[0].decode('ascii')
+                domain_state = self.zk.get('/nodes/{}/domainstate'.format(hypervisor))[0].decode('ascii')
+                if hypervisor == current_hypervisor:
+                    continue
+
+                if daemon_state != 'run' or domain_state != 'ready':
+                    continue
+    
+                memfree = int(self.zk.get('/nodes/{}/memfree'.format(hypervisor))[0].decode('ascii'))
+                if memfree > most_memfree:
+                    most_memfree = memfree
+                    target_hypervisor = hypervisor
+
+            if target_hypervisor == None:
+                ansiiprint.echo('Failed to find migration target for VM "{}"; shutting down'.format(dom_uuid), '', 'e')
+                transaction = self.zk.transaction()
+                transaction.set_data('/domains/{}/state'.format(dom_uuid), 'shutdown'.encode('ascii'))
+                transaction.commit()
+            else:
+                ansiiprint.echo('Migrating VM "{}" to hypervisor "{}"'.format(dom_uuid, target_hypervisor), '', 'i')
+                transaction = self.zk.transaction()
+                transaction.set_data('/domains/{}/state'.format(dom_uuid), 'migrate'.encode('ascii'))
+                transaction.set_data('/domains/{}/hypervisor'.format(dom_uuid), target_hypervisor.encode('ascii'))
+                transaction.set_data('/domains/{}/lasthypervisor'.format(dom_uuid), current_hypervisor.encode('ascii'))
+                transaction.commit()
+
+        self.zk.set('/nodes/{}/runningdomains'.format(self.name), ''.encode('ascii'))
+        self.zk.set('/nodes/{}/domainstate'.format(self.name), 'flushed'.encode('ascii'))
+        self.inflush = False
+
+    def unflush(self):
+        self.inflush = True
+        ansiiprint.echo('Restoring node {} to active service.'.format(self.name), '', 'i')
+        self.zk.set('/nodes/{}/domainstate'.format(self.name), 'ready'.encode('ascii'))
+        for dom_uuid in self.s_domain:
+            last_hypervisor = self.zk.get('/domains/{}/lasthypervisor'.format(dom_uuid))[0].decode('ascii')
+            if last_hypervisor != self.name:
+                continue
+
+            ansiiprint.echo('Setting unmigration for VM "{}"'.format(dom_uuid), '', 'i')
+            transaction = self.zk.transaction()
+            transaction.set_data('/domains/{}/state'.format(dom_uuid), 'migrate'.encode('ascii'))
+            transaction.set_data('/domains/{}/hypervisor'.format(dom_uuid), self.name.encode('ascii'))
+            transaction.set_data('/domains/{}/lasthypervisor'.format(dom_uuid), ''.encode('ascii'))
+            transaction.commit()
+
+        self.inflush = False
+
+    def update_zookeeper(self):
+        # Connect to libvirt
+        libvirt_name = "qemu:///system"
+        conn = libvirt.open(libvirt_name)
+        if conn == None:
+            ansiiprint.echo('Failed to open connection to "{}"'.format(libvirt_name), '', 'e')
+            return
+
+        # Get past state and update if needed
+        past_state = self.zk.get('/nodes/{}/daemonstate'.format(self.name))[0].decode('ascii')
+        if past_state != 'run':
+            self.daemon_state = 'run'
+            self.zk.set('/nodes/{}/daemonstate'.format(self.name), 'run'.encode('ascii'))
+        else:
+            self.daemon_state = 'run'
+
+        # Toggle state management of dead VMs to restart them
+        for domain, instance in self.s_domain.items():
+            if instance.inshutdown == False and domain in self.domain_list:
+                if instance.getstate() == 'start' and instance.gethypervisor() == self.name:
+                    if instance.getdom() != None:
+                        try:
+                            if instance.getdom().state()[0] != libvirt.VIR_DOMAIN_RUNNING:
+                                raise
+                        except Exception as e:
+                            # Toggle a state "change"
+                            self.zk.set('/domains/{}/state'.format(domain), instance.getstate().encode('ascii'))
+
+        # Set our information in zookeeper
+        self.name = conn.getHostname()
+        self.memused = int(psutil.virtual_memory().used / 1024 / 1024)
+        self.memfree = int(psutil.virtual_memory().free / 1024 / 1024)
+        self.cpuload = os.getloadavg()[0]
+        self.domains_count = len(conn.listDomainsID())
+        keepalive_time = int(time.time())
+        try:
+            transaction = self.zk.transaction()
+            transaction.set_data('/nodes/{}/memused'.format(self.name), str(self.memused).encode('ascii'))
+            transaction.set_data('/nodes/{}/memfree'.format(self.name), str(self.memfree).encode('ascii'))
+            transaction.set_data('/nodes/{}/cpuload'.format(self.name), str(self.cpuload).encode('ascii'))
+            transaction.set_data('/nodes/{}/runningdomains'.format(self.name), ' '.join(self.domain_list).encode('ascii'))
+            transaction.set_data('/nodes/{}/domainscount'.format(self.name), str(self.domains_count).encode('ascii'))
+            transaction.set_data('/nodes/{}/keepalive'.format(self.name), str(keepalive_time).encode('ascii'))
+            transaction.commit()
+        except:
+            return
+
+        # Close the Libvirt connection
+        conn.close()
+
+        # Display node information to the terminal
+        ansiiprint.echo('{}{} keepalive{}'.format(ansiiprint.purple(), self.name, ansiiprint.end()), '', 't')
+        ansiiprint.echo('{0}Active domains:{1} {2}  {0}Free memory [MiB]:{1} {3}  {0}Used memory [MiB]:{1} {4}  {0}Load:{1} {5}'.format(ansiiprint.bold(), ansiiprint.end(), self.domains_count, self.memfree, self.memused, self.cpuload), '', 'c')
+
+        # Update our local node lists
+        for node_name in self.t_node:
+            try:
+                node_daemon_state = self.zk.get('/nodes/{}/daemonstate'.format(node_name))[0].decode('ascii')
+                node_domain_state = self.zk.get('/nodes/{}/domainstate'.format(node_name))[0].decode('ascii')
+                node_keepalive = int(self.zk.get('/nodes/{}/keepalive'.format(node_name))[0].decode('ascii'))
+            except:
+                node_daemon_state = 'unknown'
+                node_domain_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(self.config['keepalive_interval']) * 6 )
+            if node_keepalive < node_deadtime and node_daemon_state == 'run':
+                ansiiprint.echo('Node {} seems dead - starting monitor for fencing'.format(node_name), '', 'w')
+                self.zk.set('/nodes/{}/daemonstate'.format(node_name), 'dead'.encode('ascii'))
+                fence_thread = threading.Thread(target=fenceNode, args=(node_name, self.zk), kwargs={})
+                fence_thread.start()
+
+            # Update the arrays
+            if node_daemon_state == 'run' and node_domain_state != 'flushed' and node_name not in self.active_node_list:
+                self.active_node_list.append(node_name)
+                try:
+                    self.flushed_node_list.remove(node_name)
+                except ValueError:
+                    pass
+                try:
+                    self.inactive_node_list.remove(node_name)
+                except ValueError:
+                    pass
+            if node_daemon_state != 'run' and node_domain_state != 'flushed' and node_name not in self.inactive_node_list:
+                self.inactive_node_list.append(node_name)
+                try:
+                    self.active_node_list.remove(node_name)
+                except ValueError:
+                    pass
+                try:
+                    self.flushed_node_list.remove(node_name)
+                except ValueError:
+                    pass
+            if node_domain_state == 'flushed' and node_name not in self.flushed_node_list:
+                self.flushed_node_list.append(node_name)
+                try:
+                    self.active_node_list.remove(node_name)
+                except ValueError:
+                    pass
+                try:
+                    self.inactive_node_list.remove(node_name)
+                except ValueError:
+                    pass
+        
+        # Display cluster information to the terminal
+        ansiiprint.echo('{}Cluster status{}'.format(ansiiprint.purple(), ansiiprint.end()), '', 't')
+        ansiiprint.echo('{}Active nodes:{} {}'.format(ansiiprint.bold(), ansiiprint.end(), ' '.join(self.active_node_list)), '', 'c')
+        ansiiprint.echo('{}Inactive nodes:{} {}'.format(ansiiprint.bold(), ansiiprint.end(), ' '.join(self.inactive_node_list)), '', 'c')
+        ansiiprint.echo('{}Flushed nodes:{} {}'.format(ansiiprint.bold(), ansiiprint.end(), ' '.join(self.flushed_node_list)), '', 'c')
+
+#
+# Fence thread entry function
+#
+def fenceNode(node_name, zk):
+    failcount = 0
+    while failcount < 3:
+        # Wait 5 seconds
+        time.sleep(5)
+        # Get the state
+        node_daemon_state = zk.get('/nodes/{}/daemonstate'.format(node_name))[0].decode('ascii')
+        # Is it still 'dead'
+        if node_daemon_state == 'dead':
+            failcount += 1
+            ansiiprint.echo('Node "{}" failed {} saving throws'.format(node_name, failcount), '', 'w')
+        # It changed back to something else so it must be alive
+        else:
+            ansiiprint.echo('Node "{}" passed a saving throw; canceling fence'.format(node_name), '', 'o')
+            return
+
+    ansiiprint.echo('Fencing node "{}" via IPMI reboot signal'.format(node_name), '', 'e')
+
+    ipmi_hostname = zk.get('/nodes/{}/ipmihostname'.format(node_name))[0].decode('ascii')
+    ipmi_username = zk.get('/nodes/{}/ipmiusername'.format(node_name))[0].decode('ascii')
+    ipmi_password = zk.get('/nodes/{}/ipmipassword'.format(node_name))[0].decode('ascii')
+    rebootViaIPMI(ipmi_hostname, ipmi_username, ipmi_password)
+    time.sleep(5)
+
+    ansiiprint.echo('Moving VMs from dead hypervisor "{}" to new hosts'.format(node_name), '', 'i')
+    dead_node_running_domains = zk.get('/nodes/{}/runningdomains'.format(node_name))[0].decode('ascii').split()
+    for dom_uuid in dead_node_running_domains:
+        most_memfree = 0
+        hypervisor_list = zk.get_children('/nodes')
+        current_hypervisor = zk.get('/domains/{}/hypervisor'.format(dom_uuid))[0].decode('ascii')
+        for hypervisor in hypervisor_list:
+            print(hypervisor)
+            daemon_state = zk.get('/nodes/{}/daemonstate'.format(hypervisor))[0].decode('ascii')
+            domain_state = zk.get('/nodes/{}/domainstate'.format(hypervisor))[0].decode('ascii')
+            if daemon_state != 'run' or domain_state != 'ready':
+                continue
+
+            memfree = int(zk.get('/nodes/{}/memfree'.format(hypervisor))[0].decode('ascii'))
+            if memfree > most_memfree:
+                most_memfree = memfree
+                target_hypervisor = hypervisor
+
+        ansiiprint.echo('Moving VM "{}" to hypervisor "{}"'.format(dom_uuid, target_hypervisor), '', 'i')
+        transaction = zk.transaction()
+        transaction.set_data('/domains/{}/state'.format(dom_uuid), 'start'.encode('ascii'))
+        transaction.set_data('/domains/{}/hypervisor'.format(dom_uuid), target_hypervisor.encode('ascii'))
+        transaction.set_data('/domains/{}/lasthypervisor'.format(dom_uuid), current_hypervisor.encode('ascii'))
+        transaction.commit()
+
+    # Set node in flushed state for easy remigrating when it comes back
+    zk.set('/nodes/{}/domainstate'.format(node_name), 'flushed'.encode('ascii'))
+
+#
+# Perform an IPMI fence
+#
+def rebootViaIPMI(ipmi_hostname, ipmi_user, ipmi_password):
+    ipmi_command = ['/usr/bin/ipmitool', '-H', ipmi_hostname, '-U', ipmi_user, '-P', ipmi_password, 'chassis', 'power', 'reset']
+    ipmi_command_output = subprocess.run(ipmi_command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    if ipmi_command_output == 0:
+        ansiiprint.echo('Successfully rebooted dead node', '', 'o')
+    else:
+        ansiiprint.echo('Failed to reboot dead node', '', 'e')

pvcd/VMInstance.py

@@ -0,0 +1,389 @@
+#!/usr/bin/env python3
+
+# VMInstance.py - Class implementing a PVC virtual machine and run by pvcd
+# Part of the Parallel Virtual Cluster (PVC) system
+#
+#    Copyright (C) 2018  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, either version 3 of the License, or
+#    (at your option) any later version.
+#
+#    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, sys, uuid, socket, time, threading, libvirt, kazoo.client
+import pvcd.ansiiprint as ansiiprint
+
+class VMInstance:
+    # Initialization function
+    def __init__(self, domuuid, zk, config, thishypervisor):
+        # Passed-in variables on creation
+        self.domuuid = domuuid
+        self.zk = zk
+        self.config = config
+        self.thishypervisor = thishypervisor
+
+        # These will all be set later
+        self.hypervisor = None
+        self.state = None
+        self.instart = False
+        self.inrestart = False
+        self.inmigrate = False
+        self.inreceive = False
+        self.inshutdown = False
+        self.instop = False
+
+        self.dom = self.lookupByUUID(self.domuuid)
+
+        # Watch for changes to the state field in Zookeeper
+        @zk.DataWatch('/domains/{}/state'.format(self.domuuid))
+        def watch_state(data, stat, event=""):
+            # If we get a delete state, just terminate outselves
+            if data == None:
+                return
+            # Otherwise perform a management command
+            else:
+                self.manage_vm_state()
+
+    # Get data functions
+    def getstate(self):
+        return self.state
+
+    def gethypervisor(self):
+        return self.hypervisor
+
+    def getdom(self):
+        return self.dom
+
+    # Start up the VM
+    def start_vm(self):
+        ansiiprint.echo('Starting VM', '{}:'.format(self.domuuid), 'i')
+        self.instart = True
+
+        # Start up a new Libvirt connection
+        libvirt_name = "qemu:///system"
+        conn = libvirt.open(libvirt_name)
+        if conn == None:
+            ansiiprint.echo('Failed to open local libvirt connection', '{}:'.format(self.domuuid), 'e')
+            self.instart = False
+            return
+    
+        try:
+            # Grab the domain information from Zookeeper
+            xmlconfig = self.zk.get('/domains/{}/xml'.format(self.domuuid))[0].decode('ascii')
+            dom = conn.createXML(xmlconfig, 0)
+            if not self.domuuid in self.thishypervisor.domain_list:
+                self.thishypervisor.domain_list.append(self.domuuid)
+
+            ansiiprint.echo('Successfully started VM', '{}:'.format(self.domuuid), 'o')
+            self.dom = dom
+        except libvirt.libvirtError as e:
+            ansiiprint.echo('Failed to create VM', '{}:'.format(self.domuuid), 'e')
+            self.zk.set('/domains/{}/state'.format(self.domuuid), 'failed'.encode('ascii'))
+            self.dom = None
+
+        conn.close()
+        self.instart = False
+  
+    # Restart the VM
+    def restart_vm(self):
+        ansiiprint.echo('Restarting VM', '{}:'.format(self.domuuid), 'i')
+        self.inrestart = True
+
+        # Start up a new Libvirt connection
+        libvirt_name = "qemu:///system"
+        conn = libvirt.open(libvirt_name)
+        if conn == None:
+            ansiiprint.echo('Failed to open local libvirt connection', '{}:'.format(self.domuuid), 'e')
+            self.inrestart = False
+            return
+    
+        try:
+            self.shutdown_vm()
+            self.start_vm()
+            ansiiprint.echo('Successfully restarted VM', '{}:'.format(self.domuuid), 'o')
+        except libvirt.libvirtError as e:
+            ansiiprint.echo('Failed to restart VM', '{}:'.format(self.domuuid), 'e')
+
+        self.zk.set('/domains/{}/state'.format(self.domuuid), 'start'.encode('ascii'))
+        conn.close()
+        self.inrestart = False
+
+    # Stop the VM forcibly without updating state
+    def terminate_vm(self):
+        ansiiprint.echo('Terminating VM', '{}:'.format(self.domuuid), 'i')
+        self.instop = True
+        try:
+            self.dom.destroy()
+        except AttributeError:
+            ansiiprint.echo('Failed to terminate VM', '{}:'.format(self.domuuid), 'e')
+        if self.domuuid in self.thishypervisor.domain_list:
+            try:
+                self.thishypervisor.domain_list.remove(self.domuuid)
+            except ValueError:
+                pass
+        ansiiprint.echo('Successfully terminated VM', '{}:'.format(self.domuuid), 'o')
+        self.dom = None
+        self.instop = False
+
+    # Stop the VM forcibly
+    def stop_vm(self):
+        ansiiprint.echo('Forcibly stopping VM', '{}:'.format(self.domuuid), 'i')
+        self.instop = True
+        try:
+            self.dom.destroy()
+        except AttributeError:
+            ansiiprint.echo('Failed to stop VM', '{}:'.format(self.domuuid), 'e')
+        if self.domuuid in self.thishypervisor.domain_list:
+            try:
+                self.thishypervisor.domain_list.remove(self.domuuid)
+            except ValueError:
+                pass
+
+        if self.inrestart == False:
+            self.zk.set('/domains/{}/state'.format(self.domuuid), 'stop'.encode('ascii'))
+
+        ansiiprint.echo('Successfully stopped VM', '{}:'.format(self.domuuid), 'o')
+        self.dom = None
+        self.instop = False
+    
+    # Shutdown the VM gracefully
+    def shutdown_vm(self):
+        ansiiprint.echo('Gracefully stopping VM', '{}:'.format(self.domuuid), 'i')
+        self.inshutdown = True
+        self.dom.shutdown()
+        try:
+            tick = 0
+            while self.dom.state()[0] == libvirt.VIR_DOMAIN_RUNNING and tick < 60:
+                tick += 1
+                time.sleep(0.5)
+
+            if tick >= 60:
+                ansiiprint.echo('Shutdown timeout expired', '{}:'.format(self.domuuid), 'e')
+                self.stop_vm()
+                self.inshutdown = False
+                return
+        except:
+            pass
+
+        if self.domuuid in self.thishypervisor.domain_list:
+            try:
+                self.thishypervisor.domain_list.remove(self.domuuid)
+            except ValueError:
+                pass
+
+        if self.inrestart == False:
+            self.zk.set('/domains/{}/state'.format(self.domuuid), 'stop'.encode('ascii'))
+
+        ansiiprint.echo('Successfully shutdown VM', '{}:'.format(self.domuuid), 'o')
+        self.dom = None
+        self.inshutdown = False
+
+    def live_migrate_vm(self, dest_hypervisor):
+        try:
+            dest_conn = libvirt.open('qemu+tcp://{}/system'.format(self.hypervisor))
+            if dest_conn == None:
+                raise
+        except:
+            ansiiprint.echo('Failed to open connection to qemu+tcp://{}/system; aborting migration.'.format(self.hypervisor), '{}:'.format(self.domuuid), 'e')
+            return 1
+
+        try:
+            target_dom = self.dom.migrate(dest_conn, libvirt.VIR_MIGRATE_LIVE, None, None, 0)
+            if target_dom == None:
+                raise
+            ansiiprint.echo('Successfully migrated VM', '{}:'.format(self.domuuid), 'o')
+
+        except:
+            dest_conn.close()
+            return 1
+
+        dest_conn.close()
+        return 0
+
+    # Migrate the VM to a target host
+    def migrate_vm(self):
+        self.inmigrate = True
+        ansiiprint.echo('Migrating VM to hypervisor "{}"'.format(self.hypervisor), '{}:'.format(self.domuuid), 'i')
+        migrate_ret = self.live_migrate_vm(self.hypervisor)
+        if migrate_ret != 0:
+            ansiiprint.echo('Could not live migrate VM; shutting down to migrate instead', '{}:'.format(self.domuuid), 'e')
+            self.shutdown_vm()
+            time.sleep(1)
+            self.zk.set('/domains/{}/state'.format(self.domuuid), 'start'.encode('ascii'))
+        else:
+            try:
+                self.thishypervisor.domain_list.remove(self.domuuid)
+            except ValueError:
+                pass
+            time.sleep(1)
+            self.zk.set('/domains/{}/state'.format(self.domuuid), 'start'.encode('ascii'))
+
+        self.inmigrate = False
+
+    # Receive the migration from another host (wait until VM is running)
+    def receive_migrate(self):
+        self.inreceive = True
+        ansiiprint.echo('Receiving migration', '{}:'.format(self.domuuid), 'i')
+        while True:
+            time.sleep(0.5)
+            self.state = self.zk.get('/domains/{}/state'.format(self.domuuid))[0].decode('ascii')
+            self.dom = self.lookupByUUID(self.domuuid)
+
+            if self.dom == None and self.state == 'migrate':
+                continue
+
+            if self.state != 'migrate':
+                break
+
+            try:
+                if self.dom.state()[0] == libvirt.VIR_DOMAIN_RUNNING:
+                    break
+            except:
+                continue
+
+        if self.dom.state()[0] == libvirt.VIR_DOMAIN_RUNNING:
+            if not self.domuuid in self.thishypervisor.domain_list:
+                self.thishypervisor.domain_list.append(self.domuuid)
+            ansiiprint.echo('Successfully received migrated VM', '{}:'.format(self.domuuid), 'o')
+            self.zk.set('/domains/{}/state'.format(self.domuuid), 'start'.encode('ascii'))
+        else:
+            ansiiprint.echo('Failed to receive migrated VM', '{}:'.format(self.domuuid), 'e')
+            self.zk.set('/domains/{}/state'.format(self.domuuid), 'start'.encode('ascii'))
+
+        self.inreceive = False
+
+    #
+    # Main function to manage a VM (taking only self)
+    #
+    def manage_vm_state(self):
+        # Give ourselves a bit of leeway time
+        time.sleep(0.2)
+
+        # Get the current values from zookeeper (don't rely on the watch)
+        self.state = self.zk.get('/domains/{}/state'.format(self.domuuid))[0].decode('ascii')
+        self.hypervisor = self.zk.get('/domains/{}/hypervisor'.format(self.domuuid))[0].decode('ascii')
+
+        # Check the current state of the VM
+        try:
+            if self.dom != None:
+                running, reason = self.dom.state()
+            else:
+                raise
+        except:
+            running = libvirt.VIR_DOMAIN_NOSTATE
+
+        ansiiprint.echo('VM state change for "{}": {} {}'.format(self.domuuid, self.state, self.hypervisor), '', 'i')
+
+        #######################
+        # Handle state changes
+        #######################
+        # Valid states are:
+        #   start
+        #   migrate
+        #   restart
+        #   shutdown
+        #   stop
+
+        # Conditional pass one - Are we already performing an action
+        if self.instart == False \
+        and self.inrestart == False \
+        and self.inmigrate == False \
+        and self.inreceive == False \
+        and self.inshutdown == False \
+        and self.instop == False:
+            # Conditional pass two - Is this VM configured to run on this hypervisor
+            if self.hypervisor == self.thishypervisor.name:
+                # Conditional pass three - Is this VM currently running on this hypervisor
+                if running == libvirt.VIR_DOMAIN_RUNNING:
+                    # VM is already running and should be
+                    if self.state == "start":
+                        if not self.domuuid in self.thishypervisor.domain_list:
+                            self.thishypervisor.domain_list.append(self.domuuid)
+                    # VM is already running and should be but stuck in migrate state
+                    elif self.state == "migrate":
+                        self.zk.set('/domains/{}/state'.format(self.domuuid), 'start'.encode('ascii'))
+                        if not self.domuuid in self.thishypervisor.domain_list:
+                            self.thishypervisor.domain_list.append(self.domuuid)
+                    # VM should be restarted
+                    elif self.state == "restart":
+                        self.restart_vm()
+                    # VM should be shut down
+                    elif self.state == "shutdown":
+                        self.shutdown_vm()
+                    # VM should be stopped
+                    elif self.state == "stop":
+                        self.stop_vm()
+                else:
+                    # VM should be started
+                    if self.state == "start":
+                        self.start_vm()
+                    # VM should be migrated to this hypervisor
+                    elif self.state == "migrate":
+                        self.receive_migrate()
+                    # VM should be restarted (i.e. started since it isn't running)
+                    if self.state == "restart":
+                        self.zk.set('/domains/{}/state'.format(self.domuuid), 'start'.encode('ascii'))
+                    # VM should be shut down; ensure it's gone from this node's domain_list
+                    elif self.state == "shutdown":
+                        if self.domuuid in self.thishypervisor.domain_list:
+                            self.thishypervisor.domain_list.remove(self.domuuid)
+                    # VM should be stoped; ensure it's gone from this node's domain_list
+                    elif self.state == "stop":
+                        if self.domuuid in self.thishypervisor.domain_list:
+                            self.thishypervisor.domain_list.remove(self.domuuid)
+                        
+            else:
+                # Conditional pass three - Is this VM currently running on this hypervisor
+                if running == libvirt.VIR_DOMAIN_RUNNING:
+                    # VM should be migrated away from this hypervisor
+                    if self.state == "migrate":
+                        self.migrate_vm()
+                    # VM should be terminated
+                    else:
+                        self.terminate_vm()
+
+
+    # This function is a wrapper for libvirt.lookupByUUID which fixes some problems
+    # 1. Takes a text UUID and handles converting it to bytes
+    # 2. Try's it and returns a sensible value if not
+    def lookupByUUID(self, tuuid):
+        conn = None
+        dom = None
+        libvirt_name = "qemu:///system"
+    
+        # Convert the text UUID to bytes
+        buuid = uuid.UUID(tuuid).bytes
+    
+        # Try
+        try:
+            # Open a libvirt connection
+            conn = libvirt.open(libvirt_name)
+            if conn == None:
+                ansiiprint.echo('Failed to open local libvirt connection', '{}:'.format(self.domuuid), 'e')
+                return dom
+        
+            # Lookup the UUID
+            dom = conn.lookupByUUID(buuid)
+    
+        # Fail
+        except:
+            pass
+    
+        # After everything
+        finally:
+            # Close the libvirt connection
+            if conn != None:
+                conn.close()
+    
+        # Return the dom object (or None)
+        return dom

pvcd/ansiiprint.py

@@ -0,0 +1,80 @@
+#!/usr/bin/env python3
+
+# ansiprint.py - Printing function for formatted messages
+# Part of the Parallel Virtual Cluster (PVC) system
+#
+#    Copyright (C) 2018  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, either version 3 of the License, or
+#    (at your option) any later version.
+#
+#    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 datetime
+
+# ANSII colours for output
+def red():
+    return '\033[91m'
+def blue():
+    return '\033[94m'
+def green():
+    return '\033[92m'
+def yellow():
+    return '\033[93m'
+def purple():
+    return '\033[95m'
+def bold():
+    return '\033[1m'
+def end():
+    return '\033[0m'
+
+# Print function
+def echo(message, prefix, state):
+    # Get the date
+    date = '{} - '.format(datetime.datetime.now().strftime('%Y/%m/%d %H:%M:%S.%f'))
+    endc = end()
+
+    # Continuation
+    if state == 'c':
+        date = ''
+        colour = ''
+        prompt = '    '
+    # OK
+    elif state == 'o':
+        colour = green()
+        prompt = '>>> '
+    # Error
+    elif state == 'e':
+        colour = red()
+        prompt = '>>> '
+    # Warning
+    elif state == 'w':
+        colour = yellow()
+        prompt = '>>> '
+    # Tick
+    elif state == 't':
+        colour = purple()
+        prompt = '>>> '
+    # Information
+    elif state == 'i':
+        colour = blue()
+        prompt = '>>> '
+    else:
+        colour = bold()
+        prompt = '>>> '
+
+    # Append space to prefix
+    if prefix != '':
+        prefix = prefix + ' '
+
+    print(colour + prompt + endc + date + prefix + message)