vandonr-amz commented on code in PR #30463: URL: https://github.com/apache/airflow/pull/30463#discussion_r1235521149
########## airflow/providers/amazon/aws/utils/waiter_with_logging.py: ########## @@ -0,0 +1,90 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +from __future__ import annotations + +import logging +import time + +import jmespath +from botocore.exceptions import WaiterError +from botocore.waiter import Waiter + +from airflow.exceptions import AirflowException + + +def wait( + waiter: Waiter, + waiter_delay: int, + max_attempts: int, + state_args: dict, + failure_message: str, + status_message: dict, Review Comment: at first I didn't realize this wasn't just a message... rather than using a dict that's always going to contain the same keys, why not extend this to 2 parameters ? `status_message` could be just the message, like its name suggests, and then have an other param for `status_path` or something. ########## airflow/providers/amazon/aws/utils/waiter_with_logging.py: ########## @@ -0,0 +1,90 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +from __future__ import annotations + +import logging +import time + +import jmespath +from botocore.exceptions import WaiterError +from botocore.waiter import Waiter + +from airflow.exceptions import AirflowException + + +def wait( + waiter: Waiter, + waiter_delay: int, + max_attempts: int, + state_args: dict, + failure_message: str, + status_message: dict, +): + """ + Use a boto waiter to poll an AWS service for the specified state. Although this function + uses boto waiters to poll the state of the service, it logs the response of the service + after every attempt, which is not currently supported by boto waiters. + + :param waiter: The boto waiter to use. + :param waiter_delay: The amount of time in seconds to wait between attempts. + :param max_attempts: The maximum number of attempts to be made. + :param state_args: The arguments to pass to the waiter. + :param failure_message: The message to log if a failure state is reached. + :param status_message: A dict that specifies the status message, as well as + how to retrieve that message from the waiter's response, using jmespath syntax. + e.g. + status_message = { + "message": "Waiting for cluster to reach a desired state", + "args": ["Cluster.Status"] + } + For responses that contain a list, + status_message = { + "message": "Waiting for cluster to reach a desired state", + "args": ["Clusters[0].Status"] + } + """ + log = logging.getLogger(__name__) + attempt = 0 + while True: + attempt += 1 + try: + waiter.wait(**state_args, WaiterConfig={"MaxAttempts": 1}) + break + except WaiterError as error: + if "terminal failure" in str(error): + raise AirflowException(f"{failure_message}: {error}") + status_string = _format_status_string(status_message["args"], error.last_response) + log.info("%s: %s", status_message["message"], status_string) Review Comment: thinking about this, it's a bunch of useless work if info logs are disabled. I think a way to evaluate it only if needed would be to create a class to hold the args & response, and that'd extract the status & compute the join in its `__str__` method. see https://stackoverflow.com/questions/4148790/lazy-logger-message-string-evaluation -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
