This is an automated email from the ASF dual-hosted git repository.
henry3260 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 4bc581aab21 Allow GKE Autopilot clusters without node pools and stop
GKE pod trigger reviving should_delete_pod (#69876)
4bc581aab21 is described below
commit 4bc581aab21b60cea8dfd34cdac2a33af0b394d0
Author: Vic Wen <[email protected]>
AuthorDate: Sun Oct 18 12:20:35 2026 +0800
Allow GKE Autopilot clusters without node pools and stop GKE pod trigger
reviving should_delete_pod (#69876)
---
.../google/cloud/operators/kubernetes_engine.py | 10 +++++-
.../google/cloud/triggers/kubernetes_engine.py | 2 +-
.../kubernetes_engine/example_kubernetes_engine.py | 2 +-
.../example_kubernetes_engine_async.py | 5 ++-
.../example_kubernetes_engine_job.py | 5 ++-
.../example_kubernetes_engine_kueue.py | 2 +-
.../example_kubernetes_engine_ray.py | 6 ----
.../example_kubernetes_engine_resource.py | 5 ++-
.../cloud/operators/test_kubernetes_engine.py | 37 +++++++++++++++++++++-
.../cloud/triggers/test_kubernetes_engine.py | 19 +++++++++--
10 files changed, 77 insertions(+), 16 deletions(-)
diff --git
a/providers/google/src/airflow/providers/google/cloud/operators/kubernetes_engine.py
b/providers/google/src/airflow/providers/google/cloud/operators/kubernetes_engine.py
index 26842379936..506558b1c60 100644
---
a/providers/google/src/airflow/providers/google/cloud/operators/kubernetes_engine.py
+++
b/providers/google/src/airflow/providers/google/cloud/operators/kubernetes_engine.py
@@ -466,7 +466,9 @@ class GKECreateClusterOperator(GKEOperatorMixin,
GoogleCloudBaseOperator):
"body['node_config'] or body['initial_code_count'] are
specified."
)
- if not any([self._body_field("node_config"),
self._body_field("initial_node_count")]):
+ if not self._is_autopilot_cluster() and not any(
+ [self._body_field("node_config"),
self._body_field("initial_node_count")]
+ ):
if not self._body_field("node_pools"):
error_messages.append(
"Field body['node_pools'] is required if none of fields "
@@ -485,6 +487,12 @@ class GKECreateClusterOperator(GKEOperatorMixin,
GoogleCloudBaseOperator):
return self.body.get(field_name, default_value)
return getattr(self.body, field_name, default_value)
+ def _is_autopilot_cluster(self) -> bool:
+ autopilot = self._body_field("autopilot")
+ if isinstance(autopilot, dict):
+ return bool(autopilot.get("enabled"))
+ return bool(getattr(autopilot, "enabled", False))
+
def _alert_deprecated_body_fields(self) -> None:
"""Generate warning messages if deprecated fields were used in the
body."""
deprecated_body_fields_with_replacement = [
diff --git
a/providers/google/src/airflow/providers/google/cloud/triggers/kubernetes_engine.py
b/providers/google/src/airflow/providers/google/cloud/triggers/kubernetes_engine.py
index 27d5e6f6823..833a4cfe4a2 100644
---
a/providers/google/src/airflow/providers/google/cloud/triggers/kubernetes_engine.py
+++
b/providers/google/src/airflow/providers/google/cloud/triggers/kubernetes_engine.py
@@ -151,7 +151,7 @@ class GKEStartPodTrigger(KubernetesPodTrigger):
"schedule_timeout": self.schedule_timeout,
"trigger_start_time": self.trigger_start_time,
"base_container_name": self.base_container_name,
- "should_delete_pod": self.should_delete_pod,
+ "should_delete_pod": None,
"on_finish_action": self.on_finish_action.value,
"gcp_conn_id": self.gcp_conn_id,
"impersonation_chain": self.impersonation_chain,
diff --git
a/providers/google/tests/system/google/cloud/kubernetes_engine/example_kubernetes_engine.py
b/providers/google/tests/system/google/cloud/kubernetes_engine/example_kubernetes_engine.py
index 77b04def7ad..12c0ddde94d 100644
---
a/providers/google/tests/system/google/cloud/kubernetes_engine/example_kubernetes_engine.py
+++
b/providers/google/tests/system/google/cloud/kubernetes_engine/example_kubernetes_engine.py
@@ -64,7 +64,7 @@ EXEC_CONTAINER_NAME = "main"
EXPECTED_EXEC_OUTPUT = "command executed in existing GKE Pod"
# [START howto_operator_gcp_gke_create_cluster_definition]
-CLUSTER = {"name": CLUSTER_NAME, "initial_node_count": 1, "autopilot":
{"enabled": True}}
+CLUSTER = {"name": CLUSTER_NAME, "autopilot": {"enabled": True}}
# [END howto_operator_gcp_gke_create_cluster_definition]
EXEC_POD = f"""
diff --git
a/providers/google/tests/system/google/cloud/kubernetes_engine/example_kubernetes_engine_async.py
b/providers/google/tests/system/google/cloud/kubernetes_engine/example_kubernetes_engine_async.py
index b82ecd97fc0..12f61ea5fef 100644
---
a/providers/google/tests/system/google/cloud/kubernetes_engine/example_kubernetes_engine_async.py
+++
b/providers/google/tests/system/google/cloud/kubernetes_engine/example_kubernetes_engine_async.py
@@ -51,7 +51,10 @@ CLUSTER_NAME_BASE = f"cluster-{DAG_ID}".replace("_", "-")
CLUSTER_NAME_FULL = CLUSTER_NAME_BASE + f"-{ENV_ID}".replace("_", "-")
CLUSTER_NAME = CLUSTER_NAME_BASE if len(CLUSTER_NAME_FULL) >= 33 else
CLUSTER_NAME_FULL
-CLUSTER = {"name": CLUSTER_NAME, "initial_node_count": 1}
+CLUSTER = {
+ "name": CLUSTER_NAME,
+ "node_pools": [{"name": "default-pool", "initial_node_count": 1}],
+}
with DAG(
DAG_ID,
diff --git
a/providers/google/tests/system/google/cloud/kubernetes_engine/example_kubernetes_engine_job.py
b/providers/google/tests/system/google/cloud/kubernetes_engine/example_kubernetes_engine_job.py
index cfb56b5da6e..112fba32428 100644
---
a/providers/google/tests/system/google/cloud/kubernetes_engine/example_kubernetes_engine_job.py
+++
b/providers/google/tests/system/google/cloud/kubernetes_engine/example_kubernetes_engine_job.py
@@ -49,7 +49,10 @@ GCP_PROJECT_ID = os.environ.get("SYSTEM_TESTS_GCP_PROJECT",
"default")
GCP_LOCATION = "europe-north1-a"
CLUSTER_NAME = f"gke-job-{ENV_ID}".replace("_", "-")
-CLUSTER = {"name": CLUSTER_NAME, "initial_node_count": 1}
+CLUSTER = {
+ "name": CLUSTER_NAME,
+ "node_pools": [{"name": "default-pool", "initial_node_count": 1}],
+}
JOB_NAME = "test-pi"
JOB_NAME_DEF = "test-pi-def"
diff --git
a/providers/google/tests/system/google/cloud/kubernetes_engine/example_kubernetes_engine_kueue.py
b/providers/google/tests/system/google/cloud/kubernetes_engine/example_kubernetes_engine_kueue.py
index c48ae19cea3..60127bbd9ab 100644
---
a/providers/google/tests/system/google/cloud/kubernetes_engine/example_kubernetes_engine_kueue.py
+++
b/providers/google/tests/system/google/cloud/kubernetes_engine/example_kubernetes_engine_kueue.py
@@ -47,7 +47,7 @@ GCP_PROJECT_ID = os.environ.get("SYSTEM_TESTS_GCP_PROJECT",
"default")
GCP_LOCATION = "europe-west3"
CLUSTER_NAME = f"gke-kueue-{ENV_ID}".replace("_", "-")
-CLUSTER = {"name": CLUSTER_NAME, "initial_node_count": 1, "autopilot":
{"enabled": True}}
+CLUSTER = {"name": CLUSTER_NAME, "autopilot": {"enabled": True}}
flavor_conf = """
apiVersion: kueue.x-k8s.io/v1beta1
diff --git
a/providers/google/tests/system/google/cloud/kubernetes_engine/example_kubernetes_engine_ray.py
b/providers/google/tests/system/google/cloud/kubernetes_engine/example_kubernetes_engine_ray.py
index 77bb7798f60..c1ff25554a7 100644
---
a/providers/google/tests/system/google/cloud/kubernetes_engine/example_kubernetes_engine_ray.py
+++
b/providers/google/tests/system/google/cloud/kubernetes_engine/example_kubernetes_engine_ray.py
@@ -50,12 +50,6 @@ CLUSTER_NAME = CLUSTER_NAME_BASE if len(CLUSTER_NAME_FULL)
>= 33 else CLUSTER_NA
# [START howto_operator_gcp_gke_create_cluster_definition_with_ray]
CLUSTER = {
"name": CLUSTER_NAME,
- "node_pools": [
- {
- "name": f"{CLUSTER_NAME}-node",
- "initial_node_count": 1,
- },
- ],
"autopilot": {"enabled": True},
"addons_config": {
"ray_operator_config": {"enabled": True},
diff --git
a/providers/google/tests/system/google/cloud/kubernetes_engine/example_kubernetes_engine_resource.py
b/providers/google/tests/system/google/cloud/kubernetes_engine/example_kubernetes_engine_resource.py
index f9dae65f1cb..81666fc55ef 100644
---
a/providers/google/tests/system/google/cloud/kubernetes_engine/example_kubernetes_engine_resource.py
+++
b/providers/google/tests/system/google/cloud/kubernetes_engine/example_kubernetes_engine_resource.py
@@ -44,7 +44,10 @@ GCP_PROJECT_ID = os.environ.get("SYSTEM_TESTS_GCP_PROJECT",
"default")
GCP_LOCATION = "europe-north1-a"
CLUSTER_NAME = f"gke-resource-{ENV_ID}".replace("_", "-")
-CLUSTER = {"name": CLUSTER_NAME, "initial_node_count": 1}
+CLUSTER = {
+ "name": CLUSTER_NAME,
+ "node_pools": [{"name": "default-pool", "initial_node_count": 1}],
+}
PVC_NAME = "test-pvc-name"
PVC_CONF = f"""
diff --git
a/providers/google/tests/unit/google/cloud/operators/test_kubernetes_engine.py
b/providers/google/tests/unit/google/cloud/operators/test_kubernetes_engine.py
index f820e493bfb..eb06480c971 100644
---
a/providers/google/tests/unit/google/cloud/operators/test_kubernetes_engine.py
+++
b/providers/google/tests/unit/google/cloud/operators/test_kubernetes_engine.py
@@ -106,6 +106,11 @@ GKE_CLUSTER_CREATE_BODY_DICT = {
GKE_CLUSTER_CREATE_BODY_OBJECT = Cluster(
name=GKE_CLUSTER_NAME, node_pools=[NodePool(name="a_node_pool",
initial_node_count=1)]
)
+GKE_AUTOPILOT_CLUSTER_CREATE_BODY_DICT = {
+ "name": GKE_CLUSTER_NAME,
+ "autopilot": {"enabled": True},
+}
+GKE_AUTOPILOT_CLUSTER_CREATE_BODY_OBJECT = Cluster(name=GKE_CLUSTER_NAME,
autopilot={"enabled": True})
GKE_CLUSTER_CREATE_BODY_DICT_DEPRECATED = {"name": GKE_CLUSTER_NAME,
"initial_node_count": 1}
GKE_CLUSTER_CREATE_BODY_OBJECT_DEPRECATED = Cluster(name=GKE_CLUSTER_NAME,
initial_node_count=1)
@@ -423,7 +428,15 @@ class TestGKECreateClusterOperator:
)
assert set(GKECreateClusterOperator.template_fields) ==
expected_template_fields
- @pytest.mark.parametrize("body", [GKE_CLUSTER_CREATE_BODY_DICT,
GKE_CLUSTER_CREATE_BODY_OBJECT])
+ @pytest.mark.parametrize(
+ "body",
+ [
+ GKE_CLUSTER_CREATE_BODY_DICT,
+ GKE_CLUSTER_CREATE_BODY_OBJECT,
+ GKE_AUTOPILOT_CLUSTER_CREATE_BODY_DICT,
+ GKE_AUTOPILOT_CLUSTER_CREATE_BODY_OBJECT,
+ ],
+ )
def test_body(self, body):
op = GKECreateClusterOperator(
task_id=TEST_TASK_ID,
@@ -435,6 +448,28 @@ class TestGKECreateClusterOperator:
)
assert op.cluster_name == GKE_CLUSTER_NAME
+ @pytest.mark.parametrize(
+ "body",
+ [
+ pytest.param(
+ {"name": GKE_CLUSTER_NAME, "autopilot": {"enabled": False}},
+ id="dict",
+ ),
+ pytest.param(
+ Cluster(name=GKE_CLUSTER_NAME, autopilot={"enabled": False}),
+ id="cluster",
+ ),
+ ],
+ )
+ def test_body_with_autopilot_disabled_without_node_pool_raises(self, body):
+ with pytest.raises(AirflowException):
+ GKECreateClusterOperator(
+ project_id=TEST_PROJECT_ID,
+ location=TEST_LOCATION,
+ body=body,
+ task_id=TEST_TASK_ID,
+ )
+
@pytest.mark.parametrize(
"body",
[
diff --git
a/providers/google/tests/unit/google/cloud/triggers/test_kubernetes_engine.py
b/providers/google/tests/unit/google/cloud/triggers/test_kubernetes_engine.py
index 05ebb8e30a8..a577dccf0db 100644
---
a/providers/google/tests/unit/google/cloud/triggers/test_kubernetes_engine.py
+++
b/providers/google/tests/unit/google/cloud/triggers/test_kubernetes_engine.py
@@ -52,7 +52,6 @@ NAMESPACE = "default"
POLL_INTERVAL = 2
CLUSTER_CONTEXT = "test-context"
IN_CLUSTER = False
-SHOULD_DELETE_POD = True
GET_LOGS = True
STARTUP_TIMEOUT_SECS = 120
SCHEDULE_TIMEOUT_SECS = 60
@@ -139,7 +138,7 @@ class TestGKEStartPodTrigger:
"ssl_ca_cert": SSL_CA_CERT,
"base_container_name": BASE_CONTAINER_NAME,
"on_finish_action": ON_FINISH_ACTION,
- "should_delete_pod": SHOULD_DELETE_POD,
+ "should_delete_pod": None,
"gcp_conn_id": GCP_CONN_ID,
"impersonation_chain": IMPERSONATION_CHAIN,
"last_log_time": None,
@@ -147,6 +146,22 @@ class TestGKEStartPodTrigger:
"use_dns_endpoint": False,
}
+ def test_serialize_round_trip_preserves_on_finish_action(self):
+ trigger = GKEStartPodTrigger(
+ pod_name=POD_NAME,
+ pod_namespace=NAMESPACE,
+ trigger_start_time=TRIGGER_START_TIME,
+ cluster_url=CLUSTER_URL,
+ ssl_ca_cert=SSL_CA_CERT,
+ base_container_name=BASE_CONTAINER_NAME,
+ on_finish_action="delete_succeeded_pod",
+ )
+
+ _, kwargs_dict = trigger.serialize()
+ reconstructed_trigger = GKEStartPodTrigger(**kwargs_dict)
+
+ assert reconstructed_trigger.on_finish_action.value ==
"delete_succeeded_pod"
+
@pytest.mark.asyncio
@mock.patch(f"{TRIGGER_KUB_POD_PATH}._wait_for_pod_start")
@mock.patch(f"{TRIGGER_GKE_POD_PATH}.hook")
