kvs: support non-environment variable mechanism to choose namespace

doc/man1/flux-kvs.adoc

@@ -39,20 +39,20 @@ their parent in the overlay network.
 
 Different KVS namespaces can be created in which kvs values can be
 read from/written to.  By default, all KVS operations operate on the
-default KVS namespace "primary".  To use a different namespace, set
-the environment variable FLUX_KVS_NAMESPACE to the namespace you wish
-to use.  When a KVS namespace is first created, it is only created on
-the leader.  Followers will initialize itself against a namespace the
-first time it is used.  When a namespace is removed, a namespace is
-marked for removal on the leader and an event is multicast to
-followers to similarly mark the namespace for removal.  The namespace
-will then be garbage collected in the background.  There may be a
-delay in the namespace being recognized as removed.  Similar to
-commits, it is "eventually consistent".
+default KVS namespace "primary".  An alternate namespace can be
+specified via the '--namespace' option or the environment variable
+FLUX_KVS_NAMESPACE.
 
 flux-kvs(1) runs a KVS 'COMMAND'.  The possible commands and their
 arguments are described below.
 
+COMMON OPTIONS
+--------------
+*-N, --namespace*='name'::
+Specify a namespace to be used in all KVS commands.  A namespace
+specified via this option overrides a namespace set in the environment
+variable FLUX_KVS_NAMESPACE.
+
 COMMANDS
 --------
 *namespace-create* [-o owner] 'name' ['name...']::

doc/man3/Makefile.am

@@ -45,7 +45,8 @@ MAN3_FILES_PRIMARY = \
 	flux_kvs_lookup.3 \
 	flux_kvs_commit.3 \
 	flux_kvs_txn_create.3 \
-	flux_kvs_namespace_create.3
+	flux_kvs_namespace_create.3 \
+	flux_kvs_set_namespace.3
 
 # These files are generated as roff .so includes of a primary page.
 # A2X handles this automatically if mentioned in NAME section
@@ -141,7 +142,8 @@ MAN3_FILES_SECONDARY = \
 	flux_kvs_txn_symlink.3 \
 	flux_kvs_txn_put_raw.3 \
 	# flux_kvs_txn_put_treeobj.3 \
-	flux_kvs_namespace_remove.3
+	flux_kvs_namespace_remove.3 \
+	flux_kvs_get_namespace.3
 
 ADOC_FILES  = $(MAN3_FILES_PRIMARY:%.3=%.adoc)
 XML_FILES   = $(MAN3_FILES_PRIMARY:%.3=%.xml)
@@ -257,6 +259,7 @@ flux_kvs_txn_symlink.3: flux_kvs_txn_create.3
 flux_kvs_txn_put_raw.3: flux_kvs_txn_create.3
 # flux_kvs_txn_put_treeobj.3: flux_kvs_txn_create.3
 flux_kvs_namespace_remove.3: flux_kvs_namespace_create.3
+flux_kvs_get_namespace.3: flux_kvs_set_namespace.3
 
 flux_open.3: topen.c
 flux_send.3: tsend.c

doc/man3/flux_kvs_commit.adoc

@@ -49,9 +49,9 @@ was successful, indicating the entire transaction was committed, or -1
 on failure, indicating none of the transaction was committed.
 
 By default, both `flux_kvs_commit()` and `flux_kvs_fence()` operate on
-the default KVS namespace.  To use a different namespace, set the
-environment variable FLUX_KVS_NAMESPACE to the namespace you wish to
-use.
+the default KVS namespace.  To use a different namespace, see the
+`flux_kvs_set_namespace()` function or set the environment variable
+FLUX_KVS_NAMESPACE to the namespace you wish to use.
 
 FLAGS
 -----
@@ -112,4 +112,4 @@ include::COPYRIGHT.adoc[]
 
 SEE ALSO
 ---------
-flux_future_then(3), flux_kvs_txn_create(3)
+flux_future_then(3), flux_kvs_txn_create(3), flux_kvs_set_namespace(3)

doc/man3/flux_kvs_lookup.adoc

@@ -49,9 +49,9 @@ static set of content within the KVS, effectively a snapshot.
 See `flux_kvs_lookup_get_treeobj()` below.
 
 By default, both `flux_kvs_lookup()` and `flux_kvs_lookupat()` operate
-on the default KVS namespace.  To use a different namespace, set the
-environment variable FLUX_KVS_NAMESPACE to the namespace you wish to
-use.
+on the default KVS namespace.  To use a different namespace, see the
+`flux_kvs_set_namespace()` function or set the environment variable
+FLUX_KVS_NAMESPACE to the namespace you wish to use.
 
 All the functions below are variations on a common theme.  First they
 complete the lookup RPC by blocking on the response, if not already received.
@@ -170,6 +170,6 @@ include::COPYRIGHT.adoc[]
 
 SEE ALSO
 ---------
-flux_rpc(3), flux_future_then(3)
+flux_rpc(3), flux_future_then(3), flux_kvs_set_namespace(3)
 
 https://github.com/flux-framework/rfc/blob/master/spec_11.adoc[RFC 11: Key Value Store Tree Object Format v1]

doc/man3/flux_kvs_set_namespace.adoc

@@ -0,0 +1,67 @@
+flux_kvs_set_namespace(3)
+=========================
+:doctype: manpage
+
+
+NAME
+----
+flux_kvs_set_namespace, flux_kvs_get_namespace - set/get KVS namespace
+
+
+SYNOPSIS
+--------
+ #include <flux/core.h>
+
+ int flux_kvs_set_namespace (flux_t *h, const char *namespace);
+
+ const char *flux_kvs_get_namespace (flux_t *h);
+
+DESCRIPTION
+-----------
+
+`flux_kvs_set_namespace()` sets the KVS namespace to use for all
+KVS operations within a flux handle.  By setting a KVS namespace
+in the flux handle, this namespace will override any KVS namespace
+specified in the environment variable FLUX_KVS_NAMESPACE.
+
+`flux_kvs_get_namespace()` will determine the current namespace to
+use, whether set via `flux_kvs_set_namespace()` or the environment
+variable FLUX_KVS_NAMESPACE.  If neither is set, the default
+namespace of "primary" will be used.
+
+
+RETURN VALUE
+------------
+
+`flux_kvs_set_namespace()` returns 0 on success, or -1 on failure.
+`flux_kvs_get_namespace()` returns namespace on success, or NULL on
+failure.
+
+
+ERRORS
+------
+
+EINVAL::
+One of the arguments was invalid.
+
+ENOMEM::
+Out of memory.
+
+AUTHOR
+------
+This page is maintained by the Flux community.
+
+
+RESOURCES
+---------
+Github: <http://github.com/flux-framework>
+
+
+COPYRIGHT
+---------
+include::COPYRIGHT.adoc[]
+
+
+SEE ALSO
+---------
+flux_kvs_lookup(3), flux_kvs_commit(3)

src/cmd/flux-kvs.c

@@ -60,6 +60,13 @@ static void dump_kvs_dir (const flux_kvsdir_t *dir, int maxcol,
 
 #define min(a,b) ((a)<(b)?(a):(b))
 
+static struct optparse_option global_opts[] =  {
+    { .name = "namespace", .key = 'N', .has_arg = 1,
+      .usage = "Specify KVS namespace to use.",
+    },
+    OPTPARSE_TABLE_END
+};
+
 static struct optparse_option namespace_create_opts[] =  {
     { .name = "owner", .key = 'o', .has_arg = 1,
       .usage = "Specify alternate namespace owner via userid",
@@ -314,11 +321,15 @@ int main (int argc, char *argv[])
     optparse_t *p;
     int optindex;
     int exitval;
+    const char *namespace;
 
     log_init ("flux-kvs");
 
     p = optparse_create ("flux-kvs");
 
+    if (optparse_add_option_table (p, global_opts) != OPTPARSE_SUCCESS)
+        log_msg_exit ("optparse_add_option_table() failed");
+
     /* Override help option for our own */
     if (optparse_set (p, OPTPARSE_USAGE, cmdusage) != OPTPARSE_SUCCESS)
         log_msg_exit ("optparse_set (USAGE)");
@@ -338,14 +349,19 @@ int main (int argc, char *argv[])
         exit (1);
 
     if ((argc - optindex == 0)
-        || !optparse_get_subcommand (p, argv[optind])) {
+        || !optparse_get_subcommand (p, argv[optindex])) {
         usage (p, NULL, NULL);
         exit (1);
     }
 
     if (!(h = flux_open (NULL, 0)))
         log_err_exit ("flux_open");
 
+    if ((namespace = optparse_get_str (p, "namespace", NULL))) {
+        if (flux_kvs_set_namespace (h, namespace) < 0)
+            log_msg_exit ("flux_kvs_set_namespace");
+    }
+
     optparse_set_data (p, "flux_handle", h);
 
     if ((exitval = optparse_run_subcommand (p, argc, argv)) < 0)

src/common/libkvs/Makefile.am

@@ -13,7 +13,6 @@ noinst_LTLIBRARIES = libkvs.la
 
 libkvs_la_SOURCES = \
 	kvs.c \
-	kvs_private.h \
 	kvs_lookup.c \
 	kvs_dir.c \
 	kvs_dir_private.h \
@@ -39,6 +38,8 @@ TESTS = \
 	test_kvs_txn.t \
 	test_kvs_lookup.t \
 	test_kvs_dir.t \
+	test_kvs_commit.t \
+	test_kvs_watch.t \
 	test_treeobj.t
 
 check_PROGRAMS = \
@@ -75,6 +76,14 @@ test_kvs_dir_t_SOURCES = test/kvs_dir.c
 test_kvs_dir_t_CPPFLAGS = $(test_cppflags)
 test_kvs_dir_t_LDADD = $(test_ldadd) $(LIBDL)
 
+test_kvs_commit_t_SOURCES = test/kvs_commit.c
+test_kvs_commit_t_CPPFLAGS = $(test_cppflags)
+test_kvs_commit_t_LDADD = $(test_ldadd) $(LIBDL)
+
+test_kvs_watch_t_SOURCES = test/kvs_watch.c
+test_kvs_watch_t_CPPFLAGS = $(test_cppflags)
+test_kvs_watch_t_LDADD = $(test_ldadd) $(LIBDL)
+
 test_treeobj_t_SOURCES = test/treeobj.c
 test_treeobj_t_CPPFLAGS = $(test_cppflags)
 test_treeobj_t_LDADD = $(test_ldadd) $(LIBDL)

src/common/libkvs/kvs.c

@@ -29,12 +29,7 @@
 
 #include <flux/core.h>
 
-const char *get_kvs_namespace (void)
-{
-    if (getenv ("FLUX_KVS_NAMESPACE"))
-        return getenv ("FLUX_KVS_NAMESPACE");
-    return KVS_PRIMARY_NAMESPACE;
-}
+#define FLUX_HANDLE_KVS_NAMESPACE "kvsnamespace"
 
 flux_future_t *flux_kvs_namespace_create (flux_t *h, const char *namespace,
                                           uint32_t owner, int flags)
@@ -64,13 +59,51 @@ flux_future_t *flux_kvs_namespace_remove (flux_t *h, const char *namespace)
                           "namespace", namespace);
 }
 
+int flux_kvs_set_namespace (flux_t *h, const char *namespace)
+{
+    char *str;
+
+    if (!h || !namespace) {
+        errno = EINVAL;
+        return -1;
+    }
+
+    if (!(str = strdup (namespace))) {
+        errno = ENOMEM;
+        return -1;
+    }
+
+    flux_aux_set (h, FLUX_HANDLE_KVS_NAMESPACE, str, free);
+    return 0;
+}
+
+const char *flux_kvs_get_namespace (flux_t *h)
+{
+    const char *namespace;
+
+    if (!h) {
+        errno = EINVAL;
+        return NULL;
+    }
+
+    if ((namespace = flux_aux_get (h, FLUX_HANDLE_KVS_NAMESPACE)))
+        return namespace;
+
+    if ((namespace = getenv ("FLUX_KVS_NAMESPACE")))
+        return namespace;
+
+    return KVS_PRIMARY_NAMESPACE;
+}
+
 int flux_kvs_get_version (flux_t *h, int *versionp)
 {
     flux_future_t *f;
-    const char *namespace = get_kvs_namespace ();
+    const char *namespace;
     int version;
     int rc = -1;
 
+    if (!(namespace = flux_kvs_get_namespace (h)))
+        return -1;
     if (!(f = flux_rpc_pack (h, "kvs.getroot", FLUX_NODEID_ANY, 0, "{ s:s }",
                              "namespace", namespace)))
         goto done;
@@ -87,9 +120,11 @@ int flux_kvs_get_version (flux_t *h, int *versionp)
 int flux_kvs_wait_version (flux_t *h, int version)
 {
     flux_future_t *f;
-    const char *namespace = get_kvs_namespace ();
+    const char *namespace;
     int ret = -1;
 
+    if (!(namespace = flux_kvs_get_namespace (h)))
+        return -1;
     if (!(f = flux_rpc_pack (h, "kvs.sync", FLUX_NODEID_ANY, 0, "{ s:i s:s }",
                              "rootseq", version,
                              "namespace", namespace)))

src/common/libkvs/kvs.h

@@ -36,6 +36,16 @@ flux_future_t *flux_kvs_namespace_create (flux_t *h, const char *namespace,
                                           uint32_t owner, int flags);
 flux_future_t *flux_kvs_namespace_remove (flux_t *h, const char *namespace);
 
+/* Namespace Selection
+ * - configure a KVS namespace to use in all kvs operations using this
+ *   handle.
+ * - if never set, the value from the environment variable
+ *   FLUX_KVS_NAMESPACE is used.
+ * - if FLUX_KVS_NAMESPACE is not set, KVS_PRIMARY_NAMESPACE is assumed.
+ */
+int flux_kvs_set_namespace (flux_t *h, const char *namespace);
+const char *flux_kvs_get_namespace (flux_t *h);
+
 /* Synchronization:
  * Process A commits data, then gets the store version V and sends it to B.
  * Process B waits for the store version to be >= V, then reads data.

src/common/libkvs/kvs_commit.c

@@ -29,14 +29,16 @@
 #include <czmq.h>
 #include <flux/core.h>
 
-#include "kvs_private.h"
 #include "kvs_txn_private.h"
 #include "src/common/libutil/blobref.h"
 
 flux_future_t *flux_kvs_fence (flux_t *h, int flags, const char *name,
                                int nprocs, flux_kvs_txn_t *txn)
 {
-    const char *namespace = get_kvs_namespace ();
+    const char *namespace;
+
+    if (!(namespace = flux_kvs_get_namespace (h)))
+        return NULL;
 
     if (txn) {
         json_t *ops;

src/common/libkvs/kvs_commit.h

@@ -9,8 +9,6 @@ enum kvs_commit_flags {
     FLUX_KVS_NO_MERGE = 1, /* disallow commits to be mergeable with others */
 };
 
-/* To use an alternate namespace, set environment variable FLUX_KVS_NAMESPACE */
-
 flux_future_t *flux_kvs_commit (flux_t *h, int flags, flux_kvs_txn_t *txn);
 
 flux_future_t *flux_kvs_fence (flux_t *h, int flags, const char *name,