Merge pull request #1432 from chu11/issue1341-part5

kvs: Support namespace prefixes in symlinks

doc/man1/flux-kvs.adoc

@@ -40,8 +40,9 @@ 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".  An alternate namespace can be
-specified via the '--namespace' option or the environment variable
-FLUX_KVS_NAMESPACE.
+prefixing the name to a key using the format 'ns:<NAMESPACE>/<KEY>',
+specifying the namespace via the '--namespace' option, or by setting
+the namespace in the environment variable FLUX_KVS_NAMESPACE.
 
 flux-kvs(1) runs a KVS 'COMMAND'.  The possible commands and their
 arguments are described below.
@@ -51,7 +52,8 @@ 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.
+variable FLUX_KVS_NAMESPACE.  This does not override the namespace
+specified via a prefix to a key.
 
 COMMANDS
 --------

doc/man3/flux_kvs_commit.adoc

@@ -49,9 +49,13 @@ 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, see the
-`flux_kvs_set_namespace()` function or set the environment variable
-FLUX_KVS_NAMESPACE to the namespace you wish to use.
+the default KVS namespace.  A different namespace can be chosen
+several ways.  A namespace can be set via a special prefix to a key,
+using the format 'ns:<NAMESPACE>/<KEY>', using the
+`flux_kvs_set_namespace()` function, or set via the environment
+variable FLUX_KVS_NAMESPACE.  Note that all transactions must operate
+on the same namespace.  It is an error if multiple different
+namespaces are specified via a namespace prefix.
 
 FLAGS
 -----

doc/man3/flux_kvs_lookup.adoc

@@ -51,9 +51,11 @@ 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, see the
-`flux_kvs_set_namespace()` function or set the environment variable
-FLUX_KVS_NAMESPACE to the namespace you wish to use.
+on the default KVS namespace.  A different namespace can be chosen
+several ways.  A namespace can be set via a special prefix to a key,
+using the format 'ns:<NAMESPACE>/<KEY>', using the
+`flux_kvs_set_namespace()` function, or set via the environment
+variable FLUX_KVS_NAMESPACE.
 
 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.

doc/man3/flux_kvs_set_namespace.adoc

@@ -22,7 +22,8 @@ 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.
+specified in the environment variable FLUX_KVS_NAMESPACE.  However,
+a namespace specified via a key prefix will override this setting.
 
 `flux_kvs_get_namespace()` will determine the current namespace to
 use, whether set via `flux_kvs_set_namespace()` or the environment

doc/test/spell.en.pws

@@ -410,3 +410,4 @@ ENOTSUP
 EOVERFLOW
 itr
 JOBIDS
+ns

src/cmd/flux-kvs.c

@@ -1347,6 +1347,21 @@ static int sort_cmp (void *item1, void *item2)
     return strcmp (item1, item2);
 }
 
+static void contains_namespace_prefix (const char *key, char **key_suffix)
+{
+    char *ptr;
+
+    if (!strncmp (key, "ns:", 3)
+        && (ptr = strchr (key, '/'))) {
+
+        /* No key suffix */
+        if (*(ptr + 1) == '\0')
+            log_err_exit ("%s: %s\n", key, flux_strerror (EINVAL));
+
+        (*key_suffix) = ptr + 1;
+    }
+}
+
 /* Put key in 'dirs' or 'singles' list, depending on whether
  * its contents are to be listed or not.  If -F is specified,
  * 'singles' key names are decorated based on their type.
@@ -1360,16 +1375,21 @@ static int categorize_key (optparse_t *p, const char *key,
     char *nkey;
     json_t *treeobj = NULL;
     bool require_directory = false;
+    char *key_ptr;
 
     if (!(nkey = malloc (strlen (key) + 2))) // room for decoration char + null
         log_err_exit ("malloc");
     strcpy (nkey, key);
 
+    key_ptr = nkey;
+
+    contains_namespace_prefix (nkey, &key_ptr);
+
     /* If the key has a "." suffix, strip it off, but require
      * that the key be a directory type.
      */
-    while (strlen (nkey) > 1 && nkey[strlen (nkey) - 1] == '.') {
-        nkey[strlen (nkey) - 1] = '\0';
+    while (strlen (key_ptr) > 1 && key_ptr[strlen (key_ptr) - 1] == '.') {
+        key_ptr[strlen (key_ptr) - 1] = '\0';
         require_directory = true;
     }
     if (!(f = flux_kvs_lookup (h, FLUX_KVS_TREEOBJ, nkey)))

src/modules/kvs/kvstxn.c

@@ -539,6 +539,31 @@ static int kvstxn_append (kvstxn_t *kt, int current_epoch, json_t *dirent,
     return 0;
 }
 
+static int check_cross_namespace (kvstxn_t *kt,
+                                  const char *key,
+                                  char **key_suffixp)
+{
+    char *ns_prefix = NULL, *key_suffix = NULL;
+    int pret;
+
+    if ((pret = kvs_namespace_prefix (key, &ns_prefix, &key_suffix)) < 0)
+        return -1;
+
+    if (pret) {
+        /* Cannot cross namespaces */
+        if (strcmp (ns_prefix, kt->ktm->namespace)) {
+            free (ns_prefix);
+            free (key_suffix);
+            errno = EINVAL;
+            return -1;
+        }
+    }
+
+    free (ns_prefix);
+    (*key_suffixp) = key_suffix;
+    return 0;
+}
+
 /* link (key, dirent) into directory 'dir'.
  */
 static int kvstxn_link_dirent (kvstxn_t *kt, int current_epoch,
@@ -551,22 +576,13 @@ static int kvstxn_link_dirent (kvstxn_t *kt, int current_epoch,
     json_t *dir = rootdir;
     json_t *subdir = NULL, *dir_entry;
     int saved_errno, rc = -1;
-    char *ns_prefix = NULL, *key_suffix = NULL;
-    int pret;
+    char *key_suffix = NULL;
 
-    if ((pret = kvs_namespace_prefix (key, &ns_prefix, &key_suffix)) < 0) {
+    if (check_cross_namespace (kt, key, &key_suffix) < 0) {
         saved_errno = errno;
         goto done;
     }
 
-    if (pret) {
-        /* Cannot cross namespaces */
-        if (strcmp (ns_prefix, kt->ktm->namespace)) {
-            saved_errno = EINVAL;
-            goto done;
-        }
-    }
-
     if (!(cpy = kvs_util_normalize_key (key_suffix ? key_suffix : key, NULL))) {
         saved_errno = errno;
         goto done;
@@ -658,6 +674,7 @@ static int kvstxn_link_dirent (kvstxn_t *kt, int current_epoch,
             json_t *symlink = treeobj_get_data (dir_entry);
             const char *symlinkstr;
             char *nkey = NULL;
+            char *sym_suffix = NULL;
 
             if (!symlink) {
                 saved_errno = errno;
@@ -667,8 +684,18 @@ static int kvstxn_link_dirent (kvstxn_t *kt, int current_epoch,
             assert (json_is_string (symlink));
 
             symlinkstr = json_string_value (symlink);
+
+            if (check_cross_namespace (kt, symlinkstr, &sym_suffix) < 0) {
+                saved_errno = errno;
+                goto done;
+            }
+
+            if (sym_suffix)
+                symlinkstr = sym_suffix;
+
             if (asprintf (&nkey, "%s.%s", symlinkstr, next) < 0) {
                 saved_errno = ENOMEM;
+                free (sym_suffix);
                 goto done;
             }
             if (kvstxn_link_dirent (kt,
@@ -679,9 +706,11 @@ static int kvstxn_link_dirent (kvstxn_t *kt, int current_epoch,
                                     flags,
                                     missing_ref) < 0) {
                 saved_errno = errno;
+                free (sym_suffix);
                 free (nkey);
                 goto done;
             }
+            free (sym_suffix);
             free (nkey);
             goto success;
         } else {
@@ -731,7 +760,6 @@ static int kvstxn_link_dirent (kvstxn_t *kt, int current_epoch,
  success:
     rc = 0;
  done:
-    free (ns_prefix);
     free (key_suffix);
     free (cpy);
     if (rc < 0)