shell/stage-in: use new KVS archive format

Problem: the shell stage-in plugin only works with mmapped
files.

Rename the stage-in.tags option to stage-in.keys and change
the code so it reads the main archive from the KVS.

Rework the stage-in sharness test so it does most of the same
things as before but using the new tooling.

Author: garlick

src/shell/stage-in.c

@@ -8,13 +8,14 @@
  * SPDX-License-Identifier: LGPL-3.0
 \************************************************************/
 
-/* stage-in.c - copy previously mapped files for job */
+/* stage-in.c - copy previously archived files for job */
 
 #define FLUX_SHELL_PLUGIN_NAME "stage-in"
 
 #if HAVE_CONFIG_H
 #include "config.h"
 #endif
+#include <fnmatch.h>
 #include <unistd.h>
 #include <limits.h>
 #include <stdlib.h>
@@ -42,15 +43,15 @@
 #include "info.h"
 
 struct stage_in {
-    json_t *tags;
+    json_t *names;
     const char *pattern;
     const char *destdir;
     flux_t *h;
     int count;
     size_t total_size;
 };
 
-json_t *parse_tags (const char *s, const char *default_value)
+json_t *parse_names (const char *s, const char *default_value)
 {
     char *argz = NULL;
     size_t argz_len;
@@ -107,37 +108,51 @@ static void trace_cb (void *arg,
 
 static int extract (struct stage_in *ctx)
 {
-    flux_future_t *f;
+    size_t i;
+    json_t *nameobj;
 
-    if (!(f = filemap_mmap_list (ctx->h,
-                                 false,
-                                 ctx->tags,
-                                 ctx->pattern))) {
-        shell_log_error ("mmap-list: %s", strerror (errno));
-        return -1;
-    }
-    for (;;) {
-        json_t *files;
+    json_array_foreach (ctx->names, i, nameobj) {
+        char *key = NULL;
+        flux_future_t *f = NULL;
+        json_t *archive;
         flux_error_t error;
 
-        if (flux_rpc_get_unpack (f, "{s:o}", "files", &files) < 0) {
-            if (errno == ENODATA)
-                break; // end of stream
-            shell_log_error ("mmap-list: %s", future_strerror (f, errno));
+        if (asprintf (&key, "archive.%s", json_string_value (nameobj)) < 0
+            || !(f = flux_kvs_lookup (ctx->h, "primary", 0, key))
+            || flux_kvs_lookup_get_unpack (f, "o", &archive) < 0) {
+            shell_log_error ("could not lookup %s in primary KVS namespace: %s",
+                             key,
+                             future_strerror (f, errno));
+            flux_future_destroy (f);
             return -1;
         }
+        if (ctx->pattern) {
+            size_t index = 0;
+            while (index < json_array_size (archive)) {
+                json_t *entry;
+                const char *path;
+
+                if (!(entry = json_array_get (archive, index))
+                    || json_unpack (entry, "{s:s}", "path", &path) < 0
+                    || fnmatch (ctx->pattern, path, 0) != 0) {
+                    json_array_remove (archive, index);
+                    continue;
+                }
+                index++;
+            }
+        }
         if (filemap_extract (ctx->h,
-                             files,
+                             archive,
                              0,
                              &error,
                              trace_cb,
                              ctx) < 0) {
             shell_log_error ("%s", error.text);
+            flux_future_destroy (f);
             return -1;
         }
-        flux_future_reset (f);
+        flux_future_destroy (f);
     }
-    flux_future_destroy (f);
     return 0;
 }
 
@@ -181,7 +196,7 @@ static int extract_files (struct stage_in *ctx)
 static int stage_in (flux_shell_t *shell, json_t *config)
 {
     struct stage_in ctx;
-    const char *tags = NULL;
+    const char *names = NULL;
     const char *destination = NULL;
     bool leader_only = false;
 
@@ -191,15 +206,15 @@ static int stage_in (flux_shell_t *shell, json_t *config)
     if (json_is_object (config)) {
         if (json_unpack (config,
                          "{s?s s?s s?s}",
-                         "tags", &tags,
+                         "names", &names,
                          "pattern", &ctx.pattern,
                          "destination", &destination)) {
             shell_log_error ("Error parsing stage_in shell option");
             goto error;
         }
     }
-    if (!(ctx.tags = parse_tags (tags, "main"))) {
-        shell_log_error ("Error parsing stage_in.tags shell option");
+    if (!(ctx.names = parse_names (names, "main"))) {
+        shell_log_error ("Error parsing stage_in.names shell option");
         goto error;
     }
     if (destination) {
@@ -228,10 +243,10 @@ static int stage_in (flux_shell_t *shell, json_t *config)
             goto error;
     }
 
-    json_decref (ctx.tags);
+    json_decref (ctx.names);
     return 0;
 error:
-    json_decref (ctx.tags);
+    json_decref (ctx.names);
     return -1;
 }
 

t/t2617-job-shell-stage-in.t

@@ -8,29 +8,31 @@ lptest=${FLUX_BUILD_DIR}/t/shell/lptest
 
 test_under_flux 4 job
 
-test_expect_success 'map the red test files' '
+test_expect_success 'archive the red test files' '
 	mkdir red &&
 	touch red/empty &&
 	truncate --size 8192 red/holy &&
 	$lptest >red/lptest &&
 	echo foo >red/small &&
 	mkdir red/dir &&
 	ln -s dir red/link &&
-	flux filemap map -vv --tags red red
+	flux archive create -v --name=red red
 '
-test_expect_success 'map the blue test files' '
+test_expect_success 'archive the blue test files' '
 	dir=blue/a/b/c/d/e/f/g/h &&
 	mkdir -p $dir &&
 	echo bar >$dir/test &&
-	flux filemap map -vv --tags blue blue
+	flux archive create -v --name=blue blue
 '
-test_expect_success 'map the main test files' '
+test_expect_success 'archive the main test files' '
 	mkdir main &&
 	echo "Hello world!" >main/hello &&
-	flux filemap map -vv main
+	flux archive create -v main
 '
 test_expect_success 'list all the files' '
-	flux filemap list --long --tags=red,blue,main
+	flux archive extract --list-only -v --name=red &&
+	flux archive extract --list-only -v --name=blue &&
+	flux archive extract --list-only -v
 '
 test_expect_success 'create file tree checker script' '
 	cat >check.sh <<-EOT &&
@@ -41,24 +43,18 @@ test_expect_success 'create file tree checker script' '
 	EOT
 	chmod 755 check.sh
 '
-test_expect_success 'verify that stage-in works with default tag (main)' '
+test_expect_success 'verify that stage-in works with default key (main)' '
 	flux run -N4 -ostage-in ./check.sh main
 '
-test_expect_success 'verify that stage-in works with tags=red' '
-	flux run -N2 -ostage-in.tags=red ./check.sh red
+test_expect_success 'verify that stage-in works with names=red' '
+	flux run -N2 -ostage-in.names=red ./check.sh red
 '
-test_expect_success 'verify that stage-in works with tags=red,blue' '
-	flux run -N1 -ostage-in.tags=red,blue ./check.sh red blue
-'
-test_expect_success 'verify that stage-in.direct works' '
-	flux run -N1 \
-	    -ostage-in.tags=red \
-	    -ostage-in.direct \
-	    ./check.sh red
+test_expect_success 'verify that stage-in works with names=red,blue' '
+	flux run -N1 -ostage-in.names=red,blue ./check.sh red blue
 '
 test_expect_success 'verify that stage-in.pattern works' '
 	flux run -N1 \
-	    -ostage-in.tags=red,blue \
+	    -ostage-in.names=red,blue \
 	    -ostage-in.pattern=red/* \
 	    -overbose=2 \
             ./check.sh red 2>pattern.err
@@ -99,17 +95,22 @@ test_expect_success 'verify that stage-in.destination fails on bad prefix' '
 	    -o stage-in.destination=wrong:$(pwd)/testdest \
 	    /bin/true
 '
-test_expect_success 'unmap all' '
-	flux filemap unmap --tags=red,blue,main
+test_expect_success 'remove archives' '
+	flux archive remove &&
+	flux archive remove --name=blue &&
+	flux archive remove --name=red
+'
+test_expect_success 'create a test file with random content' '
+        dd if=/dev/random of=foo bs=4096 count=1 conv=notrunc
 '
-test_expect_success 'map a test file and access it to prime the cache' '
+test_expect_success 'map test file and access it to prime the cache' '
 	mkdir -p copydir &&
-        flux filemap map ./red/lptest &&
-	flux filemap get -C copydir &&
-	cmp red/lptest copydir/red/lptest
+        flux archive create --mmap ./foo &&
+	flux archive extract -C copydir &&
+	cmp foo copydir/foo
 '
 test_expect_success 'modify mapped test file without reducing its size' '
-        dd if=/dev/zero of=red/lptest bs=4096 count=1 conv=notrunc
+        dd if=/dev/zero of=foo bs=4096 count=1 conv=notrunc
 '
 test_expect_success 'content change should cause an error' '
         test_must_fail flux run -N1 -o stage-in /bin/true 2>changed.err &&