Readd 1 second queue get timeout

Otherwise daemon stops will sometimes inexplicably block.

.version

@@ -1 +1 @@
-0.9.24
+0.9.26

README.md

@@ -42,6 +42,20 @@ To get started with PVC, please see the [About](https://parallelvirtualcluster.r
 
 ## Changelog
 
+#### v0.9.26
+
+  * [Node Daemon] Corrects some bad assumptions about fencing results during hardware failures
+  * [All] Implements VM tagging functionality
+  * [All] Implements Node log access via PVC functionality
+
+#### v0.9.25
+
+  * [Node Daemon] Returns to Rados library calls for Ceph due to performance problems
+  * [Node Daemon] Adds a date output to keepalive messages
+  * [Daemons] Configures ZK connection logging only for persistent connections
+  * [API Provisioner] Add context manager-based chroot to Debootstrap example script
+  * [Node Daemon] Fixes a bug where shutdown daemon state was overwritten
+
 #### v0.9.24
 
   * [Node Daemon] Removes Rados module polling of Ceph cluster and returns to command-based polling for timeout purposes, and removes some flaky return statements

api-daemon/provisioner/examples/debootstrap_script.py

@@ -34,6 +34,29 @@
 # with that.
 
 import os
+from contextlib import contextmanager
+
+
+# Create a chroot context manager
+# This can be used later in the script to chroot to the destination directory
+# for instance to run commands within the target.
+@contextmanager
+def chroot_target(destination):
+    try:
+        real_root = os.open("/", os.O_RDONLY)
+        os.chroot(destination)
+        fake_root = os.open("/", os.O_RDONLY)
+        os.fchdir(fake_root)
+        yield
+    finally:
+        os.fchdir(real_root)
+        os.chroot(".")
+        os.fchdir(real_root)
+        os.close(fake_root)
+        os.close(real_root)
+        del fake_root
+        del real_root
+
 
 # Installation function - performs a debootstrap install of a Debian system
 # Note that the only arguments are keyword arguments.
@@ -193,13 +216,7 @@ GRUB_DISABLE_LINUX_UUID=false
         fh.write(data)
 
     # Chroot, do some in-root tasks, then exit the chroot
-    # EXITING THE CHROOT IS VERY IMPORTANT OR THE FOLLOWING STAGES OF THE PROVISIONER
-    # WILL FAIL IN UNEXPECTED WAYS! Keep this in mind when using chroot in your scripts.
-    real_root = os.open("/", os.O_RDONLY)
-    os.chroot(temporary_directory)
-    fake_root = os.open("/", os.O_RDONLY)
-    os.fchdir(fake_root)
-
+    with chroot_target(temporary_directory):
         # Install and update GRUB
         os.system(
             "grub-install --force /dev/rbd/{}/{}_{}".format(root_disk['pool'], vm_name, root_disk['disk_id'])
@@ -219,15 +236,6 @@ GRUB_DISABLE_LINUX_UUID=false
             "systemctl enable cloud-init.target"
         )
 
-    # Restore our original root/exit the chroot
-    # EXITING THE CHROOT IS VERY IMPORTANT OR THE FOLLOWING STAGES OF THE PROVISIONER
-    # WILL FAIL IN UNEXPECTED WAYS! Keep this in mind when using chroot in your scripts.
-    os.fchdir(real_root)
-    os.chroot(".")
-    os.fchdir(real_root)
-    os.close(fake_root)
-    os.close(real_root)
-
     # Unmount the bound devfs
     os.system(
         "umount {}/dev".format(
@@ -235,8 +243,4 @@ GRUB_DISABLE_LINUX_UUID=false
         )
     )
 
-    # Clean up file handles so paths can be unmounted
-    del fake_root
-    del real_root
-
     # Everything else is done via cloud-init user-data

api-daemon/provisioner/examples/dummy_script.py

@@ -29,7 +29,7 @@
 # This script will run under root privileges as the provisioner does. Be careful
 # with that.
 
-# Installation function - performs a debootstrap install of a Debian system
+# Installation function - performs no actions then returns
 # Note that the only arguments are keyword arguments.
 def install(**kwargs):
     # The provisioner has already mounted the disks on kwargs['temporary_directory'].

api-daemon/pvcapid/Daemon.py

@@ -25,7 +25,7 @@ import yaml
 from distutils.util import strtobool as dustrtobool
 
 # Daemon version
-version = '0.9.24'
+version = '0.9.26'
 
 # API version
 API_VERSION = 1.0

api-daemon/pvcapid/flaskapi.py

@@ -592,7 +592,7 @@ class API_Node_Root(Resource):
             name: limit
             type: string
             required: false
-            description: A search limit; fuzzy by default, use ^/$ to force exact matches
+            description: A search limit in the name, tags, or an exact UUID; fuzzy by default, use ^/$ to force exact matches
           - in: query
             name: daemon_state
             type: string
@@ -834,6 +834,52 @@ class API_Node_DomainState(Resource):
 api.add_resource(API_Node_DomainState, '/node/<node>/domain-state')
 
 
+# /node/<node</log
+class API_Node_Log(Resource):
+    @RequestParser([
+        {'name': 'lines'}
+    ])
+    @Authenticator
+    def get(self, node, reqargs):
+        """
+        Return the recent logs of {node}
+        ---
+        tags:
+          - node
+        parameters:
+          - in: query
+            name: lines
+            type: integer
+            required: false
+            description: The number of lines to retrieve
+        responses:
+          200:
+            description: OK
+            schema:
+              type: object
+              id: NodeLog
+              properties:
+                name:
+                  type: string
+                  description: The name of the Node
+                data:
+                  type: string
+                  description: The recent log text
+          404:
+            description: Node not found
+            schema:
+              type: object
+              id: Message
+        """
+        return api_helper.node_log(
+            node,
+            reqargs.get('lines', None)
+        )
+
+
+api.add_resource(API_Node_Log, '/node/<node>/log')
+
+
 ##########################################################
 # Client API - VM
 ##########################################################
@@ -844,6 +890,7 @@ class API_VM_Root(Resource):
         {'name': 'limit'},
         {'name': 'node'},
         {'name': 'state'},
+        {'name': 'tag'},
     ])
     @Authenticator
     def get(self, reqargs):
@@ -892,6 +939,22 @@ class API_VM_Root(Resource):
                 migration_method:
                   type: string
                   description: The preferred migration method (live, shutdown, none)
+                tags:
+                  type: array
+                  description: The tag(s) of the VM
+                  items:
+                    type: object
+                    id: VMTag
+                    properties:
+                      name:
+                        type: string
+                        description: The name of the tag
+                      type:
+                        type: string
+                        description: The type of the tag (user, system)
+                      protected:
+                        type: boolean
+                        description: Whether the tag is protected or not
                 description:
                   type: string
                   description: The description of the VM
@@ -1076,7 +1139,7 @@ class API_VM_Root(Resource):
             name: limit
             type: string
             required: false
-            description: A name search limit; fuzzy by default, use ^/$ to force exact matches
+            description: A search limit in the name, tags, or an exact UUID; fuzzy by default, use ^/$ to force exact matches
           - in: query
             name: node
             type: string
@@ -1087,6 +1150,11 @@ class API_VM_Root(Resource):
             type: string
             required: false
             description: Limit list to VMs in this state
+          - in: query
+            name: tag
+            type: string
+            required: false
+            description: Limit list to VMs with this tag
         responses:
           200:
             description: OK
@@ -1098,6 +1166,7 @@ class API_VM_Root(Resource):
         return api_helper.vm_list(
             reqargs.get('node', None),
             reqargs.get('state', None),
+            reqargs.get('tag', None),
             reqargs.get('limit', None)
         )
 
@@ -1107,6 +1176,8 @@ class API_VM_Root(Resource):
         {'name': 'selector', 'choices': ('mem', 'vcpus', 'load', 'vms', 'none'), 'helptext': "A valid selector must be specified"},
         {'name': 'autostart'},
         {'name': 'migration_method', 'choices': ('live', 'shutdown', 'none'), 'helptext': "A valid migration_method must be specified"},
+        {'name': 'user_tags', 'action': 'append'},
+        {'name': 'protected_tags', 'action': 'append'},
         {'name': 'xml', 'required': True, 'helptext': "A Libvirt XML document must be specified"},
     ])
     @Authenticator
@@ -1158,6 +1229,20 @@ class API_VM_Root(Resource):
               - live
               - shutdown
               - none
+          - in: query
+            name: user_tags
+            type: array
+            required: false
+            description: The user tag(s) of the VM
+            items:
+              type: string
+          - in: query
+            name: protected_tags
+            type: array
+            required: false
+            description: The protected user tag(s) of the VM
+            items:
+              type: string
         responses:
           200:
             description: OK
@@ -1170,13 +1255,22 @@ class API_VM_Root(Resource):
               type: object
               id: Message
         """
+        user_tags = reqargs.get('user_tags', None)
+        if user_tags is None:
+            user_tags = []
+        protected_tags = reqargs.get('protected_tags', None)
+        if protected_tags is None:
+            protected_tags = []
+
         return api_helper.vm_define(
             reqargs.get('xml'),
             reqargs.get('node', None),
             reqargs.get('limit', None),
             reqargs.get('selector', 'none'),
             bool(strtobool(reqargs.get('autostart', 'false'))),
-            reqargs.get('migration_method', 'none')
+            reqargs.get('migration_method', 'none'),
+            user_tags,
+            protected_tags
         )
 
 
@@ -1203,7 +1297,7 @@ class API_VM_Element(Resource):
               type: object
               id: Message
         """
-        return api_helper.vm_list(None, None, vm, is_fuzzy=False)
+        return api_helper.vm_list(None, None, None, vm, is_fuzzy=False)
 
     @RequestParser([
         {'name': 'limit'},
@@ -1211,6 +1305,8 @@ class API_VM_Element(Resource):
         {'name': 'selector', 'choices': ('mem', 'vcpus', 'load', 'vms', 'none'), 'helptext': "A valid selector must be specified"},
         {'name': 'autostart'},
         {'name': 'migration_method', 'choices': ('live', 'shutdown', 'none'), 'helptext': "A valid migration_method must be specified"},
+        {'name': 'user_tags', 'action': 'append'},
+        {'name': 'protected_tags', 'action': 'append'},
         {'name': 'xml', 'required': True, 'helptext': "A Libvirt XML document must be specified"},
     ])
     @Authenticator
@@ -1265,6 +1361,20 @@ class API_VM_Element(Resource):
               - live
               - shutdown
               - none
+          - in: query
+            name: user_tags
+            type: array
+            required: false
+            description: The user tag(s) of the VM
+            items:
+              type: string
+          - in: query
+            name: protected_tags
+            type: array
+            required: false
+            description: The protected user tag(s) of the VM
+            items:
+              type: string
         responses:
           200:
             description: OK
@@ -1277,13 +1387,22 @@ class API_VM_Element(Resource):
               type: object
               id: Message
         """
+        user_tags = reqargs.get('user_tags', None)
+        if user_tags is None:
+            user_tags = []
+        protected_tags = reqargs.get('protected_tags', None)
+        if protected_tags is None:
+            protected_tags = []
+
         return api_helper.vm_define(
             reqargs.get('xml'),
             reqargs.get('node', None),
             reqargs.get('limit', None),
             reqargs.get('selector', 'none'),
             bool(strtobool(reqargs.get('autostart', 'false'))),
-            reqargs.get('migration_method', 'none')
+            reqargs.get('migration_method', 'none'),
+            user_tags,
+            protected_tags
         )
 
     @RequestParser([
@@ -1401,7 +1520,7 @@ class API_VM_Metadata(Resource):
                   type: string
                   description: The preferred migration method (live, shutdown, none)
           404:
-            description: Not found
+            description: VM not found
             schema:
               type: object
               id: Message
@@ -1469,6 +1588,11 @@ class API_VM_Metadata(Resource):
             schema:
               type: object
               id: Message
+          404:
+            description: VM not found
+            schema:
+              type: object
+              id: Message
         """
         return api_helper.update_vm_meta(
             vm,
@@ -1483,6 +1607,99 @@ class API_VM_Metadata(Resource):
 api.add_resource(API_VM_Metadata, '/vm/<vm>/meta')
 
 
+# /vm/<vm>/tags
+class API_VM_Tags(Resource):
+    @Authenticator
+    def get(self, vm):
+        """
+        Return the tags of {vm}
+        ---
+        tags:
+          - vm
+        responses:
+          200:
+            description: OK
+            schema:
+              type: object
+              id: VMTags
+              properties:
+                name:
+                  type: string
+                  description: The name of the VM
+                tags:
+                  type: array
+                  description: The tag(s) of the VM
+                  items:
+                    type: object
+                    id: VMTag
+          404:
+            description: VM not found
+            schema:
+              type: object
+              id: Message
+        """
+        return api_helper.get_vm_tags(vm)
+
+    @RequestParser([
+        {'name': 'action', 'choices': ('add', 'remove'), 'helptext': "A valid action must be specified"},
+        {'name': 'tag'},
+        {'name': 'protected'}
+    ])
+    @Authenticator
+    def post(self, vm, reqargs):
+        """
+        Set the tags of {vm}
+        ---
+        tags:
+          - vm
+        parameters:
+          - in: query
+            name: action
+            type: string
+            required: true
+            description: The action to perform with the tag
+            enum:
+              - add
+              - remove
+          - in: query
+            name: tag
+            type: string
+            required: true
+            description: The text value of the tag
+          - in: query
+            name: protected
+            type: boolean
+            required: false
+            default: false
+            description: Set the protected state of the tag
+        responses:
+          200:
+            description: OK
+            schema:
+              type: object
+              id: Message
+          400:
+            description: Bad request
+            schema:
+              type: object
+              id: Message
+          404:
+            description: VM not found
+            schema:
+              type: object
+              id: Message
+        """
+        return api_helper.update_vm_tag(
+            vm,
+            reqargs.get('action'),
+            reqargs.get('tag'),
+            reqargs.get('protected', False)
+        )
+
+
+api.add_resource(API_VM_Tags, '/vm/<vm>/tags')
+
+
 # /vm/<vm</state
 class API_VM_State(Resource):
     @Authenticator

api-daemon/pvcapid/helper.py

@@ -307,6 +307,34 @@ def node_ready(zkhandler, node, wait):
     return output, retcode
 
 
+@ZKConnection(config)
+def node_log(zkhandler, node, lines=None):
+    """
+    Return the current logs for Node.
+    """
+    # Default to 10 lines of log if not set
+    try:
+        lines = int(lines)
+    except TypeError:
+        lines = 10
+
+    retflag, retdata = pvc_node.get_node_log(zkhandler, node, lines)
+
+    if retflag:
+        retcode = 200
+        retdata = {
+            'name': node,
+            'data': retdata
+        }
+    else:
+        retcode = 400
+        retdata = {
+            'message': retdata
+        }
+
+    return retdata, retcode
+
+
 #
 # VM functions
 #
@@ -326,7 +354,7 @@ def vm_state(zkhandler, vm):
     """
     Return the state of virtual machine VM.
     """
-    retflag, retdata = pvc_vm.get_list(zkhandler, None, None, vm, is_fuzzy=False)
+    retflag, retdata = pvc_vm.get_list(zkhandler, None, None, None, vm, is_fuzzy=False)
 
     if retflag:
         if retdata:
@@ -355,7 +383,7 @@ def vm_node(zkhandler, vm):
     """
     Return the current node of virtual machine VM.
     """
-    retflag, retdata = pvc_vm.get_list(zkhandler, None, None, vm, is_fuzzy=False)
+    retflag, retdata = pvc_vm.get_list(zkhandler, None, None, None, vm, is_fuzzy=False)
 
     if retflag:
         if retdata:
@@ -409,11 +437,11 @@ def vm_console(zkhandler, vm, lines=None):
 
 @pvc_common.Profiler(config)
 @ZKConnection(config)
-def vm_list(zkhandler, node=None, state=None, limit=None, is_fuzzy=True):
+def vm_list(zkhandler, node=None, state=None, tag=None, limit=None, is_fuzzy=True):
     """
     Return a list of VMs with limit LIMIT.
     """
-    retflag, retdata = pvc_vm.get_list(zkhandler, node, state, limit, is_fuzzy)
+    retflag, retdata = pvc_vm.get_list(zkhandler, node, state, tag, limit, is_fuzzy)
 
     if retflag:
         if retdata:
@@ -433,7 +461,7 @@ def vm_list(zkhandler, node=None, state=None, limit=None, is_fuzzy=True):
 
 
 @ZKConnection(config)
-def vm_define(zkhandler, xml, node, limit, selector, autostart, migration_method):
+def vm_define(zkhandler, xml, node, limit, selector, autostart, migration_method, user_tags=[], protected_tags=[]):
     """
     Define a VM from Libvirt XML in the PVC cluster.
     """
@@ -444,7 +472,13 @@ def vm_define(zkhandler, xml, node, limit, selector, autostart, migration_method
     except Exception as e:
         return {'message': 'XML is malformed or incorrect: {}'.format(e)}, 400
 
-    retflag, retdata = pvc_vm.define_vm(zkhandler, new_cfg, node, limit, selector, autostart, migration_method, profile=None)
+    tags = list()
+    for tag in user_tags:
+        tags.append({'name': tag, 'type': 'user', 'protected': False})
+    for tag in protected_tags:
+        tags.append({'name': tag, 'type': 'user', 'protected': True})
+
+    retflag, retdata = pvc_vm.define_vm(zkhandler, new_cfg, node, limit, selector, autostart, migration_method, profile=None, tags=tags)
 
     if retflag:
         retcode = 200
@@ -463,27 +497,19 @@ def get_vm_meta(zkhandler, vm):
     """
     Get metadata of a VM.
     """
-    retflag, retdata = pvc_vm.get_list(zkhandler, None, None, vm, is_fuzzy=False)
+    dom_uuid = pvc_vm.getDomainUUID(zkhandler, vm)
+    if not dom_uuid:
+        return {"message": "VM not found."}, 404
+
+    domain_node_limit, domain_node_selector, domain_node_autostart, domain_migrate_method = pvc_common.getDomainMetadata(zkhandler, dom_uuid)
 
-    if retflag:
-        if retdata:
     retcode = 200
     retdata = {
         'name': vm,
-                'node_limit': retdata['node_limit'],
-                'node_selector': retdata['node_selector'],
-                'node_autostart': retdata['node_autostart'],
-                'migration_method': retdata['migration_method']
-            }
-        else:
-            retcode = 404
-            retdata = {
-                'message': 'VM not found.'
-            }
-    else:
-        retcode = 400
-        retdata = {
-            'message': retdata
+        'node_limit': domain_node_limit,
+        'node_selector': domain_node_selector,
+        'node_autostart': domain_node_autostart,
+        'migration_method': domain_migrate_method
     }
 
     return retdata, retcode
@@ -494,11 +520,16 @@ def update_vm_meta(zkhandler, vm, limit, selector, autostart, provisioner_profil
     """
     Update metadata of a VM.
     """
+    dom_uuid = pvc_vm.getDomainUUID(zkhandler, vm)
+    if not dom_uuid:
+        return {"message": "VM not found."}, 404
+
     if autostart is not None:
         try:
             autostart = bool(strtobool(autostart))
         except Exception:
             autostart = False
+
     retflag, retdata = pvc_vm.modify_vm_metadata(zkhandler, vm, limit, selector, autostart, provisioner_profile, migration_method)
 
     if retflag:
@@ -512,6 +543,51 @@ def update_vm_meta(zkhandler, vm, limit, selector, autostart, provisioner_profil
     return output, retcode
 
 
+@ZKConnection(config)
+def get_vm_tags(zkhandler, vm):
+    """
+    Get the tags of a VM.
+    """
+    dom_uuid = pvc_vm.getDomainUUID(zkhandler, vm)
+    if not dom_uuid:
+        return {"message": "VM not found."}, 404
+
+    tags = pvc_common.getDomainTags(zkhandler, dom_uuid)
+
+    retcode = 200
+    retdata = {
+        'name': vm,
+        'tags': tags
+    }
+
+    return retdata, retcode
+
+
+@ZKConnection(config)
+def update_vm_tag(zkhandler, vm, action, tag, protected=False):
+    """
+    Update a tag of a VM.
+    """
+    if action not in ['add', 'remove']:
+        return {"message": "Tag action must be one of 'add', 'remove'."}, 400
+
+    dom_uuid = pvc_vm.getDomainUUID(zkhandler, vm)
+    if not dom_uuid:
+        return {"message": "VM not found."}, 404
+
+    retflag, retdata = pvc_vm.modify_vm_tag(zkhandler, vm, action, tag, protected=protected)
+
+    if retflag:
+        retcode = 200
+    else:
+        retcode = 400
+
+    output = {
+        'message': retdata.replace('\"', '\'')
+    }
+    return output, retcode
+
+
 @ZKConnection(config)
 def vm_modify(zkhandler, name, restart, xml):
     """
@@ -752,7 +828,7 @@ def vm_flush_locks(zkhandler, vm):
     """
     Flush locks of a (stopped) VM.
     """
-    retflag, retdata = pvc_vm.get_list(zkhandler, None, None, vm, is_fuzzy=False)
+    retflag, retdata = pvc_vm.get_list(zkhandler, None, None, None, vm, is_fuzzy=False)
 
     if retdata[0].get('state') not in ['stop', 'disable']:
         return {"message": "VM must be stopped to flush locks"}, 400

api-daemon/pvcapid/libvirt_schema.py

@@ -41,6 +41,7 @@ libvirt_header = """<domain type='kvm'>
     <bootmenu enable='yes'/>
     <boot dev='cdrom'/>
     <boot dev='hd'/>
+    <bios useserial='yes' rebootTimeout='5'/>
   </os>
   <features>
     <acpi/>

client-cli/pvc/cli_lib/node.py

@@ -19,6 +19,8 @@
 #
 ###############################################################################
 
+import time
+
 import pvc.cli_lib.ansiprint as ansiprint
 from pvc.cli_lib.common import call_api
 
@@ -69,6 +71,89 @@ def node_domain_state(config, node, action, wait):
     return retstatus, response.json().get('message', '')
 
 
+def view_node_log(config, node, lines=100):
+    """
+    Return node log lines from the API (and display them in a pager in the main CLI)
+
+    API endpoint: GET /node/{node}/log
+    API arguments: lines={lines}
+    API schema: {"name":"{node}","data":"{node_log}"}
+    """
+    params = {
+        'lines': lines
+    }
+    response = call_api(config, 'get', '/node/{node}/log'.format(node=node), params=params)
+
+    if response.status_code != 200:
+        return False, response.json().get('message', '')
+
+    node_log = response.json()['data']
+
+    # Shrink the log buffer to length lines
+    shrunk_log = node_log.split('\n')[-lines:]
+    loglines = '\n'.join(shrunk_log)
+
+    return True, loglines
+
+
+def follow_node_log(config, node, lines=10):
+    """
+    Return and follow node log lines from the API
+
+    API endpoint: GET /node/{node}/log
+    API arguments: lines={lines}
+    API schema: {"name":"{nodename}","data":"{node_log}"}
+    """
+    # We always grab 200 to match the follow call, but only _show_ `lines` number
+    params = {
+        'lines': 200
+    }
+    response = call_api(config, 'get', '/node/{node}/log'.format(node=node), params=params)
+
+    if response.status_code != 200:
+        return False, response.json().get('message', '')
+
+    # Shrink the log buffer to length lines
+    node_log = response.json()['data']
+    shrunk_log = node_log.split('\n')[-int(lines):]
+    loglines = '\n'.join(shrunk_log)
+
+    # Print the initial data and begin following
+    print(loglines, end='')
+    print('\n', end='')
+
+    while True:
+        # Grab the next line set (200 is a reasonable number of lines per half-second; any more are skipped)
+        try:
+            params = {
+                'lines': 200
+            }
+            response = call_api(config, 'get', '/node/{node}/log'.format(node=node), params=params)
+            new_node_log = response.json()['data']
+        except Exception:
+            break
+        # Split the new and old log strings into constitutent lines
+        old_node_loglines = node_log.split('\n')
+        new_node_loglines = new_node_log.split('\n')
+
+        # Set the node log to the new log value for the next iteration
+        node_log = new_node_log
+
+        # Get the difference between the two sets of lines
+        old_node_loglines_set = set(old_node_loglines)
+        diff_node_loglines = [x for x in new_node_loglines if x not in old_node_loglines_set]
+
+        # If there's a difference, print it out
+        if len(diff_node_loglines) > 0:
+            print('\n'.join(diff_node_loglines), end='')
+            print('\n', end='')
+
+        # Wait half a second
+        time.sleep(0.5)
+
+    return True, ''
+
+
 def node_info(config, node):
     """
     Get information about node

client-cli/pvc/cli_lib/vm.py

@@ -54,12 +54,12 @@ def vm_info(config, vm):
         return False, response.json().get('message', '')
 
 
-def vm_list(config, limit, target_node, target_state):
+def vm_list(config, limit, target_node, target_state, target_tag):
     """
     Get list information about VMs (limited by {limit}, {target_node}, or {target_state})
 
     API endpoint: GET /api/v1/vm
-    API arguments: limit={limit}, node={target_node}, state={target_state}
+    API arguments: limit={limit}, node={target_node}, state={target_state}, tag={target_tag}
     API schema: [{json_data_object},{json_data_object},etc.]
     """
     params = dict()
@@ -69,6 +69,8 @@ def vm_list(config, limit, target_node, target_state):
         params['node'] = target_node
     if target_state:
         params['state'] = target_state
+    if target_tag:
+        params['tag'] = target_tag
 
     response = call_api(config, 'get', '/vm', params=params)
 
@@ -78,12 +80,12 @@ def vm_list(config, limit, target_node, target_state):
         return False, response.json().get('message', '')
 
 
-def vm_define(config, xml, node, node_limit, node_selector, node_autostart, migration_method):
+def vm_define(config, xml, node, node_limit, node_selector, node_autostart, migration_method, user_tags, protected_tags):
     """
     Define a new VM on the cluster
 
     API endpoint: POST /vm
-    API arguments: xml={xml}, node={node}, limit={node_limit}, selector={node_selector}, autostart={node_autostart}, migration_method={migration_method}
+    API arguments: xml={xml}, node={node}, limit={node_limit}, selector={node_selector}, autostart={node_autostart}, migration_method={migration_method}, user_tags={user_tags}, protected_tags={protected_tags}
     API schema: {"message":"{data}"}
     """
     params = {
@@ -91,7 +93,9 @@ def vm_define(config, xml, node, node_limit, node_selector, node_autostart, migr
         'limit': node_limit,
         'selector': node_selector,
         'autostart': node_autostart,
-        'migration_method': migration_method
+        'migration_method': migration_method,
+        'user_tags': user_tags,
+        'protected_tags': protected_tags
     }
     data = {
         'xml': xml
@@ -155,7 +159,7 @@ def vm_metadata(config, vm, node_limit, node_selector, node_autostart, migration
     """
     Modify PVC metadata of a VM
 
-    API endpoint: GET /vm/{vm}/meta,  POST /vm/{vm}/meta
+    API endpoint: POST /vm/{vm}/meta
     API arguments: limit={node_limit}, selector={node_selector}, autostart={node_autostart}, migration_method={migration_method} profile={provisioner_profile}
     API schema: {"message":"{data}"}
     """
@@ -188,6 +192,119 @@ def vm_metadata(config, vm, node_limit, node_selector, node_autostart, migration
     return retstatus, response.json().get('message', '')
 
 
+def vm_tags_get(config, vm):
+    """
+    Get PVC tags of a VM
+
+    API endpoint: GET /vm/{vm}/tags
+    API arguments:
+    API schema: {{"name": "{name}", "type": "{type}"},...}
+    """
+
+    response = call_api(config, 'get', '/vm/{vm}/tags'.format(vm=vm))
+
+    if response.status_code == 200:
+        retstatus = True
+        retdata = response.json()
+    else:
+        retstatus = False
+        retdata = response.json().get('message', '')
+
+    return retstatus, retdata
+
+
+def vm_tag_set(config, vm, action, tag, protected=False):
+    """
+    Modify PVC tags of a VM
+
+    API endpoint: POST /vm/{vm}/tags
+    API arguments: action={action}, tag={tag}, protected={protected}
+    API schema: {"message":"{data}"}
+    """
+
+    params = {
+        'action': action,
+        'tag': tag,
+        'protected': protected
+    }
+
+    # Update the tags
+    response = call_api(config, 'post', '/vm/{vm}/tags'.format(vm=vm), params=params)
+
+    if response.status_code == 200:
+        retstatus = True
+    else:
+        retstatus = False
+
+    return retstatus, response.json().get('message', '')
+
+
+def format_vm_tags(config, name, tags):
+    """
+    Format the output of a tags dictionary in a nice table
+    """
+    if len(tags) < 1:
+        return "No tags found."
+
+    output_list = []
+
+    name_length = 5
+    _name_length = len(name) + 1
+    if _name_length > name_length:
+        name_length = _name_length
+
+    tags_name_length = 4
+    tags_type_length = 5
+    tags_protected_length = 10
+    for tag in tags:
+        _tags_name_length = len(tag['name']) + 1
+        if _tags_name_length > tags_name_length:
+            tags_name_length = _tags_name_length
+
+        _tags_type_length = len(tag['type']) + 1
+        if _tags_type_length > tags_type_length:
+            tags_type_length = _tags_type_length
+
+        _tags_protected_length = len(str(tag['protected'])) + 1
+        if _tags_protected_length > tags_protected_length:
+            tags_protected_length = _tags_protected_length
+
+    output_list.append(
+        '{bold}{tags_name: <{tags_name_length}}  \
+{tags_type: <{tags_type_length}}  \
+{tags_protected: <{tags_protected_length}}{end_bold}'.format(
+            name_length=name_length,
+            tags_name_length=tags_name_length,
+            tags_type_length=tags_type_length,
+            tags_protected_length=tags_protected_length,
+            bold=ansiprint.bold(),
+            end_bold=ansiprint.end(),
+            tags_name='Name',
+            tags_type='Type',
+            tags_protected='Protected'
+        )
+    )
+
+    for tag in sorted(tags, key=lambda t: t['name']):
+        output_list.append(
+            '{bold}{tags_name: <{tags_name_length}}  \
+{tags_type: <{tags_type_length}}  \
+{tags_protected: <{tags_protected_length}}{end_bold}'.format(
+                name_length=name_length,
+                tags_type_length=tags_type_length,
+                tags_name_length=tags_name_length,
+                tags_protected_length=tags_protected_length,
+                bold='',
+                end_bold='',
+                tags_name=tag['name'],
+                tags_type=tag['type'],
+                tags_protected=str(tag['protected'])
+            )
+        )
+
+    return '\n'.join(output_list)
+
+
 def vm_remove(config, vm, delete_disks=False):
     """
     Remove a VM
@@ -1098,9 +1215,9 @@ def follow_console_log(config, vm, lines=10):
     API arguments: lines={lines}
     API schema: {"name":"{vmname}","data":"{console_log}"}
     """
-    # We always grab 500 to match the follow call, but only _show_ `lines` number
+    # We always grab 200 to match the follow call, but only _show_ `lines` number
     params = {
-        'lines': 500
+        'lines': 200
     }
     response = call_api(config, 'get', '/vm/{vm}/console'.format(vm=vm), params=params)
 
@@ -1116,10 +1233,10 @@ def follow_console_log(config, vm, lines=10):
     print(loglines, end='')
 
     while True:
-        # Grab the next line set (500 is a reasonable number of lines per second; any more are skipped)
+        # Grab the next line set (200 is a reasonable number of lines per half-second; any more are skipped)
         try:
             params = {
-                'lines': 500
+                'lines': 200
             }
             response = call_api(config, 'get', '/vm/{vm}/console'.format(vm=vm), params=params)
             new_console_log = response.json()['data']
@@ -1128,8 +1245,10 @@ def follow_console_log(config, vm, lines=10):
         # Split the new and old log strings into constitutent lines
         old_console_loglines = console_log.split('\n')
         new_console_loglines = new_console_log.split('\n')
+
         # Set the console log to the new log value for the next iteration
         console_log = new_console_log
+
         # Remove the lines from the old log until we hit the first line of the new log; this
         # ensures that the old log is a string that we can remove from the new log entirely
         for index, line in enumerate(old_console_loglines, start=0):
@@ -1144,8 +1263,8 @@ def follow_console_log(config, vm, lines=10):
         # If there's a difference, print it out
         if diff_console_log:
             print(diff_console_log, end='')
-        # Wait a second
-        time.sleep(1)
+        # Wait half a second
+        time.sleep(0.5)
 
     return True, ''
 
@@ -1248,6 +1367,54 @@ def format_info(config, domain_information, long_output):
     ainformation.append('{}Autostart:{}          {}'.format(ansiprint.purple(), ansiprint.end(), formatted_node_autostart))
     ainformation.append('{}Migration Method:{}   {}'.format(ansiprint.purple(), ansiprint.end(), formatted_migration_method))
 
+    # Tag list
+    tags_name_length = 5
+    tags_type_length = 5
+    tags_protected_length = 10
+    for tag in domain_information['tags']:
+        _tags_name_length = len(tag['name']) + 1
+        if _tags_name_length > tags_name_length:
+            tags_name_length = _tags_name_length
+
+        _tags_type_length = len(tag['type']) + 1
+        if _tags_type_length > tags_type_length:
+            tags_type_length = _tags_type_length
+
+        _tags_protected_length = len(str(tag['protected'])) + 1
+        if _tags_protected_length > tags_protected_length:
+            tags_protected_length = _tags_protected_length
+
+    if len(domain_information['tags']) > 0:
+        ainformation.append('')
+        ainformation.append('{purple}Tags:{end}               {bold}{tags_name: <{tags_name_length}} {tags_type: <{tags_type_length}} {tags_protected: <{tags_protected_length}}{end}'.format(
+            purple=ansiprint.purple(),
+            bold=ansiprint.bold(),
+            end=ansiprint.end(),
+            tags_name_length=tags_name_length,
+            tags_type_length=tags_type_length,
+            tags_protected_length=tags_protected_length,
+            tags_name='Name',
+            tags_type='Type',
+            tags_protected='Protected'
+        ))
+
+        for tag in sorted(domain_information['tags'], key=lambda t: t['type'] + t['name']):
+            ainformation.append('                    {tags_name: <{tags_name_length}} {tags_type: <{tags_type_length}} {tags_protected: <{tags_protected_length}}'.format(
+                tags_name_length=tags_name_length,
+                tags_type_length=tags_type_length,
+                tags_protected_length=tags_protected_length,
+                tags_name=tag['name'],
+                tags_type=tag['type'],
+                tags_protected=str(tag['protected'])
+            ))
+    else:
+        ainformation.append('')
+        ainformation.append('{purple}Tags:{end}               N/A'.format(
+            purple=ansiprint.purple(),
+            bold=ansiprint.bold(),
+            end=ansiprint.end(),
+        ))
+
     # Network list
     net_list = []
     cluster_net_list = call_api(config, 'get', '/network').json()
@@ -1331,6 +1498,14 @@ def format_list(config, vm_list, raw):
             net_list.append(net['vni'])
         return net_list
 
+    # Function to get tag names and returna  nicer list
+    def getNiceTagName(domain_information):
+        # Tag list
+        tag_list = []
+        for tag in sorted(domain_information['tags'], key=lambda t: t['type'] + t['name']):
+            tag_list.append(tag['name'])
+        return tag_list
+
     # Handle raw mode since it just lists the names
     if raw:
         ainformation = list()
@@ -1344,6 +1519,7 @@ def format_list(config, vm_list, raw):
     # Dynamic columns: node_name, node, migrated
     vm_name_length = 5
     vm_state_length = 6
+    vm_tags_length = 5
     vm_nets_length = 9
     vm_ram_length = 8
     vm_vcpu_length = 6
@@ -1351,6 +1527,7 @@ def format_list(config, vm_list, raw):
     vm_migrated_length = 9
     for domain_information in vm_list:
         net_list = getNiceNetID(domain_information)
+        tag_list = getNiceTagName(domain_information)
         # vm_name column
         _vm_name_length = len(domain_information['name']) + 1
         if _vm_name_length > vm_name_length:
@@ -1359,6 +1536,10 @@ def format_list(config, vm_list, raw):
         _vm_state_length = len(domain_information['state']) + 1
         if _vm_state_length > vm_state_length:
             vm_state_length = _vm_state_length
+        # vm_tags column
+        _vm_tags_length = len(','.join(tag_list)) + 1
+        if _vm_tags_length > vm_tags_length:
+            vm_tags_length = _vm_tags_length
         # vm_nets column
         _vm_nets_length = len(','.join(net_list)) + 1
         if _vm_nets_length > vm_nets_length:
@@ -1375,12 +1556,12 @@ def format_list(config, vm_list, raw):
     # Format the string (header)
     vm_list_output.append(
         '{bold}{vm_header: <{vm_header_length}} {resource_header: <{resource_header_length}} {node_header: <{node_header_length}}{end_bold}'.format(
-            vm_header_length=vm_name_length + vm_state_length + 1,
+            vm_header_length=vm_name_length + vm_state_length + vm_tags_length + 2,
             resource_header_length=vm_nets_length + vm_ram_length + vm_vcpu_length + 2,
             node_header_length=vm_node_length + vm_migrated_length + 1,
             bold=ansiprint.bold(),
             end_bold=ansiprint.end(),
-            vm_header='VMs ' + ''.join(['-' for _ in range(4, vm_name_length + vm_state_length)]),
+            vm_header='VMs ' + ''.join(['-' for _ in range(4, vm_name_length + vm_state_length + vm_tags_length + 1)]),
             resource_header='Resources ' + ''.join(['-' for _ in range(10, vm_nets_length + vm_ram_length + vm_vcpu_length + 1)]),
             node_header='Node ' + ''.join(['-' for _ in range(5, vm_node_length + vm_migrated_length)])
         )
@@ -1389,12 +1570,14 @@ def format_list(config, vm_list, raw):
     vm_list_output.append(
         '{bold}{vm_name: <{vm_name_length}} \
 {vm_state_colour}{vm_state: <{vm_state_length}}{end_colour} \
+{vm_tags: <{vm_tags_length}} \
 {vm_networks: <{vm_nets_length}} \
 {vm_memory: <{vm_ram_length}} {vm_vcpu: <{vm_vcpu_length}} \
 {vm_node: <{vm_node_length}} \
 {vm_migrated: <{vm_migrated_length}}{end_bold}'.format(
             vm_name_length=vm_name_length,
             vm_state_length=vm_state_length,
+            vm_tags_length=vm_tags_length,
             vm_nets_length=vm_nets_length,
             vm_ram_length=vm_ram_length,
             vm_vcpu_length=vm_vcpu_length,
@@ -1406,6 +1589,7 @@ def format_list(config, vm_list, raw):
             end_colour='',
             vm_name='Name',
             vm_state='State',
+            vm_tags='Tags',
             vm_networks='Networks',
             vm_memory='RAM (M)',
             vm_vcpu='vCPUs',
@@ -1434,6 +1618,9 @@ def format_list(config, vm_list, raw):
 
         # Handle colouring for an invalid network config
         net_list = getNiceNetID(domain_information)
+        tag_list = getNiceTagName(domain_information)
+        if len(tag_list) < 1:
+            tag_list = ['N/A']
         vm_net_colour = ''
         for net_vni in net_list:
             if net_vni not in ['cluster', 'storage', 'upstream'] and not re.match(r'^macvtap:.*', net_vni) and not re.match(r'^hostdev:.*', net_vni):
@@ -1443,12 +1630,14 @@ def format_list(config, vm_list, raw):
         vm_list_output.append(
             '{bold}{vm_name: <{vm_name_length}} \
 {vm_state_colour}{vm_state: <{vm_state_length}}{end_colour} \
+{vm_tags: <{vm_tags_length}} \
 {vm_net_colour}{vm_networks: <{vm_nets_length}}{end_colour} \
 {vm_memory: <{vm_ram_length}} {vm_vcpu: <{vm_vcpu_length}} \
 {vm_node: <{vm_node_length}} \
 {vm_migrated: <{vm_migrated_length}}{end_bold}'.format(
                 vm_name_length=vm_name_length,
                 vm_state_length=vm_state_length,
+                vm_tags_length=vm_tags_length,
                 vm_nets_length=vm_nets_length,
                 vm_ram_length=vm_ram_length,
                 vm_vcpu_length=vm_vcpu_length,
@@ -1460,6 +1649,7 @@ def format_list(config, vm_list, raw):
                 end_colour=ansiprint.end(),
                 vm_name=domain_information['name'],
                 vm_state=domain_information['state'],
+                vm_tags=','.join(tag_list),
                 vm_net_colour=vm_net_colour,
                 vm_networks=','.join(net_list),
                 vm_memory=domain_information['memory'],

client-cli/pvc/pvc.py

@@ -540,6 +540,43 @@ def node_unflush(node, wait):
     cleanup(retcode, retmsg)
 
 
+###############################################################################
+# pvc node log
+###############################################################################
+@click.command(name='log', short_help='Show logs of a node.')
+@click.argument(
+    'node'
+)
+@click.option(
+    '-l', '--lines', 'lines', default=None, show_default=False,
+    help='Display this many log lines from the end of the log buffer.  [default: 1000; with follow: 10]'
+)
+@click.option(
+    '-f', '--follow', 'follow', is_flag=True, default=False,
+    help='Follow the log buffer; output may be delayed by a few seconds relative to the live system. The --lines value defaults to 10 for the initial output.'
+)
+@cluster_req
+def node_log(node, lines, follow):
+    """
+    Show node logs of virtual machine DOMAIN on its current node in a pager or continuously. DOMAIN may be a UUID or name. Note that migrating a VM to a different node will cause the log buffer to be overwritten by entries from the new node.
+    """
+
+    # Set the default here so we can handle it
+    if lines is None:
+        if follow:
+            lines = 10
+        else:
+            lines = 1000
+
+    if follow:
+        retcode, retmsg = pvc_node.follow_node_log(config, node, lines)
+    else:
+        retcode, retmsg = pvc_node.view_node_log(config, node, lines)
+        click.echo_via_pager(retmsg)
+        retmsg = ''
+    cleanup(retcode, retmsg)
+
+
 ###############################################################################
 # pvc node info
 ###############################################################################
@@ -638,11 +675,21 @@ def cli_vm():
     type=click.Choice(['none', 'live', 'shutdown']),
     help='The preferred migration method of the VM between nodes; saved with VM.'
 )
+@click.option(
+    '-g', '--tag', 'user_tags',
+    default=[], multiple=True,
+    help='User tag for the VM; can be specified multiple times, once per tag.'
+)
+@click.option(
+    '-G', '--protected-tag', 'protected_tags',
+    default=[], multiple=True,
+    help='Protected user tag for the VM; can be specified multiple times, once per tag.'
+)
 @click.argument(
     'vmconfig', type=click.File()
 )
 @cluster_req
-def vm_define(vmconfig, target_node, node_limit, node_selector, node_autostart, migration_method):
+def vm_define(vmconfig, target_node, node_limit, node_selector, node_autostart, migration_method, user_tags, protected_tags):
     """
     Define a new virtual machine from Libvirt XML configuration file VMCONFIG.
     """
@@ -658,7 +705,7 @@ def vm_define(vmconfig, target_node, node_limit, node_selector, node_autostart,
     except Exception:
         cleanup(False, 'Error: XML is malformed or invalid')
 
-    retcode, retmsg = pvc_vm.vm_define(config, new_cfg, target_node, node_limit, node_selector, node_autostart, migration_method)
+    retcode, retmsg = pvc_vm.vm_define(config, new_cfg, target_node, node_limit, node_selector, node_autostart, migration_method, user_tags, protected_tags)
     cleanup(retcode, retmsg)
 
 
@@ -1111,6 +1158,90 @@ def vm_flush_locks(domain):
     cleanup(retcode, retmsg)
 
 
+###############################################################################
+# pvc vm tag
+###############################################################################
+@click.group(name='tag', short_help='Manage tags of a virtual machine.', context_settings=CONTEXT_SETTINGS)
+def vm_tags():
+    """
+    Manage the tags of a virtual machine in the PVC cluster."
+    """
+    pass
+
+
+###############################################################################
+# pvc vm tag get
+###############################################################################
+@click.command(name='get', short_help='Get the current tags of a virtual machine.')
+@click.argument(
+    'domain'
+)
+@click.option(
+    '-r', '--raw', 'raw', is_flag=True, default=False,
+    help='Display the raw value only without formatting.'
+)
+@cluster_req
+def vm_tags_get(domain, raw):
+    """
+    Get the current tags of the virtual machine DOMAIN.
+    """
+
+    retcode, retdata = pvc_vm.vm_tags_get(config, domain)
+    if retcode:
+        if not raw:
+            retdata = pvc_vm.format_vm_tags(config, domain, retdata['tags'])
+        else:
+            if len(retdata['tags']) > 0:
+                retdata = '\n'.join([tag['name'] for tag in retdata['tags']])
+            else:
+                retdata = 'No tags found.'
+    cleanup(retcode, retdata)
+
+
+###############################################################################
+# pvc vm tag add
+###############################################################################
+@click.command(name='add', short_help='Add new tags to a virtual machine.')
+@click.argument(
+    'domain'
+)
+@click.argument(
+    'tag'
+)
+@click.option(
+    '-p', '--protected', 'protected', is_flag=True, required=False, default=False,
+    help="Set this tag as protected; protected tags cannot be removed."
+)
+@cluster_req
+def vm_tags_add(domain, tag, protected):
+    """
+    Add TAG to the virtual machine DOMAIN.
+    """
+
+    retcode, retmsg = pvc_vm.vm_tag_set(config, domain, 'add', tag, protected)
+    cleanup(retcode, retmsg)
+
+
+###############################################################################
+# pvc vm tag remove
+###############################################################################
+@click.command(name='remove', short_help='Remove tags from a virtual machine.')
+@click.argument(
+    'domain'
+)
+@click.argument(
+    'tag'
+)
+@cluster_req
+def vm_tags_remove(domain, tag):
+    """
+    Remove TAG from the virtual machine DOMAIN.
+    """
+
+    retcode, retmsg = pvc_vm.vm_tag_set(config, domain, 'remove', tag)
+    cleanup(retcode, retmsg)
+
+
 ###############################################################################
 # pvc vm vcpu
 ###############################################################################
@@ -1653,19 +1784,23 @@ def vm_dump(filename, domain):
     '-s', '--state', 'target_state', default=None,
     help='Limit list to VMs in the specified state.'
 )
+@click.option(
+    '-g', '--tag', 'target_tag', default=None,
+    help='Limit list to VMs with the specified tag.'
+)
 @click.option(
     '-r', '--raw', 'raw', is_flag=True, default=False,
     help='Display the raw list of VM names only.'
 )
 @cluster_req
-def vm_list(target_node, target_state, limit, raw):
+def vm_list(target_node, target_state, target_tag, limit, raw):
     """
     List all virtual machines; optionally only match names or full UUIDs matching regex LIMIT.
 
     NOTE: Red-coloured network lists indicate one or more configured networks are missing/invalid.
     """
 
-    retcode, retdata = pvc_vm.vm_list(config, limit, target_node, target_state)
+    retcode, retdata = pvc_vm.vm_list(config, limit, target_node, target_state, target_tag)
     if retcode:
         retdata = pvc_vm.format_list(config, retdata, raw)
     else:
@@ -4609,9 +4744,14 @@ cli_node.add_command(node_primary)
 cli_node.add_command(node_flush)
 cli_node.add_command(node_ready)
 cli_node.add_command(node_unflush)
+cli_node.add_command(node_log)
 cli_node.add_command(node_info)
 cli_node.add_command(node_list)
 
+vm_tags.add_command(vm_tags_get)
+vm_tags.add_command(vm_tags_add)
+vm_tags.add_command(vm_tags_remove)
+
 vm_vcpu.add_command(vm_vcpu_get)
 vm_vcpu.add_command(vm_vcpu_set)
 
@@ -4642,6 +4782,7 @@ cli_vm.add_command(vm_move)
 cli_vm.add_command(vm_migrate)
 cli_vm.add_command(vm_unmigrate)
 cli_vm.add_command(vm_flush_locks)
+cli_vm.add_command(vm_tags)
 cli_vm.add_command(vm_vcpu)
 cli_vm.add_command(vm_memory)
 cli_vm.add_command(vm_network)

client-cli/setup.py

@@ -2,7 +2,7 @@ from setuptools import setup
 
 setup(
     name='pvc',
-    version='0.9.24',
+    version='0.9.26',
     packages=['pvc', 'pvc.cli_lib'],
     install_requires=[
         'Click',

daemon-common/cluster.py

@@ -60,7 +60,7 @@ def getClusterInformation(zkhandler):
     retcode, node_list = pvc_node.get_list(zkhandler, None)
 
     # Get vm information object list
-    retcode, vm_list = pvc_vm.get_list(zkhandler, None, None, None)
+    retcode, vm_list = pvc_vm.get_list(zkhandler, None, None, None, None)
 
     # Get network information object list
     retcode, network_list = pvc_network.get_list(zkhandler, None, None)

daemon-common/common.py

@@ -306,6 +306,50 @@ def getDomainDiskList(zkhandler, dom_uuid):
     return disk_list
 
 
+#
+# Get a list of domain tags
+#
+def getDomainTags(zkhandler, dom_uuid):
+    """
+    Get a list of tags for domain dom_uuid
+
+    The UUID must be validated before calling this function!
+    """
+    tags = list()
+
+    for tag in zkhandler.children(('domain.meta.tags', dom_uuid)):
+        tag_type = zkhandler.read(('domain.meta.tags', dom_uuid, 'tag.type', tag))
+        protected = bool(strtobool(zkhandler.read(('domain.meta.tags', dom_uuid, 'tag.protected', tag))))
+        tags.append({'name': tag, 'type': tag_type, 'protected': protected})
+
+    return tags
+
+
+#
+# Get a set of domain metadata
+#
+def getDomainMetadata(zkhandler, dom_uuid):
+    """
+    Get the domain metadata for domain dom_uuid
+
+    The UUID must be validated before calling this function!
+    """
+    domain_node_limit = zkhandler.read(('domain.meta.node_limit', dom_uuid))
+    domain_node_selector = zkhandler.read(('domain.meta.node_selector', dom_uuid))
+    domain_node_autostart = zkhandler.read(('domain.meta.autostart', dom_uuid))
+    domain_migration_method = zkhandler.read(('domain.meta.migrate_method', dom_uuid))
+
+    if not domain_node_limit:
+        domain_node_limit = None
+    else:
+        domain_node_limit = domain_node_limit.split(',')
+
+    if not domain_node_autostart:
+        domain_node_autostart = None
+
+    return domain_node_limit, domain_node_selector, domain_node_autostart, domain_migration_method
+
+
 #
 # Get domain information from XML
 #
@@ -319,19 +363,8 @@ def getInformationFromXML(zkhandler, uuid):
     domain_lastnode = zkhandler.read(('domain.last_node', uuid))
     domain_failedreason = zkhandler.read(('domain.failed_reason', uuid))
 
-    domain_node_limit = zkhandler.read(('domain.meta.node_limit', uuid))
-    domain_node_selector = zkhandler.read(('domain.meta.node_selector', uuid))
-    domain_node_autostart = zkhandler.read(('domain.meta.autostart', uuid))
-    domain_migration_method = zkhandler.read(('domain.meta.migrate_method', uuid))
-
-    if not domain_node_limit:
-        domain_node_limit = None
-    else:
-        domain_node_limit = domain_node_limit.split(',')
-
-    if not domain_node_autostart:
-        domain_node_autostart = None
-
+    domain_node_limit, domain_node_selector, domain_node_autostart, domain_migration_method = getDomainMetadata(zkhandler, uuid)
+    domain_tags = getDomainTags(zkhandler, uuid)
     domain_profile = zkhandler.read(('domain.profile', uuid))
 
     domain_vnc = zkhandler.read(('domain.console.vnc', uuid))
@@ -378,6 +411,7 @@ def getInformationFromXML(zkhandler, uuid):
         'node_selector': domain_node_selector,
         'node_autostart': bool(strtobool(domain_node_autostart)),
         'migration_method': domain_migration_method,
+        'tags': domain_tags,
         'description': domain_description,
         'profile': domain_profile,
         'memory': int(domain_memory),

daemon-common/log.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python3
 
-# log.py - Output (stdout + logfile) functions
+# log.py - PVC daemon logger functions
 # Part of the Parallel Virtual Cluster (PVC) system
 #
 #    Copyright (C) 2018-2021 Joshua M. Boniface <joshua@boniface.me>
@@ -19,7 +19,12 @@
 #
 ###############################################################################
 
-import datetime
+from collections import deque
+from threading import Thread
+from queue import Queue
+from datetime import datetime
+
+from daemon_lib.zkhandler import ZKHandler
 
 
 class Logger(object):
@@ -77,17 +82,32 @@ class Logger(object):
         self.last_colour = ''
         self.last_prompt = ''
 
+        if self.config['zookeeper_logging']:
+            self.zookeeper_logger = ZookeeperLogger(config)
+            self.zookeeper_logger.start()
+
     # Provide a hup function to close and reopen the writer
     def hup(self):
         self.writer.close()
         self.writer = open(self.logfile, 'a', buffering=0)
 
+    # Provide a termination function so all messages are flushed before terminating the main daemon
+    def terminate(self):
+        if self.config['file_logging']:
+            self.writer.close()
+        if self.config['zookeeper_logging']:
+            self.out("Waiting for Zookeeper message queue to drain", state='s')
+            while not self.zookeeper_logger.queue.empty():
+                pass
+            self.zookeeper_logger.stop()
+            self.zookeeper_logger.join()
+
     # Output function
     def out(self, message, state=None, prefix=''):
 
         # Get the date
         if self.config['log_dates']:
-            date = '{} - '.format(datetime.datetime.now().strftime('%Y/%m/%d %H:%M:%S.%f'))
+            date = '{} '.format(datetime.now().strftime('%Y/%m/%d %H:%M:%S.%f'))
         else:
             date = ''
 
@@ -123,6 +143,71 @@ class Logger(object):
         if self.config['file_logging']:
             self.writer.write(message + '\n')
 
+        # Log to Zookeeper
+        if self.config['zookeeper_logging']:
+            self.zookeeper_logger.queue.put(message)
+
         # Set last message variables
         self.last_colour = colour
         self.last_prompt = prompt
+
+
+class ZookeeperLogger(Thread):
+    """
+    Defines a threaded writer for Zookeeper locks. Threading prevents the blocking of other
+    daemon events while the records are written. They will be eventually-consistent
+    """
+    def __init__(self, config):
+        self.config = config
+        self.node = self.config['node']
+        self.max_lines = self.config['node_log_lines']
+        self.queue = Queue()
+        self.zkhandler = None
+        self.start_zkhandler()
+        # Ensure the root keys for this are instantiated
+        self.zkhandler.write([
+            ('base.logs', ''),
+            (('logs', self.node), '')
+        ])
+        self.running = False
+        Thread.__init__(self, args=(), kwargs=None)
+
+    def start_zkhandler(self):
+        # We must open our own dedicated Zookeeper instance because we can't guarantee one already exists when this starts
+        if self.zkhandler is not None:
+            try:
+                self.zkhandler.disconnect()
+            except Exception:
+                pass
+        self.zkhandler = ZKHandler(self.config, logger=None)
+        self.zkhandler.connect(persistent=True)
+
+    def run(self):
+        self.running = True
+        # Get the logs that are currently in Zookeeper and populate our deque
+        raw_logs = self.zkhandler.read(('logs.messages', self.node))
+        if raw_logs is None:
+            raw_logs = ''
+        logs = deque(raw_logs.split('\n'), self.max_lines)
+        while self.running:
+            # Get a new message
+            try:
+                message = self.queue.get(timeout=1)
+                if not message:
+                    continue
+            except Exception:
+                continue
+
+            if not self.config['log_dates']:
+                # We want to log dates here, even if the log_dates config is not set
+                date = '{} '.format(datetime.now().strftime('%Y/%m/%d %H:%M:%S.%f'))
+            else:
+                date = ''
+            # Add the message to the deque
+            logs.append(f'{date}{message}')
+            # Write the updated messages into Zookeeper
+            self.zkhandler.write([(('logs.messages', self.node), '\n'.join(logs))])
+        return
+
+    def stop(self):
+        self.running = False

daemon-common/migrations/versions/3.json

@@ -0,0 +1 @@
+{"version": "3", "root": "", "base": {"root": "", "schema": "/schema", "schema.version": "/schema/version", "config": "/config", "config.maintenance": "/config/maintenance", "config.primary_node": "/config/primary_node", "config.primary_node.sync_lock": "/config/primary_node/sync_lock", "config.upstream_ip": "/config/upstream_ip", "config.migration_target_selector": "/config/migration_target_selector", "cmd": "/cmd", "cmd.node": "/cmd/nodes", "cmd.domain": "/cmd/domains", "cmd.ceph": "/cmd/ceph", "node": "/nodes", "domain": "/domains", "network": "/networks", "storage": "/ceph", "storage.util": "/ceph/util", "osd": "/ceph/osds", "pool": "/ceph/pools", "volume": "/ceph/volumes", "snapshot": "/ceph/snapshots"}, "node": {"name": "", "keepalive": "/keepalive", "mode": "/daemonmode", "data.active_schema": "/activeschema", "data.latest_schema": "/latestschema", "data.static": "/staticdata", "data.pvc_version": "/pvcversion", "running_domains": "/runningdomains", "count.provisioned_domains": "/domainscount", "count.networks": "/networkscount", "state.daemon": "/daemonstate", "state.router": "/routerstate", "state.domain": "/domainstate", "cpu.load": "/cpuload", "vcpu.allocated": "/vcpualloc", "memory.total": "/memtotal", "memory.used": "/memused", "memory.free": "/memfree", "memory.allocated": "/memalloc", "memory.provisioned": "/memprov", "ipmi.hostname": "/ipmihostname", "ipmi.username": "/ipmiusername", "ipmi.password": "/ipmipassword", "sriov": "/sriov", "sriov.pf": "/sriov/pf", "sriov.vf": "/sriov/vf"}, "sriov_pf": {"phy": "", "mtu": "/mtu", "vfcount": "/vfcount"}, "sriov_vf": {"phy": "", "pf": "/pf", "mtu": "/mtu", "mac": "/mac", "phy_mac": "/phy_mac", "config": "/config", "config.vlan_id": "/config/vlan_id", "config.vlan_qos": "/config/vlan_qos", "config.tx_rate_min": "/config/tx_rate_min", "config.tx_rate_max": "/config/tx_rate_max", "config.spoof_check": "/config/spoof_check", "config.link_state": "/config/link_state", "config.trust": "/config/trust", "config.query_rss": "/config/query_rss", "pci": "/pci", "pci.domain": "/pci/domain", "pci.bus": "/pci/bus", "pci.slot": "/pci/slot", "pci.function": "/pci/function", "used": "/used", "used_by": "/used_by"}, "domain": {"name": "", "xml": "/xml", "state": "/state", "profile": "/profile", "stats": "/stats", "node": "/node", "last_node": "/lastnode", "failed_reason": "/failedreason", "storage.volumes": "/rbdlist", "console.log": "/consolelog", "console.vnc": "/vnc", "meta.autostart": "/node_autostart", "meta.migrate_method": "/migration_method", "meta.node_selector": "/node_selector", "meta.node_limit": "/node_limit", "meta.tags": "/tags", "migrate.sync_lock": "/migrate_sync_lock"}, "tag": {"name": "", "type": "/type", "protected": "/protected"}, "network": {"vni": "", "type": "/nettype", "rule": "/firewall_rules", "rule.in": "/firewall_rules/in", "rule.out": "/firewall_rules/out", "nameservers": "/name_servers", "domain": "/domain", "reservation": "/dhcp4_reservations", "lease": "/dhcp4_leases", "ip4.gateway": "/ip4_gateway", "ip4.network": "/ip4_network", "ip4.dhcp": "/dhcp4_flag", "ip4.dhcp_start": "/dhcp4_start", "ip4.dhcp_end": "/dhcp4_end", "ip6.gateway": "/ip6_gateway", "ip6.network": "/ip6_network", "ip6.dhcp": "/dhcp6_flag"}, "reservation": {"mac": "", "ip": "/ipaddr", "hostname": "/hostname"}, "lease": {"mac": "", "ip": "/ipaddr", "hostname": "/hostname", "expiry": "/expiry", "client_id": "/clientid"}, "rule": {"description": "", "rule": "/rule", "order": "/order"}, "osd": {"id": "", "node": "/node", "device": "/device", "stats": "/stats"}, "pool": {"name": "", "pgs": "/pgs", "stats": "/stats"}, "volume": {"name": "", "stats": "/stats"}, "snapshot": {"name": "", "stats": "/stats"}}

daemon-common/migrations/versions/4.json

@@ -0,0 +1 @@
+{"version": "4", "root": "", "base": {"root": "", "schema": "/schema", "schema.version": "/schema/version", "config": "/config", "config.maintenance": "/config/maintenance", "config.primary_node": "/config/primary_node", "config.primary_node.sync_lock": "/config/primary_node/sync_lock", "config.upstream_ip": "/config/upstream_ip", "config.migration_target_selector": "/config/migration_target_selector", "cmd": "/cmd", "cmd.node": "/cmd/nodes", "cmd.domain": "/cmd/domains", "cmd.ceph": "/cmd/ceph", "logs": "/logs", "node": "/nodes", "domain": "/domains", "network": "/networks", "storage": "/ceph", "storage.util": "/ceph/util", "osd": "/ceph/osds", "pool": "/ceph/pools", "volume": "/ceph/volumes", "snapshot": "/ceph/snapshots"}, "logs": {"node": "", "messages": "/messages"}, "node": {"name": "", "keepalive": "/keepalive", "mode": "/daemonmode", "data.active_schema": "/activeschema", "data.latest_schema": "/latestschema", "data.static": "/staticdata", "data.pvc_version": "/pvcversion", "running_domains": "/runningdomains", "count.provisioned_domains": "/domainscount", "count.networks": "/networkscount", "state.daemon": "/daemonstate", "state.router": "/routerstate", "state.domain": "/domainstate", "cpu.load": "/cpuload", "vcpu.allocated": "/vcpualloc", "memory.total": "/memtotal", "memory.used": "/memused", "memory.free": "/memfree", "memory.allocated": "/memalloc", "memory.provisioned": "/memprov", "ipmi.hostname": "/ipmihostname", "ipmi.username": "/ipmiusername", "ipmi.password": "/ipmipassword", "sriov": "/sriov", "sriov.pf": "/sriov/pf", "sriov.vf": "/sriov/vf"}, "sriov_pf": {"phy": "", "mtu": "/mtu", "vfcount": "/vfcount"}, "sriov_vf": {"phy": "", "pf": "/pf", "mtu": "/mtu", "mac": "/mac", "phy_mac": "/phy_mac", "config": "/config", "config.vlan_id": "/config/vlan_id", "config.vlan_qos": "/config/vlan_qos", "config.tx_rate_min": "/config/tx_rate_min", "config.tx_rate_max": "/config/tx_rate_max", "config.spoof_check": "/config/spoof_check", "config.link_state": "/config/link_state", "config.trust": "/config/trust", "config.query_rss": "/config/query_rss", "pci": "/pci", "pci.domain": "/pci/domain", "pci.bus": "/pci/bus", "pci.slot": "/pci/slot", "pci.function": "/pci/function", "used": "/used", "used_by": "/used_by"}, "domain": {"name": "", "xml": "/xml", "state": "/state", "profile": "/profile", "stats": "/stats", "node": "/node", "last_node": "/lastnode", "failed_reason": "/failedreason", "storage.volumes": "/rbdlist", "console.log": "/consolelog", "console.vnc": "/vnc", "meta.autostart": "/node_autostart", "meta.migrate_method": "/migration_method", "meta.node_selector": "/node_selector", "meta.node_limit": "/node_limit", "meta.tags": "/tags", "migrate.sync_lock": "/migrate_sync_lock"}, "tag": {"name": "", "type": "/type", "protected": "/protected"}, "network": {"vni": "", "type": "/nettype", "rule": "/firewall_rules", "rule.in": "/firewall_rules/in", "rule.out": "/firewall_rules/out", "nameservers": "/name_servers", "domain": "/domain", "reservation": "/dhcp4_reservations", "lease": "/dhcp4_leases", "ip4.gateway": "/ip4_gateway", "ip4.network": "/ip4_network", "ip4.dhcp": "/dhcp4_flag", "ip4.dhcp_start": "/dhcp4_start", "ip4.dhcp_end": "/dhcp4_end", "ip6.gateway": "/ip6_gateway", "ip6.network": "/ip6_network", "ip6.dhcp": "/dhcp6_flag"}, "reservation": {"mac": "", "ip": "/ipaddr", "hostname": "/hostname"}, "lease": {"mac": "", "ip": "/ipaddr", "hostname": "/hostname", "expiry": "/expiry", "client_id": "/clientid"}, "rule": {"description": "", "rule": "/rule", "order": "/order"}, "osd": {"id": "", "node": "/node", "device": "/device", "stats": "/stats"}, "pool": {"name": "", "pgs": "/pgs", "stats": "/stats"}, "volume": {"name": "", "stats": "/stats"}, "snapshot": {"name": "", "stats": "/stats"}}

daemon-common/node.py

@@ -182,6 +182,24 @@ def ready_node(zkhandler, node, wait=False):
     return True, retmsg
 
 
+def get_node_log(zkhandler, node, lines=2000):
+    # Verify node is valid
+    if not common.verifyNode(zkhandler, node):
+        return False, 'ERROR: No node named "{}" is present in the cluster.'.format(node)
+
+    # Get the data from ZK
+    node_log = zkhandler.read(('logs.messages', node))
+
+    if node_log is None:
+        return True, ''
+
+    # Shrink the log buffer to length lines
+    shrunk_log = node_log.split('\n')[-lines:]
+    loglines = '\n'.join(shrunk_log)
+
+    return True, loglines
+
+
 def get_info(zkhandler, node):
     # Verify node is valid
     if not common.verifyNode(zkhandler, node):

daemon-common/vm.py

@@ -24,6 +24,7 @@ import re
 import lxml.objectify
 import lxml.etree
 
+from distutils.util import strtobool
 from uuid import UUID
 from concurrent.futures import ThreadPoolExecutor
 
@@ -174,7 +175,7 @@ def flush_locks(zkhandler, domain):
     return success, message
 
 
-def define_vm(zkhandler, config_data, target_node, node_limit, node_selector, node_autostart, migration_method=None, profile=None, initial_state='stop'):
+def define_vm(zkhandler, config_data, target_node, node_limit, node_selector, node_autostart, migration_method=None, profile=None, tags=[], initial_state='stop'):
     # Parse the XML data
     try:
         parsed_xml = lxml.objectify.fromstring(config_data)
@@ -246,9 +247,18 @@ def define_vm(zkhandler, config_data, target_node, node_limit, node_selector, no
         (('domain.meta.migrate_method', dom_uuid), migration_method),
         (('domain.meta.node_limit', dom_uuid), formatted_node_limit),
         (('domain.meta.node_selector', dom_uuid), node_selector),
+        (('domain.meta.tags', dom_uuid), ''),
         (('domain.migrate.sync_lock', dom_uuid), ''),
     ])
 
+    for tag in tags:
+        tag_name = tag['name']
+        zkhandler.write([
+            (('domain.meta.tags', dom_uuid, 'tag.name', tag_name), tag['name']),
+            (('domain.meta.tags', dom_uuid, 'tag.type', tag_name), tag['type']),
+            (('domain.meta.tags', dom_uuid, 'tag.protected', tag_name), tag['protected']),
+        ])
+
     return True, 'Added new VM with Name "{}" and UUID "{}" to database.'.format(dom_name, dom_uuid)
 
 
@@ -282,6 +292,38 @@ def modify_vm_metadata(zkhandler, domain, node_limit, node_selector, node_autost
     return True, 'Successfully modified PVC metadata of VM "{}".'.format(domain)
 
 
+def modify_vm_tag(zkhandler, domain, action, tag, protected=False):
+    dom_uuid = getDomainUUID(zkhandler, domain)
+    if not dom_uuid:
+        return False, 'ERROR: Could not find VM "{}" in the cluster!'.format(domain)
+
+    if action == 'add':
+        zkhandler.write([
+            (('domain.meta.tags', dom_uuid, 'tag.name', tag), tag),
+            (('domain.meta.tags', dom_uuid, 'tag.type', tag), 'user'),
+            (('domain.meta.tags', dom_uuid, 'tag.protected', tag), protected),
+        ])
+
+        return True, 'Successfully added tag "{}" to VM "{}".'.format(tag, domain)
+    elif action == 'remove':
+        if not zkhandler.exists(('domain.meta.tags', dom_uuid, 'tag', tag)):
+            return False, 'The tag "{}" does not exist.'.format(tag)
+
+        if zkhandler.read(('domain.meta.tags', dom_uuid, 'tag.type', tag)) != 'user':
+            return False, 'The tag "{}" is not a user tag and cannot be removed.'.format(tag)
+
+        if bool(strtobool(zkhandler.read(('domain.meta.tags', dom_uuid, 'tag.protected', tag)))):
+            return False, 'The tag "{}" is protected and cannot be removed.'.format(tag)
+
+        zkhandler.delete([
+            (('domain.meta.tags', dom_uuid, 'tag', tag))
+        ])
+
+        return True, 'Successfully removed tag "{}" from VM "{}".'.format(tag, domain)
+    else:
+        return False, 'Specified tag action is not available.'
+
+
 def modify_vm(zkhandler, domain, restart, new_vm_config):
     dom_uuid = getDomainUUID(zkhandler, domain)
     if not dom_uuid:
@@ -403,7 +445,7 @@ def rename_vm(zkhandler, domain, new_domain):
     undefine_vm(zkhandler, dom_uuid)
 
     # Define the new VM
-    define_vm(zkhandler, vm_config_new, dom_info['node'], dom_info['node_limit'], dom_info['node_selector'], dom_info['node_autostart'], migration_method=dom_info['migration_method'], profile=dom_info['profile'], initial_state='stop')
+    define_vm(zkhandler, vm_config_new, dom_info['node'], dom_info['node_limit'], dom_info['node_selector'], dom_info['node_autostart'], migration_method=dom_info['migration_method'], profile=dom_info['profile'], tags=dom_info['tags'], initial_state='stop')
 
     # If the VM is migrated, store that
     if dom_info['migrated'] != 'no':
@@ -824,7 +866,7 @@ def get_info(zkhandler, domain):
     return True, domain_information
 
 
-def get_list(zkhandler, node, state, limit, is_fuzzy=True):
+def get_list(zkhandler, node, state, tag, limit, is_fuzzy=True):
     if node:
         # Verify node is valid
         if not common.verifyNode(zkhandler, node):
@@ -862,6 +904,7 @@ def get_list(zkhandler, node, state, limit, is_fuzzy=True):
     for vm in full_vm_list:
         name = zkhandler.read(('domain', vm))
         is_limit_match = False
+        is_tag_match = False
         is_node_match = False
         is_state_match = False
 
@@ -878,6 +921,13 @@ def get_list(zkhandler, node, state, limit, is_fuzzy=True):
         else:
             is_limit_match = True
 
+        if tag:
+            vm_tags = zkhandler.children(('domain.meta.tags', vm))
+            if tag in vm_tags:
+                is_tag_match = True
+        else:
+            is_tag_match = True
+
         # Check on node
         if node:
             vm_node = zkhandler.read(('domain.node', vm))
@@ -894,7 +944,7 @@ def get_list(zkhandler, node, state, limit, is_fuzzy=True):
         else:
             is_state_match = True
 
-        get_vm_info[vm] = True if is_limit_match and is_node_match and is_state_match else False
+        get_vm_info[vm] = True if is_limit_match and is_tag_match and is_node_match and is_state_match else False
 
     # Obtain our VM data in a thread pool
     # This helps parallelize the numerous Zookeeper calls a bit, within the bounds of the GIL, and

daemon-common/zkhandler.py

@@ -140,13 +140,13 @@ class ZKHandler(object):
         """
         try:
             self.zk_conn.start()
-            self.log('Connection to Zookeeper started', state='o')
             if persistent:
+                self.log('Connection to Zookeeper started', state='o')
                 self.zk_conn.add_listener(self.listener)
         except Exception as e:
             raise ZKConnectionException(self, e)
 
-    def disconnect(self):
+    def disconnect(self, persistent=False):
         """
         Stop and close the zk_conn object and disconnect from the cluster
 
@@ -154,6 +154,7 @@ class ZKHandler(object):
         """
         self.zk_conn.stop()
         self.zk_conn.close()
+        if persistent:
             self.log('Connection to Zookeeper terminated', state='o')
 
     #
@@ -465,7 +466,7 @@ class ZKHandler(object):
 #
 class ZKSchema(object):
     # Current version
-    _version = 2
+    _version = 4
 
     # Root for doing nested keys
     _schema_root = ''
@@ -489,6 +490,7 @@ class ZKSchema(object):
             'cmd.node': f'{_schema_root}/cmd/nodes',
             'cmd.domain': f'{_schema_root}/cmd/domains',
             'cmd.ceph': f'{_schema_root}/cmd/ceph',
+            'logs': '/logs',
             'node': f'{_schema_root}/nodes',
             'domain': f'{_schema_root}/domains',
             'network': f'{_schema_root}/networks',
@@ -499,6 +501,11 @@ class ZKSchema(object):
             'volume': f'{_schema_root}/ceph/volumes',
             'snapshot': f'{_schema_root}/ceph/snapshots',
         },
+        # The schema of an individual logs entry (/logs/{node_name})
+        'logs': {
+            'node': '',  # The root key
+            'messages': '/messages',
+        },
         # The schema of an individual node entry (/nodes/{node_name})
         'node': {
             'name': '',  # The root key
@@ -575,8 +582,15 @@ class ZKSchema(object):
             'meta.migrate_method': '/migration_method',
             'meta.node_selector': '/node_selector',
             'meta.node_limit': '/node_limit',
+            'meta.tags': '/tags',
             'migrate.sync_lock': '/migrate_sync_lock'
         },
+        # The schema of an individual domain tag entry (/domains/{domain}/tags/{tag})
+        'tag': {
+            'name': '',  # The root key
+            'type': '/type',
+            'protected': '/protected'
+        },
         # The schema of an individual network entry (/networks/{vni})
         'network': {
             'vni': '',  # The root key
@@ -763,7 +777,7 @@ class ZKSchema(object):
                         logger.out(f'Key not found: {self.path(kpath)}', state='w')
                     result = False
 
-        for elem in ['node', 'domain', 'network', 'osd', 'pool']:
+        for elem in ['logs', 'node', 'domain', 'network', 'osd', 'pool']:
             # First read all the subelements of the key class
             for child in zkhandler.zk_conn.get_children(self.path(f'base.{elem}')):
                 # For each key in the schema for that particular elem
@@ -842,7 +856,7 @@ class ZKSchema(object):
                         data = ''
                     zkhandler.zk_conn.create(self.path(kpath), data.encode(zkhandler.encoding))
 
-        for elem in ['node', 'domain', 'network', 'osd', 'pool']:
+        for elem in ['logs', 'node', 'domain', 'network', 'osd', 'pool']:
             # First read all the subelements of the key class
             for child in zkhandler.zk_conn.get_children(self.path(f'base.{elem}')):
                 # For each key in the schema for that particular elem

debian/changelog

@@ -1,3 +1,21 @@
+pvc (0.9.26-0) unstable; urgency=high
+
+  * [Node Daemon] Corrects some bad assumptions about fencing results during hardware failures
+  * [All] Implements VM tagging functionality
+  * [All] Implements Node log access via PVC functionality
+
+ -- Joshua M. Boniface <joshua@boniface.me>  Sun, 18 Jul 2021 20:49:52 -0400
+
+pvc (0.9.25-0) unstable; urgency=high
+
+  * [Node Daemon] Returns to Rados library calls for Ceph due to performance problems
+  * [Node Daemon] Adds a date output to keepalive messages
+  * [Daemons] Configures ZK connection logging only for persistent connections
+  * [API Provisioner] Add context manager-based chroot to Debootstrap example script
+  * [Node Daemon] Fixes a bug where shutdown daemon state was overwritten
+
+ -- Joshua M. Boniface <joshua@boniface.me>  Sun, 11 Jul 2021 23:19:09 -0400
+
 pvc (0.9.24-0) unstable; urgency=high
 
   * [Node Daemon] Removes Rados module polling of Ceph cluster and returns to command-based polling for timeout purposes, and removes some flaky return statements

debian/control

@@ -8,7 +8,7 @@ X-Python3-Version: >= 3.2
 
 Package: pvc-daemon-node
 Architecture: all
-Depends: systemd, pvc-daemon-common, python3-kazoo, python3-psutil, python3-apscheduler, python3-libvirt, python3-psycopg2, python3-dnspython, python3-yaml, python3-distutils, python3-gevent, ipmitool, libvirt-daemon-system, arping, vlan, bridge-utils, dnsmasq, nftables, pdns-server, pdns-backend-pgsql
+Depends: systemd, pvc-daemon-common, python3-kazoo, python3-psutil, python3-apscheduler, python3-libvirt, python3-psycopg2, python3-dnspython, python3-yaml, python3-distutils, python3-rados, python3-gevent, ipmitool, libvirt-daemon-system, arping, vlan, bridge-utils, dnsmasq, nftables, pdns-server, pdns-backend-pgsql
 Suggests: pvc-client-api, pvc-client-cli
 Description: Parallel Virtual Cluster node daemon (Python 3)
  A KVM/Zookeeper/Ceph-based VM and private cloud manager

docs/index.md

@@ -42,6 +42,20 @@ To get started with PVC, please see the [About](https://parallelvirtualcluster.r
 
 ## Changelog
 
+#### v0.9.26
+
+  * [Node Daemon] Corrects some bad assumptions about fencing results during hardware failures
+  * [All] Implements VM tagging functionality
+  * [All] Implements Node log access via PVC functionality
+
+#### v0.9.25
+
+  * [Node Daemon] Returns to Rados library calls for Ceph due to performance problems
+  * [Node Daemon] Adds a date output to keepalive messages
+  * [Daemons] Configures ZK connection logging only for persistent connections
+  * [API Provisioner] Add context manager-based chroot to Debootstrap example script
+  * [Node Daemon] Fixes a bug where shutdown daemon state was overwritten
+
 #### v0.9.24
 
   * [Node Daemon] Removes Rados module polling of Ceph cluster and returns to command-based polling for timeout purposes, and removes some flaky return statements

docs/manuals/swagger.json

@@ -144,6 +144,19 @@
             },
             "type": "object"
         },
+        "NodeLog": {
+            "properties": {
+                "data": {
+                    "description": "The recent log text",
+                    "type": "string"
+                },
+                "name": {
+                    "description": "The name of the Node",
+                    "type": "string"
+                }
+            },
+            "type": "object"
+        },
         "VMLog": {
             "properties": {
                 "data": {
@@ -215,6 +228,23 @@
             },
             "type": "object"
         },
+        "VMTags": {
+            "properties": {
+                "name": {
+                    "description": "The name of the VM",
+                    "type": "string"
+                },
+                "tags": {
+                    "description": "The tag(s) of the VM",
+                    "items": {
+                        "id": "VMTag",
+                        "type": "object"
+                    },
+                    "type": "array"
+                }
+            },
+            "type": "object"
+        },
         "acl": {
             "properties": {
                 "description": {
@@ -1370,6 +1400,28 @@
                     "description": "The current state of the VM",
                     "type": "string"
                 },
+                "tags": {
+                    "description": "The tag(s) of the VM",
+                    "items": {
+                        "id": "VMTag",
+                        "properties": {
+                            "name": {
+                                "description": "The name of the tag",
+                                "type": "string"
+                            },
+                            "protected": {
+                                "description": "Whether the tag is protected or not",
+                                "type": "boolean"
+                            },
+                            "type": {
+                                "description": "The type of the tag (user, system)",
+                                "type": "string"
+                            }
+                        },
+                        "type": "object"
+                    },
+                    "type": "array"
+                },
                 "type": {
                     "description": "The type of the VM",
                     "type": "string"
@@ -2415,7 +2467,7 @@
                 "description": "",
                 "parameters": [
                     {
-                        "description": "A search limit; fuzzy by default, use ^/$ to force exact matches",
+                        "description": "A search limit in the name, tags, or an exact UUID; fuzzy by default, use ^/$ to force exact matches",
                         "in": "query",
                         "name": "limit",
                         "required": false,
@@ -2626,6 +2678,38 @@
                 ]
             }
         },
+        "/api/v1/node/{node}/log": {
+            "get": {
+                "description": "",
+                "parameters": [
+                    {
+                        "description": "The number of lines to retrieve",
+                        "in": "query",
+                        "name": "lines",
+                        "required": false,
+                        "type": "integer"
+                    }
+                ],
+                "responses": {
+                    "200": {
+                        "description": "OK",
+                        "schema": {
+                            "$ref": "#/definitions/NodeLog"
+                        }
+                    },
+                    "404": {
+                        "description": "Node not found",
+                        "schema": {
+                            "$ref": "#/definitions/Message"
+                        }
+                    }
+                },
+                "summary": "Return the recent logs of {node}",
+                "tags": [
+                    "node"
+                ]
+            }
+        },
         "/api/v1/provisioner/create": {
             "post": {
                 "description": "Note: Starts a background job in the pvc-provisioner-worker Celery worker while returning a task ID; the task ID can be used to query the \"GET /provisioner/status/<task_id>\" endpoint for the job status",
@@ -5795,7 +5879,7 @@
                 "description": "",
                 "parameters": [
                     {
-                        "description": "A name search limit; fuzzy by default, use ^/$ to force exact matches",
+                        "description": "A search limit in the name, tags, or an exact UUID; fuzzy by default, use ^/$ to force exact matches",
                         "in": "query",
                         "name": "limit",
                         "required": false,
@@ -5814,6 +5898,13 @@
                         "name": "state",
                         "required": false,
                         "type": "string"
+                    },
+                    {
+                        "description": "Limit list to VMs with this tag",
+                        "in": "query",
+                        "name": "tag",
+                        "required": false,
+                        "type": "string"
                     }
                 ],
                 "responses": {
@@ -5889,6 +5980,26 @@
                         "name": "migration_method",
                         "required": false,
                         "type": "string"
+                    },
+                    {
+                        "description": "The user tag(s) of the VM",
+                        "in": "query",
+                        "items": {
+                            "type": "string"
+                        },
+                        "name": "user_tags",
+                        "required": false,
+                        "type": "array"
+                    },
+                    {
+                        "description": "The protected user tag(s) of the VM",
+                        "in": "query",
+                        "items": {
+                            "type": "string"
+                        },
+                        "name": "protected_tags",
+                        "required": false,
+                        "type": "array"
                     }
                 ],
                 "responses": {
@@ -6027,6 +6138,26 @@
                         "name": "migration_method",
                         "required": false,
                         "type": "string"
+                    },
+                    {
+                        "description": "The user tag(s) of the VM",
+                        "in": "query",
+                        "items": {
+                            "type": "string"
+                        },
+                        "name": "user_tags",
+                        "required": false,
+                        "type": "array"
+                    },
+                    {
+                        "description": "The protected user tag(s) of the VM",
+                        "in": "query",
+                        "items": {
+                            "type": "string"
+                        },
+                        "name": "protected_tags",
+                        "required": false,
+                        "type": "array"
                     }
                 ],
                 "responses": {
@@ -6151,7 +6282,7 @@
                         }
                     },
                     "404": {
-                        "description": "Not found",
+                        "description": "VM not found",
                         "schema": {
                             "$ref": "#/definitions/Message"
                         }
@@ -6225,6 +6356,12 @@
                         "schema": {
                             "$ref": "#/definitions/Message"
                         }
+                    },
+                    "404": {
+                        "description": "VM not found",
+                        "schema": {
+                            "$ref": "#/definitions/Message"
+                        }
                     }
                 },
                 "summary": "Set the metadata of {vm}",
@@ -6412,6 +6549,84 @@
                     "vm"
                 ]
             }
+        },
+        "/api/v1/vm/{vm}/tags": {
+            "get": {
+                "description": "",
+                "responses": {
+                    "200": {
+                        "description": "OK",
+                        "schema": {
+                            "$ref": "#/definitions/VMTags"
+                        }
+                    },
+                    "404": {
+                        "description": "VM not found",
+                        "schema": {
+                            "$ref": "#/definitions/Message"
+                        }
+                    }
+                },
+                "summary": "Return the tags of {vm}",
+                "tags": [
+                    "vm"
+                ]
+            },
+            "post": {
+                "description": "",
+                "parameters": [
+                    {
+                        "description": "The action to perform with the tag",
+                        "enum": [
+                            "add",
+                            "remove"
+                        ],
+                        "in": "query",
+                        "name": "action",
+                        "required": true,
+                        "type": "string"
+                    },
+                    {
+                        "description": "The text value of the tag",
+                        "in": "query",
+                        "name": "tag",
+                        "required": true,
+                        "type": "string"
+                    },
+                    {
+                        "default": false,
+                        "description": "Set the protected state of the tag",
+                        "in": "query",
+                        "name": "protected",
+                        "required": false,
+                        "type": "boolean"
+                    }
+                ],
+                "responses": {
+                    "200": {
+                        "description": "OK",
+                        "schema": {
+                            "$ref": "#/definitions/Message"
+                        }
+                    },
+                    "400": {
+                        "description": "Bad request",
+                        "schema": {
+                            "$ref": "#/definitions/Message"
+                        }
+                    },
+                    "404": {
+                        "description": "VM not found",
+                        "schema": {
+                            "$ref": "#/definitions/Message"
+                        }
+                    }
+                },
+                "summary": "Set the tags of {vm}",
+                "tags": [
+                    "vm"
+                ]
+            }
         }
     },
     "swagger": "2.0"

node-daemon/pvcnoded.sample.yaml

@@ -140,6 +140,8 @@ pvc:
         file_logging: True
         # stdout_logging: Enable or disable logging to stdout (i.e. journald)
         stdout_logging: True
+        # zookeeper_logging: Enable ot disable logging to Zookeeper (for `pvc node log` functionality)
+        zookeeper_logging: True
         # log_colours: Enable or disable ANSI colours in log output
         log_colours: True
         # log_dates: Enable or disable date strings in log output
@@ -152,10 +154,12 @@ pvc:
         log_keepalive_storage_details: True
         # console_log_lines: Number of console log lines to store in Zookeeper per VM
         console_log_lines: 1000
+        # node_log_lines: Number of node log lines to store in Zookeeper per node
+        node_log_lines: 2000
       # networking: PVC networking configuration
       # OPTIONAL if enable_networking: False
       networking:
-        # bridge_device: Underlying device to use for bridged vLAN networks; usually the device underlying <cluster>
+        # bridge_device: Underlying device to use for bridged vLAN networks; usually the device of <cluster>
         bridge_device: ens4
         # sriov_enable: Enable or disable (default if absent) SR-IOV network support
         sriov_enable: False

node-daemon/pvcnoded/Daemon.py

@@ -32,12 +32,14 @@ import yaml
 import json
 
 from socket import gethostname
+from datetime import datetime
 from threading import Thread
 from ipaddress import ip_address, ip_network
 from apscheduler.schedulers.background import BackgroundScheduler
 from distutils.util import strtobool
 from queue import Queue
 from xml.etree import ElementTree
+from rados import Rados
 
 from daemon_lib.zkhandler import ZKHandler
 
@@ -54,7 +56,7 @@ import pvcnoded.CephInstance as CephInstance
 import pvcnoded.MetadataAPIInstance as MetadataAPIInstance
 
 # Version string for startup output
-version = '0.9.24'
+version = '0.9.26'
 
 ###############################################################################
 # PVCD - node daemon startup program
@@ -145,6 +147,7 @@ def readConfig(pvcnoded_config_file, myhostname):
     # Handle the basic config (hypervisor-only)
     try:
         config_general = {
+            'node': o_config['pvc']['node'],
             'coordinators': o_config['pvc']['cluster']['coordinators'],
             'enable_hypervisor': o_config['pvc']['functions']['enable_hypervisor'],
             'enable_networking': o_config['pvc']['functions']['enable_networking'],
@@ -155,12 +158,14 @@ def readConfig(pvcnoded_config_file, myhostname):
             'console_log_directory': o_config['pvc']['system']['configuration']['directories']['console_log_directory'],
             'file_logging': o_config['pvc']['system']['configuration']['logging']['file_logging'],
             'stdout_logging': o_config['pvc']['system']['configuration']['logging']['stdout_logging'],
+            'zookeeper_logging': o_config['pvc']['system']['configuration']['logging'].get('zookeeper_logging', False),
             'log_colours': o_config['pvc']['system']['configuration']['logging']['log_colours'],
             'log_dates': o_config['pvc']['system']['configuration']['logging']['log_dates'],
             'log_keepalives': o_config['pvc']['system']['configuration']['logging']['log_keepalives'],
             'log_keepalive_cluster_details': o_config['pvc']['system']['configuration']['logging']['log_keepalive_cluster_details'],
             'log_keepalive_storage_details': o_config['pvc']['system']['configuration']['logging']['log_keepalive_storage_details'],
             'console_log_lines': o_config['pvc']['system']['configuration']['logging']['console_log_lines'],
+            'node_log_lines': o_config['pvc']['system']['configuration']['logging'].get('node_log_lines', 0),
             'vm_shutdown_timeout': int(o_config['pvc']['system']['intervals']['vm_shutdown_timeout']),
             'keepalive_interval': int(o_config['pvc']['system']['intervals']['keepalive_interval']),
             'fence_intervals': int(o_config['pvc']['system']['intervals']['fence_intervals']),
@@ -657,7 +662,7 @@ def update_schema(new_schema_version, stat, event=''):
     # Restart ourselves with the new schema
     logger.out('Reloading node daemon', state='s')
     try:
-        zkhandler.disconnect()
+        zkhandler.disconnect(persistent=True)
         del zkhandler
     except Exception:
         pass
@@ -692,7 +697,7 @@ else:
 
 # Cleanup function
 def cleanup():
-    global zkhandler, update_timer, d_domain
+    global logger, zkhandler, update_timer, d_domain
 
     logger.out('Terminating pvcnoded and cleaning up', state='s')
 
@@ -750,12 +755,14 @@ def cleanup():
 
     # Close the Zookeeper connection
     try:
-        zkhandler.disconnect()
+        zkhandler.disconnect(persistent=True)
         del zkhandler
     except Exception:
         pass
 
     logger.out('Terminated pvc daemon', state='s')
+    logger.terminate()
+
     os._exit(0)
 
 
@@ -1313,13 +1320,24 @@ def collect_ceph_stats(queue):
     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
-    _, stdout, _ = common.run_os_command('ceph health --format json', timeout=1)
+    command = {"prefix": "health", "format": "json"}
     try:
-        ceph_health = json.loads(stdout)['status']
+        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'
@@ -1338,7 +1356,8 @@ def collect_ceph_stats(queue):
         if debug:
             logger.out("Set ceph health information in zookeeper (primary only)", state='d', prefix='ceph-thread')
 
-        _, ceph_status, _ = common.run_os_command('ceph status --format plain', timeout=1)
+        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))
@@ -1350,7 +1369,8 @@ def collect_ceph_stats(queue):
             logger.out("Set ceph rados df information in zookeeper (primary only)", state='d', prefix='ceph-thread')
 
         # Get rados df info
-        _, ceph_df, _ = common.run_os_command('ceph df --format plain', timeout=1)
+        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))
@@ -1362,14 +1382,15 @@ def collect_ceph_stats(queue):
             logger.out("Set pool information in zookeeper (primary only)", state='d', prefix='ceph-thread')
 
         # Get pool info
-        _, stdout, _ = common.run_os_command('ceph df --format json', timeout=1)
+        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(stdout)['pools']
+            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 = []
 
-        _, stdout, _ = common.run_os_command('rados df --format json', timeout=1)
+        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:
@@ -1434,9 +1455,10 @@ def collect_ceph_stats(queue):
         # Parse the dump data
         osd_dump = dict()
 
-        _, stdout, _ = common.run_os_command('ceph osd dump --format json --connect-timeout 1', timeout=1)
+        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(stdout)['osds']
+            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 = []
@@ -1459,9 +1481,9 @@ def collect_ceph_stats(queue):
 
         osd_df = dict()
 
-        _, osd_df_out, _ = common.run_os_command('ceph osd df --format json', timeout=1)
+        command = {"prefix": "osd df", "format": "json"}
         try:
-            osd_df_raw = json.loads(osd_df_out)['nodes']
+            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 = []
@@ -1486,10 +1508,12 @@ def collect_ceph_stats(queue):
 
         osd_status = dict()
 
-        retcode, osd_status_raw, stderr = common.run_os_command('ceph osd status --format plain', timeout=1)
-        if retcode != 0:
-            logger.out('Failed to obtain OSD status data: {}'.format(stderr), state='w')
-            osd_status_raw = ''
+        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')
@@ -1556,6 +1580,8 @@ def collect_ceph_stats(queue):
                     # 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)
@@ -1758,8 +1784,9 @@ def node_keepalive():
     # 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':
+    if past_state != 'run' and past_state != 'shutdown':
         this_node.daemon_state = 'run'
         zkhandler.write([
             (('node.state.daemon', this_node.name), 'run')
@@ -1858,9 +1885,10 @@ def node_keepalive():
         else:
             cst_colour = fmt_cyan
         logger.out(
-            '{}{} keepalive{} [{}{}{}]'.format(
+            '{}{} keepalive @ {}{} [{}{}{}]'.format(
                 fmt_purple,
                 myhostname,
+                datetime.now(),
                 fmt_end,
                 fmt_bold + cst_colour,
                 this_node.router_state,

node-daemon/pvcnoded/MetadataAPIInstance.py

@@ -180,7 +180,7 @@ class MetadataAPIInstance(object):
         client_macaddr = host_information.get('mac_address', None)
 
         # Find the VM with that MAC address - we can't assume that the hostname is actually right
-        _discard, vm_list = pvc_vm.get_list(self.zkhandler, None, None, None)
+        _discard, vm_list = pvc_vm.get_list(self.zkhandler, None, None, None, None)
         vm_details = dict()
         for vm in vm_list:
             try:

node-daemon/pvcnoded/fencing.py

@@ -133,31 +133,46 @@ def rebootViaIPMI(ipmi_hostname, ipmi_user, ipmi_password, logger):
     if ipmi_reset_retcode != 0:
         logger.out('Failed to reboot dead node', state='e')
         print(ipmi_reset_stderr)
-        return False
 
-    time.sleep(2)
+    time.sleep(1)
 
-    # Ensure the node is powered on
-    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)
-
-    # Trigger a power start if needed
-    if ipmi_status_stdout != "Chassis Power is on":
+    # 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)
 
-        if ipmi_start_retcode != 0:
-            logger.out('Failed to start powered-off dead node', state='e')
-            print(ipmi_reset_stderr)
-            return False
+    time.sleep(2)
 
-    # Declare success
+    # 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
 
 
 #

test-cluster.sh

@@ -38,7 +38,7 @@ sleep 30
 _pvc vm stop --yes testX
 _pvc vm disable testX
 _pvc vm undefine --yes testX
-_pvc vm define --target hv3 ${vm_tmp}
+_pvc vm define --target hv3 --tag pvc-test ${vm_tmp}
 _pvc vm start testX
 sleep 30
 _pvc vm restart --yes --wait testX
@@ -50,6 +50,10 @@ sleep 5
 _pvc vm move --wait --target hv1 testX
 sleep 5
 _pvc vm meta testX --limit hv1 --selector vms --method live --profile test --no-autostart
+_pvc vm tag add testX mytag
+_pvc vm tag get testX
+_pvc vm list --tag mytag
+_pvc vm tag remove testX mytag
 _pvc vm network get testX
 _pvc vm vcpu set testX 4
 _pvc vm vcpu get testX