Pulsar relay implementation

lib/galaxy/jobs/runners/pulsar.py

@@ -197,6 +197,18 @@ PULSAR_PARAM_SPECS = dict(
         map=int,
         default=None,
     ),
+    relay_url=dict(
+        map=specs.to_str_or_none,
+        default=None,
+    ),
+    relay_username=dict(
+        map=specs.to_str_or_none,
+        default=None,
+    ),
+    relay_password=dict(
+        map=specs.to_str_or_none,
+        default=None,
+    ),
 )
 
 
@@ -260,7 +272,7 @@ class PulsarJobRunner(AsynchronousJobRunner[AsynchronousJobState]):
             client_manager_kwargs[kwd] = self.runner_params[kwd]
 
         for kwd in self.runner_params.keys():
-            if kwd.startswith("amqp_") or kwd.startswith("transport_"):
+            if kwd.startswith("amqp_") or kwd.startswith("transport_") or kwd.startswith("relay_"):
                 client_manager_kwargs[kwd] = self.runner_params[kwd]
 
         return client_manager_kwargs

lib/galaxy_test/driver/integration_util.py

@@ -8,6 +8,7 @@ testing configuration.
 import os
 import re
 import string
+import subprocess
 import sys
 from collections.abc import Iterator
 from typing import (
@@ -49,6 +50,56 @@ SCRIPT_DIRECTORY = os.path.abspath(os.path.dirname(__file__))
 VAULT_CONF = os.path.join(SCRIPT_DIRECTORY, "vault_conf.yml")
 
 
+def docker_run(image, name, *args, detach=True, remove=True, ports=None, env_vars: Optional[dict[str, str]] = None):
+    cmd = ["docker", "run"]
+
+    if ports:
+        for host_port, container_port in ports:
+            cmd.extend(["-p", f"{host_port}:{container_port}"])
+
+    if detach:
+        cmd.append("-d")
+
+    cmd.extend(["--name", name])
+
+    if remove:
+        cmd.append("--rm")
+    if env_vars:
+        for key, value in env_vars.items():
+            cmd.extend(["-e", f"{key}={value}"])
+
+    cmd.append(image)
+    cmd.extend(args)
+    print("Running docker command:", " ".join(cmd))
+
+    subprocess.check_call(cmd)
+
+
+def docker_exec(container_name, *args, output=True):
+    cmd = ["docker", "exec", container_name]
+    cmd.extend(args)
+
+    if output:
+        return subprocess.check_output(cmd)
+    else:
+        subprocess.check_call(cmd)
+
+
+def docker_ip_address(container_name):
+    cmd = [
+        "docker",
+        "inspect",
+        "-f",
+        "{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}",
+        container_name,
+    ]
+    return subprocess.check_output(cmd).decode("utf-8").strip()
+
+
+def docker_rm(container_name):
+    subprocess.check_call(["docker", "rm", "-f", container_name])
+
+
 def skip_if_jenkins(cls):
     if os.environ.get("BUILD_NUMBER", ""):
         return skip

test/integration/objectstore/_base.py

@@ -6,6 +6,12 @@ import time
 
 from galaxy_test.base.populators import DatasetPopulator
 from galaxy_test.driver import integration_util
+from galaxy_test.driver.integration_util import (
+    docker_exec,
+    docker_ip_address,
+    docker_rm,
+    docker_run,
+)
 
 OBJECT_STORE_HOST = os.environ.get("GALAXY_INTEGRATION_OBJECT_STORE_HOST", "127.0.0.1")
 OBJECT_STORE_PORT = int(os.environ.get("GALAXY_INTEGRATION_OBJECT_STORE_PORT", 9000))
@@ -377,49 +383,3 @@ def await_oneprovider_demo_readiness(op_container_name):
 
 def get_onedata_access_token(oz_container_name):
     return docker_exec(oz_container_name, "demo-access-token").decode("utf-8").strip()
-
-
-def docker_run(image, name, *args, detach=True, remove=True, ports=None):
-    cmd = ["docker", "run"]
-
-    if ports:
-        for container_port, host_port in ports:
-            cmd.extend(["-p", f"{container_port}:{host_port}"])
-
-    if detach:
-        cmd.append("-d")
-
-    cmd.extend(["--name", name])
-
-    if remove:
-        cmd.append("--rm")
-
-    cmd.append(image)
-    cmd.extend(args)
-
-    subprocess.check_call(cmd)
-
-
-def docker_exec(container_name, *args, output=True):
-    cmd = ["docker", "exec", container_name]
-    cmd.extend(args)
-
-    if output:
-        return subprocess.check_output(cmd)
-    else:
-        subprocess.check_call(cmd)
-
-
-def docker_ip_address(container_name):
-    cmd = [
-        "docker",
-        "inspect",
-        "-f",
-        "{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}",
-        container_name,
-    ]
-    return subprocess.check_output(cmd).decode("utf-8").strip()
-
-
-def docker_rm(container_name):
-    subprocess.check_call(["docker", "rm", "-f", container_name])

test/integration/test_pulsar_embedded_relay.py

@@ -0,0 +1,109 @@
+"""Integration tests for the Pulsar embedded runner with outputs to working directory."""
+
+import os
+import string
+import tempfile
+import uuid
+
+from galaxy.util import safe_makedirs
+from galaxy_test.base.populators import DatasetPopulator
+from galaxy_test.driver import integration_util
+from galaxy_test.driver.integration_util import (
+    docker_rm,
+    docker_run,
+)
+
+SCRIPT_DIRECTORY = os.path.abspath(os.path.dirname(__file__))
+EMBEDDED_PULSAR_JOB_CONFIG_FILE = os.path.join(SCRIPT_DIRECTORY, "embedded_pulsar_mq_job_conf.yml")
+AMQP_URL = os.environ.get("GALAXY_TEST_PULSAR_RELAY")
+
+JOB_CONF_TEMPLATE = """
+runners:
+  local:
+    load: galaxy.jobs.runners.local:LocalJobRunner
+    workers: 1
+  pulsar:
+    load: galaxy.jobs.runners.pulsar:PulsarEmbeddedMQJobRunner
+    pulsar_app_config:
+      tool_dependency_dir: none
+      conda_auto_init: false
+      conda_auto_install: false
+      message_queue_url: ${amqp_url}
+      message_queue_username: admin
+      message_queue_password: ${relay_password}
+      staging_directory: ${jobs_directory}
+    relay_url: ${amqp_url}
+    relay_username: admin
+    relay_password: ${relay_password}
+execution:
+  default: pulsar_mq_environment
+  environments:
+    pulsar_mq_environment:
+      runner: pulsar
+      rewrite_parameters: true
+      dependency_resolution: none
+      default_file_action: remote_transfer
+      jobs_directory: ${jobs_directory}
+      remote_metadata: true
+      remote_property_galaxy_home: ${galaxy_home}
+    local_environment:
+      runner: local
+tools:
+  - id: __DATA_FETCH__
+    environment: local_environment
+"""
+
+
+def _handle_galaxy_config_kwds(cls, config):
+    random_password = uuid.uuid4().hex
+    docker_run(
+        "mvdbeek/pulsar-relay",
+        cls.container_name,
+        env_vars={"PULSAR_BOOTSTRAP_ADMIN_USERNAME": "admin", "PULSAR_BOOTSTRAP_ADMIN_PASSWORD": random_password},
+        remove=False,
+        ports=[(9001, 8080)],
+    )
+    jobs_directory = os.path.join(cls._test_driver.mkdtemp(), "pulsar_staging")
+    safe_makedirs(jobs_directory)
+    job_conf_template = string.Template(JOB_CONF_TEMPLATE)
+    job_conf_str = job_conf_template.substitute(
+        amqp_url="http://localhost:9001",
+        jobs_directory=jobs_directory,
+        galaxy_home=os.path.join(SCRIPT_DIRECTORY, os.pardir),
+        relay_password=random_password,
+    )
+    with tempfile.NamedTemporaryFile(suffix="_mq_job_conf.yml", mode="w", delete=False) as job_conf:
+        job_conf.write(job_conf_str)
+    config["job_config_file"] = job_conf.name
+    infrastructure_url = "http://localhost:$GALAXY_WEB_PORT"
+    config["galaxy_infrastructure_url"] = infrastructure_url
+
+
+class EmbeddedPulsarRelayIntegrationInstance(integration_util.IntegrationInstance):
+    """Describe a Galaxy test instance with embedded pulsar configured.
+
+    $ Setup RabbitMQ (e.g. https://www.rabbitmq.com/install-homebrew.html)
+    $ GALAXY_TEST_AMQP_URL='amqp://guest:guest@localhost:5672//' pytest -s test/integration/test_pulsar_embedded_mq.py
+    """
+
+    container_name = "pulsar-relay"
+    dataset_populator: DatasetPopulator
+    framework_tool_and_types = True
+
+    @classmethod
+    def handle_galaxy_config_kwds(cls, config):
+        _handle_galaxy_config_kwds(cls, config)
+
+    @classmethod
+    def tearDownClass(cls):
+        try:
+            super().tearDownClass()
+        finally:
+            docker_rm(cls.container_name)
+
+
+instance = integration_util.integration_module_instance(EmbeddedPulsarRelayIntegrationInstance)
+
+test_tools = integration_util.integration_tool_runner(
+    ["simple_constructs", "composite_output_tests", "all_output_types"]
+)