WORM/Retention Translator: Implementation of file level WORM

To activate the file level worm feature, the features.read-only and
features.worm options should be switched "off" on the volume and
the features.worm-file-level should be switched "on". Both read-only
and worm or worm-file-level cannot be switched "on" together. The
files which are created when the worm-file-level option is set on the
volume will have their own retention profile.

If both worm and worm-file-level are "on" at that time the worm
which is the volume level worm will have priority over file level
worm. If worm-file level is switched "off" after some time and the
read-only option is switched "on" then read-only will have priority.

The current implementation allows the users to manually transmit
a file to a WORM-Retained state by removing all the write bits of
the file using the chmod command. The file will have a retention
profile which contains the state of the file, mode of retention,
and the default retention time.

The file will be made WORM-Retained for a default of 120 seconds
during which it will be immutable and undeletable and it sets the
atime of the file to the time till which it is retained.
After that period if any fop request comes for that file, will
make the transition from WORM-Retained state to WORM state, where
the file will be immutable but deletable and, it will reset
the atime to the actual atime of the file. If a WORM file needs
to be made undeletable again, it can be done by using the chmod
command with all the write bits removed.

There are two modes of retention:
1. Relax: where the retention time of a WORM-Retained file can be
   increased or decreased.
2. Enterprise: where the retention time of a WORM-Retained file
   can be increased but not be decreased.
Whenever a utime change(touch -a, -t, ...)request comes for a
file it checks the mode of retention before setting the utimes.
This is done only if the file is WORM-Retained but for a WORM file
it will change the utimes.

Lazy auto commit:
Whenever a file gets created it will store the creation time of the
file or if a file already exists then any of the next unlink, link,
truncate or rename fops will set the current time as the start time
in an xattr. The next rename/unlink/truncate/link call will check for the
auto commit period and if is is expired, then it will automatically do
the state transition. If it is a normal file then it gets converted
to WORM-Retained state. If it is a WORM-Retained file and its retention
period is expired, then it gets converted to WORM state.

Added the volume set options for the WORM translator. It allows the users
to change the default values of auto-commit-period, default-retention-period,
retention-mode. To make use of the file-level WORM first we have to set the
'worm-file' option to 'on'. The files which are created when the worm-file
option is set on the volume will get WORM-Retained. Other files will work
as usual and will not be WORMed. The auto-commit-period, retention-mode,
and the default-retention-period values for the file will be set to the values
which are set on the volume when the file is created.

Added the tests to check the basic functionalities of the WORM/Retention feature.

Change-Id: I77bd9777f9395a944d76b5cc35a5b48a3c14d148
BUG: 1326308
Signed-off-by: karthik-us <ksubrahm@redhat.com>
Reviewed-on: http://review.gluster.org/13429
Reviewed-by: Niels de Vos <ndevos@redhat.com>
Smoke: Gluster Build System <jenkins@build.gluster.com>
NetBSD-regression: NetBSD Build System <jenkins@build.gluster.org>
CentOS-regression: Gluster Build System <jenkins@build.gluster.com>

7 files changed, 1044 insertions, 41 deletions

diff --git a/xlators/features/read-only/src/Makefile.am b/xlators/features/read-only/src/Makefile.am
index 5aad7079344..3edac3f8a1d 100644
--- a/xlators/features/read-only/src/Makefile.am
+++ b/xlators/features/read-only/src/Makefile.am
@@ -2,7 +2,7 @@ xlator_LTLIBRARIES = read-only.la worm.la
 
 xlatordir = $(libdir)/glusterfs/$(PACKAGE_VERSION)/xlator/features
 
-noinst_HEADERS = read-only.h read-only-mem-types.h read-only-common.h
+noinst_HEADERS = read-only.h read-only-mem-types.h read-only-common.h worm-helper.h
 
 read_only_la_LDFLAGS = -module $(GF_XLATOR_DEFAULT_LDFLAGS)
 
@@ -11,7 +11,7 @@ read_only_la_LIBADD = $(top_builddir)/libglusterfs/src/libglusterfs.la
 
 worm_la_LDFLAGS = -module $(GF_XLATOR_DEFAULT_LDFLAGS)
 
-worm_la_SOURCES = read-only-common.c worm.c
+worm_la_SOURCES = read-only-common.c worm-helper.c worm.c
 worm_la_LIBADD = $(top_builddir)/libglusterfs/src/libglusterfs.la
 
 AM_CPPFLAGS = $(GF_CPPFLAGS) -I$(top_srcdir)/libglusterfs/src
diff --git a/xlators/features/read-only/src/read-only.h b/xlators/features/read-only/src/read-only.h
index 8e7e1b68081..3178bb26715 100644
--- a/xlators/features/read-only/src/read-only.h
+++ b/xlators/features/read-only/src/read-only.h
@@ -15,7 +15,22 @@
 #include "xlator.h"
 
 typedef struct {
-        gf_boolean_t      readonly_or_worm_enabled;
+        uint8_t worm : 1;
+        uint8_t retain : 1;
+        uint8_t legal_hold :1;
+        uint8_t ret_mode : 1;
+        uint64_t ret_period;
+        uint64_t auto_commit_period;
+} worm_reten_state_t;
+
+
+typedef struct {
+        gf_boolean_t          readonly_or_worm_enabled;
+        gf_boolean_t          worm_file;
+        uint64_t              reten_period;
+        uint64_t              com_period;
+        char                  *reten_mode;
+        time_t                start_time;
 } read_only_priv_t;
 
 #endif
diff --git a/xlators/features/read-only/src/worm-helper.c b/xlators/features/read-only/src/worm-helper.c
new file mode 100644
index 00000000000..b5b1c628d73
--- /dev/null
+++ b/xlators/features/read-only/src/worm-helper.c
@@ -0,0 +1,422 @@
+/*
+   Copyright (c) 2016 Red Hat, Inc. <http://www.redhat.com>
+   This file is part of GlusterFS.
+
+   This file is licensed to you under your choice of the GNU Lesser
+   General Public License, version 3 or any later version (LGPLv3 or
+   later), or the GNU General Public License, version 2 (GPLv2), in all
+   cases as published by the Free Software Foundation.
+*/
+#include "read-only-mem-types.h"
+#include "read-only.h"
+#include "xlator.h"
+#include "syncop.h"
+#include "worm-helper.h"
+
+/*Function to check whether file is read-only.
+ * The input *stbuf contains the attributes of the file, which is used to check
+ * the write protection bits for all the users of the file.
+ * Return true if all the write bits are disabled,false otherwise*/
+gf_boolean_t
+is_write_disabled (struct iatt *stbuf)
+{
+        gf_boolean_t ret        =       _gf_false;
+
+        GF_VALIDATE_OR_GOTO ("worm", stbuf, out);
+
+        if (stbuf->ia_prot.owner.write == 0 &&
+            stbuf->ia_prot.group.write == 0 &&
+            stbuf->ia_prot.other.write == 0)
+                ret = _gf_true;
+out:
+        return ret;
+}
+
+
+int32_t
+worm_init_state (xlator_t *this, gf_boolean_t fop_with_fd, void *file_ptr)
+{
+        int ret                 =      -1;
+        uint64_t start_time     =       0;
+        dict_t *dict            =       NULL;
+
+        GF_VALIDATE_OR_GOTO ("worm", this, out);
+        GF_VALIDATE_OR_GOTO (this->name, file_ptr, out);
+
+        start_time = time (NULL);
+        dict = dict_new ();
+        if (!dict) {
+                gf_log (this->name, GF_LOG_ERROR, "Error creating the dict");
+                goto out;
+        }
+        ret = dict_set_uint64 (dict, "trusted.start_time", start_time);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Error in setting the dict");
+                goto out;
+        }
+        if (fop_with_fd)
+                ret = syncop_fsetxattr (this, (fd_t *)file_ptr, dict, 0,
+                                        NULL, NULL);
+        else
+                ret = syncop_setxattr (this, (loc_t *)file_ptr, dict, 0, NULL,
+                                       NULL);
+out:
+        if (dict)
+                dict_destroy (dict);
+        return ret;
+}
+
+
+/*Function to set the retention state for a file.
+ * It loads the WORM/Retention state into the retention_state pointer.*/
+int32_t
+worm_set_state (xlator_t *this, gf_boolean_t fop_with_fd, void *file_ptr,
+                worm_reten_state_t *retention_state, struct iatt *stbuf)
+{
+        read_only_priv_t *priv   =      NULL;
+        struct iatt stpre        =      {0,};
+        int ret                  =      -1;
+
+        GF_VALIDATE_OR_GOTO ("worm", this, out);
+        GF_VALIDATE_OR_GOTO (this->name, file_ptr, out);
+        GF_VALIDATE_OR_GOTO (this->name, retention_state, out);
+        GF_VALIDATE_OR_GOTO (this->name, stbuf, out);
+
+        priv = this->private;
+        GF_ASSERT (priv);
+        retention_state->worm = 1;
+        retention_state->retain = 1;
+        retention_state->legal_hold = 0;
+        if (strcmp (priv->reten_mode, "relax") == 0)
+                retention_state->ret_mode = 0;
+        else
+                retention_state->ret_mode = 1;
+        retention_state->ret_period = priv->reten_period;
+        retention_state->auto_commit_period = priv->com_period;
+        if (fop_with_fd)
+                ret = syncop_fstat (this, (fd_t *)file_ptr, &stpre, NULL, NULL);
+        else
+                ret = syncop_stat (this, (loc_t *)file_ptr, &stpre, NULL, NULL);
+        if (ret)
+                goto out;
+        stbuf->ia_mtime = stpre.ia_mtime;
+        stbuf->ia_atime = time (NULL) + retention_state->ret_period;
+
+        if (fop_with_fd)
+                ret = syncop_fsetattr (this, (fd_t *)file_ptr, stbuf,
+                                       GF_SET_ATTR_ATIME, NULL, NULL,
+                                       NULL, NULL);
+        else
+                ret = syncop_setattr (this, (loc_t *)file_ptr, stbuf,
+                                      GF_SET_ATTR_ATIME, NULL, NULL,
+                                      NULL, NULL);
+        if (ret)
+                goto out;
+
+        ret = set_xattr (this, retention_state, fop_with_fd, file_ptr);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Error setting xattr");
+                goto out;
+        }
+        ret = 0;
+out:
+        return ret;
+}
+
+
+/*This function gets the state of the WORM/Retention xattr and loads it in the
+ * dict pointer.*/
+int32_t
+worm_get_state (xlator_t *this, gf_boolean_t fop_with_fd, void *file_ptr,
+                worm_reten_state_t *reten_state)
+{
+        dict_t *dict    =       NULL;
+        char *val       =       NULL;
+        int ret         =       -1;
+
+        GF_VALIDATE_OR_GOTO ("worm", this, out);
+        GF_VALIDATE_OR_GOTO (this->name, file_ptr, out);
+        GF_VALIDATE_OR_GOTO (this->name, reten_state, out);
+
+        if (fop_with_fd)
+                ret = syncop_fgetxattr (this, (fd_t *)file_ptr, &dict,
+                                        "trusted.reten_state", NULL, NULL);
+        else
+                ret = syncop_getxattr (this, (loc_t *)file_ptr, &dict,
+                                       "trusted.reten_state", NULL, NULL);
+        if (ret < 0 || !dict) {
+                ret = -1;
+                goto out;
+        }
+        ret = dict_get_str (dict, "trusted.reten_state", &val);
+        if (ret) {
+                ret = -2;
+                gf_log (this->name, GF_LOG_ERROR, "Empty val");
+        }
+        deserialize_state (val, reten_state);
+out:
+        if (dict)
+                dict_unref (dict);
+        return ret;
+}
+
+
+/*Function to lookup the current state of the WORM/Retention profile.
+ * Based on the retain value and the access time of the file, the transition
+ * from WORM/Retention to WORM is made.*/
+void
+state_lookup (xlator_t *this, gf_boolean_t fop_with_fd, void *file_ptr,
+              worm_reten_state_t *reten_state)
+{
+        int ret                                 =       -1;
+        struct iatt stbuf                       =       {0,};
+
+        GF_VALIDATE_OR_GOTO ("worm", this, out);
+        GF_VALIDATE_OR_GOTO (this->name, file_ptr, out);
+        GF_VALIDATE_OR_GOTO (this->name, reten_state, out);
+
+        if (fop_with_fd)
+                ret = syncop_fstat (this, (fd_t *)file_ptr, &stbuf, NULL, NULL);
+        else
+                ret = syncop_stat (this, (loc_t *)file_ptr, &stbuf, NULL, NULL);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Stat lookup error: %s",
+                        strerror (-ret));
+                        goto out;
+        }
+        if (time (NULL) < stbuf.ia_atime)
+                goto out;
+
+        stbuf.ia_atime -= reten_state->ret_period;
+        reten_state->retain = 0;
+        reten_state->ret_period = 0;
+        reten_state->auto_commit_period = 0;
+        ret = set_xattr (this, reten_state, fop_with_fd, file_ptr);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Error setting xattr");
+                goto out;
+        }
+
+        if (fop_with_fd)
+                ret = syncop_fsetattr (this, (fd_t *)file_ptr, &stbuf,
+                                       GF_SET_ATTR_ATIME, NULL, NULL,
+                                       NULL, NULL);
+        else
+                ret = syncop_setattr (this, (loc_t *)file_ptr, &stbuf,
+                                      GF_SET_ATTR_ATIME, NULL, NULL,
+                                      NULL, NULL);
+        if (ret)
+                goto out;
+        gf_log (this->name, GF_LOG_INFO, "Retention state reset");
+out:
+        return;
+}
+
+
+/*This function serializes and stores the WORM/Retention state of a file in an
+ * uint64_t variable by setting the bits using the bitwise operations.*/
+void
+serialize_state (worm_reten_state_t *reten_state, char *val)
+{
+        uint32_t state     =       0;
+
+        GF_VALIDATE_OR_GOTO ("worm", reten_state, out);
+        GF_VALIDATE_OR_GOTO ("worm", val, out);
+
+        state |= reten_state->worm << 0;
+        state |= reten_state->retain << 1;
+        state |= reten_state->legal_hold << 2;
+        state |= reten_state->ret_mode << 3;
+        sprintf (val, "%d/%ld/%ld", state, reten_state->ret_period,
+                 reten_state->auto_commit_period);
+
+out:
+        return;
+}
+
+
+/*This function deserializes the data stored in the xattr of the file and loads
+ * the value to the reten_state structure.*/
+void deserialize_state (char *val, worm_reten_state_t *reten_state)
+{
+        char *token     =       NULL;
+        uint32_t state  =       0;
+
+        GF_VALIDATE_OR_GOTO ("worm", val, out);
+        GF_VALIDATE_OR_GOTO ("worm", reten_state, out);
+
+        token = strtok (val, "/");
+        state = atoi (token);
+        reten_state->worm = (state >> 0) & 1;
+        reten_state->retain = (state >> 1) & 1;
+        reten_state->legal_hold = (state >> 2) & 1;
+        reten_state->ret_mode = (state >> 3) & 1;
+        token = strtok (NULL, "/");
+        reten_state->ret_period = atoi (token);
+        token = strtok (NULL, "/");
+        reten_state->auto_commit_period = atoi (token);
+
+out:
+        return;
+}
+
+
+/*Function to set the xattr for a file.
+ * If the xattr is already present then it will replace that.*/
+int32_t
+set_xattr (xlator_t *this, worm_reten_state_t *reten_state,
+           gf_boolean_t fop_with_fd, void *file_ptr)
+{
+        char val[100]   =        "";
+        int ret         =        -1;
+        dict_t *dict    =        NULL;
+
+        GF_VALIDATE_OR_GOTO ("worm", this, out);
+        GF_VALIDATE_OR_GOTO (this->name, reten_state, out);
+        GF_VALIDATE_OR_GOTO (this->name, file_ptr, out);
+
+        serialize_state (reten_state, val);
+        dict = dict_new ();
+        if (!dict) {
+                gf_log (this->name, GF_LOG_ERROR, "Error creating the dict");
+                goto out;
+        }
+        ret = dict_set_str (dict, "trusted.reten_state", val);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Error in setting the dict");
+                goto out;
+        }
+        if (fop_with_fd)
+                ret = syncop_fsetxattr (this, (fd_t *)file_ptr, dict, 0,
+                                        NULL, NULL);
+        else
+                ret = syncop_setxattr (this, (loc_t *)file_ptr, dict, 0, NULL,
+                                       NULL);
+out:
+        if (dict)
+                dict_destroy (dict);
+        return ret;
+}
+
+
+/*This function checks whether a file's timeout is happend for the state
+ * transition and if yes, then it will do the transition from the current state
+ * to the appropriate state. It also decides whether to continue or to block
+ * the FOP.
+ * Return:
+ * 0 : If the FOP should continue i.e., if the file is not in the WORM-Retained
+ *     state or if the FOP is unlink and the file is not in the Retained state.
+ * 1: If the FOP sholud block i.e., if the file is in WORM-Retained/WORM state.
+ * 2: Blocks the FOP if any operation fails while doing the state transition or
+ *    fails to get the state of the file.*/
+int32_t
+state_transition (xlator_t *this, gf_boolean_t fop_with_fd, void *file_ptr,
+                  glusterfs_fop_t op, int *ret_val)
+{
+        int label                         =       -1;
+        int ret                           =       -1;
+        uint64_t com_period               =       0;
+        uint64_t ret_period               =       0;
+        uint64_t start_time               =       0;
+        dict_t *dict                      =       NULL;
+        worm_reten_state_t reten_state    =       {0,};
+        read_only_priv_t *priv            =       NULL;
+        struct iatt stbuf                 =       {0,};
+
+        priv = this->private;
+        GF_ASSERT (priv);
+
+        if (fop_with_fd)
+                ret = syncop_fgetxattr (this, (fd_t *)file_ptr, &dict,
+                                        "trusted.start_time", NULL, NULL);
+        else
+                ret = syncop_getxattr (this, (loc_t *)file_ptr, &dict,
+                                       "trusted.start_time", NULL, NULL);
+        if (ret < 0 || !dict) {
+                ret = -2;
+                label = 2;
+                goto out;
+        }
+        ret = dict_get_uint64 (dict, "trusted.start_time", &start_time);
+        if (ret) {
+                label = 2;
+                goto out;
+        }
+
+        ret = worm_get_state (this, fop_with_fd, file_ptr, &reten_state);
+        if (ret == -2) {
+                ret = -1;
+                label = 2;
+                goto out;
+        }
+        com_period = priv->com_period;
+        if (ret == -1 && (time (NULL) - start_time) >= com_period) {
+                if (fop_with_fd)
+                        ret = syncop_fstat (this, (fd_t *)file_ptr, &stbuf,
+                                            NULL, NULL);
+                else
+                        ret = syncop_stat (this, (loc_t *)file_ptr, &stbuf,
+                                           NULL, NULL);
+                if (ret) {
+                        label = 2;
+                        goto out;
+                }
+                ret_period = priv->reten_period;
+                if ((time (NULL) - stbuf.ia_mtime) >= ret_period) {
+                        ret = worm_set_state(this, fop_with_fd, file_ptr,
+                                             &reten_state, &stbuf);
+                        if (ret) {
+                                label = 2;
+                                goto out;
+                        }
+                        label = 1;
+                        goto out;
+                } else {
+                        label = 0;
+                        goto out;
+                }
+        } else if (ret == -1 && (time (NULL) - start_time)
+                   < com_period) {
+                label = 0;
+                goto out;
+        } else if (reten_state.retain &&
+                   (time (NULL) - start_time) >=
+                    reten_state.auto_commit_period) {
+                state_lookup (this, fop_with_fd, file_ptr, &reten_state);
+        }
+        if (reten_state.retain)
+                label = 1;
+        else if (reten_state.worm && !reten_state.retain &&
+                 op == GF_FOP_UNLINK)
+                label = 0;
+        else
+                label = 1;
+
+out:
+        if (dict)
+                dict_unref (dict);
+        *ret_val = ret;
+        return label;
+}
+
+
+/*Function to check whether a file is independently WORMed (i.e., file level
+ * WORM is set on the file). */
+int32_t
+is_wormfile (xlator_t *this, gf_boolean_t fop_with_fd, void *file_ptr)
+{
+        int ret         =       -1;
+        dict_t *dict    =       NULL;
+
+        if (fop_with_fd)
+                ret = syncop_fgetxattr (this, (fd_t *)file_ptr, &dict,
+                                       "trusted.worm_file", NULL, NULL);
+        else
+                ret = syncop_getxattr (this, (loc_t *)file_ptr, &dict,
+                                       "trusted.worm_file", NULL, NULL);
+        if (dict) {
+                ret = 0;
+                dict_unref (dict);
+        }
+        return ret;
+}
\ No newline at end of file
diff --git a/xlators/features/read-only/src/worm-helper.h b/xlators/features/read-only/src/worm-helper.h
new file mode 100644
index 00000000000..a06a1d38063
--- /dev/null
+++ b/xlators/features/read-only/src/worm-helper.h
@@ -0,0 +1,36 @@
+/*
+   Copyright (c) 2016 Red Hat, Inc. <http://www.redhat.com>
+   This file is part of GlusterFS.
+
+   This file is licensed to you under your choice of the GNU Lesser
+   General Public License, version 3 or any later version (LGPLv3 or
+   later), or the GNU General Public License, version 2 (GPLv2), in all
+   cases as published by the Free Software Foundation.
+*/
+
+gf_boolean_t is_write_disabled (struct iatt *stbuf);
+
+int32_t worm_init_state (xlator_t *this, gf_boolean_t fop_with_fd,
+                         void *file_ptr);
+
+int32_t worm_set_state (xlator_t *this, gf_boolean_t fop_with_fd,
+                        void *file_ptr, worm_reten_state_t *retention_state,
+                        struct iatt *stbuf);
+
+int32_t worm_get_state (xlator_t *this, gf_boolean_t fop_with_fd,
+                        void *file_ptr, worm_reten_state_t *reten_state);
+
+void state_lookup (xlator_t *this, gf_boolean_t fop_with_fd, void *file_ptr,
+                   worm_reten_state_t *reten_state);
+
+void serialize_state (worm_reten_state_t *reten_state, char *val);
+
+void deserialize_state (char *val, worm_reten_state_t *reten_state);
+
+int32_t set_xattr (xlator_t *this, worm_reten_state_t *reten_state,
+                   gf_boolean_t fop_with_fd, void *file_ptr);
+
+int32_t state_transition (xlator_t *this, gf_boolean_t fop_with_fd,
+                          void *file_ptr, glusterfs_fop_t op, int *ret_val);
+
+int32_t is_wormfile (xlator_t *this, gf_boolean_t fop_with_fd, void *file_ptr);
diff --git a/xlators/features/read-only/src/worm.c b/xlators/features/read-only/src/worm.c
index f117e206285..2f35f1fc9f1 100644
--- a/xlators/features/read-only/src/worm.c
+++ b/xlators/features/read-only/src/worm.c
@@ -1,5 +1,5 @@
 /*
-   Copyright (c) 2008-2012 Red Hat, Inc. <http://www.redhat.com>
+   Copyright (c) 2008-2012, 2016 Red Hat, Inc. <http://www.redhat.com>
    This file is part of GlusterFS.
 
    This file is licensed to you under your choice of the GNU Lesser
@@ -12,6 +12,9 @@
 #include "read-only-common.h"
 #include "read-only-mem-types.h"
 #include "read-only.h"
+#include "syncop.h"
+#include "worm-helper.h"
+
 
 int32_t
 mem_acct_init (xlator_t *this)
@@ -26,35 +29,393 @@ mem_acct_init (xlator_t *this)
         return ret;
 }
 
-static int32_t
-worm_open_cbk (call_frame_t *frame, void *cookie, xlator_t *this, int32_t op_ret,
-               int32_t op_errno, fd_t *fd, dict_t *xdata)
-{
-        STACK_UNWIND_STRICT (open, frame, op_ret, op_errno, fd, xdata);
-        return 0;
-}
 
 int32_t
 worm_open (call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t flags,
            fd_t *fd, dict_t *xdata)
 {
         if (is_readonly_or_worm_enabled (this) &&
-            ((((flags & O_ACCMODE) == O_WRONLY) ||
-              ((flags & O_ACCMODE) == O_RDWR)) &&
-              !(flags & O_APPEND))) {
+            (flags & (O_WRONLY | O_RDWR | O_APPEND))) {
                 STACK_UNWIND_STRICT (open, frame, -1, EROFS, NULL, NULL);
                 return 0;
         }
 
-        STACK_WIND (frame, worm_open_cbk, FIRST_CHILD(this),
+        STACK_WIND_TAIL (frame, FIRST_CHILD(this),
                     FIRST_CHILD(this)->fops->open, loc, flags, fd, xdata);
         return 0;
 }
 
+
+int32_t
+worm_link (call_frame_t *frame, xlator_t *this, loc_t *oldloc, loc_t *newloc,
+           dict_t *xdata)
+{
+        int ret                     =       -1;
+        int label                   =       -1;
+
+        if (is_readonly_or_worm_enabled (this))
+                goto unwind;
+        gf_uuid_copy (oldloc->gfid, oldloc->inode->gfid);
+        if (is_wormfile (this, _gf_false, oldloc))
+                goto wind;
+        label = state_transition (this, _gf_false, oldloc, GF_FOP_LINK, &ret);
+        if (label == 0)
+                goto wind;
+        if (label == 1)
+                goto unwind;
+        if (label == 2)
+                goto out;
+
+unwind:
+        STACK_UNWIND_STRICT (link, frame, -1, EROFS, NULL, NULL, NULL, NULL,
+                             NULL);
+        ret = 0;
+        goto out;
+wind:
+        STACK_WIND_TAIL (frame, FIRST_CHILD(this),
+                         FIRST_CHILD(this)->fops->link,
+                         oldloc, newloc, xdata);
+        ret = 0;
+out:
+        if (label == 2)
+                STACK_UNWIND_STRICT (link, frame, -1, ret, NULL, NULL,
+                                     NULL, NULL, NULL);
+        return ret;
+}
+
+
+int32_t
+worm_unlink (call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t flags,
+             dict_t *xdata)
+{
+        int ret                     =       -1;
+        int label                   =       -1;
+
+        if (is_readonly_or_worm_enabled (this))
+                goto unwind;
+        gf_uuid_copy (loc->gfid, loc->inode->gfid);
+        if (is_wormfile (this, _gf_false, loc))
+                goto wind;
+        label = state_transition (this, _gf_false, loc, GF_FOP_UNLINK, &ret);
+        if (label == 0)
+                goto wind;
+        if (label == 1)
+                goto unwind;
+        if (label == 2)
+                goto out;
+
+unwind:
+        STACK_UNWIND_STRICT (unlink, frame, -1, EROFS, NULL, NULL, NULL);
+        ret = 0;
+        goto out;
+wind:
+        STACK_WIND_TAIL (frame, FIRST_CHILD(this),
+                         FIRST_CHILD(this)->fops->unlink,
+                         loc, flags, xdata);
+        ret = 0;
+out:
+        if (label == 2)
+                STACK_UNWIND_STRICT (unlink, frame, -1, ret,
+                                     NULL, NULL, NULL);
+        return ret;
+}
+
+
+int32_t
+worm_rename (call_frame_t *frame, xlator_t *this,
+             loc_t *oldloc, loc_t *newloc, dict_t *xdata)
+{
+        int ret                     =       -1;
+        int label                   =       -1;
+
+        if (is_readonly_or_worm_enabled (this))
+                goto unwind;
+        gf_uuid_copy (oldloc->gfid, oldloc->inode->gfid);
+        if (is_wormfile (this, _gf_false, oldloc))
+                goto wind;
+        label = state_transition (this, _gf_false, oldloc, GF_FOP_RENAME,
+                                  &ret);
+        if (label == 0)
+                goto wind;
+        if (label == 1)
+                goto unwind;
+        if (label == 2)
+                goto out;
+
+unwind:
+        STACK_UNWIND_STRICT (rename, frame, -1, EROFS, NULL,
+                             NULL, NULL, NULL, NULL, NULL);
+        ret = 0;
+        goto out;
+wind:
+        STACK_WIND_TAIL (frame, FIRST_CHILD (this),
+                         FIRST_CHILD (this)->fops->rename,
+                         oldloc, newloc, xdata);
+        ret = 0;
+out:
+        if (label == 2)
+                STACK_UNWIND_STRICT (rename, frame, -1, ret, NULL,
+                                     NULL, NULL, NULL, NULL, NULL);
+        return ret;
+}
+
+
+int32_t
+worm_truncate (call_frame_t *frame, xlator_t *this, loc_t *loc, off_t offset,
+               dict_t *xdata)
+{
+        int ret                     =       -1;
+        int label                   =       -1;
+
+        if (is_readonly_or_worm_enabled (this))
+                goto unwind;
+        if (is_wormfile (this, _gf_false, loc))
+                goto wind;
+        label = state_transition (this, _gf_false, loc, GF_FOP_TRUNCATE,
+                                  &ret);
+        if (label == 0)
+                goto wind;
+        if (label == 1)
+                goto unwind;
+        if (label == 2)
+                goto out;
+
+unwind:
+        STACK_UNWIND_STRICT (truncate, frame, -1, EROFS,
+                                     NULL, NULL, NULL);
+        ret = 0;
+        goto out;
+
+wind:
+        STACK_WIND_TAIL (frame,
+                         FIRST_CHILD (this),
+                         FIRST_CHILD (this)->fops->truncate,
+                         loc, offset, xdata);
+        ret = 0;
+out:
+        if (label == 2)
+                STACK_UNWIND_STRICT (truncate, frame, -1, ret,
+                                     NULL, NULL, NULL);
+        return ret;
+}
+
+
+int32_t
+worm_setattr (call_frame_t *frame, xlator_t *this, loc_t *loc,
+              struct iatt *stbuf, int32_t valid, dict_t *xdata)
+{
+        gf_boolean_t rd_only            =       _gf_false;
+        worm_reten_state_t reten_state  =       {0,};
+        struct iatt stpre               =       {0,};
+        int ret                         =       -1;
+
+        if (is_wormfile (this, _gf_false, loc))
+                goto wind;
+        if (valid & GF_SET_ATTR_MODE) {
+                rd_only = is_write_disabled (stbuf);
+                if (!rd_only)
+                        goto wind;
+
+                ret = worm_set_state (this, _gf_false, loc,
+                                      &reten_state, stbuf);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "Error setting worm state");
+                        goto out;
+                }
+        } else if (valid & GF_SET_ATTR_ATIME) {
+                ret = worm_get_state (this, _gf_false, loc, &reten_state);
+                if (ret)
+                        goto wind;
+                if (reten_state.retain) {
+                        ret = syncop_stat (this, loc, &stpre, NULL, NULL);
+                        if (ret)
+                                goto out;
+                        if (reten_state.ret_mode == 0) {
+                                if (stbuf->ia_atime < stpre.ia_mtime) {
+                                        gf_log (this->name, GF_LOG_ERROR,
+                                                "Cannot set atime less than "
+                                                "the mtime for a WORM-Retained "
+                                                "file");
+                                        goto unwind;
+                                }
+                        } else {
+                                if (stbuf->ia_atime < stpre.ia_atime) {
+                                        gf_log (this->name, GF_LOG_ERROR,
+                                                "Cannot decrease the atime of a"
+                                                " WORM-Retained file in "
+                                                "Enterprise mode");
+                                        goto unwind;
+                                }
+                        }
+                        stbuf->ia_mtime = stpre.ia_mtime;
+                }
+        }
+
+wind:
+        STACK_WIND_TAIL (frame, FIRST_CHILD (this),
+                    FIRST_CHILD (this)->fops->setattr,
+                    loc, stbuf, valid, xdata);
+        ret = 0;
+        goto out;
+unwind:
+        STACK_UNWIND_STRICT (setattr, frame, -1, EROFS, NULL, NULL, NULL);
+out:
+        return ret;
+}
+
+
+int32_t
+worm_fsetattr (call_frame_t *frame, xlator_t *this, fd_t *fd,
+               struct iatt *stbuf, int32_t valid, dict_t *xdata)
+{
+        gf_boolean_t rd_only            =       _gf_false;
+        worm_reten_state_t reten_state  =       {0,};
+        struct iatt stpre               =       {0,};
+        int ret                         =       -1;
+
+        if (is_wormfile (this, _gf_true, fd))
+                goto wind;
+        if (valid & GF_SET_ATTR_MODE) {
+                rd_only = is_write_disabled (stbuf);
+                if (!rd_only)
+                        goto wind;
+
+                ret = worm_set_state (this, _gf_true, fd,
+                                      &reten_state, stbuf);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "Error setting worm state");
+                        goto out;
+                }
+        } else if (valid & GF_SET_ATTR_ATIME) {
+                ret = worm_get_state (this, _gf_true, fd, &reten_state);
+                if (ret)
+                        goto wind;
+                if (reten_state.retain) {
+                        ret = syncop_fstat (this, fd, &stpre, NULL, NULL);
+                        if (ret)
+                                goto out;
+                        if (reten_state.ret_mode == 0) {
+                                if (stbuf->ia_atime < stpre.ia_mtime) {
+                                        gf_log (this->name, GF_LOG_ERROR,
+                                                "Cannot set atime less than "
+                                                "the mtime for a WORM-Retained "
+                                                "file");
+                                        goto unwind;
+                                }
+                        } else {
+                                if (stbuf->ia_atime < stpre.ia_atime) {
+                                        gf_log (this->name, GF_LOG_ERROR,
+                                                "Cannot decrease the atime of a"
+                                                " WORM-Retained file in "
+                                                "Enterprise mode");
+                                        goto unwind;
+                                }
+                        }
+                        stbuf->ia_mtime = stpre.ia_mtime;
+                }
+        }
+
+wind:
+        STACK_WIND_TAIL (frame, FIRST_CHILD (this),
+                    FIRST_CHILD (this)->fops->fsetattr,
+                    fd, stbuf, valid, xdata);
+        ret = 0;
+        goto out;
+unwind:
+        STACK_UNWIND_STRICT (fsetattr, frame, -1, EROFS, NULL, NULL, NULL);
+out:
+        return ret;
+}
+
+
+int32_t
+worm_writev (call_frame_t *frame, xlator_t *this, fd_t *fd,
+             struct iovec *vector, int32_t count, off_t offset, uint32_t flags,
+             struct iobref *iobref, dict_t *xdata)
+{
+        worm_reten_state_t reten_state    =       {0,};
+        int ret                           =       -1;
+
+        if (is_readonly_or_worm_enabled (this))
+                goto unwind;
+        if (is_wormfile (this, _gf_true, fd))
+                goto wind;
+        ret = worm_get_state (this, _gf_true, fd, &reten_state);
+        if (!reten_state.worm)
+                goto wind;
+
+unwind:
+        STACK_UNWIND_STRICT (writev, frame, -1, EROFS, NULL, NULL, NULL);
+        goto out;
+
+wind:
+        STACK_WIND_TAIL (frame,
+                         FIRST_CHILD (this),
+                         FIRST_CHILD (this)->fops->writev,
+                         fd, vector, count, offset, flags,
+                         iobref, xdata);
+        gf_log (this->name, GF_LOG_INFO, "WORM writev");
+        ret = 0;
+
+out:
+        return ret;
+}
+
+
+int32_t
+worm_create (call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t flags,
+             mode_t mode, mode_t umask, fd_t *fd, dict_t *xdata)
+{
+        int ret                   =       -1;
+        read_only_priv_t *priv    =       NULL;
+        dict_t *dict              =       NULL;
+
+        STACK_WIND_TAIL (frame, FIRST_CHILD (this),
+                         FIRST_CHILD(this)->fops->create, loc, flags,
+                         mode, umask, fd, xdata);
+        priv = this->private;
+        GF_ASSERT (priv);
+
+        if (priv->worm_file) {
+                dict = dict_new ();
+                if (!dict) {
+                        gf_log (this->name, GF_LOG_ERROR, "Error creating the "
+                                "dict");
+                        goto out;
+                }
+                GF_VALIDATE_OR_GOTO (this->name, dict, out);
+                ret = dict_set_int8 (dict, "trusted.worm_file", 1);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR, "Error in setting "
+                                "the dict");
+                        goto out;
+                }
+                ret = syncop_fsetxattr (this, fd, dict, 0, NULL, NULL);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "Error setting xattr");
+                        goto out;
+                }
+                ret = worm_init_state (this, _gf_true, fd);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "Error initializing state");
+                }
+        }
+
+out:
+        if (dict)
+                dict_destroy (dict);
+        return ret;
+}
+
+
 int32_t
 init (xlator_t *this)
 {
-        int                     ret     = -1;
+        int                    ret      = -1;
         read_only_priv_t       *priv    = NULL;
 
         if (!this->children || this->children->next) {
@@ -68,11 +429,34 @@ init (xlator_t *this)
                         "dangling volume. check volfile ");
         }
 
-        priv = GF_CALLOC (1, sizeof (*priv), gf_read_only_mt_priv_t);
-        if (!priv)
+        this->local_pool = mem_pool_new (read_only_priv_t, 64);
+        if (!this->local_pool) {
+                ret = -1;
+                gf_log (this->name, GF_LOG_ERROR,
+                        "failed to create read_only_priv_t's memory pool");
+                goto out;
+        }
+
+        priv = mem_get0 (this->local_pool);
+        if (!priv) {
+                gf_log (this->name, GF_LOG_ERROR, "Error allocating priv");
+                goto out;
+        }
+
+        priv->reten_mode = mem_get0 (this->local_pool);
+        if (!priv->reten_mode) {
+                gf_log (this->name, GF_LOG_ERROR, "Error allocating "
+                        "reten_mode");
                 goto out;
+        }
 
-        GF_OPTION_INIT ("worm", priv->readonly_or_worm_enabled, bool, out);
+        GF_OPTION_INIT ("worm", priv->readonly_or_worm_enabled,
+                        bool, out);
+        GF_OPTION_INIT ("worm-file-level", priv->worm_file, bool, out);
+        GF_OPTION_INIT ("default-retention-period", priv->reten_period,
+                        uint64, out);
+        GF_OPTION_INIT ("auto-commit-period", priv->com_period, uint64, out);
+        GF_OPTION_INIT ("retention-mode", priv->reten_mode, str, out);
 
         this->private = priv;
         ret = 0;
@@ -80,25 +464,33 @@ out:
         return ret;
 }
 
+
 int
 reconfigure (xlator_t *this, dict_t *options)
 {
-        read_only_priv_t  *priv                     = NULL;
+        read_only_priv_t   *priv                    = NULL;
         int                ret                      = -1;
-        gf_boolean_t       readonly_or_worm_enabled = _gf_false;
 
         priv = this->private;
         GF_ASSERT (priv);
 
-        GF_OPTION_RECONF ("worm", readonly_or_worm_enabled, options, bool, out);
-
-        priv->readonly_or_worm_enabled = readonly_or_worm_enabled;
+        GF_OPTION_RECONF ("worm", priv->readonly_or_worm_enabled,
+                          options, bool, out);
+        GF_OPTION_RECONF ("worm-file-level", priv->worm_file, options, bool,
+                          out);
+        GF_OPTION_RECONF ("default-retention-period", priv->reten_period,
+                          options, uint64, out);
+        GF_OPTION_RECONF ("retention-mode", priv->reten_mode, options, str,
+                          out);
+        GF_OPTION_RECONF ("auto-commit-period", priv->com_period, options,
+                          uint64, out);
         ret = 0;
 out:
         gf_log (this->name, GF_LOG_DEBUG, "returning %d", ret);
         return ret;
 }
 
+
 void
 fini (xlator_t *this)
 {
@@ -106,21 +498,31 @@ fini (xlator_t *this)
 
         priv = this->private;
         if (!priv)
-                return;
-
+                goto out;
+        if (priv->reten_mode != NULL) {
+                mem_put (priv->reten_mode);
+                priv->reten_mode = NULL;
+        }
+        mem_put (priv);
         this->private = NULL;
-        GF_FREE (priv);
-
+        mem_pool_destroy (this->local_pool);
+out:
         return;
 }
 
+
 struct xlator_fops fops = {
         .open        = worm_open,
+        .writev      = worm_writev,
+        .setattr     = worm_setattr,
+        .fsetattr    = worm_fsetattr,
+        .rename      = worm_rename,
+        .link        = worm_link,
+        .unlink      = worm_unlink,
+        .truncate    = worm_truncate,
+        .create      = worm_create,
 
-        .unlink      = ro_unlink,
         .rmdir       = ro_rmdir,
-        .rename      = ro_rename,
-        .truncate    = ro_truncate,
         .removexattr = ro_removexattr,
         .fsyncdir    = ro_fsyncdir,
         .xattrop     = ro_xattrop,
@@ -131,8 +533,10 @@ struct xlator_fops fops = {
         .lk          = ro_lk,
 };
 
+
 struct xlator_cbks cbks;
 
+
 struct volume_options options[] = {
         { .key  = {"worm"},
           .type = GF_OPTION_TYPE_BOOL,
@@ -140,5 +544,26 @@ struct volume_options options[] = {
           .description = "When \"on\", makes a volume get write once read many "
                          " feature. It is turned \"off\" by default."
         },
+        { .key = {"worm-file-level"},
+          .type = GF_OPTION_TYPE_BOOL,
+          .default_value = "off",
+          .description = "When \"on\", activates the file level worm. "
+                         "It is turned \"off\" by default."
+        },
+        { .key = {"default-retention-period"},
+          .type = GF_OPTION_TYPE_TIME,
+          .default_value = "120",
+          .description = "The default retention period for the files."
+        },
+        { .key = {"retention-mode"},
+          .type = GF_OPTION_TYPE_STR,
+          .default_value = "relax",
+          .description = "The mode of retention (relax/enterprise). "
+                         "It is relax by default."
+        },
+        { .key = {"auto-commit-period"},
+          .type = GF_OPTION_TYPE_TIME,
+          .default_value = "180",
+          .description = "Auto commit period for the files."
+        },
 };
-
diff --git a/xlators/mgmt/glusterd/src/glusterd-volgen.c b/xlators/mgmt/glusterd/src/glusterd-volgen.c
index 23890238933..e88116e02ba 100644
--- a/xlators/mgmt/glusterd/src/glusterd-volgen.c
+++ b/xlators/mgmt/glusterd/src/glusterd-volgen.c
@@ -2075,7 +2075,8 @@ brick_graph_add_ro (volgen_graph_t *graph, glusterd_volinfo_t *volinfo,
                 goto out;
 
         if (dict_get_str_boolean (set_dict, "features.read-only", 0) &&
-            dict_get_str_boolean (set_dict, "features.worm", 0)) {
+            (dict_get_str_boolean (set_dict, "features.worm", 0) ||
+             dict_get_str_boolean (set_dict, "features.worm-file-level", 0))) {
                 gf_msg (THIS->name, GF_LOG_ERROR, errno,
                         GD_MSG_DICT_GET_FAILED,
                         "read-only and worm cannot be set together");
@@ -2107,7 +2108,8 @@ brick_graph_add_worm (volgen_graph_t *graph, glusterd_volinfo_t *volinfo,
                 goto out;
 
         if (dict_get_str_boolean (set_dict, "features.read-only", 0) &&
-            dict_get_str_boolean (set_dict, "features.worm", 0)) {
+            (dict_get_str_boolean (set_dict, "features.worm", 0) ||
+             dict_get_str_boolean (set_dict, "features.worm-file-level", 0))) {
                 gf_msg (THIS->name, GF_LOG_ERROR, 0,
                         GD_MSG_INCOMPATIBLE_VALUE,
                         "read-only and worm cannot be set together");
@@ -2402,8 +2404,6 @@ static volgen_brick_xlator_t server_graph_table[] = {
         {brick_graph_add_server, NULL},
         {brick_graph_add_io_stats, NULL},
         {brick_graph_add_cdc, NULL},
-        {brick_graph_add_ro, NULL},
-        {brick_graph_add_worm, NULL},
         {brick_graph_add_quota, "quota"},
         {brick_graph_add_index, "index"},
         {brick_graph_add_barrier, NULL},
@@ -2412,6 +2412,8 @@ static volgen_brick_xlator_t server_graph_table[] = {
         {brick_graph_add_iot, "io-threads"},
         {brick_graph_add_upcall, "upcall"},
         {brick_graph_add_pump, NULL},
+        {brick_graph_add_ro, NULL},
+        {brick_graph_add_worm, NULL},
         {brick_graph_add_locks, "locks"},
         {brick_graph_add_acl, "acl"},
         {brick_graph_add_bitrot_stub, "bitrot-stub"},
diff --git a/xlators/mgmt/glusterd/src/glusterd-volume-set.c b/xlators/mgmt/glusterd/src/glusterd-volume-set.c
index 5bb9d9077b5..68dec22ecaf 100644
--- a/xlators/mgmt/glusterd/src/glusterd-volume-set.c
+++ b/xlators/mgmt/glusterd/src/glusterd-volume-set.c
@@ -863,6 +863,82 @@ out:
         return ret;
 }
 
+
+static int
+validate_worm (glusterd_volinfo_t *volinfo, dict_t *dict, char *key,
+               char *value, char **op_errstr)
+{
+        xlator_t *this      =       NULL;
+        gf_boolean_t b      =       _gf_false;
+        int ret             =       -1;
+
+        this = THIS;
+        GF_VALIDATE_OR_GOTO ("glusterd", this, out);
+        ret = gf_string2boolean (value, &b);
+        if (ret) {
+                gf_asprintf (op_errstr, "%s is not a valid boolean value. %s "
+                             "expects a valid boolean value.", value, key);
+                gf_msg (this->name, GF_LOG_ERROR, 0,
+                        GD_MSG_INVALID_ENTRY, "%s", *op_errstr);
+        }
+out:
+        gf_msg_debug ("glusterd", 0, "Returning %d", ret);
+
+        return ret;
+}
+
+
+static int
+validate_worm_period (glusterd_volinfo_t *volinfo, dict_t *dict, char *key,
+               char *value, char **op_errstr)
+{
+        xlator_t *this      =       NULL;
+        uint64_t period     =       -1;
+        int ret             =       -1;
+
+        this = THIS;
+        GF_VALIDATE_OR_GOTO ("glusterd", this, out);
+        ret = gf_string2uint64 (value, &period);
+        if (ret) {
+                gf_asprintf (op_errstr, "%s is not a valid uint64_t value."
+                          " %s expects a valid uint64_t value.", value, key);
+                gf_msg (this->name, GF_LOG_ERROR, 0,
+                        GD_MSG_INVALID_ENTRY, "%s", *op_errstr);
+        }
+out:
+        gf_msg_debug ("glusterd", 0, "Returning %d", ret);
+
+        return ret;
+}
+
+
+static int
+validate_reten_mode (glusterd_volinfo_t *volinfo, dict_t *dict, char *key,
+               char *value, char **op_errstr)
+{
+        xlator_t *this      =       NULL;
+        int ret             =       -1;
+
+        this = THIS;
+        GF_VALIDATE_OR_GOTO ("glusterd", this, out);
+        if ((strcmp (value, "relax") &&
+            strcmp (value, "enterprise"))) {
+                gf_asprintf (op_errstr, "The value of retention mode should be "
+                             "either relax or enterprise. But the value"
+                             " of %s is %s", key, value);
+                gf_msg (this->name, GF_LOG_ERROR, 0, GD_MSG_INVALID_ENTRY,
+                        "%s", *op_errstr);
+                ret = -1;
+                goto out;
+        }
+        ret = 0;
+out:
+        gf_msg_debug ("glusterd", 0, "Returning %d", ret);
+
+        return ret;
+}
+
+
 /* dispatch table for VOLUME SET
  * -----------------------------
  *
@@ -2314,13 +2390,40 @@ struct volopt_map_entry glusterd_volopt_map[] = {
           .op_version = 1,
           .flags      = OPT_FLAG_CLIENT_OPT | OPT_FLAG_XLATOR_OPT
         },
-        { .key        = "features.worm",
-          .voltype    = "features/worm",
-          .option     = "worm",
-          .value      = "off",
-          .op_version = 2,
+        { .key         = "features.worm",
+          .voltype     = "features/worm",
+          .option      = "worm",
+          .value       = "off",
+          .validate_fn = validate_worm,
+          .op_version  = 2,
+          .flags       = OPT_FLAG_CLIENT_OPT | OPT_FLAG_XLATOR_OPT
+        },
+        { .key         = "features.worm-file-level",
+          .voltype     = "features/worm",
+          .option      = "worm-file-level",
+          .value       = "off",
+          .validate_fn = validate_worm,
+          .op_version  = GD_OP_VERSION_3_8_0,
           .flags      = OPT_FLAG_CLIENT_OPT | OPT_FLAG_XLATOR_OPT
         },
+        { .key         = "features.default-retention-period",
+          .voltype     = "features/worm",
+          .option      = "default-retention-period",
+          .validate_fn = validate_worm_period,
+          .op_version  = GD_OP_VERSION_3_8_0,
+        },
+        { .key         = "features.retention-mode",
+          .voltype     = "features/worm",
+          .option      = "retention-mode",
+          .validate_fn = validate_reten_mode,
+          .op_version  = GD_OP_VERSION_3_8_0,
+        },
+        { .key         = "features.auto-commit-period",
+          .voltype     = "features/worm",
+          .option      = "auto-commit-period",
+          .validate_fn = validate_worm_period,
+          .op_version  = GD_OP_VERSION_3_8_0,
+        },
         { .key         = "storage.linux-aio",
           .voltype     = "storage/posix",
           .op_version  = 1