2 * QEMU Block driver for RADOS (Ceph)
4 * Copyright (C) 2010-2011 Christian Brunner <chb@muc.de>,
5 * Josh Durgin <josh.durgin@dreamhost.com>
7 * This work is licensed under the terms of the GNU GPL, version 2. See
8 * the COPYING file in the top-level directory.
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.
14 #include "qemu/osdep.h"
16 #include "qapi/error.h"
17 #include "qemu/error-report.h"
18 #include "block/block_int.h"
19 #include "crypto/secret.h"
20 #include "qemu/cutils.h"
21 #include "qapi/qmp/qstring.h"
23 #include <rbd/librbd.h>
26 * When specifying the image filename use:
28 * rbd:poolname/devicename[@snapshotname][:option1=value1[:option2=value2...]]
30 * poolname must be the name of an existing rados pool.
32 * devicename is the name of the rbd image.
34 * Each option given is used to configure rados, and may be any valid
35 * Ceph option, "id", or "conf".
37 * The "id" option indicates what user we should authenticate as to
38 * the Ceph cluster. If it is excluded we will use the Ceph default
41 * The "conf" option specifies a Ceph configuration file to read. If
42 * it is not specified, we will read from the default Ceph locations
43 * (e.g., /etc/ceph/ceph.conf). To avoid reading _any_ configuration
44 * file, specify conf=/dev/null.
46 * Configuration values containing :, @, or = can be escaped with a
50 /* rbd_aio_discard added in 0.1.2 */
51 #if LIBRBD_VERSION_CODE >= LIBRBD_VERSION(0, 1, 2)
52 #define LIBRBD_SUPPORTS_DISCARD
54 #undef LIBRBD_SUPPORTS_DISCARD
57 #define OBJ_MAX_SIZE (1UL << OBJ_DEFAULT_OBJ_ORDER)
59 #define RBD_MAX_CONF_NAME_SIZE 128
60 #define RBD_MAX_CONF_VAL_SIZE 512
61 #define RBD_MAX_CONF_SIZE 1024
62 #define RBD_MAX_POOL_NAME_SIZE 128
63 #define RBD_MAX_SNAP_NAME_SIZE 128
64 #define RBD_MAX_SNAPS 100
66 /* The LIBRBD_SUPPORTS_IOVEC is defined in librbd.h */
67 #ifdef LIBRBD_SUPPORTS_IOVEC
68 #define LIBRBD_USE_IOVEC 1
70 #define LIBRBD_USE_IOVEC 0
80 typedef struct RBDAIOCB
{
87 struct BDRVRBDState
*s
;
90 typedef struct RADOSCB
{
92 struct BDRVRBDState
*s
;
98 typedef struct BDRVRBDState
{
100 rados_ioctx_t io_ctx
;
102 char name
[RBD_MAX_IMAGE_NAME_SIZE
];
106 static char *qemu_rbd_next_tok(int max_len
,
107 char *src
, char delim
,
109 char **p
, Error
**errp
)
117 for (end
= src
; *end
; ++end
) {
121 if (*end
== '\\' && end
[1] != '\0') {
132 error_setg(errp
, "%s too long", name
);
135 error_setg(errp
, "%s too short", name
);
142 static void qemu_rbd_unescape(char *src
)
146 for (p
= src
; *src
; ++src
, ++p
) {
147 if (*src
== '\\' && src
[1] != '\0') {
155 static void qemu_rbd_parse_filename(const char *filename
, QDict
*options
,
159 char *p
, *buf
, *keypairs
;
161 size_t max_keypair_size
;
162 Error
*local_err
= NULL
;
164 if (!strstart(filename
, "rbd:", &start
)) {
165 error_setg(errp
, "File name must start with 'rbd:'");
169 max_keypair_size
= strlen(start
) + 1;
170 buf
= g_strdup(start
);
171 keypairs
= g_malloc0(max_keypair_size
);
174 found_str
= qemu_rbd_next_tok(RBD_MAX_POOL_NAME_SIZE
, p
,
175 '/', "pool name", &p
, &local_err
);
180 error_setg(errp
, "Pool name is required");
183 qemu_rbd_unescape(found_str
);
184 qdict_put(options
, "pool", qstring_from_str(found_str
));
186 if (strchr(p
, '@')) {
187 found_str
= qemu_rbd_next_tok(RBD_MAX_IMAGE_NAME_SIZE
, p
,
188 '@', "object name", &p
, &local_err
);
192 qemu_rbd_unescape(found_str
);
193 qdict_put(options
, "image", qstring_from_str(found_str
));
195 found_str
= qemu_rbd_next_tok(RBD_MAX_SNAP_NAME_SIZE
, p
,
196 ':', "snap name", &p
, &local_err
);
200 qemu_rbd_unescape(found_str
);
201 qdict_put(options
, "snapshot", qstring_from_str(found_str
));
203 found_str
= qemu_rbd_next_tok(RBD_MAX_IMAGE_NAME_SIZE
, p
,
204 ':', "object name", &p
, &local_err
);
208 qemu_rbd_unescape(found_str
);
209 qdict_put(options
, "image", qstring_from_str(found_str
));
215 found_str
= qemu_rbd_next_tok(RBD_MAX_CONF_NAME_SIZE
, p
,
216 '\0', "configuration", &p
, &local_err
);
223 /* The following are essentially all key/value pairs, and we treat
224 * 'id' and 'conf' a bit special. Key/value pairs may be in any order. */
227 name
= qemu_rbd_next_tok(RBD_MAX_CONF_NAME_SIZE
, p
,
228 '=', "conf option name", &p
, &local_err
);
234 error_setg(errp
, "conf option %s has no value", name
);
238 qemu_rbd_unescape(name
);
240 value
= qemu_rbd_next_tok(RBD_MAX_CONF_VAL_SIZE
, p
,
241 ':', "conf option value", &p
, &local_err
);
245 qemu_rbd_unescape(value
);
247 if (!strcmp(name
, "conf")) {
248 qdict_put(options
, "conf", qstring_from_str(value
));
249 } else if (!strcmp(name
, "id")) {
250 qdict_put(options
, "user" , qstring_from_str(value
));
252 /* FIXME: This is pretty ugly, and not the right way to do this.
253 * These should be contained in a structure, and then
254 * passed explicitly as individual key/value pairs to
255 * rados. Consider this legacy code that needs to be
257 char *tmp
= g_malloc0(max_keypair_size
);
258 /* only use a delimiter if it is not the first keypair found */
259 /* These are sets of unknown key/value pairs we'll pass along
262 snprintf(tmp
, max_keypair_size
, ":%s=%s", name
, value
);
263 pstrcat(keypairs
, max_keypair_size
, tmp
);
265 snprintf(keypairs
, max_keypair_size
, "%s=%s", name
, value
);
272 qdict_put(options
, "keyvalue-pairs", qstring_from_str(keypairs
));
278 error_propagate(errp
, local_err
);
286 static int qemu_rbd_set_auth(rados_t cluster
, const char *secretid
,
293 gchar
*secret
= qcrypto_secret_lookup_as_base64(secretid
,
299 rados_conf_set(cluster
, "key", secret
);
305 static int qemu_rbd_set_keypairs(rados_t cluster
, const char *keypairs
,
311 Error
*local_err
= NULL
;
314 buf
= g_strdup(keypairs
);
318 name
= qemu_rbd_next_tok(RBD_MAX_CONF_NAME_SIZE
, p
,
319 '=', "conf option name", &p
, &local_err
);
325 error_setg(errp
, "conf option %s has no value", name
);
330 value
= qemu_rbd_next_tok(RBD_MAX_CONF_VAL_SIZE
, p
,
331 ':', "conf option value", &p
, &local_err
);
336 ret
= rados_conf_set(cluster
, name
, value
);
338 error_setg_errno(errp
, -ret
, "invalid conf option %s", name
);
345 error_propagate(errp
, local_err
);
352 static void qemu_rbd_memset(RADOSCB
*rcb
, int64_t offs
)
354 if (LIBRBD_USE_IOVEC
) {
355 RBDAIOCB
*acb
= rcb
->acb
;
356 iov_memset(acb
->qiov
->iov
, acb
->qiov
->niov
, offs
, 0,
357 acb
->qiov
->size
- offs
);
359 memset(rcb
->buf
+ offs
, 0, rcb
->size
- offs
);
363 static QemuOptsList runtime_opts
= {
365 .head
= QTAILQ_HEAD_INITIALIZER(runtime_opts
.head
),
369 .type
= QEMU_OPT_STRING
,
370 .help
= "Specification of the rbd image",
373 .name
= "password-secret",
374 .type
= QEMU_OPT_STRING
,
375 .help
= "ID of secret providing the password",
379 .type
= QEMU_OPT_STRING
,
380 .help
= "Rados config file location",
384 .type
= QEMU_OPT_STRING
,
385 .help
= "Rados pool name",
389 .type
= QEMU_OPT_STRING
,
390 .help
= "Image name in the pool",
394 .type
= QEMU_OPT_STRING
,
395 .help
= "Ceph snapshot name",
398 /* maps to 'id' in rados_create() */
400 .type
= QEMU_OPT_STRING
,
401 .help
= "Rados id name",
404 .name
= "keyvalue-pairs",
405 .type
= QEMU_OPT_STRING
,
406 .help
= "Legacy rados key/value option parameters",
410 .type
= QEMU_OPT_STRING
,
414 .type
= QEMU_OPT_STRING
,
418 .type
= QEMU_OPT_STRING
,
419 .help
= "Supported authentication method, either cephx or none",
421 { /* end of list */ }
425 static int qemu_rbd_create(const char *filename
, QemuOpts
*opts
, Error
**errp
)
427 Error
*local_err
= NULL
;
431 const char *pool
, *name
, *conf
, *clientname
, *keypairs
;
432 const char *secretid
;
434 rados_ioctx_t io_ctx
;
435 QDict
*options
= NULL
;
436 QemuOpts
*rbd_opts
= NULL
;
439 secretid
= qemu_opt_get(opts
, "password-secret");
441 /* Read out options */
442 bytes
= ROUND_UP(qemu_opt_get_size_del(opts
, BLOCK_OPT_SIZE
, 0),
444 objsize
= qemu_opt_get_size_del(opts
, BLOCK_OPT_CLUSTER_SIZE
, 0);
446 if ((objsize
- 1) & objsize
) { /* not a power of 2? */
447 error_setg(errp
, "obj size needs to be power of 2");
451 if (objsize
< 4096) {
452 error_setg(errp
, "obj size too small");
456 obj_order
= ctz32(objsize
);
459 options
= qdict_new();
460 qemu_rbd_parse_filename(filename
, options
, &local_err
);
463 error_propagate(errp
, local_err
);
467 rbd_opts
= qemu_opts_create(&runtime_opts
, NULL
, 0, &error_abort
);
468 qemu_opts_absorb_qdict(rbd_opts
, options
, &local_err
);
470 error_propagate(errp
, local_err
);
475 pool
= qemu_opt_get(rbd_opts
, "pool");
476 conf
= qemu_opt_get(rbd_opts
, "conf");
477 clientname
= qemu_opt_get(rbd_opts
, "user");
478 name
= qemu_opt_get(rbd_opts
, "image");
479 keypairs
= qemu_opt_get(rbd_opts
, "keyvalue-pairs");
481 ret
= rados_create(&cluster
, clientname
);
483 error_setg_errno(errp
, -ret
, "error initializing");
487 /* try default location when conf=NULL, but ignore failure */
488 ret
= rados_conf_read_file(cluster
, conf
);
489 if (conf
&& ret
< 0) {
490 error_setg_errno(errp
, -ret
, "error reading conf file %s", conf
);
495 ret
= qemu_rbd_set_keypairs(cluster
, keypairs
, errp
);
501 if (qemu_rbd_set_auth(cluster
, secretid
, errp
) < 0) {
506 ret
= rados_connect(cluster
);
508 error_setg_errno(errp
, -ret
, "error connecting");
512 ret
= rados_ioctx_create(cluster
, pool
, &io_ctx
);
514 error_setg_errno(errp
, -ret
, "error opening pool %s", pool
);
518 ret
= rbd_create(io_ctx
, name
, bytes
, &obj_order
);
520 error_setg_errno(errp
, -ret
, "error rbd create");
523 rados_ioctx_destroy(io_ctx
);
526 rados_shutdown(cluster
);
530 qemu_opts_del(rbd_opts
);
535 * This aio completion is being called from rbd_finish_bh() and runs in qemu
538 static void qemu_rbd_complete_aio(RADOSCB
*rcb
)
540 RBDAIOCB
*acb
= rcb
->acb
;
545 if (acb
->cmd
!= RBD_AIO_READ
) {
549 } else if (!acb
->error
) {
550 acb
->ret
= rcb
->size
;
554 qemu_rbd_memset(rcb
, 0);
557 } else if (r
< rcb
->size
) {
558 qemu_rbd_memset(rcb
, r
);
560 acb
->ret
= rcb
->size
;
562 } else if (!acb
->error
) {
569 if (!LIBRBD_USE_IOVEC
) {
570 if (acb
->cmd
== RBD_AIO_READ
) {
571 qemu_iovec_from_buf(acb
->qiov
, 0, acb
->bounce
, acb
->qiov
->size
);
573 qemu_vfree(acb
->bounce
);
576 acb
->common
.cb(acb
->common
.opaque
, (acb
->ret
> 0 ? 0 : acb
->ret
));
581 #define RBD_MON_HOST 0
582 #define RBD_AUTH_SUPPORTED 1
584 static char *qemu_rbd_array_opts(QDict
*options
, const char *prefix
, int type
,
588 QemuOpts
*opts
= NULL
;
593 char *rados_str
= NULL
;
594 Error
*local_err
= NULL
;
597 assert(type
== RBD_MON_HOST
|| type
== RBD_AUTH_SUPPORTED
);
599 num_entries
= qdict_array_entries(options
, prefix
);
601 if (num_entries
< 0) {
602 error_setg(errp
, "Parse error on RBD QDict array");
606 for (i
= 0; i
< num_entries
; i
++) {
611 str
= g_strdup_printf("%s%d.", prefix
, i
);
612 qdict_extract_subqdict(options
, &sub_options
, str
);
615 opts
= qemu_opts_create(&runtime_opts
, NULL
, 0, &error_abort
);
616 qemu_opts_absorb_qdict(opts
, sub_options
, &local_err
);
617 QDECREF(sub_options
);
619 error_propagate(errp
, local_err
);
625 if (type
== RBD_MON_HOST
) {
626 host
= qemu_opt_get(opts
, "host");
627 port
= qemu_opt_get(opts
, "port");
632 if (strchr(host
, ':')) {
633 strbuf
= g_strdup_printf("[%s]:%s", host
, port
);
635 strbuf
= g_strdup_printf("%s:%s", host
, port
);
638 } else if (strchr(host
, ':')) {
639 strbuf
= g_strdup_printf("[%s]", host
);
643 value
= qemu_opt_get(opts
, "auth");
647 /* each iteration in the for loop will build upon the string, and if
648 * rados_str is NULL then it is our first pass */
650 /* separate options with ';', as that is what rados_conf_set()
652 rados_str_tmp
= rados_str
;
653 rados_str
= g_strdup_printf("%s;%s", rados_str_tmp
, value
);
654 g_free(rados_str_tmp
);
656 rados_str
= g_strdup(value
);
669 static int qemu_rbd_open(BlockDriverState
*bs
, QDict
*options
, int flags
,
672 BDRVRBDState
*s
= bs
->opaque
;
673 const char *pool
, *snap
, *conf
, *clientname
, *name
, *keypairs
;
674 const char *secretid
;
676 Error
*local_err
= NULL
;
677 char *mon_host
= NULL
;
678 char *auth_supported
= NULL
;
681 opts
= qemu_opts_create(&runtime_opts
, NULL
, 0, &error_abort
);
682 qemu_opts_absorb_qdict(opts
, options
, &local_err
);
684 error_propagate(errp
, local_err
);
689 auth_supported
= qemu_rbd_array_opts(options
, "auth-supported.",
690 RBD_AUTH_SUPPORTED
, &local_err
);
692 error_propagate(errp
, local_err
);
697 mon_host
= qemu_rbd_array_opts(options
, "server.",
698 RBD_MON_HOST
, &local_err
);
700 error_propagate(errp
, local_err
);
705 secretid
= qemu_opt_get(opts
, "password-secret");
707 pool
= qemu_opt_get(opts
, "pool");
708 conf
= qemu_opt_get(opts
, "conf");
709 snap
= qemu_opt_get(opts
, "snapshot");
710 clientname
= qemu_opt_get(opts
, "user");
711 name
= qemu_opt_get(opts
, "image");
712 keypairs
= qemu_opt_get(opts
, "keyvalue-pairs");
714 r
= rados_create(&s
->cluster
, clientname
);
716 error_setg_errno(errp
, -r
, "error initializing");
720 s
->snap
= g_strdup(snap
);
722 pstrcpy(s
->name
, RBD_MAX_IMAGE_NAME_SIZE
, name
);
725 /* try default location when conf=NULL, but ignore failure */
726 r
= rados_conf_read_file(s
->cluster
, conf
);
728 error_setg_errno(errp
, -r
, "error reading conf file %s", conf
);
729 goto failed_shutdown
;
732 r
= qemu_rbd_set_keypairs(s
->cluster
, keypairs
, errp
);
734 goto failed_shutdown
;
738 r
= rados_conf_set(s
->cluster
, "mon_host", mon_host
);
740 goto failed_shutdown
;
744 if (auth_supported
) {
745 r
= rados_conf_set(s
->cluster
, "auth_supported", auth_supported
);
747 goto failed_shutdown
;
751 if (qemu_rbd_set_auth(s
->cluster
, secretid
, errp
) < 0) {
753 goto failed_shutdown
;
757 * Fallback to more conservative semantics if setting cache
758 * options fails. Ignore errors from setting rbd_cache because the
759 * only possible error is that the option does not exist, and
760 * librbd defaults to no caching. If write through caching cannot
761 * be set up, fall back to no caching.
763 if (flags
& BDRV_O_NOCACHE
) {
764 rados_conf_set(s
->cluster
, "rbd_cache", "false");
766 rados_conf_set(s
->cluster
, "rbd_cache", "true");
769 r
= rados_connect(s
->cluster
);
771 error_setg_errno(errp
, -r
, "error connecting");
772 goto failed_shutdown
;
775 r
= rados_ioctx_create(s
->cluster
, pool
, &s
->io_ctx
);
777 error_setg_errno(errp
, -r
, "error opening pool %s", pool
);
778 goto failed_shutdown
;
781 r
= rbd_open(s
->io_ctx
, s
->name
, &s
->image
, s
->snap
);
783 error_setg_errno(errp
, -r
, "error reading header from %s", s
->name
);
787 bs
->read_only
= (s
->snap
!= NULL
);
793 rados_ioctx_destroy(s
->io_ctx
);
795 rados_shutdown(s
->cluster
);
800 g_free(auth_supported
);
804 static void qemu_rbd_close(BlockDriverState
*bs
)
806 BDRVRBDState
*s
= bs
->opaque
;
809 rados_ioctx_destroy(s
->io_ctx
);
811 rados_shutdown(s
->cluster
);
814 static const AIOCBInfo rbd_aiocb_info
= {
815 .aiocb_size
= sizeof(RBDAIOCB
),
818 static void rbd_finish_bh(void *opaque
)
820 RADOSCB
*rcb
= opaque
;
821 qemu_rbd_complete_aio(rcb
);
825 * This is the callback function for rbd_aio_read and _write
827 * Note: this function is being called from a non qemu thread so
828 * we need to be careful about what we do here. Generally we only
829 * schedule a BH, and do the rest of the io completion handling
830 * from rbd_finish_bh() which runs in a qemu context.
832 static void rbd_finish_aiocb(rbd_completion_t c
, RADOSCB
*rcb
)
834 RBDAIOCB
*acb
= rcb
->acb
;
836 rcb
->ret
= rbd_aio_get_return_value(c
);
839 aio_bh_schedule_oneshot(bdrv_get_aio_context(acb
->common
.bs
),
843 static int rbd_aio_discard_wrapper(rbd_image_t image
,
846 rbd_completion_t comp
)
848 #ifdef LIBRBD_SUPPORTS_DISCARD
849 return rbd_aio_discard(image
, off
, len
, comp
);
855 static int rbd_aio_flush_wrapper(rbd_image_t image
,
856 rbd_completion_t comp
)
858 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
859 return rbd_aio_flush(image
, comp
);
865 static BlockAIOCB
*rbd_start_aio(BlockDriverState
*bs
,
869 BlockCompletionFunc
*cb
,
878 BDRVRBDState
*s
= bs
->opaque
;
880 acb
= qemu_aio_get(&rbd_aiocb_info
, bs
, cb
, opaque
);
883 assert(!qiov
|| qiov
->size
== size
);
885 rcb
= g_new(RADOSCB
, 1);
887 if (!LIBRBD_USE_IOVEC
) {
888 if (cmd
== RBD_AIO_DISCARD
|| cmd
== RBD_AIO_FLUSH
) {
891 acb
->bounce
= qemu_try_blockalign(bs
, qiov
->size
);
892 if (acb
->bounce
== NULL
) {
896 if (cmd
== RBD_AIO_WRITE
) {
897 qemu_iovec_to_buf(acb
->qiov
, 0, acb
->bounce
, qiov
->size
);
899 rcb
->buf
= acb
->bounce
;
909 r
= rbd_aio_create_completion(rcb
, (rbd_callback_t
) rbd_finish_aiocb
, &c
);
916 #ifdef LIBRBD_SUPPORTS_IOVEC
917 r
= rbd_aio_writev(s
->image
, qiov
->iov
, qiov
->niov
, off
, c
);
919 r
= rbd_aio_write(s
->image
, off
, size
, rcb
->buf
, c
);
923 #ifdef LIBRBD_SUPPORTS_IOVEC
924 r
= rbd_aio_readv(s
->image
, qiov
->iov
, qiov
->niov
, off
, c
);
926 r
= rbd_aio_read(s
->image
, off
, size
, rcb
->buf
, c
);
929 case RBD_AIO_DISCARD
:
930 r
= rbd_aio_discard_wrapper(s
->image
, off
, size
, c
);
933 r
= rbd_aio_flush_wrapper(s
->image
, c
);
940 goto failed_completion
;
948 if (!LIBRBD_USE_IOVEC
) {
949 qemu_vfree(acb
->bounce
);
956 static BlockAIOCB
*qemu_rbd_aio_readv(BlockDriverState
*bs
,
960 BlockCompletionFunc
*cb
,
963 return rbd_start_aio(bs
, sector_num
<< BDRV_SECTOR_BITS
, qiov
,
964 (int64_t) nb_sectors
<< BDRV_SECTOR_BITS
, cb
, opaque
,
968 static BlockAIOCB
*qemu_rbd_aio_writev(BlockDriverState
*bs
,
972 BlockCompletionFunc
*cb
,
975 return rbd_start_aio(bs
, sector_num
<< BDRV_SECTOR_BITS
, qiov
,
976 (int64_t) nb_sectors
<< BDRV_SECTOR_BITS
, cb
, opaque
,
980 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
981 static BlockAIOCB
*qemu_rbd_aio_flush(BlockDriverState
*bs
,
982 BlockCompletionFunc
*cb
,
985 return rbd_start_aio(bs
, 0, NULL
, 0, cb
, opaque
, RBD_AIO_FLUSH
);
990 static int qemu_rbd_co_flush(BlockDriverState
*bs
)
992 #if LIBRBD_VERSION_CODE >= LIBRBD_VERSION(0, 1, 1)
993 /* rbd_flush added in 0.1.1 */
994 BDRVRBDState
*s
= bs
->opaque
;
995 return rbd_flush(s
->image
);
1002 static int qemu_rbd_getinfo(BlockDriverState
*bs
, BlockDriverInfo
*bdi
)
1004 BDRVRBDState
*s
= bs
->opaque
;
1005 rbd_image_info_t info
;
1008 r
= rbd_stat(s
->image
, &info
, sizeof(info
));
1013 bdi
->cluster_size
= info
.obj_size
;
1017 static int64_t qemu_rbd_getlength(BlockDriverState
*bs
)
1019 BDRVRBDState
*s
= bs
->opaque
;
1020 rbd_image_info_t info
;
1023 r
= rbd_stat(s
->image
, &info
, sizeof(info
));
1031 static int qemu_rbd_truncate(BlockDriverState
*bs
, int64_t offset
)
1033 BDRVRBDState
*s
= bs
->opaque
;
1036 r
= rbd_resize(s
->image
, offset
);
1044 static int qemu_rbd_snap_create(BlockDriverState
*bs
,
1045 QEMUSnapshotInfo
*sn_info
)
1047 BDRVRBDState
*s
= bs
->opaque
;
1050 if (sn_info
->name
[0] == '\0') {
1051 return -EINVAL
; /* we need a name for rbd snapshots */
1055 * rbd snapshots are using the name as the user controlled unique identifier
1056 * we can't use the rbd snapid for that purpose, as it can't be set
1058 if (sn_info
->id_str
[0] != '\0' &&
1059 strcmp(sn_info
->id_str
, sn_info
->name
) != 0) {
1063 if (strlen(sn_info
->name
) >= sizeof(sn_info
->id_str
)) {
1067 r
= rbd_snap_create(s
->image
, sn_info
->name
);
1069 error_report("failed to create snap: %s", strerror(-r
));
1076 static int qemu_rbd_snap_remove(BlockDriverState
*bs
,
1077 const char *snapshot_id
,
1078 const char *snapshot_name
,
1081 BDRVRBDState
*s
= bs
->opaque
;
1084 if (!snapshot_name
) {
1085 error_setg(errp
, "rbd need a valid snapshot name");
1089 /* If snapshot_id is specified, it must be equal to name, see
1090 qemu_rbd_snap_list() */
1091 if (snapshot_id
&& strcmp(snapshot_id
, snapshot_name
)) {
1093 "rbd do not support snapshot id, it should be NULL or "
1094 "equal to snapshot name");
1098 r
= rbd_snap_remove(s
->image
, snapshot_name
);
1100 error_setg_errno(errp
, -r
, "Failed to remove the snapshot");
1105 static int qemu_rbd_snap_rollback(BlockDriverState
*bs
,
1106 const char *snapshot_name
)
1108 BDRVRBDState
*s
= bs
->opaque
;
1110 return rbd_snap_rollback(s
->image
, snapshot_name
);
1113 static int qemu_rbd_snap_list(BlockDriverState
*bs
,
1114 QEMUSnapshotInfo
**psn_tab
)
1116 BDRVRBDState
*s
= bs
->opaque
;
1117 QEMUSnapshotInfo
*sn_info
, *sn_tab
= NULL
;
1119 rbd_snap_info_t
*snaps
;
1120 int max_snaps
= RBD_MAX_SNAPS
;
1123 snaps
= g_new(rbd_snap_info_t
, max_snaps
);
1124 snap_count
= rbd_snap_list(s
->image
, snaps
, &max_snaps
);
1125 if (snap_count
<= 0) {
1128 } while (snap_count
== -ERANGE
);
1130 if (snap_count
<= 0) {
1134 sn_tab
= g_new0(QEMUSnapshotInfo
, snap_count
);
1136 for (i
= 0; i
< snap_count
; i
++) {
1137 const char *snap_name
= snaps
[i
].name
;
1139 sn_info
= sn_tab
+ i
;
1140 pstrcpy(sn_info
->id_str
, sizeof(sn_info
->id_str
), snap_name
);
1141 pstrcpy(sn_info
->name
, sizeof(sn_info
->name
), snap_name
);
1143 sn_info
->vm_state_size
= snaps
[i
].size
;
1144 sn_info
->date_sec
= 0;
1145 sn_info
->date_nsec
= 0;
1146 sn_info
->vm_clock_nsec
= 0;
1148 rbd_snap_list_end(snaps
);
1156 #ifdef LIBRBD_SUPPORTS_DISCARD
1157 static BlockAIOCB
*qemu_rbd_aio_pdiscard(BlockDriverState
*bs
,
1160 BlockCompletionFunc
*cb
,
1163 return rbd_start_aio(bs
, offset
, NULL
, count
, cb
, opaque
,
1168 #ifdef LIBRBD_SUPPORTS_INVALIDATE
1169 static void qemu_rbd_invalidate_cache(BlockDriverState
*bs
,
1172 BDRVRBDState
*s
= bs
->opaque
;
1173 int r
= rbd_invalidate_cache(s
->image
);
1175 error_setg_errno(errp
, -r
, "Failed to invalidate the cache");
1180 static QemuOptsList qemu_rbd_create_opts
= {
1181 .name
= "rbd-create-opts",
1182 .head
= QTAILQ_HEAD_INITIALIZER(qemu_rbd_create_opts
.head
),
1185 .name
= BLOCK_OPT_SIZE
,
1186 .type
= QEMU_OPT_SIZE
,
1187 .help
= "Virtual disk size"
1190 .name
= BLOCK_OPT_CLUSTER_SIZE
,
1191 .type
= QEMU_OPT_SIZE
,
1192 .help
= "RBD object size"
1195 .name
= "password-secret",
1196 .type
= QEMU_OPT_STRING
,
1197 .help
= "ID of secret providing the password",
1199 { /* end of list */ }
1203 static BlockDriver bdrv_rbd
= {
1204 .format_name
= "rbd",
1205 .instance_size
= sizeof(BDRVRBDState
),
1206 .bdrv_parse_filename
= qemu_rbd_parse_filename
,
1207 .bdrv_file_open
= qemu_rbd_open
,
1208 .bdrv_close
= qemu_rbd_close
,
1209 .bdrv_create
= qemu_rbd_create
,
1210 .bdrv_has_zero_init
= bdrv_has_zero_init_1
,
1211 .bdrv_get_info
= qemu_rbd_getinfo
,
1212 .create_opts
= &qemu_rbd_create_opts
,
1213 .bdrv_getlength
= qemu_rbd_getlength
,
1214 .bdrv_truncate
= qemu_rbd_truncate
,
1215 .protocol_name
= "rbd",
1217 .bdrv_aio_readv
= qemu_rbd_aio_readv
,
1218 .bdrv_aio_writev
= qemu_rbd_aio_writev
,
1220 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
1221 .bdrv_aio_flush
= qemu_rbd_aio_flush
,
1223 .bdrv_co_flush_to_disk
= qemu_rbd_co_flush
,
1226 #ifdef LIBRBD_SUPPORTS_DISCARD
1227 .bdrv_aio_pdiscard
= qemu_rbd_aio_pdiscard
,
1230 .bdrv_snapshot_create
= qemu_rbd_snap_create
,
1231 .bdrv_snapshot_delete
= qemu_rbd_snap_remove
,
1232 .bdrv_snapshot_list
= qemu_rbd_snap_list
,
1233 .bdrv_snapshot_goto
= qemu_rbd_snap_rollback
,
1234 #ifdef LIBRBD_SUPPORTS_INVALIDATE
1235 .bdrv_invalidate_cache
= qemu_rbd_invalidate_cache
,
1239 static void bdrv_rbd_init(void)
1241 bdrv_register(&bdrv_rbd
);
1244 block_init(bdrv_rbd_init
);