return effective queue config in job-manager.queue-list and use it in flux resource

src/bindings/python/flux/queue.py

@@ -176,6 +176,51 @@ def __getattr__(self, attr):
         return getattr(self._resource_list, attr)
 
 
+def queue_conf_from_config(config):
+    """Build the job-manager.queue-list "conf" object from a raw broker config.
+
+    This helper creates a job-manager.queue-list response ``conf`` object
+    directly from broker config. For use in testing and as a fallback when
+    the job-manager is from an older version of Flux that does not provide
+    the queue config directly.
+
+    Returns ``{"queues": [{"name": str, "requires"?: list, "parent"?: str},
+    ...]}`` in config declaration order. A virtual queue (RFC 33) has no
+    "requires" of its own, so its effective "requires" is resolved from its
+    parent.
+
+    Args:
+        config (dict): a broker config, e.g. from the ``config.get`` RPC.
+
+    Raises:
+        ValueError: a virtual queue names a parent that is not configured.
+    """
+    queues = config.get("queues", {})
+    entries = []
+    # dict iteration preserves insertion order (Python 3.7+), and both
+    # tomllib and json.load preserve config file order, so this matches the
+    # job-manager's config declaration order.
+    for name, entry in queues.items():
+        conf = {"name": name}
+        parent = entry.get("parent")
+        if parent is not None:
+            # Virtual queue: inherit the parent's requires. An unresolvable
+            # parent is fatal (fail closed) - falling through would report
+            # the vqueue as covering the full instance resource set.
+            if parent not in queues:
+                raise ValueError(
+                    f"queue '{name}': parent queue '{parent}' is not configured"
+                )
+            conf["parent"] = parent
+            requires = queues[parent].get("requires")
+        else:
+            requires = entry.get("requires")
+        if requires is not None:
+            conf["requires"] = requires
+        entries.append(conf)
+    return {"queues": entries}
+
+
 class QueueInfo:
     """
     Information for a single queue.

src/cmd/flux-resource.py

@@ -24,6 +24,7 @@
 from flux.eventlog import EventLogFormatter
 from flux.hostlist import Hostlist
 from flux.idset import IDset
+from flux.queue import queue_conf_from_config
 from flux.resource import (
     ResourceJournalConsumer,
     ResourceSet,
@@ -196,61 +197,67 @@ def undrain(args):
     RPC(flux.Flux(), "resource.undrain", payload, nodeid=0).get()
 
 
-def queue_effective_entry(queues, name):
+def queue_conf(args, handle, queue_list_rpc):
     """
-    Return the effective [queues] config entry for queue ``name``: its own
-    entry, or its parent's entry if it is an RFC 33 virtual queue.
-
-    A virtual queue has no "requires" of its own, so resolving to the
-    parent's entry is what maps it to the parent's resource slice rather
-    than wrongly matching every node as an unconstrained queue would. An
-    unresolvable parent raises ValueError (defense in depth: config
-    validation rejects it in a live instance, but --config-file and
-    --from-stdin input is not validated).
+    Return the effective queue configuration as a name->entry dict, where
+    each entry has its effective "requires" (an RFC 33 virtual queue's is
+    resolved from its parent) and, for a virtual queue, its "parent". This
+    is the job-manager's authoritative view.
+
+    The primary source is the job-manager.queue-list RPC "conf" object. An
+    older job-manager omits "conf", so fall back to deriving it from the
+    [queues] config via queue_conf_from_config(). The hidden --config-file/
+    --from-stdin test options are also derived from the raw config.
+
+    'queue_list_rpc' is a job-manager.queue-list future the caller has
+    already sent (so it can overlap with other RPCs). It is required on the
+    online path and must be non-None whenever ``handle`` is set and
+    ``--config-file`` is not. It is ignored (and may be None) otherwise.
     """
-    entry = queues[name]
-    if "parent" in entry:
-        parent = entry["parent"]
-        if parent not in queues:
-            raise ValueError(
-                f"queue '{name}': parent queue '{parent}' is not configured"
-            )
-        entry = queues[parent]
-    return entry
+    if args.config_file:
+        with open(args.config_file) as fp:
+            conf = queue_conf_from_config(json.load(fp))
+    elif handle is None:
+        return {}
+    else:
+        resp = queue_list_rpc.get()
+        # Get conf from response or derive from config (fallback)
+        conf = resp.get("conf") or queue_conf_from_config(
+            handle.rpc("config.get").get()
+        )
+    return {entry["name"]: entry for entry in conf["queues"]}
 
 
 class QueueResources:
     """
     Convenience class to map queues to resource sets
     """
 
-    def __init__(self, resource_set, config):
+    def __init__(self, resource_set, queues):
         self._queues = {}
-        if "queues" not in config:
-            return
-        for queue in config["queues"]:
-            entry = queue_effective_entry(config["queues"], queue)
+        for name, entry in queues.items():
             if "requires" in entry:
                 result = resource_set.copy_constraint({"properties": entry["requires"]})
             else:
                 result = resource_set.copy()
-            self._queues[queue] = result
+            self._queues[name] = result
 
     def queue(self, queue):
         if queue not in self._queues:
             raise ValueError(f"{queue}: no such queue")
         return self._queues[queue]
 
 
-def ranks_by_queue(resource_set, config, queues):
+def ranks_by_queue(resource_set, queue_config, queues):
     """
     Return all ranks associated with a list of queues
     Args:
         resource_set: The resource set to query
-        config: a Flux config object including queue configuration, if any.
+        queue_config: effective queue config as a name->entry dict (see
+            queue_conf())
         queues: one or more queues specified as a comma-separated string
     """
-    queue_resources = QueueResources(resource_set, config)
+    queue_resources = QueueResources(resource_set, queue_config)
     ranks = IDset()
     for queue in queues:
         ranks.add(queue_resources.queue(queue).ranks)
@@ -452,25 +459,29 @@ def status(args):
     fmt = FluxResourceConfig("status").load().get_format_string(args.format)
 
     handle = None
+    queue_list_rpc = None
 
     #  Get payload from stdin or from resource.status RPC:
     if args.from_stdin:
         input_str = sys.stdin.read()
         rstatus = ResourceStatus(json.loads(input_str) if input_str else None)
     else:
         handle = flux.Flux()
+        #  Send the queue-list RPC (only needed for --queue) before draining
+        #  resource.status, so the two overlap:
+        if args.queue:
+            queue_list_rpc = handle.rpc("job-manager.queue-list")
         rstatus = resource_status(handle).get()
 
     if args.queue:
-        if args.config_file:
-            with open(args.config_file) as fp:
-                config = json.load(fp)
-        else:
-            config = {}
-            if handle is not None:
-                config = handle.rpc("config.get").get()
         try:
-            rstatus.filter(ranks_by_queue(rstatus.rset, config, args.queue))
+            rstatus.filter(
+                ranks_by_queue(
+                    rstatus.rset,
+                    queue_conf(args, handle, queue_list_rpc),
+                    args.queue,
+                )
+            )
         except ValueError as exc:
             raise ValueError(f"--queue: {exc}") from None
 
@@ -534,12 +545,12 @@ def __init__(  # lgtm[py/missing-call-to-init]
         self,
         arg=None,
         version=1,
-        flux_config=None,
+        queue_config=None,
         queue=None,
         queue_filter=None,
         hidden_queues=None,
     ):
-        self.flux_config = flux_config
+        self.queue_config = queue_config or {}
         self._queue = queue
         self._queue_filter = queue_filter
         self._hidden_queues = hidden_queues or set()
@@ -558,9 +569,8 @@ def propertiesx(self):
         properties = json.loads(self.get_properties())
         #  Strip all configured queue names from properties, so that
         #  properties used only for queue membership are not displayed.
-        if self.flux_config and "queues" in self.flux_config:
-            for q in self.flux_config["queues"]:
-                properties.pop(q, None)
+        for q in self.queue_config:
+            properties.pop(q, None)
         return ",".join(properties.keys())
 
     @property
@@ -573,11 +583,11 @@ def queue(self):
         #  If self._queue is not set, then build list of queues from
         #  set properties and queue configuration:
         queues = ""
-        if self.flux_config and "queues" in self.flux_config:
+        if self.queue_config:
             if not self.ranks:
                 return ""
             properties = json.loads(self.get_properties())
-            for key, value in self.flux_config["queues"].items():
+            for key, entry in self.queue_config.items():
                 if self._queue_filter:
                     if key not in self._queue_filter:
                         continue
@@ -587,9 +597,8 @@ def queue(self):
                 # was explicitly requested via -q (i.e. is in the filter),
                 # otherwise it would match every node in the default QUEUE
                 # column via its resolved parent's requires:
-                if "parent" in value and not self._queue_filter:
+                if "parent" in entry and not self._queue_filter:
                     continue
-                entry = queue_effective_entry(self.flux_config["queues"], key)
                 if "requires" not in entry or set(entry["requires"]).issubset(
                     set(properties)
                 ):
@@ -626,7 +635,7 @@ def constraint_combinations(rset):
 
 
 def resources_uniq_lines(
-    resources, states, formatter, config, queues=None, hidden_queues=None
+    resources, states, formatter, queue_config, queues=None, hidden_queues=None
 ):
     """
     Generate a set of resource sets that would produce unique lines given
@@ -670,10 +679,10 @@ def resources_uniq_lines(
     #  If no queues are configured then one "anonymous" queue is simulated
     #  with [None].
     if not queues:
-        if config and "queues" in config:
+        if queue_config:
             queues = [
                 q
-                for q, entry in config["queues"].items()
+                for q, entry in queue_config.items()
                 if q not in hidden_queues and "parent" not in entry
             ]
             if not queues:
@@ -691,7 +700,7 @@ def resources_uniq_lines(
             #   state would be suppressed.
             #
             for queue in queues:
-                rset = ResourceSetExtra(flux_config=config, queue=queue)
+                rset = ResourceSetExtra(queue_config=queue_config, queue=queue)
                 rset.state = state
                 key = fmt.format(rset)
                 if key not in lines:
@@ -706,7 +715,7 @@ def resources_uniq_lines(
             rset.state = state
             rset = ResourceSetExtra(
                 rset,
-                flux_config=config,
+                queue_config=queue_config,
                 queue_filter=queue_filter,
                 hidden_queues=hidden_queues,
             )
@@ -725,31 +734,36 @@ def get_resource_list(args):
     Common function for list_handler() and emit_R()
     """
     valid_states = ["up", "down", "allocated", "free", "all"]
-    config = None
+    handle = None
 
     args.states = args.states.split(",")
     for state in args.states:
         if state not in valid_states:
             LOGGER.error("Invalid resource state %s specified", state)
             sys.exit(1)
 
+    queue_list_rpc = None
     if args.from_stdin:
         resources = SchedResourceList(json.load(sys.stdin))
-        if args.config_file:
-            with open(args.config_file) as fp:
-                config = json.load(fp)
     else:
         handle = flux.Flux()
-        rpcs = [resource_list(handle), handle.rpc("config.get")]
-        resources = rpcs[0].get()
-        try:
-            config = rpcs[1].get()
-        except Exception as e:
-            LOGGER.warning("Could not get flux config: " + str(e))
+        #  Send both RPCs before draining either, so they overlap:
+        queue_list_rpc = handle.rpc("job-manager.queue-list")
+        resources = resource_list(handle).get()
+
+    #  Tolerate a failure to reach the job-manager/broker so listing still
+    #  works without queue annotation. A bad queue config (e.g. an
+    #  unresolvable virtual queue parent) raises ValueError, not OSError, and
+    #  so still propagates.
+    try:
+        queue_config = queue_conf(args, handle, queue_list_rpc)
+    except OSError as e:
+        LOGGER.warning("Could not get queue configuration: " + str(e))
+        queue_config = {}
 
     if args.queue:
         try:
-            resources.filter(ranks_by_queue(resources.all, config, args.queue))
+            resources.filter(ranks_by_queue(resources.all, queue_config, args.queue))
         except ValueError as exc:
             raise ValueError(f"--queue: {exc}") from None
 
@@ -759,7 +773,7 @@ def get_resource_list(args):
         except (ValueError, TypeError) as exc:
             raise ValueError(f"--include: {exc}") from None
 
-    return resources, config
+    return resources, queue_config
 
 
 def sort_output(args, items):
@@ -783,7 +797,7 @@ def list_handler(args):
         "nodelist": "NODELIST",
         "rlist": "LIST",
     }
-    resources, config = get_resource_list(args)
+    resources, queue_config = get_resource_list(args)
 
     list_config = FluxResourceConfig("list").load()
     fmt = list_config.get_format_string(args.format)
@@ -794,7 +808,7 @@ def list_handler(args):
         resources,
         args.states,
         formatter,
-        config,
+        queue_config,
         queues=args.queue,
         hidden_queues=hidden_queues,
     )
@@ -815,7 +829,7 @@ def info(args):
 
 def emit_R(args):
     """Emit R in JSON on stdout for requested set of resources"""
-    resources, config = get_resource_list(args)
+    resources, _ = get_resource_list(args)
 
     rset = ResourceSet()
     rset.starttime = resources["all"].starttime

src/modules/job-manager/queue.c

@@ -144,20 +144,21 @@ static void queue_list_cb (flux_t *h,
                            void *arg)
 {
     struct queue_ctx *qctx = arg;
-    json_t *a = NULL;
+    json_t *resp;
 
     if (flux_request_decode (msg, NULL, NULL) < 0)
         goto error;
-    if (!(a = queues_list_encode (qctx->queues)))
+    /* resp is a borrowed reference owned by the queues cache; "O"
+     * increfs it for the response, so it is not decref'd here.
+     */
+    if (!(resp = queues_list_response (qctx->queues)))
         goto error;
-    if (flux_respond_pack (h, msg, "{s:O}", "queues", a) < 0)
+    if (flux_respond_pack (h, msg, "O", resp) < 0)
         flux_log_error (h, "error responding to job-manager.queue-list");
-    json_decref (a);
     return;
 error:
     if (flux_respond_error (h, msg, errno, NULL) < 0)
         flux_log_error (h, "error responding to job-manager.queue-list");
-    json_decref (a);
 }
 
 static void queue_status_cb (flux_t *h,