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