glusterd: Added the validation function for subvols-per-directory

Change-Id: Ie2259023b9001311a2032792639c3093054f6750
BUG: 896431
Signed-off-by: Avra Sengupta <asengupt@redhat.com>
Reviewed-on: http://review.gluster.org/4552
Reviewed-by: Jeff Darcy <jdarcy@redhat.com>
Tested-by: Gluster Build System <jenkins@build.gluster.com>

9 files changed, 214 insertions, 26 deletions

diff --git a/tests/bugs/bug-896431.t b/tests/bugs/bug-896431.t
new file mode 100755
index 000000000..f968e59c1
--- /dev/null
+++ b/tests/bugs/bug-896431.t
@@ -0,0 +1,124 @@
+#!/bin/bash
+
+. $(dirname $0)/../include.rc
+. $(dirname $0)/../volume.rc
+
+cleanup;
+
+## Start and create a volume
+TEST glusterd;
+TEST pidof glusterd;
+TEST $CLI volume create $V0 replica 2 stripe 2 $H0:$B0/${V0}{1,2,3,4,5,6,7,8};
+
+## Verify volume is created
+EXPECT "$V0" volinfo_field $V0 'Volume Name';
+EXPECT 'Created' volinfo_field $V0 'Status';
+
+## Start volume and verify
+TEST $CLI volume start $V0;
+EXPECT 'Started' volinfo_field $V0 'Status';
+
+## Setting cluster.subvols-per-directory as -5
+TEST ! $CLI volume set $V0 cluster.subvols-per-directory -5
+EXPECT '' volinfo_field $V0 'cluster.subvols-per-directory';
+TEST ! $CLI volume set $V0 subvols-per-directory -5
+EXPECT '' volinfo_field $V0 'cluster.subvols-per-directory';
+
+## Setting cluster.subvols-per-directory as 0
+TEST ! $CLI volume set $V0 cluster.subvols-per-directory 0
+EXPECT '' volinfo_field $V0 'cluster.subvols-per-directory';
+TEST ! $CLI volume set $V0 subvols-per-directory 0
+EXPECT '' volinfo_field $V0 'cluster.subvols-per-directory';
+
+## Setting cluster.subvols-per-directory as 4 (the total number of bricks)
+TEST ! $CLI volume set $V0 cluster.subvols-per-directory 4
+EXPECT '' volinfo_field $V0 'cluster.subvols-per-directory';
+TEST ! $CLI volume set $V0 subvols-per-directory 4
+EXPECT '' volinfo_field $V0 'cluster.subvols-per-directory';
+
+## Setting cluster.subvols-per-directory as 2 (the total number of subvolumes)
+TEST $CLI volume set $V0 cluster.subvols-per-directory 2
+EXPECT '2' volinfo_field $V0 'cluster.subvols-per-directory';
+
+## Setting cluster.subvols-per-directory as 1
+TEST $CLI volume set $V0 subvols-per-directory 1
+EXPECT '1' volinfo_field $V0 'cluster.subvols-per-directory';
+
+## Finish up
+TEST $CLI volume stop $V0;
+EXPECT 'Stopped' volinfo_field $V0 'Status';
+
+TEST $CLI volume delete $V0;
+TEST ! $CLI volume info $V0;
+
+cleanup;
+
+## Start and create a pure replicate volume
+TEST glusterd;
+TEST pidof glusterd;
+TEST $CLI volume create $V0 replica 8 $H0:$B0/${V0}{1,2,3,4,5,6,7,8};
+
+## Verify volume is created
+EXPECT "$V0" volinfo_field $V0 'Volume Name';
+EXPECT 'Created' volinfo_field $V0 'Status';
+EXPECT 'Replicate' volinfo_field $V0 'Type';
+
+## Start volume and verify
+TEST $CLI volume start $V0;
+EXPECT 'Started' volinfo_field $V0 'Status';
+
+## Setting cluster.subvols-per-directory as 8 for a replicate volume
+TEST ! $CLI volume set $V0 cluster.subvols-per-directory 8
+EXPECT '' volinfo_field $V0 'cluster.subvols-per-directory';
+TEST ! $CLI volume set $V0 subvols-per-directory 8
+EXPECT '' volinfo_field $V0 'cluster.subvols-per-directory';
+
+## Setting cluster.subvols-per-directory as 1 for a replicate volume
+TEST $CLI volume set $V0 cluster.subvols-per-directory 1
+EXPECT '1' volinfo_field $V0 'cluster.subvols-per-directory';
+TEST $CLI volume set $V0 subvols-per-directory 1
+EXPECT '1' volinfo_field $V0 'cluster.subvols-per-directory';
+
+## Finish up
+TEST $CLI volume stop $V0;
+EXPECT 'Stopped' volinfo_field $V0 'Status';
+
+TEST $CLI volume delete $V0;
+TEST ! $CLI volume info $V0;
+
+cleanup;
+
+## Start and create a pure stripe volume
+TEST glusterd;
+TEST pidof glusterd;
+TEST $CLI volume create $V0 stripe 8 $H0:$B0/${V0}{1,2,3,4,5,6,7,8};
+
+## Verify volume is created
+EXPECT "$V0" volinfo_field $V0 'Volume Name';
+EXPECT 'Created' volinfo_field $V0 'Status';
+EXPECT 'Stripe' volinfo_field $V0 'Type';
+
+## Start volume and verify
+TEST $CLI volume start $V0;
+EXPECT 'Started' volinfo_field $V0 'Status';
+
+## Setting cluster.subvols-per-directory as 8 for a stripe volume
+TEST ! $CLI volume set $V0 cluster.subvols-per-directory 8
+EXPECT '' volinfo_field $V0 'cluster.subvols-per-directory';
+TEST ! $CLI volume set $V0 subvols-per-directory 8
+EXPECT '' volinfo_field $V0 'cluster.subvols-per-directory';
+
+## Setting cluster.subvols-per-directory as 1 for a stripe volume
+TEST $CLI volume set $V0 cluster.subvols-per-directory 1
+EXPECT '1' volinfo_field $V0 'cluster.subvols-per-directory';
+TEST $CLI volume set $V0 subvols-per-directory 1
+EXPECT '1' volinfo_field $V0 'cluster.subvols-per-directory';
+
+## Finish up
+TEST $CLI volume stop $V0;
+EXPECT 'Stopped' volinfo_field $V0 'Status';
+
+TEST $CLI volume delete $V0;
+TEST ! $CLI volume info $V0;
+
+cleanup;
diff --git a/xlators/cluster/dht/src/dht.c b/xlators/cluster/dht/src/dht.c
index 2425341b0..14f3eb1d1 100644
--- a/xlators/cluster/dht/src/dht.c
+++ b/xlators/cluster/dht/src/dht.c
@@ -733,6 +733,8 @@ struct volume_options options[] = {
         },
         { .key  = {"directory-layout-spread"},
           .type = GF_OPTION_TYPE_INT,
+          .min  = 1,
+          .validate = GF_OPT_VALIDATE_MIN,
           .description = "Specifies the directory layout spread."
         },
         { .key  = {"decommissioned-bricks"},
diff --git a/xlators/mgmt/glusterd/src/glusterd-brick-ops.c b/xlators/mgmt/glusterd/src/glusterd-brick-ops.c
index 6a56432d1..2ac2699cf 100644
--- a/xlators/mgmt/glusterd/src/glusterd-brick-ops.c
+++ b/xlators/mgmt/glusterd/src/glusterd-brick-ops.c
@@ -919,13 +919,15 @@ glusterd_op_perform_add_bricks (glusterd_volinfo_t *volinfo, int32_t count,
         if (stripe_count) {
                 volinfo->stripe_count = stripe_count;
         }
-        volinfo->dist_leaf_count = (volinfo->stripe_count *
-                                    volinfo->replica_count);
+        volinfo->dist_leaf_count = glusterd_get_dist_leaf_count (volinfo);
 
         /* backward compatibility */
         volinfo->sub_count = ((volinfo->dist_leaf_count == 1) ? 0:
                               volinfo->dist_leaf_count);
 
+        volinfo->subvol_count = (volinfo->brick_count /
+                                 volinfo->dist_leaf_count);
+
         ret = glusterd_create_volfiles_and_notify_services (volinfo);
         if (ret)
                 goto out;
@@ -1616,8 +1618,10 @@ glusterd_op_remove_brick (dict_t *dict, char **op_errstr)
                         volinfo->replica_count, replica_count,
                         volinfo->volname);
                 volinfo->replica_count = replica_count;
-                volinfo->dist_leaf_count = (volinfo->stripe_count *
-                                            replica_count);
+                volinfo->dist_leaf_count = glusterd_get_dist_leaf_count (volinfo);
+                volinfo->subvol_count = (volinfo->brick_count /
+                                         volinfo->dist_leaf_count);
+
                 if (replica_count == 1) {
                         if (volinfo->type == GF_CLUSTER_TYPE_REPLICATE) {
                                 volinfo->type = GF_CLUSTER_TYPE_NONE;
diff --git a/xlators/mgmt/glusterd/src/glusterd-store.c b/xlators/mgmt/glusterd/src/glusterd-store.c
index 863b70c7c..085e3e85d 100644
--- a/xlators/mgmt/glusterd/src/glusterd-store.c
+++ b/xlators/mgmt/glusterd/src/glusterd-store.c
@@ -2530,8 +2530,11 @@ glusterd_store_retrieve_volume (char    *volname)
                         break;
                 }
 
-                volinfo->dist_leaf_count = (volinfo->stripe_count *
-                                            volinfo->replica_count);
+                volinfo->dist_leaf_count = glusterd_get_dist_leaf_count (volinfo);
+
+                volinfo->subvol_count = (volinfo->brick_count /
+                                         volinfo->dist_leaf_count);
+
         }
 
         if (op_errno != GD_STORE_EOF)
diff --git a/xlators/mgmt/glusterd/src/glusterd-utils.c b/xlators/mgmt/glusterd/src/glusterd-utils.c
index 64f45f247..e151b2f71 100644
--- a/xlators/mgmt/glusterd/src/glusterd-utils.c
+++ b/xlators/mgmt/glusterd/src/glusterd-utils.c
@@ -4106,6 +4106,15 @@ glusterd_restart_gsyncds (glusterd_conf_t *conf)
         return ret;
 }
 
+inline int
+glusterd_get_dist_leaf_count (glusterd_volinfo_t *volinfo)
+{
+    int rcount = volinfo->replica_count;
+    int scount = volinfo->stripe_count;
+
+    return (rcount ? rcount : 1) * (scount ? scount : 1);
+}
+
 int
 glusterd_get_brickinfo (xlator_t *this, const char *brickname, int port,
                         gf_boolean_t localhost, glusterd_brickinfo_t **brickinfo)
diff --git a/xlators/mgmt/glusterd/src/glusterd-utils.h b/xlators/mgmt/glusterd/src/glusterd-utils.h
index b6a8675b2..c7fbe447e 100644
--- a/xlators/mgmt/glusterd/src/glusterd-utils.h
+++ b/xlators/mgmt/glusterd/src/glusterd-utils.h
@@ -424,6 +424,9 @@ glusterd_add_node_to_dict (char *server, dict_t *dict, int count,
 char *
 glusterd_uuid_to_hostname (uuid_t uuid);
 
+int
+glusterd_get_dist_leaf_count (glusterd_volinfo_t *volinfo);
+
 glusterd_brickinfo_t*
 glusterd_get_brickinfo_by_position (glusterd_volinfo_t *volinfo, uint32_t pos);
 
diff --git a/xlators/mgmt/glusterd/src/glusterd-volume-ops.c b/xlators/mgmt/glusterd/src/glusterd-volume-ops.c
index ac8925e54..bc5f8f2a7 100644
--- a/xlators/mgmt/glusterd/src/glusterd-volume-ops.c
+++ b/xlators/mgmt/glusterd/src/glusterd-volume-ops.c
@@ -1504,8 +1504,12 @@ glusterd_op_create_volume (dict_t *dict, char **op_errstr)
 
         /* dist-leaf-count is the count of brick nodes for a given
            subvolume of distribute */
-        volinfo->dist_leaf_count = (volinfo->stripe_count *
-                                    volinfo->replica_count);
+        volinfo->dist_leaf_count = glusterd_get_dist_leaf_count (volinfo);
+
+        /* subvol_count is the count of number of subvolumes present
+           for a given distribute volume */
+        volinfo->subvol_count = (volinfo->brick_count /
+                                 volinfo->dist_leaf_count);
 
         /* Keep sub-count same as earlier, for the sake of backward
            compatibility */
diff --git a/xlators/mgmt/glusterd/src/glusterd-volume-set.c b/xlators/mgmt/glusterd/src/glusterd-volume-set.c
index a5efb55f8..49d2158ff 100644
--- a/xlators/mgmt/glusterd/src/glusterd-volume-set.c
+++ b/xlators/mgmt/glusterd/src/glusterd-volume-set.c
@@ -19,7 +19,6 @@
 static int
 check_dict_key_value (dict_t *dict, char *key, char *value)
 {
-        char                 errstr[2048]  = "";
         glusterd_conf_t     *priv          = NULL;
         int                  ret           = 0;
         xlator_t            *this          = NULL;
@@ -30,25 +29,19 @@ check_dict_key_value (dict_t *dict, char *key, char *value)
         GF_ASSERT (priv);
 
         if (!dict) {
-                snprintf (errstr, 2048, "Received Empty Dict.");
-                gf_log (this->name, GF_LOG_ERROR,
-                        "%s.", errstr);
+                gf_log (this->name, GF_LOG_ERROR, "Received Empty Dict.");
                 ret = -1;
                 goto out;
         }
 
         if (!key) {
-                snprintf (errstr, 2048, "Received Empty Key.");
-                gf_log (this->name, GF_LOG_ERROR,
-                        "%s.", errstr);
+                gf_log (this->name, GF_LOG_ERROR, "Received Empty Key.");
                 ret = -1;
                 goto out;
         }
 
         if (!value) {
-                snprintf (errstr, 2048, "Received Empty Value.");
-                gf_log (this->name, GF_LOG_ERROR,
-                        "%s.", errstr);
+                gf_log (this->name, GF_LOG_ERROR, "Received Empty Value.");
                 ret = -1;
                 goto out;
         }
@@ -62,7 +55,6 @@ out:
 static int
 get_volname_volinfo (dict_t *dict, char **volname, glusterd_volinfo_t **volinfo)
 {
-        char                 errstr[2048]  = "";
         glusterd_conf_t     *priv          = NULL;
         int                  ret           = 0;
         xlator_t            *this          = NULL;
@@ -74,17 +66,13 @@ get_volname_volinfo (dict_t *dict, char **volname, glusterd_volinfo_t **volinfo)
 
         ret = dict_get_str (dict, "volname", volname);
         if (ret) {
-                snprintf (errstr, 2048, "Unable to get volume name");
-                gf_log (this->name, GF_LOG_ERROR,
-                        "%s.", errstr);
+                gf_log (this->name, GF_LOG_ERROR, "Unable to get volume name");
                 goto out;
         }
 
         ret = glusterd_volinfo_find (*volname, volinfo);
         if (ret) {
-                snprintf (errstr, 2048, "Unable to allocate memory");
-                gf_log (this->name, GF_LOG_ERROR,
-                        "%s.", errstr);
+                gf_log (this->name, GF_LOG_ERROR, "Unable to allocate memory");
                 goto out;
         }
 
@@ -246,6 +234,54 @@ out:
         return ret;
 }
 
+static int
+validate_subvols_per_directory (dict_t *dict, char *key, char *value,
+                                char **op_errstr)
+{
+        char                 errstr[2048]  = "";
+        char                *volname       = NULL;
+        glusterd_conf_t     *priv          = NULL;
+        glusterd_volinfo_t  *volinfo       = NULL;
+        int                  ret           = 0;
+        int                  subvols       = 0;
+        xlator_t            *this          = NULL;
+
+        this = THIS;
+        GF_ASSERT (this);
+        priv = this->private;
+        GF_ASSERT (priv);
+
+        ret = check_dict_key_value (dict, key, value);
+        if (ret)
+                goto out;
+
+        ret = get_volname_volinfo (dict, &volname, &volinfo);
+        if (ret)
+                goto out;
+
+        subvols = atoi(value);
+
+        /* Checking if the subvols-per-directory exceed the total
+           number of subvolumes. */
+        if (subvols > volinfo->subvol_count) {
+                snprintf (errstr, sizeof(errstr),
+                          "subvols-per-directory(%d) is greater "
+                          "than the number of subvolumes(%d).",
+                          subvols, volinfo->subvol_count);
+                gf_log (this->name, GF_LOG_ERROR,
+                        "%s.", errstr);
+                *op_errstr = gf_strdup (errstr);
+                ret = -1;
+                goto out;
+        }
+
+out:
+        gf_log (this->name, GF_LOG_DEBUG, "Returning %d", ret);
+
+        return ret;
+}
+
+
 /* dispatch table for VOLUME SET
  * -----------------------------
  *
@@ -330,7 +366,8 @@ struct volopt_map_entry glusterd_volopt_map[] = {
         { .key         = "cluster.subvols-per-directory",
           .voltype     = "cluster/distribute",
           .option      = "directory-layout-spread",
-          .op_version  = 2
+          .op_version  = 2,
+          .validate_fn = validate_subvols_per_directory
         },
         { .key         = "cluster.readdir-optimize",
           .voltype     = "cluster/distribute",
diff --git a/xlators/mgmt/glusterd/src/glusterd.h b/xlators/mgmt/glusterd/src/glusterd.h
index a2067250c..c9e8d42d3 100644
--- a/xlators/mgmt/glusterd/src/glusterd.h
+++ b/xlators/mgmt/glusterd/src/glusterd.h
@@ -268,6 +268,8 @@ struct glusterd_volinfo_ {
         int                     sub_count;  /* backward compatibility */
         int                     stripe_count;
         int                     replica_count;
+        int                     subvol_count; /* Number of subvolumes in a
+                                                 distribute volume */
         int                     dist_leaf_count; /* Number of bricks in one
                                                     distribute subvolume */
         int                     port;