ctr/xlator: Named lookup heal of pre-existing files, before ctr was ON.

Problem: The CTR xlator records file meta (heat/hardlinks)
into the data. This works fine for files which are created
after ctr xlator is switched ON. But for files which were
created before CTR xlator is ON, CTR xlator is not able to
record either of the meta i.e heat or hardlinks. Thus making
those files immune to promotions/demotions.

Solution: The solution that is implemented in this patch is
do ctr-db heal of all those pre-existent files, using named lookup.
For this purpose we use the inode-xlator context variable option
in gluster.
The inode-xlator context variable for ctr xlator will have the
following,
    a. A Lock for the context variable
    b. A hardlink list: This list represents the successful looked
       up hardlinks.
These are the scenarios when the hardlink list is updated:
1) Named-Lookup: Whenever a named lookup happens on a file, in the
   wind path we copy all required hardlink and inode information to
   ctr_db_record structure, which resides in the frame->local variable.
   We dont update the database in wind. During the unwind, we read the
   information from the ctr_db_record and ,
   Check if the inode context variable is created, if not we create it.
   Check if the hard link is there in the hardlink list.
      If its not there we add it to the list and send a update to the
      database using libgfdb.
      Please note: The database transaction can fail(and we ignore) as there
      already might be a record in the db. This update to the db is to heal
      if its not there.
      If its there in the list we ignore it.
2) Inode Forget: Whenever an inode forget hits we clear the hardlink list in
   the inode context variable and delete the inode context variable.
   Please note: An inode forget may happen for two reason,
   a. when the inode is delete.
   b. the in-memory inode is evicted from the inode table due to cache limits.
3) create: whenever a create happens we create the inode context variable and
   add the hardlink. The database updation is done as usual by ctr.
4) link: whenever a hardlink is created for the inode, we create the inode context
   variable, if not present, and add the hardlink to the list.
5) unlink: whenever a unlink happens we delete the hardlink from the list.
6) mknod: same as create.
7) rename: whenever a rename happens we update the hardlink in list. if the hardlink
   was not present for updation, we add the hardlink to the list.

What is pending:
1) This solution will only work for named lookups.
2) We dont track afr-self-heal/dht-rebalancer traffic for healing.

Change-Id: Ia4bbaf84128ad6ce8c3ddd70bcfa82894c79585f
BUG: 1212037
Signed-off-by: Joseph Fernandes <josferna@redhat.com>
Signed-off-by: Dan Lambright <dlambrig@redhat.com>
Reviewed-on: http://review.gluster.org/10370
Reviewed-by: Jeff Darcy <jdarcy@redhat.com>
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Tested-by: NetBSD Build System
Reviewed-by: Vijay Bellur <vbellur@redhat.com>

8 files changed, 1021 insertions, 6 deletions

diff --git a/libglusterfs/src/gfdb/gfdb_sqlite3_helper.c b/libglusterfs/src/gfdb/gfdb_sqlite3_helper.c
index 4b70b49419d..03ebb1f1c8a 100644
--- a/libglusterfs/src/gfdb/gfdb_sqlite3_helper.c
+++ b/libglusterfs/src/gfdb/gfdb_sqlite3_helper.c
@@ -373,7 +373,8 @@ gf_sql_insert_link (gf_sql_connection_t  *sql_conn,
         /*Execute the prepare statement*/
         if (sqlite3_step (insert_stmt) != SQLITE_DONE) {
                 gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
-                        "Failed executing the prepared stmt %s : %s",
+                        "Failed executing the prepared stmt %s %s %s %s %s : %s",
+                        gfid, pargfid, basename, basepath,
                         insert_str,
                         sqlite3_errmsg (sql_conn->sqlite3_db_conn));
                 ret = -1;
@@ -568,7 +569,8 @@ gf_sql_insert_write_wind_time (gf_sql_connection_t  *sql_conn,
         /*Execute the prepare statement*/
         if (sqlite3_step (insert_stmt) != SQLITE_DONE) {
                 gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
-                        "Failed executing the prepared stmt %s : %s",
+                        "Failed executing the prepared stmt GFID:%s %s : %s",
+                        gfid,
                         insert_str,
                         sqlite3_errmsg (sql_conn->sqlite3_db_conn));
                 ret = -1;
diff --git a/tests/basic/tier/tier_lookup_heal.t b/tests/basic/tier/tier_lookup_heal.t
new file mode 100755
index 00000000000..2b778f749f9
--- /dev/null
+++ b/tests/basic/tier/tier_lookup_heal.t
@@ -0,0 +1,72 @@
+#!/bin/bash
+
+. $(dirname $0)/../../include.rc
+. $(dirname $0)/../../volume.rc
+
+LAST_BRICK=1
+CACHE_BRICK_FIRST=2
+CACHE_BRICK_LAST=3
+PROMOTE_TIMEOUT=5
+
+function file_on_fast_tier {
+    local ret="1"
+
+    s1=$(md5sum $1)
+    s2=$(md5sum $B0/${V0}${CACHE_BRICK_FIRST}/$1)
+
+    if [ -e $B0/${V0}${CACHE_BRICK_FIRST}/$1 ] && ! [ "$s1" == "$s2" ]; then
+        echo "0"
+    else
+        echo "1"
+    fi
+}
+
+cleanup
+
+
+TEST glusterd
+TEST pidof glusterd
+
+TEST $CLI volume create $V0 replica 2 $H0:$B0/${V0}{0..$LAST_BRICK}
+TEST $CLI volume start $V0
+TEST $GFS --volfile-id=/$V0 --volfile-server=$H0 $M0;
+
+# Create files before CTR xlator is on.
+cd $M0
+TEST stat .
+TEST touch file1
+TEST stat file1
+
+# gf_file_tb and gf_flink_tb should be empty
+ENTRY_COUNT=$(echo "select * from gf_file_tb; select * from gf_flink_tb;" | \
+        sqlite3 $B0/${V0}$LAST_BRICK/.glusterfs/${V0}$LAST_BRICK.db | wc -l )
+TEST [ $ENTRY_COUNT -eq 0 ]
+
+
+#Attach tier and switch ON CTR Xlator.
+TEST $CLI volume attach-tier $V0 replica 2 $H0:$B0/${V0}$CACHE_BRICK_FIRST $H0:$B0/${V0}$CACHE_BRICK_LAST
+TEST $CLI volume set $V0 features.ctr-enabled on
+TEST $CLI volume set $V0 cluster.tier-demote-frequency 4
+TEST $CLI volume set $V0 cluster.tier-promote-frequency 4
+TEST $CLI volume set $V0 cluster.read-freq-threshold 0
+TEST $CLI volume set $V0 cluster.write-freq-threshold 0
+TEST $CLI volume set $V0 performance.quick-read off
+TEST $CLI volume set $V0 performance.io-cache off
+
+#The lookup should heal the database.
+TEST ls file1
+
+# gf_file_tb and gf_flink_tb should NOT be empty
+ENTRY_COUNT=$(echo "select * from gf_file_tb; select * from gf_flink_tb;" | \
+        sqlite3 $B0/${V0}$LAST_BRICK/.glusterfs/${V0}$LAST_BRICK.db | wc -l )
+TEST [ $ENTRY_COUNT -eq 2 ]
+
+# Heat-up the file
+uuidgen > file1
+TEST $CLI volume rebalance $V0 tier start
+sleep 5
+
+#Check if the file is promoted
+EXPECT_WITHIN $PROMOTE_TIMEOUT "0" file_on_fast_tier file1
+
+cleanup
diff --git a/xlators/features/changetimerecorder/src/Makefile.am b/xlators/features/changetimerecorder/src/Makefile.am
index e79be53c063..630dd8e6e15 100644
--- a/xlators/features/changetimerecorder/src/Makefile.am
+++ b/xlators/features/changetimerecorder/src/Makefile.am
@@ -7,12 +7,12 @@ endif
 
 changetimerecorder_la_LDFLAGS = -module -avoid-version
 
-changetimerecorder_la_SOURCES = changetimerecorder.c ctr-helper.c
+changetimerecorder_la_SOURCES = changetimerecorder.c ctr-helper.c ctr-xlator-ctx.c
 
 changetimerecorder_la_LIBADD = $(top_builddir)/libglusterfs/src/libglusterfs.la\
 			$(top_builddir)/libglusterfs/src/gfdb/libgfdb.la
 
-noinst_HEADERS = changetimerecorder.h ctr_mem_types.h ctr-helper.h
+noinst_HEADERS = changetimerecorder.h ctr_mem_types.h ctr-helper.h ctr-xlator-ctx.h
 
 AM_CPPFLAGS = $(GF_CPPFLAGS) -I$(top_srcdir)/libglusterfs/src \
 		-I$(top_srcdir)/libglusterfs/src/gfdb \
diff --git a/xlators/features/changetimerecorder/src/changetimerecorder.c b/xlators/features/changetimerecorder/src/changetimerecorder.c
index fd57a445e08..dc23132b7f3 100644
--- a/xlators/features/changetimerecorder/src/changetimerecorder.c
+++ b/xlators/features/changetimerecorder/src/changetimerecorder.c
@@ -13,6 +13,308 @@
 #include "gfdb_sqlite3.h"
 #include "ctr-helper.h"
 
+/*******************************inode forget***********************************/
+
+int
+ctr_forget (xlator_t *this, inode_t *inode)
+{
+        fini_ctr_xlator_ctx (this, inode);
+        return 0;
+}
+
+/************************** Look up heal **************************************/
+/*
+Problem: The CTR xlator records file meta (heat/hardlinks)
+into the data. This works fine for files which are created
+after ctr xlator is switched ON. But for files which were
+created before CTR xlator is ON, CTR xlator is not able to
+record either of the meta i.e heat or hardlinks. Thus making
+those files immune to promotions/demotions.
+
+Solution: The solution that is implemented in this patch is
+do ctr-db heal of all those pre-existent files, using named lookup.
+For this purpose we use the inode-xlator context variable option
+in gluster.
+The inode-xlator context variable for ctr xlator will have the
+following,
+    a. A Lock for the context variable
+    b. A hardlink list: This list represents the successful looked
+       up hardlinks.
+These are the scenarios when the hardlink list is updated:
+1) Named-Lookup: Whenever a named lookup happens on a file, in the
+   wind path we copy all required hardlink and inode information to
+   ctr_db_record structure, which resides in the frame->local variable.
+   We dont update the database in wind. During the unwind, we read the
+   information from the ctr_db_record and ,
+   Check if the inode context variable is created, if not we create it.
+   Check if the hard link is there in the hardlink list.
+      If its not there we add it to the list and send a update to the
+      database using libgfdb.
+      Please note: The database transaction can fail(and we ignore) as there
+      already might be a record in the db. This update to the db is to heal
+      if its not there.
+      If its there in the list we ignore it.
+2) Inode Forget: Whenever an inode forget hits we clear the hardlink list in
+   the inode context variable and delete the inode context variable.
+   Please note: An inode forget may happen for two reason,
+   a. when the inode is delete.
+   b. the in-memory inode is evicted from the inode table due to cache limits.
+3) create: whenever a create happens we create the inode context variable and
+   add the hardlink. The database updation is done as usual by ctr.
+4) link: whenever a hardlink is created for the inode, we create the inode
+ context variable, if not present, and add the hardlink to the list.
+5) unlink: whenever a unlink happens we delete the hardlink from the list.
+6) mknod: same as create.
+7) rename: whenever a rename happens we update the hardlink in list. if the
+   hardlink was not present for updation, we add the hardlink to the list.
+
+What is pending:
+1) This solution will only work for named lookups.
+2) We dont track afr-self-heal/dht-rebalancer traffic for healing.
+
+*/
+
+
+/* This function doesnot write anything to the db,
+ * just created the local variable
+ * for the frame and sets values for the ctr_db_record */
+static inline int
+ctr_lookup_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(frame->root);
+        GF_ASSERT(this);
+        IS_CTR_INODE_CX_SANE(ctr_inode_cx);
+
+        _priv = this->private;
+        GF_ASSERT (_priv);
+
+        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;
+                ctr_local->client_pid = frame->root->pid;
+                /*Definately no internal fops will reach here*/
+                ctr_local->is_internal_fop = _gf_false;
+                /*Dont record counters*/
+                CTR_DB_REC(ctr_local).do_record_counters = _gf_false;
+                /*Don't record time at all*/
+                CTR_DB_REC(ctr_local).do_record_times = _gf_false;
+
+                /*Copy gfid into db record*/
+                gf_uuid_copy (CTR_DB_REC(ctr_local).gfid,
+                                *(ctr_inode_cx->gfid));
+
+                /* Set fop_path and fop_type, required by libgfdb to make
+                * decision while inserting the record */
+                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;
+
+                /* Copy hard link info*/
+                gf_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);
+
+        }
+
+        ret = 0;
+
+out:
+
+        if (ret) {
+                free_ctr_local (ctr_local);
+                frame->local = NULL;
+        }
+
+        return ret;
+}
+
+
+/* This function inserts the ctr_db_record populated by ctr_lookup_wind
+ * in to the db. It also destroys the frame->local created by ctr_lookup_wind */
+static inline int
+ctr_lookup_unwind (call_frame_t          *frame,
+                   xlator_t              *this)
+{
+        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 && (ctr_local->ia_inode_type != IA_IFDIR)) {
+
+                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;
+}
+
+/******************************************************************************
+ *
+ *                        FOPS HANDLING BELOW
+ *
+ * ***************************************************************************/
+
+/****************************LOOKUP********************************************/
+
+
+int32_t
+ctr_lookup_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                   int32_t op_ret, int32_t op_errno, inode_t *inode,
+                   struct iatt *buf, dict_t *dict, struct iatt *postparent)
+{
+        int ret                         = -1;
+        ctr_xlator_ctx_t *ctr_xlator_ctx;
+        gf_ctr_local_t *ctr_local       = NULL;
+
+        CTR_IS_DISABLED_THEN_GOTO(this, out);
+        CTR_IF_INTERNAL_FOP_THEN_GOTO (frame, dict, out);
+
+        /* if the lookup failed lookup dont do anything*/
+        if (op_ret == -1) {
+                gf_log (this->name, GF_LOG_TRACE, "lookup failed with %s",
+                        strerror (op_errno));
+                goto out;
+        }
+
+        /* Ignore directory lookups */
+        if (inode->ia_type == IA_IFDIR) {
+                goto out;
+        }
+
+        /* if frame local was not set by the ctr_lookup()
+         * so dont so anything*/
+        if (!frame->local) {
+                goto out;
+        }
+
+        ctr_local = frame->local;
+        /*Assign the proper inode type*/
+        ctr_local->ia_inode_type = inode->ia_type;
+
+        /* if its a first entry
+         * then mark the ctr_record for create
+         * A create will attempt a file and a hard link created in the db*/
+        ctr_xlator_ctx = get_ctr_xlator_ctx (this, inode);
+        if (!ctr_xlator_ctx) {
+                CTR_DB_REC(ctr_local).gfdb_fop_type = GFDB_FOP_CREATE_WRITE;
+        }
+
+        /* Copy the correct gfid from resolved inode */
+        gf_uuid_copy (CTR_DB_REC(ctr_local).gfid, inode->gfid);
+
+        /* Add hard link to the list */
+        ret = add_hard_link_ctx (frame, this, inode);
+        if (ret < 0) {
+                gf_log (this->name, GF_LOG_TRACE, "Failed adding hard link");
+                goto out;
+        }
+
+        /* Inserts the ctr_db_record populated by ctr_lookup_wind
+        * in to the db. It also destroys the frame->local
+        * created by ctr_lookup_wind */
+        ret = ctr_lookup_unwind(frame, this);
+        if (ret) {
+                gf_log (this->name, GF_LOG_TRACE,
+                                "Failed inserting link wind");
+        }
+
+
+out:
+        free_ctr_local ((gf_ctr_local_t *)frame->local);
+        frame->local = NULL;
+
+        STACK_UNWIND_STRICT (lookup, frame, op_ret, op_errno, inode, buf,
+                             dict, postparent);
+
+        return 0;
+}
+
+
+
+int32_t
+ctr_lookup (call_frame_t *frame, xlator_t *this,
+              loc_t *loc, dict_t *xdata)
+{
+        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;
+        int ret                                 = -1;
+
+        CTR_IS_DISABLED_THEN_GOTO(this, out);
+        CTR_IF_INTERNAL_FOP_THEN_GOTO (frame, xdata, out);
+
+        GF_ASSERT(frame);
+        GF_ASSERT(frame->root);
+
+        /* Dont handle nameless lookups*/
+        if (!loc->parent)
+                goto out;
+
+        /*fill ctr link context*/
+        FILL_CTR_LINK_CX(_link_cx, loc->pargfid, loc->name,
+                        loc->path);
+
+         /* Fill ctr inode context*/
+         /* IA_IFREG : We assume its a file in the wind
+          * but in the unwind we are sure what the inode is a file
+          * or directory
+          * gfid: we are just filling loc->gfid which is not correct.
+          * In unwind we fill the correct gfid for successful lookup*/
+        FILL_CTR_INODE_CONTEXT(_inode_cx, IA_IFREG,
+                loc->gfid, _link_cx, NULL,
+                GFDB_FOP_DENTRY_WRITE, GFDB_FOP_WIND);
+
+        /* Create the frame->local and populate ctr_db_record
+         * No writing to the db yet */
+        ret = ctr_lookup_wind(frame, this, _inode_cx);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                                "Failed inserting link wind");
+        }
+
+out:
+        STACK_WIND (frame, ctr_lookup_cbk, FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->lookup, loc, xdata);
+        return 0;
+}
+
+
+
+
 /****************************WRITEV********************************************/
 int32_t
 ctr_writev_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
@@ -415,6 +717,15 @@ ctr_rename (call_frame_t *frame, xlator_t *this, loc_t *oldloc,
         if (ret) {
                 gf_log (this->name, GF_LOG_ERROR,
                                 "Failed inserting rename wind");
+        } else {
+                /* We are doing updation of hard link in inode context in wind
+                 * As we dont get the "inode" in the call back for rename */
+                ret = update_hard_link_ctx (frame, this, oldloc->inode);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "Failed updating hard link in"
+                                "ctr inode context");
+                }
         }
 
 out:
@@ -509,6 +820,15 @@ ctr_unlink (call_frame_t *frame, xlator_t *this,
         if (ret) {
                 gf_log (this->name, GF_LOG_ERROR,
                                 "Failed inserting unlink wind");
+        } else {
+                /* We are doing delete of hard link in inode context in wind
+                 * As we dont get the "inode" in the call back for rename */
+                ret = delete_hard_link_ctx (frame, this, loc->inode);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "Failed deleting hard link from ctr "
+                                "inode context");
+                }
         }
 
         /*
@@ -672,6 +992,12 @@ ctr_mknod_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 
         CTR_IS_DISABLED_THEN_GOTO(this, out);
 
+        /* Add hard link to the list */
+        ret = add_hard_link_ctx (frame, this, inode);
+        if (ret) {
+                gf_log (this->name, GF_LOG_TRACE, "Failed adding hard link");
+        }
+
         ret = ctr_insert_unwind(frame, this, GFDB_FOP_CREATE_WRITE,
                                 GFDB_FOP_UNWIND);
         if (ret) {
@@ -751,6 +1077,12 @@ ctr_create_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 
         CTR_IS_DISABLED_THEN_GOTO(this, out);
 
+
+        ret = add_hard_link_ctx (frame, this, inode);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed adding hard link");
+        }
+
         ret = ctr_insert_unwind(frame, this, GFDB_FOP_CREATE_WRITE,
                                 GFDB_FOP_UNWIND);
         if (ret) {
@@ -830,6 +1162,12 @@ ctr_link_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 
         CTR_IS_DISABLED_THEN_GOTO(this, out);
 
+        /* Add hard link to the list */
+        ret = add_hard_link_ctx (frame, this, inode);
+        if (ret) {
+                gf_log (this->name, GF_LOG_TRACE, "Failed adding hard link");
+        }
+
         ret = ctr_insert_unwind(frame, this, GFDB_FOP_DENTRY_WRITE,
                                 GFDB_FOP_UNWIND);
         if (ret) {
@@ -1117,6 +1455,8 @@ fini (xlator_t *this)
 }
 
 struct xlator_fops fops = {
+        /*lookup*/
+        .lookup      = ctr_lookup,
         /*write fops */
         .mknod       = ctr_mknod,
         .create      = ctr_create,
@@ -1133,7 +1473,9 @@ struct xlator_fops fops = {
         .readv       = ctr_readv
 };
 
-struct xlator_cbks cbks;
+struct xlator_cbks cbks = {
+        .forget = ctr_forget
+};
 
 struct volume_options options[] = {
         { .key  = {"ctr-enabled",},
diff --git a/xlators/features/changetimerecorder/src/ctr-helper.h b/xlators/features/changetimerecorder/src/ctr-helper.h
index 85ee872fa33..87b80e60354 100644
--- a/xlators/features/changetimerecorder/src/ctr-helper.h
+++ b/xlators/features/changetimerecorder/src/ctr-helper.h
@@ -28,6 +28,7 @@
 #include <sys/time.h>
 
 #include "gfdb_data_store.h"
+#include "ctr-xlator-ctx.h"
 
 /*CTR Xlator Private structure*/
 typedef struct gf_ctr_private {
@@ -384,7 +385,7 @@ ctr_insert_wind (call_frame_t                    *frame,
 
                 /*Insert the db record*/
                 ret = insert_record (_priv->_db_conn,
-                                        &ctr_local->gfdb_db_record);
+                                &ctr_local->gfdb_db_record);
                 if (ret) {
                         gf_log (this->name, GF_LOG_ERROR,
                                 "WIND: Inserting of record failed!");
@@ -462,6 +463,146 @@ out:
         return ret;
 }
 
+/******************************* Hard link function ***************************/
+
+static inline int
+add_hard_link_ctx (call_frame_t *frame,
+                   xlator_t     *this,
+                   inode_t      *inode)
+{
+        int ret = -1;
+        gf_ctr_local_t   *ctr_local       = NULL;
+        ctr_xlator_ctx_t *ctr_xlator_ctx  = NULL;
+        ctr_hard_link_t  *ctr_hard_link   = NULL;
+
+        GF_ASSERT (frame);
+        GF_ASSERT (this);
+        GF_ASSERT (inode);
+
+        ctr_local = frame->local;
+        if (!ctr_local) {
+                goto out;
+        }
+
+        ctr_xlator_ctx  = init_ctr_xlator_ctx (this, inode);
+        if (!ctr_xlator_ctx) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Failed accessing ctr inode context");
+                goto out;
+        }
+
+        LOCK (&ctr_xlator_ctx->lock);
+
+        /* Check if the hard link already exists
+         * in the ctr inode context*/
+        ctr_hard_link = ctr_search_hard_link_ctx (this,
+                                ctr_xlator_ctx,
+                                CTR_DB_REC(ctr_local).pargfid,
+                                CTR_DB_REC(ctr_local).file_name);
+        /* if there then ignore */
+        if (ctr_hard_link) {
+                ret = 1;
+                goto unlock;
+        }
+
+        /* Add the hard link to the list*/
+        ret = ctr_add_hard_link (this, ctr_xlator_ctx,
+                        CTR_DB_REC(ctr_local).pargfid,
+                        CTR_DB_REC(ctr_local).file_name);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Failed to add hardlink to the ctr inode context");
+                goto unlock;
+        }
+
+        ret = 0;
+unlock:
+        UNLOCK (&ctr_xlator_ctx->lock);
+out:
+        return ret;
+}
+
+static inline int
+delete_hard_link_ctx (call_frame_t *frame,
+                      xlator_t     *this,
+                      inode_t      *inode)
+{
+        int ret = -1;
+        ctr_xlator_ctx_t *ctr_xlator_ctx  = NULL;
+        gf_ctr_local_t   *ctr_local       = NULL;
+
+        GF_ASSERT (frame);
+        GF_ASSERT (this);
+        GF_ASSERT (inode);
+
+        ctr_local = frame->local;
+        if (!ctr_local) {
+                goto out;
+        }
+
+        ctr_xlator_ctx = get_ctr_xlator_ctx (this, inode);
+        if (!ctr_xlator_ctx) {
+                /* Since there is no ctr inode context so nothing more to do */
+                ret = 0;
+                goto out;
+        }
+
+        ret = ctr_delete_hard_link (this, ctr_xlator_ctx,
+                        CTR_DB_REC(ctr_local).pargfid,
+                        CTR_DB_REC(ctr_local).file_name);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed to delete hard link");
+                goto out;
+        }
+
+        ret = 0;
+
+out:
+        return ret;
+}
+
+static inline int
+update_hard_link_ctx (call_frame_t *frame,
+                      xlator_t     *this,
+                      inode_t      *inode)
+{
+        int ret = -1;
+        ctr_xlator_ctx_t *ctr_xlator_ctx  = NULL;
+        gf_ctr_local_t   *ctr_local       = NULL;
+
+        GF_ASSERT (frame);
+        GF_ASSERT (this);
+        GF_ASSERT (inode);
+
+        ctr_local = frame->local;
+        if (!ctr_local) {
+                goto out;
+        }
+
+        ctr_xlator_ctx  = init_ctr_xlator_ctx (this, inode);
+        if (!ctr_xlator_ctx) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Failed accessing ctr inode context");
+                goto out;
+        }
+
+        ret = ctr_update_hard_link (this, ctr_xlator_ctx,
+                        CTR_DB_REC(ctr_local).pargfid,
+                        CTR_DB_REC(ctr_local).file_name,
+                        CTR_DB_REC(ctr_local).old_pargfid,
+                        CTR_DB_REC(ctr_local).old_file_name);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed to delete hard link");
+                goto out;
+        }
+
+        ret = 0;
+
+out:
+        return ret;
+}
+
+
 /******************************************************************************
  *
  *                      CTR xlator init related functions
diff --git a/xlators/features/changetimerecorder/src/ctr-xlator-ctx.c b/xlators/features/changetimerecorder/src/ctr-xlator-ctx.c
new file mode 100644
index 00000000000..927c8542c4e
--- /dev/null
+++ b/xlators/features/changetimerecorder/src/ctr-xlator-ctx.c
@@ -0,0 +1,370 @@
+/*
+   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-xlator-ctx.h"
+
+#define IS_THE_ONLY_HARDLINK(ctr_hard_link)\
+        (ctr_hard_link->list.next == ctr_hard_link->list.prev)
+
+
+static inline void
+fini_ctr_hard_link (ctr_hard_link_t **ctr_hard_link) {
+
+        GF_ASSERT (ctr_hard_link);
+
+        if (*ctr_hard_link)
+                return;
+        GF_FREE ((*ctr_hard_link)->base_name);
+        GF_FREE (*ctr_hard_link);
+        *ctr_hard_link = NULL;
+}
+
+
+/* Please lock the ctr_xlator_ctx before using this function */
+ctr_hard_link_t *
+ctr_search_hard_link_ctx (xlator_t                  *this,
+                            ctr_xlator_ctx_t        *ctr_xlator_ctx,
+                            uuid_t                  pgfid,
+                            const char              *base_name)
+{
+        ctr_hard_link_t *_hard_link             = NULL;
+        ctr_hard_link_t *searched_hardlink      = NULL;
+
+        GF_ASSERT (this);
+        GF_ASSERT (ctr_xlator_ctx);
+
+        if (pgfid == NULL || base_name == NULL)
+                goto out;
+
+        /*linear search*/
+        list_for_each_entry (_hard_link, &ctr_xlator_ctx->hardlink_list, list) {
+                if (gf_uuid_compare (_hard_link->pgfid, pgfid) == 0
+                        && _hard_link->base_name
+                        && strcmp(_hard_link->base_name, base_name) == 0) {
+                        searched_hardlink = _hard_link;
+                        break;
+                }
+        }
+
+out:
+        return searched_hardlink;
+}
+
+
+
+
+/* Please lock the ctr_xlator_ctx before using this function */
+int
+ctr_add_hard_link (xlator_t           *this,
+               ctr_xlator_ctx_t         *ctr_xlator_ctx,
+               uuid_t                   pgfid,
+               const char               *base_name)
+{
+        int ret                                 = -1;
+        ctr_hard_link_t *ctr_hard_link          = NULL;
+
+        GF_ASSERT (this);
+        GF_ASSERT (ctr_xlator_ctx);
+
+        if (pgfid == NULL || base_name == NULL)
+                goto out;
+
+        ctr_hard_link = GF_CALLOC (1, sizeof (*ctr_hard_link),
+                                        gf_ctr_mt_hard_link_t);
+        if (!ctr_hard_link) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed allocating "
+                                                "ctr_hard_link");
+                goto out;
+        }
+
+        /*Initialize the ctr_hard_link object and
+         * Assign the values : parent GFID and basename*/
+        INIT_LIST_HEAD (&ctr_hard_link->list);
+        gf_uuid_copy (ctr_hard_link->pgfid, pgfid);
+        ret = gf_asprintf(&ctr_hard_link->base_name, "%s", base_name);
+        if (ret < 0) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed copying basename"
+                                        "to ctr_hard_link");
+                goto error;
+        }
+
+        /*Add the hard link to the list*/
+        list_add_tail (&ctr_hard_link->list,
+                        &ctr_xlator_ctx->hardlink_list);
+
+        /*aal izz well!*/
+        ret = 0;
+        goto out;
+error:
+        GF_FREE (ctr_hard_link);
+out:
+        return ret;
+}
+
+static void
+__delete_hard_link_from_list (ctr_hard_link_t  **ctr_hard_link)
+{
+        GF_ASSERT (ctr_hard_link);
+        GF_ASSERT (*ctr_hard_link);
+
+        /*Remove hard link from list*/
+        list_del(&(*ctr_hard_link)->list);
+        fini_ctr_hard_link (ctr_hard_link);
+}
+
+
+int
+ctr_delete_hard_link (xlator_t                *this,
+                  ctr_xlator_ctx_t      *ctr_xlator_ctx,
+                  uuid_t                pgfid,
+                  const char            *base_name)
+{
+        int ret                         = -1;
+        ctr_hard_link_t *ctr_hard_link     = NULL;
+
+        GF_ASSERT (this);
+        GF_ASSERT (ctr_xlator_ctx);
+
+
+        LOCK (&ctr_xlator_ctx->lock);
+
+        /*Check if the hard link is present */
+        ctr_hard_link = ctr_search_hard_link_ctx (this, ctr_xlator_ctx,
+                                                        pgfid, base_name);
+        if (!ctr_hard_link) {
+                gf_log (this->name, GF_LOG_ERROR, "Hard link doesnt exist"
+                                              " in the list");
+                goto out;
+        }
+
+        __delete_hard_link_from_list (&ctr_hard_link);
+        ctr_hard_link = NULL;
+
+        ret = 0;
+out:
+        UNLOCK (&ctr_xlator_ctx->lock);
+
+        return ret;
+}
+
+
+
+
+int
+ctr_update_hard_link (xlator_t                *this,
+                  ctr_xlator_ctx_t      *ctr_xlator_ctx,
+                  uuid_t                pgfid,
+                  const char            *base_name,
+                  uuid_t                old_pgfid,
+                  const char            *old_base_name)
+{
+        int ret                         = -1;
+        ctr_hard_link_t *ctr_hard_link     = NULL;
+
+        GF_ASSERT (this);
+        GF_ASSERT (ctr_xlator_ctx);
+
+
+        LOCK (&ctr_xlator_ctx->lock);
+
+        /*Check if the hard link is present */
+        ctr_hard_link = ctr_search_hard_link_ctx (this, ctr_xlator_ctx,
+                                                old_pgfid, old_base_name);
+        if (!ctr_hard_link) {
+                gf_log (this->name, GF_LOG_TRACE, "Hard link doesnt exist"
+                                              " in the list");
+                /* Since the hard link is not present in the list
+                 * we add it to the list */
+                ret = ctr_add_hard_link (this, ctr_xlator_ctx,
+                                        pgfid, base_name);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR, "Failed adding"
+                                        "hard link to the list");
+                        goto out;
+                }
+                ret = 0;
+                goto out;
+        }
+
+        /* update the hard link */
+        gf_uuid_copy (ctr_hard_link->pgfid, pgfid);
+        GF_FREE (&ctr_hard_link->base_name);
+        ret = gf_asprintf(&ctr_hard_link->base_name, "%s", base_name);
+        if (ret < 0) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed copying basename"
+                                        "to ctr_hard_link");
+                /* delete the corrupted entry */
+                __delete_hard_link_from_list (&ctr_hard_link);
+                ctr_hard_link = NULL;
+                goto out;
+        }
+
+        ret = 0;
+
+out:
+        UNLOCK (&ctr_xlator_ctx->lock);
+
+        return ret;
+}
+
+
+
+
+/* Delete all hardlinks */
+static inline int
+ctr_delete_all_hard_link (xlator_t *this,
+                          ctr_xlator_ctx_t *ctr_xlator_ctx)
+{
+        int ret = -1;
+        ctr_hard_link_t *ctr_hard_link     = NULL;
+        ctr_hard_link_t *tmp               = NULL;
+
+        GF_ASSERT (ctr_xlator_ctx);
+
+        LOCK (&ctr_xlator_ctx->lock);
+
+        list_for_each_entry_safe(ctr_hard_link, tmp,
+                        &ctr_xlator_ctx->hardlink_list, list)
+        {
+                /*Remove hard link from list*/
+                __delete_hard_link_from_list (&ctr_hard_link);
+                ctr_hard_link = NULL;
+
+        }
+
+
+        UNLOCK (&ctr_xlator_ctx->lock);
+
+        ret = 0;
+
+        return ret;
+}
+
+
+/* Please lock the inode before using this function */
+static inline ctr_xlator_ctx_t *
+__get_ctr_xlator_ctx (xlator_t  *this,
+                      inode_t   *inode)
+{
+        int ret                                 = 0;
+        uint64_t _addr                          = 0;
+        ctr_xlator_ctx_t *ctr_xlator_ctx        = NULL;
+
+        GF_ASSERT (this);
+        GF_ASSERT (inode);
+
+        ret = __inode_ctx_get (inode, this, &_addr);
+        if (ret < 0)
+                _addr = 0;
+        if (_addr != 0) {
+                ctr_xlator_ctx = (ctr_xlator_ctx_t *) (long)_addr;
+        }
+
+        return ctr_xlator_ctx;
+}
+
+
+ctr_xlator_ctx_t *
+init_ctr_xlator_ctx (xlator_t *this,
+                     inode_t *inode)
+{
+        int ret                                 = -1;
+        uint64_t _addr                          = 0;
+        ctr_xlator_ctx_t *ctr_xlator_ctx        = NULL;
+
+        GF_ASSERT (this);
+        GF_ASSERT (inode);
+
+        LOCK (&inode->lock);
+        {
+                ctr_xlator_ctx = __get_ctr_xlator_ctx (this, inode);
+                if (ctr_xlator_ctx) {
+                        ret = 0;
+                        goto out;
+                }
+                ctr_xlator_ctx = GF_CALLOC (1, sizeof (*ctr_xlator_ctx),
+                                                        gf_ctr_mt_xlator_ctx);
+                if (!ctr_xlator_ctx)
+                        goto out;
+
+                ret = LOCK_INIT (&ctr_xlator_ctx->lock);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                        "Failed init lock %s", strerror(ret));
+                        goto out;
+                }
+                _addr = (uint64_t) ctr_xlator_ctx;
+
+                ret = __inode_ctx_set (inode, this, &_addr);
+                if (ret) {
+                        goto out;
+                }
+
+                INIT_LIST_HEAD (&ctr_xlator_ctx->hardlink_list);
+
+        }
+        ret = 0;
+out:
+        if (ret) {
+                GF_FREE (ctr_xlator_ctx);
+                ctr_xlator_ctx = NULL;
+        }
+
+        UNLOCK (&inode->lock);
+
+        return ctr_xlator_ctx;
+}
+
+
+
+
+void
+fini_ctr_xlator_ctx (xlator_t *this,
+                     inode_t *inode)
+{
+        int ret                                 = 0;
+        uint64_t _addr                          = 0;
+        ctr_xlator_ctx_t *ctr_xlator_ctx        = NULL;
+
+
+        inode_ctx_del (inode, this, &_addr);
+        if (!_addr)
+                return;
+
+        ctr_xlator_ctx = (ctr_xlator_ctx_t *) (long) _addr;
+
+        ret = ctr_delete_all_hard_link (this, ctr_xlator_ctx);
+        if (ret) {
+                gf_log (this->name, GF_LOG_WARNING , "Failed deleting all hard"
+                                              " links from inode context");
+        }
+
+        LOCK_DESTROY (&ctr_xlator_ctx->lock);
+
+        GF_FREE (ctr_xlator_ctx);
+
+}
+
+
+
+
+ctr_xlator_ctx_t *
+get_ctr_xlator_ctx (xlator_t *this,
+                    inode_t *inode)
+{
+        ctr_xlator_ctx_t *ctr_xlator_ctx = NULL;
+
+        LOCK (&inode->lock);
+        ctr_xlator_ctx = __get_ctr_xlator_ctx (this, inode);
+        UNLOCK (&inode->lock);
+
+        return ctr_xlator_ctx;
+}
+
diff --git a/xlators/features/changetimerecorder/src/ctr-xlator-ctx.h b/xlators/features/changetimerecorder/src/ctr-xlator-ctx.h
new file mode 100644
index 00000000000..bc935014fc6
--- /dev/null
+++ b/xlators/features/changetimerecorder/src/ctr-xlator-ctx.h
@@ -0,0 +1,86 @@
+/*
+   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_XLATOR_CTX_H
+#define __CTR_XLATOR_CTX_H
+
+#include "xlator.h"
+#include "ctr_mem_types.h"
+#include "iatt.h"
+#include "glusterfs.h"
+#include "xlator.h"
+#include "logging.h"
+#include "locking.h"
+#include "common-utils.h"
+#include <time.h>
+#include <sys/time.h>
+
+typedef struct ctr_hard_link {
+        uuid_t                  pgfid;
+        char                    *base_name;
+        struct list_head        list;
+} ctr_hard_link_t;
+
+typedef struct ctr_xlator_ctx {
+        /* This represents the looked up hardlinks
+         * NOTE: This doesn't represent all physical hardlinks of the inode*/
+        struct list_head        hardlink_list;
+        gf_lock_t               lock;
+} ctr_xlator_ctx_t;
+
+
+ctr_hard_link_t *
+ctr_search_hard_link_ctx (xlator_t                  *this,
+                          ctr_xlator_ctx_t        *ctr_xlator_ctx,
+                          uuid_t                  pgfid,
+                          const char              *base_name);
+
+
+int
+ctr_add_hard_link (xlator_t          *this,
+               ctr_xlator_ctx_t         *ctr_xlator_ctx,
+               uuid_t                   pgfid,
+               const char               *base_name);
+
+
+
+int
+ctr_delete_hard_link (xlator_t                *this,
+                  ctr_xlator_ctx_t      *ctr_xlator_ctx,
+                  uuid_t                pgfid,
+                  const char            *base_name);
+
+
+int
+ctr_update_hard_link (xlator_t                *this,
+                  ctr_xlator_ctx_t      *ctr_xlator_ctx,
+                  uuid_t                pgfid,
+                  const char            *base_name,
+                  uuid_t                old_pgfid,
+                  const char            *old_base_name);
+
+
+ctr_xlator_ctx_t *
+get_ctr_xlator_ctx (xlator_t *this,
+                    inode_t *inode);
+
+
+
+
+ctr_xlator_ctx_t *
+init_ctr_xlator_ctx (xlator_t *this,
+                     inode_t *inode);
+
+
+void
+fini_ctr_xlator_ctx (xlator_t *this,
+                     inode_t *inode);
+
+#endif
diff --git a/xlators/features/changetimerecorder/src/ctr_mem_types.h b/xlators/features/changetimerecorder/src/ctr_mem_types.h
index 48387597814..f408c028e24 100644
--- a/xlators/features/changetimerecorder/src/ctr_mem_types.h
+++ b/xlators/features/changetimerecorder/src/ctr_mem_types.h
@@ -16,6 +16,8 @@
 
 enum gf_ctr_mem_types_ {
         gf_ctr_mt_private_t = gfdb_mt_end + 1,
+        gf_ctr_mt_xlator_ctx,
+        gf_ctr_mt_hard_link_t,
         gf_ctr_mt_end
 };
 #endif