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