Implement virDomainBlockPull for the qemu driver

The virDomainBlockPull* family of commands are enabled by the
'block_stream' and 'info block_stream' qemu monitor commands.

* src/qemu/qemu_driver.c src/qemu/qemu_monitor_text.[ch]: implement disk
  streaming by using the stream and info stream text monitor commands
* src/qemu/qemu_monitor_json.[ch]: implement commands using the qmp monitor

Signed-off-by: Adam Litke <agl@us.ibm.com>
Acked-by: Daniel P. Berrange <berrange@redhat.com>

src/qemu/qemu_driver.c

@@ -8016,6 +8016,114 @@ cleanup:
     return ret;
 }
 
+static const char *
+qemuDiskPathToAlias(virDomainObjPtr vm, const char *path) {
+    int i;
+    char *ret = NULL;
+
+    for (i = 0 ; i < vm->def->ndisks ; i++) {
+        virDomainDiskDefPtr disk = vm->def->disks[i];
+
+        if (disk->type != VIR_DOMAIN_DISK_TYPE_BLOCK &&
+            disk->type != VIR_DOMAIN_DISK_TYPE_FILE)
+            continue;
+
+        if (disk->src != NULL && STREQ(disk->src, path)) {
+            if (virAsprintf(&ret, "drive-%s", disk->info.alias) < 0) {
+                virReportOOMError();
+                return NULL;
+            }
+            break;
+        }
+    }
+
+    if (!ret) {
+        qemuReportError(VIR_ERR_INVALID_ARG,
+                        "%s", _("No device found for specified path"));
+    }
+    return ret;
+}
+
+static int
+qemuDomainBlockPullImpl(virDomainPtr dom, const char *path,
+                        virDomainBlockPullInfoPtr info,
+                        int mode)
+{
+    struct qemud_driver *driver = dom->conn->privateData;
+    virDomainObjPtr vm = NULL;
+    qemuDomainObjPrivatePtr priv;
+    char uuidstr[VIR_UUID_STRING_BUFLEN];
+    const char *device = NULL;
+    int ret = -1;
+
+    qemuDriverLock(driver);
+    virUUIDFormat(dom->uuid, uuidstr);
+    vm = virDomainFindByUUID(&driver->domains, dom->uuid);
+    if (!vm) {
+        qemuReportError(VIR_ERR_NO_DOMAIN,
+                        _("no domain with matching uuid '%s'"), uuidstr);
+        goto cleanup;
+    }
+
+    if (!virDomainObjIsActive(vm)) {
+        qemuReportError(VIR_ERR_OPERATION_INVALID,
+                        "%s", _("domain is not running"));
+        goto cleanup;
+    }
+
+    device = qemuDiskPathToAlias(vm, path);
+    if (!device) {
+        goto cleanup;
+    }
+
+    if (qemuDomainObjBeginJobWithDriver(driver, vm) < 0)
+        goto cleanup;
+    qemuDomainObjEnterMonitorWithDriver(driver, vm);
+    priv = vm->privateData;
+    ret = qemuMonitorBlockPull(priv->mon, device, info, mode);
+    qemuDomainObjExitMonitorWithDriver(driver, vm);
+    if (qemuDomainObjEndJob(vm) == 0) {
+        vm = NULL;
+        goto cleanup;
+    }
+
+cleanup:
+    VIR_FREE(device);
+    if (vm)
+        virDomainObjUnlock(vm);
+    qemuDriverUnlock(driver);
+    return ret;
+}
+
+static int
+qemuDomainBlockPull(virDomainPtr dom, const char *path,
+                    virDomainBlockPullInfoPtr info, unsigned int flags)
+{
+    virCheckFlags(0, -1);
+    return qemuDomainBlockPullImpl(dom, path, info, BLOCK_PULL_MODE_ONE);
+}
+
+static int
+qemuDomainBlockPullAll(virDomainPtr dom, const char *path, unsigned int flags)
+{
+    virCheckFlags(0, -1);
+    return qemuDomainBlockPullImpl(dom, path, NULL, BLOCK_PULL_MODE_ALL);
+}
+
+static int
+qemuDomainBlockPullAbort(virDomainPtr dom, const char *path, unsigned int flags)
+{
+    virCheckFlags(0, -1);
+    return qemuDomainBlockPullImpl(dom, path, NULL, BLOCK_PULL_MODE_ABORT);
+}
+
+static int
+qemuDomainGetBlockPullInfo(virDomainPtr dom, const char *path,
+                           virDomainBlockPullInfoPtr info, unsigned int flags)
+{
+    virCheckFlags(0, -1);
+    return qemuDomainBlockPullImpl(dom, path, info, BLOCK_PULL_MODE_INFO);
+}
 
 static virDriver qemuDriver = {
     .no = VIR_DRV_QEMU,
@@ -8138,6 +8246,10 @@ static virDriver qemuDriver = {
     .domainMigratePerform3 = qemuDomainMigratePerform3, /* 0.9.2 */
     .domainMigrateFinish3 = qemuDomainMigrateFinish3, /* 0.9.2 */
     .domainMigrateConfirm3 = qemuDomainMigrateConfirm3, /* 0.9.2 */
+    .domainBlockPull = qemuDomainBlockPull, /* 0.9.3 */
+    .domainBlockPullAll = qemuDomainBlockPullAll, /* 0.9.3 */
+    .domainBlockPullAbort = qemuDomainBlockPullAbort, /* 0.9.3 */
+    .domainGetBlockPullInfo = qemuDomainGetBlockPullInfo, /* 0.9.3 */
 };
 
 

src/qemu/qemu_monitor.c

@@ -2377,3 +2377,19 @@ int qemuMonitorScreendump(qemuMonitorPtr mon,
         ret = qemuMonitorTextScreendump(mon, file);
     return ret;
 }
+
+int qemuMonitorBlockPull(qemuMonitorPtr mon,
+                         const char *path,
+                         virDomainBlockPullInfoPtr info,
+                         int mode)
+{
+    int ret;
+
+    VIR_DEBUG("mon=%p, path=%p, info=%p, mode=%i", mon, path, info, mode);
+
+    if (mon->json)
+        ret = qemuMonitorJSONBlockPull(mon, path, info, mode);
+    else
+        ret = qemuMonitorTextBlockPull(mon, path, info, mode);
+    return ret;
+}

src/qemu/qemu_monitor.h

@@ -441,6 +441,19 @@ int qemuMonitorInjectNMI(qemuMonitorPtr mon);
 int qemuMonitorScreendump(qemuMonitorPtr mon,
                           const char *file);
 
+typedef enum {
+    BLOCK_PULL_MODE_ONE = 0,
+    BLOCK_PULL_MODE_ALL = 1,
+    BLOCK_PULL_MODE_ABORT = 2,
+    BLOCK_PULL_MODE_INFO = 3,
+} BLOCK_PULL_MODE;
+
+
+int qemuMonitorBlockPull(qemuMonitorPtr mon,
+                         const char *path,
+                         virDomainBlockPullInfoPtr info,
+                         int mode);
+
 /**
  * When running two dd process and using <> redirection, we need a
  * shell that will not truncate files.  These two strings serve that

src/qemu/qemu_monitor_json.c

@@ -2653,3 +2653,134 @@ int qemuMonitorJSONScreendump(qemuMonitorPtr mon,
     virJSONValueFree(reply);
     return ret;
 }
+
+static int qemuMonitorJSONGetBlockPullInfoOne(virJSONValuePtr entry,
+                                              const char *device,
+                                              virDomainBlockPullInfoPtr info)
+{
+    const char *this_dev;
+
+    if ((this_dev = virJSONValueObjectGetString(entry, "device")) == NULL) {
+        qemuReportError(VIR_ERR_INTERNAL_ERROR, "%s",
+                        _("entry was missing 'device'"));
+        return -1;
+    }
+    if (!STREQ(this_dev, device))
+        return -1;
+
+    if (virJSONValueObjectGetNumberUlong(entry, "offset", &info->cur) < 0) {
+        qemuReportError(VIR_ERR_INTERNAL_ERROR, "%s",
+                        _("entry was missing 'offset'"));
+        return -1;
+    }
+
+    if (virJSONValueObjectGetNumberUlong(entry, "len", &info->end) < 0) {
+        qemuReportError(VIR_ERR_INTERNAL_ERROR, "%s",
+                        _("entry was missing 'len'"));
+        return -1;
+    }
+    return 0;
+}
+
+/** qemuMonitorJSONGetBlockPullInfo:
+ * Parse Block Pull information.
+ * The reply can be a JSON array of objects or just an object.
+ */
+static int qemuMonitorJSONGetBlockPullInfo(virJSONValuePtr reply,
+                                           const char *device,
+                                           virDomainBlockPullInfoPtr info)
+{
+    virJSONValuePtr data;
+    int nr_results, i = 0;
+
+    if (!info)
+        return -1;
+
+    if ((data = virJSONValueObjectGet(reply, "return")) == NULL) {
+        qemuReportError(VIR_ERR_INTERNAL_ERROR, "%s",
+                        _("reply was missing block_pull progress information"));
+        return -1;
+    }
+
+    if (data->type == VIR_JSON_TYPE_OBJECT) {
+        if (qemuMonitorJSONGetBlockPullInfoOne(data, device, info) != 0)
+            goto not_found;
+        else
+            return 0;
+    } else if (data->type != VIR_JSON_TYPE_ARRAY) {
+        qemuReportError(VIR_ERR_INTERNAL_ERROR, "%s",
+                        _("urecognized format of block pull information"));
+        return -1;
+    }
+
+    if ((nr_results = virJSONValueArraySize(data)) < 0) {
+        qemuReportError(VIR_ERR_INTERNAL_ERROR, "%s",
+                        _("unable to determine array size"));
+        return -1;
+    }
+
+    for (i = 0; i < nr_results; i++) {
+        virJSONValuePtr entry = virJSONValueArrayGet(data, i);
+        if (qemuMonitorJSONGetBlockPullInfoOne(entry, device, info) == 0)
+            return 0;
+    }
+
+not_found:
+    qemuReportError(VIR_ERR_OPERATION_FAILED, "%s",
+                    _("No associated information for the specified disk"));
+    return -1;
+}
+
+
+int qemuMonitorJSONBlockPull(qemuMonitorPtr mon,
+                             const char *device,
+                             virDomainBlockPullInfoPtr info,
+                             int mode)
+{
+    int ret = -1;
+    virJSONValuePtr cmd = NULL;
+    virJSONValuePtr reply = NULL;
+    int parse_info = 0;
+
+    if (mode == BLOCK_PULL_MODE_ONE) {
+        cmd = qemuMonitorJSONMakeCommand("block_stream", "s:device", device, NULL);
+        parse_info = 1;
+    } else if (mode == BLOCK_PULL_MODE_ALL) {
+        cmd = qemuMonitorJSONMakeCommand("block_stream", "s:device", device,
+                                         "b:all", 1, NULL);
+    } else if (mode == BLOCK_PULL_MODE_ABORT) {
+        cmd = qemuMonitorJSONMakeCommand("block_stream", "s:device", device,
+                                         "b:stop", 1, NULL);
+    } else if (mode == BLOCK_PULL_MODE_INFO) {
+        cmd = qemuMonitorJSONMakeCommand("query-block-stream", NULL);
+        parse_info = 1;
+    }
+
+    if (!cmd)
+        return -1;
+
+    ret = qemuMonitorJSONCommand(mon, cmd, &reply);
+
+    if (ret == 0 && virJSONValueObjectHasKey(reply, "error")) {
+        if (qemuMonitorJSONHasError(reply, "DeviceNotActive"))
+            qemuReportError(VIR_ERR_OPERATION_INVALID,
+                _("No active operation on device: %s"), device);
+        else if (qemuMonitorJSONHasError(reply, "DeviceInUse"))
+            qemuReportError(VIR_ERR_OPERATION_FAILED,
+                _("Device %s in use"), device);
+        else if (qemuMonitorJSONHasError(reply, "NotSupported"))
+            qemuReportError(VIR_ERR_OPERATION_INVALID,
+                _("Operation is not supported for device: %s"), device);
+        else
+            qemuReportError(VIR_ERR_INTERNAL_ERROR, "%s",
+                _("Unexpected error"));
+        ret = -1;
+    }
+
+    if (ret == 0 && parse_info)
+        ret = qemuMonitorJSONGetBlockPullInfo(reply, device, info);
+
+    virJSONValueFree(cmd);
+    virJSONValueFree(reply);
+    return ret;
+}

src/qemu/qemu_monitor_json.h

@@ -217,5 +217,9 @@ int qemuMonitorJSONInjectNMI(qemuMonitorPtr mon);
 int qemuMonitorJSONScreendump(qemuMonitorPtr mon,
                               const char *file);
 
+int qemuMonitorJSONBlockPull(qemuMonitorPtr mon,
+                             const char *device,
+                             virDomainBlockPullInfoPtr info,
+                             int mode);
 
 #endif /* QEMU_MONITOR_JSON_H */

src/qemu/qemu_monitor_text.c

@@ -2750,3 +2750,159 @@ cleanup:
     VIR_FREE(cmd);
     return ret;
 }
+
+static int qemuMonitorTextParseBlockPullOne(const char *text,
+                                            const char *device,
+                                            virDomainBlockPullInfoPtr info,
+                                            const char **next)
+{
+    virDomainBlockPullInfo tmp;
+    char *p;
+    int mismatch = 0;
+
+    if (next == NULL)
+        return -1;
+    *next = NULL;
+
+    /*
+     * Each active stream will appear on its own line in the following format:
+     * Streaming device <device>: Completed <cur> of <end> bytes
+     */
+    if ((text = STRSKIP(text, "Streaming device ")) == NULL)
+        return -EINVAL;
+
+    if (!STREQLEN(text, device, strlen(device)))
+        mismatch = 1;
+
+    if ((text = strstr(text, ": Completed ")) == NULL)
+        return -EINVAL;
+    text += 11;
+
+    if (virStrToLong_ull (text, &p, 10, &tmp.cur))
+        return -EINVAL;
+    text = p;
+
+    if (!STRPREFIX(text, " of "))
+        return -EINVAL;
+    text += 4;
+
+    if (virStrToLong_ull (text, &p, 10, &tmp.end))
+        return -EINVAL;
+    text = p;
+
+    if (!STRPREFIX(text, " bytes"))
+        return -EINVAL;
+
+    if (mismatch) {
+        *next = STRSKIP(text, "\n");
+        return -EAGAIN;
+    }
+
+    if (info) {
+        info->cur = tmp.cur;
+        info->end = tmp.end;
+    }
+    return 0;
+}
+
+static int qemuMonitorTextParseBlockPull(const char *text,
+                                         const char *device,
+                                         virDomainBlockPullInfoPtr info)
+{
+    const char *next = NULL;
+    int ret = 0;
+
+    /* Check error: Device not found */
+    if (strstr(text, "Device '") && strstr(text, "' not found")) {
+        qemuReportError(VIR_ERR_OPERATION_INVALID, "%s", _("Device not found"));
+        return -1;
+    }
+
+    /* Check error: Already streaming this device */
+    if (strstr(text, "Device '") && strstr(text, "' is in use")) {
+        qemuReportError(VIR_ERR_OPERATION_FAILED, _("Device %s in use"),
+            device);
+        return -1;
+    }
+
+    /* Check error: Stop non-existent stream */
+    if (strstr(text, "has not been activated")) {
+        qemuReportError(VIR_ERR_OPERATION_INVALID,\
+            _("No active operation on device: %s"), device);
+        return -1;
+    }
+
+    /*
+     * Check: No active streams when calling info block_stream
+     * This is not an error condition, there are just no results to report.
+     */
+    if (strstr(text, "No active stream")) {
+        return -1;
+    }
+
+    /* Check for unsupported operation */
+    if (strstr(text, "Operation is not supported")) {
+        qemuReportError(VIR_ERR_OPERATION_INVALID,
+            _("Operation is not supported for device: %s"), device);
+        return -1;
+    }
+
+    /* No output indicates success for BlockPullAll and BlockPullAbort */
+    if (STREQ(text, ""))
+        return 0;
+
+    /* Now try to parse lines of block_stream output */
+    do {
+        ret = qemuMonitorTextParseBlockPullOne(text, device, info, &next);
+        text = next;
+    } while (text && ret == -EAGAIN);
+
+    if (ret != 0) {
+        qemuReportError(VIR_ERR_OPERATION_FAILED, "%s",
+                        _("No associated information for the specified disk"));
+        ret = -1;
+    }
+    return ret;
+}
+
+int qemuMonitorTextBlockPull(qemuMonitorPtr mon,
+                             const char *device,
+                             virDomainBlockPullInfoPtr info,
+                             int mode)
+{
+    char *cmd = NULL;
+    char *reply = NULL;
+    int ret;
+
+    if (mode == BLOCK_PULL_MODE_ONE)
+        ret = virAsprintf(&cmd, "block_stream %s", device);
+    else if (mode == BLOCK_PULL_MODE_ALL)
+        ret = virAsprintf(&cmd, "block_stream -a %s", device);
+    else if (mode == BLOCK_PULL_MODE_ABORT)
+        ret = virAsprintf(&cmd, "block_stream -s %s", device);
+    else if (mode == BLOCK_PULL_MODE_INFO)
+        ret = virAsprintf(&cmd, "info block_stream");
+    else
+        return -1;
+
+    if (ret < 0) {
+        virReportOOMError();
+        return -1;
+    }
+
+    ret = 0;
+    if (qemuMonitorHMPCommand(mon, cmd, &reply) < 0) {
+        qemuReportError(VIR_ERR_INTERNAL_ERROR,
+                        "%s", _("cannot run monitor command"));
+        ret = -1;
+        goto cleanup;
+    }
+
+    if (qemuMonitorTextParseBlockPull(reply, device, info) != 0)
+        ret = -1;
+
+cleanup:
+    VIR_FREE(cmd);
+    VIR_FREE(reply);
+    return ret;
+}

src/qemu/qemu_monitor_text.h

@@ -210,4 +210,9 @@ int qemuMonitorTextInjectNMI(qemuMonitorPtr mon);
 
 int qemuMonitorTextScreendump(qemuMonitorPtr mon, const char *file);
 
+int qemuMonitorTextBlockPull(qemuMonitorPtr mon,
+                             const char *device,
+                             virDomainBlockPullInfoPtr info,
+                             int mode);
+
 #endif /* QEMU_MONITOR_TEXT_H */