features/quota: Send the immediate parent with limit in quota statfs adjustment

Problem:
Assume the directory structure /quota_limit_dir/subdir and quota_limit_dir is
set with some limit. When quota-deem-statfs is enabled the output of
'df /quota_limit_dir' would display quota modified values wrt to
quota_limit_dir where as 'df /quota_limit_subdir/subdir' would display the
quota modified values wrt volume root (/).

This behaviour is not expected since, when mounted with subdirectory admin
doesn't want users to know information above the subdirectory mounted.

Solution:
Any subdirectory within a quota_limit_dir would show the modified values as in
the /quota_limit_dir. It searches for the nearest parent that has quota limit
set and modifies the statvfs wrt that.

Change-Id: Ie10fae8999bddbb766b1dbeb881723ed80dce688
BUG: 1080296
Signed-off-by: Varun Shastry <vshastry@redhat.com>
Reviewed-on: http://review.gluster.org/7330
Reviewed-by: Raghavendra G <rgowdapp@redhat.com>
Tested-by: Raghavendra G <rgowdapp@redhat.com>

2 files changed, 184 insertions, 47 deletions

diff --git a/xlators/features/quota/src/quota.c b/xlators/features/quota/src/quota.c
index d93911dee08..913741e0fda 100644
--- a/xlators/features/quota/src/quota.c
+++ b/xlators/features/quota/src/quota.c
@@ -14,6 +14,9 @@
 #include "defaults.h"
 #include "statedump.h"
 
+void
+quota_get_limit_dir (call_frame_t *frame, inode_t *cur_inode, xlator_t *this);
+
 int32_t
 quota_check_limit (call_frame_t *frame, inode_t *inode, xlator_t *this,
                    char *name, uuid_t par);
@@ -178,6 +181,10 @@ quota_local_cleanup (xlator_t *this, quota_local_t *local)
         loc_wipe (&local->validate_loc);
 
         inode_unref (local->inode);
+
+        if (local->xdata)
+                dict_unref (local->xdata);
+
         LOCK_DESTROY (&local->lock);
 
         mem_put (local);
@@ -3516,14 +3523,15 @@ quota_statfs_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                   int32_t op_ret, int32_t op_errno, struct statvfs *buf,
                   dict_t *xdata)
 {
-	inode_t           *inode  = NULL;
-        uint64_t           value  = 0;
-	int64_t            usage  = -1;
-	int64_t            avail  = -1;
-        int64_t            blocks = 0;
-        quota_inode_ctx_t *ctx    = NULL;
-        int                ret    = 0;
+	inode_t           *inode        = NULL;
+        uint64_t           value        = 0;
+	int64_t            usage        = -1;
+	int64_t            avail        = -1;
+        int64_t            blocks       = 0;
+        quota_inode_ctx_t *ctx          = NULL;
+        int                ret          = 0;
         gf_boolean_t       dict_created = _gf_false;
+        quota_local_t     *local        = frame->local;
 
         inode = cookie;
 
@@ -3543,20 +3551,9 @@ quota_statfs_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 	}
 
         inode_ctx_get (inode, this, &value);
-	if (!value) {
-		goto unwind;
-	}
-
-        /* if limit is set on this inode, report statfs based on this inode
-         * else report based on root.
-         */
         ctx = (quota_inode_ctx_t *)(unsigned long)value;
-        if (ctx->hard_lim <= 0) {
-                inode_ctx_get (inode->table->root, this, &value);
-                ctx = (quota_inode_ctx_t *)(unsigned long) value;
-                if (!ctx || ctx->hard_lim < 0)
-                        goto unwind;
-        }
+        if (!ctx || ctx->hard_lim <= 0)
+                goto unwind;
 
         { /* statfs is adjusted in this code block */
                 usage = (ctx->size) / buf->f_bsize;
@@ -3596,6 +3593,7 @@ unwind:
 
         if (dict_created)
                 dict_unref (xdata);
+
         return 0;
 }
 
@@ -3604,22 +3602,23 @@ int32_t
 quota_statfs_helper (call_frame_t *frame, xlator_t *this, loc_t *loc,
                      dict_t *xdata)
 {
-        quota_local_t   *local          = NULL;
+        quota_local_t   *local          = frame->local;
         int              op_errno       = EINVAL;
 
-        GF_VALIDATE_OR_GOTO ("quota", (local = frame->local), err);
+        GF_VALIDATE_OR_GOTO ("quota", local, err);
 
         if (-1 == local->op_ret) {
                 op_errno = local->op_errno;
                 goto err;
         }
 
-        STACK_WIND_COOKIE (frame, quota_statfs_cbk, loc->inode,
+        STACK_WIND_COOKIE (frame, quota_statfs_cbk, local->inode,
                            FIRST_CHILD(this),
                            FIRST_CHILD(this)->fops->statfs, loc, xdata);
         return 0;
 err:
         QUOTA_STACK_UNWIND (statfs, frame, -1, op_errno, NULL, NULL);
+
         return 0;
 }
 
@@ -3678,16 +3677,143 @@ resume:
         return 0;
 }
 
+void
+quota_get_limit_dir_continuation (struct list_head *parents, inode_t *inode,
+                                  int32_t op_ret, int32_t op_errno, void *data)
+{
+        call_frame_t   *frame        = NULL;
+        xlator_t       *this         = NULL;
+        quota_local_t  *local        = NULL;
+        quota_dentry_t *entry        = NULL;
+        inode_t        *parent       = NULL;
+
+        frame = data;
+        local = frame->local;
+        this = THIS;
+
+        if ((op_ret < 0) || list_empty (parents)) {
+                if (op_ret >= 0) {
+                        gf_log (this->name, GF_LOG_WARNING,
+                                "Couldn't build ancestry for inode (gfid:%s). "
+                                "Without knowing ancestors till root, quota "
+                                "cannot be enforced. "
+                                "Hence, failing fop with EIO",
+                                uuid_utoa (inode->gfid));
+                        op_errno = EIO;
+                }
+
+                quota_handle_validate_error (local, -1, op_errno);
+                goto out;
+        }
+
+        entry = list_entry (parents, quota_dentry_t, next);
+        parent = inode_find (inode->table, entry->par);
+
+        quota_get_limit_dir (frame, parent, this);
+
+        inode_unref (parent);
+out:
+        return;
+}
+
+void
+quota_statfs_continue (call_frame_t *frame, xlator_t *this, inode_t *inode)
+{
+        call_stub_t     *stub           = NULL;
+        quota_local_t   *local          = frame->local;
+        int              ret            = -1;
+
+        stub = fop_statfs_stub (frame, quota_statfs_helper,
+                                &local->loc, local->xdata);
+        if (!stub)
+                goto err;
+
+        LOCK (&local->lock);
+        {
+                local->inode = inode_ref (inode);
+                local->link_count = 1;
+                local->stub = stub;
+        }
+        UNLOCK (&local->lock);
+
+        ret = quota_validate (frame, local->inode, this,
+                              quota_statfs_validate_cbk);
+        if (0 > ret)
+                quota_handle_validate_error (local, -1, -ret);
+        return;
+
+err:
+        QUOTA_STACK_UNWIND (statfs, frame, -1, ENOMEM, NULL, NULL);
+
+        return;
+}
+
+void
+quota_get_limit_dir (call_frame_t *frame, inode_t *cur_inode, xlator_t *this)
+{
+        inode_t                *inode           = NULL;
+        inode_t                *parent          = NULL;
+        uint64_t                value           = 0;
+        quota_inode_ctx_t      *ctx             = NULL;
+        int                     ret             = -1;
+        quota_local_t          *local           = frame->local;
+
+        if (!cur_inode)
+                goto out;
+
+        inode = inode_ref (cur_inode);
+        while (inode) {
+                value = 0;
+                inode_ctx_get (inode, this, &value);
+
+                if (value) {
+                        ctx = (quota_inode_ctx_t *)(unsigned long)value;
+                        if (ctx->hard_lim > 0)
+                                break;
+                }
+
+                if (__is_root_gfid (inode->gfid))
+                        goto off;
+
+                parent = inode_parent (inode, 0, NULL);
+                if (!parent) {
+                        ret = quota_build_ancestry
+                               (inode, quota_get_limit_dir_continuation,
+                                (void *)frame);
+                        goto out;
+                }
+
+                inode_unref (inode);
+                inode = parent;
+        }
+
+        quota_statfs_continue (frame, this, inode);
+        inode_unref (inode);
+        return;
+
+off:
+        gf_log (this->name, GF_LOG_DEBUG,
+                "No limit set on the inode or it's parents.");
+
+        QUOTA_STACK_WIND_TAIL (frame, FIRST_CHILD(this),
+                               FIRST_CHILD(this)->fops->statfs,
+                               &local->loc, local->xdata);
+out:
+        inode_unref (inode);
+
+        return;
+}
+
 int32_t
 quota_statfs (call_frame_t *frame, xlator_t *this, loc_t *loc, dict_t *xdata)
 {
-        quota_local_t   *local  = NULL;
+        quota_local_t   *local          = NULL;
         int              op_errno       = 0;
-        call_stub_t     *stub           = NULL;
-        quota_priv_t *priv  = NULL;
-        int           ret       = 0;
+        int              ret            = -1;
+        quota_priv_t    *priv           = NULL;
 
         priv = this->private;
+        GF_ASSERT (loc);
 
         WIND_IF_QUOTAOFF (priv->is_quota_on, off);
 
@@ -3697,27 +3823,21 @@ quota_statfs (call_frame_t *frame, xlator_t *this, loc_t *loc, dict_t *xdata)
                         op_errno = ENOMEM;
                         goto err;
                 }
-                frame->local = local;
 
-                stub = fop_statfs_stub (frame, quota_statfs_helper, loc, xdata);
-                if (!stub) {
+                ret = loc_copy (&local->loc, loc);
+                if (-1 == ret) {
                         op_errno = ENOMEM;
                         goto err;
                 }
 
-                LOCK (&local->lock);
-                {
-                        local->inode = inode_ref (loc->inode);
-                        local->link_count = 1;
-                        local->stub = stub;
-                }
-                UNLOCK (&local->lock);
+                if (xdata)
+                        local->xdata = dict_ref (xdata);
 
-                ret = quota_validate (frame, local->inode, this,
-                                      quota_statfs_validate_cbk);
-                if (0 > ret) {
-                        quota_handle_validate_error (local, -1, -ret);
-                }
+                local->link_count = 1;
+
+                frame->local = local;
+
+                quota_get_limit_dir (frame, loc->inode, this);
 
                 return 0;
 	}
@@ -3733,8 +3853,9 @@ quota_statfs (call_frame_t *frame, xlator_t *this, loc_t *loc, dict_t *xdata)
          * bother calculating quota limit on / in statfs_cbk.
          */
         if (priv->consider_statfs)
-                gf_log (this->name,GF_LOG_WARNING,
-                        "missing inode, cannot adjust for quota");
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Missing inode, can't adjust for quota");
+
 
 off:
         STACK_WIND_TAIL (frame, FIRST_CHILD(this),
@@ -3742,10 +3863,8 @@ off:
         return 0;
 
 err:
-        STACK_UNWIND_STRICT (statfs, frame, -1, op_errno, NULL, NULL);
+        QUOTA_STACK_UNWIND (statfs, frame, -1, op_errno, NULL, NULL);
 
-        if (local)
-                quota_local_cleanup (this, local);
         return 0;
 }
 
diff --git a/xlators/features/quota/src/quota.h b/xlators/features/quota/src/quota.h
index 84c3257feec..5a4bcb2b1e0 100644
--- a/xlators/features/quota/src/quota.h
+++ b/xlators/features/quota/src/quota.h
@@ -80,6 +80,23 @@
                 }                                       \
         } while (0);
 
+#define QUOTA_STACK_WIND_TAIL(frame, params...)                         \
+        do {                                                            \
+                quota_local_t *_local = NULL;                           \
+                xlator_t      *_this  = NULL;                           \
+                                                                        \
+                if (frame) {                                            \
+                        _local = frame->local;                          \
+                        _this  = frame->this;                           \
+                        frame->local = NULL;                            \
+                }                                                       \
+                                                                        \
+                STACK_WIND_TAIL (frame, params);                        \
+                                                                        \
+                if (_local)                                             \
+                        quota_local_cleanup (_this, _local);            \
+        } while (0)
+
 #define QUOTA_STACK_UNWIND(fop, frame, params...)                       \
         do {                                                            \
                 quota_local_t *_local = NULL;                           \
@@ -186,6 +203,7 @@ struct quota_local {
         int64_t                 space_available;
         quota_ancestry_built_t  ancestry_cbk;
         void                   *ancestry_data;
+        dict_t                 *xdata;
 };
 typedef struct quota_local      quota_local_t;