Prevent duplicate edge workers unless existing worker is offline or unkown

[dheerajturaga]

Add validation to the edge worker registration endpoint to prevent
launching multiple workers with the same hostname. If a worker with
the same name already exists in an active state (running, idle,
starting, terminating, or maintenance), the registration will fail
with HTTP 409 CONFLICT. Workers can only reuse a name if the existing
worker is in OFFLINE, UNKNOWN, or OFFLINE_MAINTENANCE state.

[jscheffl]

Thanks! This is a very good point that is good to be fixed to prevent deployment errors to mix-up hostnames.

One nit only:
When I start a second worker, the pure HTTP error is in the logs like:

root@bbd99465f97f:/opt/airflow# airflow edge worker --pid another
2025-11-23T09:31:39.223750Z [info     ] Starting worker with API endpoint http://localhost:8080/edge_worker/v1/rpcapi [airflow.providers.edge3.cli.edge_command] loc=edge_command.py:80
  ____________       _____________
 ____    |__( )_________  __/__  /________      __
____  /| |_  /__  ___/_  /_ __  /_  __ \_ | /| / /
___  ___ |  / _  /   _  __/ _  / / /_/ /_ |/ |/ /
 _/_/  |_/_/  /_/    /_/    /_/  \____/____/|__/
   ____   __           _      __         __
  / __/__/ /__ ____   | | /| / /__  ____/ /_____ ____
 / _// _  / _ `/ -_)  | |/ |/ / _ \/ __/  '_/ -_) __/
/___/\_,_/\_, /\__/   |__/|__/\___/_/ /_/\_\\__/_/
         /___/

409 Client Error: Conflict for url: http://localhost:8080/edge_worker/v1/worker/bbd99465f97f

Can you explicitly handle the exception on the client and generate a better log message in the console? e.g. like we have for version conflicts in providers/edge3/src/airflow/providers/edge3/cli/api_client.py:133

[jscheffl]

Just realized: This PR will now lower a bit comfort though, when using breeze and then stop_airflow the docker is killed and if you then re-start the previous worker state is not cleaned and restarting directly with breeze will raise the HTTP 409 then :-) But still OK in my view.

[dheerajturaga]

Can you explicitly handle the exception on the client and generate a better log message in the console? e.g. like we have for version conflicts in providers/edge3/src/airflow/providers/edge3/cli/api_client.py:133

Does this look good to you? I made this update

2025-11-23T18:30:05.778730Z [info     ] Starting worker with API endpoint http://localhost:8080/edge_worker/v1/rpcapi [airflow.providers.edge3.cli.edge_command] loc=edge_command.py:80
  ____________       _____________
 ____    |__( )_________  __/__  /________      __
____  /| |_  /__  ___/_  /_ __  /_  __ \_ | /| / /
___  ___ |  / _  /   _  __/ _  / / /_/ /_ |/ |/ /
 _/_/  |_/_/  /_/    /_/    /_/  \____/____/|__/
   ____   __           _      __         __
  / __/__/ /__ ____   | | /| / /__  ____/ /_____ ____
 / _// _  / _ `/ -_)  | |/ |/ / _ \/ __/  '_/ -_) __/
/___/\_,_/\_, /\__/   |__/|__/\___/_/ /_/\_\\__/_/
         /___/

2025-11-23T18:30:05.924089Z [error    ] A worker with the name 'breeze' is already active. Please ensure worker names are unique, or stop the existing worker before starting a new one. [airflow.providers.edge3.cli.worker] loc=worker.py:270
A worker with the name 'breeze' is already active. Please ensure worker names are unique, or stop the existing worker before starting a new one.

[dheerajturaga]

Just realized: This PR will now lower a bit comfort though, when using breeze and then stop_airflow the docker is killed and if you then re-start the previous worker state is not cleaned and restarting directly with breeze will raise the HTTP 409 then :-) But still OK in my view.

Thats true, looks like the state does not get cleared. You would need to wait a few minutes for the api server to determine worker hearbeat is missing and change to unknown state, then you will be able to launch. OR just launch with -d (which I generally always use when testing).

I don't think this is a big deal though.

2025-11-23T18:42:55.005572Z [error    ] A worker with the name 'breeze' is already active. Please ensure worker names are unique, or stop the existing worker before starting a new one. [airflow.providers.edge3.cli.worker] loc=worker.py:270
A worker with the name 'breeze' is already active. Please ensure worker names are unique, or stop the existing worker before starting a new one.
root@99fa52cb933e:/opt/airflow# airflow edge worker --edge-hostname breeze --queues default
2025-11-23T18:45:21.402431Z [info     ] Starting worker with API endpoint http://localhost:8080/edge_worker/v1/rpcapi [airflow.providers.edge3.cli.edge_command] loc=edge_command.py:80
  ____________       _____________
 ____    |__( )_________  __/__  /________      __
____  /| |_  /__  ___/_  /_ __  /_  __ \_ | /| / /
___  ___ |  / _  /   _  __/ _  / / /_/ /_ |/ |/ /
 _/_/  |_/_/  /_/    /_/    /_/  \____/____/|__/
   ____   __           _      __         __
  / __/__/ /__ ____   | | /| / /__  ____/ /_____ ____
 / _// _  / _ `/ -_)  | |/ |/ / _ \/ __/  '_/ -_) __/
/___/\_,_/\_, /\__/   |__/|__/\___/_/ /_/\_\\__/_/
         /___/

2025-11-23T18:45:21.593074Z [info     ] An existing PID file has been found: /root/airflow/airflow-edge-worker.pid. [airflow.providers.edge3.cli.signalling] loc=signalling.py:53
2025-11-23T18:45:21.593307Z [warning  ] PID file is orphaned. Cleaning up. [airflow.providers.edge3.cli.signalling] loc=signalling.py:65
2025-11-23T18:45:21.618799Z [info     ] No new job to process          [airflow.providers.edge3.cli.worker] loc=worker.py:336

[jscheffl]

That works! Thanks!