Galaxy not including appropriate routing_key for RabbitMQ

[peterg1t]

Hi,
I am not 100 % sure if this is a bug or not but I think only a Galaxy developer can shine some light for me here. I'm trying to setup a galaxy server with a pulsar setup. I am following the documentation here when trying to setup a Docker deployment of Galaxy that uses pulsar:
Galaxy training
I have encountered an issue though. After initializing the rabbitmq server I can see three channels connecting to the broker. One for each of the queues (pulsar__kill_ack, pulsar__setup_ack, pulsar__status_update)

My pulsar exchange points to 3 queues

So far so good. I do not have a pulsar server yet listening to the queues so I would expect that every tool emits a message that ends up in one of these 3 queues. However every message sent shows up in a new channel and in consequence is dropped (unrouted).

I contrast when I use the simple code

#!/usr/bin/env python
import pika
import ssl


context = ssl.create_default_context(cafile="/root/docker-galaxy/galaxy/traefik/config/ssl/galaxy.rcs.ucalgary.ca-combined.crt")
context.verify_mode = ssl.CERT_REQUIRED
context.load_cert_chain("/root/docker-galaxy/galaxy/traefik/config/ssl/galaxy.rcs.ucalgary.ca.crt",
                        "/root/docker-galaxy/galaxy/traefik/config/ssl/galaxy.rcs.ucalgary.ca.key")

credentials = pika.PlainCredentials('pulsar_au', 'pulsar_pass')

ssl_options = pika.SSLOptions(context, "galaxy.rcs.ucalgary.ca")
conn_params = pika.ConnectionParameters(port=5671,
                                        ssl_options=ssl_options,
                                        virtual_host='/pulsar/pulsar_au',
                                        credentials=credentials)

with pika.BlockingConnection(conn_params) as conn:
    channel = conn.channel()
    channel.basic_publish(exchange='pulsar',
                          routing_key='pulsar__setup_ack',
                          body='Hello World!')

print(" [x] Sent 'Hello World!'")

Where the routing key is specified correctly according to the queue name, everything works just fine. AS far as I understand these routing keys are sent by Galaxy and cannot be specified in a tool by the admin.
I was wondering if there is a bug with Galaxy not sending the appropriate routing_key to RabbitMQ in its messages (using a pulsar-runner) or there is an obvious miss-configuration issue, or the Dockerfile is not up to date?

Here are all my configuration files for the galaxy deployment:
galaxy-container.tar.gz

I would appreciate any insights into this issue and my apologies if this is not the correct forum to raise this issue.

Best regards,

Pedro Martinez

[mvdbeek]

It's a little unclear if you're actually having a problem when pulsar is running ?

[peterg1t]

Hi @mvdbeek,
That is a good point. In this example pulsar is not running in the remote system. The way I understand this (and please correct me if I'm wrong) is based on the following diagram

If we focus on the left side only I should be able to send messages from the pulsar runner and these will be queued in the RabbitMQ (lets assume for a moment that the right side is not setup so messages are not retrieved). However following the training document the messages do not make it to the queue. I'm trying to understand why. Which routing keys do galaxy use to send the messages? I think I've seen the following.

Which are different from the routing keys pulsar uses for the queues. Where are these configured? It is not clear in the tutorials and documentation.
Hope this illustrate a bit more the question I'm trying to answer. Thanks for taking a look into this.
Regards,
Pedro

[mvdbeek]

Yes, the diagram looks roughly correct, but the screenshot of the code is for Galaxy's internal IPC system that is entirely unrelated to job running. When you're saying

the messages do not make it to the queue.

how are you producing the messages ? You've set up a pulsar job runner, mapped a tool to that job runner and then started a job ?

[peterg1t]

Yes sorry for that I'm fairly new to Galaxy and trying to understand the inner workings takes a bit of time. This is how I have setup the pulsar job runner.
In my galaxy.yml config file I have the following configuration:

job_config:
        runners:
            local_runner:
                load: galaxy.jobs.runners.local:LocalJobRunner
                workers: 4
            pulsar_runner:
                load: galaxy.jobs.runners.pulsar:PulsarMQJobRunner
                amqp_url: "pyamqp://pulsar_au:pulsar_pass@galaxy.rcs.ucalgary.ca:5671//pulsar/pulsar_au?ssl=1"
                amqp_acknowledge: True
                amqp_ack_republish_time: 300
                amqp_consumer_timeout: 2
                amqp_publish_retry: True
                amqp_publish_retry_max_retries: 60
                galaxy_url: "https://galaxy.rcs.ucalgary.ca"
                manager: _default_ # pulsar_<manager>__queue
        execution:
            default: tpv_dispatcher
            environments:
                local_env:
                    runner: local_runner
                    tmp_dir: True
                tpv_dispatcher:
                    runner: dynamic
                    type: python
                    function: map_tool_to_destination
                    rules_module: tpv.rules
                    tpv_config_files:
                        - https://gxy.io/tpv/db.yml
                        - /galaxy/server/config/tpv_rules_local.yml
        tools:
        -   class: local
            environment: local_env

and my tpv_rules_local.yml looks like this:

tools:
  default:
    abstract: true
    cores: 1
    mem: cores * 4
    params:
      walltime: 16
    scheduling:
      require:
        - pulsar
  
destinations:
  local_env:
    runner: local_runner
    max_accepted_cores: 1
    params:
      tmp_dir: true
  pulsar:
    runner: pulsar_runner
    params:
      default_file_action: remote_transfer
      dependency_resolution: remote
      jobs_directory: /mnt/pulsar/files/staging
      persistence_directory: /mnt/pulsar/files/persisted_data
      remote_metadata: false
      rewrite_parameters: true
      transport: curl
      outputs_to_working_directory: false
    scheduling:
      require:
        - pulsar

If my understanding is correct by setting the default requiring pulsar. All tools will use it? When I start a simple job the messages are sent to RabbitMQ correctly but there is no routing key and they get discarded instead of going to one of the three pulsar queues.

[peterg1t]

I shouldn't say there is no routing key but the routing key values do not seem to match the queue names. I also tried to replace the routing key to * in RabbitMQ just to see if any message gets routed to a queue to no avail. The messages are still unroutable but they do make it to the exchange.

[peterg1t]

So I have done a bit more work setting up Pulsar and I here is more info on what is happening.

Pulsar expects the following names for the queues (pulsar__setup, pulsar__kill, pulsar__status) which do not match the names setup by default in rabbitmq (pulsar__setup_ack, pulsar__kill_ack, pulsar__status_update). Is there a reason for this difference or is this a bug? I can redefine the queues explicitly like this in my definitions file:

    "vhosts": [
        {
            "limits": [],
            "metadata": {
                "description": "pulsar vhost",
                "tags": []
            },
            "name": "/pulsar/pulsar_au"
        },
        {
            "limits": [],
            "metadata": {
                "description": "Default virtual host",
                "tags": []
            },
            "name": "/"
        }
    ],
    "permissions": [
        {
            "user": "admin",
            "vhost": "/",
            "configure": ".*",
            "read": ".*",
            "write": ".*"
        },
        {
            "user": "admin",
            "vhost": "/pulsar/pulsar_au",
            "configure": ".*",
            "read": ".*",
            "write": ".*"
        },
        {
            "user": "pulsar_au",
            "vhost": "/pulsar/pulsar_au",
            "configure": ".*",
            "read": ".*",
            "write": ".*"
        }
    ],
    "parameters": [],
    "policies": [],
    "exchanges": [
        {
          "name": "pulsar",
          "vhost": "/pulsar/pulsar_au",
          "type": "direct",
          "durable": true,
          "auto_delete": false,
          "internal": false,
          "arguments": {}
        }
      ],
      "queues": [
        {
          "name": "pulsar__kill",
          "vhost": "/pulsar/pulsar_au",
          "durable": true,
          "auto_delete": false,
          "arguments": {}
        },
        {
            "name": "pulsar__setup",
            "vhost": "/pulsar/pulsar_au",
            "durable": true,
            "auto_delete": false,
            "arguments": {}
        },
        {
            "name": "pulsar__status",
            "vhost": "/pulsar/pulsar_au",
            "durable": true,
            "auto_delete": false,
            "arguments": {}
        }
      ],
      "bindings": [
        {
          "source": "pulsar",
          "vhost": "/pulsar/pulsar_au",
          "destination": "pulsar__kill",
          "destination_type": "queue",
          "routing_key": "pulsar__kill",
          "arguments": {}
        },
        {
            "source": "pulsar",
            "vhost": "/pulsar/pulsar_au",
            "destination": "pulsar__setup",
            "destination_type": "queue",
            "routing_key": "pulsar__setup",
            "arguments": {}
        },
        {
            "source": "pulsar",
            "vhost": "/pulsar/pulsar_au",
            "destination": "pulsar__status",
            "destination_type": "queue",
            "routing_key": "pulsar__status",
            "arguments": {}
        }
      ]
}

But things looks ugly in rabbitmq

I can probably remove the older queues but it would make sense to deploy the right ones on the first try.
Cheers,