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 *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(options, "pool", qstring_from_str(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(options, "image", qstring_from_str(found_str)); 163 164 found_str = qemu_rbd_next_tok(p, ':', &p); 165 qemu_rbd_unescape(found_str); 166 qdict_put(options, "snapshot", qstring_from_str(found_str)); 167 } else { 168 found_str = qemu_rbd_next_tok(p, ':', &p); 169 qemu_rbd_unescape(found_str); 170 qdict_put(options, "image", qstring_from_str(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(options, "conf", qstring_from_str(value)); 193 } else if (!strcmp(name, "id")) { 194 qdict_put(options, "user" , qstring_from_str(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(keypairs, qstring_from_str(name)); 208 qlist_append(keypairs, qstring_from_str(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 { /* end of list */ } 344 }, 345 }; 346 347 static int qemu_rbd_create(const char *filename, QemuOpts *opts, Error **errp) 348 { 349 Error *local_err = NULL; 350 int64_t bytes = 0; 351 int64_t objsize; 352 int obj_order = 0; 353 const char *pool, *name, *conf, *clientname, *keypairs; 354 const char *secretid; 355 rados_t cluster; 356 rados_ioctx_t io_ctx; 357 QDict *options = NULL; 358 int ret = 0; 359 360 secretid = qemu_opt_get(opts, "password-secret"); 361 362 /* Read out options */ 363 bytes = ROUND_UP(qemu_opt_get_size_del(opts, BLOCK_OPT_SIZE, 0), 364 BDRV_SECTOR_SIZE); 365 objsize = qemu_opt_get_size_del(opts, BLOCK_OPT_CLUSTER_SIZE, 0); 366 if (objsize) { 367 if ((objsize - 1) & objsize) { /* not a power of 2? */ 368 error_setg(errp, "obj size needs to be power of 2"); 369 ret = -EINVAL; 370 goto exit; 371 } 372 if (objsize < 4096) { 373 error_setg(errp, "obj size too small"); 374 ret = -EINVAL; 375 goto exit; 376 } 377 obj_order = ctz32(objsize); 378 } 379 380 options = qdict_new(); 381 qemu_rbd_parse_filename(filename, options, &local_err); 382 if (local_err) { 383 ret = -EINVAL; 384 error_propagate(errp, local_err); 385 goto exit; 386 } 387 388 /* 389 * Caution: while qdict_get_try_str() is fine, getting non-string 390 * types would require more care. When @options come from -blockdev 391 * or blockdev_add, its members are typed according to the QAPI 392 * schema, but when they come from -drive, they're all QString. 393 */ 394 pool = qdict_get_try_str(options, "pool"); 395 conf = qdict_get_try_str(options, "conf"); 396 clientname = qdict_get_try_str(options, "user"); 397 name = qdict_get_try_str(options, "image"); 398 keypairs = qdict_get_try_str(options, "=keyvalue-pairs"); 399 400 ret = rados_create(&cluster, clientname); 401 if (ret < 0) { 402 error_setg_errno(errp, -ret, "error initializing"); 403 goto exit; 404 } 405 406 /* try default location when conf=NULL, but ignore failure */ 407 ret = rados_conf_read_file(cluster, conf); 408 if (conf && ret < 0) { 409 error_setg_errno(errp, -ret, "error reading conf file %s", conf); 410 ret = -EIO; 411 goto shutdown; 412 } 413 414 ret = qemu_rbd_set_keypairs(cluster, keypairs, errp); 415 if (ret < 0) { 416 ret = -EIO; 417 goto shutdown; 418 } 419 420 if (qemu_rbd_set_auth(cluster, secretid, errp) < 0) { 421 ret = -EIO; 422 goto shutdown; 423 } 424 425 ret = rados_connect(cluster); 426 if (ret < 0) { 427 error_setg_errno(errp, -ret, "error connecting"); 428 goto shutdown; 429 } 430 431 ret = rados_ioctx_create(cluster, pool, &io_ctx); 432 if (ret < 0) { 433 error_setg_errno(errp, -ret, "error opening pool %s", pool); 434 goto shutdown; 435 } 436 437 ret = rbd_create(io_ctx, name, bytes, &obj_order); 438 if (ret < 0) { 439 error_setg_errno(errp, -ret, "error rbd create"); 440 } 441 442 rados_ioctx_destroy(io_ctx); 443 444 shutdown: 445 rados_shutdown(cluster); 446 447 exit: 448 QDECREF(options); 449 return ret; 450 } 451 452 /* 453 * This aio completion is being called from rbd_finish_bh() and runs in qemu 454 * BH context. 455 */ 456 static void qemu_rbd_complete_aio(RADOSCB *rcb) 457 { 458 RBDAIOCB *acb = rcb->acb; 459 int64_t r; 460 461 r = rcb->ret; 462 463 if (acb->cmd != RBD_AIO_READ) { 464 if (r < 0) { 465 acb->ret = r; 466 acb->error = 1; 467 } else if (!acb->error) { 468 acb->ret = rcb->size; 469 } 470 } else { 471 if (r < 0) { 472 qemu_rbd_memset(rcb, 0); 473 acb->ret = r; 474 acb->error = 1; 475 } else if (r < rcb->size) { 476 qemu_rbd_memset(rcb, r); 477 if (!acb->error) { 478 acb->ret = rcb->size; 479 } 480 } else if (!acb->error) { 481 acb->ret = r; 482 } 483 } 484 485 g_free(rcb); 486 487 if (!LIBRBD_USE_IOVEC) { 488 if (acb->cmd == RBD_AIO_READ) { 489 qemu_iovec_from_buf(acb->qiov, 0, acb->bounce, acb->qiov->size); 490 } 491 qemu_vfree(acb->bounce); 492 } 493 494 acb->common.cb(acb->common.opaque, (acb->ret > 0 ? 0 : acb->ret)); 495 496 qemu_aio_unref(acb); 497 } 498 499 static char *qemu_rbd_mon_host(QDict *options, Error **errp) 500 { 501 const char **vals = g_new(const char *, qdict_size(options) + 1); 502 char keybuf[32]; 503 const char *host, *port; 504 char *rados_str; 505 int i; 506 507 for (i = 0;; i++) { 508 sprintf(keybuf, "server.%d.host", i); 509 host = qdict_get_try_str(options, keybuf); 510 qdict_del(options, keybuf); 511 sprintf(keybuf, "server.%d.port", i); 512 port = qdict_get_try_str(options, keybuf); 513 qdict_del(options, keybuf); 514 if (!host && !port) { 515 break; 516 } 517 if (!host) { 518 error_setg(errp, "Parameter server.%d.host is missing", i); 519 rados_str = NULL; 520 goto out; 521 } 522 523 if (strchr(host, ':')) { 524 vals[i] = port ? g_strdup_printf("[%s]:%s", host, port) 525 : g_strdup_printf("[%s]", host); 526 } else { 527 vals[i] = port ? g_strdup_printf("%s:%s", host, port) 528 : g_strdup(host); 529 } 530 } 531 vals[i] = NULL; 532 533 rados_str = i ? g_strjoinv(";", (char **)vals) : NULL; 534 out: 535 g_strfreev((char **)vals); 536 return rados_str; 537 } 538 539 static int qemu_rbd_open(BlockDriverState *bs, QDict *options, int flags, 540 Error **errp) 541 { 542 BDRVRBDState *s = bs->opaque; 543 const char *pool, *snap, *conf, *clientname, *name, *keypairs; 544 const char *secretid; 545 QemuOpts *opts; 546 Error *local_err = NULL; 547 char *mon_host = NULL; 548 int r; 549 550 opts = qemu_opts_create(&runtime_opts, NULL, 0, &error_abort); 551 qemu_opts_absorb_qdict(opts, options, &local_err); 552 if (local_err) { 553 error_propagate(errp, local_err); 554 r = -EINVAL; 555 goto failed_opts; 556 } 557 558 mon_host = qemu_rbd_mon_host(options, &local_err); 559 if (local_err) { 560 error_propagate(errp, local_err); 561 r = -EINVAL; 562 goto failed_opts; 563 } 564 565 secretid = qemu_opt_get(opts, "password-secret"); 566 567 pool = qemu_opt_get(opts, "pool"); 568 conf = qemu_opt_get(opts, "conf"); 569 snap = qemu_opt_get(opts, "snapshot"); 570 clientname = qemu_opt_get(opts, "user"); 571 name = qemu_opt_get(opts, "image"); 572 keypairs = qemu_opt_get(opts, "=keyvalue-pairs"); 573 574 if (!pool || !name) { 575 error_setg(errp, "Parameters 'pool' and 'image' are required"); 576 r = -EINVAL; 577 goto failed_opts; 578 } 579 580 r = rados_create(&s->cluster, clientname); 581 if (r < 0) { 582 error_setg_errno(errp, -r, "error initializing"); 583 goto failed_opts; 584 } 585 586 s->snap = g_strdup(snap); 587 s->name = g_strdup(name); 588 589 /* try default location when conf=NULL, but ignore failure */ 590 r = rados_conf_read_file(s->cluster, conf); 591 if (conf && r < 0) { 592 error_setg_errno(errp, -r, "error reading conf file %s", conf); 593 goto failed_shutdown; 594 } 595 596 r = qemu_rbd_set_keypairs(s->cluster, keypairs, errp); 597 if (r < 0) { 598 goto failed_shutdown; 599 } 600 601 if (mon_host) { 602 r = rados_conf_set(s->cluster, "mon_host", mon_host); 603 if (r < 0) { 604 goto failed_shutdown; 605 } 606 } 607 608 if (qemu_rbd_set_auth(s->cluster, secretid, errp) < 0) { 609 r = -EIO; 610 goto failed_shutdown; 611 } 612 613 /* 614 * Fallback to more conservative semantics if setting cache 615 * options fails. Ignore errors from setting rbd_cache because the 616 * only possible error is that the option does not exist, and 617 * librbd defaults to no caching. If write through caching cannot 618 * be set up, fall back to no caching. 619 */ 620 if (flags & BDRV_O_NOCACHE) { 621 rados_conf_set(s->cluster, "rbd_cache", "false"); 622 } else { 623 rados_conf_set(s->cluster, "rbd_cache", "true"); 624 } 625 626 r = rados_connect(s->cluster); 627 if (r < 0) { 628 error_setg_errno(errp, -r, "error connecting"); 629 goto failed_shutdown; 630 } 631 632 r = rados_ioctx_create(s->cluster, pool, &s->io_ctx); 633 if (r < 0) { 634 error_setg_errno(errp, -r, "error opening pool %s", pool); 635 goto failed_shutdown; 636 } 637 638 r = rbd_open(s->io_ctx, s->name, &s->image, s->snap); 639 if (r < 0) { 640 error_setg_errno(errp, -r, "error reading header from %s", s->name); 641 goto failed_open; 642 } 643 644 bs->read_only = (s->snap != NULL); 645 646 qemu_opts_del(opts); 647 return 0; 648 649 failed_open: 650 rados_ioctx_destroy(s->io_ctx); 651 failed_shutdown: 652 rados_shutdown(s->cluster); 653 g_free(s->snap); 654 g_free(s->name); 655 failed_opts: 656 qemu_opts_del(opts); 657 g_free(mon_host); 658 return r; 659 } 660 661 static void qemu_rbd_close(BlockDriverState *bs) 662 { 663 BDRVRBDState *s = bs->opaque; 664 665 rbd_close(s->image); 666 rados_ioctx_destroy(s->io_ctx); 667 g_free(s->snap); 668 g_free(s->name); 669 rados_shutdown(s->cluster); 670 } 671 672 static const AIOCBInfo rbd_aiocb_info = { 673 .aiocb_size = sizeof(RBDAIOCB), 674 }; 675 676 static void rbd_finish_bh(void *opaque) 677 { 678 RADOSCB *rcb = opaque; 679 qemu_rbd_complete_aio(rcb); 680 } 681 682 /* 683 * This is the callback function for rbd_aio_read and _write 684 * 685 * Note: this function is being called from a non qemu thread so 686 * we need to be careful about what we do here. Generally we only 687 * schedule a BH, and do the rest of the io completion handling 688 * from rbd_finish_bh() which runs in a qemu context. 689 */ 690 static void rbd_finish_aiocb(rbd_completion_t c, RADOSCB *rcb) 691 { 692 RBDAIOCB *acb = rcb->acb; 693 694 rcb->ret = rbd_aio_get_return_value(c); 695 rbd_aio_release(c); 696 697 aio_bh_schedule_oneshot(bdrv_get_aio_context(acb->common.bs), 698 rbd_finish_bh, rcb); 699 } 700 701 static int rbd_aio_discard_wrapper(rbd_image_t image, 702 uint64_t off, 703 uint64_t len, 704 rbd_completion_t comp) 705 { 706 #ifdef LIBRBD_SUPPORTS_DISCARD 707 return rbd_aio_discard(image, off, len, comp); 708 #else 709 return -ENOTSUP; 710 #endif 711 } 712 713 static int rbd_aio_flush_wrapper(rbd_image_t image, 714 rbd_completion_t comp) 715 { 716 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH 717 return rbd_aio_flush(image, comp); 718 #else 719 return -ENOTSUP; 720 #endif 721 } 722 723 static BlockAIOCB *rbd_start_aio(BlockDriverState *bs, 724 int64_t off, 725 QEMUIOVector *qiov, 726 int64_t size, 727 BlockCompletionFunc *cb, 728 void *opaque, 729 RBDAIOCmd cmd) 730 { 731 RBDAIOCB *acb; 732 RADOSCB *rcb = NULL; 733 rbd_completion_t c; 734 int r; 735 736 BDRVRBDState *s = bs->opaque; 737 738 acb = qemu_aio_get(&rbd_aiocb_info, bs, cb, opaque); 739 acb->cmd = cmd; 740 acb->qiov = qiov; 741 assert(!qiov || qiov->size == size); 742 743 rcb = g_new(RADOSCB, 1); 744 745 if (!LIBRBD_USE_IOVEC) { 746 if (cmd == RBD_AIO_DISCARD || cmd == RBD_AIO_FLUSH) { 747 acb->bounce = NULL; 748 } else { 749 acb->bounce = qemu_try_blockalign(bs, qiov->size); 750 if (acb->bounce == NULL) { 751 goto failed; 752 } 753 } 754 if (cmd == RBD_AIO_WRITE) { 755 qemu_iovec_to_buf(acb->qiov, 0, acb->bounce, qiov->size); 756 } 757 rcb->buf = acb->bounce; 758 } 759 760 acb->ret = 0; 761 acb->error = 0; 762 acb->s = s; 763 764 rcb->acb = acb; 765 rcb->s = acb->s; 766 rcb->size = size; 767 r = rbd_aio_create_completion(rcb, (rbd_callback_t) rbd_finish_aiocb, &c); 768 if (r < 0) { 769 goto failed; 770 } 771 772 switch (cmd) { 773 case RBD_AIO_WRITE: 774 #ifdef LIBRBD_SUPPORTS_IOVEC 775 r = rbd_aio_writev(s->image, qiov->iov, qiov->niov, off, c); 776 #else 777 r = rbd_aio_write(s->image, off, size, rcb->buf, c); 778 #endif 779 break; 780 case RBD_AIO_READ: 781 #ifdef LIBRBD_SUPPORTS_IOVEC 782 r = rbd_aio_readv(s->image, qiov->iov, qiov->niov, off, c); 783 #else 784 r = rbd_aio_read(s->image, off, size, rcb->buf, c); 785 #endif 786 break; 787 case RBD_AIO_DISCARD: 788 r = rbd_aio_discard_wrapper(s->image, off, size, c); 789 break; 790 case RBD_AIO_FLUSH: 791 r = rbd_aio_flush_wrapper(s->image, c); 792 break; 793 default: 794 r = -EINVAL; 795 } 796 797 if (r < 0) { 798 goto failed_completion; 799 } 800 return &acb->common; 801 802 failed_completion: 803 rbd_aio_release(c); 804 failed: 805 g_free(rcb); 806 if (!LIBRBD_USE_IOVEC) { 807 qemu_vfree(acb->bounce); 808 } 809 810 qemu_aio_unref(acb); 811 return NULL; 812 } 813 814 static BlockAIOCB *qemu_rbd_aio_readv(BlockDriverState *bs, 815 int64_t sector_num, 816 QEMUIOVector *qiov, 817 int nb_sectors, 818 BlockCompletionFunc *cb, 819 void *opaque) 820 { 821 return rbd_start_aio(bs, sector_num << BDRV_SECTOR_BITS, qiov, 822 (int64_t) nb_sectors << BDRV_SECTOR_BITS, cb, opaque, 823 RBD_AIO_READ); 824 } 825 826 static BlockAIOCB *qemu_rbd_aio_writev(BlockDriverState *bs, 827 int64_t sector_num, 828 QEMUIOVector *qiov, 829 int nb_sectors, 830 BlockCompletionFunc *cb, 831 void *opaque) 832 { 833 return rbd_start_aio(bs, sector_num << BDRV_SECTOR_BITS, qiov, 834 (int64_t) nb_sectors << BDRV_SECTOR_BITS, cb, opaque, 835 RBD_AIO_WRITE); 836 } 837 838 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH 839 static BlockAIOCB *qemu_rbd_aio_flush(BlockDriverState *bs, 840 BlockCompletionFunc *cb, 841 void *opaque) 842 { 843 return rbd_start_aio(bs, 0, NULL, 0, cb, opaque, RBD_AIO_FLUSH); 844 } 845 846 #else 847 848 static int qemu_rbd_co_flush(BlockDriverState *bs) 849 { 850 #if LIBRBD_VERSION_CODE >= LIBRBD_VERSION(0, 1, 1) 851 /* rbd_flush added in 0.1.1 */ 852 BDRVRBDState *s = bs->opaque; 853 return rbd_flush(s->image); 854 #else 855 return 0; 856 #endif 857 } 858 #endif 859 860 static int qemu_rbd_getinfo(BlockDriverState *bs, BlockDriverInfo *bdi) 861 { 862 BDRVRBDState *s = bs->opaque; 863 rbd_image_info_t info; 864 int r; 865 866 r = rbd_stat(s->image, &info, sizeof(info)); 867 if (r < 0) { 868 return r; 869 } 870 871 bdi->cluster_size = info.obj_size; 872 return 0; 873 } 874 875 static int64_t qemu_rbd_getlength(BlockDriverState *bs) 876 { 877 BDRVRBDState *s = bs->opaque; 878 rbd_image_info_t info; 879 int r; 880 881 r = rbd_stat(s->image, &info, sizeof(info)); 882 if (r < 0) { 883 return r; 884 } 885 886 return info.size; 887 } 888 889 static int qemu_rbd_truncate(BlockDriverState *bs, int64_t offset) 890 { 891 BDRVRBDState *s = bs->opaque; 892 int r; 893 894 r = rbd_resize(s->image, offset); 895 if (r < 0) { 896 return r; 897 } 898 899 return 0; 900 } 901 902 static int qemu_rbd_snap_create(BlockDriverState *bs, 903 QEMUSnapshotInfo *sn_info) 904 { 905 BDRVRBDState *s = bs->opaque; 906 int r; 907 908 if (sn_info->name[0] == '\0') { 909 return -EINVAL; /* we need a name for rbd snapshots */ 910 } 911 912 /* 913 * rbd snapshots are using the name as the user controlled unique identifier 914 * we can't use the rbd snapid for that purpose, as it can't be set 915 */ 916 if (sn_info->id_str[0] != '\0' && 917 strcmp(sn_info->id_str, sn_info->name) != 0) { 918 return -EINVAL; 919 } 920 921 if (strlen(sn_info->name) >= sizeof(sn_info->id_str)) { 922 return -ERANGE; 923 } 924 925 r = rbd_snap_create(s->image, sn_info->name); 926 if (r < 0) { 927 error_report("failed to create snap: %s", strerror(-r)); 928 return r; 929 } 930 931 return 0; 932 } 933 934 static int qemu_rbd_snap_remove(BlockDriverState *bs, 935 const char *snapshot_id, 936 const char *snapshot_name, 937 Error **errp) 938 { 939 BDRVRBDState *s = bs->opaque; 940 int r; 941 942 if (!snapshot_name) { 943 error_setg(errp, "rbd need a valid snapshot name"); 944 return -EINVAL; 945 } 946 947 /* If snapshot_id is specified, it must be equal to name, see 948 qemu_rbd_snap_list() */ 949 if (snapshot_id && strcmp(snapshot_id, snapshot_name)) { 950 error_setg(errp, 951 "rbd do not support snapshot id, it should be NULL or " 952 "equal to snapshot name"); 953 return -EINVAL; 954 } 955 956 r = rbd_snap_remove(s->image, snapshot_name); 957 if (r < 0) { 958 error_setg_errno(errp, -r, "Failed to remove the snapshot"); 959 } 960 return r; 961 } 962 963 static int qemu_rbd_snap_rollback(BlockDriverState *bs, 964 const char *snapshot_name) 965 { 966 BDRVRBDState *s = bs->opaque; 967 968 return rbd_snap_rollback(s->image, snapshot_name); 969 } 970 971 static int qemu_rbd_snap_list(BlockDriverState *bs, 972 QEMUSnapshotInfo **psn_tab) 973 { 974 BDRVRBDState *s = bs->opaque; 975 QEMUSnapshotInfo *sn_info, *sn_tab = NULL; 976 int i, snap_count; 977 rbd_snap_info_t *snaps; 978 int max_snaps = RBD_MAX_SNAPS; 979 980 do { 981 snaps = g_new(rbd_snap_info_t, max_snaps); 982 snap_count = rbd_snap_list(s->image, snaps, &max_snaps); 983 if (snap_count <= 0) { 984 g_free(snaps); 985 } 986 } while (snap_count == -ERANGE); 987 988 if (snap_count <= 0) { 989 goto done; 990 } 991 992 sn_tab = g_new0(QEMUSnapshotInfo, snap_count); 993 994 for (i = 0; i < snap_count; i++) { 995 const char *snap_name = snaps[i].name; 996 997 sn_info = sn_tab + i; 998 pstrcpy(sn_info->id_str, sizeof(sn_info->id_str), snap_name); 999 pstrcpy(sn_info->name, sizeof(sn_info->name), snap_name); 1000 1001 sn_info->vm_state_size = snaps[i].size; 1002 sn_info->date_sec = 0; 1003 sn_info->date_nsec = 0; 1004 sn_info->vm_clock_nsec = 0; 1005 } 1006 rbd_snap_list_end(snaps); 1007 g_free(snaps); 1008 1009 done: 1010 *psn_tab = sn_tab; 1011 return snap_count; 1012 } 1013 1014 #ifdef LIBRBD_SUPPORTS_DISCARD 1015 static BlockAIOCB *qemu_rbd_aio_pdiscard(BlockDriverState *bs, 1016 int64_t offset, 1017 int count, 1018 BlockCompletionFunc *cb, 1019 void *opaque) 1020 { 1021 return rbd_start_aio(bs, offset, NULL, count, cb, opaque, 1022 RBD_AIO_DISCARD); 1023 } 1024 #endif 1025 1026 #ifdef LIBRBD_SUPPORTS_INVALIDATE 1027 static void qemu_rbd_invalidate_cache(BlockDriverState *bs, 1028 Error **errp) 1029 { 1030 BDRVRBDState *s = bs->opaque; 1031 int r = rbd_invalidate_cache(s->image); 1032 if (r < 0) { 1033 error_setg_errno(errp, -r, "Failed to invalidate the cache"); 1034 } 1035 } 1036 #endif 1037 1038 static QemuOptsList qemu_rbd_create_opts = { 1039 .name = "rbd-create-opts", 1040 .head = QTAILQ_HEAD_INITIALIZER(qemu_rbd_create_opts.head), 1041 .desc = { 1042 { 1043 .name = BLOCK_OPT_SIZE, 1044 .type = QEMU_OPT_SIZE, 1045 .help = "Virtual disk size" 1046 }, 1047 { 1048 .name = BLOCK_OPT_CLUSTER_SIZE, 1049 .type = QEMU_OPT_SIZE, 1050 .help = "RBD object size" 1051 }, 1052 { 1053 .name = "password-secret", 1054 .type = QEMU_OPT_STRING, 1055 .help = "ID of secret providing the password", 1056 }, 1057 { /* end of list */ } 1058 } 1059 }; 1060 1061 static BlockDriver bdrv_rbd = { 1062 .format_name = "rbd", 1063 .instance_size = sizeof(BDRVRBDState), 1064 .bdrv_parse_filename = qemu_rbd_parse_filename, 1065 .bdrv_file_open = qemu_rbd_open, 1066 .bdrv_close = qemu_rbd_close, 1067 .bdrv_create = qemu_rbd_create, 1068 .bdrv_has_zero_init = bdrv_has_zero_init_1, 1069 .bdrv_get_info = qemu_rbd_getinfo, 1070 .create_opts = &qemu_rbd_create_opts, 1071 .bdrv_getlength = qemu_rbd_getlength, 1072 .bdrv_truncate = qemu_rbd_truncate, 1073 .protocol_name = "rbd", 1074 1075 .bdrv_aio_readv = qemu_rbd_aio_readv, 1076 .bdrv_aio_writev = qemu_rbd_aio_writev, 1077 1078 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH 1079 .bdrv_aio_flush = qemu_rbd_aio_flush, 1080 #else 1081 .bdrv_co_flush_to_disk = qemu_rbd_co_flush, 1082 #endif 1083 1084 #ifdef LIBRBD_SUPPORTS_DISCARD 1085 .bdrv_aio_pdiscard = qemu_rbd_aio_pdiscard, 1086 #endif 1087 1088 .bdrv_snapshot_create = qemu_rbd_snap_create, 1089 .bdrv_snapshot_delete = qemu_rbd_snap_remove, 1090 .bdrv_snapshot_list = qemu_rbd_snap_list, 1091 .bdrv_snapshot_goto = qemu_rbd_snap_rollback, 1092 #ifdef LIBRBD_SUPPORTS_INVALIDATE 1093 .bdrv_invalidate_cache = qemu_rbd_invalidate_cache, 1094 #endif 1095 }; 1096 1097 static void bdrv_rbd_init(void) 1098 { 1099 bdrv_register(&bdrv_rbd); 1100 } 1101 1102 block_init(bdrv_rbd_init); 1103