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         if (!bdrv_is_read_only(bs)) {
 784             error_report("Opening rbd snapshots without an explicit "
 785                          "read-only=on option is deprecated. Future versions "
 786                          "will refuse to open the image instead of "
 787                          "automatically marking the image read-only.");
 788             r = bdrv_set_read_only(bs, true, &local_err);
 789             if (r < 0) {
 790                 error_propagate(errp, local_err);
 791                 goto failed_open;
 792             }
 793         }
 794     }
 795
 796     r = 0;
 797     goto out;
 798
 799 failed_open:
 800     rados_ioctx_destroy(s->io_ctx);
 801     g_free(s->snap);
 802     g_free(s->image_name);
 803     rados_shutdown(s->cluster);
 804 out:
 805     qapi_free_BlockdevOptionsRbd(opts);
 806     g_free(keypairs);
 807     g_free(secretid);
 808     return r;
 809 }
 810
 811
 812 /* Since RBD is currently always opened R/W via the API,
 813  * we just need to check if we are using a snapshot or not, in
 814  * order to determine if we will allow it to be R/W */
 815 static int qemu_rbd_reopen_prepare(BDRVReopenState *state,
 816                                    BlockReopenQueue *queue, Error **errp)
 817 {
 818     BDRVRBDState *s = state->bs->opaque;
 819     int ret = 0;
 820
 821     if (s->snap && state->flags & BDRV_O_RDWR) {
 822         error_setg(errp,
 823                    "Cannot change node '%s' to r/w when using RBD snapshot",
 824                    bdrv_get_device_or_node_name(state->bs));
 825         ret = -EINVAL;
 826     }
 827
 828     return ret;
 829 }
 830
 831 static void qemu_rbd_close(BlockDriverState *bs)
 832 {
 833     BDRVRBDState *s = bs->opaque;
 834
 835     rbd_close(s->image);
 836     rados_ioctx_destroy(s->io_ctx);
 837     g_free(s->snap);
 838     g_free(s->image_name);
 839     rados_shutdown(s->cluster);
 840 }
 841
 842 static const AIOCBInfo rbd_aiocb_info = {
 843     .aiocb_size = sizeof(RBDAIOCB),
 844 };
 845
 846 static void rbd_finish_bh(void *opaque)
 847 {
 848     RADOSCB *rcb = opaque;
 849     qemu_rbd_complete_aio(rcb);
 850 }
 851
 852 /*
 853  * This is the callback function for rbd_aio_read and _write
 854  *
 855  * Note: this function is being called from a non qemu thread so
 856  * we need to be careful about what we do here. Generally we only
 857  * schedule a BH, and do the rest of the io completion handling
 858  * from rbd_finish_bh() which runs in a qemu context.
 859  */
 860 static void rbd_finish_aiocb(rbd_completion_t c, RADOSCB *rcb)
 861 {
 862     RBDAIOCB *acb = rcb->acb;
 863
 864     rcb->ret = rbd_aio_get_return_value(c);
 865     rbd_aio_release(c);
 866
 867     aio_bh_schedule_oneshot(bdrv_get_aio_context(acb->common.bs),
 868                             rbd_finish_bh, rcb);
 869 }
 870
 871 static int rbd_aio_discard_wrapper(rbd_image_t image,
 872                                    uint64_t off,
 873                                    uint64_t len,
 874                                    rbd_completion_t comp)
 875 {
 876 #ifdef LIBRBD_SUPPORTS_DISCARD
 877     return rbd_aio_discard(image, off, len, comp);
 878 #else
 879     return -ENOTSUP;
 880 #endif
 881 }
 882
 883 static int rbd_aio_flush_wrapper(rbd_image_t image,
 884                                  rbd_completion_t comp)
 885 {
 886 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
 887     return rbd_aio_flush(image, comp);
 888 #else
 889     return -ENOTSUP;
 890 #endif
 891 }
 892
 893 static BlockAIOCB *rbd_start_aio(BlockDriverState *bs,
 894                                  int64_t off,
 895                                  QEMUIOVector *qiov,
 896                                  int64_t size,
 897                                  BlockCompletionFunc *cb,
 898                                  void *opaque,
 899                                  RBDAIOCmd cmd)
 900 {
 901     RBDAIOCB *acb;
 902     RADOSCB *rcb = NULL;
 903     rbd_completion_t c;
 904     int r;
 905
 906     BDRVRBDState *s = bs->opaque;
 907
 908     acb = qemu_aio_get(&rbd_aiocb_info, bs, cb, opaque);
 909     acb->cmd = cmd;
 910     acb->qiov = qiov;
 911     assert(!qiov || qiov->size == size);
 912
 913     rcb = g_new(RADOSCB, 1);
 914
 915     if (!LIBRBD_USE_IOVEC) {
 916         if (cmd == RBD_AIO_DISCARD || cmd == RBD_AIO_FLUSH) {
 917             acb->bounce = NULL;
 918         } else {
 919             acb->bounce = qemu_try_blockalign(bs, qiov->size);
 920             if (acb->bounce == NULL) {
 921                 goto failed;
 922             }
 923         }
 924         if (cmd == RBD_AIO_WRITE) {
 925             qemu_iovec_to_buf(acb->qiov, 0, acb->bounce, qiov->size);
 926         }
 927         rcb->buf = acb->bounce;
 928     }
 929
 930     acb->ret = 0;
 931     acb->error = 0;
 932     acb->s = s;
 933
 934     rcb->acb = acb;
 935     rcb->s = acb->s;
 936     rcb->size = size;
 937     r = rbd_aio_create_completion(rcb, (rbd_callback_t) rbd_finish_aiocb, &c);
 938     if (r < 0) {
 939         goto failed;
 940     }
 941
 942     switch (cmd) {
 943     case RBD_AIO_WRITE:
 944 #ifdef LIBRBD_SUPPORTS_IOVEC
 945             r = rbd_aio_writev(s->image, qiov->iov, qiov->niov, off, c);
 946 #else
 947             r = rbd_aio_write(s->image, off, size, rcb->buf, c);
 948 #endif
 949         break;
 950     case RBD_AIO_READ:
 951 #ifdef LIBRBD_SUPPORTS_IOVEC
 952             r = rbd_aio_readv(s->image, qiov->iov, qiov->niov, off, c);
 953 #else
 954             r = rbd_aio_read(s->image, off, size, rcb->buf, c);
 955 #endif
 956         break;
 957     case RBD_AIO_DISCARD:
 958         r = rbd_aio_discard_wrapper(s->image, off, size, c);
 959         break;
 960     case RBD_AIO_FLUSH:
 961         r = rbd_aio_flush_wrapper(s->image, c);
 962         break;
 963     default:
 964         r = -EINVAL;
 965     }
 966
 967     if (r < 0) {
 968         goto failed_completion;
 969     }
 970     return &acb->common;
 971
 972 failed_completion:
 973     rbd_aio_release(c);
 974 failed:
 975     g_free(rcb);
 976     if (!LIBRBD_USE_IOVEC) {
 977         qemu_vfree(acb->bounce);
 978     }
 979
 980     qemu_aio_unref(acb);
 981     return NULL;
 982 }
 983
 984 static BlockAIOCB *qemu_rbd_aio_preadv(BlockDriverState *bs,
 985                                        uint64_t offset, uint64_t bytes,
 986                                        QEMUIOVector *qiov, int flags,
 987                                        BlockCompletionFunc *cb,
 988                                        void *opaque)
 989 {
 990     return rbd_start_aio(bs, offset, qiov, bytes, cb, opaque,
 991                          RBD_AIO_READ);
 992 }
 993
 994 static BlockAIOCB *qemu_rbd_aio_pwritev(BlockDriverState *bs,
 995                                         uint64_t offset, uint64_t bytes,
 996                                         QEMUIOVector *qiov, int flags,
 997                                         BlockCompletionFunc *cb,
 998                                         void *opaque)
 999 {
1000     return rbd_start_aio(bs, offset, qiov, bytes, cb, opaque,
1001                          RBD_AIO_WRITE);
1002 }
1003
1004 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
1005 static BlockAIOCB *qemu_rbd_aio_flush(BlockDriverState *bs,
1006                                       BlockCompletionFunc *cb,
1007                                       void *opaque)
1008 {
1009     return rbd_start_aio(bs, 0, NULL, 0, cb, opaque, RBD_AIO_FLUSH);
1010 }
1011
1012 #else
1013
1014 static int qemu_rbd_co_flush(BlockDriverState *bs)
1015 {
1016 #if LIBRBD_VERSION_CODE >= LIBRBD_VERSION(0, 1, 1)
1017     /* rbd_flush added in 0.1.1 */
1018     BDRVRBDState *s = bs->opaque;
1019     return rbd_flush(s->image);
1020 #else
1021     return 0;
1022 #endif
1023 }
1024 #endif
1025
1026 static int qemu_rbd_getinfo(BlockDriverState *bs, BlockDriverInfo *bdi)
1027 {
1028     BDRVRBDState *s = bs->opaque;
1029     rbd_image_info_t info;
1030     int r;
1031
1032     r = rbd_stat(s->image, &info, sizeof(info));
1033     if (r < 0) {
1034         return r;
1035     }
1036
1037     bdi->cluster_size = info.obj_size;
1038     return 0;
1039 }
1040
1041 static int64_t qemu_rbd_getlength(BlockDriverState *bs)
1042 {
1043     BDRVRBDState *s = bs->opaque;
1044     rbd_image_info_t info;
1045     int r;
1046
1047     r = rbd_stat(s->image, &info, sizeof(info));
1048     if (r < 0) {
1049         return r;
1050     }
1051
1052     return info.size;
1053 }
1054
1055 static int coroutine_fn qemu_rbd_co_truncate(BlockDriverState *bs,
1056                                              int64_t offset,
1057                                              PreallocMode prealloc,
1058                                              Error **errp)
1059 {
1060     BDRVRBDState *s = bs->opaque;
1061     int r;
1062
1063     if (prealloc != PREALLOC_MODE_OFF) {
1064         error_setg(errp, "Unsupported preallocation mode '%s'",
1065                    PreallocMode_str(prealloc));
1066         return -ENOTSUP;
1067     }
1068
1069     r = rbd_resize(s->image, offset);
1070     if (r < 0) {
1071         error_setg_errno(errp, -r, "Failed to resize file");
1072         return r;
1073     }
1074
1075     return 0;
1076 }
1077
1078 static int qemu_rbd_snap_create(BlockDriverState *bs,
1079                                 QEMUSnapshotInfo *sn_info)
1080 {
1081     BDRVRBDState *s = bs->opaque;
1082     int r;
1083
1084     if (sn_info->name[0] == '\0') {
1085         return -EINVAL; /* we need a name for rbd snapshots */
1086     }
1087
1088     /*
1089      * rbd snapshots are using the name as the user controlled unique identifier
1090      * we can't use the rbd snapid for that purpose, as it can't be set
1091      */
1092     if (sn_info->id_str[0] != '\0' &&
1093         strcmp(sn_info->id_str, sn_info->name) != 0) {
1094         return -EINVAL;
1095     }
1096
1097     if (strlen(sn_info->name) >= sizeof(sn_info->id_str)) {
1098         return -ERANGE;
1099     }
1100
1101     r = rbd_snap_create(s->image, sn_info->name);
1102     if (r < 0) {
1103         error_report("failed to create snap: %s", strerror(-r));
1104         return r;
1105     }
1106
1107     return 0;
1108 }
1109
1110 static int qemu_rbd_snap_remove(BlockDriverState *bs,
1111                                 const char *snapshot_id,
1112                                 const char *snapshot_name,
1113                                 Error **errp)
1114 {
1115     BDRVRBDState *s = bs->opaque;
1116     int r;
1117
1118     if (!snapshot_name) {
1119         error_setg(errp, "rbd need a valid snapshot name");
1120         return -EINVAL;
1121     }
1122
1123     /* If snapshot_id is specified, it must be equal to name, see
1124        qemu_rbd_snap_list() */
1125     if (snapshot_id && strcmp(snapshot_id, snapshot_name)) {
1126         error_setg(errp,
1127                    "rbd do not support snapshot id, it should be NULL or "
1128                    "equal to snapshot name");
1129         return -EINVAL;
1130     }
1131
1132     r = rbd_snap_remove(s->image, snapshot_name);
1133     if (r < 0) {
1134         error_setg_errno(errp, -r, "Failed to remove the snapshot");
1135     }
1136     return r;
1137 }
1138
1139 static int qemu_rbd_snap_rollback(BlockDriverState *bs,
1140                                   const char *snapshot_name)
1141 {
1142     BDRVRBDState *s = bs->opaque;
1143
1144     return rbd_snap_rollback(s->image, snapshot_name);
1145 }
1146
1147 static int qemu_rbd_snap_list(BlockDriverState *bs,
1148                               QEMUSnapshotInfo **psn_tab)
1149 {
1150     BDRVRBDState *s = bs->opaque;
1151     QEMUSnapshotInfo *sn_info, *sn_tab = NULL;
1152     int i, snap_count;
1153     rbd_snap_info_t *snaps;
1154     int max_snaps = RBD_MAX_SNAPS;
1155
1156     do {
1157         snaps = g_new(rbd_snap_info_t, max_snaps);
1158         snap_count = rbd_snap_list(s->image, snaps, &max_snaps);
1159         if (snap_count <= 0) {
1160             g_free(snaps);
1161         }
1162     } while (snap_count == -ERANGE);
1163
1164     if (snap_count <= 0) {
1165         goto done;
1166     }
1167
1168     sn_tab = g_new0(QEMUSnapshotInfo, snap_count);
1169
1170     for (i = 0; i < snap_count; i++) {
1171         const char *snap_name = snaps[i].name;
1172
1173         sn_info = sn_tab + i;
1174         pstrcpy(sn_info->id_str, sizeof(sn_info->id_str), snap_name);
1175         pstrcpy(sn_info->name, sizeof(sn_info->name), snap_name);
1176
1177         sn_info->vm_state_size = snaps[i].size;
1178         sn_info->date_sec = 0;
1179         sn_info->date_nsec = 0;
1180         sn_info->vm_clock_nsec = 0;
1181     }
1182     rbd_snap_list_end(snaps);
1183     g_free(snaps);
1184
1185  done:
1186     *psn_tab = sn_tab;
1187     return snap_count;
1188 }
1189
1190 #ifdef LIBRBD_SUPPORTS_DISCARD
1191 static BlockAIOCB *qemu_rbd_aio_pdiscard(BlockDriverState *bs,
1192                                          int64_t offset,
1193                                          int bytes,
1194                                          BlockCompletionFunc *cb,
1195                                          void *opaque)
1196 {
1197     return rbd_start_aio(bs, offset, NULL, bytes, cb, opaque,
1198                          RBD_AIO_DISCARD);
1199 }
1200 #endif
1201
1202 #ifdef LIBRBD_SUPPORTS_INVALIDATE
1203 static void coroutine_fn qemu_rbd_co_invalidate_cache(BlockDriverState *bs,
1204                                                       Error **errp)
1205 {
1206     BDRVRBDState *s = bs->opaque;
1207     int r = rbd_invalidate_cache(s->image);
1208     if (r < 0) {
1209         error_setg_errno(errp, -r, "Failed to invalidate the cache");
1210     }
1211 }
1212 #endif
1213
1214 static QemuOptsList qemu_rbd_create_opts = {
1215     .name = "rbd-create-opts",
1216     .head = QTAILQ_HEAD_INITIALIZER(qemu_rbd_create_opts.head),
1217     .desc = {
1218         {
1219             .name = BLOCK_OPT_SIZE,
1220             .type = QEMU_OPT_SIZE,
1221             .help = "Virtual disk size"
1222         },
1223         {
1224             .name = BLOCK_OPT_CLUSTER_SIZE,
1225             .type = QEMU_OPT_SIZE,
1226             .help = "RBD object size"
1227         },
1228         {
1229             .name = "password-secret",
1230             .type = QEMU_OPT_STRING,
1231             .help = "ID of secret providing the password",
1232         },
1233         { /* end of list */ }
1234     }
1235 };
1236
1237 static BlockDriver bdrv_rbd = {
1238     .format_name            = "rbd",
1239     .instance_size          = sizeof(BDRVRBDState),
1240     .bdrv_parse_filename    = qemu_rbd_parse_filename,
1241     .bdrv_refresh_limits    = qemu_rbd_refresh_limits,
1242     .bdrv_file_open         = qemu_rbd_open,
1243     .bdrv_close             = qemu_rbd_close,
1244     .bdrv_reopen_prepare    = qemu_rbd_reopen_prepare,
1245     .bdrv_co_create         = qemu_rbd_co_create,
1246     .bdrv_co_create_opts    = qemu_rbd_co_create_opts,
1247     .bdrv_has_zero_init     = bdrv_has_zero_init_1,
1248     .bdrv_get_info          = qemu_rbd_getinfo,
1249     .create_opts            = &qemu_rbd_create_opts,
1250     .bdrv_getlength         = qemu_rbd_getlength,
1251     .bdrv_co_truncate       = qemu_rbd_co_truncate,
1252     .protocol_name          = "rbd",
1253
1254     .bdrv_aio_preadv        = qemu_rbd_aio_preadv,
1255     .bdrv_aio_pwritev       = qemu_rbd_aio_pwritev,
1256
1257 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
1258     .bdrv_aio_flush         = qemu_rbd_aio_flush,
1259 #else
1260     .bdrv_co_flush_to_disk  = qemu_rbd_co_flush,
1261 #endif
1262
1263 #ifdef LIBRBD_SUPPORTS_DISCARD
1264     .bdrv_aio_pdiscard      = qemu_rbd_aio_pdiscard,
1265 #endif
1266
1267     .bdrv_snapshot_create   = qemu_rbd_snap_create,
1268     .bdrv_snapshot_delete   = qemu_rbd_snap_remove,
1269     .bdrv_snapshot_list     = qemu_rbd_snap_list,
1270     .bdrv_snapshot_goto     = qemu_rbd_snap_rollback,
1271 #ifdef LIBRBD_SUPPORTS_INVALIDATE
1272     .bdrv_co_invalidate_cache = qemu_rbd_co_invalidate_cache,
1273 #endif
1274 };
1275
1276 static void bdrv_rbd_init(void)
1277 {
1278     bdrv_register(&bdrv_rbd);
1279 }
1280
1281 block_init(bdrv_rbd_init);