quota: For a link operation, do quota_check_limit only till the

common ancestor of src and dst file

In a dht_rename, if src_cached and dst_hashed are different, then
rename is split into link and unlink.
We need to handle quota_link properly.

We have fixed quota_rename in patch# 8940, we need to handle quota_link
similarly

> http://review.gluster.org/#/c/8940/
> quota: For a rename operation, do quota_check_limit only till the
> common ancestor of src and dst file

> Example:
> set quota limit set to 1GB on /
> create a file /a1/b1/file1 of 600MB
> mv /a1/b1/file1 /a1/b1/file2
> This rename fails as it takes delta into account which sums up to 1.2BG.
> Though we are not creating new file, we still get quota exceeded error.
> So quota enforce should happen only till b1.

> Similarly:
> mv /a/b/c/file /a/b/x/y/file
> quota enforce should happen only till dir 'b'

> Change-Id: Ia1e5363da876c3d71bd424e67a8bb28b7ac1c7c1
> BUG: 1153964
> Signed-off-by: vmallika <vmallika@redhat.com>
> Reviewed-on: http://review.gluster.org/8940
> Tested-by: Gluster Build System <jenkins@build.gluster.com>
> Reviewed-by: Raghavendra G <rgowdapp@redhat.com>
> Tested-by: Raghavendra G <rgowdapp@redhat.com>

Change-Id: I2c814018d17f7af1807c1d1d162d8bdcbb31e491
BUG: 1153964
Signed-off-by: vmallika <vmallika@redhat.com>
Reviewed-on: http://review.gluster.org/9419
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Raghavendra G <rgowdapp@redhat.com>
Tested-by: Raghavendra G <rgowdapp@redhat.com>

3 files changed, 226 insertions, 41 deletions

diff --git a/tests/bugs/quota/bug-1153964.t b/tests/bugs/quota/bug-1153964.t
new file mode 100644
index 00000000000..c923b71ca73
--- /dev/null
+++ b/tests/bugs/quota/bug-1153964.t
@@ -0,0 +1,83 @@
+#!/bin/bash
+
+. $(dirname $0)/../../include.rc
+. $(dirname $0)/../../volume.rc
+. $(dirname $0)/../../nfs.rc
+
+function rename_loop()
+{
+        local i=0
+        local limit=$1
+        while [ $i -lt $limit ]
+        do
+                j=$[$i + 1]
+                mv $N0/test_dir/file$i $N0/test_dir/file$j
+                if [ "$?" != "0" ]
+                then
+                        return 1
+                fi
+                i=$[$i + 1]
+        done
+        return 0
+}
+
+function createFile_and_checkLimit()
+{
+        local count_val=$1;
+        dd if=/dev/zero of="$N0/test_dir/file0" bs=1048576 count=$count_val
+        sleep 3
+        if [ -f $N0/test_dir/file0 ]
+        then
+                rename_loop 10
+                if [ "$?" == "0" ]
+                then
+                        echo "Y"
+                else
+                        echo "N"
+                fi
+        fi
+}
+
+cleanup;
+
+TEST glusterd
+TEST pidof glusterd
+
+TEST $CLI volume create $V0 $H0:$B0/${V0}1 $H0:$B0/${V0}2
+EXPECT 'Created' volinfo_field $V0 'Status'
+TEST $CLI volume start $V0
+EXPECT 'Started' volinfo_field $V0 'Status'
+
+TEST $CLI volume quota $V0 enable
+EXPECT 'on' volinfo_field $V0 'features.quota'
+
+EXPECT_WITHIN $NFS_EXPORT_TIMEOUT "1" is_nfs_export_available;
+TEST mount_nfs $H0:/$V0 $N0 nolock;
+TEST mkdir -p $N0/test_dir/
+
+# Try to rename file under various case and check if
+# quota limit exceeds or not.
+TEST $CLI volume quota $V0 limit-usage /test_dir 100MB
+# Case1 : If used size is less than hard-limit size
+# Create a 600MB file
+EXPECT 'Y' createFile_and_checkLimit 60
+
+TEST rm -rf $N0/test_dir/*
+# Case2 : If used size is equal to hard-limit size
+# Create a 100MB file
+EXPECT 'Y' createFile_and_checkLimit 100
+
+TEST rm -rf $N0/test_dir/*
+# Case3 : If used size is greater than hard-limit size
+# Create a 110MB file
+EXPECT 'Y' createFile_and_checkLimit 110
+
+# remove this directory as it has been created as part
+# of above testcase
+TEST rm -rf $N0/test_dir/
+
+EXPECT_WITHIN $UMOUNT_TIMEOUT "Y" force_umount $N0
+
+cleanup;
+
+
diff --git a/xlators/features/quota/src/quota.c b/xlators/features/quota/src/quota.c
index f903b4e57b7..245880c271c 100644
--- a/xlators/features/quota/src/quota.c
+++ b/xlators/features/quota/src/quota.c
@@ -174,6 +174,9 @@ quota_local_cleanup (xlator_t *this, quota_local_t *local)
         if (local->xdata)
                 dict_unref (local->xdata);
 
+        if (local->stub)
+                call_stub_destroy (local->stub);
+
         LOCK_DESTROY (&local->lock);
 
         mem_put (local);
@@ -345,17 +348,15 @@ check_ancestory_continue (struct list_head *parents, inode_t *inode,
         frame = data;
         local = frame->local;
 
-        if (op_ret < 0 || (parents && 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;
-                        op_ret = -1;
-                }
+        if (parents && list_empty (parents)) {
+                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;
+                op_ret = -1;
         }
 
         LOCK (&local->lock);
@@ -1699,6 +1700,7 @@ quota_create (call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t flags,
         priv = this->private;
 
         WIND_IF_QUOTAOFF (priv->is_quota_on, off);
+        QUOTA_WIND_FOR_INTERNAL_FOP (xdata, off);
 
         local = quota_local_new ();
         if (local == NULL) {
@@ -1813,10 +1815,6 @@ quota_unlink (call_frame_t *frame, xlator_t *this, loc_t *loc, int xflag,
 
         frame->local = local;
 
-        if (xdata && dict_get (xdata, GLUSTERFS_INTERNAL_FOP_KEY)) {
-                local->skip_check = _gf_true;
-        }
-
         ret = loc_copy (&local->loc, loc);
         if (ret) {
                 gf_log (this->name, GF_LOG_WARNING, "loc_copy failed");
@@ -1860,9 +1858,6 @@ quota_link_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 
         local = (quota_local_t *) frame->local;
 
-        if (local->skip_check)
-                goto out;
-
         ret = quota_inode_ctx_get (inode, this, &ctx, 0);
         if ((ret == -1) || (ctx == NULL)) {
                 gf_log (this->name, GF_LOG_DEBUG, "quota context is NULL on "
@@ -1950,32 +1945,120 @@ unwind:
         return 0;
 }
 
-
-int32_t
-quota_link (call_frame_t *frame, xlator_t *this, loc_t *oldloc, loc_t *newloc,
-            dict_t *xdata)
+void
+quota_link_continue (call_frame_t *frame)
 {
-        quota_priv_t      *priv  = NULL;
-        int32_t            ret   = -1, op_errno = ENOMEM;
-        quota_local_t     *local = NULL;
-        quota_inode_ctx_t *ctx   = NULL;
-        call_stub_t       *stub  = NULL;
+        int32_t            ret               = -1;
+        int32_t            op_errno          = EIO;
+        quota_local_t     *local             = NULL;
+        uuid_t             common_ancestor   = {0};
+        xlator_t          *this              = NULL;
+        quota_inode_ctx_t *ctx               = NULL;
+        inode_t           *src_parent        = NULL;
+        inode_t           *dst_parent        = NULL;
 
-        priv = this->private;
+        local = frame->local;
+        this = THIS;
 
-        WIND_IF_QUOTAOFF (priv->is_quota_on, off);
+        if (local->op_ret < 0) {
+                op_errno = local->op_errno;
+                goto err;
+        }
 
-        if (xdata && dict_get (xdata, GLUSTERFS_INTERNAL_FOP_KEY)) {
-                goto off;
+        if (local->xdata &&
+            dict_get (local->xdata, GLUSTERFS_INTERNAL_FOP_KEY)) {
+                /* Treat link as rename, crawl upwards only till common ancestor
+                 */
+                ret = quota_find_common_ancestor (local->oldloc.inode,
+                                                  local->newloc.parent,
+                                                  &common_ancestor);
+                if (ret < 0 || uuid_is_null(common_ancestor)) {
+                        gf_log (this->name, GF_LOG_ERROR, "failed to get "
+                                "common_ancestor for %s and %s",
+                                local->oldloc.path, local->newloc.path);
+                        op_errno = ESTALE;
+                        goto err;
+                }
+        } else {
+                /* Treat link as a new file.
+                 * TODO: Currently marker accounts twice for the links created
+                 * across directories.
+                 * This needs re-vist if marker accounts only once
+                 * for the links created across directories
+                 */
+                src_parent = inode_parent (local->oldloc.inode, 0, NULL);
+                dst_parent = inode_parent (local->newloc.inode, 0, NULL);
+
+                /* No need to check quota limit if src and dst parents are same
+                 */
+                if (src_parent == dst_parent ||
+                    uuid_compare (src_parent->gfid, dst_parent->gfid) == 0) {
+                        inode_unref (src_parent);
+                        inode_unref (dst_parent);
+                        goto off;
+                }
+
+                inode_unref (src_parent);
+                inode_unref (dst_parent);
         }
 
-        quota_inode_ctx_get (oldloc->inode, this, &ctx, 0);
+        quota_inode_ctx_get (local->oldloc.inode, this, &ctx, 0);
         if (ctx == NULL) {
                 gf_log (this->name, GF_LOG_DEBUG, "quota context is NULL on "
                         "inode (%s). "
                         "If quota is not enabled recently and crawler has "
                         "finished crawling, its an error",
-                        uuid_utoa (oldloc->inode->gfid));
+                        uuid_utoa (local->oldloc.inode->gfid));
+        }
+
+        LOCK (&local->lock);
+        {
+                local->link_count = 1;
+                local->delta = (ctx != NULL) ? ctx->buf.ia_blocks * 512 : 0;
+                uuid_copy (local->common_ancestor, common_ancestor);
+        }
+        UNLOCK (&local->lock);
+
+        quota_check_limit (frame, local->newloc.parent, this, NULL, NULL);
+        return;
+
+err:
+        QUOTA_STACK_UNWIND (link, frame, -1, op_errno, NULL, NULL,
+                            NULL, NULL, NULL);
+        return;
+
+off:
+        frame->local = NULL;
+
+        STACK_WIND_TAIL (frame, FIRST_CHILD(this),
+                         FIRST_CHILD(this)->fops->link, &(local->oldloc),
+                         &(local->newloc), local->xdata);
+
+        quota_local_cleanup (this, local);
+        return;
+}
+
+int32_t
+quota_link (call_frame_t *frame, xlator_t *this, loc_t *oldloc, loc_t *newloc,
+            dict_t *xdata)
+{
+        quota_priv_t      *priv     = NULL;
+        int32_t            ret      = -1;
+        int32_t            op_errno = ENOMEM;
+        quota_local_t     *local    = NULL;
+        call_stub_t       *stub     = NULL;
+
+        priv = this->private;
+
+        WIND_IF_QUOTAOFF (priv->is_quota_on, off);
+
+        /* No need to check quota limit if src and dst parents are same */
+        if (oldloc->parent && newloc->parent &&
+            !uuid_compare(oldloc->parent->gfid, newloc->parent->gfid)) {
+                gf_log (this->name, GF_LOG_DEBUG, "link %s -> %s are "
+                        "in the same directory, so skip check limit",
+                        oldloc->path, newloc->path);
+                goto off;
         }
 
         local = quota_local_new ();
@@ -1985,6 +2068,8 @@ quota_link (call_frame_t *frame, xlator_t *this, loc_t *oldloc, loc_t *newloc,
 
         frame->local = (void *) local;
 
+        if (xdata)
+                local->xdata = dict_ref (xdata);
 
         ret = loc_copy (&local->loc, newloc);
         if (ret == -1) {
@@ -1992,6 +2077,18 @@ quota_link (call_frame_t *frame, xlator_t *this, loc_t *oldloc, loc_t *newloc,
                 goto err;
         }
 
+        ret = loc_copy (&local->oldloc, oldloc);
+        if (ret < 0) {
+                gf_log (this->name, GF_LOG_WARNING, "loc_copy failed");
+                goto err;
+        }
+
+        ret = loc_copy (&local->newloc, newloc);
+        if (ret < 0) {
+                gf_log (this->name, GF_LOG_WARNING, "loc_copy failed");
+                goto err;
+        }
+
         stub = fop_link_stub (frame, quota_link_helper, oldloc, newloc, xdata);
         if (stub == NULL) {
                 goto err;
@@ -1999,13 +2096,16 @@ quota_link (call_frame_t *frame, xlator_t *this, loc_t *oldloc, loc_t *newloc,
 
         LOCK (&local->lock);
         {
-                local->link_count = 1;
+                local->link_count = 2;
+                local->fop_continue_cbk = quota_link_continue;
                 local->stub = stub;
-                local->delta = (ctx != NULL) ? ctx->buf.ia_blocks * 512 : 0;
         }
         UNLOCK (&local->lock);
 
-        quota_check_limit (frame, newloc->parent, this, NULL, NULL);
+        check_ancestory (frame, newloc->parent);
+
+        /* source parent can be NULL, so do check_ancestory on a file */
+        check_ancestory (frame, oldloc->inode);
         return 0;
 
 err:
@@ -2277,9 +2377,6 @@ quota_rename_continue (call_frame_t *frame)
         return;
 
 err:
-        if (local && local->stub)
-                call_stub_destroy (local->stub);
-
         QUOTA_STACK_UNWIND (rename, frame, -1, op_errno, NULL,
                             NULL, NULL, NULL, NULL, NULL);
         return;
@@ -2295,7 +2392,6 @@ quota_rename (call_frame_t *frame, xlator_t *this, loc_t *oldloc,
         int32_t            op_errno          = ENOMEM;
         quota_local_t     *local             = NULL;
         call_stub_t       *stub              = NULL;
-        uuid_t             common_ancestor   = {0};
 
         priv = this->private;
 
@@ -3427,6 +3523,7 @@ quota_mknod (call_frame_t *frame, xlator_t *this, loc_t *loc, mode_t mode,
         priv = this->private;
 
         WIND_IF_QUOTAOFF (priv->is_quota_on, off);
+        QUOTA_WIND_FOR_INTERNAL_FOP (xdata, off);
 
         local = quota_local_new ();
         if (local == NULL) {
diff --git a/xlators/features/quota/src/quota.h b/xlators/features/quota/src/quota.h
index 3d6c65f8fb6..f21aed6c38f 100644
--- a/xlators/features/quota/src/quota.h
+++ b/xlators/features/quota/src/quota.h
@@ -51,6 +51,12 @@
         if (!is_quota_on)                       \
                 goto label;
 
+#define QUOTA_WIND_FOR_INTERNAL_FOP(xdata, label)                          \
+        do {                                                               \
+                if (xdata && dict_get (xdata, GLUSTERFS_INTERNAL_FOP_KEY)) \
+                goto label;                                                \
+        } while (0)
+
 #define DID_REACH_LIMIT(lim, prev_size, cur_size)               \
         ((cur_size) >= (lim) && (prev_size) < (lim))
 
@@ -196,7 +202,6 @@ struct quota_local {
         int32_t                 op_ret;
         int32_t                 op_errno;
         int64_t                 size;
-        gf_boolean_t            skip_check;
         char                    just_validated;
         fop_lookup_cbk_t        validate_cbk;
         quota_fop_continue_t    fop_continue_cbk;