You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "syedahsn (via GitHub)" <gi...@apache.org> on 2023/06/01 00:01:19 UTC
[GitHub] [airflow] syedahsn opened a new pull request, #31657: Deferrable mode for EksCreateFargateProfileOperator and EksDeleteFargateProfileOperator
syedahsn opened a new pull request, #31657:
URL: https://github.com/apache/airflow/pull/31657
This PR allows deferrable mode to be used with the `EksCreateFargateProfileOperator` and the `EksDeleteFargateProfileOperator`. It also includes unit tests to test the triggers for both of the operators.
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of an existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#pull-request-guidelines)** for more information.
In case of fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in a newsfragment file, named `{pr_number}.significant.rst` or `{issue_number}.significant.rst`, in [newsfragments](https://github.com/apache/airflow/tree/main/newsfragments).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] o-nikolas merged pull request #31657: Deferrable mode for EksCreateFargateProfileOperator and EksDeleteFargateProfileOperator
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas merged PR #31657:
URL: https://github.com/apache/airflow/pull/31657
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] vandonr-amz commented on a diff in pull request #31657: Deferrable mode for EksCreateFargateProfileOperator and EksDeleteFargateProfileOperator
Posted by "vandonr-amz (via GitHub)" <gi...@apache.org>.
vandonr-amz commented on code in PR #31657:
URL: https://github.com/apache/airflow/pull/31657#discussion_r1221944662
##########
airflow/providers/amazon/aws/operators/eks.py:
##########
@@ -401,13 +417,31 @@ def execute(self, context: Context):
selectors=self.selectors,
**self.create_fargate_profile_kwargs,
)
-
- if self.wait_for_completion:
+ if self.deferrable:
+ self.defer(
+ trigger=EksCreateFargateProfileTrigger(
+ cluster_name=self.cluster_name,
+ fargate_profile_name=self.fargate_profile_name,
+ aws_conn_id=self.aws_conn_id,
+ poll_interval=self.waiter_delay,
+ max_attempts=self.waiter_max_attempts,
+ ),
+ method_name="execute_complete",
+ timeout=timedelta(seconds=(self.waiter_max_attempts * self.waiter_delay + 60)),
Review Comment:
why 60 ? Maybe create a named constant ?
##########
airflow/providers/amazon/aws/triggers/eks.py:
##########
@@ -0,0 +1,178 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import asyncio
+from functools import cached_property
+from typing import Any
+
+from botocore.exceptions import WaiterError
+
+from airflow.providers.amazon.aws.hooks.eks import EksHook
+from airflow.triggers.base import BaseTrigger, TriggerEvent
+
+
+class EksCreateFargateProfileTrigger(BaseTrigger):
+ """
+ Trigger for EksCreateFargateProfileOperator.
+ The trigger will asynchronously wait for the fargate profile to be created.
+
+ :param cluster_name: The name of the EKS cluster
+ :param fargate_profile_name: The name of the fargate profile
+ :param poll_interval: The amount of time in seconds to wait between attempts.
+ :param max_attempts: The maximum number of attempts to be made.
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
+ """
+
+ def __init__(
+ self,
+ cluster_name: str,
+ fargate_profile_name: str,
+ poll_interval: int,
+ max_attempts: int,
+ aws_conn_id: str,
+ ):
+ self.cluster_name = cluster_name
+ self.fargate_profile_name = fargate_profile_name
+ self.poll_interval = poll_interval
+ self.max_attempts = max_attempts
+ self.aws_conn_id = aws_conn_id
+
+ def serialize(self) -> tuple[str, dict[str, Any]]:
+ return (
+ self.__class__.__module__ + "." + self.__class__.__qualname__,
+ {
+ "cluster_name": self.cluster_name,
+ "fargate_profile_name": self.fargate_profile_name,
+ "poll_interval": str(self.poll_interval),
+ "max_attempts": str(self.max_attempts),
+ "aws_conn_id": self.aws_conn_id,
+ },
+ )
+
+ @cached_property
+ def hook(self) -> EksHook:
+ return EksHook(aws_conn_id=self.aws_conn_id)
Review Comment:
mentioned it earlier, but I think caching this is useless
##########
airflow/providers/amazon/aws/operators/eks.py:
##########
@@ -622,12 +667,31 @@ def execute(self, context: Context):
eks_hook.delete_fargate_profile(
clusterName=self.cluster_name, fargateProfileName=self.fargate_profile_name
)
- if self.wait_for_completion:
Review Comment:
same comments as above to this code
##########
airflow/providers/amazon/aws/operators/eks.py:
##########
@@ -401,13 +417,31 @@ def execute(self, context: Context):
selectors=self.selectors,
**self.create_fargate_profile_kwargs,
)
-
- if self.wait_for_completion:
+ if self.deferrable:
+ self.defer(
+ trigger=EksCreateFargateProfileTrigger(
+ cluster_name=self.cluster_name,
+ fargate_profile_name=self.fargate_profile_name,
+ aws_conn_id=self.aws_conn_id,
+ poll_interval=self.waiter_delay,
+ max_attempts=self.waiter_max_attempts,
+ ),
+ method_name="execute_complete",
+ timeout=timedelta(seconds=(self.waiter_max_attempts * self.waiter_delay + 60)),
+ )
+ elif self.wait_for_completion:
self.log.info("Waiting for Fargate profile to provision. This will take some time.")
eks_hook.conn.get_waiter("fargate_profile_active").wait(
clusterName=self.cluster_name, fargateProfileName=self.fargate_profile_name
)
Review Comment:
shall we pass the delay & max attempts to this waiter as well while we're at it ?
##########
airflow/providers/amazon/aws/triggers/eks.py:
##########
@@ -0,0 +1,178 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import asyncio
+from functools import cached_property
+from typing import Any
+
+from botocore.exceptions import WaiterError
+
+from airflow.providers.amazon.aws.hooks.eks import EksHook
+from airflow.triggers.base import BaseTrigger, TriggerEvent
+
+
+class EksCreateFargateProfileTrigger(BaseTrigger):
+ """
+ Trigger for EksCreateFargateProfileOperator.
+ The trigger will asynchronously wait for the fargate profile to be created.
+
+ :param cluster_name: The name of the EKS cluster
+ :param fargate_profile_name: The name of the fargate profile
+ :param poll_interval: The amount of time in seconds to wait between attempts.
+ :param max_attempts: The maximum number of attempts to be made.
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
+ """
+
+ def __init__(
+ self,
+ cluster_name: str,
+ fargate_profile_name: str,
+ poll_interval: int,
+ max_attempts: int,
+ aws_conn_id: str,
+ ):
+ self.cluster_name = cluster_name
+ self.fargate_profile_name = fargate_profile_name
+ self.poll_interval = poll_interval
+ self.max_attempts = max_attempts
+ self.aws_conn_id = aws_conn_id
+
+ def serialize(self) -> tuple[str, dict[str, Any]]:
+ return (
+ self.__class__.__module__ + "." + self.__class__.__qualname__,
+ {
+ "cluster_name": self.cluster_name,
+ "fargate_profile_name": self.fargate_profile_name,
+ "poll_interval": str(self.poll_interval),
+ "max_attempts": str(self.max_attempts),
+ "aws_conn_id": self.aws_conn_id,
+ },
+ )
+
+ @cached_property
+ def hook(self) -> EksHook:
+ return EksHook(aws_conn_id=self.aws_conn_id)
+
+ async def run(self):
+ async with self.hook.async_conn as client:
+ attempt = 0
+ waiter = client.get_waiter("fargate_profile_active")
+ while attempt < int(self.max_attempts):
+ attempt += 1
+ try:
+ await waiter.wait(
+ clusterName=self.cluster_name,
+ fargateProfileName=self.fargate_profile_name,
+ WaiterConfig={"Delay": int(self.poll_interval), "MaxAttempts": 1},
+ )
+ break
+ except WaiterError as error:
+ if "terminal failure" in str(error):
+ yield TriggerEvent(
+ {"status": "failure", "message": f"Create Fargate Profile failed: {error}"}
+ )
Review Comment:
you should raise directly here instead of returning the error to the operator
see https://apache-airflow.slack.com/archives/CCPRP7943/p1685663968275419
##########
airflow/providers/amazon/aws/triggers/eks.py:
##########
@@ -0,0 +1,178 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import asyncio
+from functools import cached_property
+from typing import Any
+
+from botocore.exceptions import WaiterError
+
+from airflow.providers.amazon.aws.hooks.eks import EksHook
+from airflow.triggers.base import BaseTrigger, TriggerEvent
+
+
+class EksCreateFargateProfileTrigger(BaseTrigger):
+ """
+ Trigger for EksCreateFargateProfileOperator.
+ The trigger will asynchronously wait for the fargate profile to be created.
+
+ :param cluster_name: The name of the EKS cluster
+ :param fargate_profile_name: The name of the fargate profile
+ :param poll_interval: The amount of time in seconds to wait between attempts.
+ :param max_attempts: The maximum number of attempts to be made.
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
+ """
+
+ def __init__(
+ self,
+ cluster_name: str,
+ fargate_profile_name: str,
+ poll_interval: int,
+ max_attempts: int,
+ aws_conn_id: str,
+ ):
+ self.cluster_name = cluster_name
+ self.fargate_profile_name = fargate_profile_name
+ self.poll_interval = poll_interval
+ self.max_attempts = max_attempts
+ self.aws_conn_id = aws_conn_id
+
+ def serialize(self) -> tuple[str, dict[str, Any]]:
+ return (
+ self.__class__.__module__ + "." + self.__class__.__qualname__,
+ {
+ "cluster_name": self.cluster_name,
+ "fargate_profile_name": self.fargate_profile_name,
+ "poll_interval": str(self.poll_interval),
+ "max_attempts": str(self.max_attempts),
+ "aws_conn_id": self.aws_conn_id,
+ },
+ )
+
+ @cached_property
+ def hook(self) -> EksHook:
+ return EksHook(aws_conn_id=self.aws_conn_id)
+
+ async def run(self):
+ async with self.hook.async_conn as client:
+ attempt = 0
+ waiter = client.get_waiter("fargate_profile_active")
+ while attempt < int(self.max_attempts):
+ attempt += 1
+ try:
+ await waiter.wait(
+ clusterName=self.cluster_name,
+ fargateProfileName=self.fargate_profile_name,
+ WaiterConfig={"Delay": int(self.poll_interval), "MaxAttempts": 1},
+ )
+ break
+ except WaiterError as error:
+ if "terminal failure" in str(error):
+ yield TriggerEvent(
+ {"status": "failure", "message": f"Create Fargate Profile failed: {error}"}
+ )
+ break
+ self.log.info(
+ "Status of fargate profile is %s", error.last_response["fargateProfile"]["status"]
+ )
+ await asyncio.sleep(int(self.poll_interval))
+ if attempt >= int(self.max_attempts):
+ yield TriggerEvent(
+ {
+ "status": "failure",
+ "message": "Create Fargate profile Failed - max attempts reached.",
+ }
+ )
+ else:
+ yield TriggerEvent({"status": "success", "message": "Fargate Profile Created"})
+
+
+class EksDeleteFargateProfileTrigger(BaseTrigger):
Review Comment:
same comments as above
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] syedahsn commented on a diff in pull request #31657: Deferrable mode for EksCreateFargateProfileOperator and EksDeleteFargateProfileOperator
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #31657:
URL: https://github.com/apache/airflow/pull/31657#discussion_r1228706671
##########
airflow/providers/amazon/aws/triggers/eks.py:
##########
@@ -0,0 +1,178 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import asyncio
+from functools import cached_property
+from typing import Any
+
+from botocore.exceptions import WaiterError
+
+from airflow.providers.amazon.aws.hooks.eks import EksHook
+from airflow.triggers.base import BaseTrigger, TriggerEvent
+
+
+class EksCreateFargateProfileTrigger(BaseTrigger):
+ """
+ Trigger for EksCreateFargateProfileOperator.
+ The trigger will asynchronously wait for the fargate profile to be created.
+
+ :param cluster_name: The name of the EKS cluster
+ :param fargate_profile_name: The name of the fargate profile
+ :param poll_interval: The amount of time in seconds to wait between attempts.
+ :param max_attempts: The maximum number of attempts to be made.
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
+ """
+
+ def __init__(
+ self,
+ cluster_name: str,
+ fargate_profile_name: str,
+ poll_interval: int,
+ max_attempts: int,
+ aws_conn_id: str,
+ ):
+ self.cluster_name = cluster_name
+ self.fargate_profile_name = fargate_profile_name
+ self.poll_interval = poll_interval
+ self.max_attempts = max_attempts
+ self.aws_conn_id = aws_conn_id
+
+ def serialize(self) -> tuple[str, dict[str, Any]]:
+ return (
+ self.__class__.__module__ + "." + self.__class__.__qualname__,
Review Comment:
This is an improvement to what we were initially doing, which was a hard coded string with the trigger path. As for making it a separate function, I'm open to it if others feel it necessary, but considering its a one line function, I don't feel like its worth it. But I'm not opposed to it either.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] syedahsn commented on a diff in pull request #31657: Deferrable mode for EksCreateFargateProfileOperator and EksDeleteFargateProfileOperator
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #31657:
URL: https://github.com/apache/airflow/pull/31657#discussion_r1228691009
##########
airflow/providers/amazon/aws/operators/eks.py:
##########
@@ -371,11 +381,14 @@ def __init__(
cluster_name: str,
pod_execution_role_arn: str,
selectors: list,
- fargate_profile_name: str | None = DEFAULT_FARGATE_PROFILE_NAME,
+ fargate_profile_name: str = DEFAULT_FARGATE_PROFILE_NAME,
create_fargate_profile_kwargs: dict | None = None,
wait_for_completion: bool = False,
aws_conn_id: str = DEFAULT_CONN_ID,
region: str | None = None,
+ waiter_delay: int = 60,
+ waiter_max_attempts: int = 30,
+ deferrable: bool = False,
Review Comment:
There's a [PR](https://github.com/apache/airflow/pull/31712#pullrequestreview-1476832822) open to have the `deferrable` parameter being inherited from `BaseDeferrableOperator`. Nothing for the other operators yet
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] syedahsn commented on a diff in pull request #31657: Deferrable mode for EksCreateFargateProfileOperator and EksDeleteFargateProfileOperator
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #31657:
URL: https://github.com/apache/airflow/pull/31657#discussion_r1228786895
##########
airflow/providers/amazon/aws/operators/eks.py:
##########
@@ -401,13 +417,31 @@ def execute(self, context: Context):
selectors=self.selectors,
**self.create_fargate_profile_kwargs,
)
-
- if self.wait_for_completion:
+ if self.deferrable:
+ self.defer(
+ trigger=EksCreateFargateProfileTrigger(
+ cluster_name=self.cluster_name,
+ fargate_profile_name=self.fargate_profile_name,
+ aws_conn_id=self.aws_conn_id,
+ poll_interval=self.waiter_delay,
+ max_attempts=self.waiter_max_attempts,
+ ),
+ method_name="execute_complete",
+ timeout=timedelta(seconds=(self.waiter_max_attempts * self.waiter_delay + 60)),
Review Comment:
I have a comment that I used in the redshift operators. I'll use that to be consistent:
```
# timeout is set to ensure that if a trigger dies, the timeout does not restart
# 60 seconds is added to allow the trigger to exit gracefully (i.e. yield TriggerEvent)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] syedahsn commented on a diff in pull request #31657: Deferrable mode for EksCreateFargateProfileOperator and EksDeleteFargateProfileOperator
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #31657:
URL: https://github.com/apache/airflow/pull/31657#discussion_r1228695276
##########
airflow/providers/amazon/aws/operators/eks.py:
##########
@@ -371,11 +381,14 @@ def __init__(
cluster_name: str,
pod_execution_role_arn: str,
selectors: list,
- fargate_profile_name: str | None = DEFAULT_FARGATE_PROFILE_NAME,
+ fargate_profile_name: str = DEFAULT_FARGATE_PROFILE_NAME,
create_fargate_profile_kwargs: dict | None = None,
wait_for_completion: bool = False,
aws_conn_id: str = DEFAULT_CONN_ID,
region: str | None = None,
+ waiter_delay: int = 60,
+ waiter_max_attempts: int = 30,
Review Comment:
I had the defaults backwards. I'll change that. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] vandonr-amz commented on a diff in pull request #31657: Deferrable mode for EksCreateFargateProfileOperator and EksDeleteFargateProfileOperator
Posted by "vandonr-amz (via GitHub)" <gi...@apache.org>.
vandonr-amz commented on code in PR #31657:
URL: https://github.com/apache/airflow/pull/31657#discussion_r1228778172
##########
airflow/providers/amazon/aws/operators/eks.py:
##########
@@ -401,13 +417,31 @@ def execute(self, context: Context):
selectors=self.selectors,
**self.create_fargate_profile_kwargs,
)
-
- if self.wait_for_completion:
+ if self.deferrable:
+ self.defer(
+ trigger=EksCreateFargateProfileTrigger(
+ cluster_name=self.cluster_name,
+ fargate_profile_name=self.fargate_profile_name,
+ aws_conn_id=self.aws_conn_id,
+ poll_interval=self.waiter_delay,
+ max_attempts=self.waiter_max_attempts,
+ ),
+ method_name="execute_complete",
+ timeout=timedelta(seconds=(self.waiter_max_attempts * self.waiter_delay + 60)),
Review Comment:
how about a comment then ?
```suggestion
# adding 60s to theoretical value to have extra room to account for time lost here and there
timeout=timedelta(seconds=(self.waiter_max_attempts * self.waiter_delay + 60)),
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ferruzzi commented on a diff in pull request #31657: Deferrable mode for EksCreateFargateProfileOperator and EksDeleteFargateProfileOperator
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #31657:
URL: https://github.com/apache/airflow/pull/31657#discussion_r1222181870
##########
airflow/providers/amazon/aws/operators/eks.py:
##########
@@ -371,11 +381,14 @@ def __init__(
cluster_name: str,
pod_execution_role_arn: str,
selectors: list,
- fargate_profile_name: str | None = DEFAULT_FARGATE_PROFILE_NAME,
+ fargate_profile_name: str = DEFAULT_FARGATE_PROFILE_NAME,
create_fargate_profile_kwargs: dict | None = None,
wait_for_completion: bool = False,
aws_conn_id: str = DEFAULT_CONN_ID,
region: str | None = None,
+ waiter_delay: int = 60,
+ waiter_max_attempts: int = 30,
+ deferrable: bool = False,
Review Comment:
Not for this PR, but another argument for a BaseAwsOperator would be to offload a bunch of these to a central place instead of repeating them constantly in every operator. The last 6 are fairly universal.
##########
tests/providers/amazon/aws/triggers/test_eks.py:
##########
@@ -0,0 +1,312 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from unittest import mock
+from unittest.mock import AsyncMock
+
+import pytest
+from botocore.exceptions import WaiterError
+
+from airflow.providers.amazon.aws.hooks.eks import EksHook
+from airflow.providers.amazon.aws.triggers.eks import (
+ EksCreateFargateProfileTrigger,
+ EksDeleteFargateProfileTrigger,
+)
+from airflow.triggers.base import TriggerEvent
+
+TEST_CLUSTER_IDENTIFIER = "test-cluster"
+TEST_FARGATE_PROFILE_NAME = "test-fargate-profile"
+TEST_POLL_INTERVAL = 10
+TEST_MAX_ATTEMPTS = 10
+TEST_AWS_CONN_ID = "test-aws-id"
+
+
+class TestEksCreateFargateProfileTrigger:
+ def test_eks_create_fargate_profile_serialize(self):
+ eks_create_fargate_profile_trigger = EksCreateFargateProfileTrigger(
+ cluster_name=TEST_CLUSTER_IDENTIFIER,
+ fargate_profile_name=TEST_FARGATE_PROFILE_NAME,
+ aws_conn_id=TEST_AWS_CONN_ID,
+ poll_interval=TEST_POLL_INTERVAL,
+ max_attempts=TEST_MAX_ATTEMPTS,
+ )
+
+ class_path, args = eks_create_fargate_profile_trigger.serialize()
+ assert class_path == "airflow.providers.amazon.aws.triggers.eks.EksCreateFargateProfileTrigger"
+ assert args["cluster_name"] == TEST_CLUSTER_IDENTIFIER
+ assert args["fargate_profile_name"] == TEST_FARGATE_PROFILE_NAME
+ assert args["aws_conn_id"] == TEST_AWS_CONN_ID
+ assert args["poll_interval"] == str(TEST_POLL_INTERVAL)
+ assert args["max_attempts"] == str(TEST_MAX_ATTEMPTS)
+
+ @pytest.mark.asyncio
+ @mock.patch.object(EksHook, "async_conn")
+ async def test_eks_create_fargate_profile_trigger_run(self, mock_async_conn):
+ a_mock = mock.MagicMock()
+ mock_async_conn.__aenter__.return_value = a_mock
+
+ a_mock.get_waiter().wait = AsyncMock()
+
+ eks_create_fargate_profile_trigger = EksCreateFargateProfileTrigger(
+ cluster_name=TEST_CLUSTER_IDENTIFIER,
+ fargate_profile_name=TEST_FARGATE_PROFILE_NAME,
+ aws_conn_id=TEST_AWS_CONN_ID,
+ poll_interval=TEST_POLL_INTERVAL,
+ max_attempts=TEST_MAX_ATTEMPTS,
+ )
+
+ generator = eks_create_fargate_profile_trigger.run()
+ response = await generator.asend(None)
+
+ assert response == TriggerEvent({"status": "success", "message": "Fargate Profile Created"})
+
+ @pytest.mark.asyncio
+ @mock.patch("asyncio.sleep")
+ @mock.patch.object(EksHook, "async_conn")
+ async def test_eks_create_fargate_profile_trigger_run_multiple_attempts(
+ self, mock_async_conn, mock_sleep
+ ):
+ a_mock = mock.MagicMock()
Review Comment:
It seems like a lot of these tests are reusing some setup code that could go into a `setup_method`
##########
airflow/providers/amazon/aws/operators/eks.py:
##########
@@ -371,11 +381,14 @@ def __init__(
cluster_name: str,
pod_execution_role_arn: str,
selectors: list,
- fargate_profile_name: str | None = DEFAULT_FARGATE_PROFILE_NAME,
+ fargate_profile_name: str = DEFAULT_FARGATE_PROFILE_NAME,
create_fargate_profile_kwargs: dict | None = None,
wait_for_completion: bool = False,
aws_conn_id: str = DEFAULT_CONN_ID,
region: str | None = None,
+ waiter_delay: int = 60,
+ waiter_max_attempts: int = 30,
Review Comment:
Are these values arbitrary? If they are, maybe have a look at the botocore code and see what the default official waiter values are.
##########
airflow/providers/amazon/aws/triggers/eks.py:
##########
@@ -0,0 +1,178 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import asyncio
+from functools import cached_property
+from typing import Any
+
+from botocore.exceptions import WaiterError
+
+from airflow.providers.amazon.aws.hooks.eks import EksHook
+from airflow.triggers.base import BaseTrigger, TriggerEvent
+
+
+class EksCreateFargateProfileTrigger(BaseTrigger):
+ """
+ Trigger for EksCreateFargateProfileOperator.
+ The trigger will asynchronously wait for the fargate profile to be created.
+
+ :param cluster_name: The name of the EKS cluster
+ :param fargate_profile_name: The name of the fargate profile
+ :param poll_interval: The amount of time in seconds to wait between attempts.
+ :param max_attempts: The maximum number of attempts to be made.
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
+ """
+
+ def __init__(
+ self,
+ cluster_name: str,
+ fargate_profile_name: str,
+ poll_interval: int,
+ max_attempts: int,
+ aws_conn_id: str,
+ ):
+ self.cluster_name = cluster_name
+ self.fargate_profile_name = fargate_profile_name
+ self.poll_interval = poll_interval
+ self.max_attempts = max_attempts
+ self.aws_conn_id = aws_conn_id
+
+ def serialize(self) -> tuple[str, dict[str, Any]]:
+ return (
+ self.__class__.__module__ + "." + self.__class__.__qualname__,
Review Comment:
Again, not necessarily for this PR: Looks like some Triggers use this and others have a fixed string. Might be a case for making a triggers/utils.py and putting in a get_trigger_name() helper?
##########
airflow/providers/amazon/aws/operators/eks.py:
##########
@@ -401,13 +417,31 @@ def execute(self, context: Context):
selectors=self.selectors,
**self.create_fargate_profile_kwargs,
)
-
- if self.wait_for_completion:
+ if self.deferrable:
+ self.defer(
+ trigger=EksCreateFargateProfileTrigger(
+ cluster_name=self.cluster_name,
+ fargate_profile_name=self.fargate_profile_name,
+ aws_conn_id=self.aws_conn_id,
+ poll_interval=self.waiter_delay,
+ max_attempts=self.waiter_max_attempts,
+ ),
+ method_name="execute_complete",
+ timeout=timedelta(seconds=(self.waiter_max_attempts * self.waiter_delay + 60)),
+ )
+ elif self.wait_for_completion:
self.log.info("Waiting for Fargate profile to provision. This will take some time.")
eks_hook.conn.get_waiter("fargate_profile_active").wait(
clusterName=self.cluster_name, fargateProfileName=self.fargate_profile_name
)
Review Comment:
+1, I would say that is implied. Not having been part of this project, my expectation was that those settings would be applied here as well, seems reasonable that a user would as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] syedahsn commented on a diff in pull request #31657: Deferrable mode for EksCreateFargateProfileOperator and EksDeleteFargateProfileOperator
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #31657:
URL: https://github.com/apache/airflow/pull/31657#discussion_r1228699857
##########
airflow/providers/amazon/aws/operators/eks.py:
##########
@@ -401,13 +417,31 @@ def execute(self, context: Context):
selectors=self.selectors,
**self.create_fargate_profile_kwargs,
)
-
- if self.wait_for_completion:
+ if self.deferrable:
+ self.defer(
+ trigger=EksCreateFargateProfileTrigger(
+ cluster_name=self.cluster_name,
+ fargate_profile_name=self.fargate_profile_name,
+ aws_conn_id=self.aws_conn_id,
+ poll_interval=self.waiter_delay,
+ max_attempts=self.waiter_max_attempts,
+ ),
+ method_name="execute_complete",
+ timeout=timedelta(seconds=(self.waiter_max_attempts * self.waiter_delay + 60)),
Review Comment:
I don't know if we need a named constant for this. 60 is arbitrary - I think adding a new constant will just add another variable someone will need to look up the value of.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] syedahsn commented on a diff in pull request #31657: Deferrable mode for EksCreateFargateProfileOperator and EksDeleteFargateProfileOperator
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #31657:
URL: https://github.com/apache/airflow/pull/31657#discussion_r1228731137
##########
tests/providers/amazon/aws/triggers/test_eks.py:
##########
@@ -0,0 +1,312 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from unittest import mock
+from unittest.mock import AsyncMock
+
+import pytest
+from botocore.exceptions import WaiterError
+
+from airflow.providers.amazon.aws.hooks.eks import EksHook
+from airflow.providers.amazon.aws.triggers.eks import (
+ EksCreateFargateProfileTrigger,
+ EksDeleteFargateProfileTrigger,
+)
+from airflow.triggers.base import TriggerEvent
+
+TEST_CLUSTER_IDENTIFIER = "test-cluster"
+TEST_FARGATE_PROFILE_NAME = "test-fargate-profile"
+TEST_POLL_INTERVAL = 10
+TEST_MAX_ATTEMPTS = 10
+TEST_AWS_CONN_ID = "test-aws-id"
+
+
+class TestEksCreateFargateProfileTrigger:
+ def test_eks_create_fargate_profile_serialize(self):
+ eks_create_fargate_profile_trigger = EksCreateFargateProfileTrigger(
+ cluster_name=TEST_CLUSTER_IDENTIFIER,
+ fargate_profile_name=TEST_FARGATE_PROFILE_NAME,
+ aws_conn_id=TEST_AWS_CONN_ID,
+ poll_interval=TEST_POLL_INTERVAL,
+ max_attempts=TEST_MAX_ATTEMPTS,
+ )
+
+ class_path, args = eks_create_fargate_profile_trigger.serialize()
+ assert class_path == "airflow.providers.amazon.aws.triggers.eks.EksCreateFargateProfileTrigger"
+ assert args["cluster_name"] == TEST_CLUSTER_IDENTIFIER
+ assert args["fargate_profile_name"] == TEST_FARGATE_PROFILE_NAME
+ assert args["aws_conn_id"] == TEST_AWS_CONN_ID
+ assert args["poll_interval"] == str(TEST_POLL_INTERVAL)
+ assert args["max_attempts"] == str(TEST_MAX_ATTEMPTS)
+
+ @pytest.mark.asyncio
+ @mock.patch.object(EksHook, "async_conn")
+ async def test_eks_create_fargate_profile_trigger_run(self, mock_async_conn):
+ a_mock = mock.MagicMock()
+ mock_async_conn.__aenter__.return_value = a_mock
+
+ a_mock.get_waiter().wait = AsyncMock()
+
+ eks_create_fargate_profile_trigger = EksCreateFargateProfileTrigger(
+ cluster_name=TEST_CLUSTER_IDENTIFIER,
+ fargate_profile_name=TEST_FARGATE_PROFILE_NAME,
+ aws_conn_id=TEST_AWS_CONN_ID,
+ poll_interval=TEST_POLL_INTERVAL,
+ max_attempts=TEST_MAX_ATTEMPTS,
+ )
+
+ generator = eks_create_fargate_profile_trigger.run()
+ response = await generator.asend(None)
+
+ assert response == TriggerEvent({"status": "success", "message": "Fargate Profile Created"})
+
+ @pytest.mark.asyncio
+ @mock.patch("asyncio.sleep")
+ @mock.patch.object(EksHook, "async_conn")
+ async def test_eks_create_fargate_profile_trigger_run_multiple_attempts(
+ self, mock_async_conn, mock_sleep
+ ):
+ a_mock = mock.MagicMock()
Review Comment:
Thats a good point. My only concern is that often there are minor changes between each setup. I'll see if there's a way to package it all in a setup method
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org