Adding ChangeTimeRecorder(CTR) Xlator to GlusterFS

**********************************************************************
                            ChangeTimeRecorder(CTR) Xlator           |
**********************************************************************
ChangeTimeRecorder(CTR) is server side xlator(translator) which sits
just above posix xlator. The main role of this xlator is to record the
access/write patterns on a file residing the brick. It records the
read(only data) and write(data and metadata) times and also count on
how many times a file is read or written. This xlator also captures
the hard links to a file(as its required by data tiering to move
files).

CTR Xlator is the consumer of libgfdb.

To Enable/Disable CTR Xlator:
~~~~~~~~~~~~~~~~~~~~~~~~~~~~

 gluster volume set <volume-name> features.ctr-enabled {on/off}

To Enable/Disable Frequency Counter Recording in CTR Xlator:
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

 gluster volume set <volume-name> features.record-counters {on/off}

Change-Id: I5d3cf056af61ac8e3f8250321a27cb240a214ac2
BUG: 1194753
Signed-off-by: Joseph Fernandes <josferna@redhat.com>
Reviewed-on: http://review.gluster.org/9935
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Vijay Bellur <vbellur@redhat.com>

12 files changed, 2102 insertions, 8 deletions

diff --git a/xlators/features/Makefile.am b/xlators/features/Makefile.am
index 4a332555ffc..67d1338ed9e 100644
--- a/xlators/features/Makefile.am
+++ b/xlators/features/Makefile.am
@@ -1,5 +1,5 @@
 SUBDIRS = locks quota read-only mac-compat quiesce marker index barrier \
-          protect compress changelog ganesha gfid-access $(GLUPY_SUBDIR) qemu-block \
-          upcall snapview-client snapview-server trash # path-converter # filter
+          protect compress changelog changetimerecorder ganesha gfid-access $(GLUPY_SUBDIR) qemu-block \
+          upcall snapview-client snapview-server trash #path-converter # filter
 
 CLEANFILES =
diff --git a/xlators/features/changetimerecorder/Makefile.am b/xlators/features/changetimerecorder/Makefile.am
new file mode 100644
index 00000000000..a985f42a877
--- /dev/null
+++ b/xlators/features/changetimerecorder/Makefile.am
@@ -0,0 +1,3 @@
+SUBDIRS = src
+
+CLEANFILES =
diff --git a/xlators/features/changetimerecorder/src/Makefile.am b/xlators/features/changetimerecorder/src/Makefile.am
new file mode 100644
index 00000000000..b124728076f
--- /dev/null
+++ b/xlators/features/changetimerecorder/src/Makefile.am
@@ -0,0 +1,25 @@
+xlator_LTLIBRARIES = changetimerecorder.la
+xlatordir = $(libdir)/glusterfs/$(PACKAGE_VERSION)/xlator/features
+
+changetimerecorder_la_LDFLAGS = -module -avoid-version
+
+changetimerecorder_la_SOURCES = changetimerecorder.c ctr-helper.c
+
+changetimerecorder_la_LIBADD = $(top_builddir)/libglusterfs/src/libglusterfs.la\
+			$(top_builddir)/libglusterfs/src/gfdb/libgfdb.la\
+			$(top_builddir)/api/src/libgfapi.la\
+			$(top_builddir)/rpc/rpc-lib/src/libgfrpc.la\
+			-lsqlite3
+
+noinst_HEADERS = changetimerecorder.h ctr_mem_types.h ctr-helper.h
+
+AM_CPPFLAGS = $(GF_CPPFLAGS) -I$(top_srcdir)/libglusterfs/src \
+		-I$(top_srcdir)/libglusterfs/src/gfdb \
+		-I$(top_srcdir)/api/src \
+		-I$(top_srcdir)/rpc/rpc-lib/src \
+		-I$(top_srcdir)/rpc/xdr/src \
+		-DDATADIR=\"$(localstatedir)\"
+
+AM_CFLAGS = -Wall $(GF_CFLAGS)
+
+CLEANFILES =
diff --git a/xlators/features/changetimerecorder/src/changetimerecorder.c b/xlators/features/changetimerecorder/src/changetimerecorder.c
new file mode 100644
index 00000000000..4a3ee339676
--- /dev/null
+++ b/xlators/features/changetimerecorder/src/changetimerecorder.c
@@ -0,0 +1,1184 @@
+/*
+   Copyright (c) 2015 Red Hat, Inc. <http://www.redhat.com>
+   This file is part of GlusterFS.
+
+   This file is licensed to you under your choice of the GNU Lesser
+   General Public License, version 3 or any later version (LGPLv3 or
+   later), or the GNU General Public License, version 2 (GPLv2), in all
+   cases as published by the Free Software Foundation.
+*/
+#include <ctype.h>
+#include <sys/uio.h>
+
+#include "ctr-helper.h"
+
+/****************************WRITEV********************************************/
+int32_t
+ctr_writev_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                    int32_t op_ret, int32_t op_errno, struct iatt *prebuf,
+                    struct iatt *postbuf,
+                    dict_t *xdata)
+{
+        int ret = -1;
+
+        CTR_IS_DISABLED_THEN_GOTO(this, out);
+
+        ret = ctr_insert_unwind(frame, this,
+                        GFDB_FOP_INODE_WRITE, GFDB_FOP_UNWIND);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Failed inserting writev unwind");
+        }
+
+
+out:
+        STACK_UNWIND_STRICT (writev, frame, op_ret, op_errno, prebuf,
+                        postbuf, xdata);
+
+        return 0;
+}
+
+int32_t
+ctr_writev (call_frame_t *frame, xlator_t *this, fd_t *fd,
+            struct iovec *vector, int32_t count, off_t off,
+            uint32_t flags,
+            struct iobref *iobref, dict_t *xdata)
+{
+        int ret = -1;
+        gf_ctr_inode_context_t ctr_inode_cx;
+        gf_ctr_inode_context_t *_inode_cx = &ctr_inode_cx;
+
+        CTR_IS_DISABLED_THEN_GOTO(this, out);
+
+        /*Fill ctr inode context*/
+        FILL_CTR_INODE_CONTEXT(_inode_cx, fd->inode->ia_type,
+                                fd->inode->gfid, NULL, NULL,
+                                GFDB_FOP_INODE_WRITE, GFDB_FOP_WIND);
+
+        /*record into the database*/
+        ret = ctr_insert_wind(frame, this, _inode_cx);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                                "Failed inserting writev wind");
+        }
+
+out:
+        STACK_WIND (frame, ctr_writev_cbk, FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->writev, fd, vector, count,
+                    off, flags, iobref, xdata);
+
+        return 0;
+}
+
+/******************************setattr*****************************************/
+
+int32_t
+ctr_setattr_cbk (call_frame_t *frame,
+                       void *cookie, xlator_t *this, int32_t op_ret,
+                       int32_t op_errno, struct iatt *preop_stbuf,
+                       struct iatt *postop_stbuf, dict_t *xdata)
+{
+
+        int ret = -1;
+
+        CTR_IS_DISABLED_THEN_GOTO(this, out);
+
+        ret = ctr_insert_unwind(frame, this,
+                        GFDB_FOP_INODE_WRITE, GFDB_FOP_UNWIND);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Failed inserting setattr unwind");
+        }
+
+out:
+        STACK_UNWIND_STRICT (setattr, frame, op_ret, op_errno, preop_stbuf,
+                       postop_stbuf, xdata);
+
+        return 0;
+}
+
+int32_t
+ctr_setattr (call_frame_t *frame,
+                   xlator_t *this, loc_t *loc,
+                   struct iatt *stbuf, int32_t valid, dict_t *xdata)
+{
+
+        int ret = -1;
+        gf_ctr_inode_context_t ctr_inode_cx;
+        gf_ctr_inode_context_t *_inode_cx = &ctr_inode_cx;
+
+        CTR_IS_DISABLED_THEN_GOTO(this, out);
+
+         /*Fill ctr inode context*/
+        FILL_CTR_INODE_CONTEXT(_inode_cx, loc->inode->ia_type,
+                loc->inode->gfid, NULL, NULL, GFDB_FOP_INODE_WRITE,
+                GFDB_FOP_WIND);
+
+        /*record into the database*/
+        ret = ctr_insert_wind(frame, this, _inode_cx);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                                "Failed inserting setattr wind");
+        }
+out:
+
+        STACK_WIND (frame, ctr_setattr_cbk, FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->setattr, loc, stbuf,
+                    valid, xdata);
+
+        return 0;
+}
+
+/****************************fremovexattr************************************/
+
+int32_t
+ctr_fremovexattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                          int32_t op_ret, int32_t op_errno, dict_t *xdata)
+{
+        int ret = -1;
+
+        CTR_IS_DISABLED_THEN_GOTO(this, out);
+
+
+        ret = ctr_insert_unwind(frame, this,
+                        GFDB_FOP_INODE_WRITE, GFDB_FOP_UNWIND);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Failed inserting fremovexattr unwind");
+        }
+
+out:
+        STACK_UNWIND_STRICT (fremovexattr, frame, op_ret, op_errno, xdata);
+
+        return 0;
+}
+
+int32_t
+ctr_fremovexattr (call_frame_t *frame, xlator_t *this, fd_t *fd,
+                      const char *name, dict_t *xdata)
+{
+        int ret = -1;
+        gf_ctr_inode_context_t ctr_inode_cx;
+        gf_ctr_inode_context_t *_inode_cx = &ctr_inode_cx;
+
+        CTR_IS_DISABLED_THEN_GOTO(this, out);
+
+
+         /*Fill ctr inode context*/
+        FILL_CTR_INODE_CONTEXT(_inode_cx, fd->inode->ia_type,
+                fd->inode->gfid, NULL, NULL, GFDB_FOP_INODE_WRITE,
+                GFDB_FOP_WIND);
+
+        /*record into the database*/
+        ret = ctr_insert_wind(frame, this, _inode_cx);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                                "Failed inserting fremovexattr wind");
+        }
+
+out:
+        STACK_WIND (frame, ctr_fremovexattr_cbk, FIRST_CHILD (this),
+                    FIRST_CHILD (this)->fops->fremovexattr,
+                    fd, name, xdata);
+        return 0;
+}
+
+/****************************removexattr*************************************/
+
+int32_t
+ctr_removexattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                         int32_t op_ret, int32_t op_errno, dict_t *xdata)
+{
+        int ret = -1;
+
+        CTR_IS_DISABLED_THEN_GOTO(this, out);
+
+
+        ret = ctr_insert_unwind(frame, this,
+                        GFDB_FOP_INODE_WRITE, GFDB_FOP_UNWIND);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Failed inserting removexattr unwind");
+        }
+
+out:
+        STACK_UNWIND_STRICT (removexattr, frame, op_ret, op_errno, xdata);
+
+        return 0;
+}
+
+int32_t
+ctr_removexattr (call_frame_t *frame, xlator_t *this, loc_t *loc,
+                     const char *name, dict_t *xdata)
+{
+        int ret = -1;
+        gf_ctr_inode_context_t ctr_inode_cx;
+        gf_ctr_inode_context_t *_inode_cx = &ctr_inode_cx;
+
+        CTR_IS_DISABLED_THEN_GOTO(this, out);
+
+
+         /*Fill ctr inode context*/
+        FILL_CTR_INODE_CONTEXT(_inode_cx, loc->inode->ia_type,
+                loc->inode->gfid, NULL, NULL, GFDB_FOP_INODE_WRITE,
+                GFDB_FOP_WIND);
+
+        /*record into the database*/
+        ret = ctr_insert_wind(frame, this, _inode_cx);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                                "Failed inserting removexattr wind");
+        }
+
+out:
+        STACK_WIND (frame, ctr_removexattr_cbk, FIRST_CHILD (this),
+                    FIRST_CHILD (this)->fops->removexattr,
+                    loc, name, xdata);
+        return 0;
+}
+
+/****************************truncate****************************************/
+
+int32_t
+ctr_truncate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                      int32_t op_ret, int32_t op_errno, struct iatt *prebuf,
+                      struct iatt *postbuf, dict_t *xdata)
+{
+        int ret = -1;
+
+        CTR_IS_DISABLED_THEN_GOTO(this, out);
+
+
+        ret = ctr_insert_unwind(frame, this,
+                        GFDB_FOP_INODE_WRITE, GFDB_FOP_UNWIND);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Failed inserting truncate unwind");
+        }
+
+
+out:
+        STACK_UNWIND_STRICT (truncate, frame, op_ret, op_errno, prebuf,
+                      postbuf, xdata);
+
+        return 0;
+}
+
+int32_t
+ctr_truncate (call_frame_t *frame, xlator_t *this, loc_t *loc,
+                  off_t offset, dict_t *xdata)
+{
+        int ret = -1;
+        gf_ctr_inode_context_t ctr_inode_cx;
+        gf_ctr_inode_context_t *_inode_cx = &ctr_inode_cx;
+
+        CTR_IS_DISABLED_THEN_GOTO(this, out);
+
+         /*Fill ctr inode context*/
+        FILL_CTR_INODE_CONTEXT(_inode_cx, loc->inode->ia_type,
+                loc->inode->gfid, NULL, NULL, GFDB_FOP_INODE_WRITE,
+                GFDB_FOP_WIND);
+
+        /*record into the database*/
+        ret = ctr_insert_wind(frame, this, _inode_cx);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                                "Failed inserting truncate wind");
+        }
+out:
+        STACK_WIND (frame, ctr_truncate_cbk, FIRST_CHILD (this),
+                    FIRST_CHILD (this)->fops->truncate,
+                    loc, offset, xdata);
+        return 0;
+}
+
+/****************************ftruncate***************************************/
+
+int32_t
+ctr_ftruncate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                       int32_t op_ret, int32_t op_errno, struct iatt *prebuf,
+                       struct iatt *postbuf, dict_t *xdata)
+{
+        int ret = -1;
+
+        CTR_IS_DISABLED_THEN_GOTO(this, out);
+
+        ret = ctr_insert_unwind(frame, this,
+                        GFDB_FOP_INODE_WRITE, GFDB_FOP_UNWIND);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Failed inserting ftruncate unwind");
+        }
+
+out:
+        STACK_UNWIND_STRICT (ftruncate, frame, op_ret, op_errno, prebuf,
+                      postbuf, xdata);
+
+        return 0;
+}
+
+int32_t
+ctr_ftruncate (call_frame_t *frame, xlator_t *this, fd_t *fd,
+                   off_t offset, dict_t *xdata)
+{
+        int ret = -1;
+        gf_ctr_inode_context_t ctr_inode_cx;
+        gf_ctr_inode_context_t *_inode_cx = &ctr_inode_cx;
+
+        CTR_IS_DISABLED_THEN_GOTO(this, out);
+
+
+         /*Fill ctr inode context*/
+        FILL_CTR_INODE_CONTEXT(_inode_cx, fd->inode->ia_type,
+                fd->inode->gfid, NULL, NULL, GFDB_FOP_INODE_WRITE,
+                GFDB_FOP_WIND);
+
+        /*record into the database*/
+        ret = ctr_insert_wind(frame, this, _inode_cx);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                                "Failed inserting ftruncate wind");
+        }
+
+out:
+        STACK_WIND (frame, ctr_ftruncate_cbk, FIRST_CHILD (this),
+                    FIRST_CHILD (this)->fops->ftruncate,
+                    fd, offset, xdata);
+        return 0;
+}
+
+/****************************rename******************************************/
+
+int32_t
+ctr_rename_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                    int32_t op_ret, int32_t op_errno, struct iatt *buf,
+                    struct iatt *preoldparent, struct iatt *postoldparent,
+                    struct iatt *prenewparent, struct iatt *postnewparent,
+                    dict_t *xdata)
+{
+        int ret = -1;
+
+        CTR_IS_DISABLED_THEN_GOTO(this, out);
+
+        ret = ctr_insert_unwind(frame, this,
+                        GFDB_FOP_DENTRY_WRITE, GFDB_FOP_UNWIND);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Failed inserting rename unwind");
+        }
+
+out:
+        STACK_UNWIND_STRICT (rename, frame, op_ret, op_errno, buf,
+                             preoldparent, postoldparent, prenewparent,
+                             postnewparent,
+                             xdata);
+
+        return 0;
+}
+
+int32_t
+ctr_rename (call_frame_t *frame, xlator_t *this, loc_t *oldloc,
+                loc_t *newloc, dict_t *xdata)
+{
+        int ret = -1;
+        gf_ctr_inode_context_t ctr_inode_cx;
+        gf_ctr_inode_context_t *_inode_cx = &ctr_inode_cx;
+        gf_ctr_link_context_t new_link_cx, old_link_cx;
+        gf_ctr_link_context_t *_nlink_cx = &new_link_cx;
+        gf_ctr_link_context_t *_olink_cx = &old_link_cx;
+
+        CTR_IS_DISABLED_THEN_GOTO(this, out);
+
+        /*Fill old link context*/
+        FILL_CTR_LINK_CX(_olink_cx, oldloc->pargfid, oldloc->name,
+                        oldloc->path);
+
+        /*Fill new link context*/
+        FILL_CTR_LINK_CX(_nlink_cx, newloc->pargfid, newloc->name,
+                        newloc->path);
+
+         /*Fill ctr inode context*/
+        FILL_CTR_INODE_CONTEXT(_inode_cx, oldloc->inode->ia_type,
+                oldloc->inode->gfid, _nlink_cx, _olink_cx,
+                GFDB_FOP_DENTRY_WRITE, GFDB_FOP_WIND);
+
+        /*record into the database*/
+        ret = ctr_insert_wind(frame, this, _inode_cx);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                                "Failed inserting rename wind");
+        }
+
+out:
+        STACK_WIND (frame, ctr_rename_cbk, FIRST_CHILD (this),
+                    FIRST_CHILD (this)->fops->rename,
+                    oldloc, newloc, xdata);
+        return 0;
+}
+
+/****************************unlink******************************************/
+int32_t
+ctr_unlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                    int32_t op_ret, int32_t op_errno, struct iatt *preparent,
+                    struct iatt *postparent, dict_t *xdata)
+{
+        int ret                                 = -1;
+        uint32_t remaining_links                = -1;
+
+        CTR_IS_DISABLED_THEN_GOTO(this, out);
+
+        if (!xdata)
+                goto out;
+
+        /*
+         *
+         * Extracting CTR_RESPONSE_LINK_COUNT_XDATA from POSIX Xlator
+         *
+         * */
+        ret = dict_get_uint32 (xdata , CTR_RESPONSE_LINK_COUNT_XDATA,
+                                &remaining_links);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Failed to getting CTR_RESPONSE_LINK_COUNT_XDATA");
+                remaining_links = -1;
+        }
+
+        /*As the xdata is no more required by CTR Xlator.*/
+        if (xdata) {
+                dict_unref (xdata);
+        }
+
+        /*This is not the only link*/
+        if (remaining_links != 1) {
+
+                ret = ctr_insert_unwind(frame, this, GFDB_FOP_DENTRY_WRITE,
+                                        GFDB_FOP_UNDEL);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "Failed inserting unlink unwind");
+                }
+        }
+        /*Last link that was deleted*/
+        else if (remaining_links == 1) {
+                ret = ctr_insert_unwind(frame, this, GFDB_FOP_DENTRY_WRITE,
+                                        GFDB_FOP_UNDEL_ALL);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "Failed inserting unlink unwind");
+                }
+        }
+
+
+
+out:
+        STACK_UNWIND_STRICT (unlink, frame, op_ret, op_errno, preparent,
+                        postparent, NULL);
+
+        return 0;
+}
+
+int32_t
+ctr_unlink (call_frame_t *frame, xlator_t *this,
+                loc_t *loc, int xflag, dict_t *xdata)
+{
+        int ret = -1;
+        gf_ctr_inode_context_t ctr_inode_cx;
+        gf_ctr_inode_context_t *_inode_cx = &ctr_inode_cx;
+        gf_ctr_link_context_t ctr_link_cx;
+        gf_ctr_link_context_t *_link_cx = &ctr_link_cx;
+
+        CTR_IS_DISABLED_THEN_GOTO(this, out);
+
+        /*Fill link context*/
+        FILL_CTR_LINK_CX(_link_cx, loc->pargfid, loc->name, loc->path);
+
+         /*Fill ctr inode context*/
+        FILL_CTR_INODE_CONTEXT(_inode_cx, loc->inode->ia_type,
+                loc->inode->gfid, _link_cx, NULL,
+                GFDB_FOP_DENTRY_WRITE, GFDB_FOP_WDEL);
+
+        /*record into the database*/
+        ret = ctr_insert_wind(frame, this, _inode_cx);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                                "Failed inserting unlink wind");
+        }
+
+
+
+        /*
+         *
+         * Sending CTR_REQUEST_LINK_COUNT_XDATA
+         * to POSIX Xlator to send link count in unwind path
+         *
+         * */
+        /*create xdata if NULL*/
+        if (!xdata)
+                xdata = dict_new();
+        if (!xdata) {
+                gf_log (this->name, GF_LOG_ERROR,
+                                "xdata is NULL :"
+                                "Cannot send CTR_REQUEST_LINK_COUNT_XDATA"
+                                "to posix");
+                goto out;
+        }
+
+        ret = dict_set_int32 (xdata, CTR_REQUEST_LINK_COUNT_XDATA, 1);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                                "Failed setting CTR_REQUEST_LINK_COUNT_XDATA");
+                if (xdata) {
+                        dict_unref (xdata);
+                }
+                goto out;
+        }
+
+
+
+out:
+        STACK_WIND (frame, ctr_unlink_cbk, FIRST_CHILD (this),
+                    FIRST_CHILD (this)->fops->unlink,
+                    loc, xflag, xdata);
+        return 0;
+}
+
+/****************************fsync******************************************/
+int32_t
+ctr_fsync_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                   int32_t op_ret, int32_t op_errno, struct iatt *prebuf,
+                   struct iatt *postbuf, dict_t *xdata)
+{
+        int ret = -1;
+
+        CTR_IS_DISABLED_THEN_GOTO(this, out);
+
+        ret = ctr_insert_unwind(frame, this, GFDB_FOP_INODE_WRITE,
+                                GFDB_FOP_UNWIND);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Failed inserting fsync unwind");
+        }
+
+out:
+        STACK_UNWIND_STRICT (fsync, frame, op_ret, op_errno, prebuf, postbuf,
+                      xdata);
+
+        return 0;
+}
+
+int32_t
+ctr_fsync (call_frame_t *frame, xlator_t *this, fd_t *fd,
+               int32_t flags, dict_t *xdata)
+{
+        int ret = -1;
+        gf_ctr_inode_context_t ctr_inode_cx;
+        gf_ctr_inode_context_t *_inode_cx = &ctr_inode_cx;
+
+        CTR_IS_DISABLED_THEN_GOTO(this, out);
+
+         /*Fill ctr inode context*/
+        FILL_CTR_INODE_CONTEXT(_inode_cx, fd->inode->ia_type,
+                fd->inode->gfid, NULL, NULL,
+                GFDB_FOP_INODE_WRITE, GFDB_FOP_WIND);
+
+        /*record into the database*/
+        ret = ctr_insert_wind(frame, this, _inode_cx);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                                "Failed inserting fsync wind");
+        }
+
+out:
+        STACK_WIND (frame, ctr_fsync_cbk, FIRST_CHILD (this),
+                    FIRST_CHILD (this)->fops->fsync,
+                    fd, flags, xdata);
+        return 0;
+}
+
+/****************************setxattr****************************************/
+
+int
+ctr_setxattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                  int32_t op_ret, int32_t op_errno, dict_t *xdata)
+{
+        int ret = -1;
+
+        CTR_IS_DISABLED_THEN_GOTO(this, out);
+
+        ret = ctr_insert_unwind(frame, this, GFDB_FOP_INODE_WRITE,
+                                GFDB_FOP_UNWIND);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Failed inserting fsync unwind");
+        }
+
+
+out:
+        STACK_UNWIND_STRICT (setxattr, frame, op_ret, op_errno, xdata);
+
+        return 0;
+}
+
+int
+ctr_setxattr (call_frame_t *frame, xlator_t *this,
+              loc_t *loc, dict_t *xattr, int flags, dict_t *xdata)
+{
+        int ret = -1;
+        gf_ctr_inode_context_t ctr_inode_cx;
+        gf_ctr_inode_context_t *_inode_cx = &ctr_inode_cx;
+
+        CTR_IS_DISABLED_THEN_GOTO(this, out);
+
+         /*Fill ctr inode context*/
+        FILL_CTR_INODE_CONTEXT(_inode_cx, loc->inode->ia_type,
+                loc->inode->gfid, NULL, NULL,
+                GFDB_FOP_INODE_WRITE, GFDB_FOP_WIND);
+
+        /*record into the database*/
+        ret = ctr_insert_wind(frame, this, _inode_cx);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                                "Failed inserting setxattr wind");
+        }
+
+out:
+        STACK_WIND (frame, ctr_setxattr_cbk, FIRST_CHILD (this),
+                    FIRST_CHILD (this)->fops->setxattr,
+                    loc, xattr, flags, xdata);
+        return 0;
+}
+
+/****************************mknod*******************************************/
+
+
+int32_t
+ctr_mknod_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                 int32_t op_ret, int32_t op_errno, inode_t *inode,
+                 struct iatt *buf, struct iatt *preparent,
+                 struct iatt *postparent, dict_t *xdata)
+{
+        int ret = -1;
+
+        CTR_IS_DISABLED_THEN_GOTO(this, out);
+
+        ret = ctr_insert_unwind(frame, this, GFDB_FOP_CREATE_WRITE,
+                                GFDB_FOP_UNWIND);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Failed inserting mknod unwind");
+        }
+
+out:
+        STACK_UNWIND_STRICT (mknod, frame, op_ret, op_errno, inode, buf,
+                preparent, postparent, xdata);
+
+        return 0;
+}
+
+
+int
+ctr_mknod (call_frame_t *frame, xlator_t *this,
+           loc_t *loc, mode_t mode, dev_t rdev, mode_t umask, dict_t *xdata)
+{
+        int ret = -1;
+        gf_ctr_inode_context_t ctr_inode_cx;
+        gf_ctr_inode_context_t *_inode_cx = &ctr_inode_cx;
+        gf_ctr_link_context_t  ctr_link_cx;
+        gf_ctr_link_context_t  *_link_cx = &ctr_link_cx;
+        void *uuid_req                          = NULL;
+        uuid_t gfid                             = {0,};
+        uuid_t *ptr_gfid                        = &gfid;
+
+        CTR_IS_DISABLED_THEN_GOTO(this, out);
+
+        GF_ASSERT(frame);
+        GF_ASSERT(frame->root);
+        CTR_IF_REBALANCE_FOP_THEN_GOTO (frame, out);
+        CTR_IF_INTERNAL_FOP_THEN_GOTO (frame, xdata, out);
+
+        /*get gfid from xdata dict*/
+        ret = dict_get_ptr (xdata, "gfid-req", &uuid_req);
+        if (ret) {
+                gf_log (this->name, GF_LOG_DEBUG,
+                        "failed to get gfid from dict");
+                goto out;
+        }
+        uuid_copy (gfid, uuid_req);
+
+        /*fill ctr link context*/
+        FILL_CTR_LINK_CX (_link_cx, loc->pargfid, loc->name, loc->path);
+
+         /*Fill ctr inode context*/
+        FILL_CTR_INODE_CONTEXT (_inode_cx, loc->inode->ia_type,
+                *ptr_gfid, _link_cx, NULL,
+                GFDB_FOP_CREATE_WRITE, GFDB_FOP_WIND);
+
+        /*record into the database*/
+        ret = ctr_insert_wind(frame, this, _inode_cx);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                                "Failed inserting mknod wind");
+        }
+
+out:
+        STACK_WIND (frame, ctr_mknod_cbk, FIRST_CHILD (this),
+                    FIRST_CHILD (this)->fops->mknod,
+                    loc, mode, rdev, umask, xdata);
+        return 0;
+}
+
+/****************************create******************************************/
+int
+ctr_create_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                int op_ret, int op_errno,
+                fd_t *fd, inode_t *inode, struct iatt *stbuf,
+                struct iatt *preparent, struct iatt *postparent,
+                dict_t *xdata)
+{
+        int ret = -1;
+
+        CTR_IS_DISABLED_THEN_GOTO(this, out);
+
+        ret = ctr_insert_unwind(frame, this, GFDB_FOP_CREATE_WRITE,
+                                GFDB_FOP_UNWIND);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Failed inserting create unwind");
+        }
+
+out:
+        STACK_UNWIND_STRICT (create, frame, op_ret, op_errno, fd, inode,
+                             stbuf,
+                        preparent, postparent, xdata);
+
+        return 0;
+}
+
+int
+ctr_create (call_frame_t *frame, xlator_t *this,
+            loc_t *loc, int32_t flags, mode_t mode,
+            mode_t umask, fd_t *fd, dict_t *xdata)
+{
+        int ret = -1;
+        gf_ctr_inode_context_t ctr_inode_cx;
+        gf_ctr_inode_context_t *_inode_cx = &ctr_inode_cx;
+        gf_ctr_link_context_t  ctr_link_cx;
+        gf_ctr_link_context_t  *_link_cx = &ctr_link_cx;
+        void             *uuid_req              = NULL;
+        uuid_t            gfid                  = {0,};
+        uuid_t            *ptr_gfid             = &gfid;
+
+        CTR_IS_DISABLED_THEN_GOTO(this, out);
+
+        /*Get GFID from Xdata dict*/
+        ret = dict_get_ptr (xdata, "gfid-req", &uuid_req);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "failed to get gfid from dict");
+                goto out;
+        }
+        uuid_copy (gfid, uuid_req);
+
+        /*fill ctr link context*/
+        FILL_CTR_LINK_CX(_link_cx, loc->pargfid, loc->name, loc->path);
+
+         /*Fill ctr inode context*/
+        FILL_CTR_INODE_CONTEXT(_inode_cx, loc->inode->ia_type,
+                *ptr_gfid, _link_cx, NULL,
+                GFDB_FOP_CREATE_WRITE, GFDB_FOP_WIND);
+
+        /*record into the database*/
+        ret = ctr_insert_wind(frame, this, &ctr_inode_cx);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                                "Failed inserting create wind");
+        }
+out:
+        STACK_WIND (frame, ctr_create_cbk, FIRST_CHILD (this),
+                    FIRST_CHILD (this)->fops->create,
+                    loc, flags, mode, umask, fd, xdata);
+        return 0;
+}
+
+/****************************link********************************************/
+
+int
+ctr_link_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+              int op_ret, int op_errno,
+              inode_t *inode, struct iatt *stbuf, struct iatt *preparent,
+              struct iatt *postparent, dict_t *xdata)
+{
+        int ret = -1;
+
+        CTR_IS_DISABLED_THEN_GOTO(this, out);
+
+        ret = ctr_insert_unwind(frame, this, GFDB_FOP_DENTRY_WRITE,
+                                GFDB_FOP_UNWIND);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Failed inserting create unwind");
+        }
+
+out:
+        STACK_UNWIND_STRICT (link, frame, op_ret, op_errno, inode, stbuf,
+                       preparent, postparent, xdata);
+        return 0;
+}
+
+int
+ctr_link (call_frame_t *frame, xlator_t *this,
+          loc_t *oldloc, loc_t *newloc, dict_t *xdata)
+{
+        int ret = -1;
+        gf_ctr_inode_context_t ctr_inode_cx;
+        gf_ctr_inode_context_t *_inode_cx = &ctr_inode_cx;
+        gf_ctr_link_context_t  ctr_link_cx;
+        gf_ctr_link_context_t  *_link_cx = &ctr_link_cx;
+
+        CTR_IS_DISABLED_THEN_GOTO(this, out);
+
+        CTR_IF_REBALANCE_FOP_THEN_GOTO (frame, out);
+
+        /*fill ctr link context*/
+        FILL_CTR_LINK_CX(_link_cx, newloc->pargfid, newloc->name,
+                        newloc->path);
+
+         /*Fill ctr inode context*/
+        FILL_CTR_INODE_CONTEXT(_inode_cx, oldloc->inode->ia_type,
+                oldloc->inode->gfid, _link_cx, NULL,
+                GFDB_FOP_DENTRY_WRITE, GFDB_FOP_WIND);
+
+        /*record into the database*/
+        ret = ctr_insert_wind(frame, this, _inode_cx);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                                "Failed inserting link wind");
+        }
+
+out:
+        STACK_WIND (frame, ctr_link_cbk, FIRST_CHILD (this),
+                    FIRST_CHILD (this)->fops->link,
+                    oldloc, newloc, xdata);
+        return 0;
+}
+
+/******************************readv*****************************************/
+int ctr_readv_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+               int op_ret, int op_errno,
+               struct iovec *vector, int count, struct iatt *stbuf,
+               struct iobref *iobref, dict_t *xdata) {
+
+        int ret = -1;
+
+        CTR_IS_DISABLED_THEN_GOTO(this, out);
+
+        ret = ctr_insert_unwind(frame, this, GFDB_FOP_INODE_READ,
+                                GFDB_FOP_UNWIND);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Failed inserting create unwind");
+        }
+
+out:
+        STACK_UNWIND_STRICT (readv, frame, op_ret, op_errno, vector, count,
+                                stbuf, iobref, xdata);
+        return 0;
+}
+
+
+int
+ctr_readv (call_frame_t *frame, xlator_t *this,
+           fd_t *fd, size_t size, off_t off, uint32_t flags, dict_t *xdata)
+{
+        int ret = -1;
+        gf_ctr_inode_context_t ctr_inode_cx;
+        gf_ctr_inode_context_t *_inode_cx = &ctr_inode_cx;
+
+        CTR_IS_DISABLED_THEN_GOTO(this, out);
+
+         /*Fill ctr inode context*/
+        FILL_CTR_INODE_CONTEXT(_inode_cx, fd->inode->ia_type,
+                fd->inode->gfid, NULL, NULL,
+                GFDB_FOP_INODE_READ, GFDB_FOP_WIND);
+
+        /*record into the database*/
+        ret = ctr_insert_wind(frame, this, _inode_cx);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                                "Failed inserting readv wind");
+        }
+
+out:
+        STACK_WIND (frame, ctr_readv_cbk, FIRST_CHILD (this),
+                    FIRST_CHILD (this)->fops->readv,
+                    fd, size, off, flags, xdata);
+        return 0;
+}
+/******************************************************************************/
+
+int
+reconfigure (xlator_t *this, dict_t *options)
+{
+        char *temp_str = NULL;
+        int ret = 0;
+        gf_ctr_private_t *_priv = NULL;
+
+        _priv = this->private;
+        if (dict_get_str(options, "changetimerecorder.frequency",
+                         &temp_str)) {
+                gf_log(this->name, GF_LOG_INFO, "set!");
+        }
+
+        GF_OPTION_RECONF ("ctr-enabled", _priv->enabled, options,
+                          bool, out);
+
+        GF_OPTION_RECONF ("record-counters", _priv->ctr_record_counter, options,
+                          bool, out);
+
+        GF_OPTION_RECONF ("record-exit", _priv->ctr_record_unwind, options,
+                          bool, out);
+
+        GF_OPTION_RECONF ("record-entry", _priv->ctr_record_wind, options,
+                          bool, out);
+
+out:
+
+        return ret;
+}
+
+/****************************init********************************************/
+
+int32_t
+init (xlator_t *this)
+{
+        gf_ctr_private_t *_priv = NULL;
+        int ret_db              = -1;
+        dict_t *params_dict      = NULL;
+
+        GF_VALIDATE_OR_GOTO ("ctr", this, error);
+
+        if (!this->children || this->children->next) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "FATAL: ctr should have exactly one child");
+                goto error;
+        }
+
+        if (!this->parents) {
+                gf_log (this->name, GF_LOG_WARNING,
+                        "dangling volume. check volfile ");
+        }
+
+        _priv = GF_CALLOC (1, sizeof (*_priv), gf_ctr_mt_private_t);
+        if (!_priv) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Calloc didnt work!!!");
+                goto error;
+        }
+
+        /*Default values for the translator*/
+        _priv->ctr_record_wind          = _gf_true;
+        _priv->ctr_record_unwind        = _gf_false;
+        _priv->ctr_hot_brick            = _gf_false;
+        _priv->gfdb_db_type             = GFDB_SQLITE3;
+        _priv->gfdb_sync_type           = GFDB_DB_SYNC;
+        _priv->enabled                  = _gf_true;
+        _priv->_db_conn                 = NULL;
+
+        /*Extract ctr xlator options*/
+        ret_db = extract_ctr_options (this, _priv);
+        if (ret_db) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Failed extracting ctr xlator options");
+                goto error;
+        }
+
+        params_dict = dict_new ();
+        if (!params_dict) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "DB Params cannot initialized!");
+                goto error;
+        }
+
+        /*Extract db params options*/
+        ret_db = extract_db_params(this, params_dict, _priv->gfdb_db_type);
+        if (ret_db) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Failed extracting db params options");
+                goto error;
+        }
+
+        /*Create a memory pool for ctr xlator*/
+        this->local_pool = mem_pool_new (gf_ctr_local_t, 64);
+        if (!this->local_pool) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "failed to create local memory pool");
+                goto error;
+        }
+
+        /*Initialize Database Connection*/
+        _priv->_db_conn = init_db(params_dict, _priv->gfdb_db_type);
+        if (!_priv->_db_conn) {
+                gf_log (this->name, GF_LOG_ERROR,
+                       "FATAL: Failed initializing data base");
+                        goto error;
+        }
+
+        ret_db = 0;
+        goto out;
+
+/*Error handling */
+error:
+
+        if (this)
+                mem_pool_destroy (this->local_pool);
+
+        if (_priv) {
+                GF_FREE (_priv->ctr_db_path);
+        }
+        GF_FREE (_priv);
+
+        if (params_dict)
+                dict_unref (params_dict);
+
+        return -1;
+
+out:
+
+        if (params_dict)
+                dict_unref (params_dict);
+
+        this->private = (void *)_priv;
+        return 0;
+}
+
+int32_t
+mem_acct_init (xlator_t *this)
+{
+        int     ret = -1;
+
+        GF_VALIDATE_OR_GOTO ("ctr", this, out);
+
+        ret = xlator_mem_acct_init (this, gf_ctr_mt_end + 1);
+
+        if (ret != 0) {
+                gf_log (this->name, GF_LOG_ERROR, "Memory accounting init"
+                        "failed");
+                return ret;
+        }
+out:
+        return ret;
+}
+
+
+void
+fini (xlator_t *this)
+{
+        gf_ctr_private_t *priv = NULL;
+
+        priv = this->private;
+
+        if (priv) {
+                if (fini_db (priv->_db_conn)) {
+                        gf_log (this->name, GF_LOG_WARNING, "Failed closing "
+                        "db connection");
+                }
+                GF_FREE (priv->ctr_db_path);
+        }
+        GF_FREE (priv);
+        mem_pool_destroy (this->local_pool);
+
+        return;
+}
+
+struct xlator_fops fops = {
+        /*write fops */
+        .mknod       = ctr_mknod,
+        .create      = ctr_create,
+        .truncate    = ctr_truncate,
+        .ftruncate   = ctr_ftruncate,
+        .setxattr    = ctr_setxattr,
+        .removexattr = ctr_removexattr,
+        .unlink      = ctr_unlink,
+        .link        = ctr_link,
+        .rename      = ctr_rename,
+        .writev      = ctr_writev,
+        .setattr      = ctr_setattr,
+        /*read fops*/
+        .readv       = ctr_readv
+};
+
+struct xlator_cbks cbks;
+
+struct volume_options options[] = {
+        { .key  = {"ctr-enabled",},
+          .type = GF_OPTION_TYPE_BOOL,
+          .value = {"on", "off"},
+          .default_value = "off",
+          .description = "Enables the CTR"
+        },
+        { .key  = {"record-entry"},
+          .type = GF_OPTION_TYPE_BOOL,
+          .value = {"on", "off"},
+          .default_value = "on"
+        },
+        { .key  = {"record-exit"},
+          .type = GF_OPTION_TYPE_BOOL,
+          .value = {"on", "off"},
+          .default_value = "off"
+        },
+        { .key  = {"record-counters"},
+          .type = GF_OPTION_TYPE_BOOL,
+          .value = {"on", "off"},
+          .default_value = "off"
+        },
+        { .key  = {"hot-brick"},
+          .type = GF_OPTION_TYPE_BOOL,
+          .value = {"on", "off"},
+          .default_value = "off"
+        },
+        { .key  = {"db-type"},
+          .type = GF_OPTION_TYPE_STR,
+          .value = {"hashfile", "rocksdb", "changelog", "sqlite3",
+                    "hyperdex"},
+          .default_value = "sqlite3"
+        },
+        { .key  = {"db-sync"},
+          .type = GF_OPTION_TYPE_STR,
+          .value = {"sync", "async"},
+          .default_value = "sync"
+        },
+        { .key  = {"db-path"},
+          .type = GF_OPTION_TYPE_PATH
+        },
+        { .key  = {"db-name"},
+          .type = GF_OPTION_TYPE_STR
+        },
+        { .key  = {GFDB_SQL_PARAM_SYNC},
+          .type = GF_OPTION_TYPE_STR,
+          .value = {"off", "normal", "full"},
+          .default_value = "normal"
+        },
+        { .key  = {GFDB_SQL_PARAM_JOURNAL_MODE},
+          .type = GF_OPTION_TYPE_STR,
+          .value = {"delete", "truncate", "persist", "memory", "wal", "off"},
+          .default_value = "wal"
+        },
+        { .key  = {GFDB_SQL_PARAM_AUTO_VACUUM},
+          .type = GF_OPTION_TYPE_STR,
+          .value = {"off", "full", "incr"},
+          .default_value = "off"
+        },
+        { .key  = {GFDB_SQL_PARAM_WAL_AUTOCHECK},
+          .type = GF_OPTION_TYPE_INT,
+          .default_value = "1000"
+        },
+        { .key  = {GFDB_SQL_PARAM_CACHE_SIZE},
+          .type = GF_OPTION_TYPE_INT,
+          .default_value = "1000"
+        },
+        { .key  = {GFDB_SQL_PARAM_PAGE_SIZE},
+          .type = GF_OPTION_TYPE_INT,
+          .default_value = "4096"
+        },
+        { .key  = {NULL} },
+};
diff --git a/xlators/features/changetimerecorder/src/changetimerecorder.h b/xlators/features/changetimerecorder/src/changetimerecorder.h
new file mode 100644
index 00000000000..6a963a56a2f
--- /dev/null
+++ b/xlators/features/changetimerecorder/src/changetimerecorder.h
@@ -0,0 +1,26 @@
+/*
+   Copyright (c) 2006-2015 Red Hat, Inc. <http://www.redhat.com>
+   This file is part of GlusterFS.
+
+   This file is licensed to you under your choice of the GNU Lesser
+   General Public License, version 3 or any later version (LGPLv3 or
+   later), or the GNU General Public License, version 2 (GPLv2), in all
+   cases as published by the Free Software Foundation.
+*/
+
+#ifndef __CTR_H
+#define __CTR_H
+
+#ifndef _CONFIG_H
+#define _CONFIG_H
+#include "config.h"
+#endif
+
+#include "glusterfs.h"
+#include "xlator.h"
+#include "logging.h"
+#include "common-utils.h"
+#include "ctr_mem_types.h"
+#include "ctr-helper.h"
+
+#endif /* __CTR_H */
diff --git a/xlators/features/changetimerecorder/src/ctr-helper.c b/xlators/features/changetimerecorder/src/ctr-helper.c
new file mode 100644
index 00000000000..41eec753fe2
--- /dev/null
+++ b/xlators/features/changetimerecorder/src/ctr-helper.c
@@ -0,0 +1,253 @@
+/*
+   Copyright (c) 2015 Red Hat, Inc. <http://www.redhat.com>
+   This file is part of GlusterFS.
+
+   This file is licensed to you under your choice of the GNU Lesser
+   General Public License, version 3 or any later version (LGPLv3 or
+   later), or the GNU General Public License, version 2 (GPLv2), in all
+   cases as published by the Free Software Foundation.
+*/
+
+#include "ctr-helper.h"
+
+
+/*******************************************************************************
+ *
+ *                      Fill unwind into db record
+ *
+ ******************************************************************************/
+int
+fill_db_record_for_unwind(gf_ctr_local_t        *ctr_local,
+                          gfdb_fop_type_t       fop_type,
+                          gfdb_fop_path_t       fop_path)
+{
+        int ret                         = -1;
+        gfdb_time_t *ctr_uwtime         = NULL;
+
+        GF_ASSERT(ctr_local);
+
+        /*If not unwind path error*/
+        if (!isunwindpath(fop_path)) {
+                gf_log (GFDB_DATA_STORE, GF_LOG_ERROR, "Wrong fop_path."
+                        "Should be unwind");
+                goto out;
+        }
+
+        ctr_uwtime = &CTR_DB_REC(ctr_local).gfdb_unwind_change_time;
+        CTR_DB_REC(ctr_local).gfdb_fop_path = fop_path;
+        CTR_DB_REC(ctr_local).gfdb_fop_type = fop_type;
+
+        /*Time is not recorded for internal fops*/
+        if (!ctr_local->is_internal_fop) {
+                ret = gettimeofday (ctr_uwtime, NULL);
+                if (ret == -1) {
+                        gf_log (GFDB_DATA_STORE, GF_LOG_ERROR,
+                                "Error filling unwind time record %s",
+                                strerror(errno));
+                        goto out;
+                }
+        }
+        ret = 0;
+out:
+        return ret;
+}
+
+
+/*******************************************************************************
+ *
+ *                      Fill wind into db record
+ *
+ ******************************************************************************/
+int
+fill_db_record_for_wind(gf_ctr_local_t          *ctr_local,
+                        gf_ctr_inode_context_t  *ctr_inode_cx)
+{
+        int ret                                 = -1;
+        gfdb_time_t *ctr_wtime                = NULL;
+
+        GF_ASSERT(ctr_local);
+        IS_CTR_INODE_CX_SANE(ctr_inode_cx);
+
+        /*if not wind path error!*/
+        if (!iswindpath(ctr_inode_cx->fop_path)) {
+                gf_log (GFDB_DATA_STORE, GF_LOG_ERROR,
+                        "Wrong fop_path. Should be wind");
+                goto out;
+        }
+
+        ctr_wtime = &CTR_DB_REC(ctr_local).gfdb_wind_change_time;
+        CTR_DB_REC(ctr_local).gfdb_fop_path = ctr_inode_cx->fop_path;
+        CTR_DB_REC(ctr_local).gfdb_fop_type = ctr_inode_cx->fop_type;
+
+        /*Time is not recorded for internal fops*/
+        if (!ctr_local->is_internal_fop) {
+                ret = gettimeofday (ctr_wtime, NULL);
+                if (ret) {
+                        gf_log (GFDB_DATA_STORE, GF_LOG_ERROR,
+                                "Error filling wind time record %s",
+                                strerror(errno));
+                        goto out;
+                }
+        }
+
+        /*Copy gfid into db record*/
+        uuid_copy (CTR_DB_REC(ctr_local).gfid, *(ctr_inode_cx->gfid));
+
+        /*Hard Links*/
+        if (isdentryfop(ctr_inode_cx->fop_type)) {
+                /*new link fop*/
+                if (NEW_LINK_CX(ctr_inode_cx)) {
+                        uuid_copy (CTR_DB_REC(ctr_local).pargfid,
+                                *((NEW_LINK_CX(ctr_inode_cx))->pargfid));
+                        strcpy (CTR_DB_REC(ctr_local).file_name,
+                                NEW_LINK_CX(ctr_inode_cx)->basename);
+                        strcpy (CTR_DB_REC(ctr_local).file_path,
+                                NEW_LINK_CX(ctr_inode_cx)->basepath);
+                }
+                /*rename fop*/
+                if (OLD_LINK_CX(ctr_inode_cx)) {
+                        uuid_copy (CTR_DB_REC(ctr_local).old_pargfid,
+                                *((OLD_LINK_CX(ctr_inode_cx))->pargfid));
+                        strcpy (CTR_DB_REC(ctr_local).old_file_name,
+                                OLD_LINK_CX(ctr_inode_cx)->basename);
+                        strcpy (CTR_DB_REC(ctr_local).old_path,
+                                OLD_LINK_CX(ctr_inode_cx)->basepath);
+                }
+        }
+
+        ret = 0;
+out:
+        /*On error roll back and clean the record*/
+        if (ret == -1) {
+                CLEAR_CTR_DB_RECORD (ctr_local);
+        }
+        return ret;
+}
+
+
+/******************************************************************************
+ *
+ *                      CTR xlator init related functions
+ *
+ *
+ * ****************************************************************************/
+static int
+extract_sql_params(xlator_t *this, dict_t *params_dict)
+{
+        int ret                         = -1;
+        char *db_path                   = NULL;
+        char *db_name                   = NULL;
+        char *db_full_path              = NULL;
+
+        GF_ASSERT (this);
+        GF_ASSERT (params_dict);
+
+        /*Extract the path of the db*/
+        db_path = NULL;
+        GET_DB_PARAM_FROM_DICT_DEFAULT(this->name, this->options, "db-path",
+                                        db_path, "/var/run/gluster/");
+
+        /*Extract the name of the db*/
+        db_name = NULL;
+        GET_DB_PARAM_FROM_DICT_DEFAULT(this->name, this->options, "db-name",
+                                        db_name, "gf_ctr_db.db");
+
+        /*Construct full path of the db*/
+        ret = gf_asprintf(&db_full_path, "%s/%s", db_path, db_name);
+        if (ret < 0) {
+                gf_log (GFDB_DATA_STORE, GF_LOG_ERROR,
+                                "Construction of full db path failed!");
+                goto out;
+        }
+
+        /*Setting the SQL DB Path*/
+        SET_DB_PARAM_TO_DICT(this->name, params_dict, GFDB_SQL_PARAM_DBPATH,
+                                db_full_path, ret, out);
+
+        /*Extact rest of the sql params*/
+        ret = gfdb_set_sql_params(this->name, this->options, params_dict);
+        if (ret) {
+                gf_log (GFDB_DATA_STORE, GF_LOG_ERROR,
+                                "Failed setting values to sql param dict!");
+        }
+
+        ret = 0;
+
+out:
+        if (ret)
+                GF_FREE (db_full_path);
+        return ret;
+}
+
+
+
+int extract_db_params(xlator_t *this, dict_t *params_dict,
+                                                gfdb_db_type_t db_type) {
+
+        int ret = -1;
+
+        GF_ASSERT (this);
+        GF_ASSERT (params_dict);
+
+        switch (db_type) {
+        case GFDB_SQLITE3:
+                ret = extract_sql_params(this, params_dict);
+                if (ret)
+                        goto out;
+                break;
+        case GFDB_ROCKS_DB:
+        case GFDB_HYPERDEX:
+        case GFDB_HASH_FILE_STORE:
+        case GFDB_INVALID_DB:
+        case GFDB_DB_END:
+                ret = -1;
+                break;
+        }
+        ret = 0;
+out:
+        return ret;
+}
+
+int extract_ctr_options (xlator_t *this, gf_ctr_private_t *_priv) {
+        int ret         = -1;
+        char *_val_str  = NULL;
+
+        GF_ASSERT (this);
+        GF_ASSERT (_priv);
+
+        /*Checking if the CTR Translator is enabled. By Default its enabled*/
+        _priv->enabled = _gf_false;
+        GF_OPTION_INIT ("ctr-enabled", _priv->enabled, bool, out);
+        if (!_priv->enabled) {
+                gf_log (GFDB_DATA_STORE, GF_LOG_ERROR,
+                                "CTR Xlator is Disable!");
+                ret = 0;
+                goto out;
+        }
+
+        /*Extract db type*/
+        GF_OPTION_INIT ("db-type", _val_str, str, out);
+        _priv->gfdb_db_type = gf_string2gfdbdbtype(_val_str);
+
+        /*Extract flag for record on wind*/
+        GF_OPTION_INIT ("record-entry", _priv->ctr_record_wind, bool, out);
+
+        /*Extract flag for record on unwind*/
+        GF_OPTION_INIT ("record-exit", _priv->ctr_record_unwind, bool, out);
+
+        /*Extract flag for record on counters*/
+        GF_OPTION_INIT ("record-counters", _priv->ctr_record_counter, bool,
+                        out);
+
+        /*Extract flag for hot tier brick*/
+        GF_OPTION_INIT ("hot-brick", _priv->ctr_hot_brick, bool, out);
+
+        /*Extract flag for sync mode*/
+        GF_OPTION_INIT ("db-sync", _val_str, str, out);
+        _priv->gfdb_sync_type = gf_string2gfdbdbsync(_val_str);
+
+        ret = 0;
+
+out:
+        return ret;
+}
diff --git a/xlators/features/changetimerecorder/src/ctr-helper.h b/xlators/features/changetimerecorder/src/ctr-helper.h
new file mode 100644
index 00000000000..9f381e4791c
--- /dev/null
+++ b/xlators/features/changetimerecorder/src/ctr-helper.h
@@ -0,0 +1,446 @@
+/*
+   Copyright (c) 2015 Red Hat, Inc. <http://www.redhat.com>
+   This file is part of GlusterFS.
+
+   This file is licensed to you under your choice of the GNU Lesser
+   General Public License, version 3 or any later version (LGPLv3 or
+   later), or the GNU General Public License, version 2 (GPLv2), in all
+   cases as published by the Free Software Foundation.
+*/
+
+#ifndef __CTR_HELPER_H
+#define __CTR_HELPER_H
+
+
+#ifndef _CONFIG_H
+#define _CONFIG_H
+#include "config.h"
+#endif
+
+#include "xlator.h"
+#include "ctr_mem_types.h"
+#include "iatt.h"
+#include "glusterfs.h"
+#include "xlator.h"
+#include "logging.h"
+#include "common-utils.h"
+#include <time.h>
+#include <sys/time.h>
+
+#include "gfdb_data_store.h"
+
+/*CTR Xlator Private structure*/
+typedef struct gf_ctr_private {
+        gf_boolean_t                    enabled;
+        char                            *ctr_db_path;
+        gf_boolean_t                    ctr_hot_brick;
+        gf_boolean_t                    ctr_record_wind;
+        gf_boolean_t                    ctr_record_unwind;
+        gf_boolean_t                    ctr_record_counter;
+        gfdb_db_type_t                  gfdb_db_type;
+        gfdb_sync_type_t                gfdb_sync_type;
+        gfdb_conn_node_t                *_db_conn;
+} gf_ctr_private_t;
+
+
+/*
+ * gf_ctr_local_t is the ctr xlator local data structure that is stored in
+ * the call_frame of each FOP.
+ *
+ * gfdb_db_record: The gf_ctr_local contains a gfdb_db_record object, which is
+ * used by the insert_record() api from the libgfdb. The gfdb_db_record object
+ * will contain all the inode and hardlink(only for dentry fops: create,
+ * mknod,link, unlink, rename).The ctr_local is keep alive till the unwind
+ * call and will be release during the unwind. The same gfdb_db_record will
+ * used for the unwind insert_record() api, to record unwind in the database.
+ *
+ * ia_inode_type in gf_ctr_local will tell the type of the inode. This is
+ * important for during the unwind path. As we will not have the inode during
+ * the unwind path. We would have include this in the gfdb_db_record itself
+ * but currently we record only file inode information.
+ *
+ * is_internal_fop in gf_ctr_local will tell us if this is a internal fop and
+ * take special/no action. We dont record change/acces times or increement heat
+ * counter for internal fops from rebalancer.
+ * NOTE: This piece is broken with the addition of frequency counters.
+ * Any rebalancer or tiering will cause the files to get the files heated.
+ * We would require seperate identifiers for tiering FOPS.
+ * The QE have noted this issue and will raise a bug as this patch gets merged.
+ * We will fix this as a bug fix.
+ *
+ * */
+typedef struct gf_ctr_local {
+        gfdb_db_record_t        gfdb_db_record;
+        ia_type_t               ia_inode_type;
+        gf_boolean_t            is_internal_fop;
+} gf_ctr_local_t;
+/*
+ * Easy access of gfdb_db_record of ctr_local
+ * */
+#define CTR_DB_REC(ctr_local)\
+        (ctr_local->gfdb_db_record)
+
+/*Clear db record*/
+#define CLEAR_CTR_DB_RECORD(ctr_local)\
+do {\
+        ctr_local->gfdb_db_record.gfdb_fop_path = GFDB_FOP_INVALID;\
+        memset(&(ctr_local->gfdb_db_record.gfdb_wind_change_time),\
+                0, sizeof(gfdb_time_t));\
+        memset(&(ctr_local->gfdb_db_record.gfdb_unwind_change_time),\
+                0, sizeof(gfdb_time_t));\
+        uuid_clear (ctr_local->gfdb_db_record.gfid);\
+        uuid_clear (ctr_local->gfdb_db_record.pargfid);\
+        memset(ctr_local->gfdb_db_record.file_name, 0, PATH_MAX);\
+        memset(ctr_local->gfdb_db_record.old_file_name, 0, PATH_MAX);\
+        ctr_local->gfdb_db_record.gfdb_fop_type = GFDB_FOP_INVALID_OP;\
+        ctr_local->ia_inode_type = IA_INVAL;\
+} while (0)
+
+
+static gf_ctr_local_t *
+init_ctr_local_t (xlator_t *this) {
+
+        gf_ctr_local_t  *ctr_local     = NULL;
+
+        GF_ASSERT(this);
+
+        ctr_local = mem_get0 (this->local_pool);
+        if (!ctr_local) {
+                gf_log (GFDB_DATA_STORE, GF_LOG_ERROR,
+                        "Error while creating ctr local");
+                goto out;
+        }
+
+        CLEAR_CTR_DB_RECORD (ctr_local);
+out:
+        return ctr_local;
+}
+
+static void
+free_ctr_local (gf_ctr_local_t *ctr_local)
+{
+        if (ctr_local)
+                mem_put (ctr_local);
+}
+
+
+
+/******************************************************************************
+ *
+ *
+ *                      Context Carrier Structures
+ *
+ *
+ * ****************************************************************************/
+
+/*
+ * Context Carrier structures are used to carry relavent information about
+ * inodes and links from the fops calls to the ctr_insert_wind.
+ * These structure just have pointers to the original data and donot
+ * do a deep copy of any data. This info is deep copied to
+ * ctr_local->gfdb_db_record and passed to insert_record() api of libgfdb. This
+ * info remains persistent for the unwind in  ctr_local->gfdb_db_record
+ * and once used will be destroyed.
+ *
+ * gf_ctr_link_context_t : Context structure for hard links
+ * gf_ctr_inode_context_t : Context structure for inodes
+ *
+ * */
+
+ /*Context Carrier Structure for hard links*/
+typedef struct gf_ctr_link_context {
+        uuid_t                  *pargfid;
+        const char              *basename;
+        /*basepath is redundent. Will go off*/
+        const char              *basepath;
+} gf_ctr_link_context_t;
+
+ /*Context Carrier Structure for inodes*/
+typedef struct gf_ctr_inode_context {
+        ia_type_t               ia_type;
+        uuid_t                  *gfid;
+        gf_ctr_link_context_t   *new_link_cx;
+        gf_ctr_link_context_t   *old_link_cx;
+        gfdb_fop_type_t         fop_type;
+        gfdb_fop_path_t         fop_path;
+} gf_ctr_inode_context_t;
+
+
+/*******************Util Macros for Context Carrier Structures*****************/
+
+/*Checks if ctr_link_cx is sane!*/
+#define IS_CTR_LINK_CX_SANE(ctr_link_cx)\
+do {\
+        if (ctr_link_cx) {\
+                if (ctr_link_cx->pargfid)\
+                        GF_ASSERT (*(ctr_link_cx->pargfid));\
+                GF_ASSERT (ctr_link_cx->basename);\
+                GF_ASSERT (ctr_link_cx->basepath);\
+        };\
+} while (0)
+
+/*Clear and fill the ctr_link_context with values*/
+#define FILL_CTR_LINK_CX(ctr_link_cx, _pargfid, _basename, _basepath)\
+do {\
+        GF_ASSERT (ctr_link_cx);\
+        GF_ASSERT (_pargfid);\
+        GF_ASSERT (_basename);\
+        GF_ASSERT (_basepath);\
+        memset (ctr_link_cx, 0, sizeof (*ctr_link_cx));\
+        ctr_link_cx->pargfid = &_pargfid;\
+        ctr_link_cx->basename = _basename;\
+        ctr_link_cx->basepath = _basepath;\
+} while (0)
+
+#define NEW_LINK_CX(ctr_inode_cx)\
+        ctr_inode_cx->new_link_cx\
+
+#define OLD_LINK_CX(ctr_inode_cx)\
+        ctr_inode_cx->old_link_cx\
+
+/*Checks if ctr_inode_cx is sane!*/
+#define IS_CTR_INODE_CX_SANE(ctr_inode_cx)\
+do {\
+        GF_ASSERT (ctr_inode_cx);\
+        GF_ASSERT (ctr_inode_cx->gfid);\
+        GF_ASSERT (*(ctr_inode_cx->gfid));\
+        GF_ASSERT (ctr_inode_cx->fop_type != GFDB_FOP_INVALID_OP);\
+        GF_ASSERT (ctr_inode_cx->fop_path != GFDB_FOP_INVALID);\
+        IS_CTR_LINK_CX_SANE (NEW_LINK_CX(ctr_inode_cx));\
+        IS_CTR_LINK_CX_SANE (OLD_LINK_CX(ctr_inode_cx));\
+} while (0)
+
+/*Clear and fill the ctr_inode_context with values*/
+#define FILL_CTR_INODE_CONTEXT(ctr_inode_cx,\
+                                _ia_type,\
+                                _gfid,\
+                                _new_link_cx,\
+                                _old_link_cx,\
+                                _fop_type,\
+                                _fop_path)\
+do {\
+        GF_ASSERT(ctr_inode_cx);\
+        GF_ASSERT(_gfid);\
+        GF_ASSERT(_fop_type != GFDB_FOP_INVALID_OP);\
+        GF_ASSERT(_fop_path != GFDB_FOP_INVALID);\
+        memset(ctr_inode_cx, 0, sizeof(*ctr_inode_cx));\
+        ctr_inode_cx->ia_type = _ia_type;\
+        ctr_inode_cx->gfid = &_gfid;\
+        IS_CTR_LINK_CX_SANE(NEW_LINK_CX(ctr_inode_cx));\
+        if (_new_link_cx)\
+                NEW_LINK_CX(ctr_inode_cx) = _new_link_cx;\
+        IS_CTR_LINK_CX_SANE(OLD_LINK_CX(ctr_inode_cx));\
+        if (_old_link_cx)\
+                OLD_LINK_CX(ctr_inode_cx) = _old_link_cx;\
+        ctr_inode_cx->fop_type = _fop_type;\
+        ctr_inode_cx->fop_path = _fop_path;\
+} while (0)
+
+/******************************************************************************
+ *
+ *                      Util functions or macros used by
+ *                      insert wind and insert unwind
+ *
+ * ****************************************************************************/
+
+/*
+ * If a rebalancer fop
+ * */
+#define REBALANCE_FOP(frame)\
+        (frame->root->pid == GF_CLIENT_PID_DEFRAG)
+
+/*
+ * if a rebalancer fop goto
+ * */
+#define CTR_IF_REBALANCE_FOP_THEN_GOTO(frame, label)\
+do {\
+        if (REBALANCE_FOP (frame))\
+                goto label;\
+} while (0)
+
+/*
+ * Internal fop
+ *
+ * */
+#define CTR_IS_INTERNAL_FOP(frame, priv)\
+        (REBALANCE_FOP(frame) && (!priv->ctr_hot_brick))
+
+/**
+ * ignore internal fops for all clients except AFR self-heal daemon
+ */
+#define CTR_IF_INTERNAL_FOP_THEN_GOTO(frame, dict, label)\
+do {\
+                if ((frame->root->pid != GF_CLIENT_PID_AFR_SELF_HEALD)  \
+                    && dict                                             \
+                    && dict_get (dict, GLUSTERFS_INTERNAL_FOP_KEY))     \
+                        goto label;                                     \
+} while (0)
+
+
+/*
+ * IS CTR Xlator is disabled then goto to label
+ * */
+ #define CTR_IS_DISABLED_THEN_GOTO(this, label)\
+ do {\
+        gf_ctr_private_t *_priv = NULL;\
+        GF_ASSERT (this);\
+        GF_ASSERT (this->private);\
+        _priv = this->private;\
+        if (!_priv->enabled)\
+                goto label;\
+ } while (0)
+
+
+int
+fill_db_record_for_unwind(gf_ctr_local_t        *ctr_local,
+                          gfdb_fop_type_t       fop_type,
+                          gfdb_fop_path_t       fop_path);
+
+int
+fill_db_record_for_wind(gf_ctr_local_t          *ctr_local,
+                        gf_ctr_inode_context_t  *ctr_inode_cx);
+
+/*******************************************************************************
+ *                              CTR INSERT WIND
+ * *****************************************************************************
+ * Function used to insert/update record into the database during a wind fop
+ * This function creates ctr_local structure into the frame of the fop
+ * call.
+ * ****************************************************************************/
+static inline int
+ctr_insert_wind (call_frame_t                    *frame,
+                xlator_t                        *this,
+                gf_ctr_inode_context_t          *ctr_inode_cx)
+{
+        int ret                         = -1;
+        gf_ctr_private_t *_priv         = NULL;
+        gf_ctr_local_t *ctr_local       = NULL;
+
+        GF_ASSERT(frame);
+        GF_ASSERT(this);
+        IS_CTR_INODE_CX_SANE(ctr_inode_cx);
+
+        _priv = this->private;
+        GF_ASSERT (_priv);
+
+        GF_ASSERT(_priv->_db_conn);
+
+        /*If record_wind option of CTR is on record wind for
+         * regular files only*/
+        if (_priv->ctr_record_wind && ctr_inode_cx->ia_type != IA_IFDIR) {
+                frame->local = init_ctr_local_t (this);
+                if (!frame->local) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "WIND: Error while creating ctr local");
+                        goto out;
+                };
+                ctr_local = frame->local;
+
+                /*Broken please refer gf_ctr_local_t documentation*/
+                ctr_local->is_internal_fop = CTR_IS_INTERNAL_FOP(frame, _priv);
+
+                /*Broken please refer gf_ctr_local_t documentation*/
+                CTR_DB_REC(ctr_local).do_record_counters =
+                                                _priv->ctr_record_counter;
+
+                /*Fill the db record for insertion*/
+                ret = fill_db_record_for_wind (ctr_local, ctr_inode_cx);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "WIND: Error filling  ctr local");
+                        goto out;
+                }
+                /*Insert the db record*/
+                ret = insert_record (_priv->_db_conn,
+                                        &ctr_local->gfdb_db_record);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "WIND: Inserting of record failed!");
+                        goto out;
+                }
+        }
+        ret = 0;
+out:
+
+        if (ret) {
+                free_ctr_local (ctr_local);
+        }
+
+        return ret;
+}
+
+
+
+
+/*******************************************************************************
+ *                             CTR INSERT UNWIND
+ * *****************************************************************************
+ * Function used to insert/update record into the database during a unwind fop
+ * This function destroys ctr_local structure into the frame of the fop
+ * call at the end.
+ * ****************************************************************************/
+static inline int
+ctr_insert_unwind (call_frame_t          *frame,
+                  xlator_t              *this,
+                  gfdb_fop_type_t       fop_type,
+                  gfdb_fop_path_t       fop_path)
+{
+        int ret = -1;
+        gf_ctr_private_t *_priv         = NULL;
+        gf_ctr_local_t *ctr_local       = NULL;
+
+        GF_ASSERT(frame);
+        GF_ASSERT(this);
+
+        _priv = this->private;
+        GF_ASSERT (_priv);
+
+        GF_ASSERT(_priv->_db_conn);
+
+        ctr_local = frame->local;
+
+        if (ctr_local
+            && (_priv->ctr_record_unwind || isdentryfop(fop_type))
+            && (ctr_local->ia_inode_type != IA_IFDIR)) {
+
+                CTR_DB_REC(ctr_local).do_record_uwind_time =
+                                                _priv->ctr_record_unwind;
+
+                ret = fill_db_record_for_unwind(ctr_local, fop_type, fop_path);
+                if (ret == -1) {
+                        gf_log(this->name, GF_LOG_ERROR, "UNWIND: Error"
+                                "filling ctr local");
+                        goto out;
+                }
+
+                ret = insert_record(_priv->_db_conn,
+                                        &ctr_local->gfdb_db_record);
+                if (ret == -1) {
+                        gf_log(this->name, GF_LOG_ERROR, "UNWIND: Error"
+                                "filling ctr local");
+                        goto out;
+                }
+        }
+        ret = 0;
+out:
+        free_ctr_local (ctr_local);
+        frame->local = NULL;
+        return ret;
+}
+
+/******************************************************************************
+ *
+ *                      CTR xlator init related functions
+ *
+ *
+ * ****************************************************************************/
+int
+extract_db_params (xlator_t              *this,
+                  dict_t                *params_dict,
+                  gfdb_db_type_t        db_type);
+
+int
+extract_ctr_options (xlator_t            *this,
+                    gf_ctr_private_t    *_priv);
+
+#endif
diff --git a/xlators/features/changetimerecorder/src/ctr_mem_types.h b/xlators/features/changetimerecorder/src/ctr_mem_types.h
new file mode 100644
index 00000000000..48387597814
--- /dev/null
+++ b/xlators/features/changetimerecorder/src/ctr_mem_types.h
@@ -0,0 +1,22 @@
+/*
+  Copyright (c) 2008-2015 Red Hat, Inc. <http://www.redhat.com>
+  This file is part of GlusterFS.
+
+  This file is licensed to you under your choice of the GNU Lesser
+  General Public License, version 3 or any later version (LGPLv3 or
+  later), or the GNU General Public License, version 2 (GPLv2), in all
+  cases as published by the Free Software Foundation.
+*/
+
+
+#ifndef __CTR_MEM_TYPES_H__
+#define __CTR_MEM_TYPES_H__
+
+#include "gfdb_mem-types.h"
+
+enum gf_ctr_mem_types_ {
+        gf_ctr_mt_private_t = gfdb_mt_end + 1,
+        gf_ctr_mt_end
+};
+#endif
+
diff --git a/xlators/mgmt/glusterd/src/glusterd-volgen.c b/xlators/mgmt/glusterd/src/glusterd-volgen.c
index c140e680261..114e57485fc 100644
--- a/xlators/mgmt/glusterd/src/glusterd-volgen.c
+++ b/xlators/mgmt/glusterd/src/glusterd-volgen.c
@@ -1560,6 +1560,73 @@ out:
 }
 
 static int
+brick_graph_add_changetimerecorder (volgen_graph_t *graph,
+                                    glusterd_volinfo_t *volinfo,
+                                    dict_t *set_dict,
+                                    glusterd_brickinfo_t *brickinfo)
+{
+        xlator_t        *xl = NULL;
+        int             ret = -1;
+        glusterd_brickinfo_t *brickiter     = NULL;
+        glusterd_brickinfo_t *tmp           = NULL;
+        char                 *brickname     = NULL;
+        char                 *path          = NULL;
+        char                 *volname       = NULL;
+        int                   bricknum      = 0;
+        char     index_basepath[PATH_MAX]   = {0};
+
+        if (!graph || !volinfo || !set_dict || !brickinfo)
+                goto out;
+
+        path      = brickinfo->path;
+
+        xl = volgen_graph_add (graph, "features/changetimerecorder",
+                               volinfo->volname);
+
+
+        ret = xlator_set_option (xl, "db-type", "sqlite3");
+        if (ret)
+                goto out;
+
+        bricknum = 0;
+        list_for_each_entry_safe (brickiter, tmp, &volinfo->bricks,
+                                  brick_list) {
+                if (brickiter == brickinfo)
+                        break;
+                bricknum++;
+        }
+        if (bricknum < volinfo->tier_info.hot_brick_count) {
+                ret = xlator_set_option (xl, "hot-brick", "on");
+        } else {
+                ret = xlator_set_option (xl, "hot-brick", "off");
+        }
+
+        brickname = strrchr(path, '/') + 1;
+        snprintf (index_basepath, sizeof (index_basepath), "%s.db",
+                  brickname);
+        ret = xlator_set_option (xl, "db-name", index_basepath);
+        if (ret)
+                goto out;
+
+        snprintf (index_basepath, sizeof (index_basepath), "%s/%s",
+                  path, ".glusterfs/");
+        ret = xlator_set_option (xl, "db-path", index_basepath);
+        if (ret)
+                goto out;
+
+        ret = xlator_set_option (xl, "record-exit", "off");
+        if (ret)
+                goto out;
+
+        ret = xlator_set_option (xl, "record-entry", "on");
+        if (ret)
+                goto out;
+
+out:
+        return ret;
+}
+
+static int
 brick_graph_add_acl (volgen_graph_t *graph, glusterd_volinfo_t *volinfo,
                       dict_t *set_dict, glusterd_brickinfo_t *brickinfo)
 {
@@ -2071,6 +2138,7 @@ static volgen_brick_xlator_t server_graph_table[] = {
         {brick_graph_add_locks, "locks"},
         {brick_graph_add_acl, "acl"},
         {brick_graph_add_changelog, "changelog"},
+        {brick_graph_add_changetimerecorder, "changetimerecorder"},
         {brick_graph_add_bd, "bd"},
         {brick_graph_add_trash, "trash"},
         {brick_graph_add_posix, "posix"},
@@ -3554,8 +3622,7 @@ nfs_option_handler (volgen_graph_t *graph,
                         return -1;
         }
 
-
-      if ((strcmp (vme->voltype, "nfs/server") == 0) &&
+        if ((strcmp (vme->voltype, "nfs/server") == 0) &&
              (vme->option && vme->option[0]!='!') ) {
                ret = xlator_set_option (xl, vme->option, vme->value);
                 if (ret)
diff --git a/xlators/mgmt/glusterd/src/glusterd-volume-set.c b/xlators/mgmt/glusterd/src/glusterd-volume-set.c
index cae7c07ed53..ae866b7ccfc 100644
--- a/xlators/mgmt/glusterd/src/glusterd-volume-set.c
+++ b/xlators/mgmt/glusterd/src/glusterd-volume-set.c
@@ -523,8 +523,6 @@ struct volopt_map_entry glusterd_volopt_map[] = {
           .op_version = GD_OP_VERSION_3_7_0,
           .flags      = OPT_FLAG_CLIENT_OPT
         },
-
-        /* Stripe xlator options */
         { .key         = "cluster.stripe-block-size",
           .voltype     = "cluster/stripe",
           .option      = "block-size",
@@ -1692,6 +1690,18 @@ struct volopt_map_entry glusterd_volopt_map[] = {
           .voltype     = "features/trash",
           .op_version  = GD_OP_VERSION_3_7_0,
         },
+        { .key         = "features.ctr-enabled",
+          .voltype     = "features/changetimerecorder",
+          .value       = "off",
+          .option      = "ctr-enabled",
+          .op_version  = GD_OP_VERSION_3_7_0
+        },
+        { .key         = "features.record-counters",
+          .voltype     = "features/changetimerecorder",
+          .value       = "off",
+          .option      = "record-counters",
+          .op_version  = GD_OP_VERSION_3_7_0
+        },
         { .key         = "locks.trace",
           .voltype     = "features/locks",
           .type        = NO_DOC,
@@ -1727,7 +1737,6 @@ struct volopt_map_entry glusterd_volopt_map[] = {
           .type        = NO_DOC,
           .op_version  = GD_OP_VERSION_3_7_0,
         },
-
         { .key         = NULL
         }
 };
diff --git a/xlators/mgmt/glusterd/src/glusterd.h b/xlators/mgmt/glusterd/src/glusterd.h
index f70a4c85822..ad280eda053 100644
--- a/xlators/mgmt/glusterd/src/glusterd.h
+++ b/xlators/mgmt/glusterd/src/glusterd.h
@@ -293,11 +293,26 @@ typedef enum gd_quorum_status_ {
         DOESNT_MEET_QUORUM, //Follows quorum and does not meet.
 } gd_quorum_status_t;
 
+typedef struct tier_info_ {
+        int                       cold_type;
+        int                       cold_brick_count;
+        int                       cold_replica_count;
+        int                       cold_disperse_count;
+        int                       cold_dist_leaf_count;
+        int                       hot_type;
+        int                       hot_brick_count;
+        int                       hot_replica_count;
+        int                       hot_disperse_count;
+        /*Commented for now Dan's DHT Tier patch will have it*/
+        /*tier_group_t             *root;*/
+} gd_tier_info_t;
+
 struct glusterd_volinfo_ {
         gf_lock_t                 lock;
         gf_boolean_t              is_snap_volume;
         glusterd_snap_t          *snapshot;
         uuid_t                    restored_from_snap;
+        gd_tier_info_t            tier_info;
         char                      parent_volname[GD_VOLUME_NAME_MAX];
                                          /* In case of a snap volume
                                             i.e (is_snap_volume == TRUE) this
@@ -983,6 +998,7 @@ int glusterd_op_stage_clearlocks_volume (dict_t *dict, char **op_errstr);
 int glusterd_op_clearlocks_volume (dict_t *dict, char **op_errstr,
                                    dict_t *rsp_dict);
 
+
 int glusterd_op_stage_barrier (dict_t *dict, char **op_errstr);
 int glusterd_op_barrier (dict_t *dict, char **op_errstr);
 
@@ -999,6 +1015,7 @@ int glusterd_op_gsync_args_get (dict_t *dict, char **op_errstr,
 
 int glusterd_start_volume (glusterd_volinfo_t *volinfo, int flags,
                            gf_boolean_t wait);
+
 int glusterd_stop_volume (glusterd_volinfo_t *volinfo);
 
 /* Synctask part */
diff --git a/xlators/storage/posix/src/posix.c b/xlators/storage/posix/src/posix.c
index fc6ec991c44..8001f238614 100644
--- a/xlators/storage/posix/src/posix.c
+++ b/xlators/storage/posix/src/posix.c
@@ -1485,8 +1485,10 @@ posix_unlink (call_frame_t *frame, xlator_t *this,
         int32_t                unlink_if_linkto   = 0;
         int32_t                check_open_fd      = 0;
         int32_t                skip_unlink        = 0;
+        int32_t                ctr_link_req       = 0;
         ssize_t                xattr_size         = -1;
         int32_t                is_dht_linkto_file = 0;
+        dict_t                 *unwind_dict        = NULL;
 
         DECLARE_OLD_FS_ID_VAR;
 
@@ -1618,13 +1620,53 @@ posix_unlink (call_frame_t *frame, xlator_t *this,
                 goto out;
         }
 
+        /*
+         *
+         * Check if there is a CTR_REQUEST_LINK_COUNT_XDATA from CTR Xlator
+         *
+         * */
+        op_ret = dict_get_int32 (xdata, CTR_REQUEST_LINK_COUNT_XDATA,
+                                 &ctr_link_req);
+        if (op_ret) {
+                /*Since no request no response*/
+                op_ret = 0;
+                goto out;
+        }
+
+        /* Sending back inode link count to ctr_unlink(changetimerecoder xlator)
+         * via "CTR_RESPONSE_LINK_COUNT_XDATA" key using unwind_dict.
+         * CTR Xlator will clear all the records if the link count has become 1
+         * i.e this was the last hard link.
+         * */
+        unwind_dict = dict_new ();
+        /* Even if unwind_dict fails to alloc memory we will not mark the FOP
+         * unsuccessful
+         * because this dict is only used by CTR Xlator to clear
+         * all records if link count == 0*/
+        if (!unwind_dict) {
+                op_ret = 0;
+                 gf_log (this->name, GF_LOG_WARNING,
+                        "Failed to creating unwind_dict");
+                goto out;
+        }
+        /* Even if unwind_dict fails to set CTR_RESPONSE_LINK_COUNT_XDATA we will
+         * not mark the FOP unsuccessful
+         * because this dict is only used by CTR Xlator to clear
+         * all records if link count == 0*/
+        op_ret = dict_set_uint32 (unwind_dict, CTR_RESPONSE_LINK_COUNT_XDATA,
+                                stbuf.ia_nlink);
+        if (op_ret == -1) {
+                gf_log (this->name, GF_LOG_WARNING,
+                        "Failed to set CTR_RESPONSE_LINK_COUNT_XDATA");
+        }
+
         op_ret = 0;
 
 out:
         SET_TO_OLD_FS_ID ();
 
         STACK_UNWIND_STRICT (unlink, frame, op_ret, op_errno,
-                             &preparent, &postparent, NULL);
+                             &preparent, &postparent, unwind_dict);
 
         if (fd != -1) {
                 close (fd);