features/qemu-block: add qemu backing image support (clone)

Add basic backing image support to the block-format mechanism. This
is a functionality checkpoint that enables the raw mechanism
required to support client driven "snapshot" and "clone" requests.

This change enhances the block-format setxattr command to support
an additional and optional backing image reference. For example:

setxattr -n trusted.glusterfs.block-format -v "qcow2:10GB:<bimg>" ./newimage

... where <bimg> refers to the backing image for unallocated blocks
in newimage. <bimg> can be provided in one of two formats:

- a gfid string in the following format (assuming a valid gfid):

<gfid:00000000-0000-0000-0000-000000000000>

- or a filename that must be resident in the same directory as the
new clone file being formatted. E.g.,

setxattr -n trusted.glusterfs.block-format -v "qcow2:10GB:baseimg" ./newimage

This latter format is more restrictive, simply provided for
convenience or until something more refined is available.

This change makes no assumptions about the backing image file and
affords no additional protection. It is up to the user/client to
recognize the relationship between the files and manage them
appropriately (i.e., no writes to the backing image, etc.).

BUG: 986775
Change-Id: I7aff7bdc59b85a6459001a6bfeae4db6bf74f703
Signed-off-by: Brian Foster <bfoster@redhat.com>
Reviewed-on: http://review.gluster.org/5967
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Anand Avati <avati@redhat.com>

4 files changed, 166 insertions, 23 deletions

diff --git a/xlators/features/qemu-block/src/bdrv-xlator.c b/xlators/features/qemu-block/src/bdrv-xlator.c
index 145f7c79d5e..106c5977535 100644
--- a/xlators/features/qemu-block/src/bdrv-xlator.c
+++ b/xlators/features/qemu-block/src/bdrv-xlator.c
@@ -69,10 +69,13 @@ static int
 qemu_gluster_open (BlockDriverState *bs, QDict *options, int bdrv_flags)
 {
 	inode_t *inode = NULL;
-	BDRVGlusterState *s = NULL;
+	BDRVGlusterState *s = bs->opaque;
 	QemuOpts *opts = NULL;
 	Error *local_err = NULL;
 	const char *filename = NULL;
+	char gfid_str[128];
+	int ret;
+	qb_conf_t *conf = THIS->private;
 
 	opts = qemu_opts_create_nofail(&runtime_opts);
 	qemu_opts_absorb_qdict(opts, options, &local_err);
@@ -84,12 +87,40 @@ qemu_gluster_open (BlockDriverState *bs, QDict *options, int bdrv_flags)
 
 	filename = qemu_opt_get(opts, "filename");
 
-	inode = qb_inode_from_filename (filename);
-	if (!inode)
-		return -EINVAL;
+	/*
+	 * gfid:<gfid> format means we're opening a backing image.
+	 */
+	ret = sscanf(filename, "gluster://gfid:%s", gfid_str);
+	if (ret) {
+		loc_t loc = {0,};
+		struct iatt buf = {0,};
+		uuid_t gfid;
 
-	s = bs->opaque;
-	s->inode = inode_ref (inode);
+		uuid_parse(gfid_str, gfid);
+
+		loc.inode = inode_find(conf->root_inode->table, gfid);
+		if (!loc.inode) {
+			loc.inode = inode_new(conf->root_inode->table);
+			uuid_copy(loc.inode->gfid, gfid);
+		}
+
+		uuid_copy(loc.gfid, loc.inode->gfid);
+		ret = syncop_lookup(FIRST_CHILD(THIS), &loc, NULL, &buf, NULL,
+				    NULL);
+		if (ret) {
+			loc_wipe(&loc);
+			return -errno;
+		}
+
+		s->inode = inode_ref(loc.inode);
+		loc_wipe(&loc);
+	} else {
+		inode = qb_inode_from_filename (filename);
+		if (!inode)
+			return -EINVAL;
+
+		s->inode = inode_ref(inode);
+	}
 
 	return 0;
 }
diff --git a/xlators/features/qemu-block/src/qb-coroutines.c b/xlators/features/qemu-block/src/qb-coroutines.c
index 9e64fecaf5c..d29117eb5d4 100644
--- a/xlators/features/qemu-block/src/qb-coroutines.c
+++ b/xlators/features/qemu-block/src/qb-coroutines.c
@@ -35,6 +35,8 @@ qb_format_and_resume (void *opaque)
 	call_stub_t *stub = NULL;
 	inode_t *inode = NULL;
 	char filename[64];
+	char base_filename[128];
+	int use_base = 0;
 	qb_inode_t *qb_inode = NULL;
 	Error *local_err = NULL;
 	fd_t *fd = NULL;
@@ -54,8 +56,62 @@ qb_format_and_resume (void *opaque)
 
 	qb_inode = qb_inode_ctx_get (frame->this, inode);
 
-	bdrv_img_create (filename, qb_inode->fmt, 0, 0,
-			 0, qb_inode->size, 0, &local_err, true);
+	/*
+	 * See if the caller specified a backing image.
+	 */
+	if (!uuid_is_null(qb_inode->backing_gfid) || qb_inode->backing_fname) {
+		loc_t loc = {0,};
+		char gfid_str[64];
+		struct iatt buf;
+
+		if (!uuid_is_null(qb_inode->backing_gfid)) {
+			loc.inode = inode_find(qb_conf->root_inode->table,
+					qb_inode->backing_gfid);
+			if (!loc.inode) {
+				loc.inode = inode_new(qb_conf->root_inode->table);
+				uuid_copy(loc.inode->gfid,
+					qb_inode->backing_gfid);
+			}
+			uuid_copy(loc.gfid, loc.inode->gfid);
+		} else if (qb_inode->backing_fname) {
+			loc.inode = inode_new(qb_conf->root_inode->table);
+			loc.name = qb_inode->backing_fname;
+			loc.parent = inode_parent(inode, NULL, NULL);
+			loc_path(&loc, loc.name);
+		}
+
+		/*
+		 * Lookup the backing image. Verify existence and/or get the
+		 * gfid if we don't already have it.
+		 */
+		ret = syncop_lookup(FIRST_CHILD(frame->this), &loc, NULL, &buf,
+				    NULL, NULL);
+		GF_FREE(qb_inode->backing_fname);
+		if (ret) {
+			loc_wipe(&loc);
+			ret = errno;
+			goto err;
+		}
+
+		uuid_copy(qb_inode->backing_gfid, buf.ia_gfid);
+		loc_wipe(&loc);
+
+		/*
+		 * We pass the filename of the backing image into the qemu block
+		 * subsystem as the associated gfid. This is embedded into the
+		 * clone image and passed along to the gluster bdrv backend when
+		 * the block subsystem needs to operate on the backing image on
+		 * behalf of the clone.
+		 */
+		uuid_unparse(qb_inode->backing_gfid, gfid_str);
+		snprintf(base_filename, sizeof(base_filename),
+			 "gluster://gfid:%s", gfid_str);
+		use_base = 1;
+	}
+
+	bdrv_img_create (filename, qb_inode->fmt,
+			 use_base ? base_filename : NULL, 0, 0, qb_inode->size,
+			 0, &local_err, true);
 
 	if (error_is_set (&local_err)) {
 		gf_log (frame->this->name, GF_LOG_ERROR, "%s",
@@ -113,6 +169,10 @@ qb_format_and_resume (void *opaque)
 	QB_STUB_UNWIND (stub, 0, 0);
 
 	return 0;
+
+err:
+	QB_STUB_UNWIND(stub, -1, ret);
+	return 0;
 }
 
 
diff --git a/xlators/features/qemu-block/src/qemu-block.c b/xlators/features/qemu-block/src/qemu-block.c
index 416ae44383c..0edb7b9493d 100644
--- a/xlators/features/qemu-block/src/qemu-block.c
+++ b/xlators/features/qemu-block/src/qemu-block.c
@@ -108,42 +108,76 @@ qb_iatt_fixup (xlator_t *this, inode_t *inode, struct iatt *iatt)
 int
 qb_format_extract (xlator_t *this, char *format, inode_t *inode)
 {
-	char       *s = NULL;
+	char       *s, *save;
 	uint64_t    size = 0;
 	char        fmt[QB_XATTR_VAL_MAX+1] = {0, };
 	qb_inode_t *qb_inode = NULL;
+	char *formatstr = NULL;
+	uuid_t gfid = {0,};
+	char gfid_str[64] = {0,};
+	int ret;
 
-	strncpy (fmt, format, QB_XATTR_VAL_MAX);
-	s = strchr (fmt, ':');
+	strncpy(fmt, format, QB_XATTR_VAL_MAX);
+
+	s = strtok_r(fmt, ":", &save);
 	if (!s)
 		goto invalid;
-	if (s == fmt)
-		goto invalid;
+	formatstr = gf_strdup(s);
 
-	*s = 0; s++;
-	if (!*s || strchr (s, ':'))
+	s = strtok_r(NULL, ":", &save);
+	if (!s)
 		goto invalid;
-
 	if (gf_string2bytesize (s, &size))
 		goto invalid;
-
 	if (!size)
 		goto invalid;
 
+	s = strtok_r(NULL, "\0", &save);
+	if (s && !strncmp(s, "<gfid:", strlen("<gfid:"))) {
+		/*
+		 * Check for valid gfid backing image specifier.
+		 */
+		if (strlen(s) + 1 > sizeof(gfid_str))
+			goto invalid;
+		ret = sscanf(s, "<gfid:%[^>]s", gfid_str);
+		if (ret == 1) {
+			ret = uuid_parse(gfid_str, gfid);
+			if (ret < 0)
+				goto invalid;
+		}
+	}
+
 	qb_inode = qb_inode_ctx_get (this, inode);
 	if (!qb_inode)
 		qb_inode = GF_CALLOC (1, sizeof (*qb_inode),
 				      gf_qb_mt_qb_inode_t);
-	if (!qb_inode)
+	if (!qb_inode) {
+		GF_FREE(formatstr);
 		return ENOMEM;
+	}
 
-	strncpy (qb_inode->fmt, fmt, QB_XATTR_VAL_MAX);
+	strncpy(qb_inode->fmt, formatstr, QB_XATTR_VAL_MAX);
 	qb_inode->size = size;
-	qb_inode->size_str = s;
+
+	/*
+	 * If a backing gfid was not specified, interpret any remaining bytes
+	 * associated with a backing image as a filename local to the parent
+	 * directory. The format processing will validate further.
+	 */
+	if (!uuid_is_null(gfid))
+		uuid_copy(qb_inode->backing_gfid, gfid);
+	else if (s)
+		qb_inode->backing_fname = gf_strdup(s);
 
 	inode_ctx_set (inode, this, (void *)&qb_inode);
+
+	GF_FREE(formatstr);
+
 	return 0;
+
 invalid:
+	GF_FREE(formatstr);
+
 	gf_log (this->name, GF_LOG_WARNING,
 		"invalid format '%s' in inode %s", format,
 		uuid_utoa (inode->gfid));
@@ -191,6 +225,15 @@ qb_lookup_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 	if (op_ret == -1)
 		goto out;
 
+	/*
+	 * Cache the root inode for dealing with backing images. The format
+	 * coroutine and the gluster qemu backend driver both use the root inode
+	 * table to verify and/or redirect I/O to the backing image via
+	 * anonymous fd's.
+	 */
+	if (!conf->root_inode && __is_root_gfid(inode->gfid))
+		conf->root_inode = inode_ref(inode);
+
 	if (!xdata)
 		goto out;
 
@@ -249,6 +292,7 @@ qb_setxattr_format (call_frame_t *frame, xlator_t *this, call_stub_t *stub,
 	int op_errno = 0;
 	qb_local_t *qb_local = NULL;
 	data_t *data = NULL;
+	qb_inode_t *qb_inode;
 
 	if (!(data = dict_get (xattr, "trusted.glusterfs.block-format"))) {
 		QB_STUB_RESUME (stub);
@@ -264,12 +308,15 @@ qb_setxattr_format (call_frame_t *frame, xlator_t *this, call_stub_t *stub,
 		QB_STUB_UNWIND (stub, -1, op_errno);
 		return 0;
 	}
+	qb_inode = qb_inode_ctx_get(this, inode);
 
 	qb_local = frame->local;
 
 	qb_local->stub = stub;
 	qb_local->inode = inode_ref (inode);
-	strncpy (qb_local->fmt, format, QB_XATTR_VAL_MAX);
+
+	snprintf(qb_local->fmt, QB_XATTR_VAL_MAX, "%s:%lu", qb_inode->fmt,
+		 qb_inode->size);
 
 	qb_coroutine (frame, qb_format_and_resume);
 
@@ -1041,6 +1088,8 @@ fini (xlator_t *this)
 
         this->private = NULL;
 
+	if (conf->root_inode)
+		inode_unref(conf->root_inode);
         GF_FREE (conf);
 
 	return;
diff --git a/xlators/features/qemu-block/src/qemu-block.h b/xlators/features/qemu-block/src/qemu-block.h
index a91adb1ed51..55e7c23ac58 100644
--- a/xlators/features/qemu-block/src/qemu-block.h
+++ b/xlators/features/qemu-block/src/qemu-block.h
@@ -36,15 +36,16 @@
 
 #define QB_XATTR_KEY_MAX 64
 
-#define QB_XATTR_VAL_MAX 32
+#define QB_XATTR_VAL_MAX 64
 
 
 typedef struct qb_inode {
 	char fmt[QB_XATTR_VAL_MAX]; /* this is only the format, not "format:size" */
 	size_t size; /* virtual size in bytes */
-	char *size_str; /* pointer into fmt[] after ":" where size begins */
 	BlockDriverState *bs;
 	int refcnt;
+	uuid_t backing_gfid;
+	char *backing_fname;
 } qb_inode_t;
 
 
@@ -53,6 +54,7 @@ typedef struct qb_conf {
 	struct syncenv *env;
 	char qb_xattr_key[QB_XATTR_KEY_MAX];
 	char *default_password;
+	inode_t *root_inode;
 } qb_conf_t;
 
 
@@ -71,6 +73,7 @@ void qb_local_free (xlator_t *this, qb_local_t *local);
 int qb_coroutine (call_frame_t *frame, synctask_fn_t fn);
 inode_t *qb_inode_from_filename (const char *filename);
 int qb_inode_to_filename (inode_t *inode, char *filename, int size);
+int qb_format_extract (xlator_t *this, char *format, inode_t *inode);
 
 qb_inode_t *qb_inode_ctx_get (xlator_t *this, inode_t *inode);