glusterd / cli: mount-broker service

Mountbroker is configured in glusterd volfile through a DSL
which is restriced enough to be able to appear in the role
of the value of a volfile knob.

Basically the DSL describes set-theorical requirements
against the option set which is sent by the cli (in the
hope of getting a mount with these options).

If the requirements meet and the volume id and the uid
who is to "own" the mount can be unambigously deduced from
the given request, glusterd does the mount with the given
parameters.

The use case of geo-replication is sugared by means of volume
options which then generate a complete mount-broker option set.

Demo:

- add the following option to your glusterd volfile:

    option mountbroker-root /tmp/mbr
    option mountbroker.fool EQL(volfile-id=pop*|user-map-root=*|volfile-server=localhost)&MEET(user-map-root=john|user-map-root=jane)

- before starting glusterd, create /tmp/mbr owned by root with mode 0755

- with cli, do

   $ gluster system:: mount fool volfile-id=pop33 user-map-root=jane volfile-server=localhost

- on succesful completion (volume pop33 exists and is started, jane is a valid username),
  the mount path will be echoed to you

- you can get rid of the mount by

   $ gluster system:: umount <mount-path>

Change-Id: I629cf64add0a45500d05becc3316f67cdb5b42ff
BUG: 3482
Reviewed-on: http://review.gluster.com/128
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Vijay Bellur <vijay@gluster.com>

7 files changed, 1027 insertions, 3 deletions

diff --git a/xlators/mgmt/glusterd/src/Makefile.am b/xlators/mgmt/glusterd/src/Makefile.am
index 01fbbd064..95a9e7dd4 100644
--- a/xlators/mgmt/glusterd/src/Makefile.am
+++ b/xlators/mgmt/glusterd/src/Makefile.am
@@ -5,14 +5,15 @@ glusterd_la_SOURCES = glusterd.c glusterd-handler.c glusterd-sm.c glusterd-op-sm
 	glusterd-utils.c glusterd-rpc-ops.c glusterd-store.c glusterd-handshake.c \
 	glusterd-pmap.c glusterd-volgen.c glusterd-rebalance.c glusterd-quota.c \
 	glusterd-geo-rep.c glusterd-replace-brick.c glusterd-log-ops.c \
-	glusterd-volume-ops.c glusterd-brick-ops.c
+	glusterd-volume-ops.c glusterd-brick-ops.c glusterd-mountbroker.c
 
 glusterd_la_LIBADD = $(top_builddir)/libglusterfs/src/libglusterfs.la \
 		     $(top_builddir)/rpc/xdr/src/libgfxdr.la \
 		     $(top_builddir)/rpc/rpc-lib/src/libgfrpc.la
 
 noinst_HEADERS = glusterd.h glusterd-utils.h glusterd-op-sm.h glusterd-sm.h \
-	glusterd-store.h glusterd-mem-types.h glusterd-pmap.h glusterd-volgen.h
+	glusterd-store.h glusterd-mem-types.h glusterd-pmap.h glusterd-volgen.h \
+        glusterd-mountbroker.h
 
 AM_CFLAGS = -fPIC -D_FILE_OFFSET_BITS=64 -D_GNU_SOURCE -Wall -D$(GF_HOST_OS)\
 	-I$(top_srcdir)/libglusterfs/src -shared -nostartfiles $(GF_CFLAGS)\
diff --git a/xlators/mgmt/glusterd/src/glusterd-handler.c b/xlators/mgmt/glusterd/src/glusterd-handler.c
index 0547694ca..2f486dac6 100644
--- a/xlators/mgmt/glusterd/src/glusterd-handler.c
+++ b/xlators/mgmt/glusterd/src/glusterd-handler.c
@@ -36,6 +36,7 @@
 #include "compat.h"
 #include "compat-errno.h"
 #include "statedump.h"
+#include "run.h"
 #include "glusterd-mem-types.h"
 #include "glusterd.h"
 #include "glusterd-sm.h"
@@ -48,6 +49,7 @@
 #include "xdr-generic.h"
 #include "rpc-clnt.h"
 #include "glusterd-volgen.h"
+#include "glusterd-mountbroker.h"
 
 #include <sys/resource.h>
 #include <inttypes.h>
@@ -1774,6 +1776,151 @@ glusterd_handle_getwd (rpcsvc_request_t *req)
 
 
 int
+glusterd_handle_mount (rpcsvc_request_t *req)
+{
+        gf1_cli_mount_req mnt_req = {0,};
+        gf1_cli_mount_rsp rsp     = {0,};
+        dict_t *dict              = NULL;
+        int ret                   = 0;
+
+        GF_ASSERT (req);
+
+        if (!xdr_to_generic (req->msg[0], &mnt_req, (xdrproc_t)xdr_gf1_cli_mount_req)) {
+                //failed to decode msg;
+                req->rpc_err = GARBAGE_ARGS;
+                rsp.op_ret = -1;
+                rsp.op_errno = EINVAL;
+                goto out;
+        }
+
+        gf_log ("glusterd", GF_LOG_INFO, "Received mount req");
+
+        if (mnt_req.dict.dict_len) {
+                /* Unserialize the dictionary */
+                dict  = dict_new ();
+
+                ret = dict_unserialize (mnt_req.dict.dict_val,
+                                        mnt_req.dict.dict_len,
+                                        &dict);
+                if (ret < 0) {
+                        gf_log ("glusterd", GF_LOG_ERROR,
+                                "failed to "
+                                "unserialize req-buffer to dictionary");
+                        rsp.op_ret = -1;
+                        rsp.op_errno = -EINVAL;
+                        goto out;
+                } else {
+                        dict->extra_stdfree = mnt_req.dict.dict_val;
+                }
+        }
+
+        rsp.op_ret = glusterd_do_mount (mnt_req.label, dict,
+                                        &rsp.path, &rsp.op_errno);
+
+ out:
+        if (!rsp.path)
+                rsp.path = "";
+
+        ret = glusterd_submit_reply (req, &rsp, NULL, 0, NULL,
+                                     (xdrproc_t)xdr_gf1_cli_mount_rsp);
+
+        if (dict)
+                dict_unref (dict);
+        if (*rsp.path)
+                GF_FREE (rsp.path);
+
+        glusterd_friend_sm ();
+        glusterd_op_sm ();
+
+        return ret;
+}
+
+int
+glusterd_handle_umount (rpcsvc_request_t *req)
+{
+        gf1_cli_umount_req umnt_req = {0,};
+        gf1_cli_umount_rsp rsp      = {0,};
+        char *mountbroker_root      = NULL;
+        char mntp[PATH_MAX]         = {0,};
+        char *path                  = NULL;
+        runner_t runner             = {0,};
+        int ret                     = 0;
+        xlator_t *this              = THIS;
+        gf_boolean_t dir_ok         = _gf_false;
+        char *pdir                  = NULL;
+        char *t                     = NULL;
+
+        GF_ASSERT (req);
+        GF_ASSERT (this);
+
+        if (!xdr_to_generic (req->msg[0], &umnt_req, (xdrproc_t)xdr_gf1_cli_umount_req)) {
+                //failed to decode msg;
+                req->rpc_err = GARBAGE_ARGS;
+                rsp.op_ret = -1;
+                goto out;
+        }
+
+        gf_log ("glusterd", GF_LOG_INFO, "Received umount req");
+
+        if (dict_get_str (this->options, "mountbroker-root",
+                          &mountbroker_root) != 0) {
+                rsp.op_errno = ENOENT;
+                goto out;
+        }
+
+        /* check if it is allowed to umount path */
+        path = gf_strdup (umnt_req.path);
+        if (!path) {
+                rsp.op_errno = ENOMEM;
+                goto out;
+        }
+        dir_ok = _gf_false;
+        pdir = dirname (path);
+        t = strtail (pdir, mountbroker_root);
+        if (t && *t == '/') {
+                t = strtail(++t, MB_HIVE);
+                if (t && !*t)
+                        dir_ok = _gf_true;
+        }
+        GF_FREE (path);
+        if (!dir_ok) {
+                rsp.op_errno = EACCES;
+                goto out;
+        }
+
+        runinit (&runner);
+        runner_add_args (&runner, "umount", umnt_req.path, NULL);
+        if (umnt_req.lazy)
+                runner_add_arg (&runner, "-l");
+        rsp.op_ret = runner_run (&runner);
+        if (rsp.op_ret == 0) {
+                if (realpath (umnt_req.path, mntp))
+                        rmdir (mntp);
+                else {
+                        rsp.op_ret = -1;
+                        rsp.op_errno = errno;
+                }
+                if (unlink (umnt_req.path) != 0) {
+                        rsp.op_ret = -1;
+                        rsp.op_errno = errno;
+                }
+        }
+
+ out:
+        if (rsp.op_errno)
+                rsp.op_ret = -1;
+
+        ret = glusterd_submit_reply (req, &rsp, NULL, 0, NULL,
+                                     (xdrproc_t)xdr_gf1_cli_umount_rsp);
+
+        glusterd_friend_sm ();
+        glusterd_op_sm ();
+
+        return ret;
+}
+
+
+int
 glusterd_friend_remove (uuid_t uuid, char *hostname)
 {
         int                           ret = 0;
@@ -2593,6 +2740,8 @@ rpcsvc_actor_t gd_svc_cli_actors[] = {
         [GLUSTER_CLI_LOG_LEVEL]     = {"LOG_LEVEL", GLUSTER_CLI_LOG_LEVEL, glusterd_handle_log_level, NULL, NULL},
         [GLUSTER_CLI_GETWD]         = { "GETWD", GLUSTER_CLI_GETWD, glusterd_handle_getwd, NULL, NULL},
         [GLUSTER_CLI_STATUS_VOLUME]  = {"STATUS_VOLUME", GLUSTER_CLI_STATUS_VOLUME, glusterd_handle_status_volume, NULL, NULL},
+        [GLUSTER_CLI_MOUNT]         = { "MOUNT", GLUSTER_CLI_MOUNT, glusterd_handle_mount, NULL, NULL},
+        [GLUSTER_CLI_UMOUNT]        = { "UMOUNT", GLUSTER_CLI_UMOUNT, glusterd_handle_umount, NULL, NULL},
 
 };
 
diff --git a/xlators/mgmt/glusterd/src/glusterd-mem-types.h b/xlators/mgmt/glusterd/src/glusterd-mem-types.h
index 3467f9d0a..e10cf1aca 100644
--- a/xlators/mgmt/glusterd/src/glusterd-mem-types.h
+++ b/xlators/mgmt/glusterd/src/glusterd-mem-types.h
@@ -66,7 +66,11 @@ typedef enum gf_gld_mem_types_ {
         gf_gld_mt_op_allack_ctx_t               = gf_common_mt_end + 40,
         gf_gld_mt_linearr                       = gf_common_mt_end + 41,
         gf_gld_mt_linebuf                       = gf_common_mt_end + 42,
-        gf_gld_mt_end                           = gf_common_mt_end + 43
+        gf_gld_mt_mount_pattern                 = gf_common_mt_end + 43,
+        gf_gld_mt_mount_comp_container          = gf_common_mt_end + 44,
+        gf_gld_mt_mount_component               = gf_common_mt_end + 45,
+        gf_gld_mt_mount_spec                    = gf_common_mt_end + 46,
+        gf_gld_mt_end                           = gf_common_mt_end + 47,
 } gf_gld_mem_types_t;
 #endif
 
diff --git a/xlators/mgmt/glusterd/src/glusterd-mountbroker.c b/xlators/mgmt/glusterd/src/glusterd-mountbroker.c
new file mode 100644
index 000000000..03767ee6d
--- /dev/null
+++ b/xlators/mgmt/glusterd/src/glusterd-mountbroker.c
@@ -0,0 +1,647 @@
+/*
+  Copyright (c) 2011 Gluster, Inc. <http://www.gluster.com>
+  This file is part of GlusterFS.
+
+  GlusterFS is free software; you can redistribute it and/or modify
+  it under the terms of the GNU General Public License as published
+  by the Free Software Foundation; either version 3 of the License,
+  or (at your option) any later version.
+
+  GlusterFS is distributed in the hope that it will be useful, but
+  WITHOUT ANY WARRANTY; without even the implied warranty of
+  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+  General Public License for more details.
+
+  You should have received a copy of the GNU General Public License
+  along with this program.  If not, see
+  <http://www.gnu.org/licenses/>.
+*/
+
+#ifndef _CONFIG_H
+#define _CONFIG_H
+#include "config.h"
+#endif
+#include <inttypes.h>
+#include <fnmatch.h>
+#include <pwd.h>
+
+#include "globals.h"
+#include "glusterfs.h"
+#include "compat.h"
+#include "dict.h"
+#include "list.h"
+#include "logging.h"
+#include "defaults.h"
+#include "compat.h"
+#include "compat-errno.h"
+#include "run.h"
+#include "glusterd-mem-types.h"
+#include "glusterd.h"
+#include "glusterd-utils.h"
+
+#include "glusterd-mountbroker.h"
+#include "glusterd-op-sm.h"
+
+static int
+seq_dict_foreach (dict_t *dict,
+                  int (*fn)(char *str, void *data),
+                  void *data)
+{
+        char index[] = "4294967296"; // 1<<32
+        int        i = 0;
+        char    *val = NULL;
+        int      ret = 0;
+
+        for (;;i++) {
+                snprintf(index, sizeof(index), "%d", i);
+                ret = dict_get_str (dict, index, &val);
+                if (ret != 0)
+                        return ret == -ENOENT ? 0 : ret;
+                ret = fn (val, data);
+                if (ret != 0)
+                        return ret;
+        }
+}
+
+static void
+skipwhite (char **s)
+{
+        while (isspace (**s))
+                (*s)++;
+}
+
+static char *
+nwstrtail (char *str, char *pattern)
+{
+        for (;;) {
+                skipwhite (&str);
+                skipwhite (&pattern);
+
+                if (*str != *pattern || !*str)
+                        break;
+
+                str++;
+                pattern++;
+        }
+
+        return *pattern ? NULL : str;
+}
+
+
+int
+parse_mount_pattern_desc (gf_mount_spec_t *mspec, char *pdesc)
+#define SYNTAX_ERR -2
+{
+        char *curs              = NULL;
+        char *c2                = NULL;
+        char sc                 = '\0';
+        char **cc               = NULL;
+        gf_mount_pattern_t *pat = NULL;
+        int pnum                = 0;
+        int ret                 = 0;
+        int lastsup             = -1;
+        int incl                = -1;
+        char **pcc              = NULL;
+        int pnc                 = 0;
+
+        skipwhite (&pdesc);
+
+        /* a bow to theory */
+        if (!*pdesc)
+                return 0;
+
+        /* count number of components, separated by '&' */
+        mspec->len = 0;
+        for (curs = pdesc; *curs; curs++) {
+                if (*curs == ')')
+                        mspec->len++;
+        }
+
+        mspec->patterns = GF_CALLOC (mspec->len, sizeof (*mspec->patterns),
+                                     gf_gld_mt_mount_pattern);
+        if (!mspec->patterns) {
+                ret = -1;
+                goto out;
+        }
+
+        pat = mspec->patterns;
+        curs = pdesc;
+        skipwhite (&curs);
+        for (;;) {
+                incl = -1;
+
+                /* check for pattern signedness modifier */
+                if (*curs == '-') {
+                        pat->negative = _gf_true;
+                        curs++;
+                }
+
+                /* now should come condition specifier,
+                 * then opening paren
+                 */
+                c2 = nwstrtail (curs, "SUB(");
+                if (c2) {
+                        pat->condition = SET_SUB;
+                        goto got_cond;
+                }
+                c2 = nwstrtail (curs, "SUP(");
+                if (c2) {
+                        pat->condition = SET_SUPER;
+                        lastsup = pat - mspec->patterns;
+                        goto got_cond;
+                }
+                c2 = nwstrtail (curs, "EQL(");
+                if (c2) {
+                        pat->condition = SET_EQUAL;
+                        goto got_cond;
+                }
+                c2 = nwstrtail (curs, "MEET(");
+                if (c2) {
+                        pat->condition = SET_INTERSECT;
+                        goto got_cond;
+                }
+                c2 = nwstrtail (curs, "SUB+(");
+                if (c2) {
+                        pat->condition = SET_SUB;
+                        incl = lastsup;
+                        goto got_cond;
+                }
+
+                ret = SYNTAX_ERR;
+                goto out;
+
+ got_cond:
+                curs = c2;
+                skipwhite (&curs);
+                /* count the number of components for pattern */
+                pnum = *curs == ')' ? 0 : 1;
+                for (c2 = curs ;*c2 != ')';) {
+                        if (strchr ("&|", *c2)) {
+                                ret = SYNTAX_ERR;
+                                goto out;
+                        }
+                        while (!strchr ("|&)", *c2) && !isspace (*c2))
+                                c2++;
+                        skipwhite (&c2);
+                        switch (*c2) {
+                        case ')':
+                                break;
+                        case '\0':
+                        case '&':
+                                ret = SYNTAX_ERR;
+                                goto out;
+                        case '|':
+                                *c2 = ' ';
+                                skipwhite (&c2);
+                                /* fall through */
+                        default:
+                                pnum++;
+                        }
+                }
+                if (incl >= 0) {
+                        pnc = 0;
+                        for (pcc = mspec->patterns[incl].components; *pcc; pcc++)
+                                pnc++;
+                        pnum += pnc;
+                }
+                pat->components = GF_CALLOC (pnum + 1, sizeof (*pat->components),
+                                             gf_gld_mt_mount_comp_container);
+                if (!pat->components) {
+                        ret = -1;
+                        goto out;
+                }
+
+                cc = pat->components;
+                /* copy over included component set */
+                if (incl >= 0) {
+                        memcpy (pat->components,
+                                mspec->patterns[incl].components,
+                                pnc * sizeof (*pat->components));
+                        cc += pnc;
+                }
+                /* parse and add components */
+                c2 = ""; /* reset c2 */
+                while (*c2 != ')') {
+                        c2 = curs;
+                        while (!isspace (*c2) && *c2 != ')')
+                                c2++;
+                        sc = *c2;
+                        *c2 = '\0';;
+                        *cc = gf_strdup (curs);
+                        if (!*cc) {
+                                ret = -1;
+                                goto out;
+                        }
+                        *c2 = sc;
+                        skipwhite (&c2);
+                        curs = c2;
+                        cc++;
+                }
+
+                curs++;
+                skipwhite (&curs);
+                if (*curs == '&') {
+                        curs++;
+                        skipwhite (&curs);
+                }
+
+                if (!*curs)
+                        break;
+                pat++;
+        }
+
+ out:
+        if (ret == SYNTAX_ERR) {
+                gf_log ("", GF_LOG_ERROR, "cannot parse mount patterns %s",
+                        pdesc);
+        }
+
+        /* We've allocted a lotta stuff here but don't bother with freeing
+         * on error, in that case we'll terminate anyway
+         */
+        return ret ? -1 : 0;
+}
+#undef SYNTAX_ERR
+
+
+const char *georep_mnt_desc_template =
+        "SUP("
+                "xlator-option=\\*-dht.assert-no-child-down=true "
+                "volfile-server=localhost "
+                "client-pid=-1 "
+                "volfile-id=%s "
+                "user-map-root=%s "
+        ")"
+        "SUB+("
+                "log-file="DEFAULT_LOG_FILE_DIRECTORY"/"GEOREP"*/* "
+                "log-level=* "
+        ")";
+
+int
+make_georep_mountspec (gf_mount_spec_t *mspec, const char *volname,
+                       char *user)
+{
+        char *georep_mnt_desc = NULL;
+        int ret               = 0;
+
+        ret = gf_asprintf (&georep_mnt_desc, georep_mnt_desc_template,
+                           volname, user);
+        if (ret == -1)
+                return ret;
+
+        return parse_mount_pattern_desc (mspec, georep_mnt_desc);
+}
+
+static gf_boolean_t
+match_comp (char *str, char *patcomp)
+{
+        char *c1 = patcomp;
+        char *c2 = str;
+
+        GF_ASSERT (c1);
+        GF_ASSERT (c2);
+
+        while (*c1 == *c2) {
+                if (!*c1)
+                        return _gf_true;
+                c1++;
+                c2++;
+                if (c1[-1] == '=')
+                        break;
+        }
+
+        return fnmatch (c1, c2, 0) == 0 ? _gf_true : _gf_false;
+}
+
+struct gf_set_descriptor {
+        gf_boolean_t priv[2];
+        gf_boolean_t common;
+};
+
+static int
+_gf_set_dict_iter1 (char *val, void *data)
+{
+        void **dataa                 = data;
+        struct gf_set_descriptor *sd = dataa[0];
+        char **curs                  = dataa[1];
+        gf_boolean_t priv            = _gf_true;
+
+        while (*curs) {
+                if (match_comp (val, *curs)) {
+                        priv = _gf_false;
+                        sd->common = _gf_true;
+                }
+                curs++;
+        }
+
+        if (priv)
+                sd->priv[0] = _gf_true;
+
+        return 0;
+}
+
+static int
+_gf_set_dict_iter2 (char *val, void *data)
+{
+        void **dataa      = data;
+        gf_boolean_t *boo = dataa[0];
+        char *comp        = dataa[1];
+
+        if (match_comp (val, comp))
+                *boo = _gf_true;
+
+        return 0;
+}
+
+static void
+relate_sets (struct gf_set_descriptor *sd, dict_t *argdict, char **complist)
+{
+        void *dataa[] = {NULL, NULL};
+        gf_boolean_t boo = _gf_false;
+
+        memset (sd, 0, sizeof (*sd));
+
+        dataa[0] = sd;
+        dataa[1] = complist;
+        seq_dict_foreach (argdict, _gf_set_dict_iter1, dataa);
+
+        while (*complist) {
+                boo = _gf_false;
+                dataa[0] = &boo;
+                dataa[1] = *complist;
+                seq_dict_foreach (argdict, _gf_set_dict_iter2, dataa);
+
+                if (boo)
+                        sd->common = _gf_true;
+                else
+                        sd->priv[1] = _gf_true;
+
+                complist++;
+        }
+}
+
+static int
+_arg_parse_uid (char *val, void *data)
+{
+        char *user        = strtail (val, "user-map-root=");
+        struct passwd *pw = NULL;
+
+        if (!user)
+                return 0;
+        pw = getpwnam (user);
+        if (!pw)
+                return -EINVAL;
+
+        if (*(int *)data >= 0)
+                /* uid ambiguity, already found */
+                return -EINVAL;
+
+        *(int *)data = pw->pw_uid;
+        return 0;
+}
+
+static int
+evaluate_mount_request (gf_mount_spec_t *mspec, dict_t *argdict)
+{
+        struct gf_set_descriptor sd = {{0,},};
+        int i                       = 0;
+        int uid                     = -1;
+        int ret                     = 0;
+        gf_boolean_t match          = _gf_false;
+
+        for (i = 0; i < mspec->len; i++) {
+                relate_sets (&sd, argdict, mspec->patterns[i].components);
+                switch (mspec->patterns[i].condition) {
+                case SET_SUB:
+                        match = !sd.priv[0];
+                        break;
+                case SET_SUPER:
+                        match = !sd.priv[1];
+                        break;
+                case SET_EQUAL:
+                        match = (!sd.priv[0] && !sd.priv[1]);
+                        break;
+                case SET_INTERSECT:
+                        match = sd.common;
+                default:
+                        GF_ASSERT(!"unreached");
+                }
+                if (mspec->patterns[i].negative)
+                        match = !match;
+
+                if (!match)
+                        return -EPERM;
+        }
+
+        ret = seq_dict_foreach (argdict, _arg_parse_uid, &uid);
+        if (ret != 0)
+                return ret;
+
+        return uid;
+}
+
+static int
+_volname_get (char *val, void *data)
+{
+        char **volname = data;
+
+        *volname = strtail (val, "volfile-id=");
+
+        return *volname ? 1 : 0;
+}
+
+static int
+_runner_add (char *val, void *data)
+{
+        runner_t *runner = data;
+
+        runner_argprintf (runner, "--%s", val);
+
+        return 0;
+}
+
+int
+glusterd_do_mount (char *label, dict_t *argdict, char **path, int *op_errno)
+{
+        glusterd_conf_t *priv      = NULL;
+        char *mountbroker_root     = NULL;
+        gf_mount_spec_t *mspec     = NULL;
+        int uid                    = -ENOENT;
+        char *volname              = NULL;
+        glusterd_volinfo_t *vol    = NULL;
+        char *mtptemp              = NULL;
+        char *mntlink              = NULL;
+        char *cookieswitch         = NULL;
+        char *cookie               = NULL;
+        char *sla                  = NULL;
+        struct stat st             = {0,};
+        runner_t runner            = {0,};
+        int ret                    = 0;
+        xlator_t *this             = THIS;
+
+        priv = this->private;
+        GF_ASSERT (priv);
+
+        GF_ASSERT (op_errno);
+        *op_errno = 0;
+
+        if (dict_get_str (this->options, "mountbroker-root",
+                          &mountbroker_root) != 0) {
+                *op_errno = ENOENT;
+                goto out;
+        }
+
+        GF_ASSERT (label);
+        if (!*label) {
+                *op_errno = EINVAL;
+                goto out;
+        }
+
+        /* look up spec for label */
+        list_for_each_entry (mspec, &priv->mount_specs,
+                             speclist) {
+                if (strcmp (mspec->label, label) != 0)
+                        continue;
+                uid = evaluate_mount_request (mspec, argdict);
+                break;
+        }
+        if (uid < 0) {
+                *op_errno = -uid;
+                goto out;
+        }
+
+        /* some sanity check on arguments */
+        seq_dict_foreach (argdict, _volname_get, &volname);
+        if (!volname) {
+                *op_errno = EINVAL;
+                goto out;
+        }
+        if (glusterd_volinfo_find (volname, &vol) != 0 ||
+            !glusterd_is_volume_started (vol)) {
+                *op_errno = ENOENT;
+                goto out;
+        }
+
+        /* go do mount */
+
+        /** create actual mount dir */
+
+        /*** "overload" string name to be possible to used for cookie
+             creation, see below */
+        ret = gf_asprintf (&mtptemp, "%s/user%d/mtpt-%s-XXXXXX/cookie",
+                           mountbroker_root, uid, label);
+        if (ret == -1) {
+                mtptemp = NULL;
+                *op_errno = ENOMEM;
+                goto out;
+        }
+        /*** hide cookie part */
+        cookieswitch = strrchr (mtptemp, '/');
+        *cookieswitch = '\0';
+
+        sla = strrchr (mtptemp, '/');
+        *sla = '\0';
+        ret = mkdir (mtptemp, 0700);
+        if (ret == 0)
+                ret = chown (mtptemp, uid, 0);
+        else if (errno == EEXIST)
+                ret = 0;
+        if (ret == -1) {
+                *op_errno = errno;
+                goto out;
+        }
+        ret = lstat (mtptemp, &st);
+        if (ret == -1) {
+                *op_errno = errno;
+                goto out;
+        }
+        if (!(S_ISDIR (st.st_mode) && (st.st_mode & ~S_IFMT) == 0700 &&
+              st.st_uid == uid && st.st_gid == 0)) {
+                *op_errno = EACCES;
+                goto out;
+        }
+        *sla = '/';
+
+        if (!mkdtemp (mtptemp)) {
+                *op_errno = errno;
+                goto out;
+        }
+
+        /** create private "cookie" symlink */
+
+        /*** occupy an entry in the hive dir via mkstemp */
+        ret = gf_asprintf (&cookie, "%s/"MB_HIVE"/mntXXXXXX",
+                           mountbroker_root);
+        if (ret == -1) {
+                cookie = NULL;
+                *op_errno = ENOMEM;
+                goto out;
+        }
+        ret = mkstemp (cookie);
+        if (ret == -1) {
+                *op_errno = errno;
+                goto out;
+        }
+        close (ret);
+
+        /*** assembly the path from cookie to mountpoint */
+        sla = strchr (sla - 1, '/');
+        GF_ASSERT (sla);
+        ret = gf_asprintf (&mntlink, "../user%d%s", uid, sla);
+        if (ret == -1) {
+                *op_errno = ENOMEM;
+                goto out;
+        }
+
+        /*** create cookie link in (to-be) mountpoint,
+             move it over to the final place */
+        *cookieswitch = '/';
+        ret = symlink (mntlink, mtptemp);
+        if (ret != -1)
+                ret = rename (mtptemp, cookie);
+        *cookieswitch = '\0';
+        if (ret == -1) {
+                *op_errno = errno;
+                goto out;
+        }
+
+        /** invoke glusterfs on the mountpoint */
+
+        runinit (&runner);
+        runner_add_arg (&runner, GFS_PREFIX"/sbin/glusterfs");
+        seq_dict_foreach (argdict, _runner_add, &runner);
+        runner_add_arg (&runner, mtptemp);
+        ret = runner_run_reuse (&runner);
+        if (ret == -1) {
+                *op_errno = EIO; /* XXX hacky fake */
+                runner_log (&runner, "", GF_LOG_ERROR, "command failed");
+        }
+        runner_end (&runner);
+
+ out:
+
+        if (*op_errno) {
+                ret = -1;
+                gf_log ("", GF_LOG_WARNING, "unsuccessful mount request (%s)",
+                        strerror (*op_errno));
+                if (mtptemp) {
+                        *cookieswitch = '/';
+                        unlink (mtptemp);
+                        *cookieswitch = '\0';
+                        rmdir (mtptemp);
+                }
+                if (cookie) {
+                        unlink (cookie);
+                        GF_FREE (cookie);
+                }
+
+        } else {
+                ret = 0;
+                *path = cookie;
+        }
+
+        if (mtptemp)
+                GF_FREE (mtptemp);
+
+        return ret;
+}
diff --git a/xlators/mgmt/glusterd/src/glusterd-mountbroker.h b/xlators/mgmt/glusterd/src/glusterd-mountbroker.h
new file mode 100644
index 000000000..e5b3d23ae
--- /dev/null
+++ b/xlators/mgmt/glusterd/src/glusterd-mountbroker.h
@@ -0,0 +1,50 @@
+/*
+  Copyright (c) 2011 Gluster, Inc. <http://www.gluster.com>
+  This file is part of GlusterFS.
+
+  GlusterFS is free software; you can redistribute it and/or modify
+  it under the terms of the GNU General Public License as published
+  by the Free Software Foundation; either version 3 of the License,
+  or (at your option) any later version.
+
+  GlusterFS is distributed in the hope that it will be useful, but
+  WITHOUT ANY WARRANTY; without even the implied warranty of
+  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+  General Public License for more details.
+
+  You should have received a copy of the GNU General Public License
+  along with this program.  If not, see
+  <http://www.gnu.org/licenses/>.
+*/
+
+#define MB_HIVE "mb_hive"
+
+typedef enum {
+        SET_SUB = 1,
+        SET_SUPER,
+        SET_EQUAL,
+        SET_INTERSECT
+} gf_setrel_t;
+
+struct gf_mount_pattern {
+        char **components;
+        gf_setrel_t condition;
+        gf_boolean_t negative;
+};
+typedef struct gf_mount_pattern gf_mount_pattern_t;
+
+struct gf_mount_spec {
+        struct list_head speclist;
+        char *label;
+        gf_mount_pattern_t *patterns;
+        size_t len;
+};
+typedef struct gf_mount_spec gf_mount_spec_t;
+
+
+int parse_mount_pattern_desc (gf_mount_spec_t *mspec, char *pdesc);
+
+int make_georep_mountspec (gf_mount_spec_t *mspec, const char *volname,
+                           char *user);
+
+int glusterd_do_mount (char *label, dict_t *argdict, char **path, int *op_errno);
diff --git a/xlators/mgmt/glusterd/src/glusterd.c b/xlators/mgmt/glusterd/src/glusterd.c
index 09e20a186..6daf84a06 100644
--- a/xlators/mgmt/glusterd/src/glusterd.c
+++ b/xlators/mgmt/glusterd/src/glusterd.c
@@ -47,6 +47,8 @@
 #include "common-utils.h"
 #include "run.h"
 
+#include "glusterd-mountbroker.h"
+
 static uuid_t glusterd_uuid;
 extern struct rpcsvc_program glusterd1_mop_prog;
 extern struct rpcsvc_program gd_svc_mgmt_prog;
@@ -500,6 +502,151 @@ configure_syncdaemon (glusterd_conf_t *conf)
 }
 #undef RUN_GSYNCD_CMD
 
+static int
+check_prepare_mountbroker_root (char *mountbroker_root)
+{
+        int dfd0        = -1;
+        int dfd         = -1;
+        int dfd2        = -1;
+        struct stat st  = {0,};
+        struct stat st2 = {0,};
+        int ret         = 0;
+
+        ret = open (mountbroker_root, O_RDONLY);
+        if (ret != -1) {
+                dfd = ret;
+                ret = fstat (dfd, &st);
+        }
+        if (ret == -1 || !S_ISDIR (st.st_mode)) {
+                gf_log ("", GF_LOG_ERROR,
+                        "cannot access mountbroker-root directory %s",
+                        mountbroker_root);
+                ret = -1;
+                goto out;
+        }
+        if (st.st_uid != 0 ||
+            (st.st_mode & (S_IWGRP|S_IWOTH))) {
+                gf_log ("", GF_LOG_ERROR,
+                        "permissions on mountbroker-root directory %s are "
+                        "too liberal", mountbroker_root);
+                ret = -1;
+                goto out;
+        }
+
+        dfd0 = dup (dfd);
+
+        for (;;) {
+                ret = openat (dfd, "..", O_RDONLY);
+                if (ret != -1) {
+                        dfd2 = ret;
+                        ret = fstat (dfd2, &st2);
+                }
+                if (ret == -1) {
+                        gf_log ("", GF_LOG_ERROR,
+                                "error while checking mountbroker-root ancestors "
+                                "%d (%s)", errno, strerror (errno));
+                        goto out;
+                }
+
+                if (st2.st_ino == st.st_ino)
+                        break; /* arrived to root */
+
+                if (st2.st_uid != 0 ||
+                    ((st2.st_mode & (S_IWGRP|S_IWOTH)) &&
+                     !(st2.st_mode & S_ISVTX))) {
+                        gf_log ("", GF_LOG_ERROR,
+                                "permissions on ancestors of mountbroker-root "
+                                "directory are too liberal");
+                        ret = -1;
+                        goto out;
+                }
+
+                close (dfd);
+                dfd = dfd2;
+                st = st2;
+        }
+
+        ret = mkdirat (dfd0, MB_HIVE, 0711);
+        if (ret == -1 && errno == EEXIST)
+                ret = 0;
+        if (ret != -1)
+                ret = fstatat (dfd0, MB_HIVE, &st, AT_SYMLINK_NOFOLLOW);
+        if (ret == -1 || st.st_mode != (S_IFDIR|0711)) {
+                gf_log ("", GF_LOG_ERROR,
+                        "failed to set up mountbroker-root directory %s",
+                        mountbroker_root);
+                ret = -1;
+                goto out;
+        }
+
+        ret = 0;
+
+ out:
+        close (dfd0);
+        close (dfd);
+        close (dfd2);
+
+        return ret;
+}
+
+static void
+_install_mount_spec (dict_t *opts, char *key, data_t *value, void *data)
+{
+        glusterd_conf_t *priv  = THIS->private;
+        char *label            = NULL;
+        gf_boolean_t georep    = _gf_false;
+        char *pdesc            = value->data;
+        char *volname          = NULL;
+        int *ret               = data;
+        int rv                 = 0;
+        gf_mount_spec_t *mspec = NULL;
+        char *user             = NULL;
+
+        if (*ret == -1)
+                return;
+
+        label = strtail (key, "mountbroker.");
+        if (!label) {
+                georep = _gf_true;
+                label = strtail (key, "mountbroker-"GEOREP".");
+        }
+
+        if (!label)
+                return;
+
+        mspec = GF_CALLOC (1, sizeof (*mspec), gf_gld_mt_mount_spec);
+        if (!mspec)
+                goto err;
+        mspec->label = label;
+
+        if (georep) {
+                volname = gf_strdup (pdesc);
+                if (!volname)
+                        goto err;
+                user = strchr (volname, ':');
+                if (user) {
+                        *user = '\0';
+                        user++;
+                } else
+                        user = label;
+                rv = make_georep_mountspec (mspec, volname, user);
+                GF_FREE (volname);
+                if (rv != 0)
+                        goto err;
+        } else if (parse_mount_pattern_desc (mspec, pdesc) != 0)
+                goto err;
+
+        list_add_tail (&mspec->speclist, &priv->mount_specs);
+
+        return;
+ err:
+
+        gf_log ("", GF_LOG_ERROR,
+                "adding %smount spec failed: label: %s desc: %s",
+                georep ? GEOREP" " : "", label, pdesc);
+
+        *ret = -1;
+}
 
 /*
  * init - called during glusterd initialization
@@ -519,6 +666,7 @@ init (xlator_t *this)
         char               dirname [PATH_MAX];
         char               cmd_log_filename [PATH_MAX] = {0,};
         int                first_time        = 0;
+        char              *mountbroker_root  = NULL;
 
         dir_data = dict_get (this->options, "working-directory");
 
@@ -699,6 +847,19 @@ init (xlator_t *this)
         if (ret < 0)
                 goto out;
 
+        INIT_LIST_HEAD (&conf->mount_specs);
+        dict_foreach (this->options, _install_mount_spec, &ret);
+        if (ret)
+                goto out;
+        ret = dict_get_str (this->options, "mountbroker-root",
+                            &mountbroker_root);
+        if (ret)
+                ret = 0;
+        else
+                ret = check_prepare_mountbroker_root (mountbroker_root);
+        if (ret)
+                goto out;
+
         ret = configure_syncdaemon (conf);
         if (ret)
                 goto out;
@@ -830,5 +991,16 @@ struct volume_options options[] = {
         { .key = {"bind-insecure"},
           .type = GF_OPTION_TYPE_BOOL,
         },
+
+        { .key  = {"mountbroker-root"},
+          .type = GF_OPTION_TYPE_PATH,
+        },
+        { .key  = {"mountbroker.*"},
+          .type = GF_OPTION_TYPE_ANY,
+        },
+        { .key  = {"mountbroker-"GEOREP".*"},
+          .type = GF_OPTION_TYPE_ANY,
+        },
+
         { .key   = {NULL} },
 };
diff --git a/xlators/mgmt/glusterd/src/glusterd.h b/xlators/mgmt/glusterd/src/glusterd.h
index 41fbe5faa..86eeaeb1c 100644
--- a/xlators/mgmt/glusterd/src/glusterd.h
+++ b/xlators/mgmt/glusterd/src/glusterd.h
@@ -111,6 +111,7 @@ typedef struct {
         gf_timer_t *timer;
         glusterd_sm_tr_log_t op_sm_log;
         struct rpc_clnt_program *gfs_mgmt;
+        struct list_head mount_specs;
 } glusterd_conf_t;
 
 typedef enum gf_brick_status {