[PATCH 1/2] s3/modules: New vfs module: cephfs

Sam Lang samlang at gmail.com
Tue Apr 16 16:25:30 MDT 2013


Implements a vfs module for cephfs, a distributed file system
with posix semantics, built on the ceph distributed object
storage layer.  The ceph vfs module interfaces to the libcephfs
userspace API, and is primarily a lightweight wrapper around
libcephfs, translating error codes and parameters as necessary.

Signed-off-by: Sam Lang <sam.lang at inktank.com>
---
 source3/Makefile.in           |    5 +
 source3/configure.in          |   31 +
 source3/modules/vfs_ceph.c    | 1256 +++++++++++++++++++++++++++++++++++++++++
 source3/modules/wscript_build |   10 +
 source3/wscript               |   15 +
 5 files changed, 1317 insertions(+)
 create mode 100644 source3/modules/vfs_ceph.c

diff --git a/source3/Makefile.in b/source3/Makefile.in
index 42332c6..bd2ddda 100644
--- a/source3/Makefile.in
+++ b/source3/Makefile.in
@@ -913,6 +913,7 @@ VFS_LINUX_XFS_SGID_OBJ = modules/vfs_linux_xfs_sgid.o
 VFS_TIME_AUDIT_OBJ = modules/vfs_time_audit.o
 VFS_MEDIA_HARMONY_OBJ = modules/vfs_media_harmony.o
 VFS_BTRFS_OBJ = modules/vfs_btrfs.o
+VFS_CEPH_OBJ = modules/vfs_ceph.o
 
 PAM_ERRORS_OBJ = ../libcli/auth/pam_errors.o
 PLAINTEXT_AUTH_OBJ = auth/pampass.o auth/pass_check.o $(PAM_ERRORS_OBJ)
@@ -2964,6 +2965,10 @@ bin/btrfs. at SHLIBEXT@: $(BINARY_PREREQS) $(VFS_BTRFS_OBJ)
 	@echo "Building plugin $@"
 	@$(SHLD_MODULE) $(VFS_BTRFS_OBJ)
 
+bin/ceph. at SHLIBEXT@: $(BINARY_PREREQS) $(VFS_CEPH_OBJ)
+	@echo "Building plugin $@"
+	@$(SHLD_MODULE) $(VFS_CEPH_OBJ)
+
 #########################################################
 ## IdMap NSS plugins
 
diff --git a/source3/configure.in b/source3/configure.in
index 88c71d4..d2aa215 100644
--- a/source3/configure.in
+++ b/source3/configure.in
@@ -1023,6 +1023,36 @@ if test x"$ac_cv_header_gpfs_gpl_h" = x"yes"; then
     default_shared_modules="$default_shared_modules vfs_gpfs"
 fi
 
+#############################
+# check if building with libcephfs
+# fail if --with-libcephfs specified but no headers found
+printf "%s" "checking for CephFS... "
+AC_CHECK_HEADERS(cephfs/libcephfs.h)
+
+save_LIBS="$LIBS"
+LIBS="$LIBS -lcephfs"
+AC_TRY_LINK([#include <cephfs/libcephfs.h>],
+	  [ceph_version(0, 0, 0)],
+	  samba_cv_HAVE_CEPHFS=yes,
+	  samba_cv_HAVE_CEPHFS=no
+	  )
+
+if test x"$ac_cv_header_cephfs_libcephfs_h" = x"yes" && test x"$samba_cv_HAVE_CEPHFS" = x"yes"; then
+    AC_DEFINE(HAVE_CEPH, 1, [Whether CEPH headers are available])
+    default_shared_modules="$default_shared_modules vfs_ceph"
+    save_CFLAGS="$CFLAGS"
+    save_LDFLAGS="$LDFLAGS"
+    save_LIBS="$save_LIBS -lcephfs"
+else
+    if test x"$CEPHFS_PATH_SPEC" = x"yes"; then
+        AC_MSG_ERROR([No valid libcephfs found in ${CEPHFS_PATH}])
+    fi
+fi
+
+CFLAGS="$save_CFLAGS"
+LDFLAGS="$save_LDFLAGS"
+LIBS="$save_LIBS"
+
 # Note that all the libunwind symbols in the API are defined to internal
 # platform-specific version, so we must include libunwind.h before checking
 # any of them.
@@ -6455,6 +6485,7 @@ SMB_MODULE(vfs_linux_xfs_sgid, \$(VFS_LINUX_XFS_SGID_OBJ), "bin/linux_xfs_sgid.$
 SMB_MODULE(vfs_time_audit, \$(VFS_TIME_AUDIT_OBJ), "bin/time_audit.$SHLIBEXT", VFS)
 SMB_MODULE(vfs_media_harmony, \$(VFS_MEDIA_HARMONY_OBJ), "bin/media_harmony.$SHLIBEXT", VFS)
 SMB_MODULE(vfs_btrfs, \$(VFS_BTRFS_OBJ), "bin/btrfs.$SHLIBEXT", VFS)
+SMB_MODULE(vfs_ceph, \$(VFS_CEPH_OBJ), "bin/ceph.$SHLIBEXT", VFS)
 
 SMB_SUBSYSTEM(VFS,smbd/vfs.o)
 
diff --git a/source3/modules/vfs_ceph.c b/source3/modules/vfs_ceph.c
new file mode 100644
index 0000000..13bcf9d
--- /dev/null
+++ b/source3/modules/vfs_ceph.c
@@ -0,0 +1,1256 @@
+/*
+   Unix SMB/CIFS implementation.
+   Wrap disk only vfs functions to sidestep dodgy compilers.
+   Copyright (C) Tim Potter 1998
+   Copyright (C) Jeremy Allison 2007
+   Copyright (C) Brian Chrisman 2011 <bchrisman at gmail.com>
+   Copyright (C) Richard Sharpe 2011 <realrichardsharpe at gmail.com>
+
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; either version 3 of the License, or
+   (at your option) any later version.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+/*
+ * This VFS only works with the libceph.so user-space client. It is not needed
+ * if you are using the kernel client or the FUSE client.
+ *
+ * Add the following smb.conf parameter to each share that will be hosted on
+ * Ceph:
+ *
+ *   vfs objects = ceph [any others you need go here]
+ */
+
+#include "includes.h"
+#include "smbd/smbd.h"
+#include <dirent.h>
+#include <sys/statvfs.h>
+#include "cephfs/libcephfs.h"
+#include "smbprofile.h"
+
+#undef DBGC_CLASS
+#define DBGC_CLASS DBGC_VFS
+
+/*
+ * Note, libceph's return code model is to return -errno! So we have to convert
+ * to what Samba expects, with is set errno to -return and return -1
+ */
+#define WRAP_RETURN(_res) \
+	errno = 0; \
+	if (_res < 0) { \
+		errno = -_res; \
+		return -1; \
+	} \
+	return _res \
+
+/*
+ * We mount only one file system and then all shares are assumed to be in that.
+ * FIXME: If we want to support more than one FS, then we have to deal with
+ * this differently.
+ *
+ * So, cmount tells us if we have been this way before and whether
+ * we need to mount ceph and cmount_cnt tells us how many times we have
+ * connected
+ */
+static struct ceph_mount_info * cmount = NULL;
+static uint32_t cmount_cnt = 0;
+
+/* Check for NULL pointer parameters in cephwrap_* functions */
+
+/* We don't want to have NULL function pointers lying around.  Someone
+   is sure to try and execute them.  These stubs are used to prevent
+   this possibility. */
+
+static int cephwrap_connect(struct vfs_handle_struct *handle,  const char *service, const char *user)
+{
+	int ret;
+	char buf[256];
+
+	const char * conf_file;
+
+	if (cmount) {
+		handle->data = cmount; /* We have been here before */
+		cmount_cnt++;
+		return 0;
+	}
+
+	conf_file = lp_parm_const_string(SNUM(handle->conn), "ceph", "config_file", NULL);
+
+	DEBUG(2, ( "[CEPH] calling: ceph_create\n" ));
+	ret = ceph_create(&cmount, NULL);
+	if (ret)
+		goto err_out;
+
+	if (conf_file) {
+		/* Override the config file */
+		DEBUG(2, ( "[CEPH] calling: ceph_conf_read_file\n" ));
+		ret = ceph_conf_read_file(cmount, conf_file);
+	} else {
+
+		DEBUG(2, ( "[CEPH] calling: ceph_conf_read_file with %s\n", conf_file));
+		ret = ceph_conf_read_file(cmount, NULL);
+	}
+
+	if (ret)
+		goto err_out;
+
+	DEBUG(2, ( "[CEPH] calling: ceph_conf_get\n" ));
+	ret = ceph_conf_get(cmount, "log file", buf, sizeof(buf));
+	if (ret < 0)
+		goto err_out;
+
+	DEBUG(2, ("[CEPH] calling: ceph_mount\n"));
+	ret = ceph_mount(cmount, NULL);
+	if (ret < 0)
+		goto err_out;
+
+
+	/*
+	 * encode mount context/state into our vfs/connection holding structure
+	 * cmount is a ceph_mount_t*
+	 */
+	handle->data = cmount;
+	cmount_cnt++;
+
+	return 0;
+
+err_out:
+	/*
+	 * Handle the error correctly. Ceph returns -errno.
+	 */
+	DEBUG(2, ("[CEPH] Error return: %s\n", strerror(-ret)));
+	WRAP_RETURN(ret);
+}
+
+static void cephwrap_disconnect(struct vfs_handle_struct *handle)
+{
+	if (!cmount) {
+		DEBUG(0, ("[CEPH] Error, ceph not mounted\n"));
+		return;
+	}
+
+	/* Should we unmount/shutdown? Only if the last disconnect? */
+	if (--cmount_cnt) {
+		DEBUG(10, ("[CEPH] Not shuting down CEPH because still more connections\n"));
+		return;
+	}
+
+	ceph_shutdown(cmount);
+
+	cmount = NULL;  /* Make it safe */
+}
+
+/* Disk operations */
+
+static uint64_t cephwrap_disk_free(struct vfs_handle_struct *handle,  const char *path, bool small_query, uint64_t *bsize,
+			       uint64_t *dfree, uint64_t *dsize)
+{
+	struct statvfs statvfs_buf;
+	int ret;
+
+	if (!(ret = ceph_statfs(handle->data, path, &statvfs_buf))) {
+		/*
+		 * Provide all the correct values.
+		 */
+		*bsize = statvfs_buf.f_bsize;
+		*dfree = statvfs_buf.f_bsize * statvfs_buf.f_bavail;
+		*dsize = statvfs_buf.f_bsize * statvfs_buf.f_blocks;
+		DEBUG(10, ("[CEPH] bsize: %lu, dfree: %lu, dsize: %lu\n",
+			*bsize, *dfree, *dsize));
+		return *dfree;
+	} else {
+		DEBUG(10, ("[CEPH] ceph_statfs returned %d\n", ret));
+		WRAP_RETURN(ret);
+	}
+}
+
+static int cephwrap_get_quota(struct vfs_handle_struct *handle,  enum SMB_QUOTA_TYPE qtype, unid_t id, SMB_DISK_QUOTA *qt)
+{
+	/* libceph: Ceph does not implement this */
+#if 0
+/* was ifdef HAVE_SYS_QUOTAS */
+	int ret;
+
+	ret = ceph_get_quota(handle->conn->connectpath, qtype, id, qt);
+
+	if (ret) {
+		errno = -ret;
+		ret = -1;
+	}
+
+	return ret;
+#else
+	errno = ENOSYS;
+	return -1;
+#endif
+}
+
+static int cephwrap_set_quota(struct vfs_handle_struct *handle,  enum SMB_QUOTA_TYPE qtype, unid_t id, SMB_DISK_QUOTA *qt)
+{
+	/* libceph: Ceph does not implement this */
+#if 0
+/* was ifdef HAVE_SYS_QUOTAS */
+	int ret;
+
+	ret = ceph_set_quota(handle->conn->connectpath, qtype, id, qt);
+	if (ret) {
+		errno = -ret;
+		ret = -1;
+	}
+
+	return ret;
+#else
+	WRAP_RETURN(-ENOSYS);
+#endif
+}
+
+static int cephwrap_statvfs(struct vfs_handle_struct *handle,  const char *path, vfs_statvfs_struct *statbuf)
+{
+	struct statvfs statvfs_buf;
+	int ret;
+
+	ret = ceph_statfs(handle->data, path, &statvfs_buf);
+	if (ret < 0) {
+		WRAP_RETURN(ret);
+	} else {
+		statbuf->OptimalTransferSize = statvfs_buf.f_frsize;
+		statbuf->BlockSize = statvfs_buf.f_bsize;
+		statbuf->TotalBlocks = statvfs_buf.f_blocks;
+		statbuf->BlocksAvail = statvfs_buf.f_bfree;
+		statbuf->UserBlocksAvail = statvfs_buf.f_bavail;
+		statbuf->TotalFileNodes = statvfs_buf.f_files;
+		statbuf->FreeFileNodes = statvfs_buf.f_ffree;
+		statbuf->FsIdentifier = statvfs_buf.f_fsid;
+		DEBUG(10, ("[CEPH] f_bsize: %ld, f_blocks: %ld, f_bfree: %ld, f_bavail: %ld\n",
+			statvfs_buf.f_bsize, statvfs_buf.f_blocks,
+			statvfs_buf.f_bfree, statvfs_buf.f_bavail));
+	}
+	return ret;
+}
+
+/* Directory operations */
+
+static DIR *cephwrap_opendir(struct vfs_handle_struct *handle,  const char *fname, const char *mask, uint32 attr)
+{
+	int ret = 0;
+	struct ceph_dir_result *result;
+	DEBUG(10, ("[CEPH] opendir(%p, %s)\n", handle, fname));
+
+	/* Returns NULL if it does not exist or there are problems ? */
+	ret = ceph_opendir(handle->data, fname, &result);
+	if (ret < 0) {
+		result = NULL;
+		errno = -ret; /* We return result which is NULL in this case */
+	}
+
+	DEBUG(10, ("[CEPH] opendir(...) = %d\n", ret));
+	return (DIR *) result;
+}
+
+static DIR *cephwrap_fdopendir(struct vfs_handle_struct *handle,
+			       struct files_struct *fsp,
+			       const char *mask,
+			       uint32 attributes)
+{
+	int ret = 0;
+	struct ceph_dir_result *result;
+	DEBUG(10, ("[CEPH] fdopendir(%p, %p)\n", handle, fsp));
+
+	ret = ceph_opendir(handle->data, fsp->fsp_name->base_name, &result);
+	if (ret < 0) {
+		result = NULL;
+		errno = -ret; /* We return result which is NULL in this case */
+	}
+
+	DEBUG(10, ("[CEPH] fdopendir(...) = %d\n", ret));
+	return (DIR *) result;
+}
+
+static struct dirent *cephwrap_readdir(struct vfs_handle_struct *handle,
+				       DIR *dirp,
+				       SMB_STRUCT_STAT *sbuf)
+{
+	struct dirent *result;
+
+	DEBUG(10, ("[CEPH] readdir(%p, %p)\n", handle, dirp));
+	result = ceph_readdir(handle->data, (struct ceph_dir_result *) dirp);
+	DEBUG(10, ("[CEPH] readdir(...) = %p\n", result));
+
+	/* Default Posix readdir() does not give us stat info.
+	 * Set to invalid to indicate we didn't return this info. */
+	if (sbuf)
+		SET_STAT_INVALID(*sbuf);
+	return result;
+}
+
+static void cephwrap_seekdir(struct vfs_handle_struct *handle, DIR *dirp, long offset)
+{
+	DEBUG(10, ("[CEPH] seekdir(%p, %p, %ld)\n", handle, dirp, offset));
+	ceph_seekdir(handle->data, (struct ceph_dir_result *) dirp, offset);
+}
+
+static long cephwrap_telldir(struct vfs_handle_struct *handle, DIR *dirp)
+{
+	long ret;
+	DEBUG(10, ("[CEPH] telldir(%p, %p)\n", handle, dirp));
+	ret = ceph_telldir(handle->data, (struct ceph_dir_result *) dirp);
+	DEBUG(10, ("[CEPH] telldir(...) = %ld\n", ret));
+	WRAP_RETURN(ret);
+}
+
+static void cephwrap_rewinddir(struct vfs_handle_struct *handle, DIR *dirp)
+{
+	DEBUG(10, ("[CEPH] rewinddir(%p, %p)\n", handle, dirp));
+	ceph_rewinddir(handle->data, (struct ceph_dir_result *) dirp);
+}
+
+static int cephwrap_mkdir(struct vfs_handle_struct *handle,  const char *path, mode_t mode)
+{
+	int result;
+	bool has_dacl = False;
+	char *parent = NULL;
+
+	DEBUG(10, ("[CEPH] mkdir(%p, %s)\n", handle, path));
+
+	if (lp_inherit_acls(SNUM(handle->conn))
+	    && parent_dirname(talloc_tos(), path, &parent, NULL)
+	    && (has_dacl = directory_has_default_acl(handle->conn, parent)))
+		mode = 0777;
+
+	TALLOC_FREE(parent);
+
+	result = ceph_mkdir(handle->data, path, mode);
+
+	/*
+	 * Note. This order is important
+	 */
+	if (result) {
+		WRAP_RETURN(result);
+	} else if (result == 0 && !has_dacl) {
+		/*
+		 * We need to do this as the default behavior of POSIX ACLs
+		 * is to set the mask to be the requested group permission
+		 * bits, not the group permission bits to be the requested
+		 * group permission bits. This is not what we want, as it will
+		 * mess up any inherited ACL bits that were set. JRA.
+		 */
+		int saved_errno = errno; /* We may get ENOSYS */
+		if ((SMB_VFS_CHMOD_ACL(handle->conn, path, mode) == -1) && (errno == ENOSYS))
+			errno = saved_errno;
+	}
+
+	return result;
+}
+
+static int cephwrap_rmdir(struct vfs_handle_struct *handle,  const char *path)
+{
+	int result;
+
+	DEBUG(10, ("[CEPH] rmdir(%p, %s)\n", handle, path));
+	result = ceph_rmdir(handle->data, path);
+	DEBUG(10, ("[CEPH] rmdir(...) = %d\n", result));
+	WRAP_RETURN(result);
+}
+
+static int cephwrap_closedir(struct vfs_handle_struct *handle, DIR *dirp)
+{
+	int result;
+
+	DEBUG(10, ("[CEPH] closedir(%p, %p)\n", handle, dirp));
+	result = ceph_closedir(handle->data, (struct ceph_dir_result *) dirp);
+	DEBUG(10, ("[CEPH] closedir(...) = %d\n", result));
+	WRAP_RETURN(result);
+}
+
+/* File operations */
+
+static int cephwrap_open(struct vfs_handle_struct *handle,
+			struct smb_filename *smb_fname,
+			files_struct *fsp, int flags, mode_t mode)
+{
+	int result = -ENOENT;
+	DEBUG(10, ("[CEPH] open(%p, %s, %p, %d, %d)\n", handle, smb_fname_str_dbg(smb_fname), fsp, flags, mode));
+
+	if (smb_fname->stream_name) {
+		goto out;
+	}
+
+	result = ceph_open(handle->data, smb_fname->base_name, flags, mode);
+out:
+	DEBUG(10, ("[CEPH] open(...) = %d\n", result));
+	WRAP_RETURN(result);
+}
+
+static int cephwrap_close(struct vfs_handle_struct *handle, files_struct *fsp)
+{
+	int result;
+
+	DEBUG(10, ("[CEPH] close(%p, %p)\n", handle, fsp));
+	result = ceph_close(handle->data, fsp->fh->fd);
+	DEBUG(10, ("[CEPH] close(...) = %d\n", result));
+
+	WRAP_RETURN(result);
+}
+
+static ssize_t cephwrap_read(struct vfs_handle_struct *handle, files_struct *fsp, void *data, size_t n)
+{
+	ssize_t result;
+
+	DEBUG(10, ("[CEPH] read(%p, %p, %p, %lu)\n", handle, fsp, data, n));
+
+	/* Using -1 for the offset means read/write rather than pread/pwrite */
+	result = ceph_read(handle->data, fsp->fh->fd, data, n, -1);
+	DEBUG(10, ("[CEPH] read(...) = %lu\n", result));
+	WRAP_RETURN(result);
+}
+
+static ssize_t cephwrap_pread(struct vfs_handle_struct *handle, files_struct *fsp, void *data,
+			size_t n, off_t offset)
+{
+	ssize_t result;
+
+	DEBUG(10, ("[CEPH] pread(%p, %p, %p, %lu, %lu)\n", handle, fsp, data, n, offset));
+
+	result = ceph_read(handle->data, fsp->fh->fd, data, n, offset);
+	DEBUG(10, ("[CEPH] pread(...) = %lu\n", result));
+	WRAP_RETURN(result);
+}
+
+
+static ssize_t cephwrap_write(struct vfs_handle_struct *handle, files_struct *fsp, const void *data, size_t n)
+{
+	ssize_t result;
+
+	DEBUG(10, ("[CEPH] write(%p, %p, %p, %lu)\n", handle, fsp, data, n));
+
+	result = ceph_write(handle->data, fsp->fh->fd, data, n, -1);
+
+	DEBUG(10, ("[CEPH] write(...) = %lu\n", result));
+	if (result < 0) {
+		WRAP_RETURN(result);
+	}
+	fsp->fh->pos += result;
+	return result;
+}
+
+static ssize_t cephwrap_pwrite(struct vfs_handle_struct *handle, files_struct *fsp, const void *data,
+			size_t n, off_t offset)
+{
+	ssize_t result;
+
+	DEBUG(10, ("[CEPH] pwrite(%p, %p, %p, %lu, %lu)\n", handle, fsp, data, n, offset));
+	result = ceph_write(handle->data, fsp->fh->fd, data, n, offset);
+	DEBUG(10, ("[CEPH] pwrite(...) = %lu\n", result));
+	WRAP_RETURN(result);
+}
+
+static off_t cephwrap_lseek(struct vfs_handle_struct *handle, files_struct *fsp, off_t offset, int whence)
+{
+	off_t result = 0;
+
+	DEBUG(10, ("[CEPH] cephwrap_lseek\n"));
+	/* Cope with 'stat' file opens. */
+	if (fsp->fh->fd != -1) {
+		result = ceph_lseek(handle->data, fsp->fh->fd, offset, whence);
+	}
+	WRAP_RETURN(result);
+}
+
+static ssize_t cephwrap_sendfile(struct vfs_handle_struct *handle, int tofd, files_struct *fromfsp, const DATA_BLOB *hdr,
+			off_t offset, size_t n)
+{
+	/*
+	 * We cannot support sendfile because libceph is in user space.
+	 */
+	DEBUG(10, ("[CEPH] cephwrap_sendfile\n"));
+	errno = ENOTSUP;
+	return -1;
+}
+
+static ssize_t cephwrap_recvfile(struct vfs_handle_struct *handle,
+			int fromfd,
+			files_struct *tofsp,
+			off_t offset,
+			size_t n)
+{
+	/*
+	 * We cannot support recvfile because libceph is in user space.
+	 */
+	DEBUG(10, ("[CEPH] cephwrap_recvfile\n"));
+	errno=ENOTSUP;
+	return -1;
+}
+
+static int cephwrap_rename(struct vfs_handle_struct *handle,
+			  const struct smb_filename *smb_fname_src,
+			  const struct smb_filename *smb_fname_dst)
+{
+	int result = -1;
+	DEBUG(10, ("[CEPH] cephwrap_rename\n"));
+	if (smb_fname_src->stream_name || smb_fname_dst->stream_name) {
+		errno = ENOENT;
+		return result;
+	}
+
+	result = ceph_rename(handle->data, smb_fname_src->base_name, smb_fname_dst->base_name);
+	WRAP_RETURN(result);
+}
+
+static int cephwrap_fsync(struct vfs_handle_struct *handle, files_struct *fsp)
+{
+	int result;
+	DEBUG(10, ("[CEPH] cephwrap_fsync\n"));
+	result = ceph_fsync(handle->data, fsp->fh->fd, false);
+	WRAP_RETURN(result);
+}
+
+static void cephwrap_init_stat_ex_from_stat(struct stat_ex *dst, const struct stat *src)
+{
+	ZERO_STRUCT(*dst);
+
+	dst->st_ex_dev = src->st_dev;
+	dst->st_ex_ino = src->st_ino;
+	dst->st_ex_mode = src->st_mode;
+	dst->st_ex_nlink = src->st_nlink;
+	dst->st_ex_uid = src->st_uid;
+	dst->st_ex_gid = src->st_gid;
+	dst->st_ex_rdev = src->st_rdev;
+	dst->st_ex_size = src->st_size;
+	dst->st_ex_atime.tv_sec = src->st_atime;
+	dst->st_ex_mtime.tv_sec = src->st_mtime;
+	dst->st_ex_ctime.tv_sec = src->st_ctime;
+	dst->st_ex_btime.tv_sec = src->st_mtime;
+	dst->st_ex_blksize = src->st_blksize;
+	dst->st_ex_blocks = src->st_blocks;
+}
+
+static int cephwrap_stat(struct vfs_handle_struct *handle,
+			struct smb_filename *smb_fname)
+{
+	int result = -1;
+	struct stat stbuf;
+
+	DEBUG(10, ("[CEPH] stat(%p, %s)\n", handle, smb_fname_str_dbg(smb_fname)));
+
+	if (smb_fname->stream_name) {
+		errno = ENOENT;
+		return result;
+	}
+
+	result = ceph_stat(handle->data, smb_fname->base_name, (struct stat *) &stbuf);
+	DEBUG(10, ("[CEPH] stat(...) = %d\n", result));
+	if (result < 0) {
+		WRAP_RETURN(result);
+	} else {
+		DEBUG(10, ("[CEPH]\tstbuf = {dev = %lu, ino = %lu, mode = 0x%x, nlink = %lu, "
+			   "uid = %d, gid = %d, rdev = %lu, size = %lu, blksize = %lu, "
+			   "blocks = %lu, atime = %lu, mtime = %lu, ctime = %lu}\n",
+			   stbuf.st_dev, stbuf.st_ino, stbuf.st_mode, stbuf.st_nlink,
+			   stbuf.st_uid, stbuf.st_gid, stbuf.st_rdev, stbuf.st_size, stbuf.st_blksize,
+			   stbuf.st_blocks, stbuf.st_atime, stbuf.st_mtime, stbuf.st_ctime));
+	}
+	cephwrap_init_stat_ex_from_stat(&(smb_fname->st), &stbuf);
+	DEBUG(10, ("[CEPH] mode = 0x%x\n", smb_fname->st.st_ex_mode));
+	return result;
+}
+
+static int cephwrap_fstat(struct vfs_handle_struct *handle, files_struct *fsp, SMB_STRUCT_STAT *sbuf)
+{
+	int result = -1;
+	struct stat stbuf;
+
+	DEBUG(10, ("[CEPH] fstat(%p, %d)\n", handle, fsp->fh->fd));
+	result = ceph_fstat(handle->data, fsp->fh->fd, (struct stat *) &stbuf);
+	DEBUG(10, ("[CEPH] fstat(...) = %d\n", result));
+	if (result < 0) {
+		WRAP_RETURN(result);
+	} else {
+		DEBUG(10, ("[CEPH]\tstbuf = {dev = %lu, ino = %lu, mode = 0x%x, nlink = %lu, "
+			   "uid = %d, gid = %d, rdev = %lu, size = %lu, blksize = %lu, "
+			   "blocks = %lu, atime = %lu, mtime = %lu, ctime = %lu}\n",
+			   stbuf.st_dev, stbuf.st_ino, stbuf.st_mode, stbuf.st_nlink,
+			   stbuf.st_uid, stbuf.st_gid, stbuf.st_rdev, stbuf.st_size, stbuf.st_blksize,
+			   stbuf.st_blocks, stbuf.st_atime, stbuf.st_mtime, stbuf.st_ctime));
+	}
+
+	cephwrap_init_stat_ex_from_stat(sbuf, &stbuf);
+	DEBUG(10, ("[CEPH] mode = 0x%x\n", sbuf->st_ex_mode));
+	return result;
+}
+
+static int cephwrap_lstat(struct vfs_handle_struct *handle,
+			 struct smb_filename *smb_fname)
+{
+	int result = -1;
+	struct stat stbuf;
+
+	DEBUG(10, ("[CEPH] lstat(%p, %s)\n", handle, smb_fname_str_dbg(smb_fname)));
+
+	if (smb_fname->stream_name) {
+		errno = ENOENT;
+		return result;
+	}
+
+	result = ceph_lstat(handle->data, smb_fname->base_name, &stbuf);
+	DEBUG(10, ("[CEPH] lstat(...) = %d\n", result));
+	if (result < 0) {
+		WRAP_RETURN(result);
+	}
+	cephwrap_init_stat_ex_from_stat(&(smb_fname->st), &stbuf);
+	return result;
+}
+
+static int cephwrap_unlink(struct vfs_handle_struct *handle,
+			  const struct smb_filename *smb_fname)
+{
+	int result = -1;
+
+	DEBUG(10, ("[CEPH] unlink(%p, %s)\n", handle, smb_fname_str_dbg(smb_fname)));
+	if (smb_fname->stream_name) {
+		errno = ENOENT;
+		return result;
+	}
+	result = ceph_unlink(handle->data, smb_fname->base_name);
+	DEBUG(10, ("[CEPH] unlink(...) = %d\n", result));
+	WRAP_RETURN(result);
+}
+
+static int cephwrap_chmod(struct vfs_handle_struct *handle,  const char *path, mode_t mode)
+{
+	int result;
+
+	DEBUG(10, ("[CEPH] chmod(%p, %s, %d)\n", handle, path, mode));
+
+	/*
+	 * We need to do this due to the fact that the default POSIX ACL
+	 * chmod modifies the ACL *mask* for the group owner, not the
+	 * group owner bits directly. JRA.
+	 */
+
+
+	{
+		int saved_errno = errno; /* We might get ENOSYS */
+		if ((result = SMB_VFS_CHMOD_ACL(handle->conn, path, mode)) == 0) {
+			return result;
+		}
+		/* Error - return the old errno. */
+		errno = saved_errno;
+	}
+
+	result = ceph_chmod(handle->data, path, mode);
+	DEBUG(10, ("[CEPH] chmod(...) = %d\n", result));
+	WRAP_RETURN(result);
+}
+
+static int cephwrap_fchmod(struct vfs_handle_struct *handle, files_struct *fsp, mode_t mode)
+{
+	int result;
+
+	DEBUG(10, ("[CEPH] fchmod(%p, %p, %d)\n", handle, fsp, mode));
+
+	/*
+	 * We need to do this due to the fact that the default POSIX ACL
+	 * chmod modifies the ACL *mask* for the group owner, not the
+	 * group owner bits directly. JRA.
+	 */
+
+	{
+		int saved_errno = errno; /* We might get ENOSYS */
+		if ((result = SMB_VFS_FCHMOD_ACL(fsp, mode)) == 0) {
+			return result;
+		}
+		/* Error - return the old errno. */
+		errno = saved_errno;
+	}
+
+#if defined(HAVE_FCHMOD)
+	result = ceph_fchmod(handle->data, fsp->fh->fd, mode);
+	DEBUG(10, ("[CEPH] fchmod(...) = %d\n", result));
+	WRAP_RETURN(result);
+#else
+	errno = ENOSYS;
+#endif
+	return -1;
+}
+
+static int cephwrap_chown(struct vfs_handle_struct *handle, const char *path, uid_t uid, gid_t gid)
+{
+	int result;
+	DEBUG(10, ("[CEPH] chown(%p, %s, %d, %d)\n", handle, path, uid, gid));
+	result = ceph_chown(handle->data, path, uid, gid);
+	DEBUG(10, ("[CEPH] chown(...) = %d\n", result));
+	WRAP_RETURN(result);
+}
+
+static int cephwrap_fchown(struct vfs_handle_struct *handle, files_struct *fsp, uid_t uid, gid_t gid)
+{
+	int result;
+#ifdef HAVE_FCHOWN
+
+	DEBUG(10, ("[CEPH] fchown(%p, %p, %d, %d)\n", handle, fsp, uid, gid));
+	result = ceph_fchown(handle->data, fsp->fh->fd, uid, gid);
+	DEBUG(10, ("[CEPH] fchown(...) = %d\n", result));
+	WRAP_RETURN(result);
+#else
+	errno = ENOSYS;
+	result = -1;
+#endif
+	return result;
+}
+
+static int cephwrap_lchown(struct vfs_handle_struct *handle, const char *path, uid_t uid, gid_t gid)
+{
+	int result;
+
+	DEBUG(10, ("[CEPH] lchown(%p, %s, %d, %d)\n", handle, path, uid, gid));
+	result = ceph_lchown(handle->data, path, uid, gid);
+	DEBUG(10, ("[CEPH] lchown(...) = %d\n", result));
+	WRAP_RETURN(result);
+}
+
+static int cephwrap_chdir(struct vfs_handle_struct *handle,  const char *path)
+{
+	int result = -1;
+	DEBUG(10, ("[CEPH] chdir(%p, %s)\n", handle, path));
+	/*
+	 * If the path is just / use chdir because Ceph is below / and
+	 * cannot deal with changing directory above its mount point
+	 */
+	if (path && !strcmp(path, "/"))
+		return chdir(path);
+
+	result = ceph_chdir(handle->data, path);
+	DEBUG(10, ("[CEPH] chdir(...) = %d\n", result));
+	WRAP_RETURN(result);
+}
+
+static char *cephwrap_getwd(struct vfs_handle_struct *handle)
+{
+	const char *cwd = ceph_getcwd(handle->data);
+	DEBUG(10, ("[CEPH] getwd(%p) = %s\n", handle, cwd));
+	return strdup(cwd);
+}
+
+static int cephwrap_ntimes(struct vfs_handle_struct *handle,
+			 const struct smb_filename *smb_fname,
+			 struct smb_file_time *ft)
+{
+	struct utimbuf buf;
+	buf.actime = ft->atime.tv_sec;
+	buf.modtime = ft->mtime.tv_sec;
+	int result = ceph_utime(handle->data, smb_fname->base_name, &buf);
+	DEBUG(10, ("[CEPH] ntimes(%p, %s, {%ld, %ld, %ld, %ld}) = %d\n", handle, smb_fname_str_dbg(smb_fname),
+				ft->mtime.tv_sec, ft->atime.tv_sec, ft->ctime.tv_sec,
+				ft->create_time.tv_sec, result));
+	return result;
+}
+
+static int strict_allocate_ftruncate(struct vfs_handle_struct *handle, files_struct *fsp, off_t len)
+{
+	off_t space_to_write;
+	uint64_t space_avail;
+	uint64_t bsize,dfree,dsize;
+	int ret;
+	NTSTATUS status;
+	SMB_STRUCT_STAT *pst;
+
+	status = vfs_stat_fsp(fsp);
+	if (!NT_STATUS_IS_OK(status)) {
+		return -1;
+	}
+	pst = &fsp->fsp_name->st;
+
+#ifdef S_ISFIFO
+	if (S_ISFIFO(pst->st_ex_mode))
+		return 0;
+#endif
+
+	if (pst->st_ex_size == len)
+		return 0;
+
+	/* Shrink - just ftruncate. */
+	if (pst->st_ex_size > len)
+		return ftruncate(fsp->fh->fd, len);
+
+	space_to_write = len - pst->st_ex_size;
+
+	/* for allocation try fallocate first. This can fail on some
+	   platforms e.g. when the filesystem doesn't support it and no
+	   emulation is being done by the libc (like on AIX with JFS1). In that
+	   case we do our own emulation. fallocate implementations can
+	   return ENOTSUP or EINVAL in cases like that. */
+	ret = SMB_VFS_FALLOCATE(fsp, VFS_FALLOCATE_EXTEND_SIZE,
+				pst->st_ex_size, space_to_write);
+	if (ret == ENOSPC) {
+		errno = ENOSPC;
+		return -1;
+	}
+	if (ret == 0) {
+		return 0;
+	}
+	DEBUG(10,("[CEPH] strict_allocate_ftruncate: SMB_VFS_FALLOCATE failed with "
+		"error %d. Falling back to slow manual allocation\n", ret));
+
+	/* available disk space is enough or not? */
+	space_avail = get_dfree_info(fsp->conn,
+				     fsp->fsp_name->base_name, false,
+				     &bsize,&dfree,&dsize);
+	/* space_avail is 1k blocks */
+	if (space_avail == (uint64_t)-1 ||
+			((uint64_t)space_to_write/1024 > space_avail) ) {
+		errno = ENOSPC;
+		return -1;
+	}
+
+	/* Write out the real space on disk. */
+	ret = vfs_slow_fallocate(fsp, pst->st_ex_size, space_to_write);
+	if (ret != 0) {
+		errno = ret;
+		ret = -1;
+	}
+
+	return 0;
+}
+
+static int cephwrap_ftruncate(struct vfs_handle_struct *handle, files_struct *fsp, off_t len)
+{
+	int result = -1;
+	SMB_STRUCT_STAT st;
+	char c = 0;
+	off_t currpos;
+
+	DEBUG(10, ("[CEPH] ftruncate(%p, %p, %lu\n", handle, fsp, len));
+
+	if (lp_strict_allocate(SNUM(fsp->conn))) {
+		result = strict_allocate_ftruncate(handle, fsp, len);
+