Merge pull request #5202 from grondo/broker-signals

broker: foward nonfatal signals to all running jobs

Author: mergify[bot]

src/broker/broker.c

@@ -45,6 +45,7 @@
 #include "src/common/libfluxutil/method.h"
 #include "ccan/array_size/array_size.h"
 #include "ccan/str/str.h"
+#include "ccan/ptrint/ptrint.h"
 
 #include "module.h"
 #include "brokercfg.h"
@@ -1308,7 +1309,8 @@ static void broker_destroy_sigwatcher (void *data)
 
 static int broker_handle_signals (broker_ctx_t *ctx)
 {
-    int i, sigs[] = { SIGHUP, SIGINT, SIGQUIT, SIGTERM, SIGALRM };
+    int i, sigs[] = { SIGHUP, SIGINT, SIGQUIT, SIGTERM,
+                      SIGALRM, SIGUSR1, SIGUSR2 };
     int blocked[] = { SIGPIPE };
     flux_watcher_t *w;
 
@@ -1965,13 +1967,80 @@ static void module_status_cb (module_t *p, int prev_status, void *arg)
     }
 }
 
+static bool signal_is_deadly (int signum)
+{
+    int deadly_sigs[] = { SIGHUP, SIGINT, SIGQUIT, SIGTERM, SIGALRM };
+    for (int i = 0; i < ARRAY_SIZE (deadly_sigs); i++) {
+        if (signum == deadly_sigs[i])
+            return true;
+    }
+    return false;
+}
+
+static void killall_cb (flux_future_t *f, void *arg)
+{
+    broker_ctx_t *ctx = arg;
+    int count = 0;
+    if (flux_rpc_get_unpack (f, "{s:i}", "count", &count) < 0) {
+        flux_log_error (ctx->h,
+                        "job-manager.killall: %s",
+                        future_strerror (f, errno));
+    }
+    flux_future_destroy (f);
+    if (count) {
+        flux_log (ctx->h,
+                  LOG_INFO,
+                  "forwarded signal %d to %d jobs",
+                  (int) ptr2int (flux_future_aux_get (f, "signal")),
+                  count);
+    }
+}
+
+static int killall_jobs (broker_ctx_t *ctx, int signum)
+{
+    flux_future_t *f = NULL;
+    if (!(f = flux_rpc_pack (ctx->h,
+                             "job-manager.killall",
+                             FLUX_NODEID_ANY,
+                             0,
+                             "{s:b s:i s:i}",
+                             "dry_run", 0,
+                             "userid", FLUX_USERID_UNKNOWN,
+                             "signum", signum))
+        || flux_future_then (f, -1., killall_cb, ctx) < 0) {
+            flux_future_destroy (f);
+            return -1;
+    }
+    if (flux_future_aux_set (f, "signum", int2ptr (signum), NULL) < 0)
+        flux_log_error (ctx->h, "killall: future_aux_set");
+    return 0;
+}
+
 static void signal_cb (flux_reactor_t *r, flux_watcher_t *w,
-                         int revents, void *arg)
+                       int revents, void *arg)
 {
     broker_ctx_t *ctx = arg;
     int signum = flux_signal_watcher_get_signum (w);
 
     flux_log (ctx->h, LOG_INFO, "signal %d", signum);
+
+    if (ctx->rank == 0 && !signal_is_deadly (signum)) {
+        /* Attempt to forward non-deadly signals to jobs. If that fails,
+         * then fall through to state_machine_kill() so the signal is
+         * delivered somewhere.
+         */
+        if (killall_jobs (ctx, signum) == 0)
+            return;
+        /*
+         * Note: flux_rpc(3) in the rank 0 broker to the job manager module
+         *  is expected to fail immediately if the job-manager module is not
+         *  loaded due to the broker internal flux_t handle implementation.
+         */
+        flux_log (ctx->h,
+                  LOG_INFO,
+                  "killall failed, delivering signal %d locally instead",
+                  signum);
+    }
     state_machine_kill (ctx->state_machine, signum);
 }
 

src/broker/runat.c

@@ -465,9 +465,9 @@ int runat_push_shell_command (struct runat *r,
 
     /*   For shell commands run the target cmdline in a separate process
      *   group so that any processes spawned by the shell will be signaled
-     *   in runat_abort(). This is probably unnecessary for single commands,
-     *   and does not work for an interactive shell (seems to disable access
-     *   to the pty), so we set the flag only here for now.
+     *   in runat_abort(). This does not work for an interactive shell
+     *   (seems to disable access to the pty), so this flag is not set in
+     *   runat_push_shell().
      */
     cmd->flags |= FLUX_SUBPROCESS_FLAGS_SETPGRP;
 
@@ -522,6 +522,13 @@ int runat_push_command (struct runat *r,
     }
     if (!(cmd = runat_command_create (environ, flags)))
         return -1;
+
+    /*   Run the target cmdline in a separate process group so that any
+     *   processes spawned by the new process are also signaled in
+     *   runat_abort().
+     */
+    cmd->flags |= FLUX_SUBPROCESS_FLAGS_SETPGRP;
+
     if (runat_command_set_argz (cmd, argz, argz_len) < 0)
         goto error;
     if (runat_command_modenv (cmd, env_blocklist, r->local_uri) < 0)

t/Makefile.am

@@ -344,6 +344,7 @@ dist_check_SCRIPTS = \
 	issues/t4711-job-list-purge-inactive.sh \
 	issues/t4771-flux-start-bash.sh \
 	issues/t4852-t_submit-legacy.sh \
+	issues/t5105-signal-propagation.sh \
 	python/__init__.py \
 	python/subflux.py \
 	python/tap \

t/issues/t5105-signal-propagation.sh

@@ -0,0 +1,36 @@
+#!/bin/sh
+# Run an 3 level nested instance, with a test job at the final level
+
+waitfile=$SHARNESS_TEST_SRCDIR/scripts/waitfile.lua
+
+cat <<EOF >test.py
+import signal
+import time
+import flux
+import os
+
+h = flux.Flux()
+level = h.attr_get("instance-level")
+jobid = os.getenv("FLUX_JOB_ID")
+signal.signal(
+    signal.SIGUSR1,
+    lambda x, y: print(f"job {jobid} in level {level} got SIGUSR1", flush=True),
+)
+open("ready", 'a').close()
+signal.pause()
+EOF
+
+id=$(flux submit --output=log flux start \
+     flux run flux start \
+     flux run flux python ./test.py)
+
+$waitfile -t 30 -v ready
+
+flux job kill -s SIGUSR1 $id
+
+$waitfile -t 30 -v -p "got SIGUSR1" log
+
+flux job status --json -v $id
+
+# vi: ts=4 sw=4 expandtab
+