Bump version to 0.9.53

Due to the executor the previous attempt did not work.

.version

@@ -1 +1 @@
-0.9.47
+0.9.53

CHANGELOG.md

@@ -1,5 +1,41 @@
 ## PVC Changelog
 
+###### [v0.9.53](https://github.com/parallelvirtualcluster/pvc/releases/tag/v0.9.53)
+
+  * [API] Fixes sort order of VM list (for real this time)
+
+###### [v0.9.52](https://github.com/parallelvirtualcluster/pvc/releases/tag/v0.9.52)
+
+  * [CLI] Fixes a bug with vm modify not requiring a cluster
+  * [Docs] Adds a reference to the bootstrap daemon
+  * [API] Adds sorting to node and VM lists for consistency
+  * [Node Daemon/API] Adds kb_ stats values for OSD stats
+
+###### [v0.9.51](https://github.com/parallelvirtualcluster/pvc/releases/tag/v0.9.51)
+
+  * [CLI Client] Fixes a faulty literal_eval when viewing task status
+  * [CLI Client] Adds a confirmation flag to the vm disable command
+  * [Node Daemon] Removes the pvc-flush service
+
+###### [v0.9.50](https://github.com/parallelvirtualcluster/pvc/releases/tag/v0.9.50)
+
+  * [Node Daemon/API/CLI] Adds free memory node selector
+  * [Node Daemon] Fixes bug sending space-containing detect disk strings
+
+###### [v0.9.49](https://github.com/parallelvirtualcluster/pvc/releases/tag/v0.9.49)
+
+  * [Node Daemon] Fixes bugs with OSD stat population on creation
+  * [Node Daemon/API] Adds additional information to Zookeeper about OSDs
+  * [Node Daemon] Refactors OSD removal for improved safety
+  * [Node Daemon/API/CLI] Adds explicit support for replacing and refreshing (reimporting) OSDs
+  * [API/CLI] Fixes a language inconsistency about "router mode"
+
+###### [v0.9.48](https://github.com/parallelvirtualcluster/pvc/releases/tag/v0.9.48)
+
+  * [CLI] Fixes situation where only a single cluster is available
+  * [CLI/API/Daemon] Allows forcing of OSD removal ignoring errors
+  * [CLI] Fixes bug where down OSDs are not displayed
+
 ###### [v0.9.47](https://github.com/parallelvirtualcluster/pvc/releases/tag/v0.9.47)
 
   * [Node Daemon/API/CLI] Adds Ceph pool device class/tier support

README.md

@@ -19,7 +19,7 @@ As a consequence of its features, PVC makes administrating very high-uptime VMs
 
 PVC also features an optional, fully customizable VM provisioning framework, designed to automate and simplify VM deployments using custom provisioning profiles, scripts, and CloudInit userdata API support.
 
-Installation of PVC is accomplished by two main components: a [Node installer ISO](https://github.com/parallelvirtualcluster/pvc-installer) which creates on-demand installer ISOs, and an [Ansible role framework](https://github.com/parallelvirtualcluster/pvc-ansible) to configure, bootstrap, and administrate the nodes. Once up, the cluster is managed via an HTTP REST API, accessible via a Python Click CLI client or WebUI.
+Installation of PVC is accomplished by two main components: a [Node installer ISO](https://github.com/parallelvirtualcluster/pvc-installer) which creates on-demand installer ISOs, and an [Ansible role framework](https://github.com/parallelvirtualcluster/pvc-ansible) to configure, bootstrap, and administrate the nodes. Installation can also be fully automated with a companion [cluster bootstrapping system](https://github.com/parallelvirtualcluster/pvc-bootstrap). Once up, the cluster is managed via an HTTP REST API, accessible via a Python Click CLI client or WebUI.
 
 Just give it physical servers, and it will run your VMs without you having to think about it, all in just an hour or two of setup time.
 

api-daemon/pvcapid/Daemon.py

@@ -25,7 +25,7 @@ import yaml
 from distutils.util import strtobool as dustrtobool
 
 # Daemon version
-version = "0.9.47"
+version = "0.9.53"
 
 # API version
 API_VERSION = 1.0

api-daemon/pvcapid/flaskapi.py

@@ -1002,7 +1002,7 @@ class API_VM_Root(Resource):
                     type: string
                 node_selector:
                   type: string
-                  description: The selector used to determine candidate nodes during migration
+                  description: The selector used to determine candidate nodes during migration; see 'target_selector' in the node daemon configuration reference
                 node_autostart:
                   type: boolean
                   description: Whether to autostart the VM when its node returns to ready domain state
@@ -1252,7 +1252,7 @@ class API_VM_Root(Resource):
             {"name": "node"},
             {
                 "name": "selector",
-                "choices": ("mem", "vcpus", "load", "vms", "none"),
+                "choices": ("mem", "memfree", "vcpus", "load", "vms", "none"),
                 "helptext": "A valid selector must be specified",
             },
             {"name": "autostart"},
@@ -1297,13 +1297,15 @@ class API_VM_Root(Resource):
             name: selector
             type: string
             required: false
-            description: The selector used to determine candidate nodes during migration
-            default: mem
+            description: The selector used to determine candidate nodes during migration; see 'target_selector' in the node daemon configuration reference
+            default: none
             enum:
               - mem
+              - memfree
               - vcpus
               - load
               - vms
+              - none (cluster default)
           - in: query
             name: autostart
             type: boolean
@@ -1397,7 +1399,7 @@ class API_VM_Element(Resource):
             {"name": "node"},
             {
                 "name": "selector",
-                "choices": ("mem", "vcpus", "load", "vms", "none"),
+                "choices": ("mem", "memfree", "vcpus", "load", "vms", "none"),
                 "helptext": "A valid selector must be specified",
             },
             {"name": "autostart"},
@@ -1444,10 +1446,11 @@ class API_VM_Element(Resource):
             name: selector
             type: string
             required: false
-            description: The selector used to determine candidate nodes during migration
-            default: mem
+            description: The selector used to determine candidate nodes during migration; see 'target_selector' in the node daemon configuration reference
+            default: none
             enum:
               - mem
+              - memfree
               - vcpus
               - load
               - vms
@@ -1626,7 +1629,7 @@ class API_VM_Metadata(Resource):
                     type: string
                 node_selector:
                   type: string
-                  description: The selector used to determine candidate nodes during migration
+                  description: The selector used to determine candidate nodes during migration; see 'target_selector' in the node daemon configuration reference
                 node_autostart:
                   type: string
                   description: Whether to autostart the VM when its node returns to ready domain state
@@ -1646,7 +1649,7 @@ class API_VM_Metadata(Resource):
             {"name": "limit"},
             {
                 "name": "selector",
-                "choices": ("mem", "vcpus", "load", "vms", "none"),
+                "choices": ("mem", "memfree", "vcpus", "load", "vms", "none"),
                 "helptext": "A valid selector must be specified",
             },
             {"name": "autostart"},
@@ -1675,12 +1678,14 @@ class API_VM_Metadata(Resource):
             name: selector
             type: string
             required: false
-            description: The selector used to determine candidate nodes during migration
+            description: The selector used to determine candidate nodes during migration; see 'target_selector' in the node daemon configuration reference
             enum:
               - mem
+              - memfree
               - vcpus
               - load
               - vms
+              - none (cluster default)
           - in: query
             name: autostart
             type: boolean
@@ -4099,11 +4104,112 @@ class API_Storage_Ceph_OSD_Element(Resource):
 
     @RequestParser(
         [
+            {
+                "name": "device",
+                "required": True,
+                "helptext": "A valid device or detect string must be specified.",
+            },
+            {
+                "name": "weight",
+                "required": True,
+                "helptext": "An OSD weight must be specified.",
+            },
             {
                 "name": "yes-i-really-mean-it",
                 "required": True,
                 "helptext": "Please confirm that 'yes-i-really-mean-it'.",
-            }
+            },
+        ]
+    )
+    @Authenticator
+    def post(self, osdid, reqargs):
+        """
+        Replace a Ceph OSD in the cluster
+        Note: This task may take up to 30s to complete and return
+        ---
+        tags:
+          - storage / ceph
+        parameters:
+          - in: query
+            name: device
+            type: string
+            required: true
+            description: The block device (e.g. "/dev/sdb", "/dev/disk/by-path/...", etc.) or detect string ("detect:NAME:SIZE:ID") to replace the OSD onto
+          - in: query
+            name: weight
+            type: number
+            required: true
+            description: The Ceph CRUSH weight for the replaced OSD
+        responses:
+          200:
+            description: OK
+            schema:
+              type: object
+              id: Message
+          400:
+            description: Bad request
+            schema:
+              type: object
+              id: Message
+        """
+        return api_helper.ceph_osd_replace(
+            osdid,
+            reqargs.get("device", None),
+            reqargs.get("weight", None),
+        )
+
+    @RequestParser(
+        [
+            {
+                "name": "device",
+                "required": True,
+                "helptext": "A valid device or detect string must be specified.",
+            },
+        ]
+    )
+    @Authenticator
+    def put(self, osdid, reqargs):
+        """
+        Refresh (reimport) a Ceph OSD in the cluster
+        Note: This task may take up to 30s to complete and return
+        ---
+        tags:
+          - storage / ceph
+        parameters:
+          - in: query
+            name: device
+            type: string
+            required: true
+            description: The block device (e.g. "/dev/sdb", "/dev/disk/by-path/...", etc.) or detect string ("detect:NAME:SIZE:ID") that the OSD should be using
+        responses:
+          200:
+            description: OK
+            schema:
+              type: object
+              id: Message
+          400:
+            description: Bad request
+            schema:
+              type: object
+              id: Message
+        """
+        return api_helper.ceph_osd_refresh(
+            osdid,
+            reqargs.get("device", None),
+        )
+
+    @RequestParser(
+        [
+            {
+                "name": "force",
+                "required": False,
+                "helptext": "Force removal even if steps fail.",
+            },
+            {
+                "name": "yes-i-really-mean-it",
+                "required": True,
+                "helptext": "Please confirm that 'yes-i-really-mean-it'.",
+            },
         ]
     )
     @Authenticator
@@ -4116,6 +4222,11 @@ class API_Storage_Ceph_OSD_Element(Resource):
         tags:
           - storage / ceph
         parameters:
+          - in: query
+            name: force
+            type: boolean
+            required: flase
+            description: Force removal even if some step(s) fail
           - in: query
             name: yes-i-really-mean-it
             type: string
@@ -4138,7 +4249,7 @@ class API_Storage_Ceph_OSD_Element(Resource):
               type: object
               id: Message
         """
-        return api_helper.ceph_osd_remove(osdid)
+        return api_helper.ceph_osd_remove(osdid, reqargs.get("force", False))
 
 
 api.add_resource(API_Storage_Ceph_OSD_Element, "/storage/ceph/osd/<osdid>")

api-daemon/pvcapid/helper.py

@@ -224,7 +224,7 @@ def node_domain_state(zkhandler, node):
 @ZKConnection(config)
 def node_secondary(zkhandler, node):
     """
-    Take NODE out of primary router mode.
+    Take NODE out of primary coordinator mode.
     """
     retflag, retdata = pvc_node.secondary_node(zkhandler, node)
 
@@ -240,7 +240,7 @@ def node_secondary(zkhandler, node):
 @ZKConnection(config)
 def node_primary(zkhandler, node):
     """
-    Set NODE to primary router mode.
+    Set NODE to primary coordinator mode.
     """
     retflag, retdata = pvc_node.primary_node(zkhandler, node)
 
@@ -1302,11 +1302,43 @@ def ceph_osd_add(zkhandler, node, device, weight, ext_db_flag=False, ext_db_rati
 
 
 @ZKConnection(config)
-def ceph_osd_remove(zkhandler, osd_id):
+def ceph_osd_replace(zkhandler, osd_id, device, weight):
+    """
+    Replace a Ceph OSD in the PVC Ceph storage cluster.
+    """
+    retflag, retdata = pvc_ceph.replace_osd(zkhandler, osd_id, device, weight)
+
+    if retflag:
+        retcode = 200
+    else:
+        retcode = 400
+
+    output = {"message": retdata.replace('"', "'")}
+    return output, retcode
+
+
+@ZKConnection(config)
+def ceph_osd_refresh(zkhandler, osd_id, device):
+    """
+    Refresh (reimport) a Ceph OSD in the PVC Ceph storage cluster.
+    """
+    retflag, retdata = pvc_ceph.refresh_osd(zkhandler, osd_id, device)
+
+    if retflag:
+        retcode = 200
+    else:
+        retcode = 400
+
+    output = {"message": retdata.replace('"', "'")}
+    return output, retcode
+
+
+@ZKConnection(config)
+def ceph_osd_remove(zkhandler, osd_id, force_flag):
     """
     Remove a Ceph OSD from the PVC Ceph storage cluster.
     """
-    retflag, retdata = pvc_ceph.remove_osd(zkhandler, osd_id)
+    retflag, retdata = pvc_ceph.remove_osd(zkhandler, osd_id, force_flag)
 
     if retflag:
         retcode = 200

client-cli/pvc/cli_lib/ceph.py

@@ -255,7 +255,47 @@ def ceph_osd_add(config, node, device, weight, ext_db_flag, ext_db_ratio):
     return retstatus, response.json().get("message", "")
 
 
-def ceph_osd_remove(config, osdid):
+def ceph_osd_replace(config, osdid, device, weight):
+    """
+    Replace an existing Ceph OSD with a new device
+
+    API endpoint: POST /api/v1/storage/ceph/osd/{osdid}
+    API arguments: device={device}, weight={weight}
+    API schema: {"message":"{data}"}
+    """
+    params = {"device": device, "weight": weight, "yes-i-really-mean-it": "yes"}
+    response = call_api(config, "post", f"/storage/ceph/osd/{osdid}", params=params)
+
+    if response.status_code == 200:
+        retstatus = True
+    else:
+        retstatus = False
+
+    return retstatus, response.json().get("message", "")
+
+
+def ceph_osd_refresh(config, osdid, device):
+    """
+    Refresh (reimport) an existing Ceph OSD with device {device}
+
+    API endpoint: PUT /api/v1/storage/ceph/osd/{osdid}
+    API arguments: device={device}
+    API schema: {"message":"{data}"}
+    """
+    params = {
+        "device": device,
+    }
+    response = call_api(config, "put", f"/storage/ceph/osd/{osdid}", params=params)
+
+    if response.status_code == 200:
+        retstatus = True
+    else:
+        retstatus = False
+
+    return retstatus, response.json().get("message", "")
+
+
+def ceph_osd_remove(config, osdid, force_flag):
     """
     Remove Ceph OSD
 
@@ -263,7 +303,7 @@ def ceph_osd_remove(config, osdid):
     API arguments:
     API schema: {"message":"{data}"}
     """
-    params = {"yes-i-really-mean-it": "yes"}
+    params = {"force": force_flag, "yes-i-really-mean-it": "yes"}
     response = call_api(
         config, "delete", "/storage/ceph/osd/{osdid}".format(osdid=osdid), params=params
     )
@@ -367,10 +407,29 @@ def format_list_osd(osd_list):
 
     for osd_information in osd_list:
         try:
-            # If this happens, the node hasn't checked in fully yet, so just ignore it
+            # If this happens, the node hasn't checked in fully yet, so use some dummy data
             if osd_information["stats"]["node"] == "|":
-                continue
+                for key in osd_information["stats"].keys():
+                    if (
+                        osd_information["stats"][key] == "|"
+                        or osd_information["stats"][key] is None
+                    ):
+                        osd_information["stats"][key] = "N/A"
+                for key in osd_information.keys():
+                    if osd_information[key] is None:
+                        osd_information[key] = "N/A"
+            else:
+                for key in osd_information["stats"].keys():
+                    if key in ["utilization", "var"] and isinstance(
+                        osd_information["stats"][key], float
+                    ):
+                        osd_information["stats"][key] = round(
+                            osd_information["stats"][key], 2
+                        )
         except KeyError:
+            print(
+                f"Details for OSD {osd_information['id']} missing required keys, skipping."
+            )
             continue
 
         # Deal with the size to human readable
@@ -396,7 +455,7 @@ def format_list_osd(osd_list):
             osd_id_length = _osd_id_length
 
         # Set the OSD node length
-        _osd_node_length = len(osd_information["stats"]["node"]) + 1
+        _osd_node_length = len(osd_information["node"]) + 1
         if _osd_node_length > osd_node_length:
             osd_node_length = _osd_node_length
 
@@ -439,13 +498,11 @@ def format_list_osd(osd_list):
         if _osd_free_length > osd_free_length:
             osd_free_length = _osd_free_length
 
-        osd_util = round(osd_information["stats"]["utilization"], 2)
-        _osd_util_length = len(str(osd_util)) + 1
+        _osd_util_length = len(str(osd_information["stats"]["utilization"])) + 1
         if _osd_util_length > osd_util_length:
             osd_util_length = _osd_util_length
 
-        osd_var = round(osd_information["stats"]["var"], 2)
-        _osd_var_length = len(str(osd_var)) + 1
+        _osd_var_length = len(str(osd_information["stats"]["var"])) + 1
         if _osd_var_length > osd_var_length:
             osd_var_length = _osd_var_length
 
@@ -592,18 +649,9 @@ def format_list_osd(osd_list):
     )
 
     for osd_information in sorted(osd_list, key=lambda x: int(x["id"])):
-        try:
-            # If this happens, the node hasn't checked in fully yet, so just ignore it
-            if osd_information["stats"]["node"] == "|":
-                continue
-        except KeyError:
-            continue
-
         osd_up_flag, osd_up_colour, osd_in_flag, osd_in_colour = getOutputColoursOSD(
             osd_information
         )
-        osd_util = round(osd_information["stats"]["utilization"], 2)
-        osd_var = round(osd_information["stats"]["var"], 2)
 
         osd_db_device = osd_information["db_device"]
         if not osd_db_device:
@@ -653,7 +701,7 @@ def format_list_osd(osd_list):
                 osd_rdops_length=osd_rdops_length,
                 osd_rddata_length=osd_rddata_length,
                 osd_id=osd_information["id"],
-                osd_node=osd_information["stats"]["node"],
+                osd_node=osd_information["node"],
                 osd_device=osd_information["device"],
                 osd_db_device=osd_db_device,
                 osd_up_colour=osd_up_colour,
@@ -666,8 +714,8 @@ def format_list_osd(osd_list):
                 osd_reweight=osd_information["stats"]["reweight"],
                 osd_used=osd_information["stats"]["used"],
                 osd_free=osd_information["stats"]["avail"],
-                osd_util=osd_util,
-                osd_var=osd_var,
+                osd_util=osd_information["stats"]["utilization"],
+                osd_var=osd_information["stats"]["var"],
                 osd_wrops=osd_information["stats"]["wr_ops"],
                 osd_wrdata=osd_information["stats"]["wr_data"],
                 osd_rdops=osd_information["stats"]["rd_ops"],

client-cli/pvc/cli_lib/provisioner.py

@@ -23,7 +23,6 @@ from requests_toolbelt.multipart.encoder import (
     MultipartEncoder,
     MultipartEncoderMonitor,
 )
-from ast import literal_eval
 
 import pvc.cli_lib.ansiprint as ansiprint
 from pvc.cli_lib.common import UploadProgressBar, call_api
@@ -793,10 +792,10 @@ def task_status(config, task_id=None, is_watching=False):
                         task["type"] = task_type
                     task["worker"] = task_host
                     task["id"] = task_job.get("id")
-                    task_args = literal_eval(task_job.get("args"))
+                    task_args = task_job.get("args")
                     task["vm_name"] = task_args[0]
                     task["vm_profile"] = task_args[1]
-                    task_kwargs = literal_eval(task_job.get("kwargs"))
+                    task_kwargs = task_job.get("kwargs")
                     task["vm_define"] = str(bool(task_kwargs["define_vm"]))
                     task["vm_start"] = str(bool(task_kwargs["start_vm"]))
                     task_data.append(task)

client-cli/pvc/pvc.py

@@ -486,7 +486,7 @@ def cli_node():
 @cluster_req
 def node_secondary(node, wait):
     """
-    Take NODE out of primary router mode.
+    Take NODE out of primary coordinator mode.
     """
 
     task_retcode, task_retdata = pvc_provisioner.task_status(config, None)
@@ -539,7 +539,7 @@ def node_secondary(node, wait):
 @cluster_req
 def node_primary(node, wait):
     """
-    Put NODE into primary router mode.
+    Put NODE into primary coordinator mode.
     """
 
     task_retcode, task_retdata = pvc_provisioner.task_status(config, None)
@@ -803,11 +803,11 @@ def cli_vm():
 )
 @click.option(
     "-s",
-    "--selector",
+    "--node-selector",
     "node_selector",
-    default="mem",
+    default="none",
     show_default=True,
-    type=click.Choice(["mem", "load", "vcpus", "vms", "none"]),
+    type=click.Choice(["mem", "memfree", "load", "vcpus", "vms", "none"]),
     help='Method to determine optimal target node during autoselect; "none" will use the default for the cluster.',
 )
 @click.option(
@@ -857,6 +857,18 @@ def vm_define(
 ):
     """
     Define a new virtual machine from Libvirt XML configuration file VMCONFIG.
+
+    The target node selector ("--node-selector"/"-s") can be "none" to use the cluster default, or one of the following values:
+      * "mem": choose the node with the least provisioned VM memory
+      * "memfree": choose the node with the most (real) free memory
+      * "vcpus": choose the node with the least allocated VM vCPUs
+      * "load": choose the node with the lowest current load average
+      * "vms": choose the node with the least number of provisioned VMs
+
+    For most clusters, "mem" should be sufficient, but others may be used based on the cluster workload and available resources. The following caveats should be considered:
+      * "mem" looks at the provisioned memory, not the allocated memory; thus, stopped or disabled VMs are counted towards a node's memory for this selector, even though their memory is not actively in use.
+      * "memfree" looks at the free memory of the node in general, ignoring the amount provisioned to VMs; if any VM's internal memory usage changes, this value would be affected. This might be preferable to "mem" on clusters with very high memory utilization versus total capacity or if many VMs are stopped/disabled.
+      * "load" looks at the system load of the node in general, ignoring load in any particular VMs; if any VM's CPU usage changes, this value would be affected. This might be preferable on clusters with some very CPU intensive VMs.
     """
 
     # Open the XML file
@@ -898,11 +910,11 @@ def vm_define(
 )
 @click.option(
     "-s",
-    "--selector",
+    "--node-selector",
     "node_selector",
     default=None,
     show_default=False,
-    type=click.Choice(["mem", "load", "vcpus", "vms", "none"]),
+    type=click.Choice(["mem", "memfree", "load", "vcpus", "vms", "none"]),
     help='Method to determine optimal target node during autoselect; "none" will use the default for the cluster.',
 )
 @click.option(
@@ -942,6 +954,8 @@ def vm_meta(
 ):
     """
     Modify the PVC metadata of existing virtual machine DOMAIN. At least one option to update must be specified. DOMAIN may be a UUID or name.
+
+    For details on the "--node-selector"/"-s" values, please see help for the command "pvc vm define".
     """
 
     if (
@@ -1009,6 +1023,7 @@ def vm_meta(
 )
 @click.argument("domain")
 @click.argument("cfgfile", type=click.File(), default=None, required=False)
+@cluster_req
 def vm_modify(
     domain,
     cfgfile,
@@ -1343,15 +1358,31 @@ def vm_stop(domain, confirm_flag):
     default=False,
     help="Forcibly stop the VM instead of waiting for shutdown.",
 )
+@click.option(
+    "-y",
+    "--yes",
+    "confirm_flag",
+    is_flag=True,
+    default=False,
+    help="Confirm the disable",
+)
 @cluster_req
-def vm_disable(domain, force):
+def vm_disable(domain, force_flag, confirm_flag):
     """
     Shut down virtual machine DOMAIN and mark it as disabled. DOMAIN may be a UUID or name.
 
     Disabled VMs will not be counted towards a degraded cluster health status, unlike stopped VMs. Use this option for a VM that will remain off for an extended period.
     """
 
-    retcode, retmsg = pvc_vm.vm_state(config, domain, "disable", force=force)
+    if not confirm_flag and not config["unsafe"]:
+        try:
+            click.confirm(
+                "Disable VM {}".format(domain), prompt_suffix="? ", abort=True
+            )
+        except Exception:
+            exit(0)
+
+    retcode, retmsg = pvc_vm.vm_state(config, domain, "disable", force=force_flag)
     cleanup(retcode, retmsg)
 
 
@@ -3372,10 +3403,19 @@ def ceph_osd_add(node, device, weight, ext_db_flag, ext_db_ratio, confirm_flag):
 
 
 ###############################################################################
-# pvc storage osd remove
+# pvc storage osd replace
 ###############################################################################
-@click.command(name="remove", short_help="Remove OSD.")
+@click.command(name="replace", short_help="Replace OSD block device.")
 @click.argument("osdid")
+@click.argument("device")
+@click.option(
+    "-w",
+    "--weight",
+    "weight",
+    default=1.0,
+    show_default=True,
+    help="New weight of the OSD within the CRUSH map.",
+)
 @click.option(
     "-y",
     "--yes",
@@ -3385,11 +3425,80 @@ def ceph_osd_add(node, device, weight, ext_db_flag, ext_db_ratio, confirm_flag):
     help="Confirm the removal",
 )
 @cluster_req
-def ceph_osd_remove(osdid, confirm_flag):
+def ceph_osd_replace(osdid, device, weight, confirm_flag):
+    """
+    Replace the block device of an existing OSD with ID OSDID with DEVICE. Use this command to replace a failed or smaller OSD block device with a new one.
+
+    DEVICE must be a valid raw block device (e.g. '/dev/sda', '/dev/nvme0n1', '/dev/disk/by-path/...', '/dev/disk/by-id/...') or a "detect" string. Using partitions is not supported. A "detect" string is a string in the form "detect:<NAME>:<HUMAN-SIZE>:<ID>". For details, see 'pvc storage osd add --help'.
+
+    The weight of an OSD should reflect the ratio of the OSD to other OSDs in the storage cluster. For details, see 'pvc storage osd add --help'. Note that the current weight must be explicitly specified if it differs from the default.
+
+    Existing IDs, external DB devices, etc. of the OSD will be preserved; data will be lost and rebuilt from the remaining healthy OSDs.
+    """
+    if not confirm_flag and not config["unsafe"]:
+        try:
+            click.confirm(
+                "Replace OSD {} with block device {}".format(osdid, device),
+                prompt_suffix="? ",
+                abort=True,
+            )
+        except Exception:
+            exit(0)
+
+    retcode, retmsg = pvc_ceph.ceph_osd_replace(config, osdid, device, weight)
+    cleanup(retcode, retmsg)
+
+
+###############################################################################
+# pvc storage osd refresh
+###############################################################################
+@click.command(name="refresh", short_help="Refresh (reimport) OSD device.")
+@click.argument("osdid")
+@click.argument("device")
+@cluster_req
+def ceph_osd_refresh(osdid, device):
+    """
+    Refresh (reimport) the block DEVICE of an existing OSD with ID OSDID. Use this command to reimport a working OSD into a rebuilt/replaced node.
+
+    DEVICE must be a valid raw block device (e.g. '/dev/sda', '/dev/nvme0n1', '/dev/disk/by-path/...', '/dev/disk/by-id/...') or a "detect" string. Using partitions is not supported. A "detect" string is a string in the form "detect:<NAME>:<HUMAN-SIZE>:<ID>". For details, see 'pvc storage osd add --help'.
+
+    Existing data, IDs, weights, etc. of the OSD will be preserved.
+
+    NOTE: If a device had an external DB device, this is not automatically handled at this time. It is best to remove and re-add the OSD instead.
+    """
+    retcode, retmsg = pvc_ceph.ceph_osd_refresh(config, osdid, device)
+    cleanup(retcode, retmsg)
+
+
+###############################################################################
+# pvc storage osd remove
+###############################################################################
+@click.command(name="remove", short_help="Remove OSD.")
+@click.argument("osdid")
+@click.option(
+    "-f",
+    "--force",
+    "force_flag",
+    is_flag=True,
+    default=False,
+    help="Force removal even if steps fail",
+)
+@click.option(
+    "-y",
+    "--yes",
+    "confirm_flag",
+    is_flag=True,
+    default=False,
+    help="Confirm the removal",
+)
+@cluster_req
+def ceph_osd_remove(osdid, force_flag, confirm_flag):
     """
     Remove a Ceph OSD with ID OSDID.
 
     DANGER: This will completely remove the OSD from the cluster. OSDs will rebalance which will negatively affect performance and available space. It is STRONGLY RECOMMENDED to set an OSD out (using 'pvc storage osd out') and allow the cluster to fully rebalance (verified with 'pvc storage status') before removing an OSD.
+
+    NOTE: The "-f"/"--force" option is useful after replacing a failed node, to ensure the OSD is removed even if the OSD in question does not properly exist on the node after a rebuild.
     """
     if not confirm_flag and not config["unsafe"]:
         try:
@@ -3397,7 +3506,7 @@ def ceph_osd_remove(osdid, confirm_flag):
         except Exception:
             exit(0)
 
-    retcode, retmsg = pvc_ceph.ceph_osd_remove(config, osdid)
+    retcode, retmsg = pvc_ceph.ceph_osd_remove(config, osdid, force_flag)
     cleanup(retcode, retmsg)
 
 
@@ -4024,7 +4133,9 @@ def provisioner_template_system_list(limit):
 @click.option(
     "--node-selector",
     "node_selector",
-    type=click.Choice(["mem", "vcpus", "vms", "load", "none"], case_sensitive=False),
+    type=click.Choice(
+        ["mem", "memfree", "vcpus", "vms", "load", "none"], case_sensitive=False
+    ),
     default="none",
     help='Method to determine optimal target node during autoselect; "none" will use the default for the cluster.',
 )
@@ -4057,6 +4168,8 @@ def provisioner_template_system_add(
 ):
     """
     Add a new system template NAME to the PVC cluster provisioner.
+
+    For details on the possible "--node-selector" values, please see help for the command "pvc vm define".
     """
     params = dict()
     params["name"] = name
@@ -4116,7 +4229,9 @@ def provisioner_template_system_add(
 @click.option(
     "--node-selector",
     "node_selector",
-    type=click.Choice(["mem", "vcpus", "vms", "load", "none"], case_sensitive=False),
+    type=click.Choice(
+        ["mem", "memfree", "vcpus", "vms", "load", "none"], case_sensitive=False
+    ),
     help='Method to determine optimal target node during autoselect; "none" will use the default for the cluster.',
 )
 @click.option(
@@ -4148,6 +4263,8 @@ def provisioner_template_system_modify(
 ):
     """
     Add a new system template NAME to the PVC cluster provisioner.
+
+    For details on the possible "--node-selector" values, please see help for the command "pvc vm define".
     """
     params = dict()
     params["vcpus"] = vcpus
@@ -5741,6 +5858,11 @@ def cli(_cluster, _debug, _quiet, _unsafe, _colour):
     global config
     store_data = get_store(store_path)
     config = get_config(store_data, _cluster)
+
+    # There is only one cluster and no local cluster, so even if nothing was passed, use it
+    if len(store_data) == 1 and _cluster is None and config.get("badcfg", None):
+        config = get_config(store_data, list(store_data.keys())[0])
+
     if not config.get("badcfg", None):
         config["debug"] = _debug
         config["unsafe"] = _unsafe
@@ -5857,6 +5979,8 @@ ceph_benchmark.add_command(ceph_benchmark_list)
 
 ceph_osd.add_command(ceph_osd_create_db_vg)
 ceph_osd.add_command(ceph_osd_add)
+ceph_osd.add_command(ceph_osd_replace)
+ceph_osd.add_command(ceph_osd_refresh)
 ceph_osd.add_command(ceph_osd_remove)
 ceph_osd.add_command(ceph_osd_in)
 ceph_osd.add_command(ceph_osd_out)

client-cli/setup.py

@@ -2,7 +2,7 @@ from setuptools import setup
 
 setup(
     name="pvc",
-    version="0.9.47",
+    version="0.9.53",
     packages=["pvc", "pvc.cli_lib"],
     install_requires=[
         "Click",

daemon-common/ceph.py

@@ -181,6 +181,7 @@ def getClusterOSDList(zkhandler):
 
 def getOSDInformation(zkhandler, osd_id):
     # Get the devices
+    osd_node = zkhandler.read(("osd.node", osd_id))
     osd_device = zkhandler.read(("osd.device", osd_id))
     osd_db_device = zkhandler.read(("osd.db_device", osd_id))
     # Parse the stats data
@@ -189,6 +190,7 @@ def getOSDInformation(zkhandler, osd_id):
 
     osd_information = {
         "id": osd_id,
+        "node": osd_node,
         "device": osd_device,
         "db_device": osd_db_device,
         "stats": osd_stats,
@@ -234,7 +236,7 @@ def add_osd_db_vg(zkhandler, node, device):
     return success, message
 
 
-# OSD addition and removal uses the /cmd/ceph pipe
+# OSD actions use the /cmd/ceph pipe
 # These actions must occur on the specific node they reference
 def add_osd(zkhandler, node, device, weight, ext_db_flag=False, ext_db_ratio=0.05):
     # Verify the target node exists
@@ -286,14 +288,111 @@ def add_osd(zkhandler, node, device, weight, ext_db_flag=False, ext_db_ratio=0.0
     return success, message
 
 
-def remove_osd(zkhandler, osd_id):
+def replace_osd(zkhandler, osd_id, new_device, weight):
+    # Get current OSD information
+    osd_information = getOSDInformation(zkhandler, osd_id)
+    node = osd_information["node"]
+    old_device = osd_information["device"]
+    ext_db_flag = True if osd_information["db_device"] else False
+
+    # Verify target block device isn't in use
+    block_osd = verifyOSDBlock(zkhandler, node, new_device)
+    if block_osd and block_osd != osd_id:
+        return (
+            False,
+            'ERROR: Block device "{}" on node "{}" is used by OSD "{}"'.format(
+                new_device, node, block_osd
+            ),
+        )
+
+    # Tell the cluster to create a new OSD for the host
+    replace_osd_string = "osd_replace {},{},{},{},{},{}".format(
+        node, osd_id, old_device, new_device, weight, ext_db_flag
+    )
+    zkhandler.write([("base.cmd.ceph", replace_osd_string)])
+    # Wait 1/2 second for the cluster to get the message and start working
+    time.sleep(0.5)
+    # Acquire a read lock, so we get the return exclusively
+    with zkhandler.readlock("base.cmd.ceph"):
+        try:
+            result = zkhandler.read("base.cmd.ceph").split()[0]
+            if result == "success-osd_replace":
+                message = 'Replaced OSD {} with block device "{}" on node "{}".'.format(
+                    osd_id, new_device, node
+                )
+                success = True
+            else:
+                message = "ERROR: Failed to replace OSD; check node logs for details."
+                success = False
+        except Exception:
+            message = "ERROR: Command ignored by node."
+            success = False
+
+    # Acquire a write lock to ensure things go smoothly
+    with zkhandler.writelock("base.cmd.ceph"):
+        time.sleep(0.5)
+        zkhandler.write([("base.cmd.ceph", "")])
+
+    return success, message
+
+
+def refresh_osd(zkhandler, osd_id, device):
+    # Get current OSD information
+    osd_information = getOSDInformation(zkhandler, osd_id)
+    node = osd_information["node"]
+    ext_db_flag = True if osd_information["db_device"] else False
+
+    # Verify target block device isn't in use
+    block_osd = verifyOSDBlock(zkhandler, node, device)
+    if not block_osd or block_osd != osd_id:
+        return (
+            False,
+            'ERROR: Block device "{}" on node "{}" is not used by OSD "{}"; use replace instead'.format(
+                device, node, osd_id
+            ),
+        )
+
+    # Tell the cluster to create a new OSD for the host
+    refresh_osd_string = "osd_refresh {},{},{},{}".format(
+        node, osd_id, device, ext_db_flag
+    )
+    zkhandler.write([("base.cmd.ceph", refresh_osd_string)])
+    # Wait 1/2 second for the cluster to get the message and start working
+    time.sleep(0.5)
+    # Acquire a read lock, so we get the return exclusively
+    with zkhandler.readlock("base.cmd.ceph"):
+        try:
+            result = zkhandler.read("base.cmd.ceph").split()[0]
+            if result == "success-osd_refresh":
+                message = (
+                    'Refreshed OSD {} with block device "{}" on node "{}".'.format(
+                        osd_id, device, node
+                    )
+                )
+                success = True
+            else:
+                message = "ERROR: Failed to refresh OSD; check node logs for details."
+                success = False
+        except Exception:
+            message = "ERROR: Command ignored by node."
+            success = False
+
+    # Acquire a write lock to ensure things go smoothly
+    with zkhandler.writelock("base.cmd.ceph"):
+        time.sleep(0.5)
+        zkhandler.write([("base.cmd.ceph", "")])
+
+    return success, message
+
+
+def remove_osd(zkhandler, osd_id, force_flag):
     if not verifyOSD(zkhandler, osd_id):
         return False, 'ERROR: No OSD with ID "{}" is present in the cluster.'.format(
             osd_id
         )
 
     # Tell the cluster to remove an OSD
-    remove_osd_string = "osd_remove {}".format(osd_id)
+    remove_osd_string = "osd_remove {},{}".format(osd_id, str(force_flag))
     zkhandler.write([("base.cmd.ceph", remove_osd_string)])
     # Wait 1/2 second for the cluster to get the message and start working
     time.sleep(0.5)

daemon-common/common.py

@@ -639,6 +639,8 @@ def findTargetNode(zkhandler, dom_uuid):
     # Execute the search
     if search_field == "mem":
         return findTargetNodeMem(zkhandler, node_limit, dom_uuid)
+    if search_field == "memfree":
+        return findTargetNodeMemFree(zkhandler, node_limit, dom_uuid)
     if search_field == "load":
         return findTargetNodeLoad(zkhandler, node_limit, dom_uuid)
     if search_field == "vcpus":
@@ -677,7 +679,7 @@ def getNodes(zkhandler, node_limit, dom_uuid):
 
 
 #
-# via free memory (relative to allocated memory)
+# via provisioned memory
 #
 def findTargetNodeMem(zkhandler, node_limit, dom_uuid):
     most_provfree = 0
@@ -698,6 +700,24 @@ def findTargetNodeMem(zkhandler, node_limit, dom_uuid):
     return target_node
 
 
+#
+# via free memory
+#
+def findTargetNodeMemFree(zkhandler, node_limit, dom_uuid):
+    most_memfree = 0
+    target_node = None
+
+    node_list = getNodes(zkhandler, node_limit, dom_uuid)
+    for node in node_list:
+        memfree = int(zkhandler.read(("node.memory.free", node)))
+
+        if memfree > most_memfree:
+            most_memfree = memfree
+            target_node = node
+
+    return target_node
+
+
 #
 # via load average
 #

daemon-common/migrations/versions/8.json

@@ -0,0 +1 @@
+{"version": "8", "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", "mtu": "/mtu", "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", "db_device": "/db_device", "fsid": "/fsid", "ofsid": "/fsid/osd", "cfsid": "/fsid/cluster", "lvm": "/lvm", "vg": "/lvm/vg", "lv": "/lvm/lv", "stats": "/stats"}, "pool": {"name": "", "pgs": "/pgs", "tier": "/tier", "stats": "/stats"}, "volume": {"name": "", "stats": "/stats"}, "snapshot": {"name": "", "stats": "/stats"}}

daemon-common/node.py

@@ -91,7 +91,7 @@ def secondary_node(zkhandler, node):
     if daemon_mode == "hypervisor":
         return (
             False,
-            'ERROR: Cannot change router mode on non-coordinator node "{}"'.format(
+            'ERROR: Cannot change coordinator mode on non-coordinator node "{}"'.format(
                 node
             ),
         )
@@ -104,9 +104,9 @@ def secondary_node(zkhandler, node):
     # Get current state
     current_state = zkhandler.read(("node.state.router", node))
     if current_state == "secondary":
-        return True, 'Node "{}" is already in secondary router mode.'.format(node)
+        return True, 'Node "{}" is already in secondary coordinator mode.'.format(node)
 
-    retmsg = "Setting node {} in secondary router mode.".format(node)
+    retmsg = "Setting node {} in secondary coordinator mode.".format(node)
     zkhandler.write([("base.config.primary_node", "none")])
 
     return True, retmsg
@@ -124,7 +124,7 @@ def primary_node(zkhandler, node):
     if daemon_mode == "hypervisor":
         return (
             False,
-            'ERROR: Cannot change router mode on non-coordinator node "{}"'.format(
+            'ERROR: Cannot change coordinator mode on non-coordinator node "{}"'.format(
                 node
             ),
         )
@@ -137,9 +137,9 @@ def primary_node(zkhandler, node):
     # Get current state
     current_state = zkhandler.read(("node.state.router", node))
     if current_state == "primary":
-        return True, 'Node "{}" is already in primary router mode.'.format(node)
+        return True, 'Node "{}" is already in primary coordinator mode.'.format(node)
 
-    retmsg = "Setting node {} in primary router mode.".format(node)
+    retmsg = "Setting node {} in primary coordinator mode.".format(node)
     zkhandler.write([("base.config.primary_node", node)])
 
     return True, retmsg
@@ -236,6 +236,7 @@ def get_list(
 ):
     node_list = []
     full_node_list = zkhandler.children("base.node")
+    full_node_list.sort()
 
     if is_fuzzy and limit:
         # Implicitly assume fuzzy limits

daemon-common/vm.py

@@ -1193,6 +1193,7 @@ def get_list(zkhandler, node, state, tag, limit, is_fuzzy=True, negate=False):
             return False, 'VM state "{}" is not valid.'.format(state)
 
     full_vm_list = zkhandler.children("base.domain")
+    full_vm_list.sort()
 
     # Set our limit to a sensible regex
     if limit:
@@ -1291,4 +1292,4 @@ def get_list(zkhandler, node, state, tag, limit, is_fuzzy=True, negate=False):
             except Exception:
                 pass
 
-    return True, vm_data_list
+    return True, sorted(vm_data_list, key=lambda d: d["name"])

daemon-common/zkhandler.py

@@ -540,7 +540,7 @@ class ZKHandler(object):
 #
 class ZKSchema(object):
     # Current version
-    _version = 7
+    _version = 8
 
     # Root for doing nested keys
     _schema_root = ""
@@ -700,6 +700,12 @@ class ZKSchema(object):
             "node": "/node",
             "device": "/device",
             "db_device": "/db_device",
+            "fsid": "/fsid",
+            "ofsid": "/fsid/osd",
+            "cfsid": "/fsid/cluster",
+            "lvm": "/lvm",
+            "vg": "/lvm/vg",
+            "lv": "/lvm/lv",
             "stats": "/stats",
         },
         # The schema of an individual pool entry (/ceph/pools/{pool_name})

debian/changelog

@@ -1,3 +1,51 @@
+pvc (0.9.53-0) unstable; urgency=high
+
+  * [API] Fixes sort order of VM list (for real this time)
+
+ -- Joshua M. Boniface <joshua@boniface.me>  Fri, 12 Aug 2022 17:47:11 -0400
+
+pvc (0.9.52-0) unstable; urgency=high
+
+  * [CLI] Fixes a bug with vm modify not requiring a cluster
+  * [Docs] Adds a reference to the bootstrap daemon
+  * [API] Adds sorting to node and VM lists for consistency
+  * [Node Daemon/API] Adds kb_ stats values for OSD stats
+
+ -- Joshua M. Boniface <joshua@boniface.me>  Fri, 12 Aug 2022 11:09:25 -0400
+
+pvc (0.9.51-0) unstable; urgency=high
+
+  * [CLI Client] Fixes a faulty literal_eval when viewing task status
+  * [CLI Client] Adds a confirmation flag to the vm disable command
+  * [Node Daemon] Removes the pvc-flush service
+
+ -- Joshua M. Boniface <joshua@boniface.me>  Mon, 25 Jul 2022 23:25:41 -0400
+
+pvc (0.9.50-0) unstable; urgency=high
+
+  * [Node Daemon/API/CLI] Adds free memory node selector
+  * [Node Daemon] Fixes bug sending space-containing detect disk strings
+
+ -- Joshua M. Boniface <joshua@boniface.me>  Wed, 06 Jul 2022 16:01:14 -0400
+
+pvc (0.9.49-0) unstable; urgency=high
+
+  * [Node Daemon] Fixes bugs with OSD stat population on creation
+  * [Node Daemon/API] Adds additional information to Zookeeper about OSDs
+  * [Node Daemon] Refactors OSD removal for improved safety
+  * [Node Daemon/API/CLI] Adds explicit support for replacing and refreshing (reimporting) OSDs
+  * [API/CLI] Fixes a language inconsistency about "router mode"
+
+ -- Joshua M. Boniface <joshua@boniface.me>  Fri, 06 May 2022 15:49:39 -0400
+
+pvc (0.9.48-0) unstable; urgency=high
+
+  * [CLI] Fixes situation where only a single cluster is available
+  * [CLI/API/Daemon] Allows forcing of OSD removal ignoring errors
+  * [CLI] Fixes bug where down OSDs are not displayed
+
+ -- Joshua M. Boniface <joshua@boniface.me>  Fri, 29 Apr 2022 12:13:58 -0400
+
 pvc (0.9.47-0) unstable; urgency=high
 
   * [Node Daemon/API/CLI] Adds Ceph pool device class/tier support

debian/pvc-daemon-node.install

@@ -3,5 +3,4 @@ node-daemon/pvcnoded.sample.yaml etc/pvc
 node-daemon/pvcnoded usr/share/pvc
 node-daemon/pvcnoded.service lib/systemd/system
 node-daemon/pvc.target lib/systemd/system
-node-daemon/pvc-flush.service lib/systemd/system
 node-daemon/monitoring usr/share/pvc

debian/pvc-daemon-node.postinst

@@ -7,11 +7,6 @@ systemctl daemon-reload
 systemctl enable /lib/systemd/system/pvcnoded.service
 systemctl enable /lib/systemd/system/pvc.target
 
-# Inform administrator of the autoflush daemon if it is not enabled
-if ! systemctl is-active --quiet pvc-flush.service; then
-    echo "NOTE: The PVC autoflush daemon (pvc-flush.service) is not enabled by default; enable it to perform automatic flush/unflush actions on host shutdown/startup."
-fi
-
 # Inform administrator of the service restart/startup not occurring automatically
 if systemctl is-active --quiet pvcnoded.service; then
     echo "NOTE: The PVC node daemon (pvcnoded.service) has not been restarted; this is up to the administrator."

docs/index.md

@@ -18,7 +18,7 @@ As a consequence of its features, PVC makes administrating very high-uptime VMs
 
 PVC also features an optional, fully customizable VM provisioning framework, designed to automate and simplify VM deployments using custom provisioning profiles, scripts, and CloudInit userdata API support.
 
-Installation of PVC is accomplished by two main components: a [Node installer ISO](https://github.com/parallelvirtualcluster/pvc-installer) which creates on-demand installer ISOs, and an [Ansible role framework](https://github.com/parallelvirtualcluster/pvc-ansible) to configure, bootstrap, and administrate the nodes. Once up, the cluster is managed via an HTTP REST API, accessible via a Python Click CLI client or WebUI.
+Installation of PVC is accomplished by two main components: a [Node installer ISO](https://github.com/parallelvirtualcluster/pvc-installer) which creates on-demand installer ISOs, and an [Ansible role framework](https://github.com/parallelvirtualcluster/pvc-ansible) to configure, bootstrap, and administrate the nodes. Installation can also be fully automated with a companion [cluster bootstrapping system](https://github.com/parallelvirtualcluster/pvc-bootstrap). Once up, the cluster is managed via an HTTP REST API, accessible via a Python Click CLI client or WebUI.
 
 Just give it physical servers, and it will run your VMs without you having to think about it, all in just an hour or two of setup time.
 

docs/manuals/daemon.md

@@ -353,7 +353,19 @@ The password for the PVC node daemon to log in to the IPMI interface.
 
 * *required*
 
-The selector algorithm to use when migrating hosts away from the node. Valid `selector` values are: `mem`: the node with the least allocated VM memory; `vcpus`: the node with the least allocated VM vCPUs; `load`: the node with the least current load average; `vms`: the node with the least number of provisioned VMs.
+The default selector algorithm to use when migrating VMs away from a node; individual VMs can override this default.
+
+Valid `target_selector` values are:
+  * `mem`: choose the node with the least provisioned VM memory
+  * `memfree`: choose the node with the most (real) free memory
+  * `vcpus`: choose the node with the least allocated VM vCPUs
+  * `load`: choose the node with the lowest current load average
+  * `vms`: choose the node with the least number of provisioned VMs
+
+For most clusters, `mem` should be sufficient, but others may be used based on the cluster workload and available resources. The following caveats should be considered:
+  * `mem` looks at the provisioned memory, not the allocated memory; thus, stopped or disabled VMs are counted towards a node's memory for this selector, even though their memory is not actively in use.
+  * `memfree` looks at the free memory of the node in general, ignoring the amount provisioned to VMs; if any VM's internal memory usage changes, this value would be affected. This might be preferable to `mem` on clusters with very high memory utilization versus total capacity or if many VMs are stopped/disabled.
+  * `load` looks at the system load of the node in general, ignoring load in any particular VMs; if any VM's CPU usage changes, this value would be affected. This might be preferable on clusters with some very CPU intensive VMs.
 
 #### `system` → `configuration` → `directories` → `dynamic_directory`
 

docs/manuals/swagger.json

@@ -192,7 +192,7 @@
                     "type": "array"
                 },
                 "node_selector": {
-                    "description": "The selector used to determine candidate nodes during migration",
+                    "description": "The selector used to determine candidate nodes during migration; see 'target_selector' in the node daemon configuration reference",
                     "type": "string"
                 }
             },
@@ -1414,7 +1414,7 @@
                     "type": "array"
                 },
                 "node_selector": {
-                    "description": "The selector used to determine candidate nodes during migration",
+                    "description": "The selector used to determine candidate nodes during migration; see 'target_selector' in the node daemon configuration reference",
                     "type": "string"
                 },
                 "profile": {
@@ -5094,6 +5094,13 @@
             "delete": {
                 "description": "Note: This task may take up to 30s to complete and return<br/>Warning: This operation may have unintended consequences for the storage cluster; ensure the cluster can support removing the OSD before proceeding",
                 "parameters": [
+                    {
+                        "description": "Force removal even if some step(s) fail",
+                        "in": "query",
+                        "name": "force",
+                        "required": "flase",
+                        "type": "boolean"
+                    },
                     {
                         "description": "A confirmation string to ensure that the API consumer really means it",
                         "in": "query",
@@ -5141,6 +5148,73 @@
                 "tags": [
                     "storage / ceph"
                 ]
+            },
+            "post": {
+                "description": "Note: This task may take up to 30s to complete and return",
+                "parameters": [
+                    {
+                        "description": "The block device (e.g. \"/dev/sdb\", \"/dev/disk/by-path/...\", etc.) or detect string (\"detect:NAME:SIZE:ID\") to replace the OSD onto",
+                        "in": "query",
+                        "name": "device",
+                        "required": true,
+                        "type": "string"
+                    },
+                    {
+                        "description": "The Ceph CRUSH weight for the replaced OSD",
+                        "in": "query",
+                        "name": "weight",
+                        "required": true,
+                        "type": "number"
+                    }
+                ],
+                "responses": {
+                    "200": {
+                        "description": "OK",
+                        "schema": {
+                            "$ref": "#/definitions/Message"
+                        }
+                    },
+                    "400": {
+                        "description": "Bad request",
+                        "schema": {
+                            "$ref": "#/definitions/Message"
+                        }
+                    }
+                },
+                "summary": "Replace a Ceph OSD in the cluster",
+                "tags": [
+                    "storage / ceph"
+                ]
+            },
+            "put": {
+                "description": "Note: This task may take up to 30s to complete and return",
+                "parameters": [
+                    {
+                        "description": "The block device (e.g. \"/dev/sdb\", \"/dev/disk/by-path/...\", etc.) or detect string (\"detect:NAME:SIZE:ID\") that the OSD should be using",
+                        "in": "query",
+                        "name": "device",
+                        "required": true,
+                        "type": "string"
+                    }
+                ],
+                "responses": {
+                    "200": {
+                        "description": "OK",
+                        "schema": {
+                            "$ref": "#/definitions/Message"
+                        }
+                    },
+                    "400": {
+                        "description": "Bad request",
+                        "schema": {
+                            "$ref": "#/definitions/Message"
+                        }
+                    }
+                },
+                "summary": "Refresh (reimport) a Ceph OSD in the cluster",
+                "tags": [
+                    "storage / ceph"
+                ]
             }
         },
         "/api/v1/storage/ceph/osd/{osdid}/state": {
@@ -6099,13 +6173,15 @@
                         "type": "string"
                     },
                     {
-                        "default": "mem",
-                        "description": "The selector used to determine candidate nodes during migration",
+                        "default": "none",
+                        "description": "The selector used to determine candidate nodes during migration; see 'target_selector' in the node daemon configuration reference",
                         "enum": [
                             "mem",
+                            "memfree",
                             "vcpus",
                             "load",
-                            "vms"
+                            "vms",
+                            "none (cluster default)"
                         ],
                         "in": "query",
                         "name": "selector",
@@ -6256,10 +6332,11 @@
                         "type": "string"
                     },
                     {
-                        "default": "mem",
-                        "description": "The selector used to determine candidate nodes during migration",
+                        "default": "none",
+                        "description": "The selector used to determine candidate nodes during migration; see 'target_selector' in the node daemon configuration reference",
                         "enum": [
                             "mem",
+                            "memfree",
                             "vcpus",
                             "load",
                             "vms",
@@ -6517,12 +6594,14 @@
                         "type": "string"
                     },
                     {
-                        "description": "The selector used to determine candidate nodes during migration",
+                        "description": "The selector used to determine candidate nodes during migration; see 'target_selector' in the node daemon configuration reference",
                         "enum": [
                             "mem",
+                            "memfree",
                             "vcpus",
                             "load",
-                            "vms"
+                            "vms",
+                            "none (cluster default)"
                         ],
                         "in": "query",
                         "name": "selector",

node-daemon/pvc-flush.service

@@ -1,20 +0,0 @@
-# Parallel Virtual Cluster autoflush daemon
-
-[Unit]
-Description = Parallel Virtual Cluster autoflush daemon
-After = pvcnoded.service pvcapid.service zookeeper.service libvirtd.service ssh.service ceph.target network-online.target
-Wants = pvcnoded.service
-PartOf = pvc.target
-
-[Service]
-Type = oneshot
-RemainAfterExit = true
-WorkingDirectory = /usr/share/pvc
-TimeoutSec = 30min
-ExecStartPre = /bin/sleep 30
-ExecStart = /usr/bin/pvc -c local node unflush --wait
-ExecStop = /usr/bin/pvc -c local node flush --wait
-ExecStopPost = /bin/sleep 5
-
-[Install]
-WantedBy = pvc.target

node-daemon/pvcnoded.sample.yaml

@@ -122,7 +122,7 @@ pvc:
         pass: Passw0rd
     # migration: Migration option configuration
     migration:
-      # target_selector: Criteria to select the ideal migration target, options: mem, load, vcpus, vms
+      # target_selector: Criteria to select the ideal migration target, options: mem, memfree, load, vcpus, vms
       target_selector: mem
     # configuration: Local system configurations
     configuration:

node-daemon/pvcnoded/Daemon.py

@@ -48,7 +48,7 @@ import re
 import json
 
 # Daemon version
-version = "0.9.47"
+version = "0.9.53"
 
 
 ##########################################################
@@ -943,7 +943,9 @@ def entrypoint():
 
             # Add any missing OSDs to the list
             for osd in [osd for osd in new_osd_list if osd not in osd_list]:
-                d_osd[osd] = CephInstance.CephOSDInstance(zkhandler, this_node, osd)
+                d_osd[osd] = CephInstance.CephOSDInstance(
+                    zkhandler, logger, this_node, osd
+                )
 
             # Remove any deleted OSDs from the list
             for osd in [osd for osd in osd_list if osd not in new_osd_list]:
@@ -963,7 +965,9 @@ def entrypoint():
 
             # Add any missing pools to the list
             for pool in [pool for pool in new_pool_list if pool not in pool_list]:
-                d_pool[pool] = CephInstance.CephPoolInstance(zkhandler, this_node, pool)
+                d_pool[pool] = CephInstance.CephPoolInstance(
+                    zkhandler, logger, this_node, pool
+                )
                 # Prepare the volume components for this pool
                 volume_list[pool] = list()
                 d_volume[pool] = dict()
@@ -993,7 +997,7 @@ def entrypoint():
                         if volume not in volume_list[pool]
                     ]:
                         d_volume[pool][volume] = CephInstance.CephVolumeInstance(
-                            zkhandler, this_node, pool, volume
+                            zkhandler, logger, this_node, pool, volume
                         )
 
                     # Remove any deleted volumes from the list

node-daemon/pvcnoded/objects/CephInstance.py

@@ -21,7 +21,6 @@
 
 import time
 import json
-import psutil
 
 import daemon_lib.common as common
 
@@ -99,12 +98,15 @@ def get_detect_device(detect_string):
 
 
 class CephOSDInstance(object):
-    def __init__(self, zkhandler, this_node, osd_id):
+    def __init__(self, zkhandler, logger, this_node, osd_id):
         self.zkhandler = zkhandler
+        self.logger = logger
         self.this_node = this_node
         self.osd_id = osd_id
         self.node = None
-        self.size = None
+        self.device = None
+        self.vg = None
+        self.lv = None
         self.stats = dict()
 
         @self.zkhandler.zk_conn.DataWatch(
@@ -141,6 +143,121 @@ class CephOSDInstance(object):
             if data and data != self.stats:
                 self.stats = json.loads(data)
 
+        @self.zkhandler.zk_conn.DataWatch(
+            self.zkhandler.schema.path("osd.device", self.osd_id)
+        )
+        def watch_osd_device(data, stat, event=""):
+            if event and event.type == "DELETED":
+                # The key has been deleted after existing before; terminate this watcher
+                # because this class instance is about to be reaped in Daemon.py
+                return False
+
+            try:
+                data = data.decode("ascii")
+            except AttributeError:
+                data = ""
+
+            if data and data != self.device:
+                self.device = data
+
+        # Exception conditional for migration from schema v7 to schema v8
+        try:
+
+            @self.zkhandler.zk_conn.DataWatch(
+                self.zkhandler.schema.path("osd.vg", self.osd_id)
+            )
+            def watch_osd_vg(data, stat, event=""):
+                if event and event.type == "DELETED":
+                    # The key has been deleted after existing before; terminate this watcher
+                    # because this class instance is about to be reaped in Daemon.py
+                    return False
+
+                try:
+                    data = data.decode("ascii")
+                except AttributeError:
+                    data = ""
+
+                if data and data != self.vg:
+                    self.vg = data
+
+            @self.zkhandler.zk_conn.DataWatch(
+                self.zkhandler.schema.path("osd.lv", self.osd_id)
+            )
+            def watch_osd_lv(data, stat, event=""):
+                if event and event.type == "DELETED":
+                    # The key has been deleted after existing before; terminate this watcher
+                    # because this class instance is about to be reaped in Daemon.py
+                    return False
+
+                try:
+                    data = data.decode("ascii")
+                except AttributeError:
+                    data = ""
+
+                if data and data != self.lv:
+                    self.lv = data
+
+            if self.node == self.this_node.name:
+                self.update_information()
+        except TypeError:
+            return
+
+    def update_information(self):
+        if self.vg is not None and self.lv is not None:
+            find_device = f"/dev/{self.vg}/{self.lv}"
+        else:
+            find_device = self.device
+
+        self.logger.out(
+            f"Updating stored disk information for OSD {self.osd_id}",
+            state="i",
+        )
+
+        retcode, stdout, stderr = common.run_os_command(
+            f"ceph-volume lvm list {find_device}"
+        )
+        osd_blockdev = None
+        osd_fsid = None
+        osd_clusterfsid = None
+        osd_device = None
+        for line in stdout.split("\n"):
+            if "block device" in line:
+                osd_blockdev = line.split()[-1]
+            if "osd fsid" in line:
+                osd_fsid = line.split()[-1]
+            if "cluster fsid" in line:
+                osd_clusterfsid = line.split()[-1]
+            if "devices" in line:
+                osd_device = line.split()[-1]
+
+        if not osd_blockdev or not osd_fsid or not osd_clusterfsid or not osd_device:
+            self.logger.out(
+                f"Failed to find updated OSD information via ceph-volume for {find_device}",
+                state="e",
+            )
+            return
+
+        # Split OSD blockdev into VG and LV components
+        # osd_blockdev = /dev/ceph-<uuid>/osd-block-<uuid>
+        _, _, osd_vg, osd_lv = osd_blockdev.split("/")
+
+        # Except for potentially the "osd.device", this should never change, but this ensures
+        # that the data is added at lease once on initialization for existing OSDs.
+        self.zkhandler.write(
+            [
+                (("osd.device", self.osd_id), osd_device),
+                (("osd.fsid", self.osd_id), ""),
+                (("osd.ofsid", self.osd_id), osd_fsid),
+                (("osd.cfsid", self.osd_id), osd_clusterfsid),
+                (("osd.lvm", self.osd_id), ""),
+                (("osd.vg", self.osd_id), osd_vg),
+                (("osd.lv", self.osd_id), osd_lv),
+            ]
+        )
+        self.device = osd_device
+        self.vg = osd_vg
+        self.lv = osd_lv
+
     @staticmethod
     def add_osd(
         zkhandler, logger, node, device, weight, ext_db_flag=False, ext_db_ratio=0.05
@@ -230,24 +347,39 @@ class CephOSDInstance(object):
                 print(stderr)
                 raise Exception
 
-            # 4a. Get OSD FSID
+            # 4a. Get OSD information
             logger.out(
-                "Getting OSD FSID for ID {} on {}".format(osd_id, device), state="i"
+                "Getting OSD information for ID {} on {}".format(osd_id, device),
+                state="i",
             )
             retcode, stdout, stderr = common.run_os_command(
                 "ceph-volume lvm list {device}".format(device=device)
             )
             for line in stdout.split("\n"):
+                if "block device" in line:
+                    osd_blockdev = line.split()[-1]
                 if "osd fsid" in line:
                     osd_fsid = line.split()[-1]
+                if "cluster fsid" in line:
+                    osd_clusterfsid = line.split()[-1]
+                if "devices" in line:
+                    osd_device = line.split()[-1]
 
             if not osd_fsid:
                 print("ceph-volume lvm list")
-                print("Could not find OSD fsid in data:")
+                print("Could not find OSD information in data:")
                 print(stdout)
                 print(stderr)
                 raise Exception
 
+            # Split OSD blockdev into VG and LV components
+            # osd_blockdev = /dev/ceph-<uuid>/osd-block-<uuid>
+            _, _, osd_vg, osd_lv = osd_blockdev.split("/")
+
+            # Reset whatever we were given to Ceph's /dev/xdX naming
+            if device != osd_device:
+                device = osd_device
+
             # 4b. Activate the OSD
             logger.out("Activating new OSD disk with ID {}".format(osd_id), state="i")
             retcode, stdout, stderr = common.run_os_command(
@@ -275,6 +407,7 @@ class CephOSDInstance(object):
                 print(stdout)
                 print(stderr)
                 raise Exception
+
             time.sleep(0.5)
 
             # 6. Verify it started
@@ -297,7 +430,16 @@ class CephOSDInstance(object):
                     (("osd.node", osd_id), node),
                     (("osd.device", osd_id), device),
                     (("osd.db_device", osd_id), db_device),
-                    (("osd.stats", osd_id), "{}"),
+                    (("osd.fsid", osd_id), ""),
+                    (("osd.ofsid", osd_id), osd_fsid),
+                    (("osd.cfsid", osd_id), osd_clusterfsid),
+                    (("osd.lvm", osd_id), ""),
+                    (("osd.vg", osd_id), osd_vg),
+                    (("osd.lv", osd_id), osd_lv),
+                    (
+                        ("osd.stats", osd_id),
+                        '{"uuid": "|", "up": 0, "in": 0, "primary_affinity": "|", "utilization": "|", "var": "|", "pgs": "|", "kb": "|", "weight": "|", "reweight": "|", "node": "|", "used": "|", "avail": "|", "wr_ops": "|", "wr_data": "|", "rd_ops": "|", "rd_data": "|", "state": "|"}',
+                    ),
                 ]
             )
 
@@ -310,8 +452,37 @@ class CephOSDInstance(object):
             return False
 
     @staticmethod
-    def remove_osd(zkhandler, logger, osd_id, osd_obj):
-        logger.out("Removing OSD disk {}".format(osd_id), state="i")
+    def replace_osd(
+        zkhandler,
+        logger,
+        node,
+        osd_id,
+        old_device,
+        new_device,
+        weight,
+        ext_db_flag=False,
+    ):
+        # Handle a detect device if that is passed
+        if match(r"detect:", new_device):
+            ddevice = get_detect_device(new_device)
+            if ddevice is None:
+                logger.out(
+                    f"Failed to determine block device from detect string {new_device}",
+                    state="e",
+                )
+                return False
+            else:
+                logger.out(
+                    f"Determined block device {ddevice} from detect string {new_device}",
+                    state="i",
+                )
+                new_device = ddevice
+
+        # We are ready to create a new OSD on this node
+        logger.out(
+            "Replacing OSD {} disk with block device {}".format(osd_id, new_device),
+            state="i",
+        )
         try:
             # Verify the OSD is present
             retcode, stdout, stderr = common.run_os_command("ceph osd ls")
@@ -343,25 +514,371 @@ class CephOSDInstance(object):
                 print(stderr)
                 raise Exception
 
-            # 2. Wait for the OSD to flush
-            logger.out("Flushing OSD disk with ID {}".format(osd_id), state="i")
-            osd_string = str()
+            # 2. Wait for the OSD to be safe to remove (but don't wait for rebalancing to complete)
+            logger.out("Waiting for OSD {osd_id} to be safe to remove", state="i")
             while True:
-                try:
-                    retcode, stdout, stderr = common.run_os_command(
-                        "ceph pg dump osds --format json"
-                    )
-                    dump_string = json.loads(stdout)
-                    for osd in dump_string:
-                        if str(osd["osd"]) == osd_id:
-                            osd_string = osd
-                    num_pgs = osd_string["num_pgs"]
-                    if num_pgs > 0:
-                        time.sleep(5)
-                    else:
-                        raise Exception
-                except Exception:
+                retcode, stdout, stderr = common.run_os_command(
+                    f"ceph osd safe-to-destroy osd.{osd_id}"
+                )
+                if retcode in [0, 11]:
+                    # Code 0 = success
+                    # Code 11 = "Error EAGAIN: OSD(s) 5 have no reported stats, and not all PGs are active+clean; we cannot draw any conclusions." which means all PGs have been remappped but backfill is still occurring
                     break
+                else:
+                    time.sleep(5)
+
+            # 3. Stop the OSD process
+            logger.out("Stopping OSD disk with ID {}".format(osd_id), state="i")
+            retcode, stdout, stderr = common.run_os_command(
+                "systemctl stop ceph-osd@{}".format(osd_id)
+            )
+            if retcode:
+                print("systemctl stop")
+                print(stdout)
+                print(stderr)
+                raise Exception
+            time.sleep(2)
+
+            # 4. Destroy the OSD
+            logger.out("Destroying OSD with ID {osd_id}", state="i")
+            retcode, stdout, stderr = common.run_os_command(
+                f"ceph osd destroy {osd_id} --yes-i-really-mean-it"
+            )
+            if retcode:
+                print("ceph osd destroy")
+                print(stdout)
+                print(stderr)
+                raise Exception
+
+            # 5. Adjust the weight
+            logger.out(
+                "Adjusting weight of OSD disk with ID {} in CRUSH map".format(osd_id),
+                state="i",
+            )
+            retcode, stdout, stderr = common.run_os_command(
+                "ceph osd crush reweight osd.{osdid} {weight}".format(
+                    osdid=osd_id, weight=weight
+                )
+            )
+            if retcode:
+                print("ceph osd crush reweight")
+                print(stdout)
+                print(stderr)
+                raise Exception
+
+            # 6a. Zap the new disk to ensure it is ready to go
+            logger.out("Zapping disk {}".format(new_device), state="i")
+            retcode, stdout, stderr = common.run_os_command(
+                "ceph-volume lvm zap --destroy {}".format(new_device)
+            )
+            if retcode:
+                print("ceph-volume lvm zap")
+                print(stdout)
+                print(stderr)
+                raise Exception
+
+            dev_flags = "--data {}".format(new_device)
+
+            # 6b. Prepare the logical volume if ext_db_flag
+            if ext_db_flag:
+                db_device = "osd-db/osd-{}".format(osd_id)
+                dev_flags += " --block.db {}".format(db_device)
+            else:
+                db_device = ""
+
+            # 6c. Replace the OSD
+            logger.out(
+                "Preparing LVM for replaced OSD {} disk on {}".format(
+                    osd_id, new_device
+                ),
+                state="i",
+            )
+            retcode, stdout, stderr = common.run_os_command(
+                "ceph-volume lvm prepare --osd-id {osdid} --bluestore {devices}".format(
+                    osdid=osd_id, devices=dev_flags
+                )
+            )
+            if retcode:
+                print("ceph-volume lvm prepare")
+                print(stdout)
+                print(stderr)
+                raise Exception
+
+            # 7a. Get OSD information
+            logger.out(
+                "Getting OSD information for ID {} on {}".format(osd_id, new_device),
+                state="i",
+            )
+            retcode, stdout, stderr = common.run_os_command(
+                "ceph-volume lvm list {device}".format(device=new_device)
+            )
+            for line in stdout.split("\n"):
+                if "block device" in line:
+                    osd_blockdev = line.split()[-1]
+                if "osd fsid" in line:
+                    osd_fsid = line.split()[-1]
+                if "cluster fsid" in line:
+                    osd_clusterfsid = line.split()[-1]
+                if "devices" in line:
+                    osd_device = line.split()[-1]
+
+            if not osd_fsid:
+                print("ceph-volume lvm list")
+                print("Could not find OSD information in data:")
+                print(stdout)
+                print(stderr)
+                raise Exception
+
+            # Split OSD blockdev into VG and LV components
+            # osd_blockdev = /dev/ceph-<uuid>/osd-block-<uuid>
+            _, _, osd_vg, osd_lv = osd_blockdev.split("/")
+
+            # Reset whatever we were given to Ceph's /dev/xdX naming
+            if new_device != osd_device:
+                new_device = osd_device
+
+            # 7b. Activate the OSD
+            logger.out("Activating new OSD disk with ID {}".format(osd_id), state="i")
+            retcode, stdout, stderr = common.run_os_command(
+                "ceph-volume lvm activate --bluestore {osdid} {osdfsid}".format(
+                    osdid=osd_id, osdfsid=osd_fsid
+                )
+            )
+            if retcode:
+                print("ceph-volume lvm activate")
+                print(stdout)
+                print(stderr)
+                raise Exception
+
+            time.sleep(0.5)
+
+            # 8. Verify it started
+            retcode, stdout, stderr = common.run_os_command(
+                "systemctl status ceph-osd@{osdid}".format(osdid=osd_id)
+            )
+            if retcode:
+                print("systemctl status")
+                print(stdout)
+                print(stderr)
+                raise Exception
+
+            # 9. Update Zookeeper information
+            logger.out(
+                "Adding new OSD disk with ID {} to Zookeeper".format(osd_id), state="i"
+            )
+            zkhandler.write(
+                [
+                    (("osd", osd_id), ""),
+                    (("osd.node", osd_id), node),
+                    (("osd.device", osd_id), new_device),
+                    (("osd.db_device", osd_id), db_device),
+                    (("osd.fsid", osd_id), ""),
+                    (("osd.ofsid", osd_id), osd_fsid),
+                    (("osd.cfsid", osd_id), osd_clusterfsid),
+                    (("osd.lvm", osd_id), ""),
+                    (("osd.vg", osd_id), osd_vg),
+                    (("osd.lv", osd_id), osd_lv),
+                    (
+                        ("osd.stats", osd_id),
+                        '{"uuid": "|", "up": 0, "in": 0, "primary_affinity": "|", "utilization": "|", "var": "|", "pgs": "|", "kb": "|", "weight": "|", "reweight": "|", "node": "|", "used": "|", "avail": "|", "wr_ops": "|", "wr_data": "|", "rd_ops": "|", "rd_data": "|", "state": "|"}',
+                    ),
+                ]
+            )
+
+            # Log it
+            logger.out(
+                "Replaced OSD {} disk with device {}".format(osd_id, new_device),
+                state="o",
+            )
+            return True
+        except Exception as e:
+            # Log it
+            logger.out("Failed to replace OSD {} disk: {}".format(osd_id, e), state="e")
+            return False
+
+    @staticmethod
+    def refresh_osd(zkhandler, logger, node, osd_id, device, ext_db_flag):
+        # Handle a detect device if that is passed
+        if match(r"detect:", device):
+            ddevice = get_detect_device(device)
+            if ddevice is None:
+                logger.out(
+                    f"Failed to determine block device from detect string {device}",
+                    state="e",
+                )
+                return False
+            else:
+                logger.out(
+                    f"Determined block device {ddevice} from detect string {device}",
+                    state="i",
+                )
+                device = ddevice
+
+        # We are ready to create a new OSD on this node
+        logger.out(
+            "Refreshing OSD {} disk on block device {}".format(osd_id, device),
+            state="i",
+        )
+        try:
+            # 1. Verify the OSD is present
+            retcode, stdout, stderr = common.run_os_command("ceph osd ls")
+            osd_list = stdout.split("\n")
+            if osd_id not in osd_list:
+                logger.out(
+                    "Could not find OSD {} in the cluster".format(osd_id), state="e"
+                )
+                return True
+
+            dev_flags = "--data {}".format(device)
+
+            if ext_db_flag:
+                db_device = "osd-db/osd-{}".format(osd_id)
+                dev_flags += " --block.db {}".format(db_device)
+            else:
+                db_device = ""
+
+            # 2. Get OSD information
+            logger.out(
+                "Getting OSD information for ID {} on {}".format(osd_id, device),
+                state="i",
+            )
+            retcode, stdout, stderr = common.run_os_command(
+                "ceph-volume lvm list {device}".format(device=device)
+            )
+            for line in stdout.split("\n"):
+                if "block device" in line:
+                    osd_blockdev = line.split()[-1]
+                if "osd fsid" in line:
+                    osd_fsid = line.split()[-1]
+                if "cluster fsid" in line:
+                    osd_clusterfsid = line.split()[-1]
+                if "devices" in line:
+                    osd_device = line.split()[-1]
+
+            if not osd_fsid:
+                print("ceph-volume lvm list")
+                print("Could not find OSD information in data:")
+                print(stdout)
+                print(stderr)
+                raise Exception
+
+            # Split OSD blockdev into VG and LV components
+            # osd_blockdev = /dev/ceph-<uuid>/osd-block-<uuid>
+            _, _, osd_vg, osd_lv = osd_blockdev.split("/")
+
+            # Reset whatever we were given to Ceph's /dev/xdX naming
+            if device != osd_device:
+                device = osd_device
+
+            # 3. Activate the OSD
+            logger.out("Activating new OSD disk with ID {}".format(osd_id), state="i")
+            retcode, stdout, stderr = common.run_os_command(
+                "ceph-volume lvm activate --bluestore {osdid} {osdfsid}".format(
+                    osdid=osd_id, osdfsid=osd_fsid
+                )
+            )
+            if retcode:
+                print("ceph-volume lvm activate")
+                print(stdout)
+                print(stderr)
+                raise Exception
+
+            time.sleep(0.5)
+
+            # 4. Verify it started
+            retcode, stdout, stderr = common.run_os_command(
+                "systemctl status ceph-osd@{osdid}".format(osdid=osd_id)
+            )
+            if retcode:
+                print("systemctl status")
+                print(stdout)
+                print(stderr)
+                raise Exception
+
+            # 5. Update Zookeeper information
+            logger.out(
+                "Adding new OSD disk with ID {} to Zookeeper".format(osd_id), state="i"
+            )
+            zkhandler.write(
+                [
+                    (("osd", osd_id), ""),
+                    (("osd.node", osd_id), node),
+                    (("osd.device", osd_id), device),
+                    (("osd.db_device", osd_id), db_device),
+                    (("osd.fsid", osd_id), ""),
+                    (("osd.ofsid", osd_id), osd_fsid),
+                    (("osd.cfsid", osd_id), osd_clusterfsid),
+                    (("osd.lvm", osd_id), ""),
+                    (("osd.vg", osd_id), osd_vg),
+                    (("osd.lv", osd_id), osd_lv),
+                    (
+                        ("osd.stats", osd_id),
+                        '{"uuid": "|", "up": 0, "in": 0, "primary_affinity": "|", "utilization": "|", "var": "|", "pgs": "|", "kb": "|", "weight": "|", "reweight": "|", "node": "|", "used": "|", "avail": "|", "wr_ops": "|", "wr_data": "|", "rd_ops": "|", "rd_data": "|", "state": "|"}',
+                    ),
+                ]
+            )
+
+            # Log it
+            logger.out("Refreshed OSD {} disk on {}".format(osd_id, device), state="o")
+            return True
+        except Exception as e:
+            # Log it
+            logger.out("Failed to refresh OSD {} disk: {}".format(osd_id, e), state="e")
+            return False
+
+    @staticmethod
+    def remove_osd(zkhandler, logger, osd_id, osd_obj, force_flag):
+        logger.out("Removing OSD disk {}".format(osd_id), state="i")
+        try:
+            # Verify the OSD is present
+            retcode, stdout, stderr = common.run_os_command("ceph osd ls")
+            osd_list = stdout.split("\n")
+            if osd_id not in osd_list:
+                logger.out(
+                    "Could not find OSD {} in the cluster".format(osd_id), state="e"
+                )
+                if force_flag:
+                    logger.out("Ignoring error due to force flag", state="i")
+                else:
+                    return True
+
+            # 1. Set the OSD down and out so it will flush
+            logger.out("Setting down OSD disk with ID {}".format(osd_id), state="i")
+            retcode, stdout, stderr = common.run_os_command(
+                "ceph osd down {}".format(osd_id)
+            )
+            if retcode:
+                print("ceph osd down")
+                print(stdout)
+                print(stderr)
+                if force_flag:
+                    logger.out("Ignoring error due to force flag", state="i")
+                else:
+                    raise Exception
+
+            logger.out("Setting out OSD disk with ID {}".format(osd_id), state="i")
+            retcode, stdout, stderr = common.run_os_command(
+                "ceph osd out {}".format(osd_id)
+            )
+            if retcode:
+                print("ceph osd out")
+                print(stdout)
+                print(stderr)
+                if force_flag:
+                    logger.out("Ignoring error due to force flag", state="i")
+                else:
+                    raise Exception
+
+            # 2. Wait for the OSD to be safe to remove (but don't wait for rebalancing to complete)
+            logger.out("Waiting for OSD {osd_id} to be safe to remove", state="i")
+            while True:
+                retcode, stdout, stderr = common.run_os_command(
+                    f"ceph osd safe-to-destroy osd.{osd_id}"
+                )
+                if int(retcode) in [0, 11]:
+                    break
+                else:
+                    time.sleep(5)
 
             # 3. Stop the OSD process and wait for it to be terminated
             logger.out("Stopping OSD disk with ID {}".format(osd_id), state="i")
@@ -372,43 +889,55 @@ class CephOSDInstance(object):
                 print("systemctl stop")
                 print(stdout)
                 print(stderr)
-                raise Exception
-
-            # FIXME: There has to be a better way to do this /shrug
-            while True:
-                is_osd_up = False
-                # Find if there is a process named ceph-osd with arg '--id {id}'
-                for p in psutil.process_iter(attrs=["name", "cmdline"]):
-                    if "ceph-osd" == p.info["name"] and "--id {}".format(
-                        osd_id
-                    ) in " ".join(p.info["cmdline"]):
-                        is_osd_up = True
-                # If there isn't, continue
-                if not is_osd_up:
-                    break
+                if force_flag:
+                    logger.out("Ignoring error due to force flag", state="i")
+                else:
+                    raise Exception
+            time.sleep(2)
 
             # 4. Determine the block devices
-            retcode, stdout, stderr = common.run_os_command(
-                "readlink /var/lib/ceph/osd/ceph-{}/block".format(osd_id)
+            osd_vg = zkhandler.read(("osd.vg", osd_id))
+            osd_lv = zkhandler.read(("osd.lv", osd_id))
+            osd_lvm = f"/dev/{osd_vg}/{osd_lv}"
+            osd_device = None
+
+            logger.out(
+                f"Getting disk info for OSD {osd_id} LV {osd_lvm}",
+                state="i",
             )
-            vg_name = stdout.split("/")[-2]  # e.g. /dev/ceph-<uuid>/osd-block-<uuid>
             retcode, stdout, stderr = common.run_os_command(
-                "vgs --separator , --noheadings -o pv_name {}".format(vg_name)
+                f"ceph-volume lvm list {osd_lvm}"
             )
-            pv_block = stdout.strip()
+            for line in stdout.split("\n"):
+                if "devices" in line:
+                    osd_device = line.split()[-1]
+
+            if not osd_device:
+                print("ceph-volume lvm list")
+                print("Could not find OSD information in data:")
+                print(stdout)
+                print(stderr)
+                if force_flag:
+                    logger.out("Ignoring error due to force flag", state="i")
+                else:
+                    raise Exception
 
             # 5. Zap the volumes
             logger.out(
-                "Zapping OSD disk with ID {} on {}".format(osd_id, pv_block), state="i"
+                "Zapping OSD {} disk on {}".format(osd_id, osd_device),
+                state="i",
             )
             retcode, stdout, stderr = common.run_os_command(
-                "ceph-volume lvm zap --destroy {}".format(pv_block)
+                "ceph-volume lvm zap --destroy {}".format(osd_device)
             )
             if retcode:
                 print("ceph-volume lvm zap")
                 print(stdout)
                 print(stderr)
-                raise Exception
+                if force_flag:
+                    logger.out("Ignoring error due to force flag", state="i")
+                else:
+                    raise Exception
 
             # 6. Purge the OSD from Ceph
             logger.out("Purging OSD disk with ID {}".format(osd_id), state="i")
@@ -419,7 +948,10 @@ class CephOSDInstance(object):
                 print("ceph osd purge")
                 print(stdout)
                 print(stderr)
-                raise Exception
+                if force_flag:
+                    logger.out("Ignoring error due to force flag", state="i")
+                else:
+                    raise Exception
 
             # 7. Remove the DB device
             if zkhandler.exists(("osd.db_device", osd_id)):
@@ -605,8 +1137,9 @@ class CephOSDInstance(object):
 
 
 class CephPoolInstance(object):
-    def __init__(self, zkhandler, this_node, name):
+    def __init__(self, zkhandler, logger, this_node, name):
         self.zkhandler = zkhandler
+        self.logger = logger
         self.this_node = this_node
         self.name = name
         self.pgs = ""
@@ -648,8 +1181,9 @@ class CephPoolInstance(object):
 
 
 class CephVolumeInstance(object):
-    def __init__(self, zkhandler, this_node, pool, name):
+    def __init__(self, zkhandler, logger, this_node, pool, name):
         self.zkhandler = zkhandler
+        self.logger = logger
         self.this_node = this_node
         self.pool = pool
         self.name = name
@@ -705,8 +1239,9 @@ class CephSnapshotInstance(object):
 # Primary command function
 # This command pipe is only used for OSD adds and removes
 def ceph_command(zkhandler, logger, this_node, data, d_osd):
-    # Get the command and args
-    command, args = data.split()
+    # Get the command and args; the * + join ensures arguments with spaces (e.g. detect strings) are recombined right
+    command, *args = data.split()
+    args = " ".join(args)
 
     # Adding a new OSD
     if command == "osd_add":
@@ -732,9 +1267,63 @@ def ceph_command(zkhandler, logger, this_node, data, d_osd):
                 # Wait 1 seconds before we free the lock, to ensure the client hits the lock
                 time.sleep(1)
 
+    # Replacing an OSD
+    if command == "osd_replace":
+        node, osd_id, old_device, new_device, weight, ext_db_flag = args.split(",")
+        ext_db_flag = bool(strtobool(ext_db_flag))
+        if node == this_node.name:
+            # Lock the command queue
+            zk_lock = zkhandler.writelock("base.cmd.ceph")
+            with zk_lock:
+                # Add the OSD
+                result = CephOSDInstance.replace_osd(
+                    zkhandler,
+                    logger,
+                    node,
+                    osd_id,
+                    old_device,
+                    new_device,
+                    weight,
+                    ext_db_flag,
+                )
+                # Command succeeded
+                if result:
+                    # Update the command queue
+                    zkhandler.write([("base.cmd.ceph", "success-{}".format(data))])
+                # Command failed
+                else:
+                    # Update the command queue
+                    zkhandler.write([("base.cmd.ceph", "failure-{}".format(data))])
+                # Wait 1 seconds before we free the lock, to ensure the client hits the lock
+                time.sleep(1)
+
+    # Refreshing an OSD
+    if command == "osd_refresh":
+        node, osd_id, device, ext_db_flag = args.split(",")
+        ext_db_flag = bool(strtobool(ext_db_flag))
+        if node == this_node.name:
+            # Lock the command queue
+            zk_lock = zkhandler.writelock("base.cmd.ceph")
+            with zk_lock:
+                # Add the OSD
+                result = CephOSDInstance.refresh_osd(
+                    zkhandler, logger, node, osd_id, device, ext_db_flag
+                )
+                # Command succeeded
+                if result:
+                    # Update the command queue
+                    zkhandler.write([("base.cmd.ceph", "success-{}".format(data))])
+                # Command failed
+                else:
+                    # Update the command queue
+                    zkhandler.write([("base.cmd.ceph", "failure-{}".format(data))])
+                # Wait 1 seconds before we free the lock, to ensure the client hits the lock
+                time.sleep(1)
+
     # Removing an OSD
     elif command == "osd_remove":
-        osd_id = args
+        osd_id, force = args.split(",")
+        force_flag = bool(strtobool(force))
 
         # Verify osd_id is in the list
         if d_osd[osd_id] and d_osd[osd_id].node == this_node.name:
@@ -743,7 +1332,7 @@ def ceph_command(zkhandler, logger, this_node, data, d_osd):
             with zk_lock:
                 # Remove the OSD
                 result = CephOSDInstance.remove_osd(
-                    zkhandler, logger, osd_id, d_osd[osd_id]
+                    zkhandler, logger, osd_id, d_osd[osd_id], force_flag
                 )
                 # Command succeeded
                 if result:

node-daemon/pvcnoded/util/keepalive.py

@@ -307,8 +307,14 @@ def collect_ceph_stats(logger, config, zkhandler, this_node, queue):
                         "var": osd["var"],
                         "pgs": osd["pgs"],
                         "kb": osd["kb"],
+                        "kb_used": osd["kb_used"],
+                        "kb_used_data": osd["kb_used_data"],
+                        "kb_used_omap": osd["kb_used_omap"],
+                        "kb_used_meta": osd["kb_used_meta"],
+                        "kb_avail": osd["kb_avail"],
                         "weight": osd["crush_weight"],
                         "reweight": osd["reweight"],
+                        "class": osd["device_class"],
                     }
                 }
             )