[Qemu-devel] [RFC 2/2] rbd: implement bdrv_set_key

[Josh Durgin <josh.durgin@dreamhost.com>]

We can't connect to the cluster before the key is set, so this step is
moved to qemu_rbd_open_finish. Each bdrv function will fail before the
key is set.

Signed-off-by: Josh Durgin <josh.durgin@dreamhost.com>
---
 block/rbd.c |  160 +++++++++++++++++++++++++++++++++++++++++-----------------
 1 files changed, 113 insertions(+), 47 deletions(-)

diff --git a/block/rbd.c b/block/rbd.c
index c684e0c..005222a 100644
--- a/block/rbd.c
+++ b/block/rbd.c
@@ -83,6 +83,8 @@ typedef struct BDRVRBDState {
     rados_ioctx_t io_ctx;
     rbd_image_t image;
     char name[RBD_MAX_IMAGE_NAME_SIZE];
+    char pool[RBD_MAX_POOL_NAME_SIZE];
+    char conf[RBD_MAX_CONF_SIZE];
     int qemu_aio_count;
     char *snap;
     int event_reader_pos;
@@ -271,6 +273,17 @@ static int qemu_rbd_set_conf(rados_t cluster, const char *conf)
     return ret;
 }
 
+static int qemu_rbd_connect(rados_t cluster,
+                            const char *pool,
+                            rados_ioctx_t *io_ctx)
+{
+    int ret = rados_connect(cluster);
+    if (ret < 0) {
+        return ret;
+    }
+    return rados_ioctx_create(cluster, pool, io_ctx);
+}
+
 static int qemu_rbd_create(const char *filename, QEMUOptionParameter *options)
 {
     int64_t bytes = 0;
@@ -332,16 +345,15 @@ static int qemu_rbd_create(const char *filename, QEMUOptionParameter *options)
         return -EIO;
     }
 
-    if (rados_connect(cluster) < 0) {
-        error_report("error connecting");
-        rados_shutdown(cluster);
-        return -EIO;
-    }
-
-    if (rados_ioctx_create(cluster, pool, &io_ctx) < 0) {
-        error_report("error opening pool %s", pool);
+    ret = qemu_rbd_connect(cluster, pool, &io_ctx);
+    if (ret < 0) {
+        error_report("error connecting: %s", strerror(-ret));
         rados_shutdown(cluster);
-        return -EIO;
+        if (ret == -EPERM || ret == -ENOTSUP) {
+            return -EPERM;
+        } else {
+            return -EIO;
+        }
     }
 
     ret = rbd_create(io_ctx, name, bytes, &obj_order);
@@ -431,57 +443,46 @@ static int qemu_rbd_aio_flush_cb(void *opaque)
     return (s->qemu_aio_count > 0);
 }
 
-static int qemu_rbd_open(BlockDriverState *bs, const char *filename, int flags)
+
+static int qemu_rbd_finish_open(BDRVRBDState *s, const char *key)
 {
-    BDRVRBDState *s = bs->opaque;
-    char pool[RBD_MAX_POOL_NAME_SIZE];
-    char snap_buf[RBD_MAX_SNAP_NAME_SIZE];
-    char conf[RBD_MAX_CONF_SIZE];
     char clientname_buf[RBD_MAX_CONF_SIZE];
     char *clientname;
     int r;
 
-    if (qemu_rbd_parsename(filename, pool, sizeof(pool),
-                           snap_buf, sizeof(snap_buf),
-                           s->name, sizeof(s->name),
-                           conf, sizeof(conf)) < 0) {
-        return -EINVAL;
-    }
-
-    clientname = qemu_rbd_parse_clientname(conf, clientname_buf);
+    clientname = qemu_rbd_parse_clientname(s->conf, clientname_buf);
     r = rados_create(&s->cluster, clientname);
     if (r < 0) {
         error_report("error initializing");
         return r;
     }
 
-    s->snap = NULL;
-    if (snap_buf[0] != '\0') {
-        s->snap = g_strdup(snap_buf);
-    }
-
-    if (strstr(conf, "conf=") == NULL) {
+    if (strstr(s->conf, "conf=") == NULL) {
         /* try default location, but ignore failure */
         rados_conf_read_file(s->cluster, NULL);
     }
 
-    if (conf[0] != '\0') {
-        r = qemu_rbd_set_conf(s->cluster, conf);
+    if (s->conf[0] != '\0') {
+        r = qemu_rbd_set_conf(s->cluster, s->conf);
         if (r < 0) {
             error_report("error setting config options");
             goto failed_shutdown;
         }
     }
 
-    r = rados_connect(s->cluster);
-    if (r < 0) {
-        error_report("error connecting");
-        goto failed_shutdown;
+    if (key) {
+        r = rados_conf_set(s->cluster, "key", key);
+        if (r < 0) {
+            error_report("error setting key");
+            goto failed_shutdown;
+        }
     }
 
-    r = rados_ioctx_create(s->cluster, pool, &s->io_ctx);
-    if (r < 0) {
-        error_report("error opening pool %s", pool);
+    r = qemu_rbd_connect(s->cluster, s->pool, &s->io_ctx);
+    if (r == -EPERM || r == -ENOTSUP) {
+        return -EPERM;
+    } else if (r < 0) {
+        error_report("error connecting: %s", strerror(-r));
         goto failed_shutdown;
     }
 
@@ -491,8 +492,6 @@ static int qemu_rbd_open(BlockDriverState *bs, const char *filename, int flags)
         goto failed_open;
     }
 
-    bs->read_only = (s->snap != NULL);
-
     s->event_reader_pos = 0;
     r = qemu_pipe(s->fds);
     if (r < 0) {
@@ -513,23 +512,55 @@ failed_open:
     rados_ioctx_destroy(s->io_ctx);
 failed_shutdown:
     rados_shutdown(s->cluster);
-    g_free(s->snap);
     return r;
 }
 
+static int qemu_rbd_open(BlockDriverState *bs, const char *filename, int flags)
+{
+    BDRVRBDState *s = bs->opaque;
+    char snap_buf[RBD_MAX_SNAP_NAME_SIZE];
+    int r;
+
+    if (qemu_rbd_parsename(filename, s->pool, sizeof(s->pool),
+                           snap_buf, sizeof(snap_buf),
+                           s->name, sizeof(s->name),
+                           s->conf, sizeof(s->conf)) < 0) {
+        return -EINVAL;
+    }
+
+    s->snap = NULL;
+    if (snap_buf[0] != '\0') {
+        s->snap = g_strdup(snap_buf);
+    }
+
+    bs->read_only = (s->snap != NULL);
+
+    r = qemu_rbd_finish_open(s, NULL);
+    if (r == -EPERM) {
+        bs->needs_auth = 1;
+    } else if (r < 0) {
+        g_free(s->snap);
+        return r;
+    }
+
+    return 0;
+}
+
 static void qemu_rbd_close(BlockDriverState *bs)
 {
     BDRVRBDState *s = bs->opaque;
 
-    close(s->fds[0]);
-    close(s->fds[1]);
-    qemu_aio_set_fd_handler(s->fds[RBD_FD_READ], NULL , NULL, NULL, NULL,
-        NULL);
+    if (!bdrv_key_required(bs)) {
+        close(s->fds[0]);
+        close(s->fds[1]);
+        qemu_aio_set_fd_handler(s->fds[RBD_FD_READ], NULL , NULL, NULL, NULL,
+                                NULL);
 
-    rbd_close(s->image);
-    rados_ioctx_destroy(s->io_ctx);
+        rbd_close(s->image);
+        rados_ioctx_destroy(s->io_ctx);
+        rados_shutdown(s->cluster);
+    }
     g_free(s->snap);
-    rados_shutdown(s->cluster);
 }
 
 /*
@@ -692,6 +723,9 @@ static BlockDriverAIOCB *qemu_rbd_aio_readv(BlockDriverState *bs,
                                             BlockDriverCompletionFunc *cb,
                                             void *opaque)
 {
+    if (bdrv_key_required(bs)) {
+        return NULL;
+    }
     return rbd_aio_rw_vector(bs, sector_num, qiov, nb_sectors, cb, opaque, 0);
 }
 
@@ -702,11 +736,18 @@ static BlockDriverAIOCB *qemu_rbd_aio_writev(BlockDriverState *bs,
                                              BlockDriverCompletionFunc *cb,
                                              void *opaque)
 {
+    if (bdrv_key_required(bs)) {
+        return NULL;
+    }
     return rbd_aio_rw_vector(bs, sector_num, qiov, nb_sectors, cb, opaque, 1);
 }
 
 static int qemu_rbd_co_flush(BlockDriverState *bs)
 {
+    if (bdrv_key_required(bs)) {
+        return -EPERM;
+    }
+
 #if LIBRBD_VERSION_CODE >= LIBRBD_VERSION(0, 1, 1)
     /* rbd_flush added in 0.1.1 */
     BDRVRBDState *s = bs->opaque;
@@ -721,6 +762,9 @@ static int qemu_rbd_getinfo(BlockDriverState *bs, BlockDriverInfo *bdi)
     BDRVRBDState *s = bs->opaque;
     rbd_image_info_t info;
     int r;
+    if (bdrv_key_required(bs)) {
+        return -EPERM;
+    }
 
     r = rbd_stat(s->image, &info, sizeof(info));
     if (r < 0) {
@@ -736,6 +780,9 @@ static int64_t qemu_rbd_getlength(BlockDriverState *bs)
     BDRVRBDState *s = bs->opaque;
     rbd_image_info_t info;
     int r;
+    if (bdrv_key_required(bs)) {
+        return -EPERM;
+    }
 
     r = rbd_stat(s->image, &info, sizeof(info));
     if (r < 0) {
@@ -749,6 +796,9 @@ static int qemu_rbd_truncate(BlockDriverState *bs, int64_t offset)
 {
     BDRVRBDState *s = bs->opaque;
     int r;
+    if (bdrv_key_required(bs)) {
+        return -EPERM;
+    }
 
     r = rbd_resize(s->image, offset);
     if (r < 0) {
@@ -758,11 +808,22 @@ static int qemu_rbd_truncate(BlockDriverState *bs, int64_t offset)
     return 0;
 }
 
+static int qemu_rbd_set_key(BlockDriverState *bs, const char *key)
+{
+    if (bs->valid_key) {
+        return 0;
+    }
+    return qemu_rbd_finish_open(bs->opaque, key);
+}
+
 static int qemu_rbd_snap_create(BlockDriverState *bs,
                                 QEMUSnapshotInfo *sn_info)
 {
     BDRVRBDState *s = bs->opaque;
     int r;
+    if (bdrv_key_required(bs)) {
+        return -EPERM;
+    }
 
     if (sn_info->name[0] == '\0') {
         return -EINVAL; /* we need a name for rbd snapshots */
@@ -799,6 +860,10 @@ static int qemu_rbd_snap_list(BlockDriverState *bs,
     rbd_snap_info_t *snaps;
     int max_snaps = RBD_MAX_SNAPS;
 
+    if (bdrv_key_required(bs)) {
+        return -EPERM;
+    }
+
     do {
         snaps = g_malloc(sizeof(*snaps) * max_snaps);
         snap_count = rbd_snap_list(s->image, snaps, &max_snaps);
@@ -856,6 +921,7 @@ static BlockDriver bdrv_rbd = {
     .create_options     = qemu_rbd_create_options,
     .bdrv_getlength     = qemu_rbd_getlength,
     .bdrv_truncate      = qemu_rbd_truncate,
+    .bdrv_set_key       = qemu_rbd_set_key,
     .protocol_name      = "rbd",
 
     .bdrv_aio_readv     = qemu_rbd_aio_readv,
-- 
1.7.1