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