block/rbd.c

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