From e9fd069cea243707a47d9892f04e488ddae6012c Mon Sep 17 00:00:00 2001

From: Dan Lambright <dlambrig@redhat.com>

Date: Sat, 27 Jun 2015 07:36:36 -0400

Subject: [PATCH 187/190] 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: I244f3155bdb27227c060cd256c5801ec8ea7cc90

BUG: 1232625

Signed-off-by: Dan Lambright <dlambrig@redhat.com>

Reviewed-on: https://code.engineering.redhat.com/gerrit/51756

Reviewed-by: Joseph Fernandes <josferna@redhat.com>

Tested-by: Joseph Fernandes <josferna@redhat.com>

---

 libglusterfs/src/gfdb/gfdb_data_store_types.h      |   15 ++-

 libglusterfs/src/gfdb/gfdb_sqlite3.c               |    2 -

 libglusterfs/src/gfdb/gfdb_sqlite3_helper.c        |   49 ++++++---

 xlators/cluster/dht/src/tier.h                     |    5 +-

 .../changetimerecorder/src/changetimerecorder.c    |  121 +++++++++++++++++---

 .../features/changetimerecorder/src/ctr-helper.c   |   10 ++

 .../features/changetimerecorder/src/ctr-helper.h   |  108 +++++++++++++++++-

 .../changetimerecorder/src/ctr-xlator-ctx.c        |   35 ++++++-

 .../changetimerecorder/src/ctr-xlator-ctx.h        |    4 +

 xlators/mgmt/glusterd/src/glusterd-volgen.c        |    8 ++

 xlators/mgmt/glusterd/src/glusterd-volume-set.c    |   24 ++++

 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 f642c63..fb41c38 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 7eca03f..94553fb 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 f3e0e97..40ee24c 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 7d4ea6b..34631b0 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 1fdeb7c..703df49 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                        = &gfi;;

         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             = &gfi;;

+        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 c4d9769..f5d4c47 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 a3df971..82a0c8b 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 a91a2bb..b8f6f03 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,