This is an automated email from the ASF dual-hosted git repository.
vincbeck 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 9bcb6fd3ca5 Respect `waiter_delay` & `waiter_max_attempts` in EKS
delete operators (#74000)
9bcb6fd3ca5 is described below
commit 9bcb6fd3ca5a05d034c6d2ffca965f4ff620bd3d
Author: Justin Pakzad <[email protected]>
AuthorDate: Tue Oct 13 09:19:16 2026 -0400
Respect `waiter_delay` & `waiter_max_attempts` in EKS delete operators
(#74000)
---
.../airflow/providers/amazon/aws/operators/eks.py | 18 +++-
.../tests/unit/amazon/aws/operators/test_eks.py | 103 ++++++++++++++++++++-
2 files changed, 115 insertions(+), 6 deletions(-)
diff --git a/providers/amazon/src/airflow/providers/amazon/aws/operators/eks.py
b/providers/amazon/src/airflow/providers/amazon/aws/operators/eks.py
index db3e379fe6b..3eee3434b91 100644
--- a/providers/amazon/src/airflow/providers/amazon/aws/operators/eks.py
+++ b/providers/amazon/src/airflow/providers/amazon/aws/operators/eks.py
@@ -835,7 +835,10 @@ class EksDeleteClusterOperator(AwsBaseOperator[EksHook]):
if self.wait_for_completion:
self.log.info("Waiting for cluster to delete. This will take some
time.")
-
self.hook.conn.get_waiter("cluster_deleted").wait(name=self.cluster_name)
+ self.hook.conn.get_waiter("cluster_deleted").wait(
+ name=self.cluster_name,
+ WaiterConfig={"Delay": self.waiter_delay, "MaxAttempts":
self.waiter_max_attempts},
+ )
def delete_any_nodegroups(self) -> None:
"""
@@ -854,7 +857,10 @@ class EksDeleteClusterOperator(AwsBaseOperator[EksHook]):
self.hook.delete_nodegroup(clusterName=self.cluster_name, nodegroupName=group)
# Note this is a custom waiter so we're using hook.get_waiter(),
not hook.conn.get_waiter().
self.log.info("Waiting for all nodegroups to delete. This will
take some time.")
-
self.hook.get_waiter("all_nodegroups_deleted").wait(clusterName=self.cluster_name)
+ self.hook.get_waiter("all_nodegroups_deleted").wait(
+ clusterName=self.cluster_name,
+ WaiterConfig={"Delay": self.waiter_delay, "MaxAttempts":
self.waiter_max_attempts},
+ )
self.log.info(SUCCESS_MSG.format(compute=NODEGROUP_FULL_NAME))
def delete_any_fargate_profiles(self) -> None:
@@ -879,7 +885,9 @@ class EksDeleteClusterOperator(AwsBaseOperator[EksHook]):
clusterName=self.cluster_name,
fargateProfileName=profile
)
self.hook.conn.get_waiter("fargate_profile_deleted").wait(
- clusterName=self.cluster_name, fargateProfileName=profile
+ clusterName=self.cluster_name,
+ fargateProfileName=profile,
+ WaiterConfig={"Delay": self.waiter_delay, "MaxAttempts":
self.waiter_max_attempts},
)
self.log.info(SUCCESS_MSG.format(compute=FARGATE_FULL_NAME))
@@ -976,7 +984,9 @@ class EksDeleteNodegroupOperator(AwsBaseOperator[EksHook]):
elif self.wait_for_completion:
self.log.info("Waiting for nodegroup to delete. This will take
some time.")
self.hook.conn.get_waiter("nodegroup_deleted").wait(
- clusterName=self.cluster_name,
nodegroupName=self.nodegroup_name
+ clusterName=self.cluster_name,
+ nodegroupName=self.nodegroup_name,
+ WaiterConfig={"Delay": self.waiter_delay, "MaxAttempts":
self.waiter_max_attempts},
)
def execute_complete(self, context: Context, event: dict[str, Any] | None
= None) -> None:
diff --git a/providers/amazon/tests/unit/amazon/aws/operators/test_eks.py
b/providers/amazon/tests/unit/amazon/aws/operators/test_eks.py
index f95be60d51d..ebd756518b9 100644
--- a/providers/amazon/tests/unit/amazon/aws/operators/test_eks.py
+++ b/providers/amazon/tests/unit/amazon/aws/operators/test_eks.py
@@ -752,7 +752,11 @@ class TestEksDeleteClusterOperator:
mock_list_nodegroups.assert_not_called()
mock_delete_cluster.assert_called_once_with(name=self.cluster_name)
- mock_waiter.assert_called_with(mock.ANY, name=CLUSTER_NAME)
+ mock_waiter.assert_called_with(
+ mock.ANY,
+ name=CLUSTER_NAME,
+ WaiterConfig={"Delay": 30, "MaxAttempts": 40},
+ )
assert_expected_waiter_type(mock_waiter, "ClusterDeleted")
def test_eks_delete_cluster_operator_with_deferrable(self):
@@ -782,6 +786,72 @@ class TestEksDeleteClusterOperator:
assert mock_delete_cluster.call_count == 3
mock_delete_cluster.assert_called_with(name=self.cluster_name)
+ @mock.patch.object(Waiter, "wait")
+ @mock.patch.object(EksHook, "delete_cluster")
+ def test_delete_cluster_with_waiter_config(self, mock_delete_cluster,
mock_waiter):
+ op = EksDeleteClusterOperator(
+ task_id=TASK_ID,
+ cluster_name=CLUSTER_NAME,
+ wait_for_completion=True,
+ waiter_delay=15,
+ waiter_max_attempts=10,
+ )
+ op.execute({})
+ mock_delete_cluster.assert_called_once_with(name=CLUSTER_NAME)
+ mock_waiter.assert_called_with(
+ mock.ANY,
+ name=CLUSTER_NAME,
+ WaiterConfig={"Delay": 15, "MaxAttempts": 10},
+ )
+ assert_expected_waiter_type(mock_waiter, "ClusterDeleted")
+
+ @mock.patch.object(Waiter, "wait")
+ @mock.patch.object(EksHook, "get_waiter")
+ @mock.patch.object(EksHook, "list_fargate_profiles")
+ @mock.patch.object(EksHook, "list_nodegroups")
+ @mock.patch.object(EksHook, "delete_fargate_profile")
+ @mock.patch.object(EksHook, "delete_nodegroup")
+ @mock.patch.object(EksHook, "delete_cluster")
+ def test_force_delete_compute_with_waiter_config(
+ self,
+ mock_delete_cluster,
+ mock_delete_nodegroup,
+ mock_delete_fargate_profile,
+ mock_list_nodegroups,
+ mock_list_fargate_profiles,
+ mock_get_waiter,
+ mock_waiter,
+ ):
+ mock_list_nodegroups.return_value = [NODEGROUP_NAME]
+ mock_list_fargate_profiles.return_value = [FARGATE_PROFILE_NAME]
+ expected_config = {"Delay": 15, "MaxAttempts": 10}
+
+ op = EksDeleteClusterOperator(
+ task_id=TASK_ID,
+ cluster_name=CLUSTER_NAME,
+ force_delete_compute=True,
+ wait_for_completion=True,
+ waiter_delay=15,
+ waiter_max_attempts=10,
+ )
+
+ op.execute({})
+
+ mock_get_waiter.assert_called_once_with("all_nodegroups_deleted")
+ mock_get_waiter.return_value.wait.assert_called_once_with(
+ clusterName=CLUSTER_NAME,
+ WaiterConfig=expected_config,
+ )
+ assert mock_waiter.call_args_list == [
+ mock.call(
+ mock.ANY,
+ clusterName=CLUSTER_NAME,
+ fargateProfileName=FARGATE_PROFILE_NAME,
+ WaiterConfig=expected_config,
+ ),
+ mock.call(mock.ANY, name=CLUSTER_NAME,
WaiterConfig=expected_config),
+ ]
+
@mock.patch("time.sleep", return_value=None)
@mock.patch.object(EksHook, "get_waiter")
@mock.patch.object(EksHook, "list_nodegroups")
@@ -854,7 +924,12 @@ class TestEksDeleteNodegroupOperator:
mock_delete_nodegroup.assert_called_once_with(
clusterName=self.cluster_name, nodegroupName=self.nodegroup_name
)
- mock_waiter.assert_called_with(mock.ANY, clusterName=CLUSTER_NAME,
nodegroupName=NODEGROUP_NAME)
+ mock_waiter.assert_called_with(
+ mock.ANY,
+ clusterName=CLUSTER_NAME,
+ nodegroupName=NODEGROUP_NAME,
+ WaiterConfig={"Delay": 30, "MaxAttempts": 40},
+ )
assert_expected_waiter_type(mock_waiter, "NodegroupDeleted")
@mock.patch("time.sleep", return_value=None)
@@ -872,6 +947,30 @@ class TestEksDeleteNodegroupOperator:
clusterName=self.cluster_name, nodegroupName=self.nodegroup_name
)
+ @mock.patch.object(Waiter, "wait")
+ @mock.patch.object(EksHook, "delete_nodegroup")
+ def test_delete_nodegroup_with_waiter_config(self, mock_delete_nodegroup,
mock_waiter):
+ op = EksDeleteNodegroupOperator(
+ task_id=TASK_ID,
+ cluster_name=CLUSTER_NAME,
+ nodegroup_name=NODEGROUP_NAME,
+ wait_for_completion=True,
+ waiter_delay=15,
+ waiter_max_attempts=10,
+ )
+ op.execute({})
+ mock_delete_nodegroup.assert_called_once_with(
+ clusterName=CLUSTER_NAME,
+ nodegroupName=NODEGROUP_NAME,
+ )
+ mock_waiter.assert_called_with(
+ mock.ANY,
+ clusterName=CLUSTER_NAME,
+ nodegroupName=NODEGROUP_NAME,
+ WaiterConfig={"Delay": 15, "MaxAttempts": 10},
+ )
+ assert_expected_waiter_type(mock_waiter, "NodegroupDeleted")
+
def test_template_fields(self):
validate_template_fields(self.delete_nodegroup_operator)
