tier/ctr: Providing option to record or ignore metadata heat

Currently we heat up a file for both data and metadata write.
Here we provide a ctr xlator option called "ctr-record-metadata-heat"
were the admin can decide on recording metadata heat i.e heatup a
file on metadata writes or not.
Metadata data operation are
a. setattr: explicit changing of atime/mtime using utimes,
            changing of posix permissions of the file
b. rename: Renaming a file,
c. unlink, link: adding or deleting hardlinks
d. xattrs: setting or removal of xattrs.

NOTE: atime, mtime and ctime change through writev, readv, truncate, mknod
and create will not be considered here as these fops are data and primary
metadata fops.

Defaultly "ctr-record-metadata-heat" is off. Admin can
switch it on using gluster volume set command.

Backport of http://review.gluster.org/12540

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

Change-Id: I986c319f0cc337b0692a1dd02f71254e786afac4
BUG: 1282315
Reviewed-on: http://review.gluster.org/12582
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Tested-by: NetBSD Build System <jenkins@build.gluster.org>
Reviewed-by: Dan Lambright <dlambrig@redhat.com>
Tested-by: Dan Lambright <dlambrig@redhat.com>

4 files changed, 251 insertions, 54 deletions

diff --git a/xlators/features/changetimerecorder/src/changetimerecorder.c b/xlators/features/changetimerecorder/src/changetimerecorder.c
index 9602fd70649..c329812fda3 100644
--- a/xlators/features/changetimerecorder/src/changetimerecorder.c
+++ b/xlators/features/changetimerecorder/src/changetimerecorder.c
@@ -352,7 +352,7 @@ ctr_lookup (call_frame_t *frame, xlator_t *this,
         if (ret) {
                 gf_msg (this->name, GF_LOG_ERROR, 0,
                         CTR_MSG_INSERT_LINK_WIND_FAILED,
-                        "Failed inserting link wind");
+                        "Failed to insert link wind");
         }
 
 out:
@@ -380,7 +380,7 @@ ctr_writev_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         if (ret) {
                 gf_msg (this->name, GF_LOG_ERROR, 0,
                         CTR_MSG_INSERT_WRITEV_UNWIND_FAILED,
-                        "Failed inserting writev unwind");
+                        "Failed to insert writev unwind");
         }
 
 
@@ -414,7 +414,7 @@ ctr_writev (call_frame_t *frame, xlator_t *this, fd_t *fd,
         if (ret) {
                 gf_msg (this->name, GF_LOG_ERROR, 0,
                         CTR_MSG_INSERT_WRITEV_WIND_FAILED,
-                        "Failed inserting writev wind");
+                        "Failed to insert writev wind");
         }
 
 out:
@@ -443,7 +443,7 @@ ctr_setattr_cbk (call_frame_t *frame,
         if (ret) {
                 gf_msg (this->name, GF_LOG_ERROR, 0,
                         CTR_MSG_INSERT_SETATTR_UNWIND_FAILED,
-                        "Failed inserting setattr unwind");
+                        "Failed to insert setattr unwind");
         }
 
 out:
@@ -465,6 +465,7 @@ ctr_setattr (call_frame_t *frame,
 
         CTR_IS_DISABLED_THEN_GOTO(this, out);
         CTR_IF_INTERNAL_FOP_THEN_GOTO (frame, xdata, out);
+        CTR_RECORD_METADATA_HEAT_IS_DISABLED_THEN_GOTO (this, out);
 
          /*Fill ctr inode context*/
         FILL_CTR_INODE_CONTEXT(_inode_cx, loc->inode->ia_type,
@@ -476,7 +477,7 @@ ctr_setattr (call_frame_t *frame,
         if (ret) {
                 gf_msg (this->name, GF_LOG_ERROR, 0,
                         CTR_MSG_INSERT_SETATTR_WIND_FAILED,
-                        "Failed inserting setattr wind");
+                        "Failed to insert setattr wind");
         }
 out:
 
@@ -487,6 +488,65 @@ out:
         return 0;
 }
 
+/*************************** fsetattr ***************************************/
+int32_t
+ctr_fsetattr_cbk (call_frame_t *frame,
+                        void *cookie, xlator_t *this, int32_t op_ret,
+                        int32_t op_errno, struct iatt *preop_stbuf,
+                        struct iatt *postop_stbuf, dict_t *xdata)
+{
+        int ret = -1;
+
+        CTR_IS_DISABLED_THEN_GOTO(this, out);
+
+        ret = ctr_insert_unwind(frame, this,
+                        GFDB_FOP_INODE_WRITE, GFDB_FOP_UNWIND);
+        if (ret) {
+                gf_msg (this->name, GF_LOG_ERROR, 0,
+                        CTR_MSG_INSERT_SETATTR_UNWIND_FAILED,
+                        "Failed to insert fsetattr unwind");
+        }
+
+out:
+        STACK_UNWIND_STRICT (fsetattr, frame, op_ret, op_errno,
+                                preop_stbuf, postop_stbuf, xdata);
+
+        return 0;
+}
+
+
+int32_t
+ctr_fsetattr (call_frame_t *frame,
+                    xlator_t *this, fd_t *fd,
+                    struct iatt *stbuf, int32_t valid, dict_t *xdata)
+{
+        int ret = -1;
+        gf_ctr_inode_context_t ctr_inode_cx;
+        gf_ctr_inode_context_t *_inode_cx = &ctr_inode_cx;
+
+        CTR_IS_DISABLED_THEN_GOTO(this, out);
+        CTR_IF_INTERNAL_FOP_THEN_GOTO (frame, xdata, out);
+        CTR_RECORD_METADATA_HEAT_IS_DISABLED_THEN_GOTO (this, out);
+
+         /*Fill ctr inode context*/
+        FILL_CTR_INODE_CONTEXT(_inode_cx, fd->inode->ia_type,
+                fd->inode->gfid, NULL, NULL, GFDB_FOP_INODE_WRITE,
+                GFDB_FOP_WIND);
+
+        /*record into the database*/
+        ret = ctr_insert_wind(frame, this, _inode_cx);
+        if (ret) {
+                gf_msg (this->name, GF_LOG_ERROR, 0,
+                        CTR_MSG_INSERT_SETATTR_WIND_FAILED,
+                        "Failed to insert fsetattr wind");
+        }
+out:
+        STACK_WIND (frame, ctr_fsetattr_cbk,
+                    FIRST_CHILD (this), FIRST_CHILD (this)->fops->fsetattr,
+                    fd, stbuf, valid, xdata);
+
+        return 0;
+}
 /****************************fremovexattr************************************/
 
 int32_t
@@ -503,7 +563,7 @@ ctr_fremovexattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         if (ret) {
                 gf_msg (this->name, GF_LOG_ERROR, 0,
                         CTR_MSG_INSERT_FREMOVEXATTR_UNWIND_FAILED,
-                        "Failed inserting fremovexattr unwind");
+                        "Failed to insert fremovexattr unwind");
         }
 
 out:
@@ -522,7 +582,7 @@ ctr_fremovexattr (call_frame_t *frame, xlator_t *this, fd_t *fd,
 
         CTR_IS_DISABLED_THEN_GOTO(this, out);
         CTR_IF_INTERNAL_FOP_THEN_GOTO (frame, xdata, out);
-
+        CTR_RECORD_METADATA_HEAT_IS_DISABLED_THEN_GOTO (this, out);
 
          /*Fill ctr inode context*/
         FILL_CTR_INODE_CONTEXT(_inode_cx, fd->inode->ia_type,
@@ -534,7 +594,7 @@ ctr_fremovexattr (call_frame_t *frame, xlator_t *this, fd_t *fd,
         if (ret) {
                 gf_msg (this->name, GF_LOG_ERROR, 0,
                         CTR_MSG_INSERT_FREMOVEXATTR_WIND_FAILED,
-                        "Failed inserting fremovexattr wind");
+                        "Failed to insert fremovexattr wind");
         }
 
 out:
@@ -561,7 +621,7 @@ ctr_removexattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         if (ret) {
                 gf_msg (this->name, GF_LOG_ERROR, 0,
                         CTR_MSG_INSERT_REMOVEXATTR_UNWIND_FAILED,
-                        "Failed inserting removexattr unwind");
+                        "Failed to insert removexattr unwind");
         }
 
 out:
@@ -580,7 +640,7 @@ ctr_removexattr (call_frame_t *frame, xlator_t *this, loc_t *loc,
 
         CTR_IS_DISABLED_THEN_GOTO(this, out);
         CTR_IF_INTERNAL_FOP_THEN_GOTO (frame, xdata, out);
-
+        CTR_RECORD_METADATA_HEAT_IS_DISABLED_THEN_GOTO (this, out);
 
          /*Fill ctr inode context*/
         FILL_CTR_INODE_CONTEXT(_inode_cx, loc->inode->ia_type,
@@ -592,7 +652,7 @@ ctr_removexattr (call_frame_t *frame, xlator_t *this, loc_t *loc,
         if (ret) {
                 gf_msg (this->name, GF_LOG_ERROR, 0,
                         CTR_MSG_INSERT_REMOVEXATTR_WIND_FAILED,
-                        "Failed inserting removexattr wind");
+                        "Failed to insert removexattr wind");
         }
 
 out:
@@ -619,7 +679,7 @@ ctr_truncate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         if (ret) {
                 gf_msg (this->name, GF_LOG_ERROR, 0,
                         CTR_MSG_INSERT_TRUNCATE_UNWIND_FAILED,
-                        "Failed inserting truncate unwind");
+                        "Failed to insert truncate unwind");
         }
 
 
@@ -651,7 +711,7 @@ ctr_truncate (call_frame_t *frame, xlator_t *this, loc_t *loc,
         if (ret) {
                 gf_msg (this->name, GF_LOG_ERROR, 0,
                         CTR_MSG_INSERT_TRUNCATE_WIND_FAILED,
-                        "Failed inserting truncate wind");
+                        "Failed to insert truncate wind");
         }
 out:
         STACK_WIND (frame, ctr_truncate_cbk, FIRST_CHILD (this),
@@ -676,7 +736,7 @@ ctr_ftruncate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         if (ret) {
                 gf_msg (this->name, GF_LOG_ERROR, 0,
                         CTR_MSG_INSERT_FTRUNCATE_UNWIND_FAILED,
-                        "Failed inserting ftruncate unwind");
+                        "Failed to insert ftruncate unwind");
         }
 
 out:
@@ -707,7 +767,7 @@ ctr_ftruncate (call_frame_t *frame, xlator_t *this, fd_t *fd,
         if (ret) {
                 gf_msg (this->name, GF_LOG_ERROR, 0,
                         CTR_MSG_INSERT_FTRUNCATE_WIND_FAILED,
-                        "Failed inserting ftruncate wind");
+                        "Failed to insert ftruncate wind");
         }
 
 out:
@@ -735,7 +795,7 @@ ctr_rename_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         if (ret) {
                 gf_msg (this->name, GF_LOG_ERROR, 0,
                         CTR_MSG_INSERT_RENAME_UNWIND_FAILED,
-                        "Failed inserting rename unwind");
+                        "Failed to insert rename unwind");
         }
 
 out:
@@ -774,12 +834,15 @@ ctr_rename (call_frame_t *frame, xlator_t *this, loc_t *oldloc,
                 oldloc->inode->gfid, _nlink_cx, _olink_cx,
                 GFDB_FOP_DENTRY_WRITE, GFDB_FOP_WIND);
 
+        /* Is a metatdata fop */
+        _inode_cx->is_metadata_fop = _gf_true;
+
         /*record into the database*/
         ret = ctr_insert_wind(frame, this, _inode_cx);
         if (ret) {
                 gf_msg (this->name, GF_LOG_ERROR, 0,
                         CTR_MSG_INSERT_RENAME_WIND_FAILED,
-                        "Failed inserting rename wind");
+                        "Failed to insert 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 */
@@ -834,7 +897,7 @@ ctr_unlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 if (ret) {
                         gf_msg (this->name, GF_LOG_ERROR, 0,
                                 CTR_MSG_INSERT_UNLINK_UNWIND_FAILED,
-                                "Failed inserting unlink unwind");
+                                "Failed to insert unlink unwind");
                 }
         }
         /*Last link that was deleted*/
@@ -845,7 +908,7 @@ ctr_unlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 if (ret) {
                         gf_msg (this->name, GF_LOG_ERROR, 0,
                                 CTR_MSG_INSERT_UNLINK_UNWIND_FAILED,
-                                "Failed inserting unlink unwind");
+                                "Failed to insert unlink unwind");
                 }
         }
 
@@ -883,6 +946,9 @@ ctr_unlink (call_frame_t *frame, xlator_t *this,
         /*Internal FOP*/
         _inode_cx->is_internal_fop = is_internal_fop (frame, xdata);
 
+        /* Is a metadata FOP */
+        _inode_cx->is_metadata_fop = _gf_true;
+
         /* If its a internal FOP and dht link file donot record*/
         if (_inode_cx->is_internal_fop &&
                         dht_is_linkfile (&dummy_stat, xdata)) {
@@ -894,7 +960,7 @@ ctr_unlink (call_frame_t *frame, xlator_t *this,
         if (ret) {
                 gf_msg (this->name, GF_LOG_ERROR, 0,
                         CTR_MSG_INSERT_UNLINK_UNWIND_FAILED,
-                        "Failed inserting unlink wind");
+                        "Failed to insert 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 */
@@ -961,7 +1027,7 @@ ctr_fsync_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         if (ret) {
                 gf_msg (this->name, GF_LOG_ERROR, 0,
                         CTR_MSG_INSERT_FSYNC_UNWIND_FAILED,
-                        "Failed inserting fsync unwind");
+                        "Failed to insert fsync unwind");
         }
 
 out:
@@ -992,7 +1058,7 @@ ctr_fsync (call_frame_t *frame, xlator_t *this, fd_t *fd,
         if (ret) {
                 gf_msg (this->name, GF_LOG_ERROR, 0,
                         CTR_MSG_INSERT_FSYNC_WIND_FAILED,
-                        "Failed inserting fsync wind");
+                        "Failed to insert fsync wind");
         }
 
 out:
@@ -1017,10 +1083,9 @@ ctr_setxattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         if (ret) {
                 gf_msg (this->name, GF_LOG_ERROR, 0,
                         CTR_MSG_INSERT_FSYNC_UNWIND_FAILED,
-                        "Failed inserting fsync unwind");
+                        "Failed to insert setxattr unwind");
         }
 
-
 out:
         STACK_UNWIND_STRICT (setxattr, frame, op_ret, op_errno, xdata);
 
@@ -1037,6 +1102,7 @@ ctr_setxattr (call_frame_t *frame, xlator_t *this,
 
         CTR_IS_DISABLED_THEN_GOTO(this, out);
         CTR_IF_INTERNAL_FOP_THEN_GOTO (frame, xdata, out);
+        CTR_RECORD_METADATA_HEAT_IS_DISABLED_THEN_GOTO (this, out);
 
          /*Fill ctr inode context*/
         FILL_CTR_INODE_CONTEXT(_inode_cx, loc->inode->ia_type,
@@ -1048,7 +1114,7 @@ ctr_setxattr (call_frame_t *frame, xlator_t *this,
         if (ret) {
                 gf_msg (this->name, GF_LOG_ERROR, 0,
                         CTR_MSG_INSERT_SETATTR_WIND_FAILED,
-                        "Failed inserting setxattr wind");
+                        "Failed to insert setxattr wind");
         }
 
 out:
@@ -1057,7 +1123,61 @@ out:
                     loc, xattr, flags, xdata);
         return 0;
 }
+/**************************** fsetxattr *************************************/
+int32_t
+ctr_fsetxattr_cbk (call_frame_t *frame,
+                         void *cookie, xlator_t *this, int32_t op_ret,
+                         int32_t op_errno, dict_t *xdata)
+{
+        int ret = -1;
+
+        CTR_IS_DISABLED_THEN_GOTO(this, out);
+
+        ret = ctr_insert_unwind(frame, this, GFDB_FOP_INODE_WRITE,
+                                GFDB_FOP_UNWIND);
+        if (ret) {
+                gf_msg (this->name, GF_LOG_ERROR, 0,
+                        CTR_MSG_INSERT_FSYNC_UNWIND_FAILED,
+                        "Failed to insert fsetxattr unwind");
+        }
+
+out:
+        STACK_UNWIND_STRICT (fsetxattr, frame, op_ret, op_errno, xdata);
+
+        return 0;
+}
+
+int32_t
+ctr_fsetxattr (call_frame_t *frame, xlator_t *this, fd_t *fd, dict_t *dict,
+               int32_t flags, dict_t *xdata)
+{
+        int ret = -1;
+        gf_ctr_inode_context_t ctr_inode_cx;
+        gf_ctr_inode_context_t *_inode_cx = &ctr_inode_cx;
+
+        CTR_IS_DISABLED_THEN_GOTO(this, out);
+        CTR_IF_INTERNAL_FOP_THEN_GOTO (frame, xdata, out);
+        CTR_RECORD_METADATA_HEAT_IS_DISABLED_THEN_GOTO (this, out);
+
+         /*Fill ctr inode context*/
+        FILL_CTR_INODE_CONTEXT(_inode_cx, fd->inode->ia_type,
+                fd->inode->gfid, NULL, NULL,
+                GFDB_FOP_INODE_WRITE, GFDB_FOP_WIND);
+
+        /*record into the database*/
+        ret = ctr_insert_wind(frame, this, _inode_cx);
+        if (ret) {
+                gf_msg (this->name, GF_LOG_ERROR, 0,
+                        CTR_MSG_INSERT_SETATTR_WIND_FAILED,
+                        "Failed to insert fsetxattr wind");
+        }
 
+out:
+        STACK_WIND (frame, ctr_fsetxattr_cbk,
+                    FIRST_CHILD (this), FIRST_CHILD (this)->fops->fsetxattr,
+                    fd, dict, flags, xdata);
+        return 0;
+}
 /****************************mknod*******************************************/
 
 
@@ -1083,7 +1203,7 @@ ctr_mknod_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         if (ret) {
                 gf_msg (this->name, GF_LOG_ERROR, 0,
                         CTR_MSG_INSERT_MKNOD_UNWIND_FAILED,
-                        "Failed inserting mknod unwind");
+                        "Failed to insert mknod unwind");
         }
 
 out:
@@ -1134,7 +1254,7 @@ ctr_mknod (call_frame_t *frame, xlator_t *this,
         if (ret) {
                 gf_msg (this->name, GF_LOG_ERROR, 0,
                         CTR_MSG_INSERT_MKNOD_WIND_FAILED,
-                        "Failed inserting mknod wind");
+                        "Failed to insert mknod wind");
         }
 
 out:
@@ -1169,7 +1289,7 @@ ctr_create_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         if (ret) {
                 gf_msg (this->name, GF_LOG_ERROR, 0,
                         CTR_MSG_INSERT_CREATE_UNWIND_FAILED,
-                        "Failed inserting create unwind");
+                        "Failed to insert create unwind");
         }
 
 out:
@@ -1232,7 +1352,7 @@ ctr_create (call_frame_t *frame, xlator_t *this,
         if (ret) {
                 gf_msg (this->name, GF_LOG_ERROR, 0,
                         CTR_MSG_INSERT_CREATE_WIND_FAILED,
-                        "Failed inserting create wind");
+                        "Failed to insert create wind");
         }
 out:
         STACK_WIND (frame, ctr_create_cbk, FIRST_CHILD (this),
@@ -1264,7 +1384,7 @@ ctr_link_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         if (ret) {
                 gf_msg (this->name, GF_LOG_ERROR, 0,
                         CTR_MSG_INSERT_CREATE_UNWIND_FAILED,
-                        "Failed inserting create unwind");
+                        "Failed to insert create unwind");
         }
 
 out:
@@ -1301,6 +1421,9 @@ ctr_link (call_frame_t *frame, xlator_t *this,
         /*Internal FOP*/
         _inode_cx->is_internal_fop = is_internal_fop (frame, xdata);
 
+        /* Is a metadata fop */
+        _inode_cx->is_metadata_fop = _gf_true;
+
         /* If its a internal FOP and dht link file donot record*/
         if (_inode_cx->is_internal_fop &&
                         dht_is_linkfile (&dummy_stat, xdata)) {
@@ -1313,7 +1436,7 @@ ctr_link (call_frame_t *frame, xlator_t *this,
         if (ret) {
                 gf_msg (this->name, GF_LOG_ERROR, 0,
                         CTR_MSG_INSERT_LINK_WIND_FAILED,
-                        "Failed inserting link wind");
+                        "Failed to insert link wind");
         }
 
 out:
@@ -1338,7 +1461,7 @@ int ctr_readv_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         if (ret) {
                 gf_msg (this->name, GF_LOG_ERROR, 0,
                         CTR_MSG_INSERT_CREATE_UNWIND_FAILED,
-                        "Failed inserting create unwind");
+                        "Failed to insert create unwind");
         }
 
 out:
@@ -1369,7 +1492,7 @@ ctr_readv (call_frame_t *frame, xlator_t *this,
         if (ret) {
                 gf_msg (this->name, GF_LOG_ERROR, 0,
                         CTR_MSG_INSERT_READV_WIND_FAILED,
-                        "Failed inserting readv wind");
+                        "Failed to insert readv wind");
         }
 
 out:
@@ -1692,6 +1815,10 @@ reconfigure (xlator_t *this, dict_t *options)
         GF_OPTION_RECONF ("record-counters", priv->ctr_record_counter, options,
                           bool, out);
 
+        GF_OPTION_RECONF ("ctr-record-metadata-heat",
+                          priv->ctr_record_metadata_heat, options,
+                          bool, out);
+
         GF_OPTION_RECONF ("ctr_link_consistency", priv->ctr_link_consistency,
                         options, bool, out);
 
@@ -1874,23 +2001,26 @@ fini (xlator_t *this)
 
 struct xlator_fops fops = {
         /*lookup*/
-        .lookup      = ctr_lookup,
+        .lookup         = ctr_lookup,
         /*write fops */
-        .mknod       = ctr_mknod,
-        .create      = ctr_create,
-        .truncate    = ctr_truncate,
-        .ftruncate   = ctr_ftruncate,
-        .setxattr    = ctr_setxattr,
-        .removexattr = ctr_removexattr,
-        .unlink      = ctr_unlink,
-        .link        = ctr_link,
-        .rename      = ctr_rename,
-        .writev      = ctr_writev,
-        .setattr      = ctr_setattr,
+        .mknod          = ctr_mknod,
+        .create         = ctr_create,
+        .truncate       = ctr_truncate,
+        .ftruncate      = ctr_ftruncate,
+        .setxattr       = ctr_setxattr,
+        .fsetxattr      = ctr_fsetxattr,
+        .removexattr    = ctr_removexattr,
+        .fremovexattr   = ctr_fremovexattr,
+        .unlink         = ctr_unlink,
+        .link           = ctr_link,
+        .rename         = ctr_rename,
+        .writev         = ctr_writev,
+        .setattr        = ctr_setattr,
+        .fsetattr       = ctr_fsetattr,
         /*read fops*/
-        .readv       = ctr_readv,
+        .readv          = ctr_readv,
         /* IPC call*/
-        .ipc          = ctr_ipc
+        .ipc            = ctr_ipc
 };
 
 struct xlator_cbks cbks = {
@@ -1919,6 +2049,11 @@ struct volume_options options[] = {
           .value = {"on", "off"},
           .default_value = "off"
         },
+        { .key  = {"ctr-record-metadata-heat"},
+          .type = GF_OPTION_TYPE_BOOL,
+          .value = {"on", "off"},
+          .default_value = "off"
+        },
         { .key  = {"ctr_link_consistency"},
           .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 047abd9f032..ab918eac825 100644
--- a/xlators/features/changetimerecorder/src/ctr-helper.c
+++ b/xlators/features/changetimerecorder/src/ctr-helper.c
@@ -272,6 +272,11 @@ int extract_ctr_options (xlator_t *this, gf_ctr_private_t *_priv) {
         GF_OPTION_INIT ("record-counters", _priv->ctr_record_counter, bool,
                         out);
 
+        /* Extract flag for record metadata heat */
+        GF_OPTION_INIT ("ctr-record-metadata-heat",
+                        _priv->ctr_record_metadata_heat, bool,
+                        out);
+
         /*Extract flag for link consistency*/
         GF_OPTION_INIT ("ctr_link_consistency", _priv->ctr_link_consistency,
                         bool, out);
diff --git a/xlators/features/changetimerecorder/src/ctr-helper.h b/xlators/features/changetimerecorder/src/ctr-helper.h
index bee143293b3..7e51be074e9 100644
--- a/xlators/features/changetimerecorder/src/ctr-helper.h
+++ b/xlators/features/changetimerecorder/src/ctr-helper.h
@@ -50,6 +50,7 @@ typedef struct gf_ctr_private {
         gf_boolean_t                    ctr_record_wind;
         gf_boolean_t                    ctr_record_unwind;
         gf_boolean_t                    ctr_record_counter;
+        gf_boolean_t                    ctr_record_metadata_heat;
         gf_boolean_t                    ctr_link_consistency;
         gfdb_db_type_t                  gfdb_db_type;
         gfdb_sync_type_t                gfdb_sync_type;
@@ -83,7 +84,7 @@ typedef struct gf_ctr_local {
         gfdb_db_record_t        gfdb_db_record;
         ia_type_t               ia_inode_type;
         gf_boolean_t            is_internal_fop;
-        gf_client_pid_t          client_pid;
+        gf_client_pid_t         client_pid;
 } gf_ctr_local_t;
 /*
  * Easy access of gfdb_db_record of ctr_local
@@ -176,6 +177,8 @@ typedef struct gf_ctr_inode_context {
         gfdb_fop_type_t         fop_type;
         gfdb_fop_path_t         fop_path;
         gf_boolean_t            is_internal_fop;
+        /* Indicating metadata fops */
+        gf_boolean_t            is_metadata_fop;
 } gf_ctr_inode_context_t;
 
 
@@ -343,6 +346,19 @@ do {\
                 goto label;\
  } while (0)
 
+/*
+ * IS CTR record metadata heat is disabled then goto to label
+ * */
+ #define CTR_RECORD_METADATA_HEAT_IS_DISABLED_THEN_GOTO(this, label)\
+ do {\
+        gf_ctr_private_t *_priv = NULL;\
+        GF_ASSERT (this);\
+        GF_ASSERT (this->private);\
+        _priv = this->private;\
+        if (!_priv->ctr_record_metadata_heat)\
+                goto label;\
+ } while (0)
+
 int
 fill_db_record_for_unwind (xlator_t              *this,
                           gf_ctr_local_t        *ctr_local,
@@ -395,16 +411,41 @@ ctr_insert_wind (call_frame_t                    *frame,
                 ctr_local->is_internal_fop = ctr_inode_cx->is_internal_fop;
 
                 /* Decide whether to record counters or not */
-                CTR_DB_REC(ctr_local).do_record_counters =
-                                                _priv->ctr_record_counter &&
-                                                !(ctr_local->is_internal_fop);
+                CTR_DB_REC(ctr_local).do_record_counters = _gf_false;
+                /* If record counter is enabled */
+                if (_priv->ctr_record_counter) {
+                        /* If not a internal fop */
+                        if (!(ctr_local->is_internal_fop)) {
+                                /* If its a metadata fop AND
+                                 * record metadata heat
+                                 * OR
+                                 * its NOT a metadata fop */
+                                if ((ctr_inode_cx->is_metadata_fop
+                                        && _priv->ctr_record_metadata_heat)
+                                        ||
+                                        (!ctr_inode_cx->is_metadata_fop)) {
+                                        CTR_DB_REC(ctr_local).do_record_counters
+                                                = _gf_true;
+                                }
+                        }
+                }
 
                 /* Decide whether to record times or not
                  * For non internal FOPS record times as usual*/
+                CTR_DB_REC(ctr_local).do_record_times = _gf_false;
                 if (!ctr_local->is_internal_fop) {
-                        CTR_DB_REC(ctr_local).do_record_times =
-                                                (_priv->ctr_record_wind
-                                                || _priv->ctr_record_unwind);
+                        /* If its a metadata fop AND
+                        * record metadata heat
+                        * OR
+                        * its NOT a metadata fop */
+                        if ((ctr_inode_cx->is_metadata_fop &&
+                                _priv->ctr_record_metadata_heat)
+                                ||
+                                (!ctr_inode_cx->is_metadata_fop)) {
+                                CTR_DB_REC(ctr_local).do_record_times =
+                                        (_priv->ctr_record_wind
+                                        || _priv->ctr_record_unwind);
+                        }
                 }
                 /* when its a internal FOPS*/
                 else {
diff --git a/xlators/mgmt/glusterd/src/glusterd-volume-set.c b/xlators/mgmt/glusterd/src/glusterd-volume-set.c
index b942d9df944..e17a43dc835 100644
--- a/xlators/mgmt/glusterd/src/glusterd-volume-set.c
+++ b/xlators/mgmt/glusterd/src/glusterd-volume-set.c
@@ -2071,6 +2071,22 @@ struct volopt_map_entry glusterd_volopt_map[] = {
                          "of writes (or reads) to a given file are needed "
                          "before triggering migration."
         },
+        { .key         = "features.ctr-record-metadata-heat",
+          .voltype     = "features/changetimerecorder",
+          .value       = "off",
+          .option      = "ctr-record-metadata-heat",
+          .op_version  = GD_OP_VERSION_3_7_0,
+          /* Purposefully commenting the description so that this option remains
+           * hidden from the users as this is more of a developer option as of
+           * now.
+           * .description = "Its a Change Time Recorder Xlator option to "
+           *             "enable recording write heat on metadata of the file. "
+                         "The default is disabled. "
+                         "Metadata is inode atttributes like atime, mtime,"
+                         " permissions etc and "
+                         "extended attributes of a file ."
+           * */
+        },
         { .key         = "features.ctr_link_consistency",
           .voltype     = "features/changetimerecorder",
           .value       = "off",