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