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 "qemu/module.h" 20 #include "qemu/option.h" 21 #include "block/block_int.h" 22 #include "block/qdict.h" 23 #include "crypto/secret.h" 24 #include "qemu/cutils.h" 25 #include "sysemu/replay.h" 26 #include "qapi/qmp/qstring.h" 27 #include "qapi/qmp/qdict.h" 28 #include "qapi/qmp/qjson.h" 29 #include "qapi/qmp/qlist.h" 30 #include "qapi/qobject-input-visitor.h" 31 #include "qapi/qapi-visit-block-core.h" 32 33 /* 34 * When specifying the image filename use: 35 * 36 * rbd:poolname/devicename[@snapshotname][:option1=value1[:option2=value2...]] 37 * 38 * poolname must be the name of an existing rados pool. 39 * 40 * devicename is the name of the rbd image. 41 * 42 * Each option given is used to configure rados, and may be any valid 43 * Ceph option, "id", or "conf". 44 * 45 * The "id" option indicates what user we should authenticate as to 46 * the Ceph cluster. If it is excluded we will use the Ceph default 47 * (normally 'admin'). 48 * 49 * The "conf" option specifies a Ceph configuration file to read. If 50 * it is not specified, we will read from the default Ceph locations 51 * (e.g., /etc/ceph/ceph.conf). To avoid reading _any_ configuration 52 * file, specify conf=/dev/null. 53 * 54 * Configuration values containing :, @, or = can be escaped with a 55 * leading "\". 56 */ 57 58 /* rbd_aio_discard added in 0.1.2 */ 59 #if LIBRBD_VERSION_CODE >= LIBRBD_VERSION(0, 1, 2) 60 #define LIBRBD_SUPPORTS_DISCARD 61 #else 62 #undef LIBRBD_SUPPORTS_DISCARD 63 #endif 64 65 #define OBJ_MAX_SIZE (1UL << OBJ_DEFAULT_OBJ_ORDER) 66 67 #define RBD_MAX_SNAPS 100 68 69 /* The LIBRBD_SUPPORTS_IOVEC is defined in librbd.h */ 70 #ifdef LIBRBD_SUPPORTS_IOVEC 71 #define LIBRBD_USE_IOVEC 1 72 #else 73 #define LIBRBD_USE_IOVEC 0 74 #endif 75 76 #define RBD_ENCRYPTION_LUKS_HEADER_VERIFICATION_LEN 8 77 78 static const char rbd_luks_header_verification[ 79 RBD_ENCRYPTION_LUKS_HEADER_VERIFICATION_LEN] = { 80 'L', 'U', 'K', 'S', 0xBA, 0xBE, 0, 1 81 }; 82 83 static const char rbd_luks2_header_verification[ 84 RBD_ENCRYPTION_LUKS_HEADER_VERIFICATION_LEN] = { 85 'L', 'U', 'K', 'S', 0xBA, 0xBE, 0, 2 86 }; 87 88 typedef enum { 89 RBD_AIO_READ, 90 RBD_AIO_WRITE, 91 RBD_AIO_DISCARD, 92 RBD_AIO_FLUSH 93 } RBDAIOCmd; 94 95 typedef struct RBDAIOCB { 96 BlockAIOCB common; 97 int64_t ret; 98 QEMUIOVector *qiov; 99 char *bounce; 100 RBDAIOCmd cmd; 101 int error; 102 struct BDRVRBDState *s; 103 } RBDAIOCB; 104 105 typedef struct RADOSCB { 106 RBDAIOCB *acb; 107 struct BDRVRBDState *s; 108 int64_t size; 109 char *buf; 110 int64_t ret; 111 } RADOSCB; 112 113 typedef struct BDRVRBDState { 114 rados_t cluster; 115 rados_ioctx_t io_ctx; 116 rbd_image_t image; 117 char *image_name; 118 char *snap; 119 char *namespace; 120 uint64_t image_size; 121 } BDRVRBDState; 122 123 static int qemu_rbd_connect(rados_t *cluster, rados_ioctx_t *io_ctx, 124 BlockdevOptionsRbd *opts, bool cache, 125 const char *keypairs, const char *secretid, 126 Error **errp); 127 128 static char *qemu_rbd_strchr(char *src, char delim) 129 { 130 char *p; 131 132 for (p = src; *p; ++p) { 133 if (*p == delim) { 134 return p; 135 } 136 if (*p == '\\' && p[1] != '\0') { 137 ++p; 138 } 139 } 140 141 return NULL; 142 } 143 144 145 static char *qemu_rbd_next_tok(char *src, char delim, char **p) 146 { 147 char *end; 148 149 *p = NULL; 150 151 end = qemu_rbd_strchr(src, delim); 152 if (end) { 153 *p = end + 1; 154 *end = '\0'; 155 } 156 return src; 157 } 158 159 static void qemu_rbd_unescape(char *src) 160 { 161 char *p; 162 163 for (p = src; *src; ++src, ++p) { 164 if (*src == '\\' && src[1] != '\0') { 165 src++; 166 } 167 *p = *src; 168 } 169 *p = '\0'; 170 } 171 172 static void qemu_rbd_parse_filename(const char *filename, QDict *options, 173 Error **errp) 174 { 175 const char *start; 176 char *p, *buf; 177 QList *keypairs = NULL; 178 char *found_str, *image_name; 179 180 if (!strstart(filename, "rbd:", &start)) { 181 error_setg(errp, "File name must start with 'rbd:'"); 182 return; 183 } 184 185 buf = g_strdup(start); 186 p = buf; 187 188 found_str = qemu_rbd_next_tok(p, '/', &p); 189 if (!p) { 190 error_setg(errp, "Pool name is required"); 191 goto done; 192 } 193 qemu_rbd_unescape(found_str); 194 qdict_put_str(options, "pool", found_str); 195 196 if (qemu_rbd_strchr(p, '@')) { 197 image_name = qemu_rbd_next_tok(p, '@', &p); 198 199 found_str = qemu_rbd_next_tok(p, ':', &p); 200 qemu_rbd_unescape(found_str); 201 qdict_put_str(options, "snapshot", found_str); 202 } else { 203 image_name = qemu_rbd_next_tok(p, ':', &p); 204 } 205 /* Check for namespace in the image_name */ 206 if (qemu_rbd_strchr(image_name, '/')) { 207 found_str = qemu_rbd_next_tok(image_name, '/', &image_name); 208 qemu_rbd_unescape(found_str); 209 qdict_put_str(options, "namespace", found_str); 210 } else { 211 qdict_put_str(options, "namespace", ""); 212 } 213 qemu_rbd_unescape(image_name); 214 qdict_put_str(options, "image", image_name); 215 if (!p) { 216 goto done; 217 } 218 219 /* The following are essentially all key/value pairs, and we treat 220 * 'id' and 'conf' a bit special. Key/value pairs may be in any order. */ 221 while (p) { 222 char *name, *value; 223 name = qemu_rbd_next_tok(p, '=', &p); 224 if (!p) { 225 error_setg(errp, "conf option %s has no value", name); 226 break; 227 } 228 229 qemu_rbd_unescape(name); 230 231 value = qemu_rbd_next_tok(p, ':', &p); 232 qemu_rbd_unescape(value); 233 234 if (!strcmp(name, "conf")) { 235 qdict_put_str(options, "conf", value); 236 } else if (!strcmp(name, "id")) { 237 qdict_put_str(options, "user", value); 238 } else { 239 /* 240 * We pass these internally to qemu_rbd_set_keypairs(), so 241 * we can get away with the simpler list of [ "key1", 242 * "value1", "key2", "value2" ] rather than a raw dict 243 * { "key1": "value1", "key2": "value2" } where we can't 244 * guarantee order, or even a more correct but complex 245 * [ { "key1": "value1" }, { "key2": "value2" } ] 246 */ 247 if (!keypairs) { 248 keypairs = qlist_new(); 249 } 250 qlist_append_str(keypairs, name); 251 qlist_append_str(keypairs, value); 252 } 253 } 254 255 if (keypairs) { 256 qdict_put(options, "=keyvalue-pairs", 257 qstring_from_gstring(qobject_to_json(QOBJECT(keypairs)))); 258 } 259 260 done: 261 g_free(buf); 262 qobject_unref(keypairs); 263 return; 264 } 265 266 267 static void qemu_rbd_refresh_limits(BlockDriverState *bs, Error **errp) 268 { 269 /* XXX Does RBD support AIO on less than 512-byte alignment? */ 270 bs->bl.request_alignment = 512; 271 } 272 273 274 static int qemu_rbd_set_auth(rados_t cluster, BlockdevOptionsRbd *opts, 275 Error **errp) 276 { 277 char *key, *acr; 278 int r; 279 GString *accu; 280 RbdAuthModeList *auth; 281 282 if (opts->key_secret) { 283 key = qcrypto_secret_lookup_as_base64(opts->key_secret, errp); 284 if (!key) { 285 return -EIO; 286 } 287 r = rados_conf_set(cluster, "key", key); 288 g_free(key); 289 if (r < 0) { 290 error_setg_errno(errp, -r, "Could not set 'key'"); 291 return r; 292 } 293 } 294 295 if (opts->has_auth_client_required) { 296 accu = g_string_new(""); 297 for (auth = opts->auth_client_required; auth; auth = auth->next) { 298 if (accu->str[0]) { 299 g_string_append_c(accu, ';'); 300 } 301 g_string_append(accu, RbdAuthMode_str(auth->value)); 302 } 303 acr = g_string_free(accu, FALSE); 304 r = rados_conf_set(cluster, "auth_client_required", acr); 305 g_free(acr); 306 if (r < 0) { 307 error_setg_errno(errp, -r, 308 "Could not set 'auth_client_required'"); 309 return r; 310 } 311 } 312 313 return 0; 314 } 315 316 static int qemu_rbd_set_keypairs(rados_t cluster, const char *keypairs_json, 317 Error **errp) 318 { 319 QList *keypairs; 320 QString *name; 321 QString *value; 322 const char *key; 323 size_t remaining; 324 int ret = 0; 325 326 if (!keypairs_json) { 327 return ret; 328 } 329 keypairs = qobject_to(QList, 330 qobject_from_json(keypairs_json, &error_abort)); 331 remaining = qlist_size(keypairs) / 2; 332 assert(remaining); 333 334 while (remaining--) { 335 name = qobject_to(QString, qlist_pop(keypairs)); 336 value = qobject_to(QString, qlist_pop(keypairs)); 337 assert(name && value); 338 key = qstring_get_str(name); 339 340 ret = rados_conf_set(cluster, key, qstring_get_str(value)); 341 qobject_unref(value); 342 if (ret < 0) { 343 error_setg_errno(errp, -ret, "invalid conf option %s", key); 344 qobject_unref(name); 345 ret = -EINVAL; 346 break; 347 } 348 qobject_unref(name); 349 } 350 351 qobject_unref(keypairs); 352 return ret; 353 } 354 355 static void qemu_rbd_memset(RADOSCB *rcb, int64_t offs) 356 { 357 if (LIBRBD_USE_IOVEC) { 358 RBDAIOCB *acb = rcb->acb; 359 iov_memset(acb->qiov->iov, acb->qiov->niov, offs, 0, 360 acb->qiov->size - offs); 361 } else { 362 memset(rcb->buf + offs, 0, rcb->size - offs); 363 } 364 } 365 366 #ifdef LIBRBD_SUPPORTS_ENCRYPTION 367 static int qemu_rbd_convert_luks_options( 368 RbdEncryptionOptionsLUKSBase *luks_opts, 369 char **passphrase, 370 size_t *passphrase_len, 371 Error **errp) 372 { 373 return qcrypto_secret_lookup(luks_opts->key_secret, (uint8_t **)passphrase, 374 passphrase_len, errp); 375 } 376 377 static int qemu_rbd_convert_luks_create_options( 378 RbdEncryptionCreateOptionsLUKSBase *luks_opts, 379 rbd_encryption_algorithm_t *alg, 380 char **passphrase, 381 size_t *passphrase_len, 382 Error **errp) 383 { 384 int r = 0; 385 386 r = qemu_rbd_convert_luks_options( 387 qapi_RbdEncryptionCreateOptionsLUKSBase_base(luks_opts), 388 passphrase, passphrase_len, errp); 389 if (r < 0) { 390 return r; 391 } 392 393 if (luks_opts->has_cipher_alg) { 394 switch (luks_opts->cipher_alg) { 395 case QCRYPTO_CIPHER_ALG_AES_128: { 396 *alg = RBD_ENCRYPTION_ALGORITHM_AES128; 397 break; 398 } 399 case QCRYPTO_CIPHER_ALG_AES_256: { 400 *alg = RBD_ENCRYPTION_ALGORITHM_AES256; 401 break; 402 } 403 default: { 404 r = -ENOTSUP; 405 error_setg_errno(errp, -r, "unknown encryption algorithm: %u", 406 luks_opts->cipher_alg); 407 return r; 408 } 409 } 410 } else { 411 /* default alg */ 412 *alg = RBD_ENCRYPTION_ALGORITHM_AES256; 413 } 414 415 return 0; 416 } 417 418 static int qemu_rbd_encryption_format(rbd_image_t image, 419 RbdEncryptionCreateOptions *encrypt, 420 Error **errp) 421 { 422 int r = 0; 423 g_autofree char *passphrase = NULL; 424 size_t passphrase_len; 425 rbd_encryption_format_t format; 426 rbd_encryption_options_t opts; 427 rbd_encryption_luks1_format_options_t luks_opts; 428 rbd_encryption_luks2_format_options_t luks2_opts; 429 size_t opts_size; 430 uint64_t raw_size, effective_size; 431 432 r = rbd_get_size(image, &raw_size); 433 if (r < 0) { 434 error_setg_errno(errp, -r, "cannot get raw image size"); 435 return r; 436 } 437 438 switch (encrypt->format) { 439 case RBD_IMAGE_ENCRYPTION_FORMAT_LUKS: { 440 memset(&luks_opts, 0, sizeof(luks_opts)); 441 format = RBD_ENCRYPTION_FORMAT_LUKS1; 442 opts = &luks_opts; 443 opts_size = sizeof(luks_opts); 444 r = qemu_rbd_convert_luks_create_options( 445 qapi_RbdEncryptionCreateOptionsLUKS_base(&encrypt->u.luks), 446 &luks_opts.alg, &passphrase, &passphrase_len, errp); 447 if (r < 0) { 448 return r; 449 } 450 luks_opts.passphrase = passphrase; 451 luks_opts.passphrase_size = passphrase_len; 452 break; 453 } 454 case RBD_IMAGE_ENCRYPTION_FORMAT_LUKS2: { 455 memset(&luks2_opts, 0, sizeof(luks2_opts)); 456 format = RBD_ENCRYPTION_FORMAT_LUKS2; 457 opts = &luks2_opts; 458 opts_size = sizeof(luks2_opts); 459 r = qemu_rbd_convert_luks_create_options( 460 qapi_RbdEncryptionCreateOptionsLUKS2_base( 461 &encrypt->u.luks2), 462 &luks2_opts.alg, &passphrase, &passphrase_len, errp); 463 if (r < 0) { 464 return r; 465 } 466 luks2_opts.passphrase = passphrase; 467 luks2_opts.passphrase_size = passphrase_len; 468 break; 469 } 470 default: { 471 r = -ENOTSUP; 472 error_setg_errno( 473 errp, -r, "unknown image encryption format: %u", 474 encrypt->format); 475 return r; 476 } 477 } 478 479 r = rbd_encryption_format(image, format, opts, opts_size); 480 if (r < 0) { 481 error_setg_errno(errp, -r, "encryption format fail"); 482 return r; 483 } 484 485 r = rbd_get_size(image, &effective_size); 486 if (r < 0) { 487 error_setg_errno(errp, -r, "cannot get effective image size"); 488 return r; 489 } 490 491 r = rbd_resize(image, raw_size + (raw_size - effective_size)); 492 if (r < 0) { 493 error_setg_errno(errp, -r, "cannot resize image after format"); 494 return r; 495 } 496 497 return 0; 498 } 499 500 static int qemu_rbd_encryption_load(rbd_image_t image, 501 RbdEncryptionOptions *encrypt, 502 Error **errp) 503 { 504 int r = 0; 505 g_autofree char *passphrase = NULL; 506 size_t passphrase_len; 507 rbd_encryption_luks1_format_options_t luks_opts; 508 rbd_encryption_luks2_format_options_t luks2_opts; 509 rbd_encryption_format_t format; 510 rbd_encryption_options_t opts; 511 size_t opts_size; 512 513 switch (encrypt->format) { 514 case RBD_IMAGE_ENCRYPTION_FORMAT_LUKS: { 515 memset(&luks_opts, 0, sizeof(luks_opts)); 516 format = RBD_ENCRYPTION_FORMAT_LUKS1; 517 opts = &luks_opts; 518 opts_size = sizeof(luks_opts); 519 r = qemu_rbd_convert_luks_options( 520 qapi_RbdEncryptionOptionsLUKS_base(&encrypt->u.luks), 521 &passphrase, &passphrase_len, errp); 522 if (r < 0) { 523 return r; 524 } 525 luks_opts.passphrase = passphrase; 526 luks_opts.passphrase_size = passphrase_len; 527 break; 528 } 529 case RBD_IMAGE_ENCRYPTION_FORMAT_LUKS2: { 530 memset(&luks2_opts, 0, sizeof(luks2_opts)); 531 format = RBD_ENCRYPTION_FORMAT_LUKS2; 532 opts = &luks2_opts; 533 opts_size = sizeof(luks2_opts); 534 r = qemu_rbd_convert_luks_options( 535 qapi_RbdEncryptionOptionsLUKS2_base(&encrypt->u.luks2), 536 &passphrase, &passphrase_len, errp); 537 if (r < 0) { 538 return r; 539 } 540 luks2_opts.passphrase = passphrase; 541 luks2_opts.passphrase_size = passphrase_len; 542 break; 543 } 544 default: { 545 r = -ENOTSUP; 546 error_setg_errno( 547 errp, -r, "unknown image encryption format: %u", 548 encrypt->format); 549 return r; 550 } 551 } 552 553 r = rbd_encryption_load(image, format, opts, opts_size); 554 if (r < 0) { 555 error_setg_errno(errp, -r, "encryption load fail"); 556 return r; 557 } 558 559 return 0; 560 } 561 #endif 562 563 /* FIXME Deprecate and remove keypairs or make it available in QMP. */ 564 static int qemu_rbd_do_create(BlockdevCreateOptions *options, 565 const char *keypairs, const char *password_secret, 566 Error **errp) 567 { 568 BlockdevCreateOptionsRbd *opts = &options->u.rbd; 569 rados_t cluster; 570 rados_ioctx_t io_ctx; 571 int obj_order = 0; 572 int ret; 573 574 assert(options->driver == BLOCKDEV_DRIVER_RBD); 575 if (opts->location->has_snapshot) { 576 error_setg(errp, "Can't use snapshot name for image creation"); 577 return -EINVAL; 578 } 579 580 #ifndef LIBRBD_SUPPORTS_ENCRYPTION 581 if (opts->has_encrypt) { 582 error_setg(errp, "RBD library does not support image encryption"); 583 return -ENOTSUP; 584 } 585 #endif 586 587 if (opts->has_cluster_size) { 588 int64_t objsize = opts->cluster_size; 589 if ((objsize - 1) & objsize) { /* not a power of 2? */ 590 error_setg(errp, "obj size needs to be power of 2"); 591 return -EINVAL; 592 } 593 if (objsize < 4096) { 594 error_setg(errp, "obj size too small"); 595 return -EINVAL; 596 } 597 obj_order = ctz32(objsize); 598 } 599 600 ret = qemu_rbd_connect(&cluster, &io_ctx, opts->location, false, keypairs, 601 password_secret, errp); 602 if (ret < 0) { 603 return ret; 604 } 605 606 ret = rbd_create(io_ctx, opts->location->image, opts->size, &obj_order); 607 if (ret < 0) { 608 error_setg_errno(errp, -ret, "error rbd create"); 609 goto out; 610 } 611 612 #ifdef LIBRBD_SUPPORTS_ENCRYPTION 613 if (opts->has_encrypt) { 614 rbd_image_t image; 615 616 ret = rbd_open(io_ctx, opts->location->image, &image, NULL); 617 if (ret < 0) { 618 error_setg_errno(errp, -ret, 619 "error opening image '%s' for encryption format", 620 opts->location->image); 621 goto out; 622 } 623 624 ret = qemu_rbd_encryption_format(image, opts->encrypt, errp); 625 rbd_close(image); 626 if (ret < 0) { 627 /* encryption format fail, try removing the image */ 628 rbd_remove(io_ctx, opts->location->image); 629 goto out; 630 } 631 } 632 #endif 633 634 ret = 0; 635 out: 636 rados_ioctx_destroy(io_ctx); 637 rados_shutdown(cluster); 638 return ret; 639 } 640 641 static int qemu_rbd_co_create(BlockdevCreateOptions *options, Error **errp) 642 { 643 return qemu_rbd_do_create(options, NULL, NULL, errp); 644 } 645 646 static int qemu_rbd_extract_encryption_create_options( 647 QemuOpts *opts, 648 RbdEncryptionCreateOptions **spec, 649 Error **errp) 650 { 651 QDict *opts_qdict; 652 QDict *encrypt_qdict; 653 Visitor *v; 654 int ret = 0; 655 656 opts_qdict = qemu_opts_to_qdict(opts, NULL); 657 qdict_extract_subqdict(opts_qdict, &encrypt_qdict, "encrypt."); 658 qobject_unref(opts_qdict); 659 if (!qdict_size(encrypt_qdict)) { 660 *spec = NULL; 661 goto exit; 662 } 663 664 /* Convert options into a QAPI object */ 665 v = qobject_input_visitor_new_flat_confused(encrypt_qdict, errp); 666 if (!v) { 667 ret = -EINVAL; 668 goto exit; 669 } 670 671 visit_type_RbdEncryptionCreateOptions(v, NULL, spec, errp); 672 visit_free(v); 673 if (!*spec) { 674 ret = -EINVAL; 675 goto exit; 676 } 677 678 exit: 679 qobject_unref(encrypt_qdict); 680 return ret; 681 } 682 683 static int coroutine_fn qemu_rbd_co_create_opts(BlockDriver *drv, 684 const char *filename, 685 QemuOpts *opts, 686 Error **errp) 687 { 688 BlockdevCreateOptions *create_options; 689 BlockdevCreateOptionsRbd *rbd_opts; 690 BlockdevOptionsRbd *loc; 691 RbdEncryptionCreateOptions *encrypt = NULL; 692 Error *local_err = NULL; 693 const char *keypairs, *password_secret; 694 QDict *options = NULL; 695 int ret = 0; 696 697 create_options = g_new0(BlockdevCreateOptions, 1); 698 create_options->driver = BLOCKDEV_DRIVER_RBD; 699 rbd_opts = &create_options->u.rbd; 700 701 rbd_opts->location = g_new0(BlockdevOptionsRbd, 1); 702 703 password_secret = qemu_opt_get(opts, "password-secret"); 704 705 /* Read out options */ 706 rbd_opts->size = ROUND_UP(qemu_opt_get_size_del(opts, BLOCK_OPT_SIZE, 0), 707 BDRV_SECTOR_SIZE); 708 rbd_opts->cluster_size = qemu_opt_get_size_del(opts, 709 BLOCK_OPT_CLUSTER_SIZE, 0); 710 rbd_opts->has_cluster_size = (rbd_opts->cluster_size != 0); 711 712 options = qdict_new(); 713 qemu_rbd_parse_filename(filename, options, &local_err); 714 if (local_err) { 715 ret = -EINVAL; 716 error_propagate(errp, local_err); 717 goto exit; 718 } 719 720 ret = qemu_rbd_extract_encryption_create_options(opts, &encrypt, errp); 721 if (ret < 0) { 722 goto exit; 723 } 724 rbd_opts->encrypt = encrypt; 725 rbd_opts->has_encrypt = !!encrypt; 726 727 /* 728 * Caution: while qdict_get_try_str() is fine, getting non-string 729 * types would require more care. When @options come from -blockdev 730 * or blockdev_add, its members are typed according to the QAPI 731 * schema, but when they come from -drive, they're all QString. 732 */ 733 loc = rbd_opts->location; 734 loc->pool = g_strdup(qdict_get_try_str(options, "pool")); 735 loc->conf = g_strdup(qdict_get_try_str(options, "conf")); 736 loc->has_conf = !!loc->conf; 737 loc->user = g_strdup(qdict_get_try_str(options, "user")); 738 loc->has_user = !!loc->user; 739 loc->q_namespace = g_strdup(qdict_get_try_str(options, "namespace")); 740 loc->has_q_namespace = !!loc->q_namespace; 741 loc->image = g_strdup(qdict_get_try_str(options, "image")); 742 keypairs = qdict_get_try_str(options, "=keyvalue-pairs"); 743 744 ret = qemu_rbd_do_create(create_options, keypairs, password_secret, errp); 745 if (ret < 0) { 746 goto exit; 747 } 748 749 exit: 750 qobject_unref(options); 751 qapi_free_BlockdevCreateOptions(create_options); 752 return ret; 753 } 754 755 /* 756 * This aio completion is being called from rbd_finish_bh() and runs in qemu 757 * BH context. 758 */ 759 static void qemu_rbd_complete_aio(RADOSCB *rcb) 760 { 761 RBDAIOCB *acb = rcb->acb; 762 int64_t r; 763 764 r = rcb->ret; 765 766 if (acb->cmd != RBD_AIO_READ) { 767 if (r < 0) { 768 acb->ret = r; 769 acb->error = 1; 770 } else if (!acb->error) { 771 acb->ret = rcb->size; 772 } 773 } else { 774 if (r < 0) { 775 qemu_rbd_memset(rcb, 0); 776 acb->ret = r; 777 acb->error = 1; 778 } else if (r < rcb->size) { 779 qemu_rbd_memset(rcb, r); 780 if (!acb->error) { 781 acb->ret = rcb->size; 782 } 783 } else if (!acb->error) { 784 acb->ret = r; 785 } 786 } 787 788 g_free(rcb); 789 790 if (!LIBRBD_USE_IOVEC) { 791 if (acb->cmd == RBD_AIO_READ) { 792 qemu_iovec_from_buf(acb->qiov, 0, acb->bounce, acb->qiov->size); 793 } 794 qemu_vfree(acb->bounce); 795 } 796 797 acb->common.cb(acb->common.opaque, (acb->ret > 0 ? 0 : acb->ret)); 798 799 qemu_aio_unref(acb); 800 } 801 802 static char *qemu_rbd_mon_host(BlockdevOptionsRbd *opts, Error **errp) 803 { 804 const char **vals; 805 const char *host, *port; 806 char *rados_str; 807 InetSocketAddressBaseList *p; 808 int i, cnt; 809 810 if (!opts->has_server) { 811 return NULL; 812 } 813 814 for (cnt = 0, p = opts->server; p; p = p->next) { 815 cnt++; 816 } 817 818 vals = g_new(const char *, cnt + 1); 819 820 for (i = 0, p = opts->server; p; p = p->next, i++) { 821 host = p->value->host; 822 port = p->value->port; 823 824 if (strchr(host, ':')) { 825 vals[i] = g_strdup_printf("[%s]:%s", host, port); 826 } else { 827 vals[i] = g_strdup_printf("%s:%s", host, port); 828 } 829 } 830 vals[i] = NULL; 831 832 rados_str = i ? g_strjoinv(";", (char **)vals) : NULL; 833 g_strfreev((char **)vals); 834 return rados_str; 835 } 836 837 static int qemu_rbd_connect(rados_t *cluster, rados_ioctx_t *io_ctx, 838 BlockdevOptionsRbd *opts, bool cache, 839 const char *keypairs, const char *secretid, 840 Error **errp) 841 { 842 char *mon_host = NULL; 843 Error *local_err = NULL; 844 int r; 845 846 if (secretid) { 847 if (opts->key_secret) { 848 error_setg(errp, 849 "Legacy 'password-secret' clashes with 'key-secret'"); 850 return -EINVAL; 851 } 852 opts->key_secret = g_strdup(secretid); 853 opts->has_key_secret = true; 854 } 855 856 mon_host = qemu_rbd_mon_host(opts, &local_err); 857 if (local_err) { 858 error_propagate(errp, local_err); 859 r = -EINVAL; 860 goto out; 861 } 862 863 r = rados_create(cluster, opts->user); 864 if (r < 0) { 865 error_setg_errno(errp, -r, "error initializing"); 866 goto out; 867 } 868 869 /* try default location when conf=NULL, but ignore failure */ 870 r = rados_conf_read_file(*cluster, opts->conf); 871 if (opts->has_conf && r < 0) { 872 error_setg_errno(errp, -r, "error reading conf file %s", opts->conf); 873 goto failed_shutdown; 874 } 875 876 r = qemu_rbd_set_keypairs(*cluster, keypairs, errp); 877 if (r < 0) { 878 goto failed_shutdown; 879 } 880 881 if (mon_host) { 882 r = rados_conf_set(*cluster, "mon_host", mon_host); 883 if (r < 0) { 884 goto failed_shutdown; 885 } 886 } 887 888 r = qemu_rbd_set_auth(*cluster, opts, errp); 889 if (r < 0) { 890 goto failed_shutdown; 891 } 892 893 /* 894 * Fallback to more conservative semantics if setting cache 895 * options fails. Ignore errors from setting rbd_cache because the 896 * only possible error is that the option does not exist, and 897 * librbd defaults to no caching. If write through caching cannot 898 * be set up, fall back to no caching. 899 */ 900 if (cache) { 901 rados_conf_set(*cluster, "rbd_cache", "true"); 902 } else { 903 rados_conf_set(*cluster, "rbd_cache", "false"); 904 } 905 906 r = rados_connect(*cluster); 907 if (r < 0) { 908 error_setg_errno(errp, -r, "error connecting"); 909 goto failed_shutdown; 910 } 911 912 r = rados_ioctx_create(*cluster, opts->pool, io_ctx); 913 if (r < 0) { 914 error_setg_errno(errp, -r, "error opening pool %s", opts->pool); 915 goto failed_shutdown; 916 } 917 /* 918 * Set the namespace after opening the io context on the pool, 919 * if nspace == NULL or if nspace == "", it is just as we did nothing 920 */ 921 rados_ioctx_set_namespace(*io_ctx, opts->q_namespace); 922 923 r = 0; 924 goto out; 925 926 failed_shutdown: 927 rados_shutdown(*cluster); 928 out: 929 g_free(mon_host); 930 return r; 931 } 932 933 static int qemu_rbd_convert_options(QDict *options, BlockdevOptionsRbd **opts, 934 Error **errp) 935 { 936 Visitor *v; 937 938 /* Convert the remaining options into a QAPI object */ 939 v = qobject_input_visitor_new_flat_confused(options, errp); 940 if (!v) { 941 return -EINVAL; 942 } 943 944 visit_type_BlockdevOptionsRbd(v, NULL, opts, errp); 945 visit_free(v); 946 if (!opts) { 947 return -EINVAL; 948 } 949 950 return 0; 951 } 952 953 static int qemu_rbd_attempt_legacy_options(QDict *options, 954 BlockdevOptionsRbd **opts, 955 char **keypairs) 956 { 957 char *filename; 958 int r; 959 960 filename = g_strdup(qdict_get_try_str(options, "filename")); 961 if (!filename) { 962 return -EINVAL; 963 } 964 qdict_del(options, "filename"); 965 966 qemu_rbd_parse_filename(filename, options, NULL); 967 968 /* keypairs freed by caller */ 969 *keypairs = g_strdup(qdict_get_try_str(options, "=keyvalue-pairs")); 970 if (*keypairs) { 971 qdict_del(options, "=keyvalue-pairs"); 972 } 973 974 r = qemu_rbd_convert_options(options, opts, NULL); 975 976 g_free(filename); 977 return r; 978 } 979 980 static int qemu_rbd_open(BlockDriverState *bs, QDict *options, int flags, 981 Error **errp) 982 { 983 BDRVRBDState *s = bs->opaque; 984 BlockdevOptionsRbd *opts = NULL; 985 const QDictEntry *e; 986 Error *local_err = NULL; 987 char *keypairs, *secretid; 988 int r; 989 990 keypairs = g_strdup(qdict_get_try_str(options, "=keyvalue-pairs")); 991 if (keypairs) { 992 qdict_del(options, "=keyvalue-pairs"); 993 } 994 995 secretid = g_strdup(qdict_get_try_str(options, "password-secret")); 996 if (secretid) { 997 qdict_del(options, "password-secret"); 998 } 999 1000 r = qemu_rbd_convert_options(options, &opts, &local_err); 1001 if (local_err) { 1002 /* If keypairs are present, that means some options are present in 1003 * the modern option format. Don't attempt to parse legacy option 1004 * formats, as we won't support mixed usage. */ 1005 if (keypairs) { 1006 error_propagate(errp, local_err); 1007 goto out; 1008 } 1009 1010 /* If the initial attempt to convert and process the options failed, 1011 * we may be attempting to open an image file that has the rbd options 1012 * specified in the older format consisting of all key/value pairs 1013 * encoded in the filename. Go ahead and attempt to parse the 1014 * filename, and see if we can pull out the required options. */ 1015 r = qemu_rbd_attempt_legacy_options(options, &opts, &keypairs); 1016 if (r < 0) { 1017 /* Propagate the original error, not the legacy parsing fallback 1018 * error, as the latter was just a best-effort attempt. */ 1019 error_propagate(errp, local_err); 1020 goto out; 1021 } 1022 /* Take care whenever deciding to actually deprecate; once this ability 1023 * is removed, we will not be able to open any images with legacy-styled 1024 * backing image strings. */ 1025 warn_report("RBD options encoded in the filename as keyvalue pairs " 1026 "is deprecated"); 1027 } 1028 1029 /* Remove the processed options from the QDict (the visitor processes 1030 * _all_ options in the QDict) */ 1031 while ((e = qdict_first(options))) { 1032 qdict_del(options, e->key); 1033 } 1034 1035 r = qemu_rbd_connect(&s->cluster, &s->io_ctx, opts, 1036 !(flags & BDRV_O_NOCACHE), keypairs, secretid, errp); 1037 if (r < 0) { 1038 goto out; 1039 } 1040 1041 s->snap = g_strdup(opts->snapshot); 1042 s->image_name = g_strdup(opts->image); 1043 1044 /* rbd_open is always r/w */ 1045 r = rbd_open(s->io_ctx, s->image_name, &s->image, s->snap); 1046 if (r < 0) { 1047 error_setg_errno(errp, -r, "error reading header from %s", 1048 s->image_name); 1049 goto failed_open; 1050 } 1051 1052 if (opts->has_encrypt) { 1053 #ifdef LIBRBD_SUPPORTS_ENCRYPTION 1054 r = qemu_rbd_encryption_load(s->image, opts->encrypt, errp); 1055 if (r < 0) { 1056 goto failed_post_open; 1057 } 1058 #else 1059 r = -ENOTSUP; 1060 error_setg(errp, "RBD library does not support image encryption"); 1061 goto failed_post_open; 1062 #endif 1063 } 1064 1065 r = rbd_get_size(s->image, &s->image_size); 1066 if (r < 0) { 1067 error_setg_errno(errp, -r, "error getting image size from %s", 1068 s->image_name); 1069 goto failed_post_open; 1070 } 1071 1072 /* If we are using an rbd snapshot, we must be r/o, otherwise 1073 * leave as-is */ 1074 if (s->snap != NULL) { 1075 r = bdrv_apply_auto_read_only(bs, "rbd snapshots are read-only", errp); 1076 if (r < 0) { 1077 goto failed_post_open; 1078 } 1079 } 1080 1081 /* When extending regular files, we get zeros from the OS */ 1082 bs->supported_truncate_flags = BDRV_REQ_ZERO_WRITE; 1083 1084 r = 0; 1085 goto out; 1086 1087 failed_post_open: 1088 rbd_close(s->image); 1089 failed_open: 1090 rados_ioctx_destroy(s->io_ctx); 1091 g_free(s->snap); 1092 g_free(s->image_name); 1093 rados_shutdown(s->cluster); 1094 out: 1095 qapi_free_BlockdevOptionsRbd(opts); 1096 g_free(keypairs); 1097 g_free(secretid); 1098 return r; 1099 } 1100 1101 1102 /* Since RBD is currently always opened R/W via the API, 1103 * we just need to check if we are using a snapshot or not, in 1104 * order to determine if we will allow it to be R/W */ 1105 static int qemu_rbd_reopen_prepare(BDRVReopenState *state, 1106 BlockReopenQueue *queue, Error **errp) 1107 { 1108 BDRVRBDState *s = state->bs->opaque; 1109 int ret = 0; 1110 1111 if (s->snap && state->flags & BDRV_O_RDWR) { 1112 error_setg(errp, 1113 "Cannot change node '%s' to r/w when using RBD snapshot", 1114 bdrv_get_device_or_node_name(state->bs)); 1115 ret = -EINVAL; 1116 } 1117 1118 return ret; 1119 } 1120 1121 static void qemu_rbd_close(BlockDriverState *bs) 1122 { 1123 BDRVRBDState *s = bs->opaque; 1124 1125 rbd_close(s->image); 1126 rados_ioctx_destroy(s->io_ctx); 1127 g_free(s->snap); 1128 g_free(s->image_name); 1129 rados_shutdown(s->cluster); 1130 } 1131 1132 /* Resize the RBD image and update the 'image_size' with the current size */ 1133 static int qemu_rbd_resize(BlockDriverState *bs, uint64_t size) 1134 { 1135 BDRVRBDState *s = bs->opaque; 1136 int r; 1137 1138 r = rbd_resize(s->image, size); 1139 if (r < 0) { 1140 return r; 1141 } 1142 1143 s->image_size = size; 1144 1145 return 0; 1146 } 1147 1148 static const AIOCBInfo rbd_aiocb_info = { 1149 .aiocb_size = sizeof(RBDAIOCB), 1150 }; 1151 1152 static void rbd_finish_bh(void *opaque) 1153 { 1154 RADOSCB *rcb = opaque; 1155 qemu_rbd_complete_aio(rcb); 1156 } 1157 1158 /* 1159 * This is the callback function for rbd_aio_read and _write 1160 * 1161 * Note: this function is being called from a non qemu thread so 1162 * we need to be careful about what we do here. Generally we only 1163 * schedule a BH, and do the rest of the io completion handling 1164 * from rbd_finish_bh() which runs in a qemu context. 1165 */ 1166 static void rbd_finish_aiocb(rbd_completion_t c, RADOSCB *rcb) 1167 { 1168 RBDAIOCB *acb = rcb->acb; 1169 1170 rcb->ret = rbd_aio_get_return_value(c); 1171 rbd_aio_release(c); 1172 1173 replay_bh_schedule_oneshot_event(bdrv_get_aio_context(acb->common.bs), 1174 rbd_finish_bh, rcb); 1175 } 1176 1177 static int rbd_aio_discard_wrapper(rbd_image_t image, 1178 uint64_t off, 1179 uint64_t len, 1180 rbd_completion_t comp) 1181 { 1182 #ifdef LIBRBD_SUPPORTS_DISCARD 1183 return rbd_aio_discard(image, off, len, comp); 1184 #else 1185 return -ENOTSUP; 1186 #endif 1187 } 1188 1189 static int rbd_aio_flush_wrapper(rbd_image_t image, 1190 rbd_completion_t comp) 1191 { 1192 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH 1193 return rbd_aio_flush(image, comp); 1194 #else 1195 return -ENOTSUP; 1196 #endif 1197 } 1198 1199 static BlockAIOCB *rbd_start_aio(BlockDriverState *bs, 1200 int64_t off, 1201 QEMUIOVector *qiov, 1202 int64_t size, 1203 BlockCompletionFunc *cb, 1204 void *opaque, 1205 RBDAIOCmd cmd) 1206 { 1207 RBDAIOCB *acb; 1208 RADOSCB *rcb = NULL; 1209 rbd_completion_t c; 1210 int r; 1211 1212 BDRVRBDState *s = bs->opaque; 1213 1214 acb = qemu_aio_get(&rbd_aiocb_info, bs, cb, opaque); 1215 acb->cmd = cmd; 1216 acb->qiov = qiov; 1217 assert(!qiov || qiov->size == size); 1218 1219 rcb = g_new(RADOSCB, 1); 1220 1221 if (!LIBRBD_USE_IOVEC) { 1222 if (cmd == RBD_AIO_DISCARD || cmd == RBD_AIO_FLUSH) { 1223 acb->bounce = NULL; 1224 } else { 1225 acb->bounce = qemu_try_blockalign(bs, qiov->size); 1226 if (acb->bounce == NULL) { 1227 goto failed; 1228 } 1229 } 1230 if (cmd == RBD_AIO_WRITE) { 1231 qemu_iovec_to_buf(acb->qiov, 0, acb->bounce, qiov->size); 1232 } 1233 rcb->buf = acb->bounce; 1234 } 1235 1236 acb->ret = 0; 1237 acb->error = 0; 1238 acb->s = s; 1239 1240 rcb->acb = acb; 1241 rcb->s = acb->s; 1242 rcb->size = size; 1243 r = rbd_aio_create_completion(rcb, (rbd_callback_t) rbd_finish_aiocb, &c); 1244 if (r < 0) { 1245 goto failed; 1246 } 1247 1248 switch (cmd) { 1249 case RBD_AIO_WRITE: { 1250 /* 1251 * RBD APIs don't allow us to write more than actual size, so in order 1252 * to support growing images, we resize the image before write 1253 * operations that exceed the current size. 1254 */ 1255 if (off + size > s->image_size) { 1256 r = qemu_rbd_resize(bs, off + size); 1257 if (r < 0) { 1258 goto failed_completion; 1259 } 1260 } 1261 #ifdef LIBRBD_SUPPORTS_IOVEC 1262 r = rbd_aio_writev(s->image, qiov->iov, qiov->niov, off, c); 1263 #else 1264 r = rbd_aio_write(s->image, off, size, rcb->buf, c); 1265 #endif 1266 break; 1267 } 1268 case RBD_AIO_READ: 1269 #ifdef LIBRBD_SUPPORTS_IOVEC 1270 r = rbd_aio_readv(s->image, qiov->iov, qiov->niov, off, c); 1271 #else 1272 r = rbd_aio_read(s->image, off, size, rcb->buf, c); 1273 #endif 1274 break; 1275 case RBD_AIO_DISCARD: 1276 r = rbd_aio_discard_wrapper(s->image, off, size, c); 1277 break; 1278 case RBD_AIO_FLUSH: 1279 r = rbd_aio_flush_wrapper(s->image, c); 1280 break; 1281 default: 1282 r = -EINVAL; 1283 } 1284 1285 if (r < 0) { 1286 goto failed_completion; 1287 } 1288 return &acb->common; 1289 1290 failed_completion: 1291 rbd_aio_release(c); 1292 failed: 1293 g_free(rcb); 1294 if (!LIBRBD_USE_IOVEC) { 1295 qemu_vfree(acb->bounce); 1296 } 1297 1298 qemu_aio_unref(acb); 1299 return NULL; 1300 } 1301 1302 static BlockAIOCB *qemu_rbd_aio_preadv(BlockDriverState *bs, 1303 uint64_t offset, uint64_t bytes, 1304 QEMUIOVector *qiov, int flags, 1305 BlockCompletionFunc *cb, 1306 void *opaque) 1307 { 1308 return rbd_start_aio(bs, offset, qiov, bytes, cb, opaque, 1309 RBD_AIO_READ); 1310 } 1311 1312 static BlockAIOCB *qemu_rbd_aio_pwritev(BlockDriverState *bs, 1313 uint64_t offset, uint64_t bytes, 1314 QEMUIOVector *qiov, int flags, 1315 BlockCompletionFunc *cb, 1316 void *opaque) 1317 { 1318 return rbd_start_aio(bs, offset, qiov, bytes, cb, opaque, 1319 RBD_AIO_WRITE); 1320 } 1321 1322 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH 1323 static BlockAIOCB *qemu_rbd_aio_flush(BlockDriverState *bs, 1324 BlockCompletionFunc *cb, 1325 void *opaque) 1326 { 1327 return rbd_start_aio(bs, 0, NULL, 0, cb, opaque, RBD_AIO_FLUSH); 1328 } 1329 1330 #else 1331 1332 static int qemu_rbd_co_flush(BlockDriverState *bs) 1333 { 1334 #if LIBRBD_VERSION_CODE >= LIBRBD_VERSION(0, 1, 1) 1335 /* rbd_flush added in 0.1.1 */ 1336 BDRVRBDState *s = bs->opaque; 1337 return rbd_flush(s->image); 1338 #else 1339 return 0; 1340 #endif 1341 } 1342 #endif 1343 1344 static int qemu_rbd_getinfo(BlockDriverState *bs, BlockDriverInfo *bdi) 1345 { 1346 BDRVRBDState *s = bs->opaque; 1347 rbd_image_info_t info; 1348 int r; 1349 1350 r = rbd_stat(s->image, &info, sizeof(info)); 1351 if (r < 0) { 1352 return r; 1353 } 1354 1355 bdi->cluster_size = info.obj_size; 1356 return 0; 1357 } 1358 1359 static ImageInfoSpecific *qemu_rbd_get_specific_info(BlockDriverState *bs, 1360 Error **errp) 1361 { 1362 BDRVRBDState *s = bs->opaque; 1363 ImageInfoSpecific *spec_info; 1364 char buf[RBD_ENCRYPTION_LUKS_HEADER_VERIFICATION_LEN] = {0}; 1365 int r; 1366 1367 if (s->image_size >= RBD_ENCRYPTION_LUKS_HEADER_VERIFICATION_LEN) { 1368 r = rbd_read(s->image, 0, 1369 RBD_ENCRYPTION_LUKS_HEADER_VERIFICATION_LEN, buf); 1370 if (r < 0) { 1371 error_setg_errno(errp, -r, "cannot read image start for probe"); 1372 return NULL; 1373 } 1374 } 1375 1376 spec_info = g_new(ImageInfoSpecific, 1); 1377 *spec_info = (ImageInfoSpecific){ 1378 .type = IMAGE_INFO_SPECIFIC_KIND_RBD, 1379 .u.rbd.data = g_new0(ImageInfoSpecificRbd, 1), 1380 }; 1381 1382 if (memcmp(buf, rbd_luks_header_verification, 1383 RBD_ENCRYPTION_LUKS_HEADER_VERIFICATION_LEN) == 0) { 1384 spec_info->u.rbd.data->encryption_format = 1385 RBD_IMAGE_ENCRYPTION_FORMAT_LUKS; 1386 spec_info->u.rbd.data->has_encryption_format = true; 1387 } else if (memcmp(buf, rbd_luks2_header_verification, 1388 RBD_ENCRYPTION_LUKS_HEADER_VERIFICATION_LEN) == 0) { 1389 spec_info->u.rbd.data->encryption_format = 1390 RBD_IMAGE_ENCRYPTION_FORMAT_LUKS2; 1391 spec_info->u.rbd.data->has_encryption_format = true; 1392 } else { 1393 spec_info->u.rbd.data->has_encryption_format = false; 1394 } 1395 1396 return spec_info; 1397 } 1398 1399 static int64_t qemu_rbd_getlength(BlockDriverState *bs) 1400 { 1401 BDRVRBDState *s = bs->opaque; 1402 rbd_image_info_t info; 1403 int r; 1404 1405 r = rbd_stat(s->image, &info, sizeof(info)); 1406 if (r < 0) { 1407 return r; 1408 } 1409 1410 return info.size; 1411 } 1412 1413 static int coroutine_fn qemu_rbd_co_truncate(BlockDriverState *bs, 1414 int64_t offset, 1415 bool exact, 1416 PreallocMode prealloc, 1417 BdrvRequestFlags flags, 1418 Error **errp) 1419 { 1420 int r; 1421 1422 if (prealloc != PREALLOC_MODE_OFF) { 1423 error_setg(errp, "Unsupported preallocation mode '%s'", 1424 PreallocMode_str(prealloc)); 1425 return -ENOTSUP; 1426 } 1427 1428 r = qemu_rbd_resize(bs, offset); 1429 if (r < 0) { 1430 error_setg_errno(errp, -r, "Failed to resize file"); 1431 return r; 1432 } 1433 1434 return 0; 1435 } 1436 1437 static int qemu_rbd_snap_create(BlockDriverState *bs, 1438 QEMUSnapshotInfo *sn_info) 1439 { 1440 BDRVRBDState *s = bs->opaque; 1441 int r; 1442 1443 if (sn_info->name[0] == '\0') { 1444 return -EINVAL; /* we need a name for rbd snapshots */ 1445 } 1446 1447 /* 1448 * rbd snapshots are using the name as the user controlled unique identifier 1449 * we can't use the rbd snapid for that purpose, as it can't be set 1450 */ 1451 if (sn_info->id_str[0] != '\0' && 1452 strcmp(sn_info->id_str, sn_info->name) != 0) { 1453 return -EINVAL; 1454 } 1455 1456 if (strlen(sn_info->name) >= sizeof(sn_info->id_str)) { 1457 return -ERANGE; 1458 } 1459 1460 r = rbd_snap_create(s->image, sn_info->name); 1461 if (r < 0) { 1462 error_report("failed to create snap: %s", strerror(-r)); 1463 return r; 1464 } 1465 1466 return 0; 1467 } 1468 1469 static int qemu_rbd_snap_remove(BlockDriverState *bs, 1470 const char *snapshot_id, 1471 const char *snapshot_name, 1472 Error **errp) 1473 { 1474 BDRVRBDState *s = bs->opaque; 1475 int r; 1476 1477 if (!snapshot_name) { 1478 error_setg(errp, "rbd need a valid snapshot name"); 1479 return -EINVAL; 1480 } 1481 1482 /* If snapshot_id is specified, it must be equal to name, see 1483 qemu_rbd_snap_list() */ 1484 if (snapshot_id && strcmp(snapshot_id, snapshot_name)) { 1485 error_setg(errp, 1486 "rbd do not support snapshot id, it should be NULL or " 1487 "equal to snapshot name"); 1488 return -EINVAL; 1489 } 1490 1491 r = rbd_snap_remove(s->image, snapshot_name); 1492 if (r < 0) { 1493 error_setg_errno(errp, -r, "Failed to remove the snapshot"); 1494 } 1495 return r; 1496 } 1497 1498 static int qemu_rbd_snap_rollback(BlockDriverState *bs, 1499 const char *snapshot_name) 1500 { 1501 BDRVRBDState *s = bs->opaque; 1502 1503 return rbd_snap_rollback(s->image, snapshot_name); 1504 } 1505 1506 static int qemu_rbd_snap_list(BlockDriverState *bs, 1507 QEMUSnapshotInfo **psn_tab) 1508 { 1509 BDRVRBDState *s = bs->opaque; 1510 QEMUSnapshotInfo *sn_info, *sn_tab = NULL; 1511 int i, snap_count; 1512 rbd_snap_info_t *snaps; 1513 int max_snaps = RBD_MAX_SNAPS; 1514 1515 do { 1516 snaps = g_new(rbd_snap_info_t, max_snaps); 1517 snap_count = rbd_snap_list(s->image, snaps, &max_snaps); 1518 if (snap_count <= 0) { 1519 g_free(snaps); 1520 } 1521 } while (snap_count == -ERANGE); 1522 1523 if (snap_count <= 0) { 1524 goto done; 1525 } 1526 1527 sn_tab = g_new0(QEMUSnapshotInfo, snap_count); 1528 1529 for (i = 0; i < snap_count; i++) { 1530 const char *snap_name = snaps[i].name; 1531 1532 sn_info = sn_tab + i; 1533 pstrcpy(sn_info->id_str, sizeof(sn_info->id_str), snap_name); 1534 pstrcpy(sn_info->name, sizeof(sn_info->name), snap_name); 1535 1536 sn_info->vm_state_size = snaps[i].size; 1537 sn_info->date_sec = 0; 1538 sn_info->date_nsec = 0; 1539 sn_info->vm_clock_nsec = 0; 1540 } 1541 rbd_snap_list_end(snaps); 1542 g_free(snaps); 1543 1544 done: 1545 *psn_tab = sn_tab; 1546 return snap_count; 1547 } 1548 1549 #ifdef LIBRBD_SUPPORTS_DISCARD 1550 static BlockAIOCB *qemu_rbd_aio_pdiscard(BlockDriverState *bs, 1551 int64_t offset, 1552 int bytes, 1553 BlockCompletionFunc *cb, 1554 void *opaque) 1555 { 1556 return rbd_start_aio(bs, offset, NULL, bytes, cb, opaque, 1557 RBD_AIO_DISCARD); 1558 } 1559 #endif 1560 1561 #ifdef LIBRBD_SUPPORTS_INVALIDATE 1562 static void coroutine_fn qemu_rbd_co_invalidate_cache(BlockDriverState *bs, 1563 Error **errp) 1564 { 1565 BDRVRBDState *s = bs->opaque; 1566 int r = rbd_invalidate_cache(s->image); 1567 if (r < 0) { 1568 error_setg_errno(errp, -r, "Failed to invalidate the cache"); 1569 } 1570 } 1571 #endif 1572 1573 static QemuOptsList qemu_rbd_create_opts = { 1574 .name = "rbd-create-opts", 1575 .head = QTAILQ_HEAD_INITIALIZER(qemu_rbd_create_opts.head), 1576 .desc = { 1577 { 1578 .name = BLOCK_OPT_SIZE, 1579 .type = QEMU_OPT_SIZE, 1580 .help = "Virtual disk size" 1581 }, 1582 { 1583 .name = BLOCK_OPT_CLUSTER_SIZE, 1584 .type = QEMU_OPT_SIZE, 1585 .help = "RBD object size" 1586 }, 1587 { 1588 .name = "password-secret", 1589 .type = QEMU_OPT_STRING, 1590 .help = "ID of secret providing the password", 1591 }, 1592 { 1593 .name = "encrypt.format", 1594 .type = QEMU_OPT_STRING, 1595 .help = "Encrypt the image, format choices: 'luks', 'luks2'", 1596 }, 1597 { 1598 .name = "encrypt.cipher-alg", 1599 .type = QEMU_OPT_STRING, 1600 .help = "Name of encryption cipher algorithm" 1601 " (allowed values: aes-128, aes-256)", 1602 }, 1603 { 1604 .name = "encrypt.key-secret", 1605 .type = QEMU_OPT_STRING, 1606 .help = "ID of secret providing LUKS passphrase", 1607 }, 1608 { /* end of list */ } 1609 } 1610 }; 1611 1612 static const char *const qemu_rbd_strong_runtime_opts[] = { 1613 "pool", 1614 "namespace", 1615 "image", 1616 "conf", 1617 "snapshot", 1618 "user", 1619 "server.", 1620 "password-secret", 1621 1622 NULL 1623 }; 1624 1625 static BlockDriver bdrv_rbd = { 1626 .format_name = "rbd", 1627 .instance_size = sizeof(BDRVRBDState), 1628 .bdrv_parse_filename = qemu_rbd_parse_filename, 1629 .bdrv_refresh_limits = qemu_rbd_refresh_limits, 1630 .bdrv_file_open = qemu_rbd_open, 1631 .bdrv_close = qemu_rbd_close, 1632 .bdrv_reopen_prepare = qemu_rbd_reopen_prepare, 1633 .bdrv_co_create = qemu_rbd_co_create, 1634 .bdrv_co_create_opts = qemu_rbd_co_create_opts, 1635 .bdrv_has_zero_init = bdrv_has_zero_init_1, 1636 .bdrv_get_info = qemu_rbd_getinfo, 1637 .bdrv_get_specific_info = qemu_rbd_get_specific_info, 1638 .create_opts = &qemu_rbd_create_opts, 1639 .bdrv_getlength = qemu_rbd_getlength, 1640 .bdrv_co_truncate = qemu_rbd_co_truncate, 1641 .protocol_name = "rbd", 1642 1643 .bdrv_aio_preadv = qemu_rbd_aio_preadv, 1644 .bdrv_aio_pwritev = qemu_rbd_aio_pwritev, 1645 1646 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH 1647 .bdrv_aio_flush = qemu_rbd_aio_flush, 1648 #else 1649 .bdrv_co_flush_to_disk = qemu_rbd_co_flush, 1650 #endif 1651 1652 #ifdef LIBRBD_SUPPORTS_DISCARD 1653 .bdrv_aio_pdiscard = qemu_rbd_aio_pdiscard, 1654 #endif 1655 1656 .bdrv_snapshot_create = qemu_rbd_snap_create, 1657 .bdrv_snapshot_delete = qemu_rbd_snap_remove, 1658 .bdrv_snapshot_list = qemu_rbd_snap_list, 1659 .bdrv_snapshot_goto = qemu_rbd_snap_rollback, 1660 #ifdef LIBRBD_SUPPORTS_INVALIDATE 1661 .bdrv_co_invalidate_cache = qemu_rbd_co_invalidate_cache, 1662 #endif 1663 1664 .strong_runtime_opts = qemu_rbd_strong_runtime_opts, 1665 }; 1666 1667 static void bdrv_rbd_init(void) 1668 { 1669 bdrv_register(&bdrv_rbd); 1670 } 1671 1672 block_init(bdrv_rbd_init); 1673