Merge pull request #160 from jameshcorbett/disable-fluxion-scheduling

DWS: add option to disable fluxion scheduling

Author: mergify[bot]

src/modules/coral2_dws.py

@@ -306,7 +306,7 @@ def state_complete(workflow, state):
     )
 
 
-def workflow_state_change_cb(event, handle, k8s_api):
+def workflow_state_change_cb(event, handle, k8s_api, disable_fluxion):
     """Exception-catching wrapper around _workflow_state_change_cb_inner."""
     try:
         workflow = event["object"]
@@ -324,7 +324,9 @@ def workflow_state_change_cb(event, handle, k8s_api):
         del _WORKFLOWINFO_CACHE[jobid]
         return
     try:
-        _workflow_state_change_cb_inner(workflow, jobid, winfo, handle, k8s_api)
+        _workflow_state_change_cb_inner(
+            workflow, jobid, winfo, handle, k8s_api, disable_fluxion
+        )
     except Exception:
         LOGGER.exception(
             "Failed to process event update for workflow with jobid %s:", jobid
@@ -342,7 +344,9 @@ def workflow_state_change_cb(event, handle, k8s_api):
         handle.job_raise(jobid, "exception", 0, "DWS/Rabbit interactions failed")
 
 
-def _workflow_state_change_cb_inner(workflow, jobid, winfo, handle, k8s_api):
+def _workflow_state_change_cb_inner(
+    workflow, jobid, winfo, handle, k8s_api, disable_fluxion
+):
     if "state" not in workflow["status"]:
         # workflow was just submitted, DWS still needs to give workflow
         # a state of 'Proposal'
@@ -379,13 +383,15 @@ def _workflow_state_change_cb_inner(workflow, jobid, winfo, handle, k8s_api):
             resources = flux.job.kvslookup.job_kvs_lookup(handle, jobid)["jobspec"][
                 "resources"
             ]
+        if not disable_fluxion:
+            resources = directivebreakdown.apply_breakdowns(
+                k8s_api, workflow, resources, _MIN_ALLOCATION_SIZE
+            )
         handle.rpc(
             "job-manager.dws.resource-update",
             payload={
                 "id": jobid,
-                "resources": directivebreakdown.apply_breakdowns(
-                    k8s_api, workflow, resources, _MIN_ALLOCATION_SIZE
-                ),
+                "resources": resources,
             },
         ).then(log_rpc_response)
     elif state_complete(workflow, "Setup"):
@@ -537,7 +543,10 @@ def init_rabbits(k8s_api, handle, watchers, graph_path, disable_draining):
         else:
             mark_rabbit(handle, rabbit["status"]["status"], *rabbit_rpaths[name], name)
         drain_offline_nodes(
-            handle, name, rabbit["status"]["access"].get("computes", []), disable_draining
+            handle,
+            name,
+            rabbit["status"]["access"].get("computes", []),
+            disable_draining,
         )
     watchers.add_watch(
         Watch(
@@ -625,6 +634,11 @@ def setup_parsing():
         action="store_true",
         help="Disable the draining of compute nodes based on k8s status",
     )
+    parser.add_argument(
+        "--disable-fluxion",
+        action="store_true",
+        help="Disable Fluxion scheduling of rabbits",
+    )
     return parser
 
 
@@ -729,16 +743,25 @@ def main():
     # start watching k8s workflow resources and operate on them when updates occur
     # or new RPCs are received
     with Watchers(handle, watch_interval=args.watch_interval) as watchers:
-        init_rabbits(
-            k8s_api,
-            handle,
-            watchers,
-            args.resourcegraph,
-            args.disable_compute_node_draining,
-        )
+        if not args.disable_fluxion:
+            init_rabbits(
+                k8s_api,
+                handle,
+                watchers,
+                args.resourcegraph,
+                args.disable_compute_node_draining,
+            )
         services = register_services(handle, k8s_api)
         watchers.add_watch(
-            Watch(k8s_api, WORKFLOW_CRD, 0, workflow_state_change_cb, handle, k8s_api)
+            Watch(
+                k8s_api,
+                WORKFLOW_CRD,
+                0,
+                workflow_state_change_cb,
+                handle,
+                k8s_api,
+                args.disable_fluxion,
+            )
         )
         raise_self_exception(handle)
 

t/t1002-dws-workflow-obj.t

@@ -43,7 +43,50 @@ test_expect_success 'exec dws service-providing script with bad arguments' '
         -e1 -v -rR.local --foobar
 '
 
+test_expect_success 'exec dws service-providing script with fluxion scheduling disabled' '
+    R=$(flux R encode -r 0) &&
+    DWS_JOBID=$(flux submit \
+            --setattr=system.alloc-bypass.R="$R" \
+            -o per-resource.type=node --output=dws-fluxion-disabled.out \
+            --error=dws-fluxion-disabled.err python ${DWS_MODULE_PATH} -e1 \
+            -v --disable-fluxion) &&
+    flux job wait-event -vt 15 -p guest.exec.eventlog ${DWS_JOBID} shell.start &&
+    flux job wait-event -vt 15 -m "note=dws watchers setup" ${DWS_JOBID} exception &&
+    ${RPC} "dws.create"
+'
+
+test_expect_success 'job submission without DW string works with fluxion-rabbit scheduling disabled' '
+    jobid=$(flux submit -n1 /bin/true) &&
+    flux job wait-event -vt 25 -m status=0 ${jobid} finish &&
+    test_must_fail flux job wait-event -vt 5 -m description=${CREATE_DEP_NAME} \
+        ${jobid} dependency-add
+'
+
+test_expect_success 'job submission with valid DW string works with fluxion-rabbit scheduling disabled' '
+    jobid=$(flux submit --setattr=system.dw="#DW jobdw capacity=10GiB type=xfs name=project1" \
+        -N1 -n1 hostname) &&
+    flux job wait-event -vt 10 -m description=${CREATE_DEP_NAME} \
+        ${jobid} dependency-add &&
+    flux job wait-event -t 10 -m description=${CREATE_DEP_NAME} \
+        ${jobid} dependency-remove &&
+    flux job wait-event -t 10 -m rabbit_workflow=fluxjob-$(flux job id ${jobid}) \
+        ${jobid} memo &&
+    flux job wait-event -vt 15 ${jobid} depend &&
+    flux job wait-event -vt 15 ${jobid} priority &&
+    flux job wait-event -vt 15 -m description=${PROLOG_NAME} \
+        ${jobid} prolog-start &&
+    flux job wait-event -vt 25 -m description=${PROLOG_NAME} \
+        ${jobid} prolog-finish &&
+    flux job wait-event -vt 15 -m status=0 ${jobid} finish &&
+    flux job wait-event -vt 15 -m description=${EPILOG_NAME} \
+        ${jobid} epilog-start &&
+    flux job wait-event -vt 30 -m description=${EPILOG_NAME} \
+        ${jobid} epilog-finish &&
+    flux job wait-event -vt 15 ${jobid} clean
+'
+
 test_expect_success 'load fluxion with rabbits' '
+    flux cancel ${DWS_JOBID} &&
 	flux R encode -l | flux python ${FLUX_SOURCE_DIR}/src/cmd/flux-dws2jgf.py \
 	--no-validate | jq . > R.local &&
 	flux kvs put resource.R="$(cat R.local)" &&