features/changelog: NSR Journal changes

* Sequential updates to journal
* Journal update in the FOP path
* NSR specific format changes

TBD
---
* POST-OP record

Change-Id: I5b21b7624ccb095295a0c69abf00866e0d6cd818
Signed-off-by: Venky Shankar <vshankar@redhat.com>

4 files changed, 443 insertions, 117 deletions

diff --git a/xlators/features/changelog/src/changelog-encoders.c b/xlators/features/changelog/src/changelog-encoders.c
index 8d45ee1ff..ecd598e4d 100644
--- a/xlators/features/changelog/src/changelog-encoders.c
+++ b/xlators/features/changelog/src/changelog-encoders.c
@@ -95,12 +95,21 @@ changelog_encode_write_xtra (changelog_write_data_t *cwd,
                 case CHANGELOG_OPT_REC_ENTRY:
                         data = &co->co_entry;
                         break;
+                case CHANGELOG_OPT_REC_NAME:
+                        data = co->co_entry.cef_bname;
+                        break;
                 case CHANGELOG_OPT_REC_ULL:
                         data = &co->co_number;
                         break;
                 case CHANGELOG_OPT_REC_UUID:
                         data = &co->co_uuid;
                         break;
+                case CHANGELOG_OPT_REC_INT32:
+                        data = &co->co_int32;
+                        break;
+                case CHANGELOG_OPT_REC_UINT32:
+                        data = &co->co_uint32;
+                        break;
                 }
 
                 if (co->co_convert)
diff --git a/xlators/features/changelog/src/changelog-helpers.h b/xlators/features/changelog/src/changelog-helpers.h
index 4c3d5405b..f8f254cf6 100644
--- a/xlators/features/changelog/src/changelog-helpers.h
+++ b/xlators/features/changelog/src/changelog-helpers.h
@@ -373,7 +373,10 @@ typedef enum {
         CHANGELOG_OPT_REC_FOP,
         CHANGELOG_OPT_REC_ULL,
         CHANGELOG_OPT_REC_UUID,
+        CHANGELOG_OPT_REC_NAME,
         CHANGELOG_OPT_REC_ENTRY,
+        CHANGELOG_OPT_REC_INT32,
+        CHANGELOG_OPT_REC_UINT32,
 } changelog_optional_rec_type_t;
 
 struct changelog_entry_fields {
@@ -404,6 +407,8 @@ typedef struct {
         union {
                 uuid_t                        co_uuid;
                 glusterfs_fop_t               co_fop;
+                int                           co_int32;
+                unsigned int                  co_uint32;
                 unsigned long long            co_number;
                 struct changelog_entry_fields co_entry;
         };
@@ -491,6 +496,22 @@ changelog_forget (xlator_t *this, inode_t *inode);
                 }                                       \
         } while (0)
 
+#define CHANGELOG_FILL_INT32(co, number, converter, xlen) do {  \
+                co->co_convert = converter;                     \
+                co->co_free = NULL;                             \
+                co->co_type = CHANGELOG_OPT_REC_INT32;          \
+                co->co_int32 = number;                          \
+                xlen += sizeof (int);                           \
+        } while (0)
+
+#define CHANGELOG_FILL_UINT32(co, number, converter, xlen) do { \
+                co->co_convert = converter;                     \
+                co->co_free = NULL;                             \
+                co->co_type = CHANGELOG_OPT_REC_UINT32;         \
+                co->co_uint32 = number;                         \
+                xlen += sizeof (unsigned int);                  \
+        } while (0)
+
 #define CHANGELOG_FILL_FOP_NUMBER(co, fop, converter, xlen) do { \
                 co->co_convert = converter;                      \
                 co->co_free = NULL;                              \
@@ -499,6 +520,18 @@ changelog_forget (xlator_t *this, inode_t *inode);
                 xlen += sizeof (fop);                            \
         } while (0)
 
+#define CHANGELOG_FILL_NAME(co, name, freefn, xlen, label)       \
+        do {                                                     \
+                co->co_convert = NULL;                           \
+                co->co_free = freefn;                            \
+                co->co_type = CHANGELOG_OPT_REC_NAME;            \
+                co->co_entry.cef_bname = gf_strdup(name);        \
+                if (!co->co_entry.cef_bname)                     \
+                        goto label;                              \
+                co->co_len = strlen (name);                      \
+                xlen += co->co_len;                              \
+        } while(0)                                               \
+
 #define CHANGELOG_FILL_ENTRY(co, pargfid, bname,                        \
                              converter, freefn, xlen, label)            \
         do {                                                            \
diff --git a/xlators/features/changelog/src/changelog.c b/xlators/features/changelog/src/changelog.c
index f152439ab..66fdd61be 100644
--- a/xlators/features/changelog/src/changelog.c
+++ b/xlators/features/changelog/src/changelog.c
@@ -469,8 +469,6 @@ changelog_fsetattr_cbk (call_frame_t *frame,
                                 preop_stbuf, postop_stbuf, xdata);
 
         return 0;
-
-
 }
 
 int32_t
diff --git a/xlators/features/changelog/src/policy/changelog-policy-replication.c b/xlators/features/changelog/src/policy/changelog-policy-replication.c
index 529c5c822..29c049716 100644
--- a/xlators/features/changelog/src/policy/changelog-policy-replication.c
+++ b/xlators/features/changelog/src/policy/changelog-policy-replication.c
@@ -19,6 +19,46 @@
 #define PRE_OP_MARK  0x5F4552505FULL   /* _PRE_ */
 #define POST_OP_MARK 0x5F54534F505FULL /* _POST_ */
 
+/* similar to fop_fn, but... */
+size_t
+int32_fn (void *data, char *buffer, gf_boolean_t encode)
+{
+        size_t bufsz = 0;
+        int    nr    = 0;
+        char buf[20] = {0,};
+
+        nr = *(int *) data;
+
+        if (encode) {
+                (void) snprintf (buf, sizeof (buf), "%d", nr);
+                CHANGELOG_FILL_BUFFER (buffer, bufsz, buf, strlen (buf));
+        } else
+                CHANGELOG_FILL_BUFFER (buffer, bufsz,
+                                       &nr, sizeof (int));
+
+        return bufsz;
+}
+
+
+size_t
+uint32_fn (void *data, char *buffer, gf_boolean_t encode)
+{
+        size_t       bufsz = 0;
+        unsigned int nr    = 0;
+        char buf[20]       = {0,};
+
+        nr = *(unsigned int *) data;
+
+        if (encode) {
+                (void) snprintf (buf, sizeof (buf), "%u", nr);
+                CHANGELOG_FILL_BUFFER (buffer, bufsz, buf, strlen (buf));
+        } else
+                CHANGELOG_FILL_BUFFER (buffer, bufsz,
+                                       &nr, sizeof (unsigned int));
+
+        return bufsz;
+}
+
 size_t
 number_fn (void *data, char *buffer, gf_boolean_t encode)
 {
@@ -145,9 +185,32 @@ changelog_fix_term(xlator_t *this,
         return _gf_true;
 }
 
+/**
+ * Replication policy records journal entries in the FOP path. This is
+ * quite different that the default policy (used by geo-replication),
+ * which journals records in the callback path on a successfull posix
+ * operation. Additionally, each record starts with a PRE-OP marker and
+ * the index number generated by the leader.
+ * (c.f. nsr_$NAME$() ~/xlator/cluster/nsr/nsr-server/src/all-templates.c)
+ *
+ * POST-OPs are marked asynchronously and not during in the callback path
+ * Marking it in the callback path is incorrect as the actual FOP may not
+ * have been synchronized to the disk. Therefore, POST op marking is done
+ * after a successful file system sync, which is trigerred periodically
+ * by NSR server component. To keep journal updates strictly sequential,
+ * POST-OPs are separate record in the journal.
+ */
 
-
-/** override FOPS */
+/**
+ * Override File Operations
+ *
+ * NOTE: Since journal updates are done in the FOP path, there is no
+ * actual use of @local in cbk. Therefore, @local could have been
+ * declared statically for each FOP (which would remove the overhead
+ * of allocating (here) and deallocating (in cbk)). Let's not do that
+ * and keep it this way for now. Will worry about it later (for code
+ * reasonability and performance).
+ */
 
 int32_t
 changelog_replication_rmdir (call_frame_t *frame, xlator_t *this,
@@ -155,11 +218,14 @@ changelog_replication_rmdir (call_frame_t *frame, xlator_t *this,
 {
         int                ret      = -1;
         size_t             xtra_len = 0;
+        changelog_priv_t  *priv     = NULL;
         changelog_opt_t   *co       = NULL;
         changelog_local_t *local    = NULL;
 
-        /* <PRE> + FOP + GFID + Entry */
-        CHANGELOG_INIT_NOCHECK (this, local, NULL, loc->inode->gfid, 4);
+        priv = this->private;
+
+        /* <PRE> + IDX + FOP + GFID + UID + GID + Entry */
+        CHANGELOG_INIT_NOCHECK (this, local, NULL, loc->inode->gfid, 7);
         if (!local)
                 goto out;
 
@@ -173,20 +239,31 @@ changelog_replication_rmdir (call_frame_t *frame, xlator_t *this,
         CHANGELOG_FILL_USIGNLL (co, PRE_OP_MARK, NULL, xtra_len);
         co++;
 
+        CHANGELOG_FILL_UINT32 (co, local->lu.val, uint32_fn, xtra_len);
+        co++;
+
         CHANGELOG_FILL_FOP_NUMBER (co, frame->root->op, fop_fn, xtra_len);
         co++;
 
         CHANGELOG_FILL_UUID (co, loc->inode->gfid, uuid_fn, xtra_len);
         co++;
 
+        CHANGELOG_FILL_INT32 (co, frame->root->uid, int32_fn, xtra_len);
+        co++;
+
+        CHANGELOG_FILL_INT32 (co, frame->root->gid, int32_fn, xtra_len);
+        co++;
+
         CHANGELOG_FILL_ENTRY (co, loc->pargfid, loc->name,
                               entry_fn, entry_free_fn, xtra_len, out);
 
-        changelog_set_usable_record_and_length (local, xtra_len, 4);
+        changelog_set_usable_record_and_length (local, xtra_len, 7);
 
         frame->local = local;
         ret = 0;
 
+        changelog_update (this, priv, local, CHANGELOG_TYPE_ENTRY);
+
  out:
         if (ret)
                 changelog_local_cleanup (this, local);
@@ -206,11 +283,14 @@ changelog_replication_rename (call_frame_t *frame, xlator_t *this,
 {
         int                ret      = -1;
         size_t             xtra_len = 0;
+        changelog_priv_t  *priv     = NULL;
         changelog_opt_t   *co       = NULL;
         changelog_local_t *local    = NULL;
 
-        /* <PRE> + FOP + GFID + OLDLOC + NEWLOC */
-        CHANGELOG_INIT_NOCHECK (this, local, NULL, oldloc->inode->gfid, 5);
+        priv = this->private;
+
+        /* <PRE> + IDX + FOP + GFID + UID + GID + OLDLOC + NEWLOC */
+        CHANGELOG_INIT_NOCHECK (this, local, NULL, oldloc->inode->gfid, 8);
         if (!local)
                 goto out;
 
@@ -224,12 +304,21 @@ changelog_replication_rename (call_frame_t *frame, xlator_t *this,
         CHANGELOG_FILL_USIGNLL (co, PRE_OP_MARK, NULL, xtra_len);
         co++;
 
+        CHANGELOG_FILL_UINT32 (co, local->lu.val, uint32_fn, xtra_len);
+        co++;
+
         CHANGELOG_FILL_FOP_NUMBER (co, frame->root->op, fop_fn, xtra_len);
         co++;
 
         CHANGELOG_FILL_UUID (co, oldloc->inode->gfid, uuid_fn, xtra_len);
         co++;
 
+        CHANGELOG_FILL_INT32 (co, frame->root->uid, int32_fn, xtra_len);
+        co++;
+
+        CHANGELOG_FILL_INT32 (co, frame->root->gid, int32_fn, xtra_len);
+        co++;
+
         CHANGELOG_FILL_ENTRY (co, oldloc->pargfid, oldloc->name,
                               entry_fn, entry_free_fn, xtra_len, out);
         co++;
@@ -237,11 +326,13 @@ changelog_replication_rename (call_frame_t *frame, xlator_t *this,
         CHANGELOG_FILL_ENTRY (co, newloc->pargfid, newloc->name,
                               entry_fn, entry_free_fn, xtra_len, out);
 
-        changelog_set_usable_record_and_length (local, xtra_len, 5);
+        changelog_set_usable_record_and_length (local, xtra_len, 8);
 
         frame->local = local;
         ret = 0;
 
+        changelog_update (this, priv, local, CHANGELOG_TYPE_ENTRY);
+
  out:
         if (ret)
                 changelog_local_cleanup (this, local);
@@ -255,11 +346,14 @@ changelog_replication_link (call_frame_t *frame,
 {
         int                ret      = -1;
         size_t             xtra_len = 0;
+        changelog_priv_t  *priv     = NULL;
         changelog_opt_t   *co       = NULL;
         changelog_local_t *local    = NULL;
 
-        /* <PRE> + FOP + GFID + Entry */
-        CHANGELOG_INIT_NOCHECK (this, local, NULL, oldloc->gfid, 4);
+        priv = this->private;
+
+        /* <PRE> + IDX + FOP + GFID + UID + GID + Entry */
+        CHANGELOG_INIT_NOCHECK (this, local, NULL, oldloc->gfid, 7);
         if (!local)
                 goto out;
 
@@ -273,20 +367,31 @@ changelog_replication_link (call_frame_t *frame,
         CHANGELOG_FILL_USIGNLL (co, PRE_OP_MARK, NULL, xtra_len);
         co++;
 
+        CHANGELOG_FILL_UINT32 (co, local->lu.val, uint32_fn, xtra_len);
+        co++;
+
         CHANGELOG_FILL_FOP_NUMBER (co, frame->root->op, fop_fn, xtra_len);
         co++;
 
         CHANGELOG_FILL_UUID (co, oldloc->gfid, uuid_fn, xtra_len);
         co++;
 
+        CHANGELOG_FILL_INT32 (co, frame->root->uid, int32_fn, xtra_len);
+        co++;
+
+        CHANGELOG_FILL_INT32 (co, frame->root->gid, int32_fn, xtra_len);
+        co++;
+
         CHANGELOG_FILL_ENTRY (co, newloc->pargfid, newloc->name,
                               entry_fn, entry_free_fn, xtra_len, out);
 
-        changelog_set_usable_record_and_length (local, xtra_len, 4);
+        changelog_set_usable_record_and_length (local, xtra_len, 7);
 
         frame->local = local;
         ret = 0;
 
+        changelog_update (this, priv, local, CHANGELOG_TYPE_ENTRY);
+
  out:
         if (ret)
                 changelog_local_cleanup (this, local);
@@ -302,9 +407,12 @@ changelog_replication_mkdir (call_frame_t *frame,
         uuid_t             gfid     = {0,};
         void              *uuid_req = NULL;
         size_t             xtra_len = 0;
+        changelog_priv_t  *priv     = NULL;
         changelog_opt_t   *co       = NULL;
         changelog_local_t *local    = NULL;
 
+        priv = this->private;
+
         ret = dict_get_ptr (xdata, "gfid-req", &uuid_req);
         if (ret) {
                 gf_log (this->name, GF_LOG_DEBUG,
@@ -315,8 +423,8 @@ changelog_replication_mkdir (call_frame_t *frame,
 
         ret = -1;
 
-        /* <PRE> + FOP + GFID + Entry */
-        CHANGELOG_INIT_NOCHECK (this, local, NULL, gfid, 4);
+        /* <PRE> + IDX + FOP + GFID + MODE + UID + GID + Entry */
+        CHANGELOG_INIT_NOCHECK (this, local, NULL, gfid, 8);
         if (!local)
                 goto out;
 
@@ -330,20 +438,34 @@ changelog_replication_mkdir (call_frame_t *frame,
         CHANGELOG_FILL_USIGNLL (co, PRE_OP_MARK, NULL, xtra_len);
         co++;
 
+        CHANGELOG_FILL_UINT32 (co, local->lu.val, uint32_fn, xtra_len);
+        co++;
+
         CHANGELOG_FILL_FOP_NUMBER (co, frame->root->op, fop_fn, xtra_len);
         co++;
 
         CHANGELOG_FILL_UUID (co, gfid, uuid_fn, xtra_len);
         co++;
 
+        CHANGELOG_FILL_UINT32 (co, mode | S_IFDIR, uint32_fn, xtra_len);
+        co++;
+
+        CHANGELOG_FILL_INT32 (co, frame->root->uid, int32_fn, xtra_len);
+        co++;
+
+        CHANGELOG_FILL_INT32 (co, frame->root->gid, int32_fn, xtra_len);
+        co++;
+
         CHANGELOG_FILL_ENTRY (co, loc->pargfid, loc->name,
                               entry_fn, entry_free_fn, xtra_len, out);
 
-        changelog_set_usable_record_and_length (local, xtra_len, 4);
+        changelog_set_usable_record_and_length (local, xtra_len, 8);
 
         frame->local = local;
         ret = 0;
 
+        changelog_update (this, priv, local, CHANGELOG_TYPE_ENTRY);
+
  out:
         if (ret)
                 changelog_local_cleanup (this, local);
@@ -359,9 +481,12 @@ changelog_replication_symlink (call_frame_t *frame, xlator_t *this,
         size_t             xtra_len = 0;
         uuid_t             gfid     = {0,};
         void              *uuid_req = NULL;
+        changelog_priv_t  *priv     = NULL;
         changelog_opt_t   *co       = NULL;
         changelog_local_t *local    = NULL;
 
+        priv = this->private;
+
         ret = dict_get_ptr (xdata, "gfid-req", &uuid_req);
         if (ret) {
                 gf_log (this->name, GF_LOG_DEBUG,
@@ -372,8 +497,8 @@ changelog_replication_symlink (call_frame_t *frame, xlator_t *this,
 
         ret = -1;
 
-        /* <PRE> + FOP + GFID + Entry */
-        CHANGELOG_INIT_NOCHECK (this, local, NULL, gfid, 4);
+        /* <PRE> + IDX + FOP + GFID + LINKNAME + UID + GID + Entry */
+        CHANGELOG_INIT_NOCHECK (this, local, NULL, gfid, 8);
         if (!local)
                 goto out;
 
@@ -387,20 +512,34 @@ changelog_replication_symlink (call_frame_t *frame, xlator_t *this,
         CHANGELOG_FILL_USIGNLL (co, PRE_OP_MARK, NULL, xtra_len);
         co++;
 
+        CHANGELOG_FILL_UINT32 (co, local->lu.val, uint32_fn, xtra_len);
+        co++;
+
         CHANGELOG_FILL_FOP_NUMBER (co, frame->root->op, fop_fn, xtra_len);
         co++;
 
         CHANGELOG_FILL_UUID (co, gfid, uuid_fn, xtra_len);
         co++;
 
+        CHANGELOG_FILL_NAME (co, linkname, entry_free_fn, xtra_len, out);
+        co++;
+
+        CHANGELOG_FILL_INT32 (co, frame->root->uid, int32_fn, xtra_len);
+        co++;
+
+        CHANGELOG_FILL_INT32 (co, frame->root->gid, int32_fn, xtra_len);
+        co++;
+
         CHANGELOG_FILL_ENTRY (co, loc->pargfid, loc->name,
                               entry_fn, entry_free_fn, xtra_len, out);
 
-        changelog_set_usable_record_and_length (local, xtra_len, 4);
+        changelog_set_usable_record_and_length (local, xtra_len, 8);
 
         frame->local = local;
         ret = 0;
 
+        changelog_update (this, priv, local, CHANGELOG_TYPE_ENTRY);
+
  out:
         if (ret)
                 changelog_local_cleanup (this, local);
@@ -417,9 +556,12 @@ changelog_replication_mknod (call_frame_t *frame,
         uuid_t             gfid     = {0,};
         void              *uuid_req = NULL;
         size_t             xtra_len = 0;
+        changelog_priv_t  *priv     = NULL;
         changelog_opt_t   *co       = NULL;
         changelog_local_t *local    = NULL;
 
+        priv = this->private;
+
         ret = dict_get_ptr (xdata, "gfid-req", &uuid_req);
         if (ret) {
                 gf_log (this->name, GF_LOG_DEBUG,
@@ -430,33 +572,50 @@ changelog_replication_mknod (call_frame_t *frame,
 
         ret = -1;
 
-        /* <PRE> + FOP + GFID + Entry */
-        CHANGELOG_INIT_NOCHECK (this, local, NULL, gfid, 4);
+        /* <PRE> + IDX + FOP + GFID + MODE + UID + GID + DEV + Entry */
+        CHANGELOG_INIT_NOCHECK (this, local, NULL, gfid, 9);
 
         co = changelog_get_usable_buffer (local);
         if (!co)
                 goto out;
 
-        if (changelog_fix_term(this, local, xdata) == _gf_false)
+        if (changelog_fix_term (this, local, xdata) == _gf_false)
                 goto out;
 
         CHANGELOG_FILL_USIGNLL (co, PRE_OP_MARK, NULL, xtra_len);
         co++;
 
+        CHANGELOG_FILL_UINT32 (co, local->lu.val, uint32_fn, xtra_len);
+        co++;
+
         CHANGELOG_FILL_FOP_NUMBER (co, frame->root->op, fop_fn, xtra_len);
         co++;
 
         CHANGELOG_FILL_UUID (co, gfid, uuid_fn, xtra_len);
         co++;
 
+        CHANGELOG_FILL_UINT32 (co, mode, uint32_fn, xtra_len);
+        co++;
+
+        CHANGELOG_FILL_INT32 (co, frame->root->uid, int32_fn, xtra_len);
+        co++;
+
+        CHANGELOG_FILL_INT32 (co, frame->root->gid, int32_fn, xtra_len);
+        co++;
+
+        CHANGELOG_FILL_USIGNLL (co, dev, number_fn, xtra_len);
+        co++;
+
         CHANGELOG_FILL_ENTRY (co, loc->pargfid, loc->name,
                               entry_fn, entry_free_fn, xtra_len, out);
 
-        changelog_set_usable_record_and_length (local, xtra_len, 4);
+        changelog_set_usable_record_and_length (local, xtra_len, 9);
 
         frame->local = local;
         ret = 0;
 
+        changelog_update (this, priv, local, CHANGELOG_TYPE_ENTRY);
+
  out:
         if (ret)
                 changelog_local_cleanup (this, local);
@@ -472,9 +631,12 @@ changelog_replication_create (call_frame_t *frame, xlator_t *this,
         uuid_t             gfid     = {0,};
         void              *uuid_req = NULL;
         size_t             xtra_len = 0;
+        changelog_priv_t  *priv     = NULL;
         changelog_opt_t   *co       = NULL;
         changelog_local_t *local    = NULL;
 
+        priv = this->private;
+
         ret = dict_get_ptr (xdata, "gfid-req", &uuid_req);
         if (ret) {
                 gf_log (this->name, GF_LOG_DEBUG,
@@ -485,60 +647,69 @@ changelog_replication_create (call_frame_t *frame, xlator_t *this,
 
         ret = -1;
 
-        /* <PRE> + FOP + GFID + Entry */
-        CHANGELOG_INIT_NOCHECK (this, local, NULL, gfid, 4);
+        /* <PRE> + IDX + FOP + GFID + MODE + UID + GID + ENTRY */
+        CHANGELOG_INIT_NOCHECK (this, local, NULL, gfid, 8);
 
         co = changelog_get_usable_buffer (local);
         if (!co)
                 goto out;
 
-        if (changelog_fix_term(this, local, xdata) == _gf_false)
+        if (changelog_fix_term (this, local, xdata) == _gf_false)
                 goto out;
 
         CHANGELOG_FILL_USIGNLL (co, PRE_OP_MARK, NULL, xtra_len);
         co++;
 
+        CHANGELOG_FILL_UINT32 (co, local->lu.val, uint32_fn, xtra_len);
+        co++;
+
         CHANGELOG_FILL_FOP_NUMBER (co, frame->root->op, fop_fn, xtra_len);
         co++;
 
         CHANGELOG_FILL_UUID (co, gfid, uuid_fn, xtra_len);
         co++;
 
-        CHANGELOG_FILL_ENTRY (co, loc->pargfid, loc->name,
-                              entry_fn, entry_free_fn, xtra_len, out);
+        CHANGELOG_FILL_UINT32 (co, mode, uint32_fn, xtra_len);
         co++;
 
-        /*
-         * This isn't really kosher (we should have a similar but separate
-         * mode_fn) but it should do for now.
-         */
-        CHANGELOG_FILL_FOP_NUMBER (co, mode, fop_fn, xtra_len);
+        CHANGELOG_FILL_INT32 (co, frame->root->uid, int32_fn, xtra_len);
         co++;
 
-        changelog_set_usable_record_and_length (local, xtra_len, 5);
+        CHANGELOG_FILL_INT32 (co, frame->root->gid, int32_fn, xtra_len);
+        co++;
+
+        CHANGELOG_FILL_ENTRY (co, loc->pargfid, loc->name,
+                              entry_fn, entry_free_fn, xtra_len, out);
+
+        changelog_set_usable_record_and_length (local, xtra_len, 8);
 
         frame->local = local;
         ret = 0;
 
+        changelog_update (this, priv, local, CHANGELOG_TYPE_ENTRY);
+
  out:
         if (ret)
                 changelog_local_cleanup (this, local);
         return ret;
 }
 
-int32_t
-changelog_replication_fsetattr (call_frame_t *frame,
-                                xlator_t *this, fd_t *fd,
-                                struct iatt *stbuf, int32_t valid,
-                                dict_t *xdata)
+static int
+_changelog_setattr_fill_common (call_frame_t *frame, xlator_t *this,
+                                int32_t attr, struct iatt *stbuf,
+                                uuid_t gfid, dict_t *xdata)
 {
-        int                ret      = -1;
-        size_t             xtra_len = 0;
-        changelog_opt_t   *co       = NULL;
-        changelog_local_t *local    = NULL;
+        int                ret        = -1;
+        size_t             xtra_len   = 0;
+        int                used_count = 0;
+        changelog_priv_t  *priv       = NULL;
+        changelog_opt_t   *co         = NULL;
+        changelog_local_t *local      = NULL;
+
+        priv = this->private;
 
-        /* <PRE> + FOP + GFID */
-        CHANGELOG_INIT_NOCHECK (this, local, NULL, fd->inode->gfid, 3);
+        used_count = 7;
+        CHANGELOG_INIT_NOCHECK (this, local, NULL, gfid, used_count);
         if (!local)
                 goto out;
 
@@ -549,78 +720,121 @@ changelog_replication_fsetattr (call_frame_t *frame,
         if (changelog_fix_term(this, local, xdata) == _gf_false)
                 goto out;
 
+        /**
+         * - <PRE>
+         * - IDX
+         * - FOP
+         * - GFID
+         * - Valid flag
+         *     GF_SET_ATTR_MODE [chmod]
+         *       ->ia_prot
+         *       ->ia_type
+         *     GF_SET_ATTR_UID | GF_SET_ATTR_GID [chown]
+         *       ->ia_uid
+         *       ->ia_gid
+         *     GF_SET_ATTR_ATIME | GF_SET_ATTR_MTIME [utimes]
+         *       ->ia_atime
+         *       ->ia_mtime
+         */
+
         CHANGELOG_FILL_USIGNLL (co, PRE_OP_MARK, NULL, xtra_len);
         co++;
 
+        CHANGELOG_FILL_UINT32 (co, local->lu.val, uint32_fn, xtra_len);
+        co++;
+
         CHANGELOG_FILL_FOP_NUMBER (co, frame->root->op, fop_fn, xtra_len);
         co++;
 
-        CHANGELOG_FILL_UUID (co, fd->inode->gfid, uuid_fn, xtra_len);
+        CHANGELOG_FILL_UUID (co, gfid, uuid_fn, xtra_len);
+        co++;
 
-        changelog_set_usable_record_and_length (local, xtra_len, 3);
+        CHANGELOG_FILL_UINT32 (co, attr, uint32_fn, xtra_len);
+        co++;
 
-        frame->local = local;
-        ret = 0;
+        if (attr & GF_SET_ATTR_MODE) {
+                mode_t mode = 0;
 
- out:
-        if (ret)
-                changelog_local_cleanup (this, local);
-        return ret;
-}
+                /* ->ia_prot & ->ia_type stored as a consolidated value */
+                used_count--;
+                mode = st_mode_from_ia (stbuf->ia_prot, stbuf->ia_type);
 
-int32_t
-changelog_replication_setattr (call_frame_t *frame,
-                               xlator_t *this, loc_t *loc,
-                               struct iatt *stbuf, int32_t valid, dict_t *xdata)
-{
-        int                ret      = -1;
-        size_t             xtra_len = 0;
-        changelog_opt_t   *co       = NULL;
-        changelog_local_t *local    = NULL;
+                CHANGELOG_FILL_UINT32 (co, mode, uint32_fn, xtra_len);
 
-        /* <PRE> + FOP + GFID */
-        CHANGELOG_INIT_NOCHECK (this, local, NULL, loc->inode->gfid, 3);
-        if (!local)
-                goto out;
+        } else if (attr & (GF_SET_ATTR_UID | GF_SET_ATTR_GID)) {
+                uid_t uid = -1;
+                gid_t gid = -1;
 
-        co = changelog_get_usable_buffer (local);
-        if (!co)
-                goto out;
+                if (attr & GF_SET_ATTR_UID)
+                        uid = stbuf->ia_uid;
+                if (attr & GF_SET_ATTR_GID)
+                        gid = stbuf->ia_gid;
 
-        if (changelog_fix_term(this, local, xdata) == _gf_false)
-                goto out;
+                /* ->ia_uid & ->ia_gid */
+                CHANGELOG_FILL_INT32 (co, uid, int32_fn, xtra_len);
+                co++;
 
-        CHANGELOG_FILL_USIGNLL (co, PRE_OP_MARK, NULL, xtra_len);
-        co++;
+                CHANGELOG_FILL_INT32 (co, gid, int32_fn, xtra_len);
 
-        CHANGELOG_FILL_FOP_NUMBER (co, frame->root->op, fop_fn, xtra_len);
-        co++;
+        } else if (attr & (GF_SET_ATTR_ATIME | GF_SET_ATTR_MTIME)) {
 
-        CHANGELOG_FILL_UUID (co, loc->inode->gfid, uuid_fn, xtra_len);
+                /* ->ia_atime & ->ia_mtime, need usecs? */
+                CHANGELOG_FILL_UINT32 (co,
+                                       stbuf->ia_atime, uint32_fn, xtra_len);
+                co++;
+
+                CHANGELOG_FILL_UINT32 (co,
+                                       stbuf->ia_mtime, uint32_fn, xtra_len);
+        }
 
-        changelog_set_usable_record_and_length (local, xtra_len, 3);
+        changelog_set_usable_record_and_length (local, xtra_len, used_count);
 
-        frame->local = local;
         ret = 0;
+        frame->local = local;
+
+        changelog_update (this, priv, frame->local, CHANGELOG_TYPE_METADATA);
 
  out:
         if (ret)
                 changelog_local_cleanup (this, local);
+
         return ret;
 }
 
 int32_t
+changelog_replication_fsetattr (call_frame_t *frame,
+                                xlator_t *this, fd_t *fd,
+                                struct iatt *stbuf, int32_t valid,
+                                dict_t *xdata)
+{
+        return _changelog_setattr_fill_common (frame, this, valid,
+                                               stbuf, fd->inode->gfid, xdata);
+}
+
+int32_t
+changelog_replication_setattr (call_frame_t *frame,
+                               xlator_t *this, loc_t *loc,
+                               struct iatt *stbuf, int32_t valid, dict_t *xdata)
+{
+        return _changelog_setattr_fill_common (frame, this, valid,
+                                               stbuf, loc->inode->gfid, xdata);
+}
+
+int32_t
 changelog_replication_fremovexattr (call_frame_t *frame, xlator_t *this,
                                     fd_t *fd, const char *name, dict_t *xdata)
 {
         int                ret      = -1;
         size_t             xtra_len = 0;
         changelog_opt_t   *co       = NULL;
+        changelog_priv_t  *priv     = NULL;
         changelog_local_t *local    = NULL;
         int32_t            xattr_op;
 
-        /* <PRE> + FOP + GFID */
-        CHANGELOG_INIT_NOCHECK (this, local, NULL, fd->inode->gfid, 3);
+        priv = this->private;
+
+        /* <PRE> + IDX + FOP + GFID */
+        CHANGELOG_INIT_NOCHECK (this, local, NULL, fd->inode->gfid, 4);
         if (!local)
                 goto out;
 
@@ -634,23 +848,26 @@ changelog_replication_fremovexattr (call_frame_t *frame, xlator_t *this,
         CHANGELOG_FILL_USIGNLL (co, PRE_OP_MARK, NULL, xtra_len);
         co++;
 
-        if (dict_get_int32(xdata, "recon-xattr-opcode", &xattr_op) == 0) {
+        CHANGELOG_FILL_UINT32 (co, local->lu.val, uint32_fn, xtra_len);
+        co++;
+
+        if (dict_get_int32(xdata, "recon-xattr-opcode", &xattr_op) == 0)
                 CHANGELOG_FILL_FOP_NUMBER (co, (glusterfs_fop_t)xattr_op,
                                                 fop_fn, xtra_len);
-        }
-        else {
+        else
                 CHANGELOG_FILL_FOP_NUMBER (co, frame->root->op, fop_fn,
                                            xtra_len);
-        }
         co++;
 
         CHANGELOG_FILL_UUID (co, fd->inode->gfid, uuid_fn, xtra_len);
 
-        changelog_set_usable_record_and_length (local, xtra_len, 3);
+        changelog_set_usable_record_and_length (local, xtra_len, 4);
 
         frame->local = local;
         ret = 0;
 
+        changelog_update (this, priv, local, CHANGELOG_TYPE_METADATA);
+
  out:
         if (ret)
                 changelog_local_cleanup (this, local);
@@ -663,11 +880,15 @@ changelog_replication_removexattr (call_frame_t *frame, xlator_t *this,
 {
         int                ret      = -1;
         size_t             xtra_len = 0;
+        changelog_priv_t  *priv     = NULL;
         changelog_opt_t   *co       = NULL;
         changelog_local_t *local    = NULL;
         int32_t            xattr_op;
 
-        CHANGELOG_INIT_NOCHECK (this, local, NULL, loc->inode->gfid, 3);
+        priv = this->private;
+
+        /* <PRE> + IDX + FOP + GFID */
+        CHANGELOG_INIT_NOCHECK (this, local, NULL, loc->inode->gfid, 4);
         if (!local)
                 goto out;
 
@@ -681,23 +902,29 @@ changelog_replication_removexattr (call_frame_t *frame, xlator_t *this,
         CHANGELOG_FILL_USIGNLL (co, PRE_OP_MARK, NULL, xtra_len);
         co++;
 
-        if (dict_get_int32(xdata, "recon-xattr-opcode", &xattr_op) == 0) {
+        CHANGELOG_FILL_UINT32 (co, local->lu.val, uint32_fn, xtra_len);
+        co++;
+
+        CHANGELOG_FILL_UINT32 (co, local->lu.val, uint32_fn, xtra_len);
+        co++;
+
+        if (dict_get_int32(xdata, "recon-xattr-opcode", &xattr_op) == 0)
                 CHANGELOG_FILL_FOP_NUMBER (co, (glusterfs_fop_t)xattr_op,
                                            fop_fn, xtra_len);
-        }
-        else {
+        else
                 CHANGELOG_FILL_FOP_NUMBER (co, frame->root->op, fop_fn,
                                            xtra_len);
-        }
         co++;
 
         CHANGELOG_FILL_UUID (co, loc->inode->gfid, uuid_fn, xtra_len);
 
-        changelog_set_usable_record_and_length (local, xtra_len, 3);
+        changelog_set_usable_record_and_length (local, xtra_len, 4);
 
         frame->local = local;
         ret = 0;
 
+        changelog_update (this, priv, local, CHANGELOG_TYPE_METADATA);
+
  out:
         if (ret)
                 changelog_local_cleanup (this, local);
@@ -711,12 +938,15 @@ changelog_replication_setxattr (call_frame_t *frame,
 {
         int                ret      = -1;
         size_t             xtra_len = 0;
+        changelog_priv_t  *priv     = NULL;
         changelog_opt_t   *co       = NULL;
         changelog_local_t *local    = NULL;
         int32_t            xattr_op;
 
-        /* <PRE> + FOP + GFID */
-        CHANGELOG_INIT_NOCHECK (this, local, NULL, loc->inode->gfid, 3);
+        priv = this->private;
+
+        /* <PRE> + IDX + FOP + GFID */
+        CHANGELOG_INIT_NOCHECK (this, local, NULL, loc->inode->gfid, 4);
         if (!local)
                 goto out;
 
@@ -730,23 +960,26 @@ changelog_replication_setxattr (call_frame_t *frame,
         CHANGELOG_FILL_USIGNLL (co, PRE_OP_MARK, NULL, xtra_len);
         co++;
 
-        if (dict_get_int32(xdata, "recon-xattr-opcode", &xattr_op) == 0) {
+        CHANGELOG_FILL_UINT32 (co, local->lu.val, uint32_fn, xtra_len);
+        co++;
+
+        if (dict_get_int32(xdata, "recon-xattr-opcode", &xattr_op) == 0)
                 CHANGELOG_FILL_FOP_NUMBER (co, (glusterfs_fop_t)xattr_op,
                                            fop_fn, xtra_len);
-        }
-        else {
+        else
                 CHANGELOG_FILL_FOP_NUMBER (co, frame->root->op, fop_fn,
                                            xtra_len);
-        }
         co++;
 
         CHANGELOG_FILL_UUID (co, loc->inode->gfid, uuid_fn, xtra_len);
 
-        changelog_set_usable_record_and_length (local, xtra_len, 3);
+        changelog_set_usable_record_and_length (local, xtra_len, 4);
 
         frame->local = local;
         ret = 0;
 
+        changelog_update (this, priv, local, CHANGELOG_TYPE_METADATA);
+
  out:
         if (ret)
                 changelog_local_cleanup (this, local);
@@ -760,12 +993,15 @@ changelog_replication_fsetxattr (call_frame_t *frame,
 {
         int                ret      = -1;
         size_t             xtra_len = 0;
+        changelog_priv_t  *priv     = NULL;
         changelog_opt_t   *co       = NULL;
         changelog_local_t *local    = NULL;
         int32_t            xattr_op;
 
-        /* <PRE> + FOP + GFID */
-        CHANGELOG_INIT_NOCHECK (this, local, NULL, fd->inode->gfid, 3);
+        priv = this->private;
+
+        /* <PRE> + IDX + FOP + GFID */
+        CHANGELOG_INIT_NOCHECK (this, local, NULL, fd->inode->gfid, 4);
         if (!local)
                 goto out;
 
@@ -779,24 +1015,26 @@ changelog_replication_fsetxattr (call_frame_t *frame,
         CHANGELOG_FILL_USIGNLL (co, PRE_OP_MARK, NULL, xtra_len);
         co++;
 
+        CHANGELOG_FILL_UINT32 (co, local->lu.val, uint32_fn, xtra_len);
+        co++;
 
-        if (dict_get_int32(xdata, "recon-xattr-opcode", &xattr_op) == 0) {
+        if (dict_get_int32(xdata, "recon-xattr-opcode", &xattr_op) == 0)
                 CHANGELOG_FILL_FOP_NUMBER (co, (glusterfs_fop_t)xattr_op,
                                            fop_fn, xtra_len);
-        }
-        else {
+        else
                 CHANGELOG_FILL_FOP_NUMBER (co, frame->root->op, fop_fn,
                                            xtra_len);
-        }
         co++;
 
         CHANGELOG_FILL_UUID (co, fd->inode->gfid, uuid_fn, xtra_len);
 
-        changelog_set_usable_record_and_length (local, xtra_len, 3);
+        changelog_set_usable_record_and_length (local, xtra_len, 4);
 
         frame->local = local;
         ret = 0;
 
+        changelog_update (this, priv, local, CHANGELOG_TYPE_METADATA);
+
  out:
         if (ret)
                 changelog_local_cleanup (this, local);
@@ -810,11 +1048,14 @@ changelog_replication_truncate (call_frame_t *frame,
 {
         int                ret      = -1;
         size_t             xtra_len = 0;
+        changelog_priv_t *priv = NULL;
         changelog_opt_t   *co       = NULL;
         changelog_local_t *local    = NULL;
 
-        /* <PRE> + FOP + GFID + Offset */
-        CHANGELOG_INIT_NOCHECK (this, local, NULL, loc->inode->gfid, 4);
+        priv = this->private;
+
+        /* <PRE> + IDX + FOP + GFID + Offset */
+        CHANGELOG_INIT_NOCHECK (this, local, NULL, loc->inode->gfid, 5);
         if (!local)
                 goto out;
 
@@ -828,6 +1069,9 @@ changelog_replication_truncate (call_frame_t *frame,
         CHANGELOG_FILL_USIGNLL (co, PRE_OP_MARK, NULL, xtra_len);
         co++;
 
+        CHANGELOG_FILL_UINT32 (co, local->lu.val, uint32_fn, xtra_len);
+        co++;
+
         CHANGELOG_FILL_FOP_NUMBER (co, frame->root->op, fop_fn, xtra_len);
         co++;
 
@@ -836,11 +1080,13 @@ changelog_replication_truncate (call_frame_t *frame,
 
         CHANGELOG_FILL_USIGNLL (co, offset, number_fn, xtra_len);
 
-        changelog_set_usable_record_and_length (local, xtra_len, 4);
+        changelog_set_usable_record_and_length (local, xtra_len, 5);
 
         frame->local = local;
         ret = 0;
 
+        changelog_update (this, priv, local, CHANGELOG_TYPE_DATA);
+
  out:
         if (ret)
                 changelog_local_cleanup (this, local);
@@ -854,11 +1100,14 @@ changelog_replication_ftruncate (call_frame_t *frame,
 {
         int                ret      = -1;
         size_t             xtra_len = 0;
+        changelog_priv_t *priv = NULL;
         changelog_opt_t   *co       = NULL;
         changelog_local_t *local    = NULL;
 
-        /* <PRE> + FOP + GFID + Offset */
-        CHANGELOG_INIT_NOCHECK (this, local, NULL, fd->inode->gfid, 4);
+        priv = this->private;
+
+        /* <PRE> + IDX + FOP + GFID + Offset */
+        CHANGELOG_INIT_NOCHECK (this, local, NULL, fd->inode->gfid, 5);
         if (!local)
                 goto out;
 
@@ -872,6 +1121,9 @@ changelog_replication_ftruncate (call_frame_t *frame,
         CHANGELOG_FILL_USIGNLL (co, PRE_OP_MARK, NULL, xtra_len);
         co++;
 
+        CHANGELOG_FILL_UINT32 (co, local->lu.val, uint32_fn, xtra_len);
+        co++;
+
         CHANGELOG_FILL_FOP_NUMBER (co, frame->root->op, fop_fn, xtra_len);
         co++;
 
@@ -880,11 +1132,13 @@ changelog_replication_ftruncate (call_frame_t *frame,
 
         CHANGELOG_FILL_USIGNLL (co, offset, number_fn, xtra_len);
 
-        changelog_set_usable_record_and_length (local, xtra_len, 4);
+        changelog_set_usable_record_and_length (local, xtra_len, 5);
 
         frame->local = local;
         ret = 0;
 
+        changelog_update (this, priv, local, CHANGELOG_TYPE_DATA);
+
  out:
         if (ret)
                 changelog_local_cleanup (this, local);
@@ -899,11 +1153,14 @@ changelog_replication_writev (call_frame_t *frame,
 {
         int                ret      = -1;
         size_t             xtra_len = 0;
+        changelog_priv_t *priv = NULL;
         changelog_opt_t   *co       = NULL;
         changelog_local_t *local    = NULL;
 
-        /* <PRE> + FOP + GFID + Offset + Length */
-        CHANGELOG_INIT_NOCHECK (this, local, NULL, fd->inode->gfid, 5);
+        priv = this->private;
+
+        /* <PRE> + IDX + FOP + GFID + Offset + Length */
+        CHANGELOG_INIT_NOCHECK (this, local, NULL, fd->inode->gfid, 6);
         if (!local)
                 goto out;
 
@@ -917,6 +1174,9 @@ changelog_replication_writev (call_frame_t *frame,
         CHANGELOG_FILL_USIGNLL (co, PRE_OP_MARK, NULL, xtra_len);
         co++;
 
+        CHANGELOG_FILL_UINT32 (co, local->lu.val, uint32_fn, xtra_len);
+        co++;
+
         CHANGELOG_FILL_FOP_NUMBER (co, frame->root->op, fop_fn, xtra_len);
         co++;
 
@@ -929,11 +1189,13 @@ changelog_replication_writev (call_frame_t *frame,
         CHANGELOG_FILL_USIGNLL (co, iov_length (vector, count),
                                number_fn, xtra_len);
 
-        changelog_set_usable_record_and_length (local, xtra_len, 5);
+        changelog_set_usable_record_and_length (local, xtra_len, 6);
 
         frame->local = local;
         ret = 0;
 
+        changelog_update (this, priv, local, CHANGELOG_TYPE_DATA);
+
  out:
         if (ret)
                 changelog_local_cleanup (this, local);
@@ -968,6 +1230,19 @@ changelog_replication_cops_open (xlator_t *this,
 }
 
 /**
+ * NO-OP changelog write (from changelog.c). Records are journaled
+ * in the FOP path.
+ */
+int
+changelog_replication_cops_write (xlator_t *this,
+                                  changelog_priv_t *priv, void *cpriv,
+                                  changelog_local_t *local,
+                                  changelog_log_type type)
+{
+        return 0;
+}
+
+/**
  * no implicit rollover
  */
 int
@@ -975,7 +1250,7 @@ changelog_replication_cops_rollover (xlator_t *this,
                                      changelog_priv_t *priv, void *cpriv,
                                      char *name, gf_boolean_t last)
 {
-        return changelog_replication_cops_open(this, priv, cpriv, name, last);
+        return changelog_replication_cops_open (this, priv, cpriv, name, last);
 }
 
 off_t
@@ -1024,6 +1299,14 @@ changelog_replication_policy_init (xlator_t *this,
                 return -1;
         }
 
+        cp->cpriv = GF_CALLOC (1, sizeof (off_t),
+                               gf_changelog_mt_fop_policy_t);
+        if (!cp->cpriv) {
+                GF_FREE (r_fops);
+                GF_FREE (r_cops);
+                return -1;
+        }
+
         /* no roll-over, one big fat journal per term */
         priv->rollover_time = 0;
 
@@ -1033,7 +1316,7 @@ changelog_replication_policy_init (xlator_t *this,
         /* no record header: extra data (via iobufs) are always persisted */
         priv->no_gfid_hdr = _gf_true;
 
-        priv->lockless_update = _gf_true;
+        priv->lockless_update = _gf_false;
 
         memcpy (r_fops, &changelog_default_fops, sizeof (struct xlator_fops));
         memcpy (r_cops, &changelog_default_cops, sizeof (struct changelog_ops));
@@ -1067,11 +1350,13 @@ changelog_replication_policy_init (xlator_t *this,
 
         /* overload cops */
         r_cops->open         = changelog_replication_cops_open;
+        r_cops->write        = changelog_replication_cops_write;
         r_cops->rollover     = changelog_replication_cops_rollover;
         r_cops->get_offset   = changelog_replication_cops_get_offset;
         r_cops->set_offset   = changelog_replication_cops_set_offset;
         r_cops->reset_offset = changelog_replication_cops_reset_offset;
 
+
         cp->fops = r_fops;
         cp->cops = r_cops;
 
@@ -1084,5 +1369,6 @@ changelog_replication_policy_fini (xlator_t *this,
 {
         GF_FREE (cp->fops);
         GF_FREE (cp->cops);
+        GF_FREE (cp->cpriv);
         return 0;
 }