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 "qapi/error.h"
  17 #include "qemu/error-report.h"
  18 #include "block/block_int.h"
  19 #include "crypto/secret.h"
  20 #include "qemu/cutils.h"
  21 #include "qapi/qmp/qstring.h"
  22
  23 #include <rbd/librbd.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 *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, *keypairs;
 140     char *found_str;
 141     size_t max_keypair_size;
 142
 143     if (!strstart(filename, "rbd:", &start)) {
 144         error_setg(errp, "File name must start with 'rbd:'");
 145         return;
 146     }
 147
 148     max_keypair_size = strlen(start) + 1;
 149     buf = g_strdup(start);
 150     keypairs = g_malloc0(max_keypair_size);
 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(options, "pool", qstring_from_str(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(options, "image", qstring_from_str(found_str));
 165
 166         found_str = qemu_rbd_next_tok(p, ':', &p);
 167         qemu_rbd_unescape(found_str);
 168         qdict_put(options, "snapshot", qstring_from_str(found_str));
 169     } else {
 170         found_str = qemu_rbd_next_tok(p, ':', &p);
 171         qemu_rbd_unescape(found_str);
 172         qdict_put(options, "image", qstring_from_str(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(options, "conf", qstring_from_str(value));
 195         } else if (!strcmp(name, "id")) {
 196             qdict_put(options, "user" , qstring_from_str(value));
 197         } else {
 198             /* FIXME: This is pretty ugly, and not the right way to do this.
 199              *        These should be contained in a structure, and then
 200              *        passed explicitly as individual key/value pairs to
 201              *        rados.  Consider this legacy code that needs to be
 202              *        updated. */
 203             char *tmp = g_malloc0(max_keypair_size);
 204             /* only use a delimiter if it is not the first keypair found */
 205             /* These are sets of unknown key/value pairs we'll pass along
 206              * to ceph */
 207             if (keypairs[0]) {
 208                 snprintf(tmp, max_keypair_size, ":%s=%s", name, value);
 209                 pstrcat(keypairs, max_keypair_size, tmp);
 210             } else {
 211                 snprintf(keypairs, max_keypair_size, "%s=%s", name, value);
 212             }
 213             g_free(tmp);
 214         }
 215     }
 216
 217     if (keypairs[0]) {
 218         qdict_put(options, "=keyvalue-pairs", qstring_from_str(keypairs));
 219     }
 220
 221
 222 done:
 223     g_free(buf);
 224     g_free(keypairs);
 225     return;
 226 }
 227
 228
 229 static int qemu_rbd_set_auth(rados_t cluster, const char *secretid,
 230                              Error **errp)
 231 {
 232     if (secretid == 0) {
 233         return 0;
 234     }
 235
 236     gchar *secret = qcrypto_secret_lookup_as_base64(secretid,
 237                                                     errp);
 238     if (!secret) {
 239         return -1;
 240     }
 241
 242     rados_conf_set(cluster, "key", secret);
 243     g_free(secret);
 244
 245     return 0;
 246 }
 247
 248 static int qemu_rbd_set_keypairs(rados_t cluster, const char *keypairs,
 249                                  Error **errp)
 250 {
 251     char *p, *buf;
 252     char *name;
 253     char *value;
 254     int ret = 0;
 255
 256     buf = g_strdup(keypairs);
 257     p = buf;
 258
 259     while (p) {
 260         name = qemu_rbd_next_tok(p, '=', &p);
 261         if (!p) {
 262             error_setg(errp, "conf option %s has no value", name);
 263             ret = -EINVAL;
 264             break;
 265         }
 266
 267         value = qemu_rbd_next_tok(p, ':', &p);
 268
 269         ret = rados_conf_set(cluster, name, value);
 270         if (ret < 0) {
 271             error_setg_errno(errp, -ret, "invalid conf option %s", name);
 272             ret = -EINVAL;
 273             break;
 274         }
 275     }
 276
 277     g_free(buf);
 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.* and auth-supported.* extracted manually, see
 324          * qemu_rbd_array_opts()
 325          */
 326         {
 327             .name = "password-secret",
 328             .type = QEMU_OPT_STRING,
 329             .help = "ID of secret providing the password",
 330         },
 331
 332         /*
 333          * Keys for qemu_rbd_parse_filename(), not in the QAPI schema
 334          */
 335         {
 336             /*
 337              * HACK: name starts with '=' so that qemu_opts_parse()
 338              * can't set it
 339              */
 340             .name = "=keyvalue-pairs",
 341             .type = QEMU_OPT_STRING,
 342             .help = "Legacy rados key/value option parameters",
 343         },
 344
 345         /*
 346          * The remainder aren't option keys, but option sub-sub-keys,
 347          * so that qemu_rbd_array_opts() can abuse runtime_opts for
 348          * its own purposes
 349          * TODO clean this up
 350          */
 351         {
 352             .name = "host",
 353             .type = QEMU_OPT_STRING,
 354         },
 355         {
 356             .name = "port",
 357             .type = QEMU_OPT_STRING,
 358         },
 359         {
 360             .name = "auth",
 361             .type = QEMU_OPT_STRING,
 362             .help = "Supported authentication method, either cephx or none",
 363         },
 364         { /* end of list */ }
 365     },
 366 };
 367
 368 static int qemu_rbd_create(const char *filename, QemuOpts *opts, Error **errp)
 369 {
 370     Error *local_err = NULL;
 371     int64_t bytes = 0;
 372     int64_t objsize;
 373     int obj_order = 0;
 374     const char *pool, *name, *conf, *clientname, *keypairs;
 375     const char *secretid;
 376     rados_t cluster;
 377     rados_ioctx_t io_ctx;
 378     QDict *options = NULL;
 379     QemuOpts *rbd_opts = NULL;
 380     int ret = 0;
 381
 382     secretid = qemu_opt_get(opts, "password-secret");
 383
 384     /* Read out options */
 385     bytes = ROUND_UP(qemu_opt_get_size_del(opts, BLOCK_OPT_SIZE, 0),
 386                      BDRV_SECTOR_SIZE);
 387     objsize = qemu_opt_get_size_del(opts, BLOCK_OPT_CLUSTER_SIZE, 0);
 388     if (objsize) {
 389         if ((objsize - 1) & objsize) {    /* not a power of 2? */
 390             error_setg(errp, "obj size needs to be power of 2");
 391             ret = -EINVAL;
 392             goto exit;
 393         }
 394         if (objsize < 4096) {
 395             error_setg(errp, "obj size too small");
 396             ret = -EINVAL;
 397             goto exit;
 398         }
 399         obj_order = ctz32(objsize);
 400     }
 401
 402     options = qdict_new();
 403     qemu_rbd_parse_filename(filename, options, &local_err);
 404     if (local_err) {
 405         ret = -EINVAL;
 406         error_propagate(errp, local_err);
 407         goto exit;
 408     }
 409
 410     rbd_opts = qemu_opts_create(&runtime_opts, NULL, 0, &error_abort);
 411     qemu_opts_absorb_qdict(rbd_opts, options, &local_err);
 412     if (local_err) {
 413         error_propagate(errp, local_err);
 414         ret = -EINVAL;
 415         goto exit;
 416     }
 417
 418     pool       = qemu_opt_get(rbd_opts, "pool");
 419     conf       = qemu_opt_get(rbd_opts, "conf");
 420     clientname = qemu_opt_get(rbd_opts, "user");
 421     name       = qemu_opt_get(rbd_opts, "image");
 422     keypairs   = qemu_opt_get(rbd_opts, "=keyvalue-pairs");
 423
 424     ret = rados_create(&cluster, clientname);
 425     if (ret < 0) {
 426         error_setg_errno(errp, -ret, "error initializing");
 427         goto exit;
 428     }
 429
 430     /* try default location when conf=NULL, but ignore failure */
 431     ret = rados_conf_read_file(cluster, conf);
 432     if (conf && ret < 0) {
 433         error_setg_errno(errp, -ret, "error reading conf file %s", conf);
 434         ret = -EIO;
 435         goto shutdown;
 436     }
 437
 438     ret = qemu_rbd_set_keypairs(cluster, keypairs, errp);
 439     if (ret < 0) {
 440         ret = -EIO;
 441         goto shutdown;
 442     }
 443
 444     if (qemu_rbd_set_auth(cluster, secretid, errp) < 0) {
 445         ret = -EIO;
 446         goto shutdown;
 447     }
 448
 449     ret = rados_connect(cluster);
 450     if (ret < 0) {
 451         error_setg_errno(errp, -ret, "error connecting");
 452         goto shutdown;
 453     }
 454
 455     ret = rados_ioctx_create(cluster, pool, &io_ctx);
 456     if (ret < 0) {
 457         error_setg_errno(errp, -ret, "error opening pool %s", pool);
 458         goto shutdown;
 459     }
 460
 461     ret = rbd_create(io_ctx, name, bytes, &obj_order);
 462     if (ret < 0) {
 463         error_setg_errno(errp, -ret, "error rbd create");
 464     }
 465
 466     rados_ioctx_destroy(io_ctx);
 467
 468 shutdown:
 469     rados_shutdown(cluster);
 470
 471 exit:
 472     QDECREF(options);
 473     qemu_opts_del(rbd_opts);
 474     return ret;
 475 }
 476
 477 /*
 478  * This aio completion is being called from rbd_finish_bh() and runs in qemu
 479  * BH context.
 480  */
 481 static void qemu_rbd_complete_aio(RADOSCB *rcb)
 482 {
 483     RBDAIOCB *acb = rcb->acb;
 484     int64_t r;
 485
 486     r = rcb->ret;
 487
 488     if (acb->cmd != RBD_AIO_READ) {
 489         if (r < 0) {
 490             acb->ret = r;
 491             acb->error = 1;
 492         } else if (!acb->error) {
 493             acb->ret = rcb->size;
 494         }
 495     } else {
 496         if (r < 0) {
 497             qemu_rbd_memset(rcb, 0);
 498             acb->ret = r;
 499             acb->error = 1;
 500         } else if (r < rcb->size) {
 501             qemu_rbd_memset(rcb, r);
 502             if (!acb->error) {
 503                 acb->ret = rcb->size;
 504             }
 505         } else if (!acb->error) {
 506             acb->ret = r;
 507         }
 508     }
 509
 510     g_free(rcb);
 511
 512     if (!LIBRBD_USE_IOVEC) {
 513         if (acb->cmd == RBD_AIO_READ) {
 514             qemu_iovec_from_buf(acb->qiov, 0, acb->bounce, acb->qiov->size);
 515         }
 516         qemu_vfree(acb->bounce);
 517     }
 518
 519     acb->common.cb(acb->common.opaque, (acb->ret > 0 ? 0 : acb->ret));
 520
 521     qemu_aio_unref(acb);
 522 }
 523
 524 #define RBD_MON_HOST          0
 525 #define RBD_AUTH_SUPPORTED    1
 526
 527 static char *qemu_rbd_array_opts(QDict *options, const char *prefix, int type,
 528                                  Error **errp)
 529 {
 530     int num_entries;
 531     QemuOpts *opts = NULL;
 532     QDict *sub_options;
 533     const char *host;
 534     const char *port;
 535     char *str;
 536     char *rados_str = NULL;
 537     Error *local_err = NULL;
 538     int i;
 539
 540     assert(type == RBD_MON_HOST || type == RBD_AUTH_SUPPORTED);
 541
 542     num_entries = qdict_array_entries(options, prefix);
 543
 544     if (num_entries < 0) {
 545         error_setg(errp, "Parse error on RBD QDict array");
 546         return NULL;
 547     }
 548
 549     for (i = 0; i < num_entries; i++) {
 550         char *strbuf = NULL;
 551         const char *value;
 552         char *rados_str_tmp;
 553
 554         str = g_strdup_printf("%s%d.", prefix, i);
 555         qdict_extract_subqdict(options, &sub_options, str);
 556         g_free(str);
 557
 558         opts = qemu_opts_create(&runtime_opts, NULL, 0, &error_abort);
 559         qemu_opts_absorb_qdict(opts, sub_options, &local_err);
 560         QDECREF(sub_options);
 561         if (local_err) {
 562             error_propagate(errp, local_err);
 563             g_free(rados_str);
 564             rados_str = NULL;
 565             goto exit;
 566         }
 567
 568         if (type == RBD_MON_HOST) {
 569             host = qemu_opt_get(opts, "host");
 570             port = qemu_opt_get(opts, "port");
 571
 572             value = host;
 573             if (port) {
 574                 /* check for ipv6 */
 575                 if (strchr(host, ':')) {
 576                     strbuf = g_strdup_printf("[%s]:%s", host, port);
 577                 } else {
 578                     strbuf = g_strdup_printf("%s:%s", host, port);
 579                 }
 580                 value = strbuf;
 581             } else if (strchr(host, ':')) {
 582                 strbuf = g_strdup_printf("[%s]", host);
 583                 value = strbuf;
 584             }
 585         } else {
 586             value = qemu_opt_get(opts, "auth");
 587         }
 588
 589
 590         /* each iteration in the for loop will build upon the string, and if
 591          * rados_str is NULL then it is our first pass */
 592         if (rados_str) {
 593             /* separate options with ';', as that  is what rados_conf_set()
 594              * requires */
 595             rados_str_tmp = rados_str;
 596             rados_str = g_strdup_printf("%s;%s", rados_str_tmp, value);
 597             g_free(rados_str_tmp);
 598         } else {
 599             rados_str = g_strdup(value);
 600         }
 601
 602         g_free(strbuf);
 603         qemu_opts_del(opts);
 604         opts = NULL;
 605     }
 606
 607 exit:
 608     qemu_opts_del(opts);
 609     return rados_str;
 610 }
 611
 612 static int qemu_rbd_open(BlockDriverState *bs, QDict *options, int flags,
 613                          Error **errp)
 614 {
 615     BDRVRBDState *s = bs->opaque;
 616     const char *pool, *snap, *conf, *clientname, *name, *keypairs;
 617     const char *secretid;
 618     QemuOpts *opts;
 619     Error *local_err = NULL;
 620     char *mon_host = NULL;
 621     char *auth_supported = NULL;
 622     int r;
 623
 624     opts = qemu_opts_create(&runtime_opts, NULL, 0, &error_abort);
 625     qemu_opts_absorb_qdict(opts, options, &local_err);
 626     if (local_err) {
 627         error_propagate(errp, local_err);
 628         qemu_opts_del(opts);
 629         return -EINVAL;
 630     }
 631
 632     auth_supported = qemu_rbd_array_opts(options, "auth-supported.",
 633                                          RBD_AUTH_SUPPORTED, &local_err);
 634     if (local_err) {
 635         error_propagate(errp, local_err);
 636         r = -EINVAL;
 637         goto failed_opts;
 638     }
 639
 640     mon_host = qemu_rbd_array_opts(options, "server.",
 641                                    RBD_MON_HOST, &local_err);
 642     if (local_err) {
 643         error_propagate(errp, local_err);
 644         r = -EINVAL;
 645         goto failed_opts;
 646     }
 647
 648     secretid = qemu_opt_get(opts, "password-secret");
 649
 650     pool           = qemu_opt_get(opts, "pool");
 651     conf           = qemu_opt_get(opts, "conf");
 652     snap           = qemu_opt_get(opts, "snapshot");
 653     clientname     = qemu_opt_get(opts, "user");
 654     name           = qemu_opt_get(opts, "image");
 655     keypairs       = qemu_opt_get(opts, "=keyvalue-pairs");
 656
 657     if (!pool || !name) {
 658         error_setg(errp, "Parameters 'pool' and 'image' are required");
 659         r = -EINVAL;
 660         goto failed_opts;
 661     }
 662
 663     r = rados_create(&s->cluster, clientname);
 664     if (r < 0) {
 665         error_setg_errno(errp, -r, "error initializing");
 666         goto failed_opts;
 667     }
 668
 669     s->snap = g_strdup(snap);
 670     s->name = g_strdup(name);
 671
 672     /* try default location when conf=NULL, but ignore failure */
 673     r = rados_conf_read_file(s->cluster, conf);
 674     if (conf && r < 0) {
 675         error_setg_errno(errp, -r, "error reading conf file %s", conf);
 676         goto failed_shutdown;
 677     }
 678
 679     r = qemu_rbd_set_keypairs(s->cluster, keypairs, errp);
 680     if (r < 0) {
 681         goto failed_shutdown;
 682     }
 683
 684     if (mon_host) {
 685         r = rados_conf_set(s->cluster, "mon_host", mon_host);
 686         if (r < 0) {
 687             goto failed_shutdown;
 688         }
 689     }
 690
 691     if (auth_supported) {
 692         r = rados_conf_set(s->cluster, "auth_supported", auth_supported);
 693         if (r < 0) {
 694             goto failed_shutdown;
 695         }
 696     }
 697
 698     if (qemu_rbd_set_auth(s->cluster, secretid, errp) < 0) {
 699         r = -EIO;
 700         goto failed_shutdown;
 701     }
 702
 703     /*
 704      * Fallback to more conservative semantics if setting cache
 705      * options fails. Ignore errors from setting rbd_cache because the
 706      * only possible error is that the option does not exist, and
 707      * librbd defaults to no caching. If write through caching cannot
 708      * be set up, fall back to no caching.
 709      */
 710     if (flags & BDRV_O_NOCACHE) {
 711         rados_conf_set(s->cluster, "rbd_cache", "false");
 712     } else {
 713         rados_conf_set(s->cluster, "rbd_cache", "true");
 714     }
 715
 716     r = rados_connect(s->cluster);
 717     if (r < 0) {
 718         error_setg_errno(errp, -r, "error connecting");
 719         goto failed_shutdown;
 720     }
 721
 722     r = rados_ioctx_create(s->cluster, pool, &s->io_ctx);
 723     if (r < 0) {
 724         error_setg_errno(errp, -r, "error opening pool %s", pool);
 725         goto failed_shutdown;
 726     }
 727
 728     r = rbd_open(s->io_ctx, s->name, &s->image, s->snap);
 729     if (r < 0) {
 730         error_setg_errno(errp, -r, "error reading header from %s", s->name);
 731         goto failed_open;
 732     }
 733
 734     bs->read_only = (s->snap != NULL);
 735
 736     qemu_opts_del(opts);
 737     return 0;
 738
 739 failed_open:
 740     rados_ioctx_destroy(s->io_ctx);
 741 failed_shutdown:
 742     rados_shutdown(s->cluster);
 743     g_free(s->snap);
 744     g_free(s->name);
 745 failed_opts:
 746     qemu_opts_del(opts);
 747     g_free(mon_host);
 748     g_free(auth_supported);
 749     return r;
 750 }
 751
 752 static void qemu_rbd_close(BlockDriverState *bs)
 753 {
 754     BDRVRBDState *s = bs->opaque;
 755
 756     rbd_close(s->image);
 757     rados_ioctx_destroy(s->io_ctx);
 758     g_free(s->snap);
 759     g_free(s->name);
 760     rados_shutdown(s->cluster);
 761 }
 762
 763 static const AIOCBInfo rbd_aiocb_info = {
 764     .aiocb_size = sizeof(RBDAIOCB),
 765 };
 766
 767 static void rbd_finish_bh(void *opaque)
 768 {
 769     RADOSCB *rcb = opaque;
 770     qemu_rbd_complete_aio(rcb);
 771 }
 772
 773 /*
 774  * This is the callback function for rbd_aio_read and _write
 775  *
 776  * Note: this function is being called from a non qemu thread so
 777  * we need to be careful about what we do here. Generally we only
 778  * schedule a BH, and do the rest of the io completion handling
 779  * from rbd_finish_bh() which runs in a qemu context.
 780  */
 781 static void rbd_finish_aiocb(rbd_completion_t c, RADOSCB *rcb)
 782 {
 783     RBDAIOCB *acb = rcb->acb;
 784
 785     rcb->ret = rbd_aio_get_return_value(c);
 786     rbd_aio_release(c);
 787
 788     aio_bh_schedule_oneshot(bdrv_get_aio_context(acb->common.bs),
 789                             rbd_finish_bh, rcb);
 790 }
 791
 792 static int rbd_aio_discard_wrapper(rbd_image_t image,
 793                                    uint64_t off,
 794                                    uint64_t len,
 795                                    rbd_completion_t comp)
 796 {
 797 #ifdef LIBRBD_SUPPORTS_DISCARD
 798     return rbd_aio_discard(image, off, len, comp);
 799 #else
 800     return -ENOTSUP;
 801 #endif
 802 }
 803
 804 static int rbd_aio_flush_wrapper(rbd_image_t image,
 805                                  rbd_completion_t comp)
 806 {
 807 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
 808     return rbd_aio_flush(image, comp);
 809 #else
 810     return -ENOTSUP;
 811 #endif
 812 }
 813
 814 static BlockAIOCB *rbd_start_aio(BlockDriverState *bs,
 815                                  int64_t off,
 816                                  QEMUIOVector *qiov,
 817                                  int64_t size,
 818                                  BlockCompletionFunc *cb,
 819                                  void *opaque,
 820                                  RBDAIOCmd cmd)
 821 {
 822     RBDAIOCB *acb;
 823     RADOSCB *rcb = NULL;
 824     rbd_completion_t c;
 825     int r;
 826
 827     BDRVRBDState *s = bs->opaque;
 828
 829     acb = qemu_aio_get(&rbd_aiocb_info, bs, cb, opaque);
 830     acb->cmd = cmd;
 831     acb->qiov = qiov;
 832     assert(!qiov || qiov->size == size);
 833
 834     rcb = g_new(RADOSCB, 1);
 835
 836     if (!LIBRBD_USE_IOVEC) {
 837         if (cmd == RBD_AIO_DISCARD || cmd == RBD_AIO_FLUSH) {
 838             acb->bounce = NULL;
 839         } else {
 840             acb->bounce = qemu_try_blockalign(bs, qiov->size);
 841             if (acb->bounce == NULL) {
 842                 goto failed;
 843             }
 844         }
 845         if (cmd == RBD_AIO_WRITE) {
 846             qemu_iovec_to_buf(acb->qiov, 0, acb->bounce, qiov->size);
 847         }
 848         rcb->buf = acb->bounce;
 849     }
 850
 851     acb->ret = 0;
 852     acb->error = 0;
 853     acb->s = s;
 854
 855     rcb->acb = acb;
 856     rcb->s = acb->s;
 857     rcb->size = size;
 858     r = rbd_aio_create_completion(rcb, (rbd_callback_t) rbd_finish_aiocb, &c);
 859     if (r < 0) {
 860         goto failed;
 861     }
 862
 863     switch (cmd) {
 864     case RBD_AIO_WRITE:
 865 #ifdef LIBRBD_SUPPORTS_IOVEC
 866             r = rbd_aio_writev(s->image, qiov->iov, qiov->niov, off, c);
 867 #else
 868             r = rbd_aio_write(s->image, off, size, rcb->buf, c);
 869 #endif
 870         break;
 871     case RBD_AIO_READ:
 872 #ifdef LIBRBD_SUPPORTS_IOVEC
 873             r = rbd_aio_readv(s->image, qiov->iov, qiov->niov, off, c);
 874 #else
 875             r = rbd_aio_read(s->image, off, size, rcb->buf, c);
 876 #endif
 877         break;
 878     case RBD_AIO_DISCARD:
 879         r = rbd_aio_discard_wrapper(s->image, off, size, c);
 880         break;
 881     case RBD_AIO_FLUSH:
 882         r = rbd_aio_flush_wrapper(s->image, c);
 883         break;
 884     default:
 885         r = -EINVAL;
 886     }
 887
 888     if (r < 0) {
 889         goto failed_completion;
 890     }
 891     return &acb->common;
 892
 893 failed_completion:
 894     rbd_aio_release(c);
 895 failed:
 896     g_free(rcb);
 897     if (!LIBRBD_USE_IOVEC) {
 898         qemu_vfree(acb->bounce);
 899     }
 900
 901     qemu_aio_unref(acb);
 902     return NULL;
 903 }
 904
 905 static BlockAIOCB *qemu_rbd_aio_readv(BlockDriverState *bs,
 906                                       int64_t sector_num,
 907                                       QEMUIOVector *qiov,
 908                                       int nb_sectors,
 909                                       BlockCompletionFunc *cb,
 910                                       void *opaque)
 911 {
 912     return rbd_start_aio(bs, sector_num << BDRV_SECTOR_BITS, qiov,
 913                          (int64_t) nb_sectors << BDRV_SECTOR_BITS, cb, opaque,
 914                          RBD_AIO_READ);
 915 }
 916
 917 static BlockAIOCB *qemu_rbd_aio_writev(BlockDriverState *bs,
 918                                        int64_t sector_num,
 919                                        QEMUIOVector *qiov,
 920                                        int nb_sectors,
 921                                        BlockCompletionFunc *cb,
 922                                        void *opaque)
 923 {
 924     return rbd_start_aio(bs, sector_num << BDRV_SECTOR_BITS, qiov,
 925                          (int64_t) nb_sectors << BDRV_SECTOR_BITS, cb, opaque,
 926                          RBD_AIO_WRITE);
 927 }
 928
 929 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
 930 static BlockAIOCB *qemu_rbd_aio_flush(BlockDriverState *bs,
 931                                       BlockCompletionFunc *cb,
 932                                       void *opaque)
 933 {
 934     return rbd_start_aio(bs, 0, NULL, 0, cb, opaque, RBD_AIO_FLUSH);
 935 }
 936
 937 #else
 938
 939 static int qemu_rbd_co_flush(BlockDriverState *bs)
 940 {
 941 #if LIBRBD_VERSION_CODE >= LIBRBD_VERSION(0, 1, 1)
 942     /* rbd_flush added in 0.1.1 */
 943     BDRVRBDState *s = bs->opaque;
 944     return rbd_flush(s->image);
 945 #else
 946     return 0;
 947 #endif
 948 }
 949 #endif
 950
 951 static int qemu_rbd_getinfo(BlockDriverState *bs, BlockDriverInfo *bdi)
 952 {
 953     BDRVRBDState *s = bs->opaque;
 954     rbd_image_info_t info;
 955     int r;
 956
 957     r = rbd_stat(s->image, &info, sizeof(info));
 958     if (r < 0) {
 959         return r;
 960     }
 961
 962     bdi->cluster_size = info.obj_size;
 963     return 0;
 964 }
 965
 966 static int64_t qemu_rbd_getlength(BlockDriverState *bs)
 967 {
 968     BDRVRBDState *s = bs->opaque;
 969     rbd_image_info_t info;
 970     int r;
 971
 972     r = rbd_stat(s->image, &info, sizeof(info));
 973     if (r < 0) {
 974         return r;
 975     }
 976
 977     return info.size;
 978 }
 979
 980 static int qemu_rbd_truncate(BlockDriverState *bs, int64_t offset)
 981 {
 982     BDRVRBDState *s = bs->opaque;
 983     int r;
 984
 985     r = rbd_resize(s->image, offset);
 986     if (r < 0) {
 987         return r;
 988     }
 989
 990     return 0;
 991 }
 992
 993 static int qemu_rbd_snap_create(BlockDriverState *bs,
 994                                 QEMUSnapshotInfo *sn_info)
 995 {
 996     BDRVRBDState *s = bs->opaque;
 997     int r;
 998
 999     if (sn_info->name[0] == '\0') {
1000         return -EINVAL; /* we need a name for rbd snapshots */
1001     }
1002
1003     /*
1004      * rbd snapshots are using the name as the user controlled unique identifier
1005      * we can't use the rbd snapid for that purpose, as it can't be set
1006      */
1007     if (sn_info->id_str[0] != '\0' &&
1008         strcmp(sn_info->id_str, sn_info->name) != 0) {
1009         return -EINVAL;
1010     }
1011
1012     if (strlen(sn_info->name) >= sizeof(sn_info->id_str)) {
1013         return -ERANGE;
1014     }
1015
1016     r = rbd_snap_create(s->image, sn_info->name);
1017     if (r < 0) {
1018         error_report("failed to create snap: %s", strerror(-r));
1019         return r;
1020     }
1021
1022     return 0;
1023 }
1024
1025 static int qemu_rbd_snap_remove(BlockDriverState *bs,
1026                                 const char *snapshot_id,
1027                                 const char *snapshot_name,
1028                                 Error **errp)
1029 {
1030     BDRVRBDState *s = bs->opaque;
1031     int r;
1032
1033     if (!snapshot_name) {
1034         error_setg(errp, "rbd need a valid snapshot name");
1035         return -EINVAL;
1036     }
1037
1038     /* If snapshot_id is specified, it must be equal to name, see
1039        qemu_rbd_snap_list() */
1040     if (snapshot_id && strcmp(snapshot_id, snapshot_name)) {
1041         error_setg(errp,
1042                    "rbd do not support snapshot id, it should be NULL or "
1043                    "equal to snapshot name");
1044         return -EINVAL;
1045     }
1046
1047     r = rbd_snap_remove(s->image, snapshot_name);
1048     if (r < 0) {
1049         error_setg_errno(errp, -r, "Failed to remove the snapshot");
1050     }
1051     return r;
1052 }
1053
1054 static int qemu_rbd_snap_rollback(BlockDriverState *bs,
1055                                   const char *snapshot_name)
1056 {
1057     BDRVRBDState *s = bs->opaque;
1058
1059     return rbd_snap_rollback(s->image, snapshot_name);
1060 }
1061
1062 static int qemu_rbd_snap_list(BlockDriverState *bs,
1063                               QEMUSnapshotInfo **psn_tab)
1064 {
1065     BDRVRBDState *s = bs->opaque;
1066     QEMUSnapshotInfo *sn_info, *sn_tab = NULL;
1067     int i, snap_count;
1068     rbd_snap_info_t *snaps;
1069     int max_snaps = RBD_MAX_SNAPS;
1070
1071     do {
1072         snaps = g_new(rbd_snap_info_t, max_snaps);
1073         snap_count = rbd_snap_list(s->image, snaps, &max_snaps);
1074         if (snap_count <= 0) {
1075             g_free(snaps);
1076         }
1077     } while (snap_count == -ERANGE);
1078
1079     if (snap_count <= 0) {
1080         goto done;
1081     }
1082
1083     sn_tab = g_new0(QEMUSnapshotInfo, snap_count);
1084
1085     for (i = 0; i < snap_count; i++) {
1086         const char *snap_name = snaps[i].name;
1087
1088         sn_info = sn_tab + i;
1089         pstrcpy(sn_info->id_str, sizeof(sn_info->id_str), snap_name);
1090         pstrcpy(sn_info->name, sizeof(sn_info->name), snap_name);
1091
1092         sn_info->vm_state_size = snaps[i].size;
1093         sn_info->date_sec = 0;
1094         sn_info->date_nsec = 0;
1095         sn_info->vm_clock_nsec = 0;
1096     }
1097     rbd_snap_list_end(snaps);
1098     g_free(snaps);
1099
1100  done:
1101     *psn_tab = sn_tab;
1102     return snap_count;
1103 }
1104
1105 #ifdef LIBRBD_SUPPORTS_DISCARD
1106 static BlockAIOCB *qemu_rbd_aio_pdiscard(BlockDriverState *bs,
1107                                          int64_t offset,
1108                                          int count,
1109                                          BlockCompletionFunc *cb,
1110                                          void *opaque)
1111 {
1112     return rbd_start_aio(bs, offset, NULL, count, cb, opaque,
1113                          RBD_AIO_DISCARD);
1114 }
1115 #endif
1116
1117 #ifdef LIBRBD_SUPPORTS_INVALIDATE
1118 static void qemu_rbd_invalidate_cache(BlockDriverState *bs,
1119                                       Error **errp)
1120 {
1121     BDRVRBDState *s = bs->opaque;
1122     int r = rbd_invalidate_cache(s->image);
1123     if (r < 0) {
1124         error_setg_errno(errp, -r, "Failed to invalidate the cache");
1125     }
1126 }
1127 #endif
1128
1129 static QemuOptsList qemu_rbd_create_opts = {
1130     .name = "rbd-create-opts",
1131     .head = QTAILQ_HEAD_INITIALIZER(qemu_rbd_create_opts.head),
1132     .desc = {
1133         {
1134             .name = BLOCK_OPT_SIZE,
1135             .type = QEMU_OPT_SIZE,
1136             .help = "Virtual disk size"
1137         },
1138         {
1139             .name = BLOCK_OPT_CLUSTER_SIZE,
1140             .type = QEMU_OPT_SIZE,
1141             .help = "RBD object size"
1142         },
1143         {
1144             .name = "password-secret",
1145             .type = QEMU_OPT_STRING,
1146             .help = "ID of secret providing the password",
1147         },
1148         { /* end of list */ }
1149     }
1150 };
1151
1152 static BlockDriver bdrv_rbd = {
1153     .format_name            = "rbd",
1154     .instance_size          = sizeof(BDRVRBDState),
1155     .bdrv_parse_filename    = qemu_rbd_parse_filename,
1156     .bdrv_file_open         = qemu_rbd_open,
1157     .bdrv_close             = qemu_rbd_close,
1158     .bdrv_create            = qemu_rbd_create,
1159     .bdrv_has_zero_init     = bdrv_has_zero_init_1,
1160     .bdrv_get_info          = qemu_rbd_getinfo,
1161     .create_opts            = &qemu_rbd_create_opts,
1162     .bdrv_getlength         = qemu_rbd_getlength,
1163     .bdrv_truncate          = qemu_rbd_truncate,
1164     .protocol_name          = "rbd",
1165
1166     .bdrv_aio_readv         = qemu_rbd_aio_readv,
1167     .bdrv_aio_writev        = qemu_rbd_aio_writev,
1168
1169 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
1170     .bdrv_aio_flush         = qemu_rbd_aio_flush,
1171 #else
1172     .bdrv_co_flush_to_disk  = qemu_rbd_co_flush,
1173 #endif
1174
1175 #ifdef LIBRBD_SUPPORTS_DISCARD
1176     .bdrv_aio_pdiscard      = qemu_rbd_aio_pdiscard,
1177 #endif
1178
1179     .bdrv_snapshot_create   = qemu_rbd_snap_create,
1180     .bdrv_snapshot_delete   = qemu_rbd_snap_remove,
1181     .bdrv_snapshot_list     = qemu_rbd_snap_list,
1182     .bdrv_snapshot_goto     = qemu_rbd_snap_rollback,
1183 #ifdef LIBRBD_SUPPORTS_INVALIDATE
1184     .bdrv_invalidate_cache  = qemu_rbd_invalidate_cache,
1185 #endif
1186 };
1187
1188 static void bdrv_rbd_init(void)
1189 {
1190     bdrv_register(&bdrv_rbd);
1191 }
1192
1193 block_init(bdrv_rbd_init);