dagster_celery error "kombu.exceptions.OperationalError: [Errno 111] Connection refused" when materialized => [update] How to configure celery_executor ?

[rv2931]

Hello
I'm trying to execute a task on celery executor on the Open Source Dagster self hosted
I have to admit that the documentation is not really exhaustive so maybe I'm missing something

So this is docker compose stack. When up all container start correctly without error
celery container seems to be well initialized and connected to broker

[2025-01-23 22:04:27,342: INFO/MainProcess] mingle: all alone
[2025-01-23 22:04:27,364: INFO/MainProcess] dagster-45808e@celery ready.

But when I click on "materialize" on my job "all_assets" in the job list in Dagster Ui I get this following error
I don't see where it is coming from. I tried with default celery backend and postgresql backend it is the same...
I'm surely missing something as I'm not so familiar with celery too for the moment

If you have an idea of the solution it could be great
Thank you in advance
Here the error when I click "materialize" my job:

dagster-app   | kombu.exceptions.OperationalError: [Errno 111] Connection refused
dagster-app   |
dagster-app   | Stack Trace:
dagster-app   |   File "/.venv/lib/python3.8/site-packages/dagster/_core/execution/api.py", line 748, in job_execution_iterator
dagster-app   |     for event in job_context.executor.execute(job_context, execution_plan):
dagster-app   |   File "/.venv/lib/python3.8/site-packages/dagster_celery/core_execution_loop.py", line 134, in core_celery_execution_loop
dagster-app   |     step_results[step.key] = step_execution_fn(
dagster-app   |   File "/.venv/lib/python3.8/site-packages/dagster_celery/executor.py", line 125, in _submit_task
dagster-app   |     return task_signature.apply_async(
dagster-app   |   File "/.venv/lib/python3.8/site-packages/celery/canvas.py", line 400, in apply_async
dagster-app   |     return _apply(args, kwargs, **options)
dagster-app   |   File "/.venv/lib/python3.8/site-packages/celery/app/task.py", line 594, in apply_async
dagster-app   |     return app.send_task(
dagster-app   |   File "/.venv/lib/python3.8/site-packages/celery/app/base.py", line 800, in send_task
dagster-app   |     self.backend.on_task_call(P, task_id)
dagster-app   |   File "/.venv/lib/python3.8/site-packages/celery/backends/rpc.py", line 164, in on_task_call
dagster-app   |     maybe_declare(self.binding(producer.channel), retry=True)
dagster-app   |   File "/.venv/lib/python3.8/site-packages/kombu/messaging.py", line 218, in _get_channel
dagster-app   |     channel = self._channel = channel()
dagster-app   |   File "/.venv/lib/python3.8/site-packages/kombu/utils/functional.py", line 34, in __call__
dagster-app   |     value = self.__value__ = self.__contract__()
dagster-app   |   File "/.venv/lib/python3.8/site-packages/kombu/messaging.py", line 234, in <lambda>
dagster-app   |     channel = ChannelPromise(lambda: connection.default_channel)
dagster-app   |   File "/.venv/lib/python3.8/site-packages/kombu/connection.py", line 953, in default_channel
dagster-app   |     self._ensure_connection(**conn_opts)
dagster-app   |   File "/.venv/lib/python3.8/site-packages/kombu/connection.py", line 459, in _ensure_connection
dagster-app   |     return retry_over_time(
dagster-app   |   File "/usr/lib/python3.8/contextlib.py", line 131, in __exit__
dagster-app   |     self.gen.throw(type, value, traceback)
dagster-app   |   File "/.venv/lib/python3.8/site-packages/kombu/connection.py", line 476, in _reraise_as_library_errors
dagster-app   |     raise ConnectionError(str(exc)) from exc
dagster-app   |
dagster-app   | The above exception was caused by the following exception:
dagster-app   | ConnectionRefusedError: [Errno 111] Connection refused
dagster-app   |
dagster-app   | Stack Trace:
dagster-app   |   File "/.venv/lib/python3.8/site-packages/kombu/connection.py", line 472, in _reraise_as_library_errors
dagster-app   |     yield
dagster-app   |   File "/.venv/lib/python3.8/site-packages/kombu/connection.py", line 459, in _ensure_connection
dagster-app   |     return retry_over_time(
dagster-app   |   File "/.venv/lib/python3.8/site-packages/kombu/utils/functional.py", line 318, in retry_over_time
dagster-app   |     return fun(*args, **kwargs)
dagster-app   |   File "/.venv/lib/python3.8/site-packages/kombu/connection.py", line 934, in _connection_factory
dagster-app   |     self._connection = self._establish_connection()
dagster-app   |   File "/.venv/lib/python3.8/site-packages/kombu/connection.py", line 860, in _establish_connection
dagster-app   |     conn = self.transport.establish_connection()
dagster-app   |   File "/.venv/lib/python3.8/site-packages/kombu/transport/pyamqp.py", line 203, in establish_connection
dagster-app   |     conn.connect()
dagster-app   |   File "/.venv/lib/python3.8/site-packages/amqp/connection.py", line 324, in connect
dagster-app   |     self.transport.connect()
dagster-app   |   File "/.venv/lib/python3.8/site-packages/amqp/transport.py", line 132, in connect
dagster-app   |     self._connect(self.host, self.port, self.connect_timeout)
dagster-app   |   File "/.venv/lib/python3.8/site-packages/amqp/transport.py", line 187, in _connect
dagster-app   |     self.sock.connect(sa)
dagster-app   |
dagster-app   | The above exception occurred during handling of the following exception:
dagster-app   | AttributeError: 'ChannelPromise' object has no attribute '__value__'
dagster-app   |
dagster-app   | Stack Trace:
dagster-app   |   File "/.venv/lib/python3.8/site-packages/kombu/utils/functional.py", line 32, in __call__
dagster-app   |     return self.__value__

here you'll find my configurations. I think all needed files are here but don't hesitate to ask me more information if needed

# .assets.py
from dagster import AssetExecutionContext,asset

@asset
def csi_my_first_asset(context: AssetExecutionContext):
    return [1,2,3]

# definitions.py
from dagster import (
    Definitions,
    load_assets_from_modules,
    define_asset_job
)
from dagster_celery import celery_executor

from . import assets

all_assets = load_assets_from_modules([assets])

defs=Definitions(
    assets=all_assets,
    jobs=[define_asset_job("all_assets",executor_def=celery_executor,description="my first job")],
)

# celery.yaml
execution:
  config:
      broker: 'pyamqp://dagster:dagster@broker:5672/dagster-broker' 
      backend: 'db+postgresql://dagster:dagster@database/dagster'

# dagster.yaml
storage:
      postgres:
            postgres_db:
                  username:
                        env: DAGSTER_PG_USERNAME
                  password:
                        env: DAGSTER_PG_PASSWORD
                  hostname:
                        env: DAGSTER_PG_HOST
                  db_name:
                        env: DAGSTER_PG_DB
                  port:
                        env: DAGSTER_PG_PORT

# docker-compose.yml
services:
    database:
        image: postgres:16
        container_name: database
        hostname: database
        environment:
            - POSTGRES_USER=${DATABASE_USER:-dagster}
            - POSTGRES_PASSWORD=${DATABASE_PASSWORD:-dagster}
            - POSTGRES_DB=${DATABASE_NAME:-dagster}
            - TZ=utc
        ports:
            - ${DATABASE_PORT:-5432}:5432
        volumes:
            - pgdata:/var/lib/postgresql/data
    dagster:
        image: dagster-app:${DAGSTER_TAG:-latest}
        container_name: dagster-app
        hostname: dagster-app
        working_dir: /dagster
        ports:
            - ${DAGSTER_PORT:-3000}:3000
        environment:
            - DAGSTER_PG_USERNAME=${DATABASE_USER:-dagster}
            - DAGSTER_PG_PASSWORD=${DATABASE_PASSWORD:-dagster}
            - DAGSTER_PG_DB=${DATABASE_NAME:-dagster}
            - DAGSTER_PG_HOST=database
            - DAGSTER_PG_PORT=5432
        volumes:
            - ./docker/config/dagster.yaml:/dagster/dagster.yaml
            - ./projects:/dagster/projects
            - ./workspace.yaml:/dagster/workspace.yaml
        depends_on:
            - celery
    broker:
        image: rabbitmq:${RABBITMQ_TAG:-latest}
        build:
            context: .
            dockerfile: ./docker/rabbitmq/Dockerfile
        hostname: "broker"
        container_name: "broker"
        ports:
            - ${BROKER_PORT:-15672}:15672 
        environment:
            - RABBITMQ_DEFAULT_VHOST=dagster-broker
            - RABBITMQ_DEFAULT_USER=dagster
            - RABBITMQ_DEFAULT_PASS=dagster
        depends_on:
            - init
    celery:
        image: dagster-app:${DAGSTER_TAG:-latest}
        hostname: "celery"
        container_name: "celery"
        stdin_open: true
        tty: true
        command: dagster-celery worker start -A dagster_celery.app -y /dagster/celery.yaml
        volumes:
            - ./docker/config/dagster.yaml:/dagster/dagster.yaml
            - ./docker/config/celery.yaml:/dagster/celery.yaml
            - ./worker.py:/dagster/worker.py
            - ./docker/volumes/dagster_celery:/dagster/dagster_celery
        depends_on:
            - broker
    init:
        image: dagster-app:${DAGSTER_TAG:-latest}
        build:
            context: .
            dockerfile: ./docker/dagster/Dockerfile
            args:
                - DAGSTER_HOME=${BUILD_DAGSTER_HOME:-/dagster}
        container_name: dagster-init
        hostname: dagster-init
        command: /bin/sh
        working_dir: /dagster
        volumes:
            - ./docker/config/dagster.yaml:/dagster/dagster.yaml
            - ./projects:/dagster/projects
            - ./requirements.pip:/requirements.pip
            - ./workspace.yaml:/dagster/workspace.yaml
        depends_on:
            - database

volumes:
    pgdata:

[rv2931]

Hi
So I managed to make it work
So first of all: "kombu.exceptions.OperationalError: [Errno 111] Connection refused" erreor is really a problem of connection to the broker

My problem was that I didn't configure broker/backend for the celery_executor

I had a broker rabbitmq pyamqp://dagster:dagster@broker:5672/dagster-broker with user:dagster, password dagster and vhost: dagster-broker
I had a celery worker that manage to connect to the broker and backend using the celery.yaml file with correct user/password/vhost

The problem was that I had no configuration of broker/backend for the celery client that is celery_executor
I wanted to set it dagster.yaml but I didn't found any valid configuration (no execution: entry is accepted/excpected)
I didn't find any other way to do it, so celery_executor was trying to connect to default broker/backend and not mine
So the error kombu.exceptions.OperationalError: [Errno 111] Connection refused is disturbing because it shouldn't bu connexion refused because there is no rabbitmq server at default broker http://guest:[email protected]:5762//

Finally I make it work by adding celery_executor configuration directly in definition:

# definitions.py
from dagster import (
    Definitions,
    load_assets_from_modules,
    define_asset_job
)
from dagster_celery import celery_executor

from . import assets

all_assets = load_assets_from_modules([assets])

defs=Definitions(
    assets=all_assets,
    jobs=[define_asset_job("all_assets",
                           executor_def=celery_executor,
                           description="my first job",
                           config={
                               "execution":{
                                    "config":{
                                        "broker": "pyamqp://dagster:dagster@broker:5672/dagster-broker",
                                        "backend": "db+postgresql://dagster:dagster@database/dagster",
                                    }
                               }
                           })],
)

Now my question is: is there another way to configure celery_executor by config file or env variable to avoid configuring it each time/job à want to use celery_executor ?