Publish pulsar capability snapshot to relay on startup

Adds a static config + host-probe snapshot (staging dirs, dependency
resolvers, container runtimes, manager type) collected once during
PulsarApp init and POSTed once to a per-manager relay topic from
messaging.bind_app.

Galaxy can fetch the latest snapshot synchronously via the relay's
existing /api/v1/topics/{topic}/messages?limit=1&order=desc endpoint
and use it to auto-fill destination params on BYOC bootstrap and to
downgrade requests at job-build time when the remote pulsar doesn't
actually offer what the destination asks for.

The publish is fire-and-forget and gated by message_queue_publish_capabilities
(default True). Failures are logged and swallowed — capabilities are
advisory and must not block manager bind.

Author: mvdbeek

app.yml.sample

@@ -104,6 +104,15 @@
 ## configured in Galaxy's job_conf.yml if set.
 #relay_topic_prefix: production
 
+## On startup, publish a static capability snapshot (staging dirs, dependency
+## resolvers, container runtimes available, manager type) to the relay topic
+## "pulsar_capabilities[_<manager>]". Galaxy can read this to auto-fill
+## destination params and to refuse jobs that ask for capabilities this Pulsar
+## doesn't have. Advisory: a publish failure does not block startup, and
+## Galaxy keeps using operator-supplied params if no snapshot is available.
+## Set to false to disable the publish entirely.
+#message_queue_publish_capabilities: true
+
 ## Pulsar loops over waiting for queue messages for a short time before checking
 ## to see if it has been instructed to shut down. By default this is 0.2
 ## seconds. This value is used as the value of the 'timeout' parameter to

pulsar/capabilities.py

@@ -0,0 +1,195 @@
+"""Static capability snapshot collection for a Pulsar app.
+
+A ``PulsarCapabilities`` is a serializable summary of what a Pulsar
+server is configured to do — staging paths, dependency resolvers,
+container runtimes available on the host — so that Galaxy can read it
+at job-build time and adjust (or refuse) requests that the remote does
+not actually support.
+
+The data is collected once at app startup (in ``PulsarApp.__init__``),
+cached on the app, and published once to a relay topic from
+``messaging.bind_app``. It is never recomputed. Consumers fetch the
+latest snapshot via the relay's REST topic-messages endpoint.
+
+Schema is versioned (``SCHEMA_VERSION``); bumping it is a wire-breaking
+change for consumers, so add new optional fields rather than reshape
+existing ones.
+"""
+from __future__ import annotations
+
+import logging
+import os
+import shutil
+from dataclasses import (
+    asdict,
+    dataclass,
+    field,
+)
+from typing import (
+    TYPE_CHECKING,
+    Dict,
+    List,
+    Optional,
+    Union,
+)
+
+from galaxy.tool_util.deps.resolvers.conda import CondaDependencyResolver
+
+from pulsar import __version__ as pulsar_version
+
+if TYPE_CHECKING:
+    # Imported lazily to avoid the runtime cycle: pulsar.core imports this
+    # module to populate the cache.
+    from pulsar.core import PulsarApp
+    from pulsar.managers.stateful import StatefulManagerProxy
+
+log = logging.getLogger(__name__)
+
+SCHEMA_VERSION = 1
+
+
+@dataclass
+class DependencyResolverInfo:
+    type: str
+    disabled: bool = False
+    versionless: bool = False
+    auto_init: Optional[bool] = None
+    auto_install: Optional[bool] = None
+    prefix: Optional[str] = None
+
+
+@dataclass
+class ContainerRuntimeInfo:
+    docker_available: bool = False
+    singularity_available: bool = False
+    apptainer_available: bool = False
+
+
+@dataclass
+class ManagerCapabilities:
+    name: str
+    type: str
+    num_concurrent_jobs: Union[int, str, None] = None
+
+
+@dataclass
+class PulsarCapabilities:
+    schema_version: int
+    manager_name: str
+    pulsar_version: str
+    staging_directory: str
+    persistence_directory: Optional[str]
+    tool_dependency_dir: Optional[str]
+    dependency_resolvers: List[DependencyResolverInfo] = field(default_factory=list)
+    conda_available: bool = False
+    container_runtime: ContainerRuntimeInfo = field(default_factory=ContainerRuntimeInfo)
+    manager: Optional[ManagerCapabilities] = None
+
+    def to_dict(self) -> dict:
+        return asdict(self)
+
+
+def collect_capabilities(app: "PulsarApp", manager: "StatefulManagerProxy") -> PulsarCapabilities:
+    """Build the capabilities snapshot for a single manager on this app."""
+    resolvers = _collect_dependency_resolvers(app)
+    manager_caps = _collect_manager(manager)
+    return PulsarCapabilities(
+        schema_version=SCHEMA_VERSION,
+        manager_name=manager_caps.name,
+        pulsar_version=pulsar_version,
+        staging_directory=app.staging_directory,
+        persistence_directory=app.persistence_directory,
+        tool_dependency_dir=_tool_dependency_dir(app),
+        dependency_resolvers=resolvers,
+        conda_available=_conda_available(resolvers),
+        container_runtime=_detect_container_runtime(),
+        manager=manager_caps,
+    )
+
+
+def collect_all_capabilities(app: "PulsarApp") -> Dict[str, PulsarCapabilities]:
+    """Map ``manager_name -> PulsarCapabilities`` for every manager on the app.
+
+    Used by ``PulsarApp.__init__`` to populate the cache. Failures on
+    individual managers are logged and skipped so a misconfigured
+    manager cannot prevent the app from coming up.
+    """
+    out: Dict[str, PulsarCapabilities] = {}
+    for name, manager in app.managers.items():
+        try:
+            out[name] = collect_capabilities(app, manager)
+        except Exception:
+            log.exception("Failed to collect capabilities for manager %s", name)
+    return out
+
+
+def _collect_dependency_resolvers(app: "PulsarApp") -> List[DependencyResolverInfo]:
+    out: List[DependencyResolverInfo] = []
+    for r in app.dependency_manager.dependency_resolvers:
+        # ``resolver_type`` and ``versionless`` are class attributes on
+        # concrete subclasses, not the base — probe defensively rather
+        # than narrow the loop type to every known subclass.
+        rt = getattr(r, "resolver_type", None)
+        if not rt:
+            continue
+        info = DependencyResolverInfo(
+            type=str(rt),
+            disabled=r.disabled,
+            versionless=bool(getattr(r, "versionless", False)),
+        )
+        if isinstance(r, CondaDependencyResolver):
+            info.auto_init = bool(r.auto_init)
+            info.auto_install = bool(r.auto_install)
+            info.prefix = str(r.prefix) if r.prefix else None
+        out.append(info)
+    return out
+
+
+def _conda_available(resolvers: List[DependencyResolverInfo]) -> bool:
+    """A conda resolver is enabled and conda is reachable on this host.
+
+    "Reachable" means either the conda binary is on PATH or the resolver's
+    configured prefix exists on disk — Galaxy on the requesting side wants
+    to know whether ``dependency_resolution=remote`` will actually find
+    something to run.
+    """
+    for r in resolvers:
+        if r.type != "conda" or r.disabled:
+            continue
+        if shutil.which("conda"):
+            return True
+        if r.prefix and os.path.isdir(r.prefix):
+            return True
+    return False
+
+
+def _detect_container_runtime() -> ContainerRuntimeInfo:
+    return ContainerRuntimeInfo(
+        docker_available=bool(shutil.which("docker")),
+        singularity_available=bool(shutil.which("singularity")),
+        apptainer_available=bool(shutil.which("apptainer")),
+    )
+
+
+def _collect_manager(manager: "StatefulManagerProxy") -> ManagerCapabilities:
+    underlying = manager._proxied_manager
+    # ``manager_type`` is a class attribute set on each concrete manager
+    # implementation but not on the proxy or its base. The ``work_threads``
+    # attribute is queued_python-only — falling back to a public
+    # ``num_concurrent_jobs`` covers any future manager that exposes one.
+    mtype = getattr(type(underlying), "manager_type", "unknown")
+    work_threads = getattr(underlying, "work_threads", None)
+    num: Union[int, str, None]
+    if work_threads is not None:
+        try:
+            num = len(work_threads)
+        except TypeError:
+            num = None
+    else:
+        num = getattr(underlying, "num_concurrent_jobs", None)
+    return ManagerCapabilities(name=manager.name, type=str(mtype), num_concurrent_jobs=num)
+
+
+def _tool_dependency_dir(app: "PulsarApp") -> Optional[str]:
+    base = getattr(app.dependency_manager, "default_base_path", None)
+    return str(base) if base else None

pulsar/core.py

@@ -15,6 +15,7 @@
     messaging,
 )
 from pulsar.cache import Cache
+from pulsar.capabilities import collect_all_capabilities
 from pulsar.manager_factory import build_managers
 from pulsar.tools import ToolBox
 from pulsar.tools.authorization import get_authorizer
@@ -50,6 +51,7 @@ def __init__(self, **conf):
         self.__setup_job_metrics(conf)
         self.__setup_user_auth_manager(conf)
         self.__setup_managers(conf)
+        self.__setup_capabilities()
         self.__setup_file_cache(conf)
         self.__setup_bind_to_message_queue(conf)
         self.__recover_jobs()
@@ -122,6 +124,12 @@ def __setup_staging_directory(self, staging_directory):
     def __setup_managers(self, conf):
         self.managers = build_managers(self, conf)
 
+    def __setup_capabilities(self):
+        # Snapshot of static config + host probes that get_capabilities
+        # publishes to the relay. Computed once; stays valid for this
+        # process's lifetime since pulsar doesn't reload app.yml.
+        self.capabilities_by_manager = collect_all_capabilities(self)
+
     def __recover_jobs(self):
         for manager in self.managers.values():
             manager.recover_active_jobs()

pulsar/messaging/__init__.py

@@ -23,8 +23,19 @@ def bind_app(app, queue_id, conf=None):
         log.info("Detected relay connection string, binding to pulsar-relay at %s", relay_url)
 
         relay_state = RelayState()
+        merged_conf = conf or {}
         for manager in app.managers.values():
-            bind_relay.bind_manager_to_relay(manager, relay_state, relay_url, conf or {})
+            # Build the transport once per manager and reuse for both the
+            # control-message bind and the capabilities publish, so the
+            # initial token fetch + cursor file open happen exactly once.
+            relay_transport = bind_relay.build_relay_transport(manager, relay_url, merged_conf)
+            bind_relay.bind_manager_to_relay(
+                manager, relay_state, relay_url, merged_conf,
+                relay_transport=relay_transport,
+            )
+            bind_relay.publish_manager_capabilities_to_relay(
+                app, manager, relay_transport, merged_conf,
+            )
         return relay_state
     else:
         # Use AMQP binding