block/rbd.c

   1 /*
   2  * QEMU Block driver for RADOS (Ceph)
   3  *
   4  * Copyright (C) 2010-2011 Christian Brunner <chb@muc.de>,
   5  *                         Josh Durgin <josh.durgin@dreamhost.com>
   6  *
   7  * This work is licensed under the terms of the GNU GPL, version 2.  See
   8  * the COPYING file in the top-level directory.
   9  *
  10  * Contributions after 2012-01-13 are licensed under the terms of the
  11  * GNU GPL, version 2 or (at your option) any later version.
  12  */
  13
  14 #include "qemu/osdep.h"
  15
  16 #include <rbd/librbd.h>
  17 #include "qapi/error.h"
  18 #include "qemu/error-report.h"
  19 #include "qemu/option.h"
  20 #include "block/block_int.h"
  21 #include "crypto/secret.h"
  22 #include "qemu/cutils.h"
  23 #include "qapi/qmp/qstring.h"
  24 #include "qapi/qmp/qdict.h"
  25 #include "qapi/qmp/qjson.h"
  26 #include "qapi/qmp/qlist.h"
  27 #include "qapi/qobject-input-visitor.h"
  28 #include "qapi/qapi-visit-block-core.h"
  29
  30 /*
  31  * When specifying the image filename use:
  32  *
  33  * rbd:poolname/devicename[@snapshotname][:option1=value1[:option2=value2...]]
  34  *
  35  * poolname must be the name of an existing rados pool.
  36  *
  37  * devicename is the name of the rbd image.
  38  *
  39  * Each option given is used to configure rados, and may be any valid
  40  * Ceph option, "id", or "conf".
  41  *
  42  * The "id" option indicates what user we should authenticate as to
  43  * the Ceph cluster.  If it is excluded we will use the Ceph default
  44  * (normally 'admin').
  45  *
  46  * The "conf" option specifies a Ceph configuration file to read.  If
  47  * it is not specified, we will read from the default Ceph locations
  48  * (e.g., /etc/ceph/ceph.conf).  To avoid reading _any_ configuration
  49  * file, specify conf=/dev/null.
  50  *
  51  * Configuration values containing :, @, or = can be escaped with a
  52  * leading "\".
  53  */
  54
  55 /* rbd_aio_discard added in 0.1.2 */
  56 #if LIBRBD_VERSION_CODE >= LIBRBD_VERSION(0, 1, 2)
  57 #define LIBRBD_SUPPORTS_DISCARD
  58 #else
  59 #undef LIBRBD_SUPPORTS_DISCARD
  60 #endif
  61
  62 #define OBJ_MAX_SIZE (1UL << OBJ_DEFAULT_OBJ_ORDER)
  63
  64 #define RBD_MAX_SNAPS 100
  65
  66 /* The LIBRBD_SUPPORTS_IOVEC is defined in librbd.h */
  67 #ifdef LIBRBD_SUPPORTS_IOVEC
  68 #define LIBRBD_USE_IOVEC 1
  69 #else
  70 #define LIBRBD_USE_IOVEC 0
  71 #endif
  72
  73 typedef enum {
  74     RBD_AIO_READ,
  75     RBD_AIO_WRITE,
  76     RBD_AIO_DISCARD,
  77     RBD_AIO_FLUSH
  78 } RBDAIOCmd;
  79
  80 typedef struct RBDAIOCB {
  81     BlockAIOCB common;
  82     int64_t ret;
  83     QEMUIOVector *qiov;
  84     char *bounce;
  85     RBDAIOCmd cmd;
  86     int error;
  87     struct BDRVRBDState *s;
  88 } RBDAIOCB;
  89
  90 typedef struct RADOSCB {
  91     RBDAIOCB *acb;
  92     struct BDRVRBDState *s;
  93     int64_t size;
  94     char *buf;
  95     int64_t ret;
  96 } RADOSCB;
  97
  98 typedef struct BDRVRBDState {
  99     rados_t cluster;
 100     rados_ioctx_t io_ctx;
 101     rbd_image_t image;
 102     char *image_name;
 103     char *snap;
 104 } BDRVRBDState;
 105
 106 static int qemu_rbd_connect(rados_t *cluster, rados_ioctx_t *io_ctx,
 107                             BlockdevOptionsRbd *opts, bool cache,
 108                             const char *keypairs, const char *secretid,
 109                             Error **errp);
 110
 111 static char *qemu_rbd_next_tok(char *src, char delim, char **p)
 112 {
 113     char *end;
 114
 115     *p = NULL;
 116
 117     for (end = src; *end; ++end) {
 118         if (*end == delim) {
 119             break;
 120         }
 121         if (*end == '\\' && end[1] != '\0') {
 122             end++;
 123         }
 124     }
 125     if (*end == delim) {
 126         *p = end + 1;
 127         *end = '\0';
 128     }
 129     return src;
 130 }
 131
 132 static void qemu_rbd_unescape(char *src)
 133 {
 134     char *p;
 135
 136     for (p = src; *src; ++src, ++p) {
 137         if (*src == '\\' && src[1] != '\0') {
 138             src++;
 139         }
 140         *p = *src;
 141     }
 142     *p = '\0';
 143 }
 144
 145 static void qemu_rbd_parse_filename(const char *filename, QDict *options,
 146                                     Error **errp)
 147 {
 148     const char *start;
 149     char *p, *buf;
 150     QList *keypairs = NULL;
 151     char *found_str;
 152
 153     if (!strstart(filename, "rbd:", &start)) {
 154         error_setg(errp, "File name must start with 'rbd:'");
 155         return;
 156     }
 157
 158     buf = g_strdup(start);
 159     p = buf;
 160
 161     found_str = qemu_rbd_next_tok(p, '/', &p);
 162     if (!p) {
 163         error_setg(errp, "Pool name is required");
 164         goto done;
 165     }
 166     qemu_rbd_unescape(found_str);
 167     qdict_put_str(options, "pool", found_str);
 168
 169     if (strchr(p, '@')) {
 170         found_str = qemu_rbd_next_tok(p, '@', &p);
 171         qemu_rbd_unescape(found_str);
 172         qdict_put_str(options, "image", found_str);
 173
 174         found_str = qemu_rbd_next_tok(p, ':', &p);
 175         qemu_rbd_unescape(found_str);
 176         qdict_put_str(options, "snapshot", found_str);
 177     } else {
 178         found_str = qemu_rbd_next_tok(p, ':', &p);
 179         qemu_rbd_unescape(found_str);
 180         qdict_put_str(options, "image", found_str);
 181     }
 182     if (!p) {
 183         goto done;
 184     }
 185
 186     /* The following are essentially all key/value pairs, and we treat
 187      * 'id' and 'conf' a bit special.  Key/value pairs may be in any order. */
 188     while (p) {
 189         char *name, *value;
 190         name = qemu_rbd_next_tok(p, '=', &p);
 191         if (!p) {
 192             error_setg(errp, "conf option %s has no value", name);
 193             break;
 194         }
 195
 196         qemu_rbd_unescape(name);
 197
 198         value = qemu_rbd_next_tok(p, ':', &p);
 199         qemu_rbd_unescape(value);
 200
 201         if (!strcmp(name, "conf")) {
 202             qdict_put_str(options, "conf", value);
 203         } else if (!strcmp(name, "id")) {
 204             qdict_put_str(options, "user", value);
 205         } else {
 206             /*
 207              * We pass these internally to qemu_rbd_set_keypairs(), so
 208              * we can get away with the simpler list of [ "key1",
 209              * "value1", "key2", "value2" ] rather than a raw dict
 210              * { "key1": "value1", "key2": "value2" } where we can't
 211              * guarantee order, or even a more correct but complex
 212              * [ { "key1": "value1" }, { "key2": "value2" } ]
 213              */
 214             if (!keypairs) {
 215                 keypairs = qlist_new();
 216             }
 217             qlist_append_str(keypairs, name);
 218             qlist_append_str(keypairs, value);
 219         }
 220     }
 221
 222     if (keypairs) {
 223         qdict_put(options, "=keyvalue-pairs",
 224                   qobject_to_json(QOBJECT(keypairs)));
 225     }
 226
 227 done:
 228     g_free(buf);
 229     qobject_unref(keypairs);
 230     return;
 231 }
 232
 233
 234 static void qemu_rbd_refresh_limits(BlockDriverState *bs, Error **errp)
 235 {
 236     /* XXX Does RBD support AIO on less than 512-byte alignment? */
 237     bs->bl.request_alignment = 512;
 238 }
 239
 240
 241 static int qemu_rbd_set_auth(rados_t cluster, const char *secretid,
 242                              Error **errp)
 243 {
 244     if (secretid == 0) {
 245         return 0;
 246     }
 247
 248     gchar *secret = qcrypto_secret_lookup_as_base64(secretid,
 249                                                     errp);
 250     if (!secret) {
 251         return -1;
 252     }
 253
 254     rados_conf_set(cluster, "key", secret);
 255     g_free(secret);
 256
 257     return 0;
 258 }
 259
 260 static int qemu_rbd_set_keypairs(rados_t cluster, const char *keypairs_json,
 261                                  Error **errp)
 262 {
 263     QList *keypairs;
 264     QString *name;
 265     QString *value;
 266     const char *key;
 267     size_t remaining;
 268     int ret = 0;
 269
 270     if (!keypairs_json) {
 271         return ret;
 272     }
 273     keypairs = qobject_to(QList,
 274                           qobject_from_json(keypairs_json, &error_abort));
 275     remaining = qlist_size(keypairs) / 2;
 276     assert(remaining);
 277
 278     while (remaining--) {
 279         name = qobject_to(QString, qlist_pop(keypairs));
 280         value = qobject_to(QString, qlist_pop(keypairs));
 281         assert(name && value);
 282         key = qstring_get_str(name);
 283
 284         ret = rados_conf_set(cluster, key, qstring_get_str(value));
 285         qobject_unref(value);
 286         if (ret < 0) {
 287             error_setg_errno(errp, -ret, "invalid conf option %s", key);
 288             qobject_unref(name);
 289             ret = -EINVAL;
 290             break;
 291         }
 292         qobject_unref(name);
 293     }
 294
 295     qobject_unref(keypairs);
 296     return ret;
 297 }
 298
 299 static void qemu_rbd_memset(RADOSCB *rcb, int64_t offs)
 300 {
 301     if (LIBRBD_USE_IOVEC) {
 302         RBDAIOCB *acb = rcb->acb;
 303         iov_memset(acb->qiov->iov, acb->qiov->niov, offs, 0,
 304                    acb->qiov->size - offs);
 305     } else {
 306         memset(rcb->buf + offs, 0, rcb->size - offs);
 307     }
 308 }
 309
 310 static QemuOptsList runtime_opts = {
 311     .name = "rbd",
 312     .head = QTAILQ_HEAD_INITIALIZER(runtime_opts.head),
 313     .desc = {
 314         {
 315             .name = "pool",
 316             .type = QEMU_OPT_STRING,
 317             .help = "Rados pool name",
 318         },
 319         {
 320             .name = "image",
 321             .type = QEMU_OPT_STRING,
 322             .help = "Image name in the pool",
 323         },
 324         {
 325             .name = "conf",
 326             .type = QEMU_OPT_STRING,
 327             .help = "Rados config file location",
 328         },
 329         {
 330             .name = "snapshot",
 331             .type = QEMU_OPT_STRING,
 332             .help = "Ceph snapshot name",
 333         },
 334         {
 335             /* maps to 'id' in rados_create() */
 336             .name = "user",
 337             .type = QEMU_OPT_STRING,
 338             .help = "Rados id name",
 339         },
 340         /*
 341          * server.* extracted manually, see qemu_rbd_mon_host()
 342          */
 343         { /* end of list */ }
 344     },
 345 };
 346
 347 /* FIXME Deprecate and remove keypairs or make it available in QMP.
 348  * password_secret should eventually be configurable in opts->location. Support
 349  * for it in .bdrv_open will make it work here as well. */
 350 static int qemu_rbd_do_create(BlockdevCreateOptions *options,
 351                               const char *keypairs, const char *password_secret,
 352                               Error **errp)
 353 {
 354     BlockdevCreateOptionsRbd *opts = &options->u.rbd;
 355     rados_t cluster;
 356     rados_ioctx_t io_ctx;
 357     int obj_order = 0;
 358     int ret;
 359
 360     assert(options->driver == BLOCKDEV_DRIVER_RBD);
 361     if (opts->location->has_snapshot) {
 362         error_setg(errp, "Can't use snapshot name for image creation");
 363         return -EINVAL;
 364     }
 365
 366     if (opts->has_cluster_size) {
 367         int64_t objsize = opts->cluster_size;
 368         if ((objsize - 1) & objsize) {    /* not a power of 2? */
 369             error_setg(errp, "obj size needs to be power of 2");
 370             return -EINVAL;
 371         }
 372         if (objsize < 4096) {
 373             error_setg(errp, "obj size too small");
 374             return -EINVAL;
 375         }
 376         obj_order = ctz32(objsize);
 377     }
 378
 379     ret = qemu_rbd_connect(&cluster, &io_ctx, opts->location, false, keypairs,
 380                            password_secret, errp);
 381     if (ret < 0) {
 382         return ret;
 383     }
 384
 385     ret = rbd_create(io_ctx, opts->location->image, opts->size, &obj_order);
 386     if (ret < 0) {
 387         error_setg_errno(errp, -ret, "error rbd create");
 388         goto out;
 389     }
 390
 391     ret = 0;
 392 out:
 393     rados_ioctx_destroy(io_ctx);
 394     rados_shutdown(cluster);
 395     return ret;
 396 }
 397
 398 static int qemu_rbd_co_create(BlockdevCreateOptions *options, Error **errp)
 399 {
 400     return qemu_rbd_do_create(options, NULL, NULL, errp);
 401 }
 402
 403 static int coroutine_fn qemu_rbd_co_create_opts(const char *filename,
 404                                                 QemuOpts *opts,
 405                                                 Error **errp)
 406 {
 407     BlockdevCreateOptions *create_options;
 408     BlockdevCreateOptionsRbd *rbd_opts;
 409     BlockdevOptionsRbd *loc;
 410     Error *local_err = NULL;
 411     const char *keypairs, *password_secret;
 412     QDict *options = NULL;
 413     int ret = 0;
 414
 415     create_options = g_new0(BlockdevCreateOptions, 1);
 416     create_options->driver = BLOCKDEV_DRIVER_RBD;
 417     rbd_opts = &create_options->u.rbd;
 418
 419     rbd_opts->location = g_new0(BlockdevOptionsRbd, 1);
 420
 421     password_secret = qemu_opt_get(opts, "password-secret");
 422
 423     /* Read out options */
 424     rbd_opts->size = ROUND_UP(qemu_opt_get_size_del(opts, BLOCK_OPT_SIZE, 0),
 425                               BDRV_SECTOR_SIZE);
 426     rbd_opts->cluster_size = qemu_opt_get_size_del(opts,
 427                                                    BLOCK_OPT_CLUSTER_SIZE, 0);
 428     rbd_opts->has_cluster_size = (rbd_opts->cluster_size != 0);
 429
 430     options = qdict_new();
 431     qemu_rbd_parse_filename(filename, options, &local_err);
 432     if (local_err) {
 433         ret = -EINVAL;
 434         error_propagate(errp, local_err);
 435         goto exit;
 436     }
 437
 438     /*
 439      * Caution: while qdict_get_try_str() is fine, getting non-string
 440      * types would require more care.  When @options come from -blockdev
 441      * or blockdev_add, its members are typed according to the QAPI
 442      * schema, but when they come from -drive, they're all QString.
 443      */
 444     loc = rbd_opts->location;
 445     loc->pool     = g_strdup(qdict_get_try_str(options, "pool"));
 446     loc->conf     = g_strdup(qdict_get_try_str(options, "conf"));
 447     loc->has_conf = !!loc->conf;
 448     loc->user     = g_strdup(qdict_get_try_str(options, "user"));
 449     loc->has_user = !!loc->user;
 450     loc->image    = g_strdup(qdict_get_try_str(options, "image"));
 451     keypairs      = qdict_get_try_str(options, "=keyvalue-pairs");
 452
 453     ret = qemu_rbd_do_create(create_options, keypairs, password_secret, errp);
 454     if (ret < 0) {
 455         goto exit;
 456     }
 457
 458 exit:
 459     qobject_unref(options);
 460     qapi_free_BlockdevCreateOptions(create_options);
 461     return ret;
 462 }
 463
 464 /*
 465  * This aio completion is being called from rbd_finish_bh() and runs in qemu
 466  * BH context.
 467  */
 468 static void qemu_rbd_complete_aio(RADOSCB *rcb)
 469 {
 470     RBDAIOCB *acb = rcb->acb;
 471     int64_t r;
 472
 473     r = rcb->ret;
 474
 475     if (acb->cmd != RBD_AIO_READ) {
 476         if (r < 0) {
 477             acb->ret = r;
 478             acb->error = 1;
 479         } else if (!acb->error) {
 480             acb->ret = rcb->size;
 481         }
 482     } else {
 483         if (r < 0) {
 484             qemu_rbd_memset(rcb, 0);
 485             acb->ret = r;
 486             acb->error = 1;
 487         } else if (r < rcb->size) {
 488             qemu_rbd_memset(rcb, r);
 489             if (!acb->error) {
 490                 acb->ret = rcb->size;
 491             }
 492         } else if (!acb->error) {
 493             acb->ret = r;
 494         }
 495     }
 496
 497     g_free(rcb);
 498
 499     if (!LIBRBD_USE_IOVEC) {
 500         if (acb->cmd == RBD_AIO_READ) {
 501             qemu_iovec_from_buf(acb->qiov, 0, acb->bounce, acb->qiov->size);
 502         }
 503         qemu_vfree(acb->bounce);
 504     }
 505
 506     acb->common.cb(acb->common.opaque, (acb->ret > 0 ? 0 : acb->ret));
 507
 508     qemu_aio_unref(acb);
 509 }
 510
 511 static char *qemu_rbd_mon_host(BlockdevOptionsRbd *opts, Error **errp)
 512 {
 513     const char **vals;
 514     const char *host, *port;
 515     char *rados_str;
 516     InetSocketAddressBaseList *p;
 517     int i, cnt;
 518
 519     if (!opts->has_server) {
 520         return NULL;
 521     }
 522
 523     for (cnt = 0, p = opts->server; p; p = p->next) {
 524         cnt++;
 525     }
 526
 527     vals = g_new(const char *, cnt + 1);
 528
 529     for (i = 0, p = opts->server; p; p = p->next, i++) {
 530         host = p->value->host;
 531         port = p->value->port;
 532
 533         if (strchr(host, ':')) {
 534             vals[i] = g_strdup_printf("[%s]:%s", host, port);
 535         } else {
 536             vals[i] = g_strdup_printf("%s:%s", host, port);
 537         }
 538     }
 539     vals[i] = NULL;
 540
 541     rados_str = i ? g_strjoinv(";", (char **)vals) : NULL;
 542     g_strfreev((char **)vals);
 543     return rados_str;
 544 }
 545
 546 static int qemu_rbd_connect(rados_t *cluster, rados_ioctx_t *io_ctx,
 547                             BlockdevOptionsRbd *opts, bool cache,
 548                             const char *keypairs, const char *secretid,
 549                             Error **errp)
 550 {
 551     char *mon_host = NULL;
 552     Error *local_err = NULL;
 553     int r;
 554
 555     mon_host = qemu_rbd_mon_host(opts, &local_err);
 556     if (local_err) {
 557         error_propagate(errp, local_err);
 558         r = -EINVAL;
 559         goto failed_opts;
 560     }
 561
 562     r = rados_create(cluster, opts->user);
 563     if (r < 0) {
 564         error_setg_errno(errp, -r, "error initializing");
 565         goto failed_opts;
 566     }
 567
 568     /* try default location when conf=NULL, but ignore failure */
 569     r = rados_conf_read_file(*cluster, opts->conf);
 570     if (opts->has_conf && r < 0) {
 571         error_setg_errno(errp, -r, "error reading conf file %s", opts->conf);
 572         goto failed_shutdown;
 573     }
 574
 575     r = qemu_rbd_set_keypairs(*cluster, keypairs, errp);
 576     if (r < 0) {
 577         goto failed_shutdown;
 578     }
 579
 580     if (mon_host) {
 581         r = rados_conf_set(*cluster, "mon_host", mon_host);
 582         if (r < 0) {
 583             goto failed_shutdown;
 584         }
 585     }
 586
 587     if (qemu_rbd_set_auth(*cluster, secretid, errp) < 0) {
 588         r = -EIO;
 589         goto failed_shutdown;
 590     }
 591
 592     /*
 593      * Fallback to more conservative semantics if setting cache
 594      * options fails. Ignore errors from setting rbd_cache because the
 595      * only possible error is that the option does not exist, and
 596      * librbd defaults to no caching. If write through caching cannot
 597      * be set up, fall back to no caching.
 598      */
 599     if (cache) {
 600         rados_conf_set(*cluster, "rbd_cache", "true");
 601     } else {
 602         rados_conf_set(*cluster, "rbd_cache", "false");
 603     }
 604
 605     r = rados_connect(*cluster);
 606     if (r < 0) {
 607         error_setg_errno(errp, -r, "error connecting");
 608         goto failed_shutdown;
 609     }
 610
 611     r = rados_ioctx_create(*cluster, opts->pool, io_ctx);
 612     if (r < 0) {
 613         error_setg_errno(errp, -r, "error opening pool %s", opts->pool);
 614         goto failed_shutdown;
 615     }
 616
 617     return 0;
 618
 619 failed_shutdown:
 620     rados_shutdown(*cluster);
 621 failed_opts:
 622     g_free(mon_host);
 623     return r;
 624 }
 625
 626 static int qemu_rbd_open(BlockDriverState *bs, QDict *options, int flags,
 627                          Error **errp)
 628 {
 629     BDRVRBDState *s = bs->opaque;
 630     BlockdevOptionsRbd *opts = NULL;
 631     Visitor *v;
 632     QObject *crumpled = NULL;
 633     const QDictEntry *e;
 634     Error *local_err = NULL;
 635     const char *filename;
 636     char *keypairs, *secretid;
 637     int r;
 638
 639     /* If we are given a filename, parse the filename, with precedence given to
 640      * filename encoded options */
 641     filename = qdict_get_try_str(options, "filename");
 642     if (filename) {
 643         warn_report("'filename' option specified. "
 644                     "This is an unsupported option, and may be deprecated "
 645                     "in the future");
 646         qemu_rbd_parse_filename(filename, options, &local_err);
 647         qdict_del(options, "filename");
 648         if (local_err) {
 649             error_propagate(errp, local_err);
 650             return -EINVAL;
 651         }
 652     }
 653
 654     keypairs = g_strdup(qdict_get_try_str(options, "=keyvalue-pairs"));
 655     if (keypairs) {
 656         qdict_del(options, "=keyvalue-pairs");
 657     }
 658
 659     secretid = g_strdup(qdict_get_try_str(options, "password-secret"));
 660     if (secretid) {
 661         qdict_del(options, "password-secret");
 662     }
 663
 664     /* Convert the remaining options into a QAPI object */
 665     crumpled = qdict_crumple(options, errp);
 666     if (crumpled == NULL) {
 667         r = -EINVAL;
 668         goto out;
 669     }
 670
 671     v = qobject_input_visitor_new_keyval(crumpled);
 672     visit_type_BlockdevOptionsRbd(v, NULL, &opts, &local_err);
 673     visit_free(v);
 674     qobject_unref(crumpled);
 675
 676     if (local_err) {
 677         error_propagate(errp, local_err);
 678         r = -EINVAL;
 679         goto out;
 680     }
 681
 682     /* Remove the processed options from the QDict (the visitor processes
 683      * _all_ options in the QDict) */
 684     while ((e = qdict_first(options))) {
 685         qdict_del(options, e->key);
 686     }
 687
 688     r = qemu_rbd_connect(&s->cluster, &s->io_ctx, opts,
 689                          !(flags & BDRV_O_NOCACHE), keypairs, secretid, errp);
 690     if (r < 0) {
 691         goto out;
 692     }
 693
 694     s->snap = g_strdup(opts->snapshot);
 695     s->image_name = g_strdup(opts->image);
 696
 697     /* rbd_open is always r/w */
 698     r = rbd_open(s->io_ctx, s->image_name, &s->image, s->snap);
 699     if (r < 0) {
 700         error_setg_errno(errp, -r, "error reading header from %s",
 701                          s->image_name);
 702         goto failed_open;
 703     }
 704
 705     /* If we are using an rbd snapshot, we must be r/o, otherwise
 706      * leave as-is */
 707     if (s->snap != NULL) {
 708         if (!bdrv_is_read_only(bs)) {
 709             error_report("Opening rbd snapshots without an explicit "
 710                          "read-only=on option is deprecated. Future versions "
 711                          "will refuse to open the image instead of "
 712                          "automatically marking the image read-only.");
 713             r = bdrv_set_read_only(bs, true, &local_err);
 714             if (r < 0) {
 715                 error_propagate(errp, local_err);
 716                 goto failed_open;
 717             }
 718         }
 719     }
 720
 721     r = 0;
 722     goto out;
 723
 724 failed_open:
 725     rados_ioctx_destroy(s->io_ctx);
 726     g_free(s->snap);
 727     g_free(s->image_name);
 728     rados_shutdown(s->cluster);
 729 out:
 730     qapi_free_BlockdevOptionsRbd(opts);
 731     g_free(keypairs);
 732     g_free(secretid);
 733     return r;
 734 }
 735
 736
 737 /* Since RBD is currently always opened R/W via the API,
 738  * we just need to check if we are using a snapshot or not, in
 739  * order to determine if we will allow it to be R/W */
 740 static int qemu_rbd_reopen_prepare(BDRVReopenState *state,
 741                                    BlockReopenQueue *queue, Error **errp)
 742 {
 743     BDRVRBDState *s = state->bs->opaque;
 744     int ret = 0;
 745
 746     if (s->snap && state->flags & BDRV_O_RDWR) {
 747         error_setg(errp,
 748                    "Cannot change node '%s' to r/w when using RBD snapshot",
 749                    bdrv_get_device_or_node_name(state->bs));
 750         ret = -EINVAL;
 751     }
 752
 753     return ret;
 754 }
 755
 756 static void qemu_rbd_close(BlockDriverState *bs)
 757 {
 758     BDRVRBDState *s = bs->opaque;
 759
 760     rbd_close(s->image);
 761     rados_ioctx_destroy(s->io_ctx);
 762     g_free(s->snap);
 763     g_free(s->image_name);
 764     rados_shutdown(s->cluster);
 765 }
 766
 767 static const AIOCBInfo rbd_aiocb_info = {
 768     .aiocb_size = sizeof(RBDAIOCB),
 769 };
 770
 771 static void rbd_finish_bh(void *opaque)
 772 {
 773     RADOSCB *rcb = opaque;
 774     qemu_rbd_complete_aio(rcb);
 775 }
 776
 777 /*
 778  * This is the callback function for rbd_aio_read and _write
 779  *
 780  * Note: this function is being called from a non qemu thread so
 781  * we need to be careful about what we do here. Generally we only
 782  * schedule a BH, and do the rest of the io completion handling
 783  * from rbd_finish_bh() which runs in a qemu context.
 784  */
 785 static void rbd_finish_aiocb(rbd_completion_t c, RADOSCB *rcb)
 786 {
 787     RBDAIOCB *acb = rcb->acb;
 788
 789     rcb->ret = rbd_aio_get_return_value(c);
 790     rbd_aio_release(c);
 791
 792     aio_bh_schedule_oneshot(bdrv_get_aio_context(acb->common.bs),
 793                             rbd_finish_bh, rcb);
 794 }
 795
 796 static int rbd_aio_discard_wrapper(rbd_image_t image,
 797                                    uint64_t off,
 798                                    uint64_t len,
 799                                    rbd_completion_t comp)
 800 {
 801 #ifdef LIBRBD_SUPPORTS_DISCARD
 802     return rbd_aio_discard(image, off, len, comp);
 803 #else
 804     return -ENOTSUP;
 805 #endif
 806 }
 807
 808 static int rbd_aio_flush_wrapper(rbd_image_t image,
 809                                  rbd_completion_t comp)
 810 {
 811 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
 812     return rbd_aio_flush(image, comp);
 813 #else
 814     return -ENOTSUP;
 815 #endif
 816 }
 817
 818 static BlockAIOCB *rbd_start_aio(BlockDriverState *bs,
 819                                  int64_t off,
 820                                  QEMUIOVector *qiov,
 821                                  int64_t size,
 822                                  BlockCompletionFunc *cb,
 823                                  void *opaque,
 824                                  RBDAIOCmd cmd)
 825 {
 826     RBDAIOCB *acb;
 827     RADOSCB *rcb = NULL;
 828     rbd_completion_t c;
 829     int r;
 830
 831     BDRVRBDState *s = bs->opaque;
 832
 833     acb = qemu_aio_get(&rbd_aiocb_info, bs, cb, opaque);
 834     acb->cmd = cmd;
 835     acb->qiov = qiov;
 836     assert(!qiov || qiov->size == size);
 837
 838     rcb = g_new(RADOSCB, 1);
 839
 840     if (!LIBRBD_USE_IOVEC) {
 841         if (cmd == RBD_AIO_DISCARD || cmd == RBD_AIO_FLUSH) {
 842             acb->bounce = NULL;
 843         } else {
 844             acb->bounce = qemu_try_blockalign(bs, qiov->size);
 845             if (acb->bounce == NULL) {
 846                 goto failed;
 847             }
 848         }
 849         if (cmd == RBD_AIO_WRITE) {
 850             qemu_iovec_to_buf(acb->qiov, 0, acb->bounce, qiov->size);
 851         }
 852         rcb->buf = acb->bounce;
 853     }
 854
 855     acb->ret = 0;
 856     acb->error = 0;
 857     acb->s = s;
 858
 859     rcb->acb = acb;
 860     rcb->s = acb->s;
 861     rcb->size = size;
 862     r = rbd_aio_create_completion(rcb, (rbd_callback_t) rbd_finish_aiocb, &c);
 863     if (r < 0) {
 864         goto failed;
 865     }
 866
 867     switch (cmd) {
 868     case RBD_AIO_WRITE:
 869 #ifdef LIBRBD_SUPPORTS_IOVEC
 870             r = rbd_aio_writev(s->image, qiov->iov, qiov->niov, off, c);
 871 #else
 872             r = rbd_aio_write(s->image, off, size, rcb->buf, c);
 873 #endif
 874         break;
 875     case RBD_AIO_READ:
 876 #ifdef LIBRBD_SUPPORTS_IOVEC
 877             r = rbd_aio_readv(s->image, qiov->iov, qiov->niov, off, c);
 878 #else
 879             r = rbd_aio_read(s->image, off, size, rcb->buf, c);
 880 #endif
 881         break;
 882     case RBD_AIO_DISCARD:
 883         r = rbd_aio_discard_wrapper(s->image, off, size, c);
 884         break;
 885     case RBD_AIO_FLUSH:
 886         r = rbd_aio_flush_wrapper(s->image, c);
 887         break;
 888     default:
 889         r = -EINVAL;
 890     }
 891
 892     if (r < 0) {
 893         goto failed_completion;
 894     }
 895     return &acb->common;
 896
 897 failed_completion:
 898     rbd_aio_release(c);
 899 failed:
 900     g_free(rcb);
 901     if (!LIBRBD_USE_IOVEC) {
 902         qemu_vfree(acb->bounce);
 903     }
 904
 905     qemu_aio_unref(acb);
 906     return NULL;
 907 }
 908
 909 static BlockAIOCB *qemu_rbd_aio_preadv(BlockDriverState *bs,
 910                                        uint64_t offset, uint64_t bytes,
 911                                        QEMUIOVector *qiov, int flags,
 912                                        BlockCompletionFunc *cb,
 913                                        void *opaque)
 914 {
 915     return rbd_start_aio(bs, offset, qiov, bytes, cb, opaque,
 916                          RBD_AIO_READ);
 917 }
 918
 919 static BlockAIOCB *qemu_rbd_aio_pwritev(BlockDriverState *bs,
 920                                         uint64_t offset, uint64_t bytes,
 921                                         QEMUIOVector *qiov, int flags,
 922                                         BlockCompletionFunc *cb,
 923                                         void *opaque)
 924 {
 925     return rbd_start_aio(bs, offset, qiov, bytes, cb, opaque,
 926                          RBD_AIO_WRITE);
 927 }
 928
 929 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
 930 static BlockAIOCB *qemu_rbd_aio_flush(BlockDriverState *bs,
 931                                       BlockCompletionFunc *cb,
 932                                       void *opaque)
 933 {
 934     return rbd_start_aio(bs, 0, NULL, 0, cb, opaque, RBD_AIO_FLUSH);
 935 }
 936
 937 #else
 938
 939 static int qemu_rbd_co_flush(BlockDriverState *bs)
 940 {
 941 #if LIBRBD_VERSION_CODE >= LIBRBD_VERSION(0, 1, 1)
 942     /* rbd_flush added in 0.1.1 */
 943     BDRVRBDState *s = bs->opaque;
 944     return rbd_flush(s->image);
 945 #else
 946     return 0;
 947 #endif
 948 }
 949 #endif
 950
 951 static int qemu_rbd_getinfo(BlockDriverState *bs, BlockDriverInfo *bdi)
 952 {
 953     BDRVRBDState *s = bs->opaque;
 954     rbd_image_info_t info;
 955     int r;
 956
 957     r = rbd_stat(s->image, &info, sizeof(info));
 958     if (r < 0) {
 959         return r;
 960     }
 961
 962     bdi->cluster_size = info.obj_size;
 963     return 0;
 964 }
 965
 966 static int64_t qemu_rbd_getlength(BlockDriverState *bs)
 967 {
 968     BDRVRBDState *s = bs->opaque;
 969     rbd_image_info_t info;
 970     int r;
 971
 972     r = rbd_stat(s->image, &info, sizeof(info));
 973     if (r < 0) {
 974         return r;
 975     }
 976
 977     return info.size;
 978 }
 979
 980 static int qemu_rbd_truncate(BlockDriverState *bs, int64_t offset,
 981                              PreallocMode prealloc, Error **errp)
 982 {
 983     BDRVRBDState *s = bs->opaque;
 984     int r;
 985
 986     if (prealloc != PREALLOC_MODE_OFF) {
 987         error_setg(errp, "Unsupported preallocation mode '%s'",
 988                    PreallocMode_str(prealloc));
 989         return -ENOTSUP;
 990     }
 991
 992     r = rbd_resize(s->image, offset);
 993     if (r < 0) {
 994         error_setg_errno(errp, -r, "Failed to resize file");
 995         return r;
 996     }
 997
 998     return 0;
 999 }
1000
1001 static int qemu_rbd_snap_create(BlockDriverState *bs,
1002                                 QEMUSnapshotInfo *sn_info)
1003 {
1004     BDRVRBDState *s = bs->opaque;
1005     int r;
1006
1007     if (sn_info->name[0] == '\0') {
1008         return -EINVAL; /* we need a name for rbd snapshots */
1009     }
1010
1011     /*
1012      * rbd snapshots are using the name as the user controlled unique identifier
1013      * we can't use the rbd snapid for that purpose, as it can't be set
1014      */
1015     if (sn_info->id_str[0] != '\0' &&
1016         strcmp(sn_info->id_str, sn_info->name) != 0) {
1017         return -EINVAL;
1018     }
1019
1020     if (strlen(sn_info->name) >= sizeof(sn_info->id_str)) {
1021         return -ERANGE;
1022     }
1023
1024     r = rbd_snap_create(s->image, sn_info->name);
1025     if (r < 0) {
1026         error_report("failed to create snap: %s", strerror(-r));
1027         return r;
1028     }
1029
1030     return 0;
1031 }
1032
1033 static int qemu_rbd_snap_remove(BlockDriverState *bs,
1034                                 const char *snapshot_id,
1035                                 const char *snapshot_name,
1036                                 Error **errp)
1037 {
1038     BDRVRBDState *s = bs->opaque;
1039     int r;
1040
1041     if (!snapshot_name) {
1042         error_setg(errp, "rbd need a valid snapshot name");
1043         return -EINVAL;
1044     }
1045
1046     /* If snapshot_id is specified, it must be equal to name, see
1047        qemu_rbd_snap_list() */
1048     if (snapshot_id && strcmp(snapshot_id, snapshot_name)) {
1049         error_setg(errp,
1050                    "rbd do not support snapshot id, it should be NULL or "
1051                    "equal to snapshot name");
1052         return -EINVAL;
1053     }
1054
1055     r = rbd_snap_remove(s->image, snapshot_name);
1056     if (r < 0) {
1057         error_setg_errno(errp, -r, "Failed to remove the snapshot");
1058     }
1059     return r;
1060 }
1061
1062 static int qemu_rbd_snap_rollback(BlockDriverState *bs,
1063                                   const char *snapshot_name)
1064 {
1065     BDRVRBDState *s = bs->opaque;
1066
1067     return rbd_snap_rollback(s->image, snapshot_name);
1068 }
1069
1070 static int qemu_rbd_snap_list(BlockDriverState *bs,
1071                               QEMUSnapshotInfo **psn_tab)
1072 {
1073     BDRVRBDState *s = bs->opaque;
1074     QEMUSnapshotInfo *sn_info, *sn_tab = NULL;
1075     int i, snap_count;
1076     rbd_snap_info_t *snaps;
1077     int max_snaps = RBD_MAX_SNAPS;
1078
1079     do {
1080         snaps = g_new(rbd_snap_info_t, max_snaps);
1081         snap_count = rbd_snap_list(s->image, snaps, &max_snaps);
1082         if (snap_count <= 0) {
1083             g_free(snaps);
1084         }
1085     } while (snap_count == -ERANGE);
1086
1087     if (snap_count <= 0) {
1088         goto done;
1089     }
1090
1091     sn_tab = g_new0(QEMUSnapshotInfo, snap_count);
1092
1093     for (i = 0; i < snap_count; i++) {
1094         const char *snap_name = snaps[i].name;
1095
1096         sn_info = sn_tab + i;
1097         pstrcpy(sn_info->id_str, sizeof(sn_info->id_str), snap_name);
1098         pstrcpy(sn_info->name, sizeof(sn_info->name), snap_name);
1099
1100         sn_info->vm_state_size = snaps[i].size;
1101         sn_info->date_sec = 0;
1102         sn_info->date_nsec = 0;
1103         sn_info->vm_clock_nsec = 0;
1104     }
1105     rbd_snap_list_end(snaps);
1106     g_free(snaps);
1107
1108  done:
1109     *psn_tab = sn_tab;
1110     return snap_count;
1111 }
1112
1113 #ifdef LIBRBD_SUPPORTS_DISCARD
1114 static BlockAIOCB *qemu_rbd_aio_pdiscard(BlockDriverState *bs,
1115                                          int64_t offset,
1116                                          int bytes,
1117                                          BlockCompletionFunc *cb,
1118                                          void *opaque)
1119 {
1120     return rbd_start_aio(bs, offset, NULL, bytes, cb, opaque,
1121                          RBD_AIO_DISCARD);
1122 }
1123 #endif
1124
1125 #ifdef LIBRBD_SUPPORTS_INVALIDATE
1126 static void coroutine_fn qemu_rbd_co_invalidate_cache(BlockDriverState *bs,
1127                                                       Error **errp)
1128 {
1129     BDRVRBDState *s = bs->opaque;
1130     int r = rbd_invalidate_cache(s->image);
1131     if (r < 0) {
1132         error_setg_errno(errp, -r, "Failed to invalidate the cache");
1133     }
1134 }
1135 #endif
1136
1137 static QemuOptsList qemu_rbd_create_opts = {
1138     .name = "rbd-create-opts",
1139     .head = QTAILQ_HEAD_INITIALIZER(qemu_rbd_create_opts.head),
1140     .desc = {
1141         {
1142             .name = BLOCK_OPT_SIZE,
1143             .type = QEMU_OPT_SIZE,
1144             .help = "Virtual disk size"
1145         },
1146         {
1147             .name = BLOCK_OPT_CLUSTER_SIZE,
1148             .type = QEMU_OPT_SIZE,
1149             .help = "RBD object size"
1150         },
1151         {
1152             .name = "password-secret",
1153             .type = QEMU_OPT_STRING,
1154             .help = "ID of secret providing the password",
1155         },
1156         { /* end of list */ }
1157     }
1158 };
1159
1160 static BlockDriver bdrv_rbd = {
1161     .format_name            = "rbd",
1162     .instance_size          = sizeof(BDRVRBDState),
1163     .bdrv_parse_filename    = qemu_rbd_parse_filename,
1164     .bdrv_refresh_limits    = qemu_rbd_refresh_limits,
1165     .bdrv_file_open         = qemu_rbd_open,
1166     .bdrv_close             = qemu_rbd_close,
1167     .bdrv_reopen_prepare    = qemu_rbd_reopen_prepare,
1168     .bdrv_co_create         = qemu_rbd_co_create,
1169     .bdrv_co_create_opts    = qemu_rbd_co_create_opts,
1170     .bdrv_has_zero_init     = bdrv_has_zero_init_1,
1171     .bdrv_get_info          = qemu_rbd_getinfo,
1172     .create_opts            = &qemu_rbd_create_opts,
1173     .bdrv_getlength         = qemu_rbd_getlength,
1174     .bdrv_truncate          = qemu_rbd_truncate,
1175     .protocol_name          = "rbd",
1176
1177     .bdrv_aio_preadv        = qemu_rbd_aio_preadv,
1178     .bdrv_aio_pwritev       = qemu_rbd_aio_pwritev,
1179
1180 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
1181     .bdrv_aio_flush         = qemu_rbd_aio_flush,
1182 #else
1183     .bdrv_co_flush_to_disk  = qemu_rbd_co_flush,
1184 #endif
1185
1186 #ifdef LIBRBD_SUPPORTS_DISCARD
1187     .bdrv_aio_pdiscard      = qemu_rbd_aio_pdiscard,
1188 #endif
1189
1190     .bdrv_snapshot_create   = qemu_rbd_snap_create,
1191     .bdrv_snapshot_delete   = qemu_rbd_snap_remove,
1192     .bdrv_snapshot_list     = qemu_rbd_snap_list,
1193     .bdrv_snapshot_goto     = qemu_rbd_snap_rollback,
1194 #ifdef LIBRBD_SUPPORTS_INVALIDATE
1195     .bdrv_co_invalidate_cache = qemu_rbd_co_invalidate_cache,
1196 #endif
1197 };
1198
1199 static void bdrv_rbd_init(void)
1200 {
1201     bdrv_register(&bdrv_rbd);
1202 }
1203
1204 block_init(bdrv_rbd_init);