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/04/25 07:05:12 UTC
[GitHub] [airflow] syedahsn opened a new pull request, #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
syedahsn opened a new pull request, #30853:
URL: https://github.com/apache/airflow/pull/30853
Now that #30032 is merged, we can update the deferrable implementation to use custom aiobotocore waiters , as well as `async_conn` to simplify deferrable implmentation, and reduce code duplication. The method used here is described in the [README.md](https://github.com/apache/airflow/blob/main/airflow/providers/amazon/aws/triggers/README.md) included in the initial PR.
<!--
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] pankajkoti commented on a diff in pull request #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "pankajkoti (via GitHub)" <gi...@apache.org>.
pankajkoti commented on code in PR #30853:
URL: https://github.com/apache/airflow/pull/30853#discussion_r1190243960
##########
airflow/providers/amazon/aws/triggers/redshift_cluster.py:
##########
@@ -137,3 +140,73 @@ async def run(self):
},
)
yield TriggerEvent({"status": "success", "message": "Cluster Created"})
+
+
+class RedshiftPauseClusterTrigger(BaseTrigger):
+ """
+ Trigger for RedshiftPauseClusterOperator.
+ The trigger will asynchronously poll the boto3 API and wait for the
+ Redshift cluster to be in the `paused` state.
+
+ :param cluster_identifier: A unique identifier for the cluster.
+ :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_identifier: str,
+ poll_interval: int,
+ max_attempts: int,
+ aws_conn_id: str,
+ ):
+ self.cluster_identifier = cluster_identifier
+ 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 (
+ "airflow.providers.amazon.aws.triggers.redshift_cluster.RedshiftPauseClusterTrigger",
+ {
+ "cluster_identifier": self.cluster_identifier,
+ "poll_interval": str(self.poll_interval),
+ "max_attempts": str(self.max_attempts),
+ "aws_conn_id": self.aws_conn_id,
+ },
+ )
+
+ @cached_property
+ def hook(self) -> RedshiftHook:
+ return RedshiftHook(aws_conn_id=self.aws_conn_id)
+
+ async def run(self):
+ async with self.hook.async_conn as client:
+ attempt = 0
+ waiter = self.hook.get_waiter("cluster_paused", deferrable=True, client=client)
+ while attempt < int(self.max_attempts):
+ attempt = attempt + 1
+ try:
+ await waiter.wait(
+ ClusterIdentifier=self.cluster_identifier,
+ 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"Pause Cluster Failed: {error}"})
+ break
+ self.log.info(
+ "Status of cluster is %s", error.last_response["Clusters"][0]["ClusterStatus"]
+ )
+ await asyncio.sleep(int(self.poll_interval))
+ if attempt >= int(self.max_attempts):
Review Comment:
Logically, I am wondering if `attempt` would ever be `>self.max_attempts`. Should the check here be just
`if attempt == int(self.max_attempts)`?
--
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 #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "vandonr-amz (via GitHub)" <gi...@apache.org>.
vandonr-amz commented on code in PR #30853:
URL: https://github.com/apache/airflow/pull/30853#discussion_r1189016864
##########
airflow/providers/amazon/aws/operators/redshift_cluster.py:
##########
@@ -524,64 +528,55 @@ def __init__(
aws_conn_id: str = "aws_default",
deferrable: bool = False,
poll_interval: int = 10,
+ max_attempts: int = 15,
**kwargs,
):
super().__init__(**kwargs)
self.cluster_identifier = cluster_identifier
self.aws_conn_id = aws_conn_id
self.deferrable = deferrable
+ self.max_attempts = max_attempts
self.poll_interval = poll_interval
- # These parameters are added to address an issue with the boto3 API where the API
+ # These parameters are used to address an issue with the boto3 API where the API
# prematurely reports the cluster as available to receive requests. This causes the cluster
# to reject initial attempts to pause the cluster despite reporting the correct state.
- self._attempts = 10
+ self._remaining_attempts = 10
Review Comment:
I must say this is very confusing to see this next to a `max_attempts` parameter and realize that they have nothing to do with each other
--
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] dstandish commented on a diff in pull request #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #30853:
URL: https://github.com/apache/airflow/pull/30853#discussion_r1186375024
##########
airflow/providers/amazon/aws/operators/redshift_cluster.py:
##########
@@ -524,64 +527,52 @@ def __init__(
aws_conn_id: str = "aws_default",
deferrable: bool = False,
poll_interval: int = 10,
+ max_attempts: int = 15,
**kwargs,
):
super().__init__(**kwargs)
self.cluster_identifier = cluster_identifier
self.aws_conn_id = aws_conn_id
self.deferrable = deferrable
+ self.max_attempts = max_attempts
self.poll_interval = poll_interval
- # These parameters are added to address an issue with the boto3 API where the API
+ # These parameters are used to address an issue with the boto3 API where the API
# prematurely reports the cluster as available to receive requests. This causes the cluster
# to reject initial attempts to pause the cluster despite reporting the correct state.
self._attempts = 10
self._attempt_interval = 15
def execute(self, context: Context):
redshift_hook = RedshiftHook(aws_conn_id=self.aws_conn_id)
+ while self._attempts >= 1:
+ try:
+ redshift_hook.get_conn().pause_cluster(ClusterIdentifier=self.cluster_identifier)
+ break
+ except redshift_hook.get_conn().exceptions.InvalidClusterStateFault as error:
+ self._attempts = self._attempts - 1
+ if self._attempts > 0:
+ self.log.error("Unable to pause cluster. %d attempts remaining.", self._attempts)
+ time.sleep(self._attempt_interval)
+ else:
+ raise error
if self.deferrable:
self.defer(
- timeout=self.execution_timeout,
- trigger=RedshiftClusterTrigger(
- task_id=self.task_id,
+ trigger=RedshiftPauseClusterTrigger(
+ cluster_identifier=self.cluster_identifier,
poll_interval=self.poll_interval,
+ max_attempts=self.max_attempts,
Review Comment:
Like... you can still have max attempts be part of the operator interface -- user doesn't care about the trigger interface right? so just when deferring, you calculate the timeout based on supplied max attempts. maybe that makes everyone happy?
--
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 #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #30853:
URL: https://github.com/apache/airflow/pull/30853#discussion_r1185361174
##########
airflow/providers/amazon/aws/triggers/redshift_cluster.py:
##########
@@ -137,3 +140,70 @@ async def run(self):
},
)
yield TriggerEvent({"status": "success", "message": "Cluster Created"})
+
+
+class RedshiftPauseClusterTrigger(BaseTrigger):
+ """
+ Trigger for RedshiftPauseClusterOperator.
+ The trigger will asynchronously poll the boto3 API and wait for the
+ Redshift cluster to be in the `paused` state.
+
+ :param cluster_identifier: A unique identifier for the cluster.
+ :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_identifier: str,
+ poll_interval: int,
+ max_attempts: int,
+ aws_conn_id: str,
+ ):
+ self.cluster_identifier = cluster_identifier
+ 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 (
+ "airflow.providers.amazon.aws.triggers.redshift_cluster.RedshiftPauseClusterTrigger",
+ {
+ "cluster_identifier": str(self.cluster_identifier),
+ "poll_interval": str(self.poll_interval),
+ "max_attempts": str(self.max_attempts),
+ "aws_conn_id": str(self.aws_conn_id),
+ },
+ )
+
+ @cached_property
+ def hook(self) -> RedshiftHook:
+ return RedshiftHook(aws_conn_id=self.aws_conn_id)
+
+ async def run(self):
+ async with self.hook.async_conn as client:
+ attempt = 0
+ while attempt < int(self.max_attempts):
+ attempt = attempt + 1
+ try:
+ waiter = self.hook.get_waiter("cluster_paused", deferrable=True, client=client)
+ await waiter.wait(
+ ClusterIdentifier=self.cluster_identifier,
+ WaiterConfig={
+ "Delay": int(self.poll_interval),
+ "MaxAttempts": 1,
+ },
+ )
+ break
+ except WaiterError as error:
+ self.log.info(
+ "Status of cluster is %s", error.last_response["Clusters"][0]["ClusterStatus"]
+ )
+ await asyncio.sleep(int(self.poll_interval))
+ if attempt >= int(self.max_attempts):
+ yield TriggerEvent(
+ {"status": "failure", "message": "Resume Cluster Failed - max attempts reached."}
Review Comment:
sorry, that's a typo.
--
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] pankajastro commented on a diff in pull request #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "pankajastro (via GitHub)" <gi...@apache.org>.
pankajastro commented on code in PR #30853:
URL: https://github.com/apache/airflow/pull/30853#discussion_r1185321640
##########
airflow/providers/amazon/aws/triggers/redshift_cluster.py:
##########
@@ -137,3 +140,70 @@ async def run(self):
},
)
yield TriggerEvent({"status": "success", "message": "Cluster Created"})
+
+
+class RedshiftPauseClusterTrigger(BaseTrigger):
+ """
+ Trigger for RedshiftPauseClusterOperator.
+ The trigger will asynchronously poll the boto3 API and wait for the
+ Redshift cluster to be in the `paused` state.
+
+ :param cluster_identifier: A unique identifier for the cluster.
+ :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_identifier: str,
+ poll_interval: int,
+ max_attempts: int,
+ aws_conn_id: str,
+ ):
+ self.cluster_identifier = cluster_identifier
+ 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 (
+ "airflow.providers.amazon.aws.triggers.redshift_cluster.RedshiftPauseClusterTrigger",
+ {
+ "cluster_identifier": str(self.cluster_identifier),
Review Comment:
why we are casting here to str?
##########
airflow/providers/amazon/aws/triggers/redshift_cluster.py:
##########
@@ -137,3 +140,70 @@ async def run(self):
},
)
yield TriggerEvent({"status": "success", "message": "Cluster Created"})
+
+
+class RedshiftPauseClusterTrigger(BaseTrigger):
+ """
+ Trigger for RedshiftPauseClusterOperator.
+ The trigger will asynchronously poll the boto3 API and wait for the
+ Redshift cluster to be in the `paused` state.
+
+ :param cluster_identifier: A unique identifier for the cluster.
+ :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_identifier: str,
+ poll_interval: int,
+ max_attempts: int,
+ aws_conn_id: str,
+ ):
+ self.cluster_identifier = cluster_identifier
+ 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 (
+ "airflow.providers.amazon.aws.triggers.redshift_cluster.RedshiftPauseClusterTrigger",
+ {
+ "cluster_identifier": str(self.cluster_identifier),
+ "poll_interval": str(self.poll_interval),
+ "max_attempts": str(self.max_attempts),
+ "aws_conn_id": str(self.aws_conn_id),
+ },
+ )
+
+ @cached_property
+ def hook(self) -> RedshiftHook:
+ return RedshiftHook(aws_conn_id=self.aws_conn_id)
+
+ async def run(self):
+ async with self.hook.async_conn as client:
+ attempt = 0
+ while attempt < int(self.max_attempts):
+ attempt = attempt + 1
+ try:
+ waiter = self.hook.get_waiter("cluster_paused", deferrable=True, client=client)
+ await waiter.wait(
+ ClusterIdentifier=self.cluster_identifier,
+ WaiterConfig={
+ "Delay": int(self.poll_interval),
+ "MaxAttempts": 1,
+ },
+ )
+ break
+ except WaiterError as error:
+ self.log.info(
+ "Status of cluster is %s", error.last_response["Clusters"][0]["ClusterStatus"]
+ )
+ await asyncio.sleep(int(self.poll_interval))
+ if attempt >= int(self.max_attempts):
+ yield TriggerEvent(
+ {"status": "failure", "message": "Resume Cluster Failed - max attempts reached."}
+ )
+ else:
Review Comment:
does it makes sense to check status before marking success
##########
airflow/providers/amazon/aws/triggers/redshift_cluster.py:
##########
@@ -137,3 +140,70 @@ async def run(self):
},
)
yield TriggerEvent({"status": "success", "message": "Cluster Created"})
+
+
+class RedshiftPauseClusterTrigger(BaseTrigger):
+ """
+ Trigger for RedshiftPauseClusterOperator.
+ The trigger will asynchronously poll the boto3 API and wait for the
+ Redshift cluster to be in the `paused` state.
+
+ :param cluster_identifier: A unique identifier for the cluster.
+ :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_identifier: str,
+ poll_interval: int,
+ max_attempts: int,
+ aws_conn_id: str,
+ ):
+ self.cluster_identifier = cluster_identifier
+ 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 (
+ "airflow.providers.amazon.aws.triggers.redshift_cluster.RedshiftPauseClusterTrigger",
+ {
+ "cluster_identifier": str(self.cluster_identifier),
+ "poll_interval": str(self.poll_interval),
+ "max_attempts": str(self.max_attempts),
+ "aws_conn_id": str(self.aws_conn_id),
+ },
+ )
+
+ @cached_property
+ def hook(self) -> RedshiftHook:
+ return RedshiftHook(aws_conn_id=self.aws_conn_id)
+
+ async def run(self):
+ async with self.hook.async_conn as client:
+ attempt = 0
+ while attempt < int(self.max_attempts):
+ attempt = attempt + 1
+ try:
+ waiter = self.hook.get_waiter("cluster_paused", deferrable=True, client=client)
+ await waiter.wait(
+ ClusterIdentifier=self.cluster_identifier,
+ WaiterConfig={
+ "Delay": int(self.poll_interval),
+ "MaxAttempts": 1,
+ },
+ )
+ break
+ except WaiterError as error:
+ self.log.info(
+ "Status of cluster is %s", error.last_response["Clusters"][0]["ClusterStatus"]
+ )
+ await asyncio.sleep(int(self.poll_interval))
+ if attempt >= int(self.max_attempts):
+ yield TriggerEvent(
+ {"status": "failure", "message": "Resume Cluster Failed - max attempts reached."}
Review Comment:
why resume cluster, typo?
##########
airflow/providers/amazon/aws/triggers/redshift_cluster.py:
##########
@@ -137,3 +140,70 @@ async def run(self):
},
)
yield TriggerEvent({"status": "success", "message": "Cluster Created"})
+
+
+class RedshiftPauseClusterTrigger(BaseTrigger):
+ """
+ Trigger for RedshiftPauseClusterOperator.
+ The trigger will asynchronously poll the boto3 API and wait for the
+ Redshift cluster to be in the `paused` state.
+
+ :param cluster_identifier: A unique identifier for the cluster.
+ :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_identifier: str,
+ poll_interval: int,
+ max_attempts: int,
+ aws_conn_id: str,
+ ):
+ self.cluster_identifier = cluster_identifier
+ 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 (
+ "airflow.providers.amazon.aws.triggers.redshift_cluster.RedshiftPauseClusterTrigger",
+ {
+ "cluster_identifier": str(self.cluster_identifier),
+ "poll_interval": str(self.poll_interval),
+ "max_attempts": str(self.max_attempts),
+ "aws_conn_id": str(self.aws_conn_id),
+ },
+ )
+
+ @cached_property
+ def hook(self) -> RedshiftHook:
+ return RedshiftHook(aws_conn_id=self.aws_conn_id)
+
+ async def run(self):
+ async with self.hook.async_conn as client:
+ attempt = 0
+ while attempt < int(self.max_attempts):
+ attempt = attempt + 1
+ try:
+ waiter = self.hook.get_waiter("cluster_paused", deferrable=True, client=client)
Review Comment:
I think we won't be able to use waiter which Boto lib provides for example in create cluster we are using https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/redshift/waiter/ClusterAvailable.html ??
Here, I believe you are handling it but in general, for all triggers, we will have to
- wait for desired state + all terminal state
- When either we reach the desired state or terminal state or are done with all the attempts check the state and based on that mark the task fail or success
##########
airflow/providers/amazon/aws/triggers/redshift_cluster.py:
##########
@@ -137,3 +140,70 @@ async def run(self):
},
)
yield TriggerEvent({"status": "success", "message": "Cluster Created"})
+
+
+class RedshiftPauseClusterTrigger(BaseTrigger):
+ """
+ Trigger for RedshiftPauseClusterOperator.
+ The trigger will asynchronously poll the boto3 API and wait for the
+ Redshift cluster to be in the `paused` state.
+
+ :param cluster_identifier: A unique identifier for the cluster.
+ :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_identifier: str,
+ poll_interval: int,
+ max_attempts: int,
+ aws_conn_id: str,
+ ):
+ self.cluster_identifier = cluster_identifier
+ 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 (
+ "airflow.providers.amazon.aws.triggers.redshift_cluster.RedshiftPauseClusterTrigger",
+ {
+ "cluster_identifier": str(self.cluster_identifier),
+ "poll_interval": str(self.poll_interval),
+ "max_attempts": str(self.max_attempts),
+ "aws_conn_id": str(self.aws_conn_id),
+ },
+ )
+
+ @cached_property
+ def hook(self) -> RedshiftHook:
+ return RedshiftHook(aws_conn_id=self.aws_conn_id)
+
+ async def run(self):
+ async with self.hook.async_conn as client:
+ attempt = 0
+ while attempt < int(self.max_attempts):
+ attempt = attempt + 1
+ try:
+ waiter = self.hook.get_waiter("cluster_paused", deferrable=True, client=client)
+ await waiter.wait(
+ ClusterIdentifier=self.cluster_identifier,
+ WaiterConfig={
+ "Delay": int(self.poll_interval),
+ "MaxAttempts": 1,
Review Comment:
I agree with @pankajkoti waiter having 1 attempt is a bit unnatural
--
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] potiuk commented on pull request #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #30853:
URL: https://github.com/apache/airflow/pull/30853#issuecomment-1527215516
> > FYI. Airflow 2.6 implements interleaving Tirgger and Task logs together. So for those users it will be no problem, they will se trigger logs coming.
>
> that's correct. But sometimes when the operator waits for a long time in that case if we do some periodic logging like current state sleeping time etc that helps the user especially when they have limited access and feel like things are ok but if this is missing (In this case since no log in trigger run method) they might not be sure what going on in the background. It would have been nice if we could have achieved the same along with the waiter.
Quite agree. My point there was that even if there is no support for providing such waiter feedback/callback the trigger could simply slice the time and wait in (say 30 seconds) intervals, handle the timout every time it finishes and emit log entry and come back to waiting. With the new interleaving mechanism, that would end up with a log entry every 30 s.
That would not require changes in the botocore @syedahsn @pankajastro :) (as far as I understand where the discussion stalled).
--
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 pull request #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on PR #30853:
URL: https://github.com/apache/airflow/pull/30853#issuecomment-1535291932
>q: I have noticed that with waiter's performance of an operator has degraded, though I need to test it a bit more.
I haven't noticed any differences. If anything, previously, when I was testing without the custom waiters, I would notice that the Triggerer's async thread would get blocked multiple times. Using custom waiters, that happens a lot less now.
--
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] phanikumv commented on pull request #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "phanikumv (via GitHub)" <gi...@apache.org>.
phanikumv commented on PR #30853:
URL: https://github.com/apache/airflow/pull/30853#issuecomment-1522125059
> Thanks for the review! Yes the botocore waiters don't log any messages as they are polling the AWS service. Because this is implemented in the botocore code, in order to do this, we'd have to make the changes there. This is something that we've considered, and may end up implementing - unless the botocore team has a reason for not doing this by default. I imagine others have raised a similar concern,
This will be a problem for our current users who are seeing log messages on the Triggerer vs no logs post this PR.
--
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 #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #30853:
URL: https://github.com/apache/airflow/pull/30853#discussion_r1186331441
##########
airflow/providers/amazon/aws/operators/redshift_cluster.py:
##########
@@ -524,64 +527,52 @@ def __init__(
aws_conn_id: str = "aws_default",
deferrable: bool = False,
poll_interval: int = 10,
+ max_attempts: int = 15,
**kwargs,
):
super().__init__(**kwargs)
self.cluster_identifier = cluster_identifier
self.aws_conn_id = aws_conn_id
self.deferrable = deferrable
+ self.max_attempts = max_attempts
self.poll_interval = poll_interval
- # These parameters are added to address an issue with the boto3 API where the API
+ # These parameters are used to address an issue with the boto3 API where the API
# prematurely reports the cluster as available to receive requests. This causes the cluster
# to reject initial attempts to pause the cluster despite reporting the correct state.
self._attempts = 10
self._attempt_interval = 15
def execute(self, context: Context):
redshift_hook = RedshiftHook(aws_conn_id=self.aws_conn_id)
+ while self._attempts >= 1:
+ try:
+ redshift_hook.get_conn().pause_cluster(ClusterIdentifier=self.cluster_identifier)
+ break
+ except redshift_hook.get_conn().exceptions.InvalidClusterStateFault as error:
+ self._attempts = self._attempts - 1
+ if self._attempts > 0:
+ self.log.error("Unable to pause cluster. %d attempts remaining.", self._attempts)
+ time.sleep(self._attempt_interval)
+ else:
+ raise error
if self.deferrable:
self.defer(
- timeout=self.execution_timeout,
- trigger=RedshiftClusterTrigger(
- task_id=self.task_id,
+ trigger=RedshiftPauseClusterTrigger(
+ cluster_identifier=self.cluster_identifier,
poll_interval=self.poll_interval,
+ max_attempts=self.max_attempts,
Review Comment:
Using the max attempts vs the timeout was essentially a design choice due to how the boto waiter implements things. The boto waiter defines `MaxAttempts` and `Delay` so we decided to go with those parameters. Initially, when we had our own manual waiter, we were going with a `waiter_countdown` [parameter](https://github.com/apache/airflow/blob/main/airflow/providers/amazon/aws/operators/emr.py#L176) which would map to the `timeout` on the Trigger. But we are in the process of deprecating that parameter. ( As an aside, we calculate `waiter_max_attempts` as `waiter_max_attempts = waiter_countdown // waiter_check_interval_seconds`).
>one problem with using max_attempts in trigger is if trigger is killed and is respawned it will reset -- not so if we use timeout.
This is a good point. We could solve this by making the `attempt` number as part of the kwargs of the trigger - which would save the value if it was killed at any point. What do you think?
--
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 #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #30853:
URL: https://github.com/apache/airflow/pull/30853#discussion_r1186398503
##########
airflow/providers/amazon/aws/operators/redshift_cluster.py:
##########
@@ -524,64 +527,52 @@ def __init__(
aws_conn_id: str = "aws_default",
deferrable: bool = False,
poll_interval: int = 10,
+ max_attempts: int = 15,
**kwargs,
):
super().__init__(**kwargs)
self.cluster_identifier = cluster_identifier
self.aws_conn_id = aws_conn_id
self.deferrable = deferrable
+ self.max_attempts = max_attempts
self.poll_interval = poll_interval
- # These parameters are added to address an issue with the boto3 API where the API
+ # These parameters are used to address an issue with the boto3 API where the API
# prematurely reports the cluster as available to receive requests. This causes the cluster
# to reject initial attempts to pause the cluster despite reporting the correct state.
self._attempts = 10
self._attempt_interval = 15
def execute(self, context: Context):
redshift_hook = RedshiftHook(aws_conn_id=self.aws_conn_id)
+ while self._attempts >= 1:
+ try:
+ redshift_hook.get_conn().pause_cluster(ClusterIdentifier=self.cluster_identifier)
+ break
+ except redshift_hook.get_conn().exceptions.InvalidClusterStateFault as error:
+ self._attempts = self._attempts - 1
+ if self._attempts > 0:
+ self.log.error("Unable to pause cluster. %d attempts remaining.", self._attempts)
+ time.sleep(self._attempt_interval)
+ else:
+ raise error
if self.deferrable:
self.defer(
- timeout=self.execution_timeout,
- trigger=RedshiftClusterTrigger(
- task_id=self.task_id,
+ trigger=RedshiftPauseClusterTrigger(
+ cluster_identifier=self.cluster_identifier,
poll_interval=self.poll_interval,
+ max_attempts=self.max_attempts,
Review Comment:
Thinking about it a bit more, does Airflow support emitting a TriggerEvent on Trigger timeout? There is a `cleanup` method that executes after `run` but can we emit a TriggerEvent from there? Or is there a better way?
--
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 #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #30853:
URL: https://github.com/apache/airflow/pull/30853#discussion_r1187018026
##########
airflow/providers/amazon/aws/operators/redshift_cluster.py:
##########
@@ -524,64 +527,52 @@ def __init__(
aws_conn_id: str = "aws_default",
deferrable: bool = False,
poll_interval: int = 10,
+ max_attempts: int = 15,
**kwargs,
):
super().__init__(**kwargs)
self.cluster_identifier = cluster_identifier
self.aws_conn_id = aws_conn_id
self.deferrable = deferrable
+ self.max_attempts = max_attempts
self.poll_interval = poll_interval
- # These parameters are added to address an issue with the boto3 API where the API
+ # These parameters are used to address an issue with the boto3 API where the API
# prematurely reports the cluster as available to receive requests. This causes the cluster
# to reject initial attempts to pause the cluster despite reporting the correct state.
self._attempts = 10
self._attempt_interval = 15
def execute(self, context: Context):
redshift_hook = RedshiftHook(aws_conn_id=self.aws_conn_id)
+ while self._attempts >= 1:
+ try:
+ redshift_hook.get_conn().pause_cluster(ClusterIdentifier=self.cluster_identifier)
+ break
+ except redshift_hook.get_conn().exceptions.InvalidClusterStateFault as error:
+ self._attempts = self._attempts - 1
+ if self._attempts > 0:
+ self.log.error("Unable to pause cluster. %d attempts remaining.", self._attempts)
+ time.sleep(self._attempt_interval)
+ else:
+ raise error
if self.deferrable:
self.defer(
- timeout=self.execution_timeout,
- trigger=RedshiftClusterTrigger(
- task_id=self.task_id,
+ trigger=RedshiftPauseClusterTrigger(
+ cluster_identifier=self.cluster_identifier,
poll_interval=self.poll_interval,
+ max_attempts=self.max_attempts,
Review Comment:
I ended up going with adding a timeout, as you suggested. But I also left the code to track the number of attempts in as well. The reason is, when a Trigger times out, it does not yield a `TriggerEvent`. This makes the logs slightly confusing, because it seems like the Trigger succeeded.
```
[2023-05-08, 02:19:30 UTC] {credentials.py:617} INFO - Found credentials in shared credentials file: ~/.aws/credentials
[2023-05-08, 02:19:38 UTC] {redshift_cluster.py:203} INFO - Status of cluster is pausing
[2023-05-08, 02:19:57 UTC] {redshift_cluster.py:203} INFO - Status of cluster is pausing
[2023-05-08, 02:20:15 UTC] {redshift_cluster.py:203} INFO - Status of cluster is pausing
[2023-05-08, 02:20:28 UTC] {taskinstance.py:1144} INFO - Dependencies all met for dep_context=non-requeueable deps ti=<TaskInstance: example_redshift.pause_cluster manual__2023-05-08T02:11:38.809301+00:00 [queued]>
[2023-05-08, 02:20:28 UTC] {taskinstance.py:1144} INFO - Dependencies all met for dep_context=requeueable deps ti=<TaskInstance: example_redshift.pause_cluster manual__2023-05-08T02:11:38.809301+00:00 [queued]>
[2023-05-08, 02:20:28 UTC] {taskinstance.py:1343} INFO - Resuming after deferral
[2023-05-08, 02:20:28 UTC] {taskinstance.py:1364} INFO - Executing <Task(RedshiftPauseClusterOperator): pause_cluster> on 2023-05-08 02:11:38.809301+00:00
[2023-05-08, 02:20:28 UTC] {standard_task_runner.py:57} INFO - Started process 328 to run task
[2023-05-08, 02:20:28 UTC] {standard_task_runner.py:84} INFO - Running: ['***', 'tasks', 'run', 'example_redshift', 'pause_cluster', 'manual__2023-05-08T02:11:38.809301+00:00', '--job-id', '17', '--raw', '--subdir', 'DAGS_FOLDER/example_redshift.py', '--cfg-path', '/tmp/tmp582aoom9']
[2023-05-08, 02:20:28 UTC] {standard_task_runner.py:85} INFO - Job 17: Subtask pause_cluster
[2023-05-08, 02:20:28 UTC] {task_command.py:410} INFO - Running <TaskInstance: example_redshift.pause_cluster manual__2023-05-08T02:11:38.809301+00:00 [running]> on host 9ed7c7091244
[2023-05-08, 02:20:28 UTC] {taskinstance.py:1863} ERROR - Task failed with exception
airflow.exceptions.TaskDeferralError: Trigger/execution timeout
[2023-05-08, 02:20:28 UTC] {taskinstance.py:1382} INFO - Marking task as FAILED. dag_id=example_redshift, task_id=pause_cluster, execution_date=20230508T021138, start_date=20230508T021920, end_date=20230508T022028
```
I felt this was not a good user experience, so I decided to add the timeout along with the number of attempts. This way, under normal circumstances, a Trigger would fail after attempting the max number of attempts, and the Trigger would yield a `TriggerEvent` with an error message explaining the error. In the case the Trigger is killed and respawned, the timeout from the operator will persist and fail the Trigger at the correct time.
--
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 #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "vandonr-amz (via GitHub)" <gi...@apache.org>.
vandonr-amz commented on code in PR #30853:
URL: https://github.com/apache/airflow/pull/30853#discussion_r1189024588
##########
tests/providers/amazon/aws/triggers/test_redshift_cluster.py:
##########
@@ -72,3 +76,143 @@ async def test_redshift_create_cluster_trigger_run(self, mock_async_conn):
response = await generator.asend(None)
assert response == TriggerEvent({"status": "success", "message": "Cluster Created"})
+
+
+class TestRedshiftPauseClusterTrigger:
+ def test_redshift_resume_cluster_trigger_serialize(self):
+ redshift_resume_cluster_trigger = RedshiftPauseClusterTrigger(
+ cluster_identifier=TEST_CLUSTER_IDENTIFIER,
+ poll_interval=TEST_POLL_INTERVAL,
+ max_attempts=TEST_MAX_ATTEMPT,
+ aws_conn_id=TEST_AWS_CONN_ID,
+ )
+ class_path, args = redshift_resume_cluster_trigger.serialize()
+ assert (
+ class_path == "airflow.providers.amazon.aws.triggers.redshift_cluster.RedshiftPauseClusterTrigger"
+ )
+ assert args["cluster_identifier"] == TEST_CLUSTER_IDENTIFIER
+ assert args["poll_interval"] == str(TEST_POLL_INTERVAL)
+ assert args["max_attempts"] == str(TEST_MAX_ATTEMPT)
+ assert args["aws_conn_id"] == TEST_AWS_CONN_ID
+
+ @pytest.mark.asyncio
+ @async_mock.patch("airflow.providers.amazon.aws.hooks.redshift_cluster.RedshiftHook.get_waiter")
+ @async_mock.patch("airflow.providers.amazon.aws.hooks.redshift_cluster.RedshiftHook.async_conn")
Review Comment:
same remark as above for this and the other mocks in the file
--
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] pankajastro commented on pull request #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "pankajastro (via GitHub)" <gi...@apache.org>.
pankajastro commented on PR #30853:
URL: https://github.com/apache/airflow/pull/30853#issuecomment-1527050111
> FYI. Airflow 2.6 implements interleaving Tirgger and Task logs together. So for those users it will be no problem, they will se trigger logs coming.
that's correct. But sometimes when the operator waits for a long time in that case if we do some periodic logging like current state sleeping time etc that helps the user especially when they have limited access and feel like things are ok but if this is missing (In this case since no log in trigger run method) they might not be sure what going on in the background. It would have been nice if we could have achieved the same along with the waiter.
--
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] dstandish commented on a diff in pull request #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #30853:
URL: https://github.com/apache/airflow/pull/30853#discussion_r1186375024
##########
airflow/providers/amazon/aws/operators/redshift_cluster.py:
##########
@@ -524,64 +527,52 @@ def __init__(
aws_conn_id: str = "aws_default",
deferrable: bool = False,
poll_interval: int = 10,
+ max_attempts: int = 15,
**kwargs,
):
super().__init__(**kwargs)
self.cluster_identifier = cluster_identifier
self.aws_conn_id = aws_conn_id
self.deferrable = deferrable
+ self.max_attempts = max_attempts
self.poll_interval = poll_interval
- # These parameters are added to address an issue with the boto3 API where the API
+ # These parameters are used to address an issue with the boto3 API where the API
# prematurely reports the cluster as available to receive requests. This causes the cluster
# to reject initial attempts to pause the cluster despite reporting the correct state.
self._attempts = 10
self._attempt_interval = 15
def execute(self, context: Context):
redshift_hook = RedshiftHook(aws_conn_id=self.aws_conn_id)
+ while self._attempts >= 1:
+ try:
+ redshift_hook.get_conn().pause_cluster(ClusterIdentifier=self.cluster_identifier)
+ break
+ except redshift_hook.get_conn().exceptions.InvalidClusterStateFault as error:
+ self._attempts = self._attempts - 1
+ if self._attempts > 0:
+ self.log.error("Unable to pause cluster. %d attempts remaining.", self._attempts)
+ time.sleep(self._attempt_interval)
+ else:
+ raise error
if self.deferrable:
self.defer(
- timeout=self.execution_timeout,
- trigger=RedshiftClusterTrigger(
- task_id=self.task_id,
+ trigger=RedshiftPauseClusterTrigger(
+ cluster_identifier=self.cluster_identifier,
poll_interval=self.poll_interval,
+ max_attempts=self.max_attempts,
Review Comment:
Like... you can still have max attempts be part of the operator interface -- user doesn't care about the trigger interface right? so just when deferring, you calculate the timeout based on supplied max attempts. maybe that makes everyone happy? so, operator takes max attempts + poll interval, but trigger knows only poll interval, loops forever, and the triggerer system itself handles timeout.
--
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 #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #30853:
URL: https://github.com/apache/airflow/pull/30853#discussion_r1185268046
##########
airflow/providers/amazon/aws/triggers/redshift_cluster.py:
##########
@@ -137,3 +140,70 @@ async def run(self):
},
)
yield TriggerEvent({"status": "success", "message": "Cluster Created"})
+
+
+class RedshiftPauseClusterTrigger(BaseTrigger):
+ """
+ Trigger for RedshiftPauseClusterOperator.
+ The trigger will asynchronously poll the boto3 API and wait for the
+ Redshift cluster to be in the `paused` state.
+
+ :param cluster_identifier: A unique identifier for the cluster.
+ :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_identifier: str,
+ poll_interval: int,
+ max_attempts: int,
+ aws_conn_id: str,
+ ):
+ self.cluster_identifier = cluster_identifier
+ 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 (
+ "airflow.providers.amazon.aws.triggers.redshift_cluster.RedshiftPauseClusterTrigger",
+ {
+ "cluster_identifier": str(self.cluster_identifier),
+ "poll_interval": str(self.poll_interval),
+ "max_attempts": str(self.max_attempts),
+ "aws_conn_id": str(self.aws_conn_id),
+ },
+ )
+
+ @cached_property
+ def hook(self) -> RedshiftHook:
+ return RedshiftHook(aws_conn_id=self.aws_conn_id)
+
+ async def run(self):
+ async with self.hook.async_conn as client:
+ attempt = 0
+ while attempt < int(self.max_attempts):
+ attempt = attempt + 1
+ try:
+ waiter = self.hook.get_waiter("cluster_paused", deferrable=True, client=client)
+ await waiter.wait(
+ ClusterIdentifier=self.cluster_identifier,
+ WaiterConfig={
+ "Delay": int(self.poll_interval),
+ "MaxAttempts": 1,
+ },
+ )
+ break
+ except WaiterError as error:
+ self.log.info(
+ "Status of cluster is %s", error.last_response["Clusters"][0]["ClusterStatus"]
+ )
+ await asyncio.sleep(int(self.poll_interval))
Review Comment:
Yes, that's a good point. I'll make that fix. 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] syedahsn commented on pull request #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on PR #30853:
URL: https://github.com/apache/airflow/pull/30853#issuecomment-1522103025
Thanks for the review! Yes the botocore waiters don't log any messages as they are polling the AWS service. Because this is implemented in the botocore code, in order to do this, we'd have to make the changes there. This is something that we've considered, and may end up implementing - unless the botocore team has a reason for not doing this by default. I imagine others have raised a similar concern,
--
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 #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas merged PR #30853:
URL: https://github.com/apache/airflow/pull/30853
--
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] pankajkoti commented on a diff in pull request #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "pankajkoti (via GitHub)" <gi...@apache.org>.
pankajkoti commented on code in PR #30853:
URL: https://github.com/apache/airflow/pull/30853#discussion_r1184754969
##########
airflow/providers/amazon/aws/triggers/redshift_cluster.py:
##########
@@ -137,3 +140,70 @@ async def run(self):
},
)
yield TriggerEvent({"status": "success", "message": "Cluster Created"})
+
+
+class RedshiftPauseClusterTrigger(BaseTrigger):
+ """
+ Trigger for RedshiftPauseClusterOperator.
+ The trigger will asynchronously poll the boto3 API and wait for the
+ Redshift cluster to be in the `paused` state.
+
+ :param cluster_identifier: A unique identifier for the cluster.
+ :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_identifier: str,
+ poll_interval: int,
+ max_attempts: int,
+ aws_conn_id: str,
+ ):
+ self.cluster_identifier = cluster_identifier
+ 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 (
+ "airflow.providers.amazon.aws.triggers.redshift_cluster.RedshiftPauseClusterTrigger",
+ {
+ "cluster_identifier": str(self.cluster_identifier),
+ "poll_interval": str(self.poll_interval),
+ "max_attempts": str(self.max_attempts),
+ "aws_conn_id": str(self.aws_conn_id),
+ },
+ )
+
+ @cached_property
+ def hook(self) -> RedshiftHook:
+ return RedshiftHook(aws_conn_id=self.aws_conn_id)
+
+ async def run(self):
+ async with self.hook.async_conn as client:
+ attempt = 0
+ while attempt < int(self.max_attempts):
+ attempt = attempt + 1
+ try:
+ waiter = self.hook.get_waiter("cluster_paused", deferrable=True, client=client)
+ await waiter.wait(
+ ClusterIdentifier=self.cluster_identifier,
+ WaiterConfig={
+ "Delay": int(self.poll_interval),
+ "MaxAttempts": 1,
+ },
+ )
+ break
+ except WaiterError as error:
+ self.log.info(
+ "Status of cluster is %s", error.last_response["Clusters"][0]["ClusterStatus"]
+ )
+ await asyncio.sleep(int(self.poll_interval))
Review Comment:
Shouldn't we re-raise the `WaiterError` here? If the waiter reaches a failure path, it would be good if we don't wait for all the attempts to finish in these cases.
--
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] pankajkoti commented on a diff in pull request #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "pankajkoti (via GitHub)" <gi...@apache.org>.
pankajkoti commented on code in PR #30853:
URL: https://github.com/apache/airflow/pull/30853#discussion_r1190245323
##########
airflow/providers/amazon/aws/triggers/redshift_cluster.py:
##########
@@ -137,3 +140,73 @@ async def run(self):
},
)
yield TriggerEvent({"status": "success", "message": "Cluster Created"})
+
+
+class RedshiftPauseClusterTrigger(BaseTrigger):
+ """
+ Trigger for RedshiftPauseClusterOperator.
+ The trigger will asynchronously poll the boto3 API and wait for the
+ Redshift cluster to be in the `paused` state.
+
+ :param cluster_identifier: A unique identifier for the cluster.
+ :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_identifier: str,
+ poll_interval: int,
+ max_attempts: int,
+ aws_conn_id: str,
+ ):
+ self.cluster_identifier = cluster_identifier
+ 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 (
+ "airflow.providers.amazon.aws.triggers.redshift_cluster.RedshiftPauseClusterTrigger",
+ {
+ "cluster_identifier": self.cluster_identifier,
+ "poll_interval": str(self.poll_interval),
+ "max_attempts": str(self.max_attempts),
+ "aws_conn_id": self.aws_conn_id,
+ },
+ )
+
+ @cached_property
+ def hook(self) -> RedshiftHook:
+ return RedshiftHook(aws_conn_id=self.aws_conn_id)
+
+ async def run(self):
+ async with self.hook.async_conn as client:
+ attempt = 0
+ waiter = self.hook.get_waiter("cluster_paused", deferrable=True, client=client)
+ while attempt < int(self.max_attempts):
+ attempt = attempt + 1
+ try:
+ await waiter.wait(
+ ClusterIdentifier=self.cluster_identifier,
+ 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"Pause Cluster Failed: {error}"})
+ break
+ self.log.info(
+ "Status of cluster is %s", error.last_response["Clusters"][0]["ClusterStatus"]
+ )
+ await asyncio.sleep(int(self.poll_interval))
+ if attempt >= int(self.max_attempts):
Review Comment:
Also why do we need to cast `self.max_attempts` to `int`? Isn't the type int already as per type annotation?
--
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 #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #30853:
URL: https://github.com/apache/airflow/pull/30853#discussion_r1186384451
##########
airflow/providers/amazon/aws/operators/redshift_cluster.py:
##########
@@ -524,64 +527,52 @@ def __init__(
aws_conn_id: str = "aws_default",
deferrable: bool = False,
poll_interval: int = 10,
+ max_attempts: int = 15,
**kwargs,
):
super().__init__(**kwargs)
self.cluster_identifier = cluster_identifier
self.aws_conn_id = aws_conn_id
self.deferrable = deferrable
+ self.max_attempts = max_attempts
self.poll_interval = poll_interval
- # These parameters are added to address an issue with the boto3 API where the API
+ # These parameters are used to address an issue with the boto3 API where the API
# prematurely reports the cluster as available to receive requests. This causes the cluster
# to reject initial attempts to pause the cluster despite reporting the correct state.
self._attempts = 10
self._attempt_interval = 15
def execute(self, context: Context):
redshift_hook = RedshiftHook(aws_conn_id=self.aws_conn_id)
+ while self._attempts >= 1:
+ try:
+ redshift_hook.get_conn().pause_cluster(ClusterIdentifier=self.cluster_identifier)
+ break
+ except redshift_hook.get_conn().exceptions.InvalidClusterStateFault as error:
+ self._attempts = self._attempts - 1
+ if self._attempts > 0:
+ self.log.error("Unable to pause cluster. %d attempts remaining.", self._attempts)
+ time.sleep(self._attempt_interval)
+ else:
+ raise error
if self.deferrable:
self.defer(
- timeout=self.execution_timeout,
- trigger=RedshiftClusterTrigger(
- task_id=self.task_id,
+ trigger=RedshiftPauseClusterTrigger(
+ cluster_identifier=self.cluster_identifier,
poll_interval=self.poll_interval,
+ max_attempts=self.max_attempts,
Review Comment:
alright, that sounds like a fair plan. I'll make the changes
--
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] dstandish commented on a diff in pull request #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #30853:
URL: https://github.com/apache/airflow/pull/30853#discussion_r1218448419
##########
airflow/providers/amazon/aws/operators/redshift_cluster.py:
##########
@@ -524,64 +527,52 @@ def __init__(
aws_conn_id: str = "aws_default",
deferrable: bool = False,
poll_interval: int = 10,
+ max_attempts: int = 15,
**kwargs,
):
super().__init__(**kwargs)
self.cluster_identifier = cluster_identifier
self.aws_conn_id = aws_conn_id
self.deferrable = deferrable
+ self.max_attempts = max_attempts
self.poll_interval = poll_interval
- # These parameters are added to address an issue with the boto3 API where the API
+ # These parameters are used to address an issue with the boto3 API where the API
# prematurely reports the cluster as available to receive requests. This causes the cluster
# to reject initial attempts to pause the cluster despite reporting the correct state.
self._attempts = 10
self._attempt_interval = 15
def execute(self, context: Context):
redshift_hook = RedshiftHook(aws_conn_id=self.aws_conn_id)
+ while self._attempts >= 1:
+ try:
+ redshift_hook.get_conn().pause_cluster(ClusterIdentifier=self.cluster_identifier)
+ break
+ except redshift_hook.get_conn().exceptions.InvalidClusterStateFault as error:
+ self._attempts = self._attempts - 1
+ if self._attempts > 0:
+ self.log.error("Unable to pause cluster. %d attempts remaining.", self._attempts)
+ time.sleep(self._attempt_interval)
+ else:
+ raise error
if self.deferrable:
self.defer(
- timeout=self.execution_timeout,
- trigger=RedshiftClusterTrigger(
- task_id=self.task_id,
+ trigger=RedshiftPauseClusterTrigger(
+ cluster_identifier=self.cluster_identifier,
poll_interval=self.poll_interval,
+ max_attempts=self.max_attempts,
Review Comment:
I see what you mean...
Your concern is that after `[2023-05-08, 02:20:15 UTC]` there's no message saying "trigger timeout"
Perhaps that's something we should fix.
--
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 #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #30853:
URL: https://github.com/apache/airflow/pull/30853#discussion_r1185359710
##########
airflow/providers/amazon/aws/triggers/redshift_cluster.py:
##########
@@ -137,3 +140,70 @@ async def run(self):
},
)
yield TriggerEvent({"status": "success", "message": "Cluster Created"})
+
+
+class RedshiftPauseClusterTrigger(BaseTrigger):
+ """
+ Trigger for RedshiftPauseClusterOperator.
+ The trigger will asynchronously poll the boto3 API and wait for the
+ Redshift cluster to be in the `paused` state.
+
+ :param cluster_identifier: A unique identifier for the cluster.
+ :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_identifier: str,
+ poll_interval: int,
+ max_attempts: int,
+ aws_conn_id: str,
+ ):
+ self.cluster_identifier = cluster_identifier
+ 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 (
+ "airflow.providers.amazon.aws.triggers.redshift_cluster.RedshiftPauseClusterTrigger",
+ {
+ "cluster_identifier": str(self.cluster_identifier),
+ "poll_interval": str(self.poll_interval),
+ "max_attempts": str(self.max_attempts),
+ "aws_conn_id": str(self.aws_conn_id),
+ },
+ )
+
+ @cached_property
+ def hook(self) -> RedshiftHook:
+ return RedshiftHook(aws_conn_id=self.aws_conn_id)
+
+ async def run(self):
+ async with self.hook.async_conn as client:
+ attempt = 0
+ while attempt < int(self.max_attempts):
+ attempt = attempt + 1
+ try:
+ waiter = self.hook.get_waiter("cluster_paused", deferrable=True, client=client)
+ await waiter.wait(
+ ClusterIdentifier=self.cluster_identifier,
+ WaiterConfig={
+ "Delay": int(self.poll_interval),
+ "MaxAttempts": 1,
Review Comment:
Yes unfortunately we do lose the aspect of retrying due to the logging issue. Until boto supports logging internally, I can't see any way of logging during the polling period. However, we still get to use the biggest benefit of using these waiters, which is not having to write custom async code that will make the API calls - this saves us from having to write a lot of additional code that would, for the most part, be extremely redundant.
> With this, will it have an additional overhead of getting the waiter for each attempt and launching it?
I hadn't noticed any performance issues with this, but that's a good point. I moved the `get_waiter` call to outside of the while loop, so it should only get created once now. Add a new test 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] dstandish commented on a diff in pull request #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #30853:
URL: https://github.com/apache/airflow/pull/30853#discussion_r1185794972
##########
airflow/providers/amazon/aws/operators/redshift_cluster.py:
##########
@@ -524,64 +527,52 @@ def __init__(
aws_conn_id: str = "aws_default",
deferrable: bool = False,
poll_interval: int = 10,
+ max_attempts: int = 15,
**kwargs,
):
super().__init__(**kwargs)
self.cluster_identifier = cluster_identifier
self.aws_conn_id = aws_conn_id
self.deferrable = deferrable
+ self.max_attempts = max_attempts
self.poll_interval = poll_interval
- # These parameters are added to address an issue with the boto3 API where the API
+ # These parameters are used to address an issue with the boto3 API where the API
# prematurely reports the cluster as available to receive requests. This causes the cluster
# to reject initial attempts to pause the cluster despite reporting the correct state.
self._attempts = 10
self._attempt_interval = 15
def execute(self, context: Context):
redshift_hook = RedshiftHook(aws_conn_id=self.aws_conn_id)
+ while self._attempts >= 1:
+ try:
+ redshift_hook.get_conn().pause_cluster(ClusterIdentifier=self.cluster_identifier)
+ break
+ except redshift_hook.get_conn().exceptions.InvalidClusterStateFault as error:
+ self._attempts = self._attempts - 1
+ if self._attempts > 0:
+ self.log.error("Unable to pause cluster. %d attempts remaining.", self._attempts)
+ time.sleep(self._attempt_interval)
+ else:
+ raise error
if self.deferrable:
self.defer(
- timeout=self.execution_timeout,
- trigger=RedshiftClusterTrigger(
- task_id=self.task_id,
+ trigger=RedshiftPauseClusterTrigger(
+ cluster_identifier=self.cluster_identifier,
poll_interval=self.poll_interval,
+ max_attempts=self.max_attempts,
Review Comment:
so, with triggers, there's already a timeout. why don't we continue to use that? though it shouldn't be execution_timeout -- that's the timeout for the task itself. but you could calculate it from max_attempts*poll interval. and then your trigger doesn't need a max_attempts param it can just loop until it times out.
one problem with using max_attempts in trigger is if trigger is killed and is respawned it will reset -- not so if we use timeout.
##########
airflow/providers/amazon/aws/operators/redshift_cluster.py:
##########
@@ -520,64 +523,52 @@ def __init__(
aws_conn_id: str = "aws_default",
deferrable: bool = False,
poll_interval: int = 10,
+ max_attempts: int = 15,
**kwargs,
):
super().__init__(**kwargs)
self.cluster_identifier = cluster_identifier
self.aws_conn_id = aws_conn_id
self.deferrable = deferrable
+ self.max_attempts = max_attempts
self.poll_interval = poll_interval
- # These parameters are added to address an issue with the boto3 API where the API
+ # These parameters are used to address an issue with the boto3 API where the API
# prematurely reports the cluster as available to receive requests. This causes the cluster
# to reject initial attempts to pause the cluster despite reporting the correct state.
self._attempts = 10
self._attempt_interval = 15
def execute(self, context: Context):
redshift_hook = RedshiftHook(aws_conn_id=self.aws_conn_id)
+ while self._attempts >= 1:
Review Comment:
```suggestion
while self._remaining_attempts >= 1:
```
--
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 #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #30853:
URL: https://github.com/apache/airflow/pull/30853#discussion_r1189099941
##########
airflow/providers/amazon/aws/operators/redshift_cluster.py:
##########
@@ -524,64 +528,55 @@ def __init__(
aws_conn_id: str = "aws_default",
deferrable: bool = False,
poll_interval: int = 10,
+ max_attempts: int = 15,
**kwargs,
):
super().__init__(**kwargs)
self.cluster_identifier = cluster_identifier
self.aws_conn_id = aws_conn_id
self.deferrable = deferrable
+ self.max_attempts = max_attempts
self.poll_interval = poll_interval
- # These parameters are added to address an issue with the boto3 API where the API
+ # These parameters are used to address an issue with the boto3 API where the API
# prematurely reports the cluster as available to receive requests. This causes the cluster
# to reject initial attempts to pause the cluster despite reporting the correct state.
- self._attempts = 10
+ self._remaining_attempts = 10
Review Comment:
Yea the 2 are separate parameters. That's why there's a comment above describing what they are for, and why they start with a `_`. There's not too much that can be done about that.
--
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 #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "vandonr-amz (via GitHub)" <gi...@apache.org>.
vandonr-amz commented on code in PR #30853:
URL: https://github.com/apache/airflow/pull/30853#discussion_r1189020150
##########
airflow/providers/amazon/aws/waiters/redshift.json:
##########
@@ -0,0 +1,30 @@
+{
+ "version": 2,
+ "waiters": {
+ "cluster_paused": {
+ "operation": "DescribeClusters",
+ "delay": 30,
+ "maxAttempts": 60,
+ "acceptors": [
+ {
+ "matcher": "pathAll",
+ "argument": "Clusters[].ClusterStatus",
+ "expected": "paused",
+ "state": "success"
+ },
+ {
+ "expected": "ClusterNotFound",
+ "argument": "Clusters[].ClusterStatus",
+ "matcher": "error",
+ "state": "retry"
+ },
+ {
+ "expected": "deleting",
+ "matcher": "pathAny",
+ "state": "failure",
+ "argument": "Clusters[].ClusterStatus"
Review Comment:
can you organize the fields so that they are in the same order each time ? It would make it easier to read the config.
--
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] pankajkoti commented on a diff in pull request #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "pankajkoti (via GitHub)" <gi...@apache.org>.
pankajkoti commented on code in PR #30853:
URL: https://github.com/apache/airflow/pull/30853#discussion_r1184756193
##########
airflow/providers/amazon/aws/triggers/redshift_cluster.py:
##########
@@ -137,3 +140,70 @@ async def run(self):
},
)
yield TriggerEvent({"status": "success", "message": "Cluster Created"})
+
+
+class RedshiftPauseClusterTrigger(BaseTrigger):
+ """
+ Trigger for RedshiftPauseClusterOperator.
+ The trigger will asynchronously poll the boto3 API and wait for the
+ Redshift cluster to be in the `paused` state.
+
+ :param cluster_identifier: A unique identifier for the cluster.
+ :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_identifier: str,
+ poll_interval: int,
+ max_attempts: int,
+ aws_conn_id: str,
+ ):
+ self.cluster_identifier = cluster_identifier
+ 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 (
+ "airflow.providers.amazon.aws.triggers.redshift_cluster.RedshiftPauseClusterTrigger",
+ {
+ "cluster_identifier": str(self.cluster_identifier),
+ "poll_interval": str(self.poll_interval),
+ "max_attempts": str(self.max_attempts),
+ "aws_conn_id": str(self.aws_conn_id),
+ },
+ )
+
+ @cached_property
+ def hook(self) -> RedshiftHook:
+ return RedshiftHook(aws_conn_id=self.aws_conn_id)
+
+ async def run(self):
+ async with self.hook.async_conn as client:
+ attempt = 0
+ while attempt < int(self.max_attempts):
+ attempt = attempt + 1
+ try:
+ waiter = self.hook.get_waiter("cluster_paused", deferrable=True, client=client)
+ await waiter.wait(
+ ClusterIdentifier=self.cluster_identifier,
+ WaiterConfig={
+ "Delay": int(self.poll_interval),
+ "MaxAttempts": 1,
Review Comment:
I get the hack and looks like a good workaround for now, but then don't we lose the purpose of using waiters if we're setting `MaxAttempts` to 1? With this, will it have an additional overhead of getting the waiter for each attempt and launching it?
##########
airflow/providers/amazon/aws/triggers/redshift_cluster.py:
##########
@@ -137,3 +140,70 @@ async def run(self):
},
)
yield TriggerEvent({"status": "success", "message": "Cluster Created"})
+
+
+class RedshiftPauseClusterTrigger(BaseTrigger):
+ """
+ Trigger for RedshiftPauseClusterOperator.
+ The trigger will asynchronously poll the boto3 API and wait for the
+ Redshift cluster to be in the `paused` state.
+
+ :param cluster_identifier: A unique identifier for the cluster.
+ :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_identifier: str,
+ poll_interval: int,
+ max_attempts: int,
+ aws_conn_id: str,
+ ):
+ self.cluster_identifier = cluster_identifier
+ 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 (
+ "airflow.providers.amazon.aws.triggers.redshift_cluster.RedshiftPauseClusterTrigger",
+ {
+ "cluster_identifier": str(self.cluster_identifier),
+ "poll_interval": str(self.poll_interval),
+ "max_attempts": str(self.max_attempts),
+ "aws_conn_id": str(self.aws_conn_id),
+ },
+ )
+
+ @cached_property
+ def hook(self) -> RedshiftHook:
+ return RedshiftHook(aws_conn_id=self.aws_conn_id)
+
+ async def run(self):
+ async with self.hook.async_conn as client:
+ attempt = 0
+ while attempt < int(self.max_attempts):
+ attempt = attempt + 1
+ try:
+ waiter = self.hook.get_waiter("cluster_paused", deferrable=True, client=client)
+ await waiter.wait(
+ ClusterIdentifier=self.cluster_identifier,
+ WaiterConfig={
+ "Delay": int(self.poll_interval),
+ "MaxAttempts": 1,
+ },
+ )
+ break
+ except WaiterError as error:
+ self.log.info(
+ "Status of cluster is %s", error.last_response["Clusters"][0]["ClusterStatus"]
+ )
+ await asyncio.sleep(int(self.poll_interval))
Review Comment:
Shouldn't we re-raise the `WaiterError` here? If the waiter reaches a failure path, it would be good if we don't wait for all the attempts to finish.
--
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] potiuk commented on pull request #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #30853:
URL: https://github.com/apache/airflow/pull/30853#issuecomment-1526764681
FYI. Airflow 2.6 implements interleaving Tirgger and Task logs together. So for those users it will be no problem, they will se trigger logs coming.
--
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] pankajastro commented on pull request #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "pankajastro (via GitHub)" <gi...@apache.org>.
pankajastro commented on PR #30853:
URL: https://github.com/apache/airflow/pull/30853#issuecomment-1522180851
> > Thanks for the review! Yes the botocore waiters don't log any messages as they are polling the AWS service. Because this is implemented in the botocore code, in order to do this, we'd have to make the changes there. This is something that we've considered, and may end up implementing - unless the botocore team has a reason for not doing this by default. I imagine others have raised a similar concern,
>
> This will be a problem for our current users who are seeing log messages on the Triggerer vs no logs post this PR.
I agree with @phanikumv logs are important here.
--
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 #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #30853:
URL: https://github.com/apache/airflow/pull/30853#discussion_r1185423527
##########
airflow/providers/amazon/aws/triggers/redshift_cluster.py:
##########
@@ -137,3 +140,70 @@ async def run(self):
},
)
yield TriggerEvent({"status": "success", "message": "Cluster Created"})
+
+
+class RedshiftPauseClusterTrigger(BaseTrigger):
+ """
+ Trigger for RedshiftPauseClusterOperator.
+ The trigger will asynchronously poll the boto3 API and wait for the
+ Redshift cluster to be in the `paused` state.
+
+ :param cluster_identifier: A unique identifier for the cluster.
+ :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_identifier: str,
+ poll_interval: int,
+ max_attempts: int,
+ aws_conn_id: str,
+ ):
+ self.cluster_identifier = cluster_identifier
+ 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 (
+ "airflow.providers.amazon.aws.triggers.redshift_cluster.RedshiftPauseClusterTrigger",
+ {
+ "cluster_identifier": str(self.cluster_identifier),
+ "poll_interval": str(self.poll_interval),
+ "max_attempts": str(self.max_attempts),
+ "aws_conn_id": str(self.aws_conn_id),
+ },
+ )
+
+ @cached_property
+ def hook(self) -> RedshiftHook:
+ return RedshiftHook(aws_conn_id=self.aws_conn_id)
+
+ async def run(self):
+ async with self.hook.async_conn as client:
+ attempt = 0
+ while attempt < int(self.max_attempts):
+ attempt = attempt + 1
+ try:
+ waiter = self.hook.get_waiter("cluster_paused", deferrable=True, client=client)
Review Comment:
I'm not sure what you mean here. This is a custom boto waiter, for which I've created a JSON file for.
--
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 #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #30853:
URL: https://github.com/apache/airflow/pull/30853#discussion_r1192948999
##########
airflow/providers/amazon/aws/triggers/redshift_cluster.py:
##########
@@ -137,3 +140,73 @@ async def run(self):
},
)
yield TriggerEvent({"status": "success", "message": "Cluster Created"})
+
+
+class RedshiftPauseClusterTrigger(BaseTrigger):
+ """
+ Trigger for RedshiftPauseClusterOperator.
+ The trigger will asynchronously poll the boto3 API and wait for the
+ Redshift cluster to be in the `paused` state.
+
+ :param cluster_identifier: A unique identifier for the cluster.
+ :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_identifier: str,
+ poll_interval: int,
+ max_attempts: int,
+ aws_conn_id: str,
+ ):
+ self.cluster_identifier = cluster_identifier
+ 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 (
+ "airflow.providers.amazon.aws.triggers.redshift_cluster.RedshiftPauseClusterTrigger",
+ {
+ "cluster_identifier": self.cluster_identifier,
+ "poll_interval": str(self.poll_interval),
+ "max_attempts": str(self.max_attempts),
+ "aws_conn_id": self.aws_conn_id,
+ },
+ )
+
+ @cached_property
+ def hook(self) -> RedshiftHook:
+ return RedshiftHook(aws_conn_id=self.aws_conn_id)
+
+ async def run(self):
+ async with self.hook.async_conn as client:
+ attempt = 0
+ waiter = self.hook.get_waiter("cluster_paused", deferrable=True, client=client)
+ while attempt < int(self.max_attempts):
+ attempt = attempt + 1
+ try:
+ await waiter.wait(
+ ClusterIdentifier=self.cluster_identifier,
+ 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"Pause Cluster Failed: {error}"})
+ break
+ self.log.info(
+ "Status of cluster is %s", error.last_response["Clusters"][0]["ClusterStatus"]
+ )
+ await asyncio.sleep(int(self.poll_interval))
+ if attempt >= int(self.max_attempts):
Review Comment:
> Logically, I am wondering if attempt would ever be >self.max_attempts. Should the check here be just
if attempt == int(self.max_attempts)?
Yea it should never go greater than `self.max_attempts` but I thought it didn't hurt to include that in the check. Consider if future-proofing :p
>Also, why do we need to cast self.max_attempts to int? Isn't the type int already as per the type annotation?
This is due to serialization which converts all the `int` params to `str`, which then cause issues when a `str` is compared to an `int`. This way, the `str` is converted to `int` before being compared.
```
TypeError: '<' not supported between instances of 'int' and 'str'
```
--
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 #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "vandonr-amz (via GitHub)" <gi...@apache.org>.
vandonr-amz commented on code in PR #30853:
URL: https://github.com/apache/airflow/pull/30853#discussion_r1189023318
##########
tests/providers/amazon/aws/operators/test_redshift_cluster.py:
##########
@@ -389,9 +392,10 @@ def test_pause_cluster_multiple_attempts_fail(self, mock_sleep, mock_conn):
redshift_operator.execute(None)
assert mock_conn.pause_cluster.call_count == 10
- def test_pause_cluster_deferrable_mode(self):
+ @mock.patch("airflow.providers.amazon.aws.hooks.redshift_cluster.RedshiftHook.get_conn")
Review Comment:
```suggestion
@mock.patch.object(RedshiftHook, "get_conn")
```
I feel this syntax is less error-prone, as only the method name needs to be an arbitrary string.
Also, it lets me easily navigate to the hook in my IDE.
--
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] pankajkoti commented on a diff in pull request #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "pankajkoti (via GitHub)" <gi...@apache.org>.
pankajkoti commented on code in PR #30853:
URL: https://github.com/apache/airflow/pull/30853#discussion_r1190245323
##########
airflow/providers/amazon/aws/triggers/redshift_cluster.py:
##########
@@ -137,3 +140,73 @@ async def run(self):
},
)
yield TriggerEvent({"status": "success", "message": "Cluster Created"})
+
+
+class RedshiftPauseClusterTrigger(BaseTrigger):
+ """
+ Trigger for RedshiftPauseClusterOperator.
+ The trigger will asynchronously poll the boto3 API and wait for the
+ Redshift cluster to be in the `paused` state.
+
+ :param cluster_identifier: A unique identifier for the cluster.
+ :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_identifier: str,
+ poll_interval: int,
+ max_attempts: int,
+ aws_conn_id: str,
+ ):
+ self.cluster_identifier = cluster_identifier
+ 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 (
+ "airflow.providers.amazon.aws.triggers.redshift_cluster.RedshiftPauseClusterTrigger",
+ {
+ "cluster_identifier": self.cluster_identifier,
+ "poll_interval": str(self.poll_interval),
+ "max_attempts": str(self.max_attempts),
+ "aws_conn_id": self.aws_conn_id,
+ },
+ )
+
+ @cached_property
+ def hook(self) -> RedshiftHook:
+ return RedshiftHook(aws_conn_id=self.aws_conn_id)
+
+ async def run(self):
+ async with self.hook.async_conn as client:
+ attempt = 0
+ waiter = self.hook.get_waiter("cluster_paused", deferrable=True, client=client)
+ while attempt < int(self.max_attempts):
+ attempt = attempt + 1
+ try:
+ await waiter.wait(
+ ClusterIdentifier=self.cluster_identifier,
+ 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"Pause Cluster Failed: {error}"})
+ break
+ self.log.info(
+ "Status of cluster is %s", error.last_response["Clusters"][0]["ClusterStatus"]
+ )
+ await asyncio.sleep(int(self.poll_interval))
+ if attempt >= int(self.max_attempts):
Review Comment:
Also, why do we need to cast `self.max_attempts` to `int`? Isn't the type `int` already as per the type annotation?
--
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 #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #30853:
URL: https://github.com/apache/airflow/pull/30853#discussion_r1185413059
##########
airflow/providers/amazon/aws/triggers/redshift_cluster.py:
##########
@@ -137,3 +140,70 @@ async def run(self):
},
)
yield TriggerEvent({"status": "success", "message": "Cluster Created"})
+
+
+class RedshiftPauseClusterTrigger(BaseTrigger):
+ """
+ Trigger for RedshiftPauseClusterOperator.
+ The trigger will asynchronously poll the boto3 API and wait for the
+ Redshift cluster to be in the `paused` state.
+
+ :param cluster_identifier: A unique identifier for the cluster.
+ :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_identifier: str,
+ poll_interval: int,
+ max_attempts: int,
+ aws_conn_id: str,
+ ):
+ self.cluster_identifier = cluster_identifier
+ 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 (
+ "airflow.providers.amazon.aws.triggers.redshift_cluster.RedshiftPauseClusterTrigger",
+ {
+ "cluster_identifier": str(self.cluster_identifier),
+ "poll_interval": str(self.poll_interval),
+ "max_attempts": str(self.max_attempts),
+ "aws_conn_id": str(self.aws_conn_id),
+ },
+ )
+
+ @cached_property
+ def hook(self) -> RedshiftHook:
+ return RedshiftHook(aws_conn_id=self.aws_conn_id)
+
+ async def run(self):
+ async with self.hook.async_conn as client:
+ attempt = 0
+ while attempt < int(self.max_attempts):
+ attempt = attempt + 1
+ try:
+ waiter = self.hook.get_waiter("cluster_paused", deferrable=True, client=client)
+ await waiter.wait(
+ ClusterIdentifier=self.cluster_identifier,
+ WaiterConfig={
+ "Delay": int(self.poll_interval),
+ "MaxAttempts": 1,
+ },
+ )
+ break
+ except WaiterError as error:
+ self.log.info(
+ "Status of cluster is %s", error.last_response["Clusters"][0]["ClusterStatus"]
+ )
+ await asyncio.sleep(int(self.poll_interval))
+ if attempt >= int(self.max_attempts):
+ yield TriggerEvent(
+ {"status": "failure", "message": "Resume Cluster Failed - max attempts reached."}
+ )
+ else:
Review Comment:
we could, but it would be redundant. There are 2 ways of breaking out of the `while` loop:
1) the waiter does not raise a `WaiterError`, which can only happen if the success criteria is met. the `break` statement will take us out of the loop
2) The number of attempts exceeded `self.max_attempts`
In the second case, we are marking it as a failure, which only leaves the success option.
--
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 pull request #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on PR #30853:
URL: https://github.com/apache/airflow/pull/30853#issuecomment-1533857104
Thanks @potiuk for the idea about handling the error from the waiter to log a status message! I've been playing around that for a bit, and I have a solution that produces the following logs for this operator:
```
[2023-05-03, 22:53:07 UTC] {base.py:73} INFO - Using connection ID 'aws_default' for task execution.
[2023-05-03, 22:53:07 UTC] {credentials.py:617} INFO - Found credentials in shared credentials file: ~/.aws/credentials
[2023-05-03, 22:53:15 UTC] {redshift_cluster.py:200} INFO - Status of cluster is pausing
[2023-05-03, 22:53:53 UTC] {redshift_cluster.py:200} INFO - Status of cluster is pausing
[2023-05-03, 22:54:32 UTC] {redshift_cluster.py:200} INFO - Status of cluster is pausing
[2023-05-03, 22:55:10 UTC] {redshift_cluster.py:200} INFO - Status of cluster is pausing
[2023-05-03, 22:55:48 UTC] {triggerer_job_runner.py:615} INFO - Trigger example_redshift/manual__2023-05-03T22:44:17.762920+00:00/pause_cluster/-1/1 (ID 2) fired: TriggerEvent<{'status': 'success', 'message': 'Cluster paused'}>
```
This will need to be implemented on all boto waiters if we want them to log messages during the polling period, until boto waiters support logging natively.
I've opened an [issue](https://github.com/boto/botocore/issues/2923) with botocore requesting a logging feature. We'll have to see if/when they choose to add it.
--
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] dstandish commented on a diff in pull request #30853: Change Deferrable implementation for RedshiftPauseClusterOperator to follow standard
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #30853:
URL: https://github.com/apache/airflow/pull/30853#discussion_r1186374086
##########
airflow/providers/amazon/aws/operators/redshift_cluster.py:
##########
@@ -524,64 +527,52 @@ def __init__(
aws_conn_id: str = "aws_default",
deferrable: bool = False,
poll_interval: int = 10,
+ max_attempts: int = 15,
**kwargs,
):
super().__init__(**kwargs)
self.cluster_identifier = cluster_identifier
self.aws_conn_id = aws_conn_id
self.deferrable = deferrable
+ self.max_attempts = max_attempts
self.poll_interval = poll_interval
- # These parameters are added to address an issue with the boto3 API where the API
+ # These parameters are used to address an issue with the boto3 API where the API
# prematurely reports the cluster as available to receive requests. This causes the cluster
# to reject initial attempts to pause the cluster despite reporting the correct state.
self._attempts = 10
self._attempt_interval = 15
def execute(self, context: Context):
redshift_hook = RedshiftHook(aws_conn_id=self.aws_conn_id)
+ while self._attempts >= 1:
+ try:
+ redshift_hook.get_conn().pause_cluster(ClusterIdentifier=self.cluster_identifier)
+ break
+ except redshift_hook.get_conn().exceptions.InvalidClusterStateFault as error:
+ self._attempts = self._attempts - 1
+ if self._attempts > 0:
+ self.log.error("Unable to pause cluster. %d attempts remaining.", self._attempts)
+ time.sleep(self._attempt_interval)
+ else:
+ raise error
if self.deferrable:
self.defer(
- timeout=self.execution_timeout,
- trigger=RedshiftClusterTrigger(
- task_id=self.task_id,
+ trigger=RedshiftPauseClusterTrigger(
+ cluster_identifier=self.cluster_identifier,
poll_interval=self.poll_interval,
+ max_attempts=self.max_attempts,
Review Comment:
I don't see how that wolud address the issue. If it's part of the kwargs, and trigger killed, when new trigger spawned, same kwargs, same behavior. Best to use the timeout feature for triggers.
--
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