tier/ctr: Ignore creation of T file and Ctr Lookup heal improvememnts

This is a back port of 11334

1) Ignore creation of T file in ctr_mknod
2) Ignore lookup for T file in ctr_lookup
3) Ctr_lookup:
    a. If the gfid and pgfid in empty dont record
    b. Decreased log level for multiple heal attempts
    c. Inode/File heal happens after an expiry period, which is configurable.
    d. Hardlink heal happens after an expiry period, which is configurable.

> Change-Id: Id8eb5092e78beaec22d05f5283645081619e2452
> BUG: 1235269
> Signed-off-by: Joseph Fernandes <josferna@redhat.com>
> Reviewed-on: http://review.gluster.org/11334
> Tested-by: Gluster Build System <jenkins@build.gluster.com>
> Reviewed-by: Dan Lambright <dlambrig@redhat.com>
> Tested-by: Dan Lambright <dlambrig@redhat.com>

Change-Id: Ia28a5cf975e41d318906f707deca447aaa35630f
BUG: 1236288
Signed-off-by: Dan Lambright <dlambrig@redhat.com>
Reviewed-on: http://review.gluster.org/11446
Reviewed-by: Joseph Fernandes
Tested-by: Joseph Fernandes
Tested-by: Gluster Build System <jenkins@build.gluster.com>

11 files changed, 336 insertions, 45 deletions

diff --git a/libglusterfs/src/gfdb/gfdb_data_store_types.h b/libglusterfs/src/gfdb/gfdb_data_store_types.h
index f642c63b049..fb41c385808 100644
--- a/libglusterfs/src/gfdb/gfdb_data_store_types.h
+++ b/libglusterfs/src/gfdb/gfdb_data_store_types.h
@@ -26,6 +26,16 @@
 #include "dict.h"
 #include "libglusterfs-messages.h"
 
+/*
+ * Helps in dynamically choosing log level
+ * */
+static inline gf_loglevel_t
+_gfdb_log_level (gf_loglevel_t given_level,
+                 gf_boolean_t ignore_level)
+{
+        return (ignore_level) ? GF_LOG_DEBUG : given_level;
+}
+
 typedef enum gf_db_operation {
         GFDB_INVALID_DB_OP = -1,
         /* Query DB OPS : All the Query DB_OP should be added */
@@ -71,8 +81,6 @@ typedef enum gf_db_operation {
 #define GF_COL_LINK_UPDATE              "LINK_UPDATE"
 
 
-
-
 /***********************Time related********************************/
 /*1 sec = 1000000 microsec*/
 #define GFDB_MICROSEC         1000000
@@ -305,6 +313,9 @@ typedef struct gfdb_db_record {
         /* Global flag to Record/Not Record wind or wind time.
          * This flag will overrule do_record_uwind_time*/
         gf_boolean_t                    do_record_times;
+        /* Ignoring errors while inserting.
+         * */
+        gf_boolean_t                    ignore_errors;
 } gfdb_db_record_t;
 
 
diff --git a/libglusterfs/src/gfdb/gfdb_sqlite3.c b/libglusterfs/src/gfdb/gfdb_sqlite3.c
index 7eca03f40e2..94553fba70d 100644
--- a/libglusterfs/src/gfdb/gfdb_sqlite3.c
+++ b/libglusterfs/src/gfdb/gfdb_sqlite3.c
@@ -176,8 +176,6 @@ sql_stmt_fini (char **sql_stmt)
 
 /******************************************************************************
  *                      DB Essential functions used by
- *                      create/insert/delete/update/query functions
- *                      > execute_sqlstmt ()
  *                      > gf_open_sqlite3_conn ()
  *                      > gf_close_sqlite3_conn ()
  * ***************************************************************************/
diff --git a/libglusterfs/src/gfdb/gfdb_sqlite3_helper.c b/libglusterfs/src/gfdb/gfdb_sqlite3_helper.c
index f3e0e9711a9..40ee24c9627 100644
--- a/libglusterfs/src/gfdb/gfdb_sqlite3_helper.c
+++ b/libglusterfs/src/gfdb/gfdb_sqlite3_helper.c
@@ -9,7 +9,7 @@
 */
 
 #include "gfdb_sqlite3_helper.h"
-#include "libglusterfs-messages.h"
+
 
 #define GFDB_SQL_STMT_SIZE 256
 
@@ -309,7 +309,8 @@ gf_sql_insert_link (gf_sql_connection_t  *sql_conn,
                    char                 *pargfid,
                    char                 *basename,
                    char                 *basepath,
-                   gf_boolean_t         link_consistency)
+                   gf_boolean_t         link_consistency,
+                   gf_boolean_t         ignore_errors)
 {
         int ret = -1;
         sqlite3_stmt *insert_stmt = NULL;
@@ -332,8 +333,11 @@ gf_sql_insert_link (gf_sql_connection_t  *sql_conn,
         ret = sqlite3_prepare (sql_conn->sqlite3_db_conn, insert_str, -1,
                                 &insert_stmt, 0);
         if (ret != SQLITE_OK) {
-                gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
-                        LG_MSG_PREPARE_FAILED, "Failed preparing insert "
+                gf_msg (GFDB_STR_SQLITE3,
+                        _gfdb_log_level (GF_LOG_ERROR, ignore_errors),
+                        0,
+                        LG_MSG_PREPARE_FAILED,
+                        "Failed preparing insert "
                         "statment %s : %s", insert_str,
                         sqlite3_errmsg (sql_conn->sqlite3_db_conn));
                 ret = -1;
@@ -343,8 +347,11 @@ gf_sql_insert_link (gf_sql_connection_t  *sql_conn,
         /*Bind gfid*/
         ret = sqlite3_bind_text (insert_stmt, 1, gfid, -1, NULL);
         if (ret != SQLITE_OK) {
-                gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
-                        LG_MSG_BINDING_FAILED, "Failed binding gfid %s : %s",
+                gf_msg (GFDB_STR_SQLITE3,
+                        _gfdb_log_level (GF_LOG_ERROR, ignore_errors),
+                        0,
+                        LG_MSG_BINDING_FAILED,
+                        "Failed binding gfid %s : %s",
                         gfid, sqlite3_errmsg (sql_conn->sqlite3_db_conn));
                 ret = -1;
                 goto out;
@@ -353,8 +360,10 @@ gf_sql_insert_link (gf_sql_connection_t  *sql_conn,
         /*Bind pargfid*/
         ret = sqlite3_bind_text (insert_stmt, 2, pargfid, -1, NULL);
         if (ret != SQLITE_OK) {
-                gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
-                        LG_MSG_BINDING_FAILED, "Failed binding parent gfid %s "
+                gf_msg (GFDB_STR_SQLITE3,
+                        _gfdb_log_level (GF_LOG_ERROR, ignore_errors),
+                        0, LG_MSG_BINDING_FAILED,
+                        "Failed binding parent gfid %s "
                         ": %s", pargfid,
                         sqlite3_errmsg (sql_conn->sqlite3_db_conn));
                 ret = -1;
@@ -364,8 +373,9 @@ gf_sql_insert_link (gf_sql_connection_t  *sql_conn,
         /*Bind basename*/
         ret = sqlite3_bind_text (insert_stmt, 3, basename, -1, NULL);
         if (ret != SQLITE_OK) {
-                gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
-                        LG_MSG_BINDING_FAILED,
+                gf_msg (GFDB_STR_SQLITE3,
+                        _gfdb_log_level (GF_LOG_ERROR, ignore_errors),
+                        0, LG_MSG_BINDING_FAILED,
                         "Failed binding basename %s : %s", basename,
                         sqlite3_errmsg (sql_conn->sqlite3_db_conn));
                 ret = -1;
@@ -375,8 +385,10 @@ gf_sql_insert_link (gf_sql_connection_t  *sql_conn,
         /*Bind basepath*/
         ret = sqlite3_bind_text (insert_stmt, 4, basepath, -1, NULL);
         if (ret != SQLITE_OK) {
-                gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
-                        LG_MSG_BINDING_FAILED, "Failed binding basepath %s : "
+                gf_msg (GFDB_STR_SQLITE3,
+                        _gfdb_log_level (GF_LOG_ERROR, ignore_errors), 0,
+                        LG_MSG_BINDING_FAILED,
+                        "Failed binding basepath %s : "
                         "%s", basepath,
                         sqlite3_errmsg (sql_conn->sqlite3_db_conn));
                 ret = -1;
@@ -385,8 +397,11 @@ gf_sql_insert_link (gf_sql_connection_t  *sql_conn,
 
         /*Execute the prepare statement*/
         if (sqlite3_step (insert_stmt) != SQLITE_DONE) {
-                gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0, LG_MSG_EXEC_FAILED,
-                        "Failed executing the prepared stmt %s %s %s %s %s : %s",
+                gf_msg (GFDB_STR_SQLITE3,
+                        _gfdb_log_level (GF_LOG_ERROR, ignore_errors),
+                        0, LG_MSG_EXEC_FAILED,
+                        "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;
@@ -777,7 +792,8 @@ gf_sql_insert_wind (gf_sql_connection_t  *sql_conn,
                                         gfid_str, pargfid_str,
                                         gfdb_db_record->file_name,
                                         gfdb_db_record->file_path,
-                                        gfdb_db_record->link_consistency);
+                                        gfdb_db_record->link_consistency,
+                                        gfdb_db_record->ignore_errors);
                         if (ret) {
                                 gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
                                         LG_MSG_INSERT_FAILED, "Failed "
@@ -836,7 +852,8 @@ gf_sql_insert_wind (gf_sql_connection_t  *sql_conn,
                                         gfid_str, pargfid_str,
                                         gfdb_db_record->file_name,
                                         gfdb_db_record->file_path,
-                                        gfdb_db_record->link_consistency);
+                                        gfdb_db_record->link_consistency,
+                                        gfdb_db_record->ignore_errors);
                                 if (ret) {
                                         gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR,
                                                 0, LG_MSG_INSERT_FAILED,
diff --git a/xlators/cluster/dht/src/tier.h b/xlators/cluster/dht/src/tier.h
index 7d4ea6b7ebe..34631b0a841 100644
--- a/xlators/cluster/dht/src/tier.h
+++ b/xlators/cluster/dht/src/tier.h
@@ -27,9 +27,8 @@
 
 #define DEFAULT_PROMOTE_FREQ_SEC 120
 #define DEFAULT_DEMOTE_FREQ_SEC  120
-#define DEFAULT_WRITE_FREQ_SEC 120
-#define DEFAULT_READ_FREQ_SEC 120
-
+#define DEFAULT_WRITE_FREQ_SEC 0
+#define DEFAULT_READ_FREQ_SEC 0
 /*
  * Size of timer wheel. We would not promote or demote less
  * frequently than this number.
diff --git a/xlators/features/changetimerecorder/src/changetimerecorder.c b/xlators/features/changetimerecorder/src/changetimerecorder.c
index 1fdeb7c76ca..703df496e95 100644
--- a/xlators/features/changetimerecorder/src/changetimerecorder.c
+++ b/xlators/features/changetimerecorder/src/changetimerecorder.c
@@ -114,7 +114,7 @@ ctr_lookup_wind(call_frame_t                    *frame,
                 /*Don't record time at all*/
                 CTR_DB_REC(ctr_local).do_record_times = _gf_false;
 
-                /*Copy gfid into db record*/
+                /* Copy gfid into db record*/
                 gf_uuid_copy (CTR_DB_REC(ctr_local).gfid,
                                 *(ctr_inode_cx->gfid));
 
@@ -131,6 +131,12 @@ ctr_lookup_wind(call_frame_t                    *frame,
                 strcpy (CTR_DB_REC(ctr_local).file_path,
                         NEW_LINK_CX(ctr_inode_cx)->basepath);
 
+                /* Since we are in lookup we can ignore errors while
+                 * Inserting in the DB, because there may be many
+                 * to write to the DB attempts for healing.
+                 * We dont want to log all failed attempts and
+                 * bloat the log*/
+                 ctr_local->gfdb_db_record.ignore_errors = _gf_true;
         }
 
         ret = 0;
@@ -171,8 +177,11 @@ ctr_lookup_unwind (call_frame_t          *frame,
                 ret = insert_record(_priv->_db_conn,
                                 &ctr_local->gfdb_db_record);
                 if (ret == -1) {
-                        gf_msg(this->name, GF_LOG_ERROR, 0,
-                               CTR_MSG_FILL_CTR_LOCAL_ERROR_UNWIND,
+                        gf_msg (this->name,
+                                _gfdb_log_level (GF_LOG_ERROR,
+                                        ctr_local->
+                                        gfdb_db_record.ignore_errors),
+                                0, CTR_MSG_FILL_CTR_LOCAL_ERROR_UNWIND,
                                "UNWIND: Error filling ctr local");
                         goto out;
                 }
@@ -198,9 +207,11 @@ 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;
+        int ret                                 = -1;
+        ctr_xlator_ctx_t *ctr_xlator_ctx        = NULL;
+        gf_ctr_local_t *ctr_local               = NULL;
+        ctr_heal_ret_val_t  ret_val             = CTR_CTX_ERROR;
+        gf_boolean_t    _is_heal_needed         = _gf_false;
 
         CTR_IS_DISABLED_THEN_GOTO(this, out);
         CTR_IF_INTERNAL_FOP_THEN_GOTO (frame, dict, out);
@@ -223,34 +234,70 @@ ctr_lookup_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 goto out;
         }
 
+        /* if the lookup is for dht link donot record*/
+        if (dht_is_linkfile (buf, dict)) {
+                gf_msg_trace (this->name, 0, "Ignoring Lookup "
+                                "for dht link file");
+                goto out;
+        }
+
         ctr_local = frame->local;
         /*Assign the proper inode type*/
         ctr_local->ia_inode_type = inode->ia_type;
 
+        /* Copy gfid directly from inode */
+        gf_uuid_copy (CTR_DB_REC(ctr_local).gfid, inode->gfid);
+
+        /* Checking if gfid and parent gfid is valid */
+        if (gf_uuid_is_null(CTR_DB_REC(ctr_local).gfid) ||
+                gf_uuid_is_null(CTR_DB_REC(ctr_local).pargfid)) {
+                gf_msg_trace (this->name, 0,
+                        "Invalid GFID");
+                goto out;
+        }
+
         /* 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) {
+                 /* This marks inode heal */
                 CTR_DB_REC(ctr_local).gfdb_fop_type = GFDB_FOP_CREATE_WRITE;
+                _is_heal_needed = _gf_true;
         }
 
         /* 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_msg_trace (this->name, 0, "Failed adding hard link");
+        ret_val = add_hard_link_ctx (frame, this, inode);
+        if (ret_val == CTR_CTX_ERROR) {
+                gf_msg_trace (this->name, 0,
+                        "Failed adding hardlink to list");
                 goto out;
         }
+        /* If inode needs healing then heal the hardlink also */
+        else if (ret_val & CTR_TRY_INODE_HEAL) {
+                /* This marks inode heal */
+                CTR_DB_REC(ctr_local).gfdb_fop_type = GFDB_FOP_CREATE_WRITE;
+                _is_heal_needed = _gf_true;
+        }
+        /* If hardlink needs healing */
+        else if (ret_val & CTR_TRY_HARDLINK_HEAL) {
+                _is_heal_needed = _gf_true;
+        }
 
-        /* Inserts the ctr_db_record populated by ctr_lookup_wind
+        /* If lookup heal needed */
+        if (!_is_heal_needed)
+                goto out;
+
+        /* FINALLY HEAL : 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_msg_trace (this->name, 0, "Failed inserting link wind");
+                gf_msg_trace (this->name, 0,
+                        "Failed healing/inserting link");
         }
 
 
@@ -793,6 +840,7 @@ ctr_unlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         }
         /*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) {
@@ -819,6 +867,7 @@ ctr_unlink (call_frame_t *frame, xlator_t *this,
         gf_ctr_link_context_t ctr_link_cx;
         gf_ctr_link_context_t *_link_cx = &ctr_link_cx;
         gf_boolean_t is_xdata_created = _gf_false;
+        struct iatt dummy_stat        = {0};
 
         GF_ASSERT (frame);
 
@@ -835,6 +884,12 @@ ctr_unlink (call_frame_t *frame, xlator_t *this,
         /*Internal FOP*/
         _inode_cx->is_internal_fop = CTR_IS_INTERNAL_FOP(frame, xdata);
 
+        /* If its a internal FOP and dht link file donot record*/
+        if (_inode_cx->is_internal_fop &&
+                        dht_is_linkfile (&dummy_stat, xdata)) {
+                goto out;
+        }
+
         /*record into the database*/
         ret = ctr_insert_wind(frame, this, _inode_cx);
         if (ret) {
@@ -1014,12 +1069,13 @@ ctr_mknod_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                  struct iatt *postparent, dict_t *xdata)
 {
         int ret = -1;
+        ctr_heal_ret_val_t ret_val = CTR_CTX_ERROR;
 
         CTR_IS_DISABLED_THEN_GOTO(this, out);
 
         /* Add hard link to the list */
-        ret = add_hard_link_ctx (frame, this, inode);
-        if (ret) {
+        ret_val = add_hard_link_ctx (frame, this, inode);
+        if (ret_val == CTR_CTX_ERROR) {
                 gf_msg_trace (this->name, 0, "Failed adding hard link");
         }
 
@@ -1053,6 +1109,7 @@ ctr_mknod (call_frame_t *frame, xlator_t *this,
         uuid_t *ptr_gfid                        = &gfid;
 
         CTR_IS_DISABLED_THEN_GOTO(this, out);
+        CTR_IF_INTERNAL_FOP_THEN_GOTO (frame, xdata, out);
 
         GF_ASSERT(frame);
         GF_ASSERT(frame->root);
@@ -1073,9 +1130,6 @@ ctr_mknod (call_frame_t *frame, xlator_t *this,
                 *ptr_gfid, _link_cx, NULL,
                 GFDB_FOP_CREATE_WRITE, GFDB_FOP_WIND);
 
-        /*Internal FOP*/
-        _inode_cx->is_internal_fop = CTR_IS_INTERNAL_FOP(frame, xdata);
-
         /*record into the database*/
         ret = ctr_insert_wind(frame, this, _inode_cx);
         if (ret) {
@@ -1140,6 +1194,7 @@ ctr_create (call_frame_t *frame, xlator_t *this,
         void             *uuid_req              = NULL;
         uuid_t            gfid                  = {0,};
         uuid_t            *ptr_gfid             = &gfid;
+        struct iatt dummy_stat                  = {0};
 
         CTR_IS_DISABLED_THEN_GOTO(this, out);
 
@@ -1167,6 +1222,12 @@ ctr_create (call_frame_t *frame, xlator_t *this,
         /*Internal FOP*/
         _inode_cx->is_internal_fop = CTR_IS_INTERNAL_FOP(frame, xdata);
 
+        /* If its a internal FOP and dht link file donot record*/
+        if (_inode_cx->is_internal_fop &&
+                        dht_is_linkfile (&dummy_stat, xdata)) {
+                goto out;
+        }
+
         /*record into the database*/
         ret = ctr_insert_wind(frame, this, &ctr_inode_cx);
         if (ret) {
@@ -1222,6 +1283,7 @@ ctr_link (call_frame_t *frame, xlator_t *this,
         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;
+        struct iatt dummy_stat          = {0};
 
         CTR_IS_DISABLED_THEN_GOTO(this, out);
 
@@ -1240,6 +1302,13 @@ ctr_link (call_frame_t *frame, xlator_t *this,
         /*Internal FOP*/
         _inode_cx->is_internal_fop = CTR_IS_INTERNAL_FOP(frame, xdata);
 
+        /* If its a internal FOP and dht link file donot record*/
+        if (_inode_cx->is_internal_fop &&
+                        dht_is_linkfile (&dummy_stat, xdata)) {
+                goto out;
+        }
+
+
         /*record into the database*/
         ret = ctr_insert_wind(frame, this, _inode_cx);
         if (ret) {
@@ -1334,6 +1403,14 @@ reconfigure (xlator_t *this, dict_t *options)
         GF_OPTION_RECONF ("ctr_link_consistency", _priv->ctr_link_consistency,
                         options, bool, out);
 
+        GF_OPTION_RECONF ("ctr_inode_heal_expire_period",
+                                _priv->ctr_inode_heal_expire_period,
+                                options, uint64, out);
+
+        GF_OPTION_RECONF ("ctr_hardlink_heal_expire_period",
+                                _priv->ctr_hardlink_heal_expire_period,
+                                options, uint64, out);
+
         GF_OPTION_RECONF ("record-exit", _priv->ctr_record_unwind, options,
                           bool, out);
 
@@ -1385,6 +1462,10 @@ init (xlator_t *this)
         _priv->gfdb_sync_type           = GFDB_DB_SYNC;
         _priv->enabled                  = _gf_true;
         _priv->_db_conn                 = NULL;
+        _priv->ctr_hardlink_heal_expire_period =
+                                CTR_DEFAULT_HARDLINK_EXP_PERIOD;
+        _priv->ctr_inode_heal_expire_period =
+                                CTR_DEFAULT_INODE_EXP_PERIOD;
 
         /*Extract ctr xlator options*/
         ret_db = extract_ctr_options (this, _priv);
@@ -1549,6 +1630,14 @@ struct volume_options options[] = {
           .value = {"on", "off"},
           .default_value = "off"
         },
+        { .key  = {"ctr_hardlink_heal_expire_period"},
+          .type = GF_OPTION_TYPE_INT,
+          .default_value = "300"
+        },
+        { .key  = {"ctr_inode_heal_expire_period"},
+          .type = GF_OPTION_TYPE_INT,
+          .default_value = "300"
+        },
         { .key  = {"hot-brick"},
           .type = GF_OPTION_TYPE_BOOL,
           .value = {"on", "off"},
diff --git a/xlators/features/changetimerecorder/src/ctr-helper.c b/xlators/features/changetimerecorder/src/ctr-helper.c
index c4d97695bd0..f5d4c474674 100644
--- a/xlators/features/changetimerecorder/src/ctr-helper.c
+++ b/xlators/features/changetimerecorder/src/ctr-helper.c
@@ -276,6 +276,16 @@ int extract_ctr_options (xlator_t *this, gf_ctr_private_t *_priv) {
         GF_OPTION_INIT ("ctr_link_consistency", _priv->ctr_link_consistency,
                         bool, out);
 
+        /*Extract ctr_inode_heal_expire_period */
+        GF_OPTION_INIT ("ctr_inode_heal_expire_period",
+                        _priv->ctr_inode_heal_expire_period,
+                        uint64, out);
+
+        /*Extract ctr_hardlink_heal_expire_period*/
+        GF_OPTION_INIT ("ctr_hardlink_heal_expire_period",
+                        _priv->ctr_hardlink_heal_expire_period,
+                        uint64, out);
+
         /*Extract flag for hot tier brick*/
         GF_OPTION_INIT ("hot-brick", _priv->ctr_hot_brick, bool, out);
 
diff --git a/xlators/features/changetimerecorder/src/ctr-helper.h b/xlators/features/changetimerecorder/src/ctr-helper.h
index a3df9716eaa..82a0c8bc93c 100644
--- a/xlators/features/changetimerecorder/src/ctr-helper.h
+++ b/xlators/features/changetimerecorder/src/ctr-helper.h
@@ -31,6 +31,9 @@
 #include "ctr-xlator-ctx.h"
 #include "ctr-messages.h"
 
+#define CTR_DEFAULT_HARDLINK_EXP_PERIOD 300  /* Five mins */
+#define CTR_DEFAULT_INODE_EXP_PERIOD    300 /* Five mins */
+
 /*CTR Xlator Private structure*/
 typedef struct gf_ctr_private {
         gf_boolean_t                    enabled;
@@ -43,6 +46,8 @@ typedef struct gf_ctr_private {
         gfdb_db_type_t                  gfdb_db_type;
         gfdb_sync_type_t                gfdb_sync_type;
         gfdb_conn_node_t                *_db_conn;
+        uint64_t                        ctr_hardlink_heal_expire_period;
+        uint64_t                        ctr_inode_heal_expire_period;
 } gf_ctr_private_t;
 
 
@@ -472,19 +477,91 @@ out:
 
 /******************************* Hard link function ***************************/
 
-static inline int
+static inline gf_boolean_t
+__is_inode_expired (ctr_xlator_ctx_t *ctr_xlator_ctx,
+                    gf_ctr_private_t *_priv,
+                    gfdb_time_t *current_time)
+{
+        gf_boolean_t    ret       = _gf_false;
+        uint64_t        time_diff = 0;
+
+        GF_ASSERT (ctr_xlator_ctx);
+        GF_ASSERT (_priv);
+        GF_ASSERT (current_time);
+
+        time_diff = current_time->tv_sec -
+                        ctr_xlator_ctx->inode_heal_period;
+
+        ret = (time_diff >= _priv->ctr_inode_heal_expire_period) ?
+                        _gf_true : _gf_false;
+        return ret;
+}
+
+static inline gf_boolean_t
+__is_hardlink_expired (ctr_hard_link_t *ctr_hard_link,
+                       gf_ctr_private_t *_priv,
+                       gfdb_time_t *current_time)
+{
+        gf_boolean_t    ret       = _gf_false;
+        uint64_t        time_diff = 0;
+
+        GF_ASSERT (ctr_hard_link);
+        GF_ASSERT (_priv);
+        GF_ASSERT (current_time);
+
+        time_diff = current_time->tv_sec -
+                        ctr_hard_link->hardlink_heal_period;
+
+        ret = ret || (time_diff >= _priv->ctr_hardlink_heal_expire_period) ?
+                        _gf_true : _gf_false;
+
+        return ret;
+}
+
+
+/* Return values of heal*/
+typedef enum ctr_heal_ret_val {
+        CTR_CTX_ERROR = -1,
+        /* No healing required */
+        CTR_TRY_NO_HEAL = 0,
+        /* Try healing hard link */
+        CTR_TRY_HARDLINK_HEAL = 1,
+        /* Try healing inode */
+        CTR_TRY_INODE_HEAL = 2,
+} ctr_heal_ret_val_t;
+
+
+
+/**
+ * @brief Function to add hard link to the inode context variable.
+ *        The inode context maintainences a in-memory list. This is used
+ *        smart healing of database.
+ * @param frame of the FOP
+ * @param this is the Xlator instant
+ * @param inode
+ * @return Return ctr_heal_ret_val_t
+ */
+
+static inline ctr_heal_ret_val_t
 add_hard_link_ctx (call_frame_t *frame,
                    xlator_t     *this,
                    inode_t      *inode)
 {
+        ctr_heal_ret_val_t ret_val = CTR_TRY_NO_HEAL;
         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_ctr_private_t *_priv           = NULL;
+        gfdb_time_t       current_time    = {0};
+
 
         GF_ASSERT (frame);
         GF_ASSERT (this);
         GF_ASSERT (inode);
+        GF_ASSERT (this->private);
+
+        _priv = this->private;
 
         ctr_local = frame->local;
         if (!ctr_local) {
@@ -509,7 +586,29 @@ add_hard_link_ctx (call_frame_t *frame,
                                 CTR_DB_REC(ctr_local).file_name);
         /* if there then ignore */
         if (ctr_hard_link) {
-                ret = 1;
+
+                ret = gettimeofday (&current_time, NULL);
+                if (ret == -1) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "Failed to get current time");
+                        ret_val = CTR_CTX_ERROR;
+                        goto unlock;
+                }
+
+                if (__is_hardlink_expired (ctr_hard_link,
+                                           _priv, &current_time)) {
+                        ctr_hard_link->hardlink_heal_period =
+                                        current_time.tv_sec;
+                        ret_val = ret_val | CTR_TRY_HARDLINK_HEAL;
+                }
+
+                if (__is_inode_expired (ctr_xlator_ctx,
+                                           _priv, &current_time)) {
+                        ctr_xlator_ctx->inode_heal_period =
+                                                current_time.tv_sec;
+                        ret_val = ret_val | CTR_TRY_INODE_HEAL;
+                }
+
                 goto unlock;
         }
 
@@ -521,14 +620,15 @@ add_hard_link_ctx (call_frame_t *frame,
                 gf_msg (this->name, GF_LOG_ERROR, 0,
                         CTR_MSG_ADD_HARDLINK_TO_CTR_INODE_CONTEXT_FAILED,
                         "Failed to add hardlink to the ctr inode context");
+                ret_val = CTR_CTX_ERROR;
                 goto unlock;
         }
 
-        ret = 0;
+        ret_val = CTR_TRY_NO_HEAL;
 unlock:
         UNLOCK (&ctr_xlator_ctx->lock);
 out:
-        return ret;
+        return ret_val;
 }
 
 static inline int
diff --git a/xlators/features/changetimerecorder/src/ctr-xlator-ctx.c b/xlators/features/changetimerecorder/src/ctr-xlator-ctx.c
index a91a2bbefc0..b8f6f0301d9 100644
--- a/xlators/features/changetimerecorder/src/ctr-xlator-ctx.c
+++ b/xlators/features/changetimerecorder/src/ctr-xlator-ctx.c
@@ -10,6 +10,8 @@
 
 #include "ctr-xlator-ctx.h"
 #include "ctr-messages.h"
+#include <time.h>
+#include <sys/time.h>
 
 #define IS_THE_ONLY_HARDLINK(ctr_hard_link)\
         (ctr_hard_link->list.next == ctr_hard_link->list.prev)
@@ -63,13 +65,14 @@ out:
 
 /* Please lock the ctr_xlator_ctx before using this function */
 int
-ctr_add_hard_link (xlator_t           *this,
+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;
+        struct timeval  current_time                   = {0};
 
         GF_ASSERT (this);
         GF_ASSERT (ctr_xlator_ctx);
@@ -98,10 +101,19 @@ ctr_add_hard_link (xlator_t           *this,
                 goto error;
         }
 
+        ret = gettimeofday (&current_time, NULL);
+        if (ret == -1) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Failed to get current time");
+                goto error;
+        }
+
         /*Add the hard link to the list*/
         list_add_tail (&ctr_hard_link->list,
                         &ctr_xlator_ctx->hardlink_list);
 
+        ctr_hard_link->hardlink_heal_period = current_time.tv_sec;
+
         /*aal izz well!*/
         ret = 0;
         goto out;
@@ -169,8 +181,9 @@ ctr_update_hard_link (xlator_t                *this,
                   uuid_t                old_pgfid,
                   const char            *old_base_name)
 {
-        int ret                         = -1;
+        int ret                            = -1;
         ctr_hard_link_t *ctr_hard_link     = NULL;
+        struct timeval current_time        = {0};
 
         GF_ASSERT (this);
         GF_ASSERT (ctr_xlator_ctx);
@@ -212,6 +225,15 @@ ctr_update_hard_link (xlator_t                *this,
                 goto out;
         }
 
+        ret = gettimeofday (&current_time, NULL);
+        if (ret == -1) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Failed to get current time");
+                ctr_hard_link->hardlink_heal_period = 0;
+        } else {
+                ctr_hard_link->hardlink_heal_period = current_time.tv_sec;
+        }
+
         ret = 0;
 
 out:
@@ -284,6 +306,7 @@ init_ctr_xlator_ctx (xlator_t *this,
         int ret                                 = -1;
         uint64_t _addr                          = 0;
         ctr_xlator_ctx_t *ctr_xlator_ctx        = NULL;
+        struct timeval current_time             = {0};
 
         GF_ASSERT (this);
         GF_ASSERT (inode);
@@ -316,6 +339,14 @@ init_ctr_xlator_ctx (xlator_t *this,
 
                 INIT_LIST_HEAD (&ctr_xlator_ctx->hardlink_list);
 
+                ret = gettimeofday (&current_time, NULL);
+                if (ret == -1) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "Failed to get current time");
+                        goto out;
+                }
+
+                ctr_xlator_ctx->inode_heal_period = current_time.tv_sec;
         }
         ret = 0;
 out:
diff --git a/xlators/features/changetimerecorder/src/ctr-xlator-ctx.h b/xlators/features/changetimerecorder/src/ctr-xlator-ctx.h
index bc935014fc6..7f1c6cb1712 100644
--- a/xlators/features/changetimerecorder/src/ctr-xlator-ctx.h
+++ b/xlators/features/changetimerecorder/src/ctr-xlator-ctx.h
@@ -25,6 +25,9 @@
 typedef struct ctr_hard_link {
         uuid_t                  pgfid;
         char                    *base_name;
+        /* Hardlink expiry : Defines the expiry period after which a
+         * database heal is attempted. */
+        uint64_t                  hardlink_heal_period;
         struct list_head        list;
 } ctr_hard_link_t;
 
@@ -32,6 +35,7 @@ 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;
+        uint64_t                inode_heal_period;
         gf_lock_t               lock;
 } ctr_xlator_ctx_t;
 
diff --git a/xlators/mgmt/glusterd/src/glusterd-volgen.c b/xlators/mgmt/glusterd/src/glusterd-volgen.c
index 09671a83126..c3ec10b8b16 100644
--- a/xlators/mgmt/glusterd/src/glusterd-volgen.c
+++ b/xlators/mgmt/glusterd/src/glusterd-volgen.c
@@ -1717,6 +1717,14 @@ brick_graph_add_changetimerecorder (volgen_graph_t *graph,
         if (ret)
                 goto out;
 
+        ret = xlator_set_option (xl, "ctr_hardlink_heal_expire_period", "300");
+        if (ret)
+                goto out;
+
+        ret = xlator_set_option (xl, "ctr_inode_heal_expire_period", "300");
+        if (ret)
+                goto out;
+
         ret = xlator_set_option (xl, "record-entry", "on");
         if (ret)
                 goto out;
diff --git a/xlators/mgmt/glusterd/src/glusterd-volume-set.c b/xlators/mgmt/glusterd/src/glusterd-volume-set.c
index 66ae63f20a7..d418f4ab34d 100644
--- a/xlators/mgmt/glusterd/src/glusterd-volume-set.c
+++ b/xlators/mgmt/glusterd/src/glusterd-volume-set.c
@@ -1910,6 +1910,30 @@ struct volopt_map_entry glusterd_volopt_map[] = {
                          "updates by Change Time Recorder Xlator. When recording in a crash "
                          "consistent way the data operations will experience more latency."
         },
+        { .key         = "features.ctr_hardlink_heal_expire_period",
+          .voltype     = "features/changetimerecorder",
+          .value       = "300",
+          .option      = "ctr_hardlink_heal_expire_period",
+          .op_version  = GD_OP_VERSION_3_7_2,
+          .description = "Defines the expiry period of in-memory "
+                         "hardlink of an inode,"
+                         "used by lookup heal in Change Time Recorder."
+                         "Once the expiry period"
+                         "hits an attempt to heal the database per "
+                         "hardlink is done and the "
+                         "in-memory hardlink period is reset"
+        },
+        { .key         = "features.ctr_inode_heal_expire_period",
+          .voltype     = "features/changetimerecorder",
+          .value       = "300",
+          .option      = "ctr_inode_heal_expire_period",
+          .op_version  = GD_OP_VERSION_3_7_2,
+          .description = "Defines the expiry period of in-memory inode,"
+                         "used by lookup heal in Change Time Recorder. "
+                         "Once the expiry period"
+                         "hits an attempt to heal the database per "
+                         "inode is done"
+        },
 #endif /* USE_GFDB */
         { .key         = "locks.trace",
           .voltype     = "features/locks",