Merge pull request #5848 from grondo/post-event

add utility for posting job events manually if required

pyproject.toml

@@ -10,7 +10,7 @@ skip = [
   "src/bindings/python/flux/resource/__init__.py"]
 
 [tool.mypy]
-python_version = 3.6
+python_version = "3.6"
 files = ["src/cmd/**/*.py", "src/bindings/python/flux", "t/python/*.py"]
 mypy_path = ["src/bindings/python", "t/python/tap", "t/python"]
 allow_redefinition = true

src/cmd/Makefile.am

@@ -120,7 +120,8 @@ dist_fluxcmd_SCRIPTS = \
 	flux-update.py \
 	flux-imp-exec-helper \
 	py-runner.py \
-	flux-hostlist.py
+	flux-hostlist.py \
+	flux-post-job-event.py
 
 fluxcmd_PROGRAMS = \
 	flux-terminus \

src/cmd/flux-post-job-event.py

@@ -0,0 +1,55 @@
+#!/bin/false
+##############################################################
+# Copyright 2024 Lawrence Livermore National Security, LLC
+# (c.f. AUTHORS, NOTICE.LLNS, COPYING)
+#
+# This file is part of the Flux resource manager framework.
+# For details, see https://github.com/flux-framework.
+#
+# SPDX-License-Identifier: LGPL-3.0
+##############################################################
+
+import argparse
+import logging
+
+import flux
+from flux.job import JobID
+from flux.util import TreedictAction
+
+
+def post_event(args):
+    """Post an event to the job-manager.post-event.post service"""
+
+    req = {"id": JobID(args.id), "name": args.name}
+    if args.context:
+        req["context"] = args.context
+
+    try:
+        flux.Flux().rpc("job-manager.post-event.post", req).get()
+    except FileNotFoundError:
+        raise ValueError(f"No such job {args.id}")
+
+
+LOGGER = logging.getLogger("flux-job-post-event")
+
+
+@flux.util.CLIMain(LOGGER)
+def main():
+    parser = argparse.ArgumentParser(prog="flux-job-post-event")
+    parser.add_argument("id", help="jobid to which event shall be posted")
+    parser.add_argument("name", help="name of event to post")
+    parser.add_argument(
+        "context",
+        help="List of key=value pairs to set as event context",
+        action=TreedictAction,
+        nargs=argparse.REMAINDER,
+    )
+    parser.set_defaults(func=post_event)
+    args = parser.parse_args()
+    args.func(args)
+
+
+if __name__ == "__main__":
+    main()
+
+# vi: ts=4 sw=4 expandtab

src/modules/job-manager/Makefile.am

@@ -74,7 +74,8 @@ libjob_manager_la_SOURCES = \
 	plugins/begin-time.c \
 	plugins/update-duration.c \
 	plugins/validate-duration.c \
-	plugins/history.c
+	plugins/history.c \
+	plugins/post-event.c
 
 fluxinclude_HEADERS = \
 	jobtap.h

src/modules/job-manager/jobtap.c

@@ -51,6 +51,7 @@ extern int begin_time_plugin_init (flux_plugin_t *p);
 extern int validate_duration_plugin_init (flux_plugin_t *p);
 extern int update_duration_plugin_init (flux_plugin_t *p);
 extern int history_plugin_init (flux_plugin_t *p);
+extern int post_event_init (flux_plugin_t *p);
 
 struct jobtap_builtin {
     const char *name;
@@ -72,6 +73,7 @@ static struct jobtap_builtin jobtap_builtins [] = {
     { ".validate-duration", &validate_duration_plugin_init },
     { ".update-duration", &update_duration_plugin_init },
     { ".history", &history_plugin_init },
+    { ".post-event", &post_event_init },
     { 0 },
 };
 

src/modules/job-manager/plugins/post-event.c

@@ -0,0 +1,59 @@
+/************************************************************\
+ * Copyright 2024 Lawrence Livermore National Security, LLC
+ * (c.f. AUTHORS, NOTICE.LLNS, COPYING)
+ *
+ * This file is part of the Flux resource manager framework.
+ * For details, see https://github.com/flux-framework.
+ *
+ * SPDX-License-Identifier: LGPL-3.0
+\************************************************************/
+
+/* post-event.c - post manual events to job eventlog
+ */
+
+#if HAVE_CONFIG_H
+#include "config.h"
+#endif
+
+#include <jansson.h>
+#include <flux/core.h>
+#include <flux/jobtap.h>
+
+static void post_event_cb (flux_t *h,
+                           flux_msg_handler_t *mh,
+                           const flux_msg_t *msg,
+                           void *arg)
+{
+    flux_plugin_t *p = arg;
+    flux_jobid_t id;
+    const char *name;
+    json_t *context = NULL;
+
+    if (flux_msg_unpack (msg,
+                         "{s:I s:s s?o}",
+                         "id", &id,
+                         "name", &name,
+                         "context", &context) < 0)
+        goto error;
+    if (context) {
+        if (flux_jobtap_event_post_pack (p, id, name, "O", context) < 0)
+            goto error;
+    }
+    else if (flux_jobtap_event_post_pack (p, id, name, NULL) < 0)
+            goto error;
+    if (flux_respond (h, msg, NULL) < 0)
+        flux_log_error (h, "error responding to job-manager.post-event");
+error:
+    if (flux_respond_error (h, msg, errno, NULL) < 0)
+        flux_log_error (h, "error responding to job-manager.post-event");
+}
+
+
+int post_event_init (flux_plugin_t *p)
+{
+    if (flux_jobtap_service_register_ex (p, "post", 0, post_event_cb, p) < 0)
+        return -1;
+    return 0;
+}
+
+// vi:ts=4 sw=4 expandtab

t/Makefile.am

@@ -233,6 +233,7 @@ TESTSCRIPTS = \
 	t2812-flux-job-last.t \
 	t2813-flux-watch.t \
 	t2814-hostlist-cmd.t \
+	t2815-post-job-event.t \
 	t2900-job-timelimits.t \
 	t3000-mpi-basic.t \
 	t3001-mpi-personalities.t \

t/t2815-post-job-event.t

@@ -0,0 +1,41 @@
+#!/bin/sh
+
+test_description='Test flux post-job-event command'
+
+. $(dirname $0)/sharness.sh
+
+test_under_flux 1
+
+test_expect_success 'run a test job' '
+	JOBID=$(flux submit --wait-event=start sleep inf)
+'
+test_expect_success 'flux-post-job-event --help works' '
+	flux post-job-event --help >help.out &&
+	test_debug "cat help.out" &&
+	grep "name of event to post" help.out
+'
+test_expect_success 'flux-post-job-event can post a simple event' '
+	flux post-job-event $JOBID test &&
+	flux job wait-event -t15 $JOBID test
+'
+test_expect_success 'flux-post-job-event can post an event with context' '
+	flux post-job-event $JOBID test note=testing  &&
+	flux job wait-event -m note=testing -t15 $JOBID test 
+'
+test_expect_success 'flux-post-job-event can post multiple context keys' '
+	flux post-job-event $JOBID test note=test2 status=0  &&
+	flux job wait-event -m note=test2 -t15 $JOBID test  &&
+	flux job wait-event -m status=0 -t15 $JOBID test 
+'
+test_expect_success 'flux-post-job-event fails for invalid job' '
+	test_must_fail flux post-job-event f123 test note="test event"
+'
+test_expect_success 'flux-post-job-event fails for inactive job' '
+	flux cancel $JOBID &&
+	flux job wait-event $JOBID clean &&
+	test_must_fail flux post-job-event $JOBID test note="test event"
+'
+test_expect_success 'flux-post-job-event fails with invalid id' '
+	test_must_fail flux post-job-event baz test
+'
+test_done