AIP-69: Breeze adjustments for introduction of Edge Executor (#41731) (…

…#43139)

* Breeze adjustments for introduction of AIP-69 Remote Executor

* Rename Remote Executor to Edge Executor

(cherry picked from commit 8b15840)

dev/breeze/doc/images/output_k8s_deploy-airflow.txt

@@ -1 +1 @@
-f36a8a5cdfde95e03f9f42212585bd2a
+afbc231a108f6e3420d1a65703c0d783

dev/breeze/doc/images/output_k8s_run-complete-tests.txt

@@ -1 +1 @@
-ab161044db637ee2593866e55f8a8758
+1399a11a50f55112aeb52179341cca9c

dev/breeze/doc/images/output_k8s_shell.txt

@@ -1 +1 @@
-144a415f833aa9c436eff7d37782c15f
+ce3273f92e4b510d3265fa475e92933f

dev/breeze/doc/images/output_k8s_tests.txt

@@ -1 +1 @@
-959c2e1a03c42b12c268ca59ff1ead5b
+16e4adcf86ffef59e4546db68efe4599

dev/breeze/doc/images/output_shell.txt

@@ -1 +1 @@
-e147d82143c5f3d4d22a7da9a3c181fa
+3136c001725c66fed9f689440c0ff77c

dev/breeze/doc/images/output_start-airflow.txt

@@ -1 +1 @@
-057a38d65515c5db6201648467bc8178
+2fdb4b01e6d949fb40993e3cc416ca5c

dev/breeze/src/airflow_breeze/global_constants.py

@@ -94,19 +94,26 @@
 #   - https://endoflife.date/azure-kubernetes-service
 #   - https://endoflife.date/google-kubernetes-engine
 ALLOWED_KUBERNETES_VERSIONS = ["v1.27.13", "v1.28.9", "v1.29.4", "v1.30.0"]
+
+LOCAL_EXECUTOR = "LocalExecutor"
+KUBERNETES_EXECUTOR = "KubernetesExecutor"
+CELERY_EXECUTOR = "CeleryExecutor"
+CELERY_K8S_EXECUTOR = "CeleryKubernetesExecutor"
+EDGE_EXECUTOR = "EdgeExecutor"
+SEQUENTIAL_EXECUTOR = "SequentialExecutor"
 ALLOWED_EXECUTORS = [
-    "LocalExecutor",
-    "KubernetesExecutor",
-    "CeleryExecutor",
-    "CeleryKubernetesExecutor",
-    "SequentialExecutor",
+    LOCAL_EXECUTOR,
+    KUBERNETES_EXECUTOR,
+    CELERY_EXECUTOR,
+    CELERY_K8S_EXECUTOR,
+    EDGE_EXECUTOR,
+    SEQUENTIAL_EXECUTOR,
 ]
 
 DEFAULT_ALLOWED_EXECUTOR = ALLOWED_EXECUTORS[0]
-START_AIRFLOW_ALLOWED_EXECUTORS = ["LocalExecutor", "CeleryExecutor", "SequentialExecutor"]
+START_AIRFLOW_ALLOWED_EXECUTORS = [LOCAL_EXECUTOR, CELERY_EXECUTOR, EDGE_EXECUTOR, SEQUENTIAL_EXECUTOR]
 START_AIRFLOW_DEFAULT_ALLOWED_EXECUTOR = START_AIRFLOW_ALLOWED_EXECUTORS[0]
-
-SEQUENTIAL_EXECUTOR = "SequentialExecutor"
+ALLOWED_CELERY_EXECUTORS = [CELERY_EXECUTOR, CELERY_K8S_EXECUTOR]
 
 ALLOWED_KIND_OPERATIONS = ["start", "stop", "restart", "status", "deploy", "test", "shell", "k9s"]
 ALLOWED_CONSTRAINTS_MODES_CI = ["constraints-source-providers", "constraints", "constraints-no-providers"]
@@ -444,7 +451,7 @@ def get_airflow_extras():
 ENABLED_SYSTEMS = ""
 
 CURRENT_KUBERNETES_VERSIONS = ALLOWED_KUBERNETES_VERSIONS
-CURRENT_EXECUTORS = ["KubernetesExecutor"]
+CURRENT_EXECUTORS = [KUBERNETES_EXECUTOR]
 
 DEFAULT_KUBERNETES_VERSION = CURRENT_KUBERNETES_VERSIONS[0]
 DEFAULT_EXECUTOR = CURRENT_EXECUTORS[0]

dev/breeze/src/airflow_breeze/params/shell_params.py

@@ -36,10 +36,12 @@
     ALLOWED_PYTHON_MAJOR_MINOR_VERSIONS,
     APACHE_AIRFLOW_GITHUB_REPOSITORY,
     CELERY_BROKER_URLS_MAP,
+    CELERY_EXECUTOR,
     DEFAULT_CELERY_BROKER,
     DEFAULT_UV_HTTP_TIMEOUT,
     DOCKER_DEFAULT_PLATFORM,
     DRILL_HOST_PORT,
+    EDGE_EXECUTOR,
     FLOWER_HOST_PORT,
     MOUNT_ALL,
     MOUNT_PROVIDERS_AND_TESTS,
@@ -323,7 +325,7 @@ def compose_file(self) -> str:
             for backend in ALLOWED_BACKENDS:
                 backend_files.extend(self.get_backend_compose_files(backend))
 
-        if self.executor == "CeleryExecutor":
+        if self.executor == CELERY_EXECUTOR:
             compose_file_list.append(DOCKER_COMPOSE_DIR / "integration-celery.yml")
             if self.use_airflow_version:
                 current_extras = self.airflow_extras
@@ -489,6 +491,9 @@ def env_variables_for_docker_commands(self) -> dict[str, str]:
         _set_var(_env, "AIRFLOW_VERSION", self.airflow_version)
         _set_var(_env, "AIRFLOW__CELERY__BROKER_URL", self.airflow_celery_broker_url)
         _set_var(_env, "AIRFLOW__CORE__EXECUTOR", self.executor)
+        if self.executor == EDGE_EXECUTOR:
+            _set_var(_env, "AIRFLOW__EDGE__API_ENABLED", "true")
+            _set_var(_env, "AIRFLOW__EDGE__API_URL", "http://localhost:8080/edge_worker/v1/rpcapi")
         _set_var(_env, "ANSWER", get_forced_answer() or "")
         _set_var(_env, "BACKEND", self.backend)
         _set_var(_env, "BASE_BRANCH", self.base_branch, "main")

scripts/in_container/bin/run_tmux

@@ -78,6 +78,23 @@ if [[ ${INTEGRATION_CELERY} == "true" && ${CELERY_FLOWER} == "true" ]]; then
     tmux split-window -h
     tmux send-keys 'airflow celery flower' C-m
 fi
+if [[ ${AIRFLOW__CORE__EXECUTOR,,} == "edgeexecutor" ]]; then
+    tmux select-pane -t 0
+    tmux split-window -h
+
+    # Ensure we are not leaking any DB connection information to Edge Worker process
+    tmux send-keys 'unset AIRFLOW__DATABASE__SQL_ALCHEMY_CONN' C-m
+    tmux send-keys 'unset AIRFLOW__CELERY__RESULT_BACKEND' C-m
+    tmux send-keys 'unset POSTGRES_HOST_PORT' C-m
+    tmux send-keys 'unset BACKEND' C-m
+    tmux send-keys 'unset POSTGRES_VERSION' C-m
+    tmux send-keys 'unset DATABASE_ISOLATION' C-m
+
+    # Ensure logs are smelling like Remote and are not visible to other components
+    tmux send-keys 'export AIRFLOW__LOGGING__BASE_LOG_FOLDER=edge_logs' C-m
+
+    tmux send-keys 'airflow edge worker --edge-hostname breeze' C-m
+fi
 if [[ ${STANDALONE_DAG_PROCESSOR} == "true" ]]; then
     tmux select-pane -t 3
     tmux split-window -h