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