Merge pull request #6666 from garlick/sdexec_notify

sdexec: add stop timeout to handle unkillable processes

Author: mergify[bot]

configure.ac

@@ -433,7 +433,7 @@ AS_IF([test x$enable_code_coverage = xyes], [
 AC_ARG_WITH([flux-security], AS_HELP_STRING([--with-flux-security],
              [Build with flux-security]))
 AS_IF([test "x$with_flux_security" = "xyes"], [
-    PKG_CHECK_MODULES([FLUX_SECURITY], [flux-security >= 0.13.0],
+    PKG_CHECK_MODULES([FLUX_SECURITY], [flux-security >= 0.14.0],
       [flux_sec_incdir=`$PKG_CONFIG --variable=includedir flux-security`])
     AS_IF([test "x$flux_sec_incdir" = x],
           [AC_MSG_ERROR([couldn't find flux-security include directory])])

debian/control

@@ -5,7 +5,7 @@ Maintainer: Mark A. Grondona <[email protected]>
 Standards-Version: 4.1.2
 Build-Depends:
   debhelper (>= 10),
-  flux-security (>= 0.13.0),
+  flux-security (>= 0.14.0),
   libsystemd-dev (>= 0.23),
   libarchive-dev,
   uuid-dev,

doc/man5/flux-config-exec.rst

@@ -38,6 +38,21 @@ sdexec-properties
    (optional) A table of systemd properties to set for all jobs.  All values
    must be strings. See :ref:`sdexec_properties` below.
 
+sdexec-stop-timer-sec
+   (optional) Configure the length of time in seconds after a unit enters
+   deactivating state when it will be sent the sdexec-stop-timer-signal.
+   Deactivating state is entered by ``imp-shell`` units when the
+   :man1:`flux-shell` terminates.  The unit may remain there as long as
+   user processes remain in the unit's cgroup.
+
+   After the same length of time, if the unit hasn't terminated, for example
+   due to unkillable processes, the unit is abandoned and the node is drained.
+   Default 30.
+
+sdexec-stop-timer-signal
+   (optional) Configure the signal used by the stop timer.  By default,
+   10 (SIGUSR1, the IMP proxy for SIGKILL) is used.
+
 kill-timeout
    (optional) The amount of time in FSD to wait after ``SIGTERM`` is
    sent to a job before sending ``SIGKILL``. The default is "5s". See

src/common/libsdexec/start.c

@@ -285,6 +285,21 @@ static int prop_add (json_t *prop, const char *name, const char *val)
         if (prop_add_bool (prop, name, value) < 0)
             return -1;
     }
+    else if (streq (name, "TimeoutStopUSec")) {
+        if (streq (val, "infinity")) {
+            if (prop_add_u64 (prop, name, UINT64_MAX) < 0)
+                return -1;
+        }
+        else {
+            errno = 0;
+            char *endptr;
+            uint64_t u = strtoull (val, &endptr, 10);
+            if (errno != 0 || *endptr != '\0')
+                return -1;
+            if (prop_add_u64 (prop, name, u) < 0)
+                return -1;
+        }
+    }
     else {
         if (prop_add_string (prop, name, val) < 0)
             return -1;
@@ -293,7 +308,6 @@ static int prop_add (json_t *prop, const char *name, const char *val)
 }
 
 static json_t *prop_create (json_t *cmd,
-                            const char *type,
                             int stdin_fd,
                             int stdout_fd,
                             int stderr_fd,
@@ -307,6 +321,7 @@ static json_t *prop_create (json_t *cmd,
     json_t *opts;
     const char *key;
     json_t *val;
+    bool type_is_set = false;
 
     // Not unpacked: channels
     if (json_unpack_ex (cmd,
@@ -328,7 +343,6 @@ static json_t *prop_create (json_t *cmd,
         return NULL;
     }
     if (prop_add_execstart (prop, "ExecStart", cmdline) < 0
-        || prop_add_string (prop, "Type", type) < 0
         || prop_add_string (prop, "WorkingDirectory", cwd) < 0
         || prop_add_bool (prop, "RemainAfterExit", true) < 0
         || prop_add_env (prop, "Environment", env) < 0
@@ -345,6 +359,14 @@ static json_t *prop_create (json_t *cmd,
                 errprintf (error, "%s: error setting property", key);
                 goto error;
             }
+            if (streq (key + 12, "Type"))
+                type_is_set = true;
+        }
+    }
+    if (!type_is_set) {
+        if (prop_add_string (prop, "Type", "exec") < 0) {
+            errprintf (error, "error setting property Type=simple");
+            goto error;
         }
     }
     return prop;
@@ -357,7 +379,6 @@ static json_t *prop_create (json_t *cmd,
 flux_future_t *sdexec_start_transient_unit (flux_t *h,
                                             uint32_t rank,
                                             const char *mode,
-                                            const char *type,
                                             json_t *cmd,
                                             int stdin_fd,
                                             int stdout_fd,
@@ -368,13 +389,12 @@ flux_future_t *sdexec_start_transient_unit (flux_t *h,
     json_t *prop = NULL;
     const char *name;
 
-    if (!h || !mode || !type || !cmd) {
+    if (!h || !mode || !cmd) {
         errprintf (error, "invalid argument");
         errno = EINVAL;
         return NULL;
     }
     if (!(prop = prop_create (cmd,
-                              type,
                               stdin_fd,
                               stdout_fd,
                               stderr_fd,

src/common/libsdexec/start.h

@@ -21,8 +21,6 @@
  * and systemd.service(5) for more info on mode and type parameters.
  * mode may be one of:
      replace, fail, isolate, ignore-dependencies, ignore-requirements
- * type may be one of:
- *   simple, exec, forking, oneshot, dbus, notify, notify-reload, idle
  *
  * stdin_fd, stdout_fd, and stderr_fd are file descriptors to be duped and
  * passed to the new unit.  The dup should be complete on first fulfillment
@@ -50,7 +48,6 @@
 flux_future_t *sdexec_start_transient_unit (flux_t *h,
                                             uint32_t rank,
                                             const char *mode,
-                                            const char *type,
                                             json_t *cmd,
                                             int stdin_fd,
                                             int stdout_fd,

src/common/libsdexec/test/start.c

@@ -57,7 +57,6 @@ void test_inval (void)
     ok (sdexec_start_transient_unit (NULL,      // h
                                      0,         // rank
                                      "fail",    // mode
-                                     "simple",  // type
                                      cmd_o,     // cmd
                                      -1, -1, -1, // *_fd
                                      &error) == NULL
@@ -70,7 +69,6 @@ void test_inval (void)
     ok (sdexec_start_transient_unit (h,         // h
                                      0,         // rank
                                      NULL,      // mode
-                                     "simple",  // type
                                      cmd_o,     // cmd
                                      -1, -1, -1, // *_fd
                                      &error) == NULL
@@ -83,20 +81,6 @@ void test_inval (void)
     ok (sdexec_start_transient_unit (h,         // h
                                      0,         // rank
                                      "fail",    // mode
-                                     NULL,      // type
-                                     cmd_o,     // cmd
-                                     -1, -1, -1, // *_fd
-                                     &error) == NULL
-        && errno == EINVAL,
-        "sdexec_start_transient_unit type=NULL fails with EINVAL");
-    diag ("%s", error.text);
-
-    errno = 0;
-    error.text[0] = '\0';
-    ok (sdexec_start_transient_unit (h,         // h
-                                     0,         // rank
-                                     "fail",    // mode
-                                     "simple",  // type
                                      NULL,      // cmd
                                      -1, -1, -1, // *_fd
                                      &error) == NULL
@@ -109,7 +93,6 @@ void test_inval (void)
     ok (sdexec_start_transient_unit (h,         // h
                                      0,         // rank
                                      "fail",    // mode
-                                     "simple",  // type
                                      cmd_o_noname, // cmd
                                      -1, -1, -1, // *_fd
                                      &error) == NULL
@@ -122,7 +105,6 @@ void test_inval (void)
     ok (sdexec_start_transient_unit (h,         // h
                                      0,         // rank
                                      "fail",    // mode
-                                     "simple",  // type
                                      cmd_o_badprop, // cmd
                                      -1, -1, -1, // *_fd
                                      &error) == NULL

src/modules/job-exec/exec.c

@@ -347,7 +347,42 @@ static void error_cb (struct bulk_exec *exec, flux_subprocess_t *p, void *arg)
      *   create flux_cmd_t
      */
     if (cmd) {
-        if (errnum == EHOSTUNREACH) {
+        if (errnum == EDEADLK) {
+            /*  EDEADLK from sdexec means that unkillable processes were left
+             *   on the node and it must be drained.  A "finished" response
+             *   will not have been received, so after draining, treat this
+             *   like EHOSTUNREACH.
+             */
+            char ranks[16];
+            snprintf (ranks, sizeof (ranks), "%d", rank);
+            (void) jobinfo_drain_ranks (job,
+                                        ranks,
+                                        "unkillable processes from job %s",
+                                        idf58 (job->id));
+            bool critical = is_critical_rank (job, shell_rank);
+
+            /*  Always notify rank 0 shell of a lost shell.
+             */
+            lost_shell (job,
+                        critical,
+                        shell_rank,
+                        "shell exited with unkillable processes"
+                        " on %s (shell rank %d)",
+                        hostname,
+                        shell_rank);
+
+            /*  Raise a fatal error and terminate job immediately if
+             *  the lost shell was critical.
+             */
+            if (critical)
+                jobinfo_fatal_error (job,
+                                     0,
+                                     "shell exited with unkillable processes"
+                                     " on %s (rank %d)",
+                                     hostname,
+                                     rank);
+        }
+        else if (errnum == EHOSTUNREACH) {
             bool critical = is_critical_rank (job, shell_rank);
 
             /*  Always notify rank 0 shell of a lost shell.
@@ -519,6 +554,7 @@ static int parse_service_option (json_t *jobspec,
     return 0;
 }
 
+
 static struct bulk_exec_ops exec_ops = {
     .on_start =     start_cb,
     .on_exit =      exit_cb,
@@ -604,14 +640,27 @@ static int exec_init (struct jobinfo *job)
             goto err;
         }
         /* The systemd user instance running as user flux is not privileged
-         * to signal guest processes, therefore only signal the IMP and
-         * never use SIGKILL.  See flux-framework/flux-core#6399
+         * to signal guest processes, therefore:
+         * - Set the KillMode=process so only the IMP is signaled
+         * - Use Type=notify in conjunction with IMP calling sd_notify(3) so
+         *   the unit transitions to deactivating when the shell exits.
+         * - Set TimeoutStopUsec=infinity to disable systemd's stop timeout.
+         * - Enable sdexec's stop timeout which is armed at deactivating,
+         *   delivers SIGUSR1 (proxy for SIGKILL) after 30s, then abandons
+         *   the unit and terminates the exec RPC after another 30s.
          */
         if (streq (service, "sdexec")) {
             if (flux_cmd_setopt (cmd, "SDEXEC_PROP_KillMode", "process") < 0
+                || flux_cmd_setopt (cmd, "SDEXEC_PROP_Type", "notify") < 0
+                || flux_cmd_setopt (cmd,
+                                    "SDEXEC_PROP_TimeoutStopUSec",
+                                    "infinity") < 0
+                || flux_cmd_setopt (cmd,
+                                    "SDEXEC_STOP_TIMER_SIGNAL",
+                                    config_get_sdexec_stop_timer_signal ()) < 0
                 || flux_cmd_setopt (cmd,
-                                    "SDEXEC_PROP_SendSIGKILL",
-                                    "off") < 0) {
+                                    "SDEXEC_STOP_TIMER_SEC",
+                                    config_get_sdexec_stop_timer_sec ()) < 0) {
                 flux_log_error (job->h,
                                 "Unable to set multiuser sdexec options");
                 return -1;

src/modules/job-exec/exec_config.c

@@ -36,6 +36,8 @@ struct exec_config {
     const char *exec_service;
     int exec_service_override;
     json_t *sdexec_properties;
+    int sdexec_stop_timer_sec;
+    int sdexec_stop_timer_signal;
     double default_barrier_timeout;
 };
 
@@ -107,6 +109,20 @@ json_t *config_get_sdexec_properties (void)
     return exec_conf.sdexec_properties;
 }
 
+const char *config_get_sdexec_stop_timer_sec (void)
+{
+    static char buf[32];
+    snprintf (buf, sizeof (buf), "%d", exec_conf.sdexec_stop_timer_sec);
+    return buf;
+}
+
+const char *config_get_sdexec_stop_timer_signal  (void)
+{
+    static char buf[32];
+    snprintf (buf, sizeof (buf), "%d", exec_conf.sdexec_stop_timer_signal);
+    return buf;
+}
+
 double config_get_default_barrier_timeout (void)
 {
     return exec_conf.default_barrier_timeout;
@@ -116,15 +132,19 @@ int config_get_stats (json_t **config_stats)
 {
     json_t *o = NULL;
 
-    if (!(o = json_pack ("{s:s? s:s? s:s? s:s? s:i s:f}",
+    if (!(o = json_pack ("{s:s? s:s? s:s? s:s? s:i s:f s:i s:i}",
                          "default_cwd", default_cwd,
                          "default_job_shell", exec_conf.default_job_shell,
                          "flux_imp_path", exec_conf.flux_imp_path,
                          "exec_service", exec_conf.exec_service,
                          "exec_service_override",
                          exec_conf.exec_service_override,
                          "default_barrier_timeout",
-                         exec_conf.default_barrier_timeout))) {
+                         exec_conf.default_barrier_timeout,
+                         "sdexec_stop_timer_sec",
+                         exec_conf.sdexec_stop_timer_sec,
+                         "sdexec_stop_timer_signal",
+                         exec_conf.sdexec_stop_timer_signal))) {
         errno = ENOMEM;
         return -1;
     }
@@ -153,6 +173,8 @@ static void exec_config_init (struct exec_config *ec)
     ec->exec_service = "rexec";
     ec->exec_service_override = 0;
     ec->sdexec_properties = NULL;
+    ec->sdexec_stop_timer_sec = 30;
+    ec->sdexec_stop_timer_signal = 10; // SIGUSR1
     ec->default_barrier_timeout = 1800.;
 }
 
@@ -249,6 +271,22 @@ int config_setup (flux_t *h,
         }
     }
 
+    /*  Check configuration for exec.stop-timer-* */
+    if (flux_conf_unpack (conf,
+                          &err,
+                          "{s?{s?i s?i}}",
+                          "exec",
+                            "sdexec-stop-timer-sec",
+                              &tmpconf.sdexec_stop_timer_sec,
+                            "sdexec-stop-timer-signal",
+                              &tmpconf.sdexec_stop_timer_signal) < 0) {
+        errprintf (errp,
+                   "error reading config values exec.sdexec-stop-timer-sec: %s"
+                   " or exec.sdexec-stop-timer-signal",
+                   err.text);
+        return -1;
+    }
+
     /*  Check configuration for exec.barrier-timeout */
     if (flux_conf_unpack (conf,
                           &err,

src/modules/job-exec/exec_config.h

@@ -38,6 +38,10 @@ double config_get_default_barrier_timeout (void);
 
 int config_get_stats (json_t **config_stats);
 
+const char *config_get_sdexec_stop_timer_sec (void);
+
+const char *config_get_sdexec_stop_timer_signal (void);
+
 int config_setup (flux_t *h,
                   const flux_conf_t *conf,
                   int argc,