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_open(BlockDriverState *bs, QDict *options, int flags,
 659                          Error **errp)
 660 {
 661     BDRVRBDState *s = bs->opaque;
 662     BlockdevOptionsRbd *opts = NULL;
 663     Visitor *v;
 664     const QDictEntry *e;
 665     Error *local_err = NULL;
 666     char *keypairs, *secretid;
 667     int r;
 668
 669     keypairs = g_strdup(qdict_get_try_str(options, "=keyvalue-pairs"));
 670     if (keypairs) {
 671         qdict_del(options, "=keyvalue-pairs");
 672     }
 673
 674     secretid = g_strdup(qdict_get_try_str(options, "password-secret"));
 675     if (secretid) {
 676         qdict_del(options, "password-secret");
 677     }
 678
 679     /* Convert the remaining options into a QAPI object */
 680     v = qobject_input_visitor_new_flat_confused(options, errp);
 681     if (!v) {
 682         r = -EINVAL;
 683         goto out;
 684     }
 685
 686     visit_type_BlockdevOptionsRbd(v, NULL, &opts, &local_err);
 687     visit_free(v);
 688
 689     if (local_err) {
 690         error_propagate(errp, local_err);
 691         r = -EINVAL;
 692         goto out;
 693     }
 694
 695     /* Remove the processed options from the QDict (the visitor processes
 696      * _all_ options in the QDict) */
 697     while ((e = qdict_first(options))) {
 698         qdict_del(options, e->key);
 699     }
 700
 701     r = qemu_rbd_connect(&s->cluster, &s->io_ctx, opts,
 702                          !(flags & BDRV_O_NOCACHE), keypairs, secretid, errp);
 703     if (r < 0) {
 704         goto out;
 705     }
 706
 707     s->snap = g_strdup(opts->snapshot);
 708     s->image_name = g_strdup(opts->image);
 709
 710     /* rbd_open is always r/w */
 711     r = rbd_open(s->io_ctx, s->image_name, &s->image, s->snap);
 712     if (r < 0) {
 713         error_setg_errno(errp, -r, "error reading header from %s",
 714                          s->image_name);
 715         goto failed_open;
 716     }
 717
 718     /* If we are using an rbd snapshot, we must be r/o, otherwise
 719      * leave as-is */
 720     if (s->snap != NULL) {
 721         if (!bdrv_is_read_only(bs)) {
 722             error_report("Opening rbd snapshots without an explicit "
 723                          "read-only=on option is deprecated. Future versions "
 724                          "will refuse to open the image instead of "
 725                          "automatically marking the image read-only.");
 726             r = bdrv_set_read_only(bs, true, &local_err);
 727             if (r < 0) {
 728                 error_propagate(errp, local_err);
 729                 goto failed_open;
 730             }
 731         }
 732     }
 733
 734     r = 0;
 735     goto out;
 736
 737 failed_open:
 738     rados_ioctx_destroy(s->io_ctx);
 739     g_free(s->snap);
 740     g_free(s->image_name);
 741     rados_shutdown(s->cluster);
 742 out:
 743     qapi_free_BlockdevOptionsRbd(opts);
 744     g_free(keypairs);
 745     g_free(secretid);
 746     return r;
 747 }
 748
 749
 750 /* Since RBD is currently always opened R/W via the API,
 751  * we just need to check if we are using a snapshot or not, in
 752  * order to determine if we will allow it to be R/W */
 753 static int qemu_rbd_reopen_prepare(BDRVReopenState *state,
 754                                    BlockReopenQueue *queue, Error **errp)
 755 {
 756     BDRVRBDState *s = state->bs->opaque;
 757     int ret = 0;
 758
 759     if (s->snap && state->flags & BDRV_O_RDWR) {
 760         error_setg(errp,
 761                    "Cannot change node '%s' to r/w when using RBD snapshot",
 762                    bdrv_get_device_or_node_name(state->bs));
 763         ret = -EINVAL;
 764     }
 765
 766     return ret;
 767 }
 768
 769 static void qemu_rbd_close(BlockDriverState *bs)
 770 {
 771     BDRVRBDState *s = bs->opaque;
 772
 773     rbd_close(s->image);
 774     rados_ioctx_destroy(s->io_ctx);
 775     g_free(s->snap);
 776     g_free(s->image_name);
 777     rados_shutdown(s->cluster);
 778 }
 779
 780 static const AIOCBInfo rbd_aiocb_info = {
 781     .aiocb_size = sizeof(RBDAIOCB),
 782 };
 783
 784 static void rbd_finish_bh(void *opaque)
 785 {
 786     RADOSCB *rcb = opaque;
 787     qemu_rbd_complete_aio(rcb);
 788 }
 789
 790 /*
 791  * This is the callback function for rbd_aio_read and _write
 792  *
 793  * Note: this function is being called from a non qemu thread so
 794  * we need to be careful about what we do here. Generally we only
 795  * schedule a BH, and do the rest of the io completion handling
 796  * from rbd_finish_bh() which runs in a qemu context.
 797  */
 798 static void rbd_finish_aiocb(rbd_completion_t c, RADOSCB *rcb)
 799 {
 800     RBDAIOCB *acb = rcb->acb;
 801
 802     rcb->ret = rbd_aio_get_return_value(c);
 803     rbd_aio_release(c);
 804
 805     aio_bh_schedule_oneshot(bdrv_get_aio_context(acb->common.bs),
 806                             rbd_finish_bh, rcb);
 807 }
 808
 809 static int rbd_aio_discard_wrapper(rbd_image_t image,
 810                                    uint64_t off,
 811                                    uint64_t len,
 812                                    rbd_completion_t comp)
 813 {
 814 #ifdef LIBRBD_SUPPORTS_DISCARD
 815     return rbd_aio_discard(image, off, len, comp);
 816 #else
 817     return -ENOTSUP;
 818 #endif
 819 }
 820
 821 static int rbd_aio_flush_wrapper(rbd_image_t image,
 822                                  rbd_completion_t comp)
 823 {
 824 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
 825     return rbd_aio_flush(image, comp);
 826 #else
 827     return -ENOTSUP;
 828 #endif
 829 }
 830
 831 static BlockAIOCB *rbd_start_aio(BlockDriverState *bs,
 832                                  int64_t off,
 833                                  QEMUIOVector *qiov,
 834                                  int64_t size,
 835                                  BlockCompletionFunc *cb,
 836                                  void *opaque,
 837                                  RBDAIOCmd cmd)
 838 {
 839     RBDAIOCB *acb;
 840     RADOSCB *rcb = NULL;
 841     rbd_completion_t c;
 842     int r;
 843
 844     BDRVRBDState *s = bs->opaque;
 845
 846     acb = qemu_aio_get(&rbd_aiocb_info, bs, cb, opaque);
 847     acb->cmd = cmd;
 848     acb->qiov = qiov;
 849     assert(!qiov || qiov->size == size);
 850
 851     rcb = g_new(RADOSCB, 1);
 852
 853     if (!LIBRBD_USE_IOVEC) {
 854         if (cmd == RBD_AIO_DISCARD || cmd == RBD_AIO_FLUSH) {
 855             acb->bounce = NULL;
 856         } else {
 857             acb->bounce = qemu_try_blockalign(bs, qiov->size);
 858             if (acb->bounce == NULL) {
 859                 goto failed;
 860             }
 861         }
 862         if (cmd == RBD_AIO_WRITE) {
 863             qemu_iovec_to_buf(acb->qiov, 0, acb->bounce, qiov->size);
 864         }
 865         rcb->buf = acb->bounce;
 866     }
 867
 868     acb->ret = 0;
 869     acb->error = 0;
 870     acb->s = s;
 871
 872     rcb->acb = acb;
 873     rcb->s = acb->s;
 874     rcb->size = size;
 875     r = rbd_aio_create_completion(rcb, (rbd_callback_t) rbd_finish_aiocb, &c);
 876     if (r < 0) {
 877         goto failed;
 878     }
 879
 880     switch (cmd) {
 881     case RBD_AIO_WRITE:
 882 #ifdef LIBRBD_SUPPORTS_IOVEC
 883             r = rbd_aio_writev(s->image, qiov->iov, qiov->niov, off, c);
 884 #else
 885             r = rbd_aio_write(s->image, off, size, rcb->buf, c);
 886 #endif
 887         break;
 888     case RBD_AIO_READ:
 889 #ifdef LIBRBD_SUPPORTS_IOVEC
 890             r = rbd_aio_readv(s->image, qiov->iov, qiov->niov, off, c);
 891 #else
 892             r = rbd_aio_read(s->image, off, size, rcb->buf, c);
 893 #endif
 894         break;
 895     case RBD_AIO_DISCARD:
 896         r = rbd_aio_discard_wrapper(s->image, off, size, c);
 897         break;
 898     case RBD_AIO_FLUSH:
 899         r = rbd_aio_flush_wrapper(s->image, c);
 900         break;
 901     default:
 902         r = -EINVAL;
 903     }
 904
 905     if (r < 0) {
 906         goto failed_completion;
 907     }
 908     return &acb->common;
 909
 910 failed_completion:
 911     rbd_aio_release(c);
 912 failed:
 913     g_free(rcb);
 914     if (!LIBRBD_USE_IOVEC) {
 915         qemu_vfree(acb->bounce);
 916     }
 917
 918     qemu_aio_unref(acb);
 919     return NULL;
 920 }
 921
 922 static BlockAIOCB *qemu_rbd_aio_preadv(BlockDriverState *bs,
 923                                        uint64_t offset, uint64_t bytes,
 924                                        QEMUIOVector *qiov, int flags,
 925                                        BlockCompletionFunc *cb,
 926                                        void *opaque)
 927 {
 928     return rbd_start_aio(bs, offset, qiov, bytes, cb, opaque,
 929                          RBD_AIO_READ);
 930 }
 931
 932 static BlockAIOCB *qemu_rbd_aio_pwritev(BlockDriverState *bs,
 933                                         uint64_t offset, uint64_t bytes,
 934                                         QEMUIOVector *qiov, int flags,
 935                                         BlockCompletionFunc *cb,
 936                                         void *opaque)
 937 {
 938     return rbd_start_aio(bs, offset, qiov, bytes, cb, opaque,
 939                          RBD_AIO_WRITE);
 940 }
 941
 942 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
 943 static BlockAIOCB *qemu_rbd_aio_flush(BlockDriverState *bs,
 944                                       BlockCompletionFunc *cb,
 945                                       void *opaque)
 946 {
 947     return rbd_start_aio(bs, 0, NULL, 0, cb, opaque, RBD_AIO_FLUSH);
 948 }
 949
 950 #else
 951
 952 static int qemu_rbd_co_flush(BlockDriverState *bs)
 953 {
 954 #if LIBRBD_VERSION_CODE >= LIBRBD_VERSION(0, 1, 1)
 955     /* rbd_flush added in 0.1.1 */
 956     BDRVRBDState *s = bs->opaque;
 957     return rbd_flush(s->image);
 958 #else
 959     return 0;
 960 #endif
 961 }
 962 #endif
 963
 964 static int qemu_rbd_getinfo(BlockDriverState *bs, BlockDriverInfo *bdi)
 965 {
 966     BDRVRBDState *s = bs->opaque;
 967     rbd_image_info_t info;
 968     int r;
 969
 970     r = rbd_stat(s->image, &info, sizeof(info));
 971     if (r < 0) {
 972         return r;
 973     }
 974
 975     bdi->cluster_size = info.obj_size;
 976     return 0;
 977 }
 978
 979 static int64_t qemu_rbd_getlength(BlockDriverState *bs)
 980 {
 981     BDRVRBDState *s = bs->opaque;
 982     rbd_image_info_t info;
 983     int r;
 984
 985     r = rbd_stat(s->image, &info, sizeof(info));
 986     if (r < 0) {
 987         return r;
 988     }
 989
 990     return info.size;
 991 }
 992
 993 static int coroutine_fn qemu_rbd_co_truncate(BlockDriverState *bs,
 994                                              int64_t offset,
 995                                              PreallocMode prealloc,
 996                                              Error **errp)
 997 {
 998     BDRVRBDState *s = bs->opaque;
 999     int r;
1000
1001     if (prealloc != PREALLOC_MODE_OFF) {
1002         error_setg(errp, "Unsupported preallocation mode '%s'",
1003                    PreallocMode_str(prealloc));
1004         return -ENOTSUP;
1005     }
1006
1007     r = rbd_resize(s->image, offset);
1008     if (r < 0) {
1009         error_setg_errno(errp, -r, "Failed to resize file");
1010         return r;
1011     }
1012
1013     return 0;
1014 }
1015
1016 static int qemu_rbd_snap_create(BlockDriverState *bs,
1017                                 QEMUSnapshotInfo *sn_info)
1018 {
1019     BDRVRBDState *s = bs->opaque;
1020     int r;
1021
1022     if (sn_info->name[0] == '\0') {
1023         return -EINVAL; /* we need a name for rbd snapshots */
1024     }
1025
1026     /*
1027      * rbd snapshots are using the name as the user controlled unique identifier
1028      * we can't use the rbd snapid for that purpose, as it can't be set
1029      */
1030     if (sn_info->id_str[0] != '\0' &&
1031         strcmp(sn_info->id_str, sn_info->name) != 0) {
1032         return -EINVAL;
1033     }
1034
1035     if (strlen(sn_info->name) >= sizeof(sn_info->id_str)) {
1036         return -ERANGE;
1037     }
1038
1039     r = rbd_snap_create(s->image, sn_info->name);
1040     if (r < 0) {
1041         error_report("failed to create snap: %s", strerror(-r));
1042         return r;
1043     }
1044
1045     return 0;
1046 }
1047
1048 static int qemu_rbd_snap_remove(BlockDriverState *bs,
1049                                 const char *snapshot_id,
1050                                 const char *snapshot_name,
1051                                 Error **errp)
1052 {
1053     BDRVRBDState *s = bs->opaque;
1054     int r;
1055
1056     if (!snapshot_name) {
1057         error_setg(errp, "rbd need a valid snapshot name");
1058         return -EINVAL;
1059     }
1060
1061     /* If snapshot_id is specified, it must be equal to name, see
1062        qemu_rbd_snap_list() */
1063     if (snapshot_id && strcmp(snapshot_id, snapshot_name)) {
1064         error_setg(errp,
1065                    "rbd do not support snapshot id, it should be NULL or "
1066                    "equal to snapshot name");
1067         return -EINVAL;
1068     }
1069
1070     r = rbd_snap_remove(s->image, snapshot_name);
1071     if (r < 0) {
1072         error_setg_errno(errp, -r, "Failed to remove the snapshot");
1073     }
1074     return r;
1075 }
1076
1077 static int qemu_rbd_snap_rollback(BlockDriverState *bs,
1078                                   const char *snapshot_name)
1079 {
1080     BDRVRBDState *s = bs->opaque;
1081
1082     return rbd_snap_rollback(s->image, snapshot_name);
1083 }
1084
1085 static int qemu_rbd_snap_list(BlockDriverState *bs,
1086                               QEMUSnapshotInfo **psn_tab)
1087 {
1088     BDRVRBDState *s = bs->opaque;
1089     QEMUSnapshotInfo *sn_info, *sn_tab = NULL;
1090     int i, snap_count;
1091     rbd_snap_info_t *snaps;
1092     int max_snaps = RBD_MAX_SNAPS;
1093
1094     do {
1095         snaps = g_new(rbd_snap_info_t, max_snaps);
1096         snap_count = rbd_snap_list(s->image, snaps, &max_snaps);
1097         if (snap_count <= 0) {
1098             g_free(snaps);
1099         }
1100     } while (snap_count == -ERANGE);
1101
1102     if (snap_count <= 0) {
1103         goto done;
1104     }
1105
1106     sn_tab = g_new0(QEMUSnapshotInfo, snap_count);
1107
1108     for (i = 0; i < snap_count; i++) {
1109         const char *snap_name = snaps[i].name;
1110
1111         sn_info = sn_tab + i;
1112         pstrcpy(sn_info->id_str, sizeof(sn_info->id_str), snap_name);
1113         pstrcpy(sn_info->name, sizeof(sn_info->name), snap_name);
1114
1115         sn_info->vm_state_size = snaps[i].size;
1116         sn_info->date_sec = 0;
1117         sn_info->date_nsec = 0;
1118         sn_info->vm_clock_nsec = 0;
1119     }
1120     rbd_snap_list_end(snaps);
1121     g_free(snaps);
1122
1123  done:
1124     *psn_tab = sn_tab;
1125     return snap_count;
1126 }
1127
1128 #ifdef LIBRBD_SUPPORTS_DISCARD
1129 static BlockAIOCB *qemu_rbd_aio_pdiscard(BlockDriverState *bs,
1130                                          int64_t offset,
1131                                          int bytes,
1132                                          BlockCompletionFunc *cb,
1133                                          void *opaque)
1134 {
1135     return rbd_start_aio(bs, offset, NULL, bytes, cb, opaque,
1136                          RBD_AIO_DISCARD);
1137 }
1138 #endif
1139
1140 #ifdef LIBRBD_SUPPORTS_INVALIDATE
1141 static void coroutine_fn qemu_rbd_co_invalidate_cache(BlockDriverState *bs,
1142                                                       Error **errp)
1143 {
1144     BDRVRBDState *s = bs->opaque;
1145     int r = rbd_invalidate_cache(s->image);
1146     if (r < 0) {
1147         error_setg_errno(errp, -r, "Failed to invalidate the cache");
1148     }
1149 }
1150 #endif
1151
1152 static QemuOptsList qemu_rbd_create_opts = {
1153     .name = "rbd-create-opts",
1154     .head = QTAILQ_HEAD_INITIALIZER(qemu_rbd_create_opts.head),
1155     .desc = {
1156         {
1157             .name = BLOCK_OPT_SIZE,
1158             .type = QEMU_OPT_SIZE,
1159             .help = "Virtual disk size"
1160         },
1161         {
1162             .name = BLOCK_OPT_CLUSTER_SIZE,
1163             .type = QEMU_OPT_SIZE,
1164             .help = "RBD object size"
1165         },
1166         {
1167             .name = "password-secret",
1168             .type = QEMU_OPT_STRING,
1169             .help = "ID of secret providing the password",
1170         },
1171         { /* end of list */ }
1172     }
1173 };
1174
1175 static BlockDriver bdrv_rbd = {
1176     .format_name            = "rbd",
1177     .instance_size          = sizeof(BDRVRBDState),
1178     .bdrv_parse_filename    = qemu_rbd_parse_filename,
1179     .bdrv_refresh_limits    = qemu_rbd_refresh_limits,
1180     .bdrv_file_open         = qemu_rbd_open,
1181     .bdrv_close             = qemu_rbd_close,
1182     .bdrv_reopen_prepare    = qemu_rbd_reopen_prepare,
1183     .bdrv_co_create         = qemu_rbd_co_create,
1184     .bdrv_co_create_opts    = qemu_rbd_co_create_opts,
1185     .bdrv_has_zero_init     = bdrv_has_zero_init_1,
1186     .bdrv_get_info          = qemu_rbd_getinfo,
1187     .create_opts            = &qemu_rbd_create_opts,
1188     .bdrv_getlength         = qemu_rbd_getlength,
1189     .bdrv_co_truncate       = qemu_rbd_co_truncate,
1190     .protocol_name          = "rbd",
1191
1192     .bdrv_aio_preadv        = qemu_rbd_aio_preadv,
1193     .bdrv_aio_pwritev       = qemu_rbd_aio_pwritev,
1194
1195 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
1196     .bdrv_aio_flush         = qemu_rbd_aio_flush,
1197 #else
1198     .bdrv_co_flush_to_disk  = qemu_rbd_co_flush,
1199 #endif
1200
1201 #ifdef LIBRBD_SUPPORTS_DISCARD
1202     .bdrv_aio_pdiscard      = qemu_rbd_aio_pdiscard,
1203 #endif
1204
1205     .bdrv_snapshot_create   = qemu_rbd_snap_create,
1206     .bdrv_snapshot_delete   = qemu_rbd_snap_remove,
1207     .bdrv_snapshot_list     = qemu_rbd_snap_list,
1208     .bdrv_snapshot_goto     = qemu_rbd_snap_rollback,
1209 #ifdef LIBRBD_SUPPORTS_INVALIDATE
1210     .bdrv_co_invalidate_cache = qemu_rbd_co_invalidate_cache,
1211 #endif
1212 };
1213
1214 static void bdrv_rbd_init(void)
1215 {
1216     bdrv_register(&bdrv_rbd);
1217 }
1218
1219 block_init(bdrv_rbd_init);