features/quota : Introducing inode quota

==========================================================================
                             Inode quota
==========================================================================
= Currently, the only way to retrieve the number of files/objects in a   =
= directory or volume is to do a crawl of the entire directory/volume.   =
= This is expensive and is not scalable.                                 =
=                                                                        =
= The proposed mechanism will provide an easier alternative to determine =
= the count of files/objects in a directory or volume.                   =
=                                                                        =
= The new mechanism proposes to store count of objects/files as part of  =
= an extended attribute of a directory. Each directory's extended        =
= attribute value will indicate the number of files/objects present      =
= in a tree with the directory being considered as the root of the tree. =
=                                                                        =
= The count value can be accessed by performing a getxattr().            =
= Cluster translators like afr, dht and stripe will perform aggregation  =
= of count values from various bricks when getxattr() happens on the key =
= associated with file/object count.                                     =

A new interface is introduced:
------------------------------
        limit-objects  : limit the number of inodes at directory level
        list-objects   : list the directories where the limit is set
        remove-objects : remove the limit from the directory

==========================================================================

CLI COMMAND:
gluster volume quota <volname> limit-objects <path> <number> [<percent>]

* <number> is a hard-limit for number of objects limitation for path "<path>"
  If hard-limit is exceeded, creation of file/directory is no longer
permitted.

* <percent> is a soft-limit for number of objects creation for path "<path>"
  If soft-limit is exceeded, a warning is issued for each creation.

CLI COMMAND:
gluster volume quota <volname> remove-objects [path]

==========================================================================

CLI COMMAND:
gluster volume quota <volname> list-objects [path] ...

Sample output:
------------------
  Path                   Hard-limit Soft-limit   Used  Available
Soft-limit exceeded?
Hard-limit exceeded?
  ------------------------------------------------------------------------
--------------------------------------
  /dir                      10       80%          10       0
Yes
        Yes

==========================================================================

[root@snapshot-28 dir]# ls
a  b  file11  file12  file13  file14  file15  file16  file17
[root@snapshot-28 dir]# touch a1
touch: cannot touch `a1': Disk quota exceeded
* Nine files are created in directory "dir" and directory is included in
* the
count too. Hence the limit "10" is reached and further file creation
fails

==========================================================================

Note: We have also done some re-factoring in cli for volume name
validation. New function cli_validate_volname is created

==========================================================================

Change-Id: I1823497de4f790a2a20ebb1770293472ea33ee2b
BUG: 1190108
Signed-off-by: Sachin Pandit <spandit@redhat.com>
Signed-off-by: vmallika <vmallika@redhat.com>
Reviewed-on: http://review.gluster.org/9769
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Vijay Bellur <vbellur@redhat.com>

1 files changed, 81 insertions, 34 deletions

diff --git a/xlators/cluster/dht/src/dht-common.c b/xlators/cluster/dht/src/dht-common.c
index 729ab30e672..9fda4aa07d6 100644
--- a/xlators/cluster/dht/src/dht-common.c
+++ b/xlators/cluster/dht/src/dht-common.c
@@ -23,6 +23,7 @@
 #include "defaults.h"
 #include "byte-order.h"
 #include "glusterfs-acl.h"
+#include "quota-common-utils.h"
 
 #include <sys/time.h>
 #include <libgen.h>
@@ -31,47 +32,90 @@
 int dht_link2 (xlator_t *this, call_frame_t *frame, int op_ret);
 
 int
-dht_aggregate (dict_t *this, char *key, data_t *value, void *data)
+dht_aggregate_quota_xattr (dict_t *dst, char *key, data_t *value)
 {
-        dict_t  *dst  = NULL;
-        int64_t *ptr  = 0, *size = NULL;
-        int32_t  ret  = -1;
-        data_t  *dict_data = NULL;
-
-        dst = data;
+        int              ret            = -1;
+        quota_meta_t    *meta_dst       = NULL;
+        quota_meta_t    *meta_src       = NULL;
+        int64_t         *size           = NULL;
+        int64_t          dst_dir_count  = 0;
+        int64_t          src_dir_count  = 0;
+
+        if (value == NULL) {
+                gf_log ("dht", GF_LOG_WARNING, "data value is NULL");
+                ret = -1;
+                goto out;
+        }
 
-        if (strcmp (key, GF_XATTR_QUOTA_SIZE_KEY) == 0) {
-                ret = dict_get_bin (dst, key, (void **)&size);
+        ret = dict_get_bin (dst, key, (void **)&meta_dst);
+        if (ret < 0) {
+                meta_dst = GF_CALLOC (1, sizeof (quota_meta_t),
+                                      gf_common_quota_meta_t);
+                if (meta_dst == NULL) {
+                        gf_msg ("dht", GF_LOG_WARNING, 0, DHT_MSG_NO_MEMORY,
+                                "Memory allocation failed");
+                        ret = -1;
+                        goto out;
+                }
+                ret = dict_set_bin (dst, key, meta_dst,
+                                    sizeof (quota_meta_t));
                 if (ret < 0) {
-                        size = GF_CALLOC (1, sizeof (int64_t),
-                                          gf_common_mt_char);
-                        if (size == NULL) {
-                                gf_msg ("dht", GF_LOG_WARNING, 0,
-                                        DHT_MSG_NO_MEMORY,
-                                        "Memory allocation failed");
-                                return -1;
-                        }
-                        ret = dict_set_bin (dst, key, size, sizeof (int64_t));
-                        if (ret < 0) {
-                                gf_log ("dht", GF_LOG_WARNING,
-                                        "dht aggregate dict set failed");
-                                GF_FREE (size);
-                                return -1;
-                        }
+                        gf_log ("dht", GF_LOG_WARNING,
+                                "dht aggregate dict set failed");
+                        GF_FREE (meta_dst);
+                        ret = -1;
+                        goto out;
                 }
+        }
+
+        if (value->len > sizeof (int64_t)) {
+                meta_src = data_to_bin (value);
+
+                meta_dst->size = hton64 (ntoh64 (meta_dst->size) +
+                                         ntoh64 (meta_src->size));
+                meta_dst->file_count = hton64 (ntoh64 (meta_dst->file_count) +
+                                               ntoh64 (meta_src->file_count));
 
-                ptr = data_to_bin (value);
-                if (ptr == NULL) {
-                        gf_log ("dht", GF_LOG_WARNING, "data to bin failed");
-                        return -1;
+                if (value->len > (2 * sizeof (int64_t))) {
+                        dst_dir_count = ntoh64 (meta_dst->dir_count);
+                        src_dir_count = ntoh64 (meta_src->dir_count);
+
+                        if (src_dir_count > dst_dir_count)
+                                meta_dst->dir_count = meta_src->dir_count;
+                } else {
+                        meta_dst->dir_count = 0;
                 }
+        } else {
+                size = data_to_bin (value);
+                meta_dst->size = hton64 (ntoh64 (meta_dst->size) +
+                                         ntoh64 (*size));
+        }
+
+        ret = 0;
+out:
+        return ret;
+}
 
-                *size = hton64 (ntoh64 (*size) + ntoh64 (*ptr));
+int
+dht_aggregate (dict_t *this, char *key, data_t *value, void *data)
+{
+        dict_t          *dst            = NULL;
+        int32_t          ret            = -1;
+        data_t          *dict_data      = NULL;
+
+        dst = data;
 
+        if (strcmp (key, QUOTA_SIZE_KEY) == 0) {
+                ret = dht_aggregate_quota_xattr (dst, key, value);
+                if (ret) {
+                        gf_log ("dht", GF_LOG_WARNING, "Failed to "
+                                "aggregate qutoa xattr");
+                        goto out;
+                }
         } else if (fnmatch (GF_XATTR_STIME_PATTERN, key, FNM_NOESCAPE) == 0) {
                 ret = gf_get_min_stime (THIS, dst, key, value);
                 if (ret < 0)
-                        return ret;
+                        goto out;
         } else {
                 /* compare user xattrs only */
                 if (!strncmp (key, "user.", strlen ("user."))) {
@@ -85,14 +129,17 @@ dht_aggregate (dict_t *this, char *key, data_t *value, void *data)
                         }
                 }
                 ret = dict_set (dst, key, value);
-                if (ret)
+                if (ret) {
                         gf_msg ("dht", GF_LOG_WARNING, 0,
                                 DHT_MSG_DICT_SET_FAILED,
                                 "Failed to set dictionary value: key = %s",
                                 key);
+                }
         }
 
-        return 0;
+        ret = 0;
+out:
+        return ret;
 }
 
 
@@ -255,7 +302,6 @@ selfheal:
 
 }
 
-
 int
 dht_discover_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                   int op_ret, int op_errno,
@@ -2896,7 +2942,8 @@ dht_getxattr (call_frame_t *frame, xlator_t *this,
                 return 0;
         }
 
-        if (key && !strcmp (GF_XATTR_QUOTA_LIMIT_LIST, key)) {
+        if (key && (!strcmp (GF_XATTR_QUOTA_LIMIT_LIST, key) ||
+                    !strcmp (GF_XATTR_QUOTA_LIMIT_LIST_OBJECT, key))) {
                 /* quota hardlimit and aggregated size of a directory is stored
                  * in inode contexts of each brick. Hence its good enough that
                  * we send getxattr for this key to any brick.