job-ingest: add frobnicator to pipeline

Problem: job-ingest needs to allow jobspec to be modified
before the validator (optionally) checks jobspec feasibility.

Add an ingest pipeline stage that calls flux-job-frobnicator,
a command line utility similar to flux-job-validator.  The pipeline
stage fulfills a future with modified jobspec, which replaces
job->jobspec before it is sent to the validator stage.

Author: garlick

src/modules/job-ingest/pipeline.c

@@ -8,7 +8,7 @@
  * SPDX-License-Identifier: LGPL-3.0
 \************************************************************/
 
-/* pipeline.c - run jobspec through ingest pipeline: frob | validate */
+/* pipeline.c - run jobspec through ingest pipeline: frobnicator | validator */
 
 #if HAVE_CONFIG_H
 #include "config.h"
@@ -26,12 +26,15 @@
 struct pipeline {
     flux_t *h;
     struct workcrew *validate;
+    struct workcrew *frobnicate;
     int process_count;
     flux_watcher_t *shutdown_timer;
     bool validator_bypass;
+    bool frobnicate_enable;
 };
 
 static const char *cmd_validator = "job-validator";
+static const char *cmd_frobnicator = "job-frobnicator";
 
 
 /* Timeout (seconds) to wait for workers to terminate when
@@ -59,14 +62,15 @@ static void shutdown_timeout_cb (flux_reactor_t *r,
 
     flux_log (pl->h,
               LOG_ERR,
-              "shutdown timed out with %d validators active",
+              "shutdown timed out with %d workers active",
               pl->process_count);
     flux_reactor_stop (r);
 }
 
 void pipeline_shutdown (struct pipeline *pl)
 {
     pl->process_count = workcrew_stop_notify (pl->validate, exit_cb, pl);
+    pl->process_count += workcrew_stop_notify (pl->frobnicate, exit_cb, pl);
     if (pl->process_count == 0)
         flux_reactor_stop (flux_get_reactor (pl->h));
     else {
@@ -76,6 +80,13 @@ void pipeline_shutdown (struct pipeline *pl)
 
 }
 
+static bool validator_bypass (struct pipeline *pl, struct job *job)
+{
+    if ((pl->validator_bypass || (job->flags & FLUX_JOB_NOVALIDATE)))
+        return true;
+    return false;
+}
+
 static flux_future_t *validate_job (struct pipeline *pl,
                                     struct job *job,
                                     flux_error_t *error)
@@ -96,20 +107,103 @@ static flux_future_t *validate_job (struct pipeline *pl,
     return NULL;
 }
 
+static flux_future_t *frobnicate_job (struct pipeline *pl,
+                                      struct job *job,
+                                      flux_error_t *error)
+{
+    json_t *input;
+    flux_future_t *f;
+
+    if (!(input = job_json_object (job, error)))
+        return NULL;
+    if (!(f = workcrew_process_job (pl->frobnicate, input))) {
+        errprintf (error, "Error passing job to frobnicator");
+        goto error;
+    }
+    json_decref (input);
+    return f;
+error:
+    ERRNO_SAFE_WRAP (json_decref, input);
+    return NULL;
+}
+
+static void frobnicate_continuation (flux_future_t *f1, void *arg)
+{
+    struct pipeline *pl = arg;
+    struct job *job = flux_future_aux_get (f1, "job");
+    const char *s;
+    json_t *jobspec;
+    const char *errmsg = NULL;
+    flux_error_t error;
+
+    if (flux_future_get (f1, (const void **)&s) < 0) {
+        errmsg = future_strerror (f1, errno);
+        goto error;
+    }
+
+    if (!(jobspec = json_loads (s, 0, NULL))) {
+        errmsg = "error decoding jobspec from frobnicator";
+        errno = EINVAL;
+        goto error;
+    }
+    json_decref (job->jobspec);
+    job->jobspec = jobspec;
+
+    if (!validator_bypass (pl, job)) {
+        flux_future_t *f2;
+
+        if (!(f2 = validate_job (pl, job, &error))) {
+            errmsg = error.text;
+            goto error;
+        }
+        if (flux_future_continue (f1, f2) < 0) {
+            flux_future_destroy (f2);
+            errmsg = "error continuing validator";
+            goto error;
+        }
+    }
+    goto done;
+error:
+    flux_future_continue_error (f1, errno, errmsg);
+done:
+    flux_future_destroy (f1);
+}
+
+/* N.B. this function could be a little simpler if futures for the pipeline
+ * stages were unconditionally chained;  instead, minimize overhead for:
+ * - frobnicator not configured
+ * - frobnicator not configured AND validator bypassed
+ */
 int pipeline_process_job (struct pipeline *pl,
                           struct job *job,
                           flux_future_t **fp,
                           flux_error_t *error)
 {
-    if ((pl->validator_bypass || (job->flags & FLUX_JOB_NOVALIDATE))) {
-        *fp = NULL;
-        return 0;
+    if (pl->frobnicate_enable) {
+        flux_future_t *f1;
+        flux_future_t *f_comp;
+
+        if (!(f1 = frobnicate_job (pl, job, error))
+            || flux_future_aux_set (f1, "job", job, NULL) < 0
+            || !(f_comp = flux_future_and_then (f1,
+                                                frobnicate_continuation,
+                                                pl))) {
+            flux_future_destroy (f1);
+            return -1;
+        }
+        *fp = f_comp;
     }
+    else {
+        flux_future_t *f;
 
-    flux_future_t *f;
-    if (!(f = validate_job (pl, job, error)))
-        return -1;
-    *fp = f;
+        if (validator_bypass (pl, job))
+            *fp = NULL;
+        else {
+            if (!(f = validate_job (pl, job, error)))
+                return -1;
+            *fp = f;
+        }
+    }
     return 0;
 }
 
@@ -182,6 +276,9 @@ int pipeline_configure (struct pipeline *pl,
     json_t *ingest = NULL;
     char *validator_plugins = NULL;
     char *validator_args = NULL;
+    char *frobnicator_plugins = NULL;
+    char *frobnicator_args = NULL;
+    int rc = -1;
 
     /* Process toml
      */
@@ -201,6 +298,13 @@ int pipeline_configure (struct pipeline *pl,
                                 &pl->validator_bypass,
                                 error) < 0)
         return -1;
+    if (unpack_ingest_subtable (ingest,
+                                "frobnicator",
+                                &frobnicator_plugins,
+                                &frobnicator_args,
+                                NULL,
+                                error) < 0)
+        return -1;
 
     /* Process module command line
      */
@@ -217,6 +321,20 @@ int pipeline_configure (struct pipeline *pl,
             pl->validator_bypass = true;
     }
 
+    if (frobnicator_plugins && strlen (frobnicator_plugins) > 0)
+        pl->frobnicate_enable = true;
+    else
+        pl->frobnicate_enable = false;
+    if (workcrew_configure (pl->frobnicate,
+                            cmd_frobnicator,
+                            frobnicator_plugins,
+                            frobnicator_args) < 0) {
+        errprintf (error,
+                   "Error (re-)configuring frobnicator workcrew: %s",
+                   strerror (errno));
+        goto error;
+    }
+
     // Checked for by t2111-job-ingest-config.t
     flux_log (pl->h,
               LOG_DEBUG,
@@ -233,23 +351,25 @@ int pipeline_configure (struct pipeline *pl,
                    strerror (errno));
         goto error;
     }
-
-    free (validator_plugins);
-    free (validator_args);
-    return 0;
+    rc = 0;
 error:
     ERRNO_SAFE_WRAP (free, validator_plugins);
     ERRNO_SAFE_WRAP (free, validator_args);
-    return -1;
+    ERRNO_SAFE_WRAP (free, frobnicator_plugins);
+    ERRNO_SAFE_WRAP (free, frobnicator_args);
+    return rc;
 }
 
 json_t *pipeline_stats_get (struct pipeline *pl)
 {
     json_t *o = NULL;
     if (pl) {
+        json_t *fo = workcrew_stats_get (pl->frobnicate);
         json_t *vo = workcrew_stats_get (pl->validate);
-        o = json_pack ("{s:O}",
+        o = json_pack ("{s:O s:O}",
+                       "frobnicator", fo,
                        "validator", vo);
+        json_decref (fo);
         json_decref (vo);
     }
     return o ? o : json_null ();
@@ -260,6 +380,7 @@ void pipeline_destroy (struct pipeline *pl)
     if (pl) {
         int saved_errno = errno;
         workcrew_destroy (pl->validate);
+        workcrew_destroy (pl->frobnicate);
         flux_watcher_destroy (pl->shutdown_timer);
         free (pl);
         errno = saved_errno;
@@ -283,6 +404,9 @@ struct pipeline *pipeline_create (flux_t *h)
     if (!(pl->validate = workcrew_create (pl->h))
         || workcrew_configure (pl->validate, cmd_validator, NULL, NULL) < 0)
         goto error;
+    if (!(pl->frobnicate = workcrew_create (pl->h))
+        || workcrew_configure (pl->frobnicate, cmd_frobnicator, NULL, NULL) < 0)
+        goto error;
     return pl;
 error:
     pipeline_destroy (pl);