This is an automated email from the ASF dual-hosted git repository.
kaxilnaik pushed a commit to branch v3-0-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-0-test by this push:
new dbcd32b24ce Make default execution server URL be relative to API Base
URL (#49747)
dbcd32b24ce is described below
commit dbcd32b24ce8d746d8c5055b56053eeb596a30e6
Author: Kaxil Naik <[email protected]>
AuthorDate: Fri Apr 25 01:41:01 2025 +0530
Make default execution server URL be relative to API Base URL (#49747)
The execution API server URL is now more flexible and configurable:
- Remove hardcoded default URL "http://localhost:8080/execution/";
- Add fallback mechanism to construct URL from base_url config
- Default is now `"{BASE_URL}/execution/"` where BASE_URL comes from
`api.base_url` config
(cherry picked from commit ae7b3c6abc89767ffc9fdd7738e67d0d3cad446c)
---
.../src/airflow/config_templates/config.yml | 5 +--
.../src/airflow/executors/local_executor.py | 6 +++-
.../tests/unit/executors/test_local_executor.py | 38 +++++++++++++++++++++-
.../celery/executors/celery_executor_utils.py | 5 ++-
.../airflow/providers/edge3/cli/edge_command.py | 7 +++-
.../airflow/sdk/execution_time/execute_workload.py | 5 ++-
6 files changed, 59 insertions(+), 7 deletions(-)
diff --git a/airflow-core/src/airflow/config_templates/config.yml
b/airflow-core/src/airflow/config_templates/config.yml
index 61641618393..f63e39180e7 100644
--- a/airflow-core/src/airflow/config_templates/config.yml
+++ b/airflow-core/src/airflow/config_templates/config.yml
@@ -536,11 +536,12 @@ core:
default: "4096"
execution_api_server_url:
description: |
- The url of the execution api server.
+ The url of the execution api server. Default is
``{BASE_URL}/execution/``
+ where ``{BASE_URL}`` is the base url of the API Server.
version_added: 3.0.0
type: string
example: ~
- default: "http://localhost:8080/execution/";
+ default: ~
database:
description: ~
options:
diff --git a/airflow-core/src/airflow/executors/local_executor.py
b/airflow-core/src/airflow/executors/local_executor.py
index 87a8e71d358..867c1bc254e 100644
--- a/airflow-core/src/airflow/executors/local_executor.py
+++ b/airflow-core/src/airflow/executors/local_executor.py
@@ -108,6 +108,10 @@ def _execute_work(log: logging.Logger, workload:
workloads.ExecuteTask) -> None:
from airflow.sdk.execution_time.supervisor import supervise
setproctitle(f"airflow worker -- LocalExecutor: {workload.ti.id}")
+
+ base_url = conf.get("api", "base_url", fallback="/")
+ default_execution_api_server = f"{base_url.rstrip('/')}/execution/"
+
# This will return the exit code of the task process, but we don't care
about that, just if the
# _supervisor_ had an error reporting the state back (which will result in
an exception.)
supervise(
@@ -116,7 +120,7 @@ def _execute_work(log: logging.Logger, workload:
workloads.ExecuteTask) -> None:
dag_rel_path=workload.dag_rel_path,
bundle_info=workload.bundle_info,
token=workload.token,
- server=conf.get("core", "execution_api_server_url"),
+ server=conf.get("core", "execution_api_server_url",
fallback=default_execution_api_server),
log_path=workload.log_path,
)
diff --git a/airflow-core/tests/unit/executors/test_local_executor.py
b/airflow-core/tests/unit/executors/test_local_executor.py
index 6279125915b..c693d5a6f76 100644
--- a/airflow-core/tests/unit/executors/test_local_executor.py
+++ b/airflow-core/tests/unit/executors/test_local_executor.py
@@ -26,10 +26,11 @@ from kgb import spy_on
from uuid6 import uuid7
from airflow.executors import workloads
-from airflow.executors.local_executor import LocalExecutor
+from airflow.executors.local_executor import LocalExecutor, _execute_work
from airflow.utils import timezone
from airflow.utils.state import State
+from tests_common.test_utils.config import conf_vars
from tests_common.test_utils.markers import
skip_if_force_lowest_dependencies_marker
pytestmark = pytest.mark.db_test
@@ -166,3 +167,38 @@ class TestLocalExecutor:
pass
finally:
executor.end()
+
+ @pytest.mark.parametrize(
+ ["conf_values", "expected_server"],
+ [
+ (
+ {
+ ("api", "base_url"): "http://test-server";,
+ ("core", "execution_api_server_url"): None,
+ },
+ "http://test-server/execution/";,
+ ),
+ (
+ {
+ ("api", "base_url"): "http://test-server";,
+ ("core", "execution_api_server_url"):
"http://custom-server/execution/";,
+ },
+ "http://custom-server/execution/";,
+ ),
+ ],
+ ids=["base_url_fallback", "custom_server"],
+ )
+ @mock.patch("airflow.sdk.execution_time.supervisor.supervise")
+ def test_execution_api_server_url_config(self, mock_supervise,
conf_values, expected_server):
+ """Test that execution_api_server_url is correctly configured with
fallback"""
+ with conf_vars(conf_values):
+ _execute_work(log=mock.ANY, workload=mock.MagicMock())
+
+ mock_supervise.assert_called_with(
+ ti=mock.ANY,
+ dag_rel_path=mock.ANY,
+ bundle_info=mock.ANY,
+ token=mock.ANY,
+ server=expected_server,
+ log_path=mock.ANY,
+ )
diff --git
a/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py
b/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py
index 555c4147dd0..7956e6d90e7 100644
---
a/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py
+++
b/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py
@@ -158,13 +158,16 @@ def execute_workload(input: str) -> None:
log.info("[%s] Executing workload in Celery: %s", celery_task_id, workload)
+ base_url = conf.get("api", "base_url", fallback="/")
+ default_execution_api_server = f"{base_url.rstrip('/')}/execution/"
+
supervise(
# This is the "wrong" ti type, but it duck types the same. TODO:
Create a protocol for this.
ti=workload.ti, # type: ignore[arg-type]
dag_rel_path=workload.dag_rel_path,
bundle_info=workload.bundle_info,
token=workload.token,
- server=conf.get("core", "execution_api_server_url"),
+ server=conf.get("core", "execution_api_server_url",
fallback=default_execution_api_server),
log_path=workload.log_path,
)
diff --git a/providers/edge3/src/airflow/providers/edge3/cli/edge_command.py
b/providers/edge3/src/airflow/providers/edge3/cli/edge_command.py
index bd7e6c20a13..19dba52deea 100644
--- a/providers/edge3/src/airflow/providers/edge3/cli/edge_command.py
+++ b/providers/edge3/src/airflow/providers/edge3/cli/edge_command.py
@@ -271,6 +271,9 @@ class _EdgeWorkerCli:
setproctitle(f"airflow edge worker: {workload.ti.key}")
try:
+ base_url = conf.get("api", "base_url", fallback="/")
+ default_execution_api_server =
f"{base_url.rstrip('/')}/execution/"
+
supervise(
# This is the "wrong" ti type, but it duck types the same.
TODO: Create a protocol for this.
# Same like in
airflow/executors/local_executor.py:_execute_work()
@@ -278,7 +281,9 @@ class _EdgeWorkerCli:
dag_rel_path=workload.dag_rel_path,
bundle_info=workload.bundle_info,
token=workload.token,
- server=conf.get("core", "execution_api_server_url"),
+ server=conf.get(
+ "core", "execution_api_server_url",
fallback=default_execution_api_server
+ ),
log_path=workload.log_path,
)
return 0
diff --git a/task-sdk/src/airflow/sdk/execution_time/execute_workload.py
b/task-sdk/src/airflow/sdk/execution_time/execute_workload.py
index 8a6a9d0963a..220cd65c5ec 100644
--- a/task-sdk/src/airflow/sdk/execution_time/execute_workload.py
+++ b/task-sdk/src/airflow/sdk/execution_time/execute_workload.py
@@ -54,7 +54,10 @@ def execute_workload(workload: ExecuteTask) -> None:
raise ValueError(f"Executor does not know how to handle
{type(workload)}")
log.info("Executing workload", workload=workload)
- server = conf.get("core", "execution_api_server_url")
+
+ base_url = conf.get("api", "base_url", fallback="/")
+ default_execution_api_server = f"{base_url.rstrip('/')}/execution/"
+ server = conf.get("core", "execution_api_server_url",
fallback=default_execution_api_server)
log.info("Connecting to server:", server=server)
supervise(
