Merge pull request #1105 from grondo/expiration-update

resource: support resource expiration updates

Author: mergify[bot]

resource/modules/resource_match.cpp

@@ -1236,13 +1236,15 @@ static void update_resource (flux_future_t *f, void *arg)
     int rc = -1;
     const char *up = NULL;
     const char *down = NULL;
+    double expiration = -1.;
     json_t *resources = NULL;
     std::shared_ptr<resource_ctx_t> ctx = getctx ((flux_t *)arg);
 
-    if ( (rc = flux_rpc_get_unpack (f, "{s?:o s?:s s?:s}",
+    if ( (rc = flux_rpc_get_unpack (f, "{s?:o s?:s s?:s s?:F}",
                                            "resources", &resources,
                                            "up", &up,
-                                           "down", &down)) < 0) {
+                                           "down", &down,
+                                           "expiration", &expiration)) < 0) {
         flux_log_error (ctx->h, "%s: exiting due to resource.acquire failure",
                         __FUNCTION__);
         flux_reactor_stop (flux_get_reactor (ctx->h));
@@ -1252,6 +1254,12 @@ static void update_resource (flux_future_t *f, void *arg)
         flux_log_error (ctx->h, "%s: update_resource_db", __FUNCTION__);
         goto done;
     }
+    if (expiration >= 0.) {
+        /*  Update graph duration:
+         */
+        ctx->db->metadata.graph_duration.graph_end =
+            std::chrono::system_clock::from_time_t ((time_t) expiration);
+    }
     for (auto &kv : ctx->notify_msgs) {
         if ( (rc += flux_respond (ctx->h, kv.second->get_msg (), NULL)) < 0) {
             flux_log_error (ctx->h, "%s: flux_respond", __FUNCTION__);

t/t4011-match-duration.t

@@ -7,8 +7,9 @@ test_description='Test that parent duration is inherited according to RFC14'
 #
 # test_under_flux is under sharness.d/
 #
-test_under_flux 1
+test_under_flux 2
 
+flux setattr log-stderr-level 1
 export FLUX_URI_RESOLVE_LOCAL=t
 
 # Ensure fluxion modules are loaded under flux-alloc(1)
@@ -34,6 +35,35 @@ test_expect_success 'load fluxion modules in parent instance' '
 	load_qmanager &&
 	test_debug "flux dmesg -H | grep version"
 '
+# Usage: job_manager_get_R ID
+# This needs to be a shell script since it will be run under flux-proxy(1):
+cat <<'EOF' >job_manager_get_R
+#!/bin/sh
+flux python -c \
+"import flux; \
+ payload = {\"id\":$(flux job id $1),\"attrs\":[\"R\"]}; \
+ print(flux.Flux().rpc(\"job-manager.getattr\", payload).get_str()) \
+"
+EOF
+chmod +x job_manager_get_R
+
+export PATH=$(pwd):$PATH
+
+subinstance_get_R() {
+	flux proxy $1 flux kvs get resource.R
+}
+subinstance_get_expiration() {
+	subinstance_get_R $1 | jq .execution.expiration
+}
+# Usage: subinstance_get_duration ID JOBID
+subinstance_get_job_duration() {
+	flux proxy $1 job_manager_get_R $2 |
+		jq '.R.execution | .expiration - .starttime'
+}
+subinstance_get_job_expiration() {
+	flux proxy $1 job_manager_get_R $2 | jq '.R.execution.expiration'
+}
+
 test_expect_success HAVE_JQ 'parent expiration is inherited when duration=0' '
 	cat >get_R.sh <<-EOT &&
 	#!/bin/sh
@@ -59,8 +89,38 @@ test_expect_success HAVE_JQ 'parent expiration is inherited when duration=0' '
 	echo $exp2 | jq ". == $expiration" &&
 	flux shutdown --quiet $jobid
 '
+#  Check if running job updates are supported:
+id=$(flux submit sleep inf)
+flux update $id duration=1m && test_set_prereq FLUX_UPDATE_RUNNING
+flux cancel $id
+
+test_expect_success FLUX_UPDATE_RUNNING \
+	'expiration update is detected by subinstance scheduler' '
+	id=$(flux alloc --bg -t5m -n2) &&
+	exp1=$(subinstance_get_expiration $id) &&
+	test_debug "echo instance expiration is $exp1" &&
+	id1=$(flux proxy $id flux submit sleep 300) &&
+	duration1=$(subinstance_get_job_duration $id $id1) &&
+	test_debug "echo initial duration of subinstance job1 is $duration1" &&
+	echo $duration1 | jq -e ". < 300" &&
+	test_debug "echo updating duration of alloc job +5m" &&
+	flux update $id duration=+5m &&
+	test_debug "echo waiting for resource-update event" &&
+	flux proxy $id flux kvs eventlog wait-event -vt 30 \
+		resource.eventlog resource-update &&
+	exp2=$(subinstance_get_expiration $id) &&
+	test_debug "echo expiration updated from $exp1 to $exp2" &&
+	echo $exp2 | jq -e ". == $exp1 + 300" &&
+	id2=$(flux proxy $id flux submit sleep 300) &&
+	duration2=$(subinstance_get_job_duration $id $id2) &&
+	test_debug "echo duration of subinstance job2 is $duration2" &&
+	echo $duration2 | jq -e ". > 300" &&
+	flux proxy $id flux cancel --all &&
+	flux shutdown --quiet $id
+'
 test_expect_success 'unload fluxion modules' '
 	remove_qmanager &&
-	remove_resource
+	remove_resource &&
+	flux module load sched-simple
 '
 test_done