src/plugins/runtime/oci/runtime_oci.c

/*****************************************************************************\
 *  runtime_oci.c - OCI job runtime plugin.
 *****************************************************************************
 *  Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 *
 *  This file is part of Slurm, a resource management program.
 *  For details, see <https://slurm.schedmd.com/>.
 *  Please also read the included file: DISCLAIMER.
 *
 *  Slurm is free software; you can redistribute it and/or modify it under
 *  the terms of the GNU General Public License as published by the Free
 *  Software Foundation; either version 2 of the License, or (at your option)
 *  any later version.
 *
 *  In addition, as a special exception, the copyright holders give permission
 *  to link the code of portions of this program with the OpenSSL library under
 *  certain conditions as described in each individual source file, and
 *  distribute linked combinations including the two. You must obey the GNU
 *  General Public License in all respects for all of the code used other than
 *  OpenSSL. If you modify file(s) with this exception, you may extend this
 *  exception to your version of the file(s), but you are not obligated to do
 *  so. If you do not wish to do so, delete this exception statement from your
 *  version.  If you delete this exception statement from all source files in
 *  the program, then also delete it here.
 *
 *  Slurm is distributed in the hope that it will be useful, but WITHOUT ANY
 *  WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
 *  FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more
 *  details.
 *
 *  You should have received a copy of the GNU General Public License along
 *  with Slurm; if not, write to the Free Software Foundation, Inc.,
 *  51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA.
\*****************************************************************************/

#include "config.h"

#include <fcntl.h>
#include <getopt.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <unistd.h>

#include "slurm/slurm.h"

#include "src/common/data.h"
#include "src/common/fd.h"
#include "src/common/oci_config.h"
#include "src/common/optz.h"
#include "src/common/read_config.h"
#include "src/common/run_command.h"
#include "src/common/slurm_opt.h"
#include "src/common/xmalloc.h"
#include "src/common/xstring.h"

#include "src/interfaces/runtime.h"
#include "src/interfaces/serializer.h"

#include "src/slurmd/common/privileges.h"
#include "src/slurmd/slurmstepd/slurmstepd.h"

/* Required Slurm plugin symbols: */
const char plugin_name[] = "OCI runtime plugin";
const char plugin_type[] = "runtime/oci";
const uint32_t plugin_version = SLURM_VERSION_NUMBER;

static bool _on_container_arg(const runtime_arg_t *arg,
			      const uint32_t het_job_inx, data_t *value,
			      bool set_by_cli, bool set_by_env,
			      bool set_by_data, bool early_pass);

static const runtime_arg_t runtime_arg_container = {
	.argument = "container",
	.environment = {
		.salloc = "SALLOC_CONTAINER",
		.sbatch = "SBATCH_CONTAINER",
		.scrun = "SCRUN_CONTAINER",
		.slurm = "SLURM_CONTAINER",
		.srun = "SRUN_CONTAINER",
	},
	.field = "container",
	.field_desc = "Absolute path to OCI container bundle",
	.field_type = DATA_PARSER_STRING,
	.help_value = "path",
	.on_arg = _on_container_arg,
};

#define CONTAINER_BUNDLE_CNT 10
/* Path to OCI container bundle, indexed by heterogeneous component */
static char **container_bundle = NULL;
/* Entries allocated in container_bundle */
static uint32_t container_bundle_cnt = 0;

/*
 * We need a location inside of the container that is controlled by Slurm to
 * pass the startup script and I/O handling for batch steps.
 *
 * /tmp/slurm was chosen since runc will always mount it private
 */
#define SLURM_CONTAINER_BATCH_SCRIPT "/tmp/slurm/startup"
#define SLURM_CONTAINER_ENV_FILE "environment"
#define SLURM_CONTAINER_STDIN "/tmp/slurm/stdin"
#define SLURM_CONTAINER_STDOUT "/tmp/slurm/stdout"
#define SLURM_CONTAINER_STDERR "/tmp/slurm/stderr"

#define SLURM_CONTAINER_BATCH_STEP_PATTERN "oci-job%j-batch"
#define SLURM_CONTAINER_INTERACT_STEP_PATTERN "oci-job%j-interactive"
#define SLURM_CONTAINER_EXTERN_STEP_PATTERN "oci-job%j-extern"
#define SLURM_CONTAINER_STEP_PATTERN "oci-job%j-%s"
#define SLURM_CONTAINER_TASK_PATTERN "task-%t"

oci_conf_t *oci_conf = NULL;

static char *create_argv[] = {
	"/bin/sh", "-c",
	"echo 'RunTimeCreate never configured in oci.conf'; exit 1", NULL
};
static char *delete_argv[] = {
	"/bin/sh", "-c",
	"echo 'RunTimeDelete never configured in oci.conf'; exit 1", NULL
};
static char *kill_argv[] = {
	"/bin/sh", "-c",
	"echo 'RunTimeKill never configured in oci.conf'; exit 1", NULL
};
static char *query_argv[] = {
	"/bin/sh", "-c",
	"echo 'RunTimeQuery never configured in oci.conf'; exit 1", NULL
};
static char *run_argv[] = {
	"/bin/sh", "-c",
	"echo 'RunTimeRun never configured in oci.conf'; exit 1", NULL
};
static char *start_argv[] = {
	"/bin/sh", "-c",
	"echo 'RunTimeStart never configured in oci.conf'; exit 1", NULL
};

static char *_get_config_path(stepd_step_rec_t *step);
static char *_generate_spooldir(slurmd_conf_t *conf, stepd_step_rec_t *step,
				stepd_step_task_info_t *task);
static void _generate_patterns(slurmd_conf_t *conf, stepd_step_rec_t *step,
			       stepd_step_task_info_t *task);

static void _dump_command_args(run_command_args_t *args, const char *caller)
{
	if (get_log_level() < LOG_LEVEL_DEBUG3)
		return;

	for (int i = 0; args->script_argv[i]; i++)
		debug3("%s: command argv[%d]=%s",
		       caller, i, args->script_argv[i]);
}

static void _pattern_argv(char **buffer, char **offset, char **cmd_args)
{
	for (char **arg = cmd_args; arg && *arg; arg++) {
		if (arg != cmd_args)
			xstrfmtcatat(*buffer, offset, " ");

		xstrfmtcatat(*buffer, offset, "'");

		/*
		 * POSIX 1003.1 2.2.2 only bans a single quote in single quotes
		 * for escaping
		 */

		for (char *c = *arg; *c != '\0'; c++) {
			if (*c == '\'')
				xstrfmtcatat(*buffer, offset, "'\"'\"'");
			else
				xstrfmtcatat(*buffer, offset, "%c", *c);
		}

		xstrfmtcatat(*buffer, offset, "'");
	}
}

static char *_generate_pattern(slurmd_conf_t *conf, stepd_step_rec_t *step,
			       const char *pattern, int task_id,
			       char **cmd_args)
{
	step_container_t *c = step->container;
	char *buffer = NULL, *offset = NULL;

	xassert(c->magic == STEP_CONTAINER_MAGIC);

	if (!pattern)
		return NULL;

	xassert((task_id == -1) || (step->node_tasks >= task_id));

	for (const char *b = pattern; *b; b++) {
		if (*b == '%') {
			switch (*(++b)) {
			case '%':
				xstrfmtcatat(buffer, &offset, "%s", "%");
				break;
			case '@':
				if (cmd_args)
					_pattern_argv(&buffer, &offset,
						      cmd_args);
				else
					xstrfmtcatat(buffer, &offset,
						     "\"/bin/false\"");
				break;
			case 'b':
				xstrfmtcatat(buffer, &offset, "%s", c->bundle);
				break;
			case 'e':
				xstrfmtcatat(buffer, &offset, "%s/%s",
					     c->task_spool_dir,
					     SLURM_CONTAINER_ENV_FILE);
				break;
			case 'j':
				xstrfmtcatat(buffer, &offset, "%u",
					     step->step_id.job_id);
				break;
			case 'm':
				if (c->spool_dir)
					xstrfmtcatat(buffer, &offset, "%s",
						     c->spool_dir);
				else
					xstrfmtcatat(buffer, &offset, "%s",
						     conf->spooldir);
				break;
			case 'n':
				xstrfmtcatat(buffer, &offset, "%s",
					     step->node_name);
				break;
			case 'p':
				if (task_id >= 0)
					xstrfmtcatat(buffer, &offset, "%u",
						     step->task[task_id]->pid);
				else
					xstrfmtcatat(buffer, &offset, "%u",
						     INFINITE);
				break;
			case 'r':
				xstrfmtcatat(buffer, &offset, "%s", c->rootfs);
				break;
			case 's':
				xstrfmtcatat(buffer, &offset, "%u",
					     step->step_id.step_id);
				break;
			case 't':
				xstrfmtcatat(buffer, &offset, "%d", task_id);
				break;
			case 'u':
				xstrfmtcatat(buffer, &offset, "%s",
					     step->user_name);
				break;
			case 'U':
				xstrfmtcatat(buffer, &offset, "%u", step->uid);
				break;
			case 'Z':
				xstrfmtcatat(buffer, &offset, "%s",
					     c->work_dir);
				break;
			default:
				fatal("%s: unexpected replacement character: %c",
				      __func__, *b);
			}
		} else {
			xstrfmtcatat(buffer, &offset, "%c", *b);
		}
	}

	return buffer;
}

static int _mkdir(const char *pathname, mode_t mode, uid_t uid, gid_t gid)
{
	int rc;

	if ((rc = mkdir(pathname, mode)))
		rc = errno;
	else {
		/*
		 * Directory was successfully created so it needs user:group set
		 */
		if (chown(pathname, uid, gid) < 0) {
			error("%s: chown(%s): %m", __func__, pathname);
			return errno;
		}

		if (chmod(pathname, mode) < 0) {
			error("%s: chmod(%s, 750): %m", __func__, pathname);
			return errno;
		}

		debug("%s: created %s for %u:%u mode %o",
		      __func__, pathname, uid, gid, mode);

		return SLURM_SUCCESS;
	}

	if (rc == EEXIST)
		return SLURM_SUCCESS;

	error("%s: unable to mkdir(%s): %s",
	      __func__, pathname, slurm_strerror(rc));

	return rc;
}

/*
 * Create entire directory path while setting uid:gid for every newly created
 * directory.
 */
static int _mkpath(const char *pathname, uid_t uid, gid_t gid)
{
	static const mode_t mode = S_IRWXU | S_IRWXG;
	int rc;
	char *p, *dst;

	p = dst = xstrdup(pathname);

	while ((p = xstrchr(p + 1, '/'))) {
		*p = '\0';

		if ((rc = _mkdir(dst, mode, uid, gid)))
			goto cleanup;

		*p = '/';
	}

	/* final directory */
	rc = _mkdir(dst, mode, uid, gid);

cleanup:
	xfree(dst);
	return rc;
}

static int _load_config(stepd_step_rec_t *step)
{
	step_container_t *c = step->container;
	int rc;
	buf_t *buffer = NULL;
	char *path = _get_config_path(step);

	xassert(c->magic == STEP_CONTAINER_MAGIC);
	xassert(!c->config);
	xassert(path);

	errno = SLURM_SUCCESS;
	if (!(buffer = create_mmap_buf(path))) {
		rc = errno;
		error("%s: unable to open: %s", __func__, path);
		goto cleanup;
	}

	if ((rc = serialize_g_string_to_data(&c->config, get_buf_data(buffer),
					     remaining_buf(buffer),
					     MIME_TYPE_JSON))) {
		error("%s: unable to parse %s: %s",
		      __func__, path, slurm_strerror(rc));
	}

cleanup:
	FREE_NULL_BUFFER(buffer);
	xfree(path);
	return rc;
}

static int _write_config(stepd_step_rec_t *step, const char *jconfig,
			 const char *out)
{
	int outfd = -1;
	int rc = SLURM_SUCCESS;

	outfd = open(jconfig, (O_WRONLY | O_CREAT | O_EXCL), 0600);
	if (outfd < 0) {
		error("%s: unable to open %s: %m",
		      __func__, jconfig);
		goto rwfail;
	}

	safe_write(outfd, out, strlen(out));

	if (fsync_and_close(outfd, jconfig)) {
		outfd = -1;
		error("%s: failure sync and close of config: %s",
		      __func__, slurm_strerror(rc));
		goto rwfail;
	}

	outfd = -1;

	if (chown(jconfig, (uid_t) -1, (gid_t) step->gid) < 0) {
		error("%s: chown(%s): %m", __func__, jconfig);
		goto rwfail;
	}

	if (chmod(jconfig, 0750) < 0) {
		error("%s: chmod(%s, 750): %m", __func__, jconfig);
		goto rwfail;
	}

	return rc;

rwfail:
	rc = errno;

	if (outfd >= 0)
		close(outfd);

	return rc;
}

static bool _match_env(const data_t *data, void *needle)
{
	bool match;
	const char *needle_name = needle;
	char *name = NULL, *value;

	if (data_get_string_converted(data, &name)) {
		xfree(name);
		return false;
	}

	value = xstrstr(name, "=");

	if (value)
		*value = '\0';

	match = !xstrcmp(name, needle_name);

	if (match)
		debug3("%s: collapsing duplicate %s in /process/env",
		       __func__, name);

	xfree(name);

	return match;
}

static int _modify_config(stepd_step_rec_t *step, stepd_step_task_info_t *task)
{
	step_container_t *c = step->container;
	int rc = SLURM_SUCCESS;
	data_t *mnts, *env, *args;

	xassert(c->magic == STEP_CONTAINER_MAGIC);

	data_set_bool(data_define_dict_path(c->config, "/process/terminal/"),
		      (step->flags & LAUNCH_PTY));

	/* point to correct rootfs */
	data_set_string(data_define_dict_path(c->config, "/root/path/"),
			c->rootfs);

	mnts = data_define_dict_path(c->config, "/mounts/");
	if (data_get_type(mnts) != DATA_TYPE_LIST)
		data_set_list(mnts);

	if (c->mount_spool_dir) {
		data_t *mnt = data_set_dict(data_list_append(mnts));
		data_t *opt = data_set_list(data_key_set(mnt, "options"));
		data_set_string(data_key_set(mnt, "destination"),
				c->mount_spool_dir);
		data_set_string(data_key_set(mnt, "type"), "none");
		data_set_string(data_key_set(mnt, "source"), c->spool_dir);
		data_set_string(data_list_append(opt), "bind");
	}

	if (step->batch) {
		data_t *mnt, *opt;

		/*
		 * /dev/null has very special handling in runc and we must make
		 * sure to not conflict with that:
		 * https://github.com/opencontainers/runc/blob/master/libcontainer/rootfs_linux.go#L610-L613
		 */
		if (xstrcmp(step->task[0]->ifname, "/dev/null")) {
			data_t *mnt = data_set_dict(data_list_append(mnts));
			data_t *opt =
				data_set_list(data_key_set(mnt, "options"));
			data_set_string(data_key_set(mnt, "destination"),
					SLURM_CONTAINER_STDIN);
			data_set_string(data_key_set(mnt, "type"), "none");
			data_set_string(data_key_set(mnt, "source"),
					step->task[0]->ifname);
			data_set_string(data_list_append(opt), "bind");
		}

		/* Bind mount stdout */
		if (xstrcmp(step->task[0]->ofname, "/dev/null")) {
			data_t *mnt = data_set_dict(data_list_append(mnts));
			data_t *opt =
				data_set_list(data_key_set(mnt, "options"));

			data_set_string(data_key_set(mnt, "destination"),
					SLURM_CONTAINER_STDOUT);
			data_set_string(data_key_set(mnt, "type"), "none");
			data_set_string(data_key_set(mnt, "source"),
					step->task[0]->ofname);
			data_set_string(data_list_append(opt), "bind");
		}

		/* Bind mount stderr */
		if (xstrcmp(step->task[0]->efname, "/dev/null")) {
			data_t *mnt = data_set_dict(data_list_append(mnts));
			data_t *opt =
				data_set_list(data_key_set(mnt, "options"));

			data_set_string(data_key_set(mnt, "destination"),
					SLURM_CONTAINER_STDERR);
			data_set_string(data_key_set(mnt, "type"), "none");
			data_set_string(data_key_set(mnt, "source"),
					step->task[0]->efname);
			data_set_string(data_list_append(opt), "bind");
		}

		/*
		 * Add bind mount of the batch script to allow
		 * the container to execute it directly
		 */
		mnt = data_set_dict(data_list_append(mnts));
		opt = data_set_list(data_key_set(mnt, "options"));

		data_set_string(data_key_set(mnt, "destination"),
				SLURM_CONTAINER_BATCH_SCRIPT);
		data_set_string(data_key_set(mnt, "type"), "none");
		data_set_string_own(data_key_set(mnt, "source"),
				    step->task[0]->argv[0]);
		step->task[0]->argv[0] = xstrdup(SLURM_CONTAINER_BATCH_SCRIPT);
		data_set_string(data_list_append(opt), "bind");
		data_set_string(data_list_append(opt), "ro");
	}

	if (oci_conf->disable_hooks) {
		data_t *hooks = data_resolve_dict_path(c->config, "/hooks/");

		for (int i = 0; oci_conf->disable_hooks[i]; i++) {
			data_t *hook =
				data_key_get(hooks, oci_conf->disable_hooks[i]);

			if (hook) {
				int count = 0;

				if (data_get_type(hook) == DATA_TYPE_LIST) {
					count = data_get_list_length(hook);
				} else {
					error("Invalid type for hook %s",
					      oci_conf->disable_hooks[i]);
				}

				debug("%s: hook %s found and disabled %d entries",
				      __func__, oci_conf->disable_hooks[i],
				      count);

				data_key_unset(hooks,
					       oci_conf->disable_hooks[i]);
			} else {
				debug("%s: hook %s not found",
				      __func__,  oci_conf->disable_hooks[i]);
			}
		}
	}

	/* overwrite environ with the final step->env contents */
	env = data_set_list(data_define_dict_path(c->config, "/process/env/"));
	for (char **ptr = step->env; *ptr; ptr++) {
		data_t *entry;
		char *name = xstrdup(*ptr);
		char *value = xstrstr(name, "=");

		if (value)
			*value = '\0';

		if (!(entry = data_list_find_first(env, _match_env, name)))
			entry = data_list_append(env);

		data_set_string(entry, *ptr);
		xfree(name);
	}

	args = data_define_dict_path(c->config, "/process/args/");
	data_set_list(args);

	/* move args to the config.json for runtime to handle */
	for (int i = 0; i < task->argc; i++) {
		data_set_string_own(data_list_append(args), task->argv[i]);
		task->argv[i] = NULL;
	}

	return rc;
}

static int _generate_container_paths(slurmd_conf_t *conf,
				     stepd_step_rec_t *step)
{
	step_container_t *c = step->container;
	int rc = SLURM_SUCCESS;

	xassert(c->magic == STEP_CONTAINER_MAGIC);

	if (c->config) {
		if ((rc = data_retrieve_dict_path_string(c->config,
							 "/root/path/",
							 &c->rootfs))) {
			debug("%s: unable to find /root/path/", __func__);
			return rc;
		}

		if (c->rootfs[0] != '/') {
			/* always provide absolute path */
			char *t = NULL;

			xstrfmtcat(t, "%s/%s", c->bundle, c->rootfs);
			SWAP(c->rootfs, t);
			xfree(t);
		}
	} else {
		/* default to bundle path without config.json */
		c->rootfs = xstrdup(c->bundle);
	}

	/* generate step's spool_dir */
	if (oci_conf->mount_spool_dir) {
		c->mount_spool_dir =
			_generate_pattern(conf, step, oci_conf->mount_spool_dir,
					  step->task[0]->id, NULL);
	} else {
		c->mount_spool_dir = _generate_spooldir(conf, step, NULL);
	}

	if (!c->spool_dir)
		c->spool_dir = _generate_spooldir(conf, step, NULL);

	if ((rc = _mkpath(c->spool_dir, step->uid, step->gid)))
		fatal("%s: unable to create spool directory %s: %s",
		      __func__, c->spool_dir, slurm_strerror(rc));

	return rc;
}

static bool _pattern_has_taskid(const char *pattern)
{
	const char *p = pattern;

	while (*p) {
		if (!(p = xstrchr(p, '%')))
			break;

		if ((p[1] == '%') && (p[2] != '\0'))
			p += 2;
		else if (p[1] == 't')
			return true;
		else
			p++;
	}

	return false;
}

static char *_generate_spooldir_pattern(slurmd_conf_t *conf,
					stepd_step_rec_t *step,
					stepd_step_task_info_t *task)
{
	const char *step_id = NULL;
	const char *task_id = NULL;
	char *parent = NULL, *pattern = NULL;

	if (oci_conf->container_path) {
		parent = xstrdup(oci_conf->container_path);
	} else if (conf->spooldir) {
#ifdef MULTIPLE_SLURMD
		parent = slurm_conf_expand_slurmd_path(conf->spooldir,
						       conf->node_name,
						       conf->hostname);
#else /* !MULTIPLE_SLURMD */
		parent = xstrdup(conf->spooldir);
#endif /* !MULTIPLE_SLURMD */
	} else {
		parent = xstrdup(DEFAULT_SPOOLDIR);
	}

	if (step->step_id.step_id == SLURM_BATCH_SCRIPT) {
		step_id = SLURM_CONTAINER_BATCH_STEP_PATTERN;
	} else if (step->step_id.step_id == SLURM_INTERACTIVE_STEP) {
		step_id = SLURM_CONTAINER_INTERACT_STEP_PATTERN;
	} else if (step->step_id.step_id == SLURM_EXTERN_CONT) {
		step_id = SLURM_CONTAINER_EXTERN_STEP_PATTERN;
	} else {
		step_id = SLURM_CONTAINER_STEP_PATTERN;
	}

	if (task)
		task_id = SLURM_CONTAINER_TASK_PATTERN;
	else
		task_id = "";

	pattern = xstrdup_printf("%s%s%s%s%s", parent, (step_id[0] ? "/" : ""),
				 step_id, (task_id[0] ? "/" : ""), task_id);
	xfree(parent);
	return pattern;
}

static char *_generate_spooldir(slurmd_conf_t *conf, stepd_step_rec_t *step,
				stepd_step_task_info_t *task)
{
	int id = -1;
	char **argv = NULL, *path = NULL;
	char *pattern = _generate_spooldir_pattern(conf, step, task);

	if (task) {
		id = task->id;
		argv = task->argv;
	} else {
		char *start, *end, *next;

		/* trim pattern at first taskid replacement */

		if (pattern[0] == '/')
			next = pattern + 1;
		else
			next = pattern;

		while (next) {
			char term;

			start = next;

			if (!(end = xstrchr(next, '/'))) {
				end = start + strlen(start);
				next = NULL;
			} else {
				next = end + 1;
			}

			term = *end;
			*end = '\0';

			if (_pattern_has_taskid(start)) {
				/* cut pattern at this directory */
				*start = '\0';
				*end = term;
				break;
			}

			*end = term;
		}
	}

	xassert((id != -1) || !xstrstr(pattern, "%t"));
	path = _generate_pattern(conf, step, pattern, id, argv);
	debug3("%s: task:%d pattern:%s path:%s", __func__, id, pattern, path);
	xfree(pattern);

	return path;
}

static void _container_task_init(slurmd_conf_t *conf, stepd_step_rec_t *step,
				 stepd_step_task_info_t *task)
{
	int rc;
	step_container_t *c = step->container;

	if (!oci_conf) {
		debug2("%s: ignoring step container when oci.conf not configured",
		       __func__);
		return;
	}

	if (c->task_spool_dir || !c->spool_dir)
		fatal_abort("task spool dir already set or spool dir not set");

	/* generate the task_spool_dir now we know the task */
	c->task_spool_dir = _generate_spooldir(conf, step, task);

	if ((rc = _mkpath(c->task_spool_dir, step->uid, step->gid)))
		fatal("%s: unable to create spool directory %s: %s",
		      __func__, c->task_spool_dir, slurm_strerror(rc));
}

static char *_get_config_path(stepd_step_rec_t *step)
{
	step_container_t *c = step->container;
	char *path = NULL;

	if (!step->container)
		return NULL;

	xassert(c->magic == STEP_CONTAINER_MAGIC);

	/* OCI runtime spec requires config.json to be in root of bundle */
	xstrfmtcat(path, "%s/config.json", c->bundle);

	return path;
}

static data_for_each_cmd_t _foreach_config_env(const data_t *data, void *arg)
{
	stepd_step_rec_t *step = arg;
	int rc;
	char *name = NULL, *value;

	if (data_get_string_converted(data, &name))
		return DATA_FOR_EACH_FAIL;

	value = xstrstr(name, "=");

	if (value) {
		*value = '\0';
		value++;
	}

	rc = setenvf(&step->env, name, "%s", value);

	xfree(name);

	return (rc ? DATA_FOR_EACH_FAIL : DATA_FOR_EACH_CONT);
}

static int _merge_step_config_env(stepd_step_rec_t *step)
{
	step_container_t *c = step->container;
	data_t *env = data_resolve_dict_path(c->config, "/process/env/");

	xassert(c->magic == STEP_CONTAINER_MAGIC);

	if (!env)
		return SLURM_SUCCESS;

	xassert(!oci_conf->ignore_config_json);

	if (data_list_for_each_const(env, _foreach_config_env, step) < 0)
		return ESLURM_DATA_CONV_FAILED;

	return SLURM_SUCCESS;
}

static int _setup_container(slurmd_conf_t *conf, stepd_step_rec_t *step)
{
	step_container_t *c = step->container;
	int rc;

	xassert(c->magic == STEP_CONTAINER_MAGIC);

	if ((rc = get_oci_conf(&oci_conf)) && (rc != ENOENT)) {
		error("%s: error loading oci.conf: %s",
		      __func__, slurm_strerror(rc));
		return rc;
	}

	if (!oci_conf) {
		debug("%s: OCI Container not configured. Ignoring %pS requested container: %s",
		      __func__, step, c->bundle);
		return ESLURM_CONTAINER_NOT_CONFIGURED;
	}

	serializer_required(MIME_TYPE_JSON);

	if (!oci_conf->ignore_config_json) {
		if ((rc = _load_config(step)))
			goto error;

		if ((rc = _merge_step_config_env(step)))
			goto error;
	}

	if ((rc = _generate_container_paths(conf, step)))
		goto error;

error:
	if (rc)
		error("%s: container setup failed: %s",
		      __func__, slurm_strerror(rc));

	return rc;
}

static data_t *_get_container_state(void)
{
	int rc = SLURM_ERROR;
	data_t *state = NULL;
	char *out;
	run_command_args_t run_command_args = {
		.max_wait = -1,
		.script_argv = query_argv,
		.script_path = query_argv[0],
		.script_type = "RunTimeQuery",
		.status = &rc,
	};

	/* request container get deleted if known at all any more */
	_dump_command_args(&run_command_args, __func__);
	out = run_command(&run_command_args);
	debug("%s: RunTimeQuery rc:%u output:%s", __func__, rc, out);

	if (!out || !out[0] || rc) {
		error("%s: RunTimeQuery failed rc:%u output:%s", __func__, rc, out);
		return NULL;
	}

	if (serialize_g_string_to_data(&state, out, strlen(out),
				       MIME_TYPE_JSON)) {
		error("%s: unable to parse container state: %s",
		      __func__, out);
		log_flag_hex(STEPS, out, strlen(out),
			     "unable to parse container state response");
	}

	xfree(out);

	return state;
}

static char *_get_container_status(void)
{
	char *state = NULL;
	data_t *dstate = _get_container_state();

	if (!dstate)
		return NULL;

	if (data_retrieve_dict_path_string(dstate, "/status/", &state))
		debug("%s: unable to find /status", __func__);

	return state;
}

static void _kill_container(void)
{
	int stime = 2500;
	char *status = NULL;
	run_command_args_t run_command_args = {
		.max_wait = -1,
	};

	if (!oci_conf->ignore_config_json &&
	    !(status = _get_container_status())) {
		debug("container already dead");
	} else if (!xstrcasecmp(status, "running")) {
		run_command_args.script_argv = kill_argv;
		run_command_args.script_path = kill_argv[0];
		run_command_args.script_type = "RunTimeKill";

		for (int t = 0; t < 10; t++) {
			char *out;
			int kill_status = SLURM_ERROR;
			run_command_args.status = &kill_status;

			xfree(status);
			status = _get_container_status();

			if (!oci_conf->ignore_config_json &&
			    (!status || !xstrcasecmp(status, "stopped")))
				break;

			out = run_command(&run_command_args);
			debug("%s: RunTimeKill rc:%u output:%s",
			      __func__, kill_status, out);
			xfree(out);

			if (oci_conf->ignore_config_json)
				break;

			/*
			 * use exp backoff up to 1s to wait for the container to
			 * cleanup.
			 *
			 * OCI runtime doesn't provide any way but to poll to see
			 * if the container has been squashed
			 */
			debug("%s: sleeping %dusec to query state again",
			      __func__, stime);
			usleep(stime);

			if (stime > 1000000)
				stime = 1000000;
			else
				stime *= 2;
		}
	}

	if (status) {
		int delete_status = SLURM_ERROR;
		char *out;

		/* request container get deleted if known at all any more */
		run_command_args.script_argv = delete_argv;
		run_command_args.script_path = delete_argv[0];
		run_command_args.script_type = "RunTimeDelete";
		run_command_args.status = &delete_status;
		_dump_command_args(&run_command_args, __func__);
		out = run_command(&run_command_args);
		debug("%s: RunTimeDelete rc:%u output:%s",
		      __func__, delete_status, out);
		xfree(out);
		xfree(status);
	}
}

static void _run(stepd_step_task_info_t *task)
{
	debug3("%s: executing: %s", __func__, run_argv[2]);
	execv(run_argv[0], run_argv);
	fatal("execv(%s) failed: %m", run_argv[0]);
}

static void _create_start(stepd_step_task_info_t *task)
{
	int stime = 250, rc = SLURM_ERROR;
	char *out;
	run_command_args_t run_command_args = {
		.max_wait = -1,
		.status = &rc,
	};

	if (oci_conf->ignore_config_json)
		fatal("IgnoreFileConfigJson=true and RunTimeStart are mutually exclusive");

	run_command_args.script_argv = create_argv;
	run_command_args.script_path = create_argv[0];
	run_command_args.script_type = "RunTimeCreate";
	_dump_command_args(&run_command_args, __func__);
	out = run_command(&run_command_args);
	debug("%s: RunTimeCreate rc:%u output:%s", __func__, rc, out);
	xfree(out);

	/* have to wait here until state finds the container or fail out */
	for (int t = 0; t <= 10; t++) {
		char *status = _get_container_status();

		if (!status) {
			if (t == 10)
				fatal("container never started");

			/* state called before create done */
			if (stime > 1000000)
				stime = 1000000;
			else
				stime *= 2;

			usleep(stime);
			continue;
		}

		debug("container in %s state", status);

		if (!xstrcasecmp(status, "creating")) {
			/* wait for creation to finish */
			xfree(status);
			usleep(250);
		} else if (!xstrcasecmp(status, "created")) {
			xfree(status);
			break;
		} else {
			fatal("%s: unexpected container status: %s",
			      __func__, status);
		}
	}

	run_command_args.script_argv = start_argv;
	run_command_args.script_path = start_argv[0];
	run_command_args.script_type = "RunTimeStart";
	_dump_command_args(&run_command_args, __func__);
	out = run_command(&run_command_args);
	debug("%s: RunTimeStart rc:%u output:%s", __func__, rc, out);
	xfree(out);

	/*
	 * the initial PID is now dead but the container could still be running
	 * but it likely is running outside of slurmstepd's process group
	 */

	stime = 2500;
	while (true) {
		char *status = _get_container_status();

		if (!status || xstrcasecmp(status, "running")) {
			debug("container no longer running: %s", status);
			xfree(status);
			break;
		}

		xfree(status);

		/* increase wait times exp */
		if (stime > 1000000)
			stime = 1000000;
		else
			stime *= 2;

		usleep(stime);
	}

	/*
	 * since the parent process has exited, kill off the container to kill
	 * any orphan processes
	 */
	_kill_container();

	_exit(rc);
}

static void _generate_patterns(slurmd_conf_t *conf, stepd_step_rec_t *step,
			       stepd_step_task_info_t *task)
{
	char *gen;
	int id = -1;
	char **argv = NULL;

	debug2("%s: %ps TaskId=%d",
	       __func__, &step->step_id, (task ? task->id : -1));

	if (task) {
		id = task->id;
		argv = task->argv;
	}

	gen = _generate_pattern(conf, step, oci_conf->runtime_create, id, argv);
	if (gen) {
		static bool set = false;
		if (set)
			xfree(create_argv[2]);
		create_argv[2] = gen;
		set = true;
	}

	gen = _generate_pattern(conf, step, oci_conf->runtime_delete, id, argv);
	if (gen) {
		static bool set = false;
		if (set)
			xfree(delete_argv[2]);
		delete_argv[2] = gen;
		set = true;
	}

	gen = _generate_pattern(conf, step, oci_conf->runtime_kill, id, argv);
	if (gen) {
		static bool set = false;
		if (set)
			xfree(kill_argv[2]);
		kill_argv[2] = gen;
		set = true;
	}

	gen = _generate_pattern(conf, step, oci_conf->runtime_query, id, argv);
	if (gen) {
		static bool set = false;
		if (set)
			xfree(query_argv[2]);
		query_argv[2] = gen;
		set = true;
	}

	gen = _generate_pattern(conf, step, oci_conf->runtime_run, id, argv);
	if (gen) {
		static bool set = false;
		if (set)
			xfree(run_argv[2]);
		run_argv[2] = gen;
		set = true;
	}

	gen = _generate_pattern(conf, step, oci_conf->runtime_start, id, argv);
	if (gen) {
		static bool set = false;
		if (set)
			xfree(start_argv[2]);
		start_argv[2] = gen;
		set = true;
	}
}

static void _container_run(slurmd_conf_t *conf, stepd_step_rec_t *step,
			   stepd_step_task_info_t *task)
{
	step_container_t *c = step->container;
	int rc;

	xassert(c->magic == STEP_CONTAINER_MAGIC);

	if (!oci_conf) {
		debug("%s: OCI Container not configured. Ignoring %pS requested container: %s",
		      __func__, step, c->bundle);
		return;
	}

	if (!c->spool_dir || !c->task_spool_dir)
		fatal_abort("spool directory not populated");

	if (oci_conf->env_exclude_set) {
		char **env = env_array_exclude((const char **) step->env,
					       &oci_conf->env_exclude);
#ifdef MEMORY_LEAK_DEBUG
		FREE_NULL_ENV(step->env, NULL);
#endif
		step->env = env;
	}

	if (c->config) {
		int rc;
		char *out = NULL;
		char *jconfig = NULL;

		/* create new config.json in spooldir */
		xstrfmtcat(jconfig, "%s/config.json", c->task_spool_dir);

		if ((rc = _modify_config(step, task)))
			fatal("%s: configuring container failed: %s",
			      __func__, slurm_strerror(rc));

		if ((rc = serialize_g_data_to_string(&out, NULL, c->config,
						     MIME_TYPE_JSON,
						     SER_FLAGS_NONE))) {
			fatal("%s: serialization of config failed: %s",
			      __func__, slurm_strerror(rc));
		}

		FREE_NULL_DATA(c->config);

		if ((rc = _write_config(step, jconfig, out)))
			fatal("%s: unable to write %s: %s",
			      __func__, jconfig, slurm_strerror(rc));

		debug("%s: wrote %s", __func__, jconfig);

		/*
		 * Swap bundle path to spool directory to ensure runtime uses
		 * correct config.json
		 */
		xfree(c->bundle);
		c->bundle = xstrdup(c->task_spool_dir);

		xfree(out);
		xfree(jconfig);
	}

	if (oci_conf->create_env_file) {
		char *envfile = NULL;
		bool nl = (oci_conf->create_env_file ==
			   NEWLINE_TERMINATED_ENV_FILE);

		/* keep _generate_pattern() in sync with this path */
		xstrfmtcat(envfile, "%s/%s", c->task_spool_dir,
			   SLURM_CONTAINER_ENV_FILE);

		if ((rc = env_array_to_file(envfile, (const char **) step->env,
					    nl)))
			fatal("%s: unable to write %s: %s",
			      __func__, envfile, slurm_strerror(rc));

		if (chown(envfile, step->uid, step->gid) < 0)
			fatal("%s: chown(%s): %m", __func__, envfile);

		if (!rc && chmod(envfile, 0750) < 0)
			error("%s: chmod(%s, 750): %m", __func__, envfile);

		debug("%s: wrote %s", __func__, envfile);

		xfree(envfile);
	}

	if (oci_conf->runtime_env_exclude_set) {
		extern char **environ;
		char **env = env_array_exclude((const char **) environ,
					       &oci_conf->runtime_env_exclude);

#ifdef MEMORY_LEAK_DEBUG
		env_unset_environment();
#endif
		environ = env;
	}

	xfree(c->work_dir);
	c->work_dir = xstrdup(step->cwd);

	debug4("%s: setting cwd from %s to task spooldir: %s",
	       __func__, step->cwd, c->task_spool_dir);
	xfree(step->cwd);
	step->cwd = xstrdup(c->task_spool_dir);

	_generate_patterns(conf, step, task);

	if (oci_conf->runtime_run)
		_run(task);
	else
		_create_start(task);
}

/*
 * Return the portion of path below base, or NULL if path is not below base.
 * Caller must xfree() the result.
 */
static char *_relative_subpath(const char *base, const char *path)
{
	size_t len = strlen(base);

	if (xstrncmp(path, base, len))
		return NULL;

	path += len;
	while (path[0] == '/')
		path++;

	if (!path[0])
		return NULL;

	return xstrdup(path);
}

/*
 * Open the directory holding stop, descending rel from basefd one component
 * at a time with O_NOFOLLOW so no component can be swapped for a symlink.
 * IN basefd - directory to descend from
 * IN rel - path relative to basefd
 * IN stop - pointer into rel to the component to stop before
 * RET fd of the parent directory, or -1 with errno set
 */
static int _open_subdir(int basefd, char *rel, const char *stop)
{
	int fd;

	if ((fd = dup(basefd)) < 0)
		return -1;

	for (char *tok = rel; tok < stop;) {
		int next;
		char *sep = xstrchr(tok, '/');

		if (!sep)
			break;

		*sep = '\0';
		next = openat(fd, tok, (O_DIRECTORY | O_NOFOLLOW));
		*sep = '/';
		fd_close(&fd);

		if (next < 0)
			return -1;

		fd = next;
		tok = sep + 1;
	}

	return fd;
}

static void _cleanup_container(slurmd_conf_t *conf, stepd_step_rec_t *step)
{
	step_container_t *c = step->container;
	char *parent = NULL, *stepdir = NULL, *stepname = NULL;
	int basefd = -1, stepfd = -1;

	xassert(c->magic == STEP_CONTAINER_MAGIC);

	if (!oci_conf) {
		debug("%s: OCI Container not configured. Ignoring %pS requested container: %s",
		      __func__, step, c->bundle);
		return;
	}

	/* cleanup may be called without ever setting up container */

	_generate_patterns(conf, step, NULL);
	_kill_container();

	if (oci_conf->disable_cleanup)
		goto done;

	if (!c->spool_dir)
		c->spool_dir = _generate_spooldir(conf, step, NULL);

	/*
	 * The job user owns the step spool directory and can replace any
	 * component below it with a symlink, so never resolve these paths as
	 * root. Descend with O_NOFOLLOW and remove relative to the directory
	 * handles instead.
	 */
	/*
	 * ContainerPath may hold a taskid pattern, in which case the step
	 * spool dir is the pattern trimmed at that component and so ends in a
	 * '/'. Drop any trailing separators before splitting it.
	 */
	stepdir = xstrdup(c->spool_dir);
	for (int i = strlen(stepdir) - 1; (i > 0) && (stepdir[i] == '/'); i--)
		stepdir[i] = '\0';

	parent = xdirname(stepdir);
	stepname = xbasename(stepdir);

	if ((basefd = open(parent, O_DIRECTORY)) < 0) {
		if (errno != ENOENT)
			error("open(%s): %m", parent);
		goto cleanup_fds;
	}

	if ((stepfd = openat(basefd, stepname,
			     (O_DIRECTORY | O_NOFOLLOW))) < 0) {
		if (errno != ENOENT)
			error("openat(%s/%s): %m", parent, stepname);
		goto cleanup_fds;
	}

	if (step->node_tasks > 0) {
		/* clear every config.json and task dir */
		for (int i = 0; i < step->node_tasks; i++) {
			int taskfd = -1, taskparentfd = -1;
			char *rel = NULL, *taskdir = NULL;

			xfree(c->task_spool_dir);
			c->task_spool_dir =
				_generate_spooldir(conf, step, step->task[i]);

			_generate_patterns(conf, step, step->task[i]);

			/*
			 * The task dir is not always a direct child of the
			 * step dir: a taskid pattern in ContainerPath leaves
			 * intervening components. Walk them one at a time.
			 */
			rel = _relative_subpath(stepdir, c->task_spool_dir);
			if (!rel) {
				error("%s: %s is not below %s",
				      __func__, c->task_spool_dir, stepdir);
				continue;
			}

			taskdir = xbasename(rel);
			taskparentfd = _open_subdir(stepfd, rel, taskdir);

			if (taskparentfd < 0) {
				if (errno != ENOENT)
					error("openat(%s): %m",
					      c->task_spool_dir);
				xfree(rel);
				continue;
			}

			if ((taskfd = openat(taskparentfd, taskdir,
					     (O_DIRECTORY | O_NOFOLLOW))) < 0) {
				if (errno != ENOENT)
					error("openat(%s): %m",
					      c->task_spool_dir);
				fd_close(&taskparentfd);
				xfree(rel);
				continue;
			}

			if (!oci_conf->ignore_config_json &&
			    unlinkat(taskfd, "config.json", 0) &&
			    (errno != ENOENT))
				error("unlinkat(%s/config.json): %m",
				      c->task_spool_dir);

			if (oci_conf->create_env_file &&
			    unlinkat(taskfd, SLURM_CONTAINER_ENV_FILE, 0) &&
			    (errno != ENOENT))
				error("unlinkat(%s/%s): %m",
				      c->task_spool_dir,
				      SLURM_CONTAINER_ENV_FILE);

			fd_close(&taskfd);

			if (unlinkat(taskparentfd, taskdir, AT_REMOVEDIR) &&
			    (errno != ENOENT))
				error("unlinkat(%s): %m", c->task_spool_dir);

			fd_close(&taskparentfd);
			xfree(rel);
		}

		xfree(c->task_spool_dir);
	}

	if (unlinkat(basefd, stepname, AT_REMOVEDIR) && (errno != ENOENT))
		error("unlinkat(%s): %m", c->spool_dir);

cleanup_fds:
	fd_close(&stepfd);
	fd_close(&basefd);
	xfree(parent);
	xfree(stepdir);

done:
	FREE_NULL_OCI_CONF(oci_conf);
}

extern int runtime_p_init(const int idx, runtime_context_t context)
{
	runtime_register_arg(idx, &runtime_arg_container);
	return SLURM_SUCCESS;
}

extern int runtime_p_cli_apply_launch_params(const uint32_t het_job_inx,
					     runtime_context_t context,
					     slurm_step_launch_params_t
						     *launch_params)
{
	xassert(launch_params);
	xassert(het_job_inx != NO_VAL);

	if ((het_job_inx >= container_bundle_cnt) ||
	    !container_bundle[het_job_inx])
		return SLURM_SUCCESS;

	launch_params->container = container_bundle[het_job_inx];

	return SLURM_SUCCESS;
}

extern void runtime_p_cli_reset_args(const uint32_t het_job_inx)
{
	xassert(het_job_inx != NO_VAL);

	if (het_job_inx < container_bundle_cnt)
		xfree(container_bundle[het_job_inx]);
}

extern int runtime_p_cli_apply_step_req(const uint32_t het_job_inx,
					runtime_context_t context,
					step_create_req_msg_t *step_req)
{
	xassert(step_req);
	xassert(het_job_inx != NO_VAL);

	if ((het_job_inx >= container_bundle_cnt) ||
	    !container_bundle[het_job_inx])
		return SLURM_SUCCESS;

	xfree(step_req->container);
	step_req->container = xstrdup(container_bundle[het_job_inx]);

	return SLURM_SUCCESS;
}

extern int runtime_p_cli_apply_job_desc(const uint32_t het_job_inx,
					runtime_context_t context,
					job_desc_msg_t *job_desc)
{
	xassert(job_desc);
	xassert(het_job_inx != NO_VAL);

	if ((het_job_inx >= container_bundle_cnt) ||
	    !container_bundle[het_job_inx])
		return SLURM_SUCCESS;

	/* salloc has no step of its own to hand the bundle to. */
	if ((context == RUNTIME_CTXT_SALLOC) &&
	    !xstrstr(slurm_conf.launch_params, "use_interactive_step")) {
		error("--container requires LaunchParameters=use_interactive_step");
		return ESLURM_CONTAINER_NOT_CONFIGURED;
	}

	xfree(job_desc->container);
	job_desc->container = xstrdup(container_bundle[het_job_inx]);

	return SLURM_SUCCESS;
}

/*
 * RET the environment variable the container is exported under, or NULL when
 *	context (or its child, when child is true) reads no such variable
 */
static const char *_env_name(runtime_context_t context, bool child)
{
	/* SLURM_CONTAINER names the bundle to the program a client starts */
	if (child) {
		switch (context) {
		case RUNTIME_CTXT_SALLOC:
		case RUNTIME_CTXT_SBATCH:
		case RUNTIME_CTXT_SCRONTAB:
		case RUNTIME_CTXT_SCRUN:
		case RUNTIME_CTXT_SRUN:
			return runtime_arg_container.environment.slurm;
		default:
			return NULL;
		}
	}

	switch (context) {
	case RUNTIME_CTXT_SRUN:
		return runtime_arg_container.environment.srun;
	case RUNTIME_CTXT_SBATCH:
		return runtime_arg_container.environment.sbatch;
	case RUNTIME_CTXT_SALLOC:
		return runtime_arg_container.environment.salloc;
	case RUNTIME_CTXT_SCRUN:
		return runtime_arg_container.environment.scrun;
	default:
		return NULL;
	}
}

/*
 * RET name, or the per-component <name>_HET_GROUP_<het_job_inx> when
 *	het_job_inx is a component of a heterogeneous job rather than NO_VAL,
 *	to xfree()
 */
static char *_het_name(const char *name, const uint32_t het_job_inx)
{
	if (het_job_inx == NO_VAL)
		return xstrdup(name);

	return xstrdup_printf("%s_HET_GROUP_%u", name, het_job_inx);
}

/* Export name=value, or its per-component form, see _het_name() */
static void _setenv_het(char ***envp, const char *name,
			const uint32_t het_job_offset, const char *value)
{
	char *het_name = _het_name(name, het_job_offset);

	setenvf(envp, het_name, "%s", value);
	xfree(het_name);
}

extern int runtime_p_cli_apply_env(const uint32_t het_job_inx,
				   const uint32_t het_job_offset,
				   runtime_context_t context, bool child,
				   char ***envp)
{
	const char *name = _env_name(context, child);

	xassert(het_job_inx != NO_VAL);

	if (!name || (het_job_inx >= container_bundle_cnt) ||
	    !container_bundle[het_job_inx])
		return SLURM_SUCCESS;

	/* setenvf() writes to the process environment if the array is NULL */
	if (envp && !*envp)
		*envp = env_array_create();

	_setenv_het(envp, name, het_job_offset, container_bundle[het_job_inx]);

	return SLURM_SUCCESS;
}

/* RET str in single quotes for a POSIX shell, to xfree() */
static char *_shell_single_quote(const char *str)
{
	char *quoted = NULL, *pos = NULL;

	xstrcatat(quoted, &pos, "'");

	for (const char *ptr = str; *ptr; ptr++) {
		if (*ptr == '\'')
			xstrcatat(quoted, &pos, "'\\''");
		else
			xstrncatat(quoted, &pos, ptr, 1);
	}

	xstrcatat(quoted, &pos, "'");

	return quoted;
}

extern int runtime_p_cli_apply_interactive_step(const uint32_t het_job_inx,
						runtime_context_t context,
						char **command)
{
	/* A job that is not heterogeneous keeps its bundle as component 0 */
	const uint32_t inx = ((het_job_inx == NO_VAL) ? 0 : het_job_inx);
	char *name = NULL, *quoted = NULL, *line = NULL;

	xassert(context == RUNTIME_CTXT_SALLOC);

	if ((inx >= container_bundle_cnt) || !container_bundle[inx])
		return SLURM_SUCCESS;

	/*
	 * Give the bundle to the interactive step's srun alone, as the
	 * variable it reads, instead of exporting it from salloc's own
	 * environment, which every command salloc runs would inherit.
	 */
	name = _het_name(runtime_arg_container.environment.srun, het_job_inx);
	quoted = _shell_single_quote(container_bundle[inx]);
	line = xstrdup_printf("%s=%s %s", name, quoted, *command);

	xfree(*command);
	*command = line;
	xfree(quoted);
	xfree(name);

	return SLURM_SUCCESS;
}

extern int runtime_p_fini(void)
{
	for (uint32_t i = 0; i < container_bundle_cnt; i++)
		xfree(container_bundle[i]);

	xfree(container_bundle);
	container_bundle_cnt = 0;

	return SLURM_SUCCESS;
}

extern int runtime_p_setup(slurmd_conf_t *conf, stepd_step_rec_t *step,
			   slurm_addr_t *cli, slurm_msg_t *msg)
{
	struct priv_state sprivs = { 0 };
	int rc = EINVAL;

	if (!step->container)
		return SLURM_SUCCESS;

	if (drop_privileges(step, false, &sprivs, true) < 0) {
		error("%s: drop_privileges failed", __func__);
		return SLURM_ERROR;
	}
	rc = _setup_container(conf, step);
	if (reclaim_privileges(&sprivs) < 0) {
		error("%s: reclaim_privileges failed", __func__);
		return SLURM_ERROR;
	}

	if (rc == ESLURM_CONTAINER_NOT_CONFIGURED) {
		debug2("%s: container %s requested but containers are not configured on this node",
		       __func__, step->container->bundle);
	} else if (rc) {
		error("%s: container setup failed: %s",
		      __func__, slurm_strerror(rc));
		return SLURM_ERROR;
	} else {
		debug2("%s: container %s successfully setup",
		       __func__, step->container->bundle);
	}

	return SLURM_SUCCESS;
}

extern void runtime_p_cleanup(slurmd_conf_t *conf, stepd_step_rec_t *step)
{
	if (!step->container)
		return;

	_cleanup_container(conf, step);
}

extern void runtime_p_task_init(slurmd_conf_t *conf, stepd_step_rec_t *step,
				stepd_step_task_info_t *task)
{
	if (!step->container)
		return;

	_container_task_init(conf, step, task);
}

extern int runtime_p_run(slurmd_conf_t *conf, stepd_step_rec_t *step,
			 stepd_step_task_info_t *task)
{
	if (step->container)
		_container_run(conf, step, task);

	/*
	 * _container_run() only returns when no container was run (none
	 * requested or OCI not configured), so exec the task directly.
	 */
	execve(task->argv[0], task->argv, step->env);
	return errno;
}

static bool _on_container_arg(const runtime_arg_t *arg,
			      const uint32_t het_job_inx, data_t *value,
			      bool set_by_cli, bool set_by_env,
			      bool set_by_data, bool early_pass)
{
	xassert(het_job_inx != NO_VAL);

	if (het_job_inx >= container_bundle_cnt) {
		/* Try to minimize the calls to xrecalloc() */
		const int het_count = (het_job_inx + 1);
		const int count = (CONTAINER_BUNDLE_CNT *
				   ((het_count / CONTAINER_BUNDLE_CNT) + 1));

		xrecalloc(container_bundle, count, sizeof(*container_bundle));
		container_bundle_cnt = count;
	}

	xfree(container_bundle[het_job_inx]);

	if (data_get_string_converted(value, &container_bundle[het_job_inx])) {
		error("--%s requires a string value: use --%s=%s",
		      arg->argument, arg->argument, arg->help_value);
		return false;
	}

	if (!container_bundle[het_job_inx][0]) {
		error("--%s requires a non-empty string", arg->argument);
		xfree(container_bundle[het_job_inx]);
		return false;
	}

	return true;
}