Signal propagation into DockerOperator container

[rileypeterson]

I am running a docker compose file which looks like:

...
  airflow-worker:
    <<: *airflow-common
    command: celery worker --pid /opt/airflow/logs/airflow-worker.pid
    restart: unless-stopped
    environment:
      <<: *airflow-common-env
      # Required to handle warm shutdown of the celery workers properly
      # See https://airflow.apache.org/docs/docker-stack/entrypoint.html#signal-propagation
      DUMB_INIT_SETSID: "0"
  ws:
    build:
      context: ./ws
    image: ws_image:1.0

Along with a dag which runs the following:

ws_task = DockerOperator(
    task_id="ws",
    image="ws_image:1.0",
    api_version="auto",
    auto_remove="force",
    command=["python", "ws.py", "--runtime-seconds", "3600"],
    tty=True,
    dag=dag,
)

Where ws.py is:

import signal
import sys
import time
import asyncio
import random

file = "/tmp/file.txt"

def shutdown_task():
    with open(file, "w") as f:
        f.write("Finished")

def signal_handler1(sig, frame):
    shutdown_task()
    with open(file, "w") as f:
        f.write("Got SIGTERM")
    sys.exit(0)

def signal_handler2(sig, frame):
    shutdown_task()
    with open(file, "w") as f:
        f.write("Got SIGINT")
    sys.exit(0)

def signal_handler3(sig, frame):
    shutdown_task()
    with open(file, "w") as f:
        f.write("Got SIGHUP")
    sys.exit(0)

signal.signal(signal.SIGTERM, signal_handler1)
signal.signal(signal.SIGINT, signal_handler2)
signal.signal(signal.SIGHUP, signal_handler3)

async def main():
    with open(file, "w") as f:
        f.write("Started")
    now = time.time()
    try:
        while time.time() < now + 300:
            print(random.random())
            await asyncio.sleep(2)
    finally:
        shutdown_task()

if __name__ == '__main__':
    asyncio.run(main())

I need shutdown_task() to always run when the program ends or received a shutdown signal. According the comment in the docker compose (which I've taken from the official one) DUMB_INIT_SETSID: "0" is required to handle warm shutdown of the celery workers, but I need to propagate the signals to my docker image... is this possible?

I tried setting DUMB_INIT_SETSID="1" as an environment variable within DockerOperator, which didn't work. When I change it to "1" in my docker compose it does seem to work, but I think this is not recommended (described here). I've tried several other things to no avail. I found a similar unanswered stack overflow question here.

[potiuk]

You can read more how celery processes signals here https://docs.celeryq.dev/en/stable/userguide/workers.html#process-signals and decide yourself what you want to do.

If you want to terminate your tasks on TERM, surely you can set DUMB_INIT_SESSID to 1: - this will propagate the TERM signal to all running celery tasks. Also you can use QUIT instead of term to make celery do it on it's own. the DUM_INIT_SESSID: 0 is simply to preserve default Celery behaviour when if it receives TERM, it does warm shutdown when it stops picking new tasks but it will let all running tasks to complete.

You can also send SIGTERM twice BTW. When Celery is in "warm shutdown" mode second TERM signal will initiate cold shutdown

[rileypeterson]

Thank you! I think the simplest thing for me is to just set DUMB_INIT_SETSID to 1.