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/04 18:03:18 UTC
[GitHub] [airflow] syedahsn opened a new pull request, #30463: Add custom waiters to EMR Serverless
syedahsn opened a new pull request, #30463:
URL: https://github.com/apache/airflow/pull/30463
This PR replaces the manual waiter with the custom boto waiters. It also adds extra features to the `EmrServerlessStartJobOperator`:
- Prints a log message when the job is running.
- Prints the reason for a job failure if a job fails
By removing the waiters, the unit tests needed to be updated, and in some cases modified.
<!--
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] syedahsn commented on a diff in pull request #30463: Add custom waiters to EMR Serverless
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #30463:
URL: https://github.com/apache/airflow/pull/30463#discussion_r1164719523
##########
airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -992,21 +1036,40 @@ def execute(self, context: Context) -> dict:
self.log.info("EMR serverless job started: %s", response["jobRunId"])
if self.wait_for_completion:
- # This should be replaced with a boto waiter when available.
- waiter(
- get_state_callable=self.hook.conn.get_job_run,
- get_state_args={
- "applicationId": self.application_id,
- "jobRunId": response["jobRunId"],
- },
- parse_response=["jobRun", "state"],
- desired_state=EmrServerlessHook.JOB_SUCCESS_STATES,
- failure_states=EmrServerlessHook.JOB_FAILURE_STATES,
- object_type="job",
- action="run",
- countdown=self.waiter_countdown,
- check_interval_seconds=self.waiter_check_interval_seconds,
- )
+ try:
+ self.hook.get_waiter("serverless_job_running").wait(
+ applicationId=self.application_id,
+ jobRunId=response["jobRunId"],
+ WaiterConfig=prune_dict(
+ {
+ "Delay": self.waiter_delay,
+ "MaxAttempts": self.waiter_max_attempts,
+ }
+ ),
+ )
+ self.log.info("EMR serverless job is running: %s", response["jobRunId"])
+ self.hook.get_waiter("serverless_job_completed").wait(
+ applicationId=self.application_id,
+ jobRunId=response["jobRunId"],
+ WaiterConfig=prune_dict(
+ {
+ "Delay": self.waiter_delay,
+ "MaxAttempts": self.waiter_max_attempts,
+ }
+ ),
+ )
Review Comment:
The custom boto waiters don't log anything unfortunately. In order to see that a state change has happened, we need to use a separate waiter. In this case, the extra log message is important because it informs the user that the job is actually running.
--
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 #30463: Add custom waiters to EMR Serverless
Posted by "vandonr-amz (via GitHub)" <gi...@apache.org>.
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: 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 #30463: Add custom waiters to EMR Serverless
Posted by "vandonr-amz (via GitHub)" <gi...@apache.org>.
vandonr-amz commented on code in PR #30463:
URL: https://github.com/apache/airflow/pull/30463#discussion_r1228768476
##########
airflow/providers/amazon/aws/utils/waiter_with_logging.py:
##########
@@ -0,0 +1,77 @@
+# 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
+
+from botocore.exceptions import WaiterError
+
+from airflow.exceptions import AirflowException
+
+
+def wait(
+ waiter,
+ waiter_delay,
+ max_attempts,
+ state_args,
+ failure_message,
+ status_message,
+):
+ log = logging.getLogger(__name__)
+ attempt = 0
+ while attempt < max_attempts:
+ attempt += 1
+ try:
+ waiter.wait(
+ **state_args,
+ WaiterConfig={
+ "Delay": 1,
+ "MaxAttempts": 1,
+ },
+ )
+ break
+ except WaiterError as error:
+ if "terminal failure" in str(error):
+ raise AirflowException(f"{failure_message}: {error}")
+ response_data = []
+ for arg in status_message["args"]:
+ response_data.append(get_state(error.last_response, arg))
+
+ log.info(status_message["message"] + " : " + " - ".join(response_data))
+ time.sleep(waiter_delay)
+ if attempt >= max_attempts:
Review Comment:
you're writing the same check twice, wdyt about a `while True` and putting this inside the loop ?
##########
airflow/providers/amazon/aws/utils/waiter_with_logging.py:
##########
@@ -0,0 +1,77 @@
+# 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
+
+from botocore.exceptions import WaiterError
+
+from airflow.exceptions import AirflowException
+
+
+def wait(
+ waiter,
+ waiter_delay,
+ max_attempts,
+ state_args,
+ failure_message,
+ status_message,
+):
+ log = logging.getLogger(__name__)
+ attempt = 0
+ while attempt < max_attempts:
+ attempt += 1
+ try:
+ waiter.wait(
+ **state_args,
+ WaiterConfig={
+ "Delay": 1,
Review Comment:
Can we omit this parameter ? If not, maybe add a comment explaining that it's not meaningful here.
##########
airflow/providers/amazon/aws/utils/waiter_with_logging.py:
##########
@@ -0,0 +1,77 @@
+# 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
+
+from botocore.exceptions import WaiterError
+
+from airflow.exceptions import AirflowException
+
+
+def wait(
+ waiter,
+ waiter_delay,
+ max_attempts,
+ state_args,
+ failure_message,
+ status_message,
+):
+ log = logging.getLogger(__name__)
+ attempt = 0
+ while attempt < max_attempts:
+ attempt += 1
+ try:
+ waiter.wait(
+ **state_args,
+ WaiterConfig={
+ "Delay": 1,
+ "MaxAttempts": 1,
+ },
+ )
+ break
+ except WaiterError as error:
+ if "terminal failure" in str(error):
+ raise AirflowException(f"{failure_message}: {error}")
+ response_data = []
+ for arg in status_message["args"]:
+ response_data.append(get_state(error.last_response, arg))
+
+ log.info(status_message["message"] + " : " + " - ".join(response_data))
Review Comment:
you shouldn't concatenate strings for logs, use `%s` formatting instead, so that if users disable info logs, they don't pay the cost of string manipulation.
(you could even do away without the `join`, at the expense of readability, I'll let you judge)
##########
airflow/providers/amazon/aws/utils/waiter_with_logging.py:
##########
@@ -0,0 +1,77 @@
+# 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
+
+from botocore.exceptions import WaiterError
+
+from airflow.exceptions import AirflowException
+
+
+def wait(
Review Comment:
I disagree with you, and I don't think the question is whether it's _possible_ to do this with tenacity, but rather if using tenacity improves readability and maintainability.
If I have to read through pages of tenacity documentation to understand what's going on, it's not really a win even if it reduces the number of lines of code.
Tenacity is intended to retry things like network errors & such, using it here as an alternative `while` loop would break the [principle of least surprise](https://en.wikipedia.org/wiki/Principle_of_least_astonishment) for readers of this code.
##########
airflow/providers/amazon/aws/utils/waiter_with_logging.py:
##########
@@ -0,0 +1,77 @@
+# 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
+
+from botocore.exceptions import WaiterError
+
+from airflow.exceptions import AirflowException
+
+
+def wait(
+ waiter,
+ waiter_delay,
+ max_attempts,
+ state_args,
+ failure_message,
+ status_message,
Review Comment:
would be nice to have type annotations here
##########
airflow/providers/amazon/aws/utils/waiter_with_logging.py:
##########
@@ -0,0 +1,77 @@
+# 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
+
+from botocore.exceptions import WaiterError
+
+from airflow.exceptions import AirflowException
+
+
+def wait(
+ waiter,
+ waiter_delay,
+ max_attempts,
+ state_args,
+ failure_message,
+ status_message,
+):
+ log = logging.getLogger(__name__)
+ attempt = 0
+ while attempt < max_attempts:
+ attempt += 1
+ try:
+ waiter.wait(
+ **state_args,
+ WaiterConfig={
+ "Delay": 1,
+ "MaxAttempts": 1,
+ },
+ )
+ break
+ except WaiterError as error:
+ if "terminal failure" in str(error):
+ raise AirflowException(f"{failure_message}: {error}")
+ response_data = []
+ for arg in status_message["args"]:
+ response_data.append(get_state(error.last_response, arg))
+
+ log.info(status_message["message"] + " : " + " - ".join(response_data))
+ time.sleep(waiter_delay)
+ if attempt >= max_attempts:
+ raise AirflowException("Waiter error: max attempts reached")
+
+
+def get_state(response, keys) -> str:
Review Comment:
We could use a JMESPath expression here, just like we define them in waiters.
This function has relatively complicated logic, and no test, using an external package would make all this vanish ;)
--
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 #30463: Add custom waiters to EMR Serverless
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #30463:
URL: https://github.com/apache/airflow/pull/30463#issuecomment-1518737459
> let's go
> btw, this is going to conflict with https://github.com/apache/airflow/pull/30720 but happy to let you go first
As soon as the docs are building with no errors :)
--
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 #30463: Add custom waiters to EMR Serverless
Posted by "vandonr-amz (via GitHub)" <gi...@apache.org>.
vandonr-amz commented on code in PR #30463:
URL: https://github.com/apache/airflow/pull/30463#discussion_r1234498019
##########
airflow/providers/amazon/aws/utils/waiter_with_logging.py:
##########
@@ -0,0 +1,77 @@
+# 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
+
+from botocore.exceptions import WaiterError
+
+from airflow.exceptions import AirflowException
+
+
+def wait(
+ waiter,
+ waiter_delay,
+ max_attempts,
+ state_args,
+ failure_message,
+ status_message,
+):
+ log = logging.getLogger(__name__)
+ attempt = 0
+ while attempt < max_attempts:
+ attempt += 1
+ try:
+ waiter.wait(
Review Comment:
how are we going to use this in an async context btw ? I suppose we need to have 2 methods 😬
--
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 #30463: Add custom waiters to EMR Serverless
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #30463:
URL: https://github.com/apache/airflow/pull/30463#discussion_r1234566241
##########
airflow/providers/amazon/aws/utils/waiter_with_logging.py:
##########
@@ -0,0 +1,77 @@
+# 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
+
+from botocore.exceptions import WaiterError
+
+from airflow.exceptions import AirflowException
+
+
+def wait(
+ waiter,
+ waiter_delay,
+ max_attempts,
+ state_args,
+ failure_message,
+ status_message,
+):
+ log = logging.getLogger(__name__)
+ attempt = 0
+ while attempt < max_attempts:
+ attempt += 1
+ try:
+ waiter.wait(
+ **state_args,
+ WaiterConfig={
+ "Delay": 1,
+ "MaxAttempts": 1,
+ },
+ )
+ break
+ except WaiterError as error:
+ if "terminal failure" in str(error):
+ raise AirflowException(f"{failure_message}: {error}")
+ response_data = []
+ for arg in status_message["args"]:
+ response_data.append(get_state(error.last_response, arg))
+
+ log.info(status_message["message"] + " : " + " - ".join(response_data))
+ time.sleep(waiter_delay)
+ if attempt >= max_attempts:
+ raise AirflowException("Waiter error: max attempts reached")
+
+
+def get_state(response, keys) -> str:
Review Comment:
This is a really good idea. Makes the whole code a lot nicer.
--
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 #30463: Add custom waiters to EMR Serverless
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #30463:
URL: https://github.com/apache/airflow/pull/30463#discussion_r1234567369
##########
airflow/providers/amazon/aws/utils/waiter_with_logging.py:
##########
@@ -0,0 +1,77 @@
+# 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
+
+from botocore.exceptions import WaiterError
+
+from airflow.exceptions import AirflowException
+
+
+def wait(
+ waiter,
+ waiter_delay,
+ max_attempts,
+ state_args,
+ failure_message,
+ status_message,
+):
+ log = logging.getLogger(__name__)
+ attempt = 0
+ while attempt < max_attempts:
+ attempt += 1
+ try:
+ waiter.wait(
Review Comment:
yea, we need to have 2 separate methods unfortunately. Its still better than not using boto waiters at all (because we would then have to write hook code that would make the actual API calls), but we lose the simplicity of having a single shared wait function for asycn and sync cases :disappointed:
--
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 #30463: Add custom waiters to EMR Serverless
Posted by "vandonr-amz (via GitHub)" <gi...@apache.org>.
vandonr-amz commented on code in PR #30463:
URL: https://github.com/apache/airflow/pull/30463#discussion_r1235605361
##########
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:
it's nothing blocking though, I can do it in a subsequent PR to get this one moving.
--
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 #30463: Add custom waiters to EMR Serverless
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #30463:
URL: https://github.com/apache/airflow/pull/30463#discussion_r1213734899
##########
airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -901,36 +930,63 @@ def execute(self, context: Context) -> str | None:
raise AirflowException(f"Application Creation failed: {response}")
self.log.info("EMR serverless application created: %s", application_id)
-
- # This should be replaced with a boto waiter when available.
- waiter(
- get_state_callable=self.hook.conn.get_application,
- get_state_args={"applicationId": application_id},
- parse_response=["application", "state"],
- desired_state={"CREATED"},
- failure_states=EmrServerlessHook.APPLICATION_FAILURE_STATES,
- object_type="application",
- action="created",
- countdown=self.waiter_countdown,
- check_interval_seconds=self.waiter_check_interval_seconds,
- )
+ attempt = 0
+ waiter = self.hook.get_waiter("serverless_app_created")
+ while attempt < self.waiter_max_attempts:
Review Comment:
No the waiter is not pushing any log messages besides a debug message on completion, which is not really helpful.
--
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 #30463: Add custom waiters to EMR Serverless
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #30463:
URL: https://github.com/apache/airflow/pull/30463#discussion_r1212380235
##########
airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -901,36 +930,63 @@ def execute(self, context: Context) -> str | None:
raise AirflowException(f"Application Creation failed: {response}")
self.log.info("EMR serverless application created: %s", application_id)
-
- # This should be replaced with a boto waiter when available.
- waiter(
- get_state_callable=self.hook.conn.get_application,
- get_state_args={"applicationId": application_id},
- parse_response=["application", "state"],
- desired_state={"CREATED"},
- failure_states=EmrServerlessHook.APPLICATION_FAILURE_STATES,
- object_type="application",
- action="created",
- countdown=self.waiter_countdown,
- check_interval_seconds=self.waiter_check_interval_seconds,
- )
+ attempt = 0
+ waiter = self.hook.get_waiter("serverless_app_created")
+ while attempt < self.waiter_max_attempts:
Review Comment:
I get what you're saying, and this is certainly not an ideal use of the boto waiters. Ideally, the waiters would allow logging natively, and we would end up with a very clean way to wait while also getting information about the current status of the resources. Unfortunately, until botocore supports that, we have to manually work to get logging.
That being said, there is still benefit in using custom boto waiters. We get to make use of botocore's code that allows us to create new waiters without writing additional code. This is especially important with deferrable operators because aiobotocore can use the same config file to generate an async waiter without us having to write async code.
Although we don't get to enjoy the full benefit of using boto waiters, we do gain something, which I think makes it worth doing.
--
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 #30463: Add custom waiters to EMR Serverless
Posted by "vandonr-amz (via GitHub)" <gi...@apache.org>.
vandonr-amz commented on code in PR #30463:
URL: https://github.com/apache/airflow/pull/30463#discussion_r1235635163
##########
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:
I just think having the parameters laid out explicitely is more readable and more intuitive to use than having to refer to the doc each time to remember what magic beans I need to put in that bag.
If you want to reuse those, you can use a kwargs-style dict and pass it with `**EMR_SERVERLESS_APPLICATION_STATUS`.
But it's also the opportunity to write a more descriptive status message each time eventually.
--
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 #30463: Add custom waiters to EMR Serverless
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #30463:
URL: https://github.com/apache/airflow/pull/30463#discussion_r1234566063
##########
airflow/providers/amazon/aws/utils/waiter_with_logging.py:
##########
@@ -0,0 +1,77 @@
+# 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
+
+from botocore.exceptions import WaiterError
+
+from airflow.exceptions import AirflowException
+
+
+def wait(
+ waiter,
+ waiter_delay,
+ max_attempts,
+ state_args,
+ failure_message,
+ status_message,
+):
+ log = logging.getLogger(__name__)
+ attempt = 0
+ while attempt < max_attempts:
+ attempt += 1
+ try:
+ waiter.wait(
+ **state_args,
+ WaiterConfig={
+ "Delay": 1,
Review Comment:
Yes that parameter can be omitted. As long as `MaxAttempts` is 1, we don't get to the `sleep` call in 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] vincbeck commented on a diff in pull request #30463: Add custom waiters to EMR Serverless
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #30463:
URL: https://github.com/apache/airflow/pull/30463#discussion_r1212336666
##########
airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -901,36 +930,63 @@ def execute(self, context: Context) -> str | None:
raise AirflowException(f"Application Creation failed: {response}")
self.log.info("EMR serverless application created: %s", application_id)
-
- # This should be replaced with a boto waiter when available.
- waiter(
- get_state_callable=self.hook.conn.get_application,
- get_state_args={"applicationId": application_id},
- parse_response=["application", "state"],
- desired_state={"CREATED"},
- failure_states=EmrServerlessHook.APPLICATION_FAILURE_STATES,
- object_type="application",
- action="created",
- countdown=self.waiter_countdown,
- check_interval_seconds=self.waiter_check_interval_seconds,
- )
+ attempt = 0
+ waiter = self.hook.get_waiter("serverless_app_created")
+ while attempt < self.waiter_max_attempts:
Review Comment:
I see ... I just feel we are going backwards. We initially decided to start using boto3 custom waiters to get rid of these while loops and to make the code cleaner. Now, we want to keep using boto3 customer waiter but wrap them around the exact same code we tried to remove initially. I now I dont provide any solution here but should we try to find another solution? If I am the only one thinking that way, I am fine
--
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 #30463: Add custom waiters to EMR Serverless
Posted by "vandonr-amz (via GitHub)" <gi...@apache.org>.
vandonr-amz commented on code in PR #30463:
URL: https://github.com/apache/airflow/pull/30463#discussion_r1159155294
##########
airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -809,10 +810,13 @@ class EmrServerlessCreateApplicationOperator(BaseOperator):
Its value must be unique for each request.
:param config: Optional dictionary for arbitrary parameters to the boto API create_application call.
:param aws_conn_id: AWS connection to use
- :param waiter_countdown: Total amount of time, in seconds, the operator will wait for
+ :param waiter_countdown: (deprecated) Total amount of time, in seconds, the operator will wait for
the application to start. Defaults to 25 minutes.
- :param waiter_check_interval_seconds: Number of seconds between polling the state of the application.
- Defaults to 60 seconds.
+ :param waiter_check_interval_seconds: (deprecated) Number of seconds between polling the state
Review Comment:
why do you deprecate this one ? Instead of renaming it to waiter_delay but we could keep the old name, couldn't we ? Is it to harmonize the parameters names between operators ?
--
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 #30463: Add custom waiters to EMR Serverless
Posted by "vandonr-amz (via GitHub)" <gi...@apache.org>.
vandonr-amz commented on code in PR #30463:
URL: https://github.com/apache/airflow/pull/30463#discussion_r1159155707
##########
airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -809,10 +810,13 @@ class EmrServerlessCreateApplicationOperator(BaseOperator):
Its value must be unique for each request.
:param config: Optional dictionary for arbitrary parameters to the boto API create_application call.
:param aws_conn_id: AWS connection to use
- :param waiter_countdown: Total amount of time, in seconds, the operator will wait for
+ :param waiter_countdown: (deprecated) Total amount of time, in seconds, the operator will wait for
the application to start. Defaults to 25 minutes.
- :param waiter_check_interval_seconds: Number of seconds between polling the state of the application.
- Defaults to 60 seconds.
+ :param waiter_check_interval_seconds: (deprecated) Number of seconds between polling the state
Review Comment:
Not super fan of making user change their code for no reason except to standardize names.
This is adding friction to upgrading where we could very well keep the current name since there is no behavior change.
##########
airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -992,21 +1036,40 @@ def execute(self, context: Context) -> dict:
self.log.info("EMR serverless job started: %s", response["jobRunId"])
if self.wait_for_completion:
- # This should be replaced with a boto waiter when available.
- waiter(
- get_state_callable=self.hook.conn.get_job_run,
- get_state_args={
- "applicationId": self.application_id,
- "jobRunId": response["jobRunId"],
- },
- parse_response=["jobRun", "state"],
- desired_state=EmrServerlessHook.JOB_SUCCESS_STATES,
- failure_states=EmrServerlessHook.JOB_FAILURE_STATES,
- object_type="job",
- action="run",
- countdown=self.waiter_countdown,
- check_interval_seconds=self.waiter_check_interval_seconds,
- )
+ try:
+ self.hook.get_waiter("serverless_job_running").wait(
+ applicationId=self.application_id,
+ jobRunId=response["jobRunId"],
+ WaiterConfig=prune_dict(
+ {
+ "Delay": self.waiter_delay,
+ "MaxAttempts": self.waiter_max_attempts,
+ }
+ ),
+ )
+ self.log.info("EMR serverless job is running: %s", response["jobRunId"])
+ self.hook.get_waiter("serverless_job_completed").wait(
+ applicationId=self.application_id,
+ jobRunId=response["jobRunId"],
+ WaiterConfig=prune_dict(
+ {
+ "Delay": self.waiter_delay,
+ "MaxAttempts": self.waiter_max_attempts,
+ }
+ ),
+ )
Review Comment:
does this really need to be 2 waiters just to be able to log a line when the job is running ? Doesn't the waiter already log what state it's seeing ?
##########
airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -823,18 +827,41 @@ def __init__(
config: dict | None = None,
wait_for_completion: bool = True,
aws_conn_id: str = "aws_default",
- waiter_countdown: int = 25 * 60,
- waiter_check_interval_seconds: int = 60,
+ waiter_max_attempts: int | None | ArgNotSet = NOTSET,
+ waiter_delay: int | None | ArgNotSet = NOTSET,
Review Comment:
why allow None here ? It's either set to an integer value or not set, isn't it ?
(same question for other constructors below)
--
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] vincbeck commented on a diff in pull request #30463: Add custom waiters to EMR Serverless
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #30463:
URL: https://github.com/apache/airflow/pull/30463#discussion_r1212276610
##########
airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -901,36 +930,63 @@ def execute(self, context: Context) -> str | None:
raise AirflowException(f"Application Creation failed: {response}")
self.log.info("EMR serverless application created: %s", application_id)
-
- # This should be replaced with a boto waiter when available.
- waiter(
- get_state_callable=self.hook.conn.get_application,
- get_state_args={"applicationId": application_id},
- parse_response=["application", "state"],
- desired_state={"CREATED"},
- failure_states=EmrServerlessHook.APPLICATION_FAILURE_STATES,
- object_type="application",
- action="created",
- countdown=self.waiter_countdown,
- check_interval_seconds=self.waiter_check_interval_seconds,
- )
+ attempt = 0
+ waiter = self.hook.get_waiter("serverless_app_created")
+ while attempt < self.waiter_max_attempts:
Review Comment:
I would make them verbose then
--
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] vincbeck commented on a diff in pull request #30463: Add custom waiters to EMR Serverless
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #30463:
URL: https://github.com/apache/airflow/pull/30463#discussion_r1214445808
##########
airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -901,36 +930,63 @@ def execute(self, context: Context) -> str | None:
raise AirflowException(f"Application Creation failed: {response}")
self.log.info("EMR serverless application created: %s", application_id)
-
- # This should be replaced with a boto waiter when available.
- waiter(
- get_state_callable=self.hook.conn.get_application,
- get_state_args={"applicationId": application_id},
- parse_response=["application", "state"],
- desired_state={"CREATED"},
- failure_states=EmrServerlessHook.APPLICATION_FAILURE_STATES,
- object_type="application",
- action="created",
- countdown=self.waiter_countdown,
- check_interval_seconds=self.waiter_check_interval_seconds,
- )
+ attempt = 0
+ waiter = self.hook.get_waiter("serverless_app_created")
+ while attempt < self.waiter_max_attempts:
Review Comment:
I see. I have another proposal (yes as you can see I really hate this code). Why dont you move this code in a separate function and you use `tenacity` to handle the retry. The code would look like this:
```python
@tenacity.retry(
retry=tenacity.retry_if_exception_type(WaiterError),
stop=tenacity.stop_after_attempt(self.waiter_max_attempts),
wait=tenacity.wait_fixed(self.waiter_delay),
)
def wait(waiter: Waiter, application_id: str):
try:
waiter.wait(
applicationId=application_id,
WaiterConfig=prune_dict(
{
"Delay": self.waiter_delay,
"MaxAttempts": 1,
}
),
)
except WaiterError as error:
if "terminal failure" in str(error):
raise AirflowException(f"Serverless Application Creation failed: {error}")
self.log.info(
"Serverless Application status is: %s - %s",
error.last_response["application"]["state"],
error.last_response["application"]["stateDetails"],
)
raise
```
I am still not a big fan of this code but this is definitely better than the while loop
--
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 #30463: Add custom waiters to EMR Serverless
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #30463:
URL: https://github.com/apache/airflow/pull/30463#discussion_r1212325449
##########
airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -901,36 +930,63 @@ def execute(self, context: Context) -> str | None:
raise AirflowException(f"Application Creation failed: {response}")
self.log.info("EMR serverless application created: %s", application_id)
-
- # This should be replaced with a boto waiter when available.
- waiter(
- get_state_callable=self.hook.conn.get_application,
- get_state_args={"applicationId": application_id},
- parse_response=["application", "state"],
- desired_state={"CREATED"},
- failure_states=EmrServerlessHook.APPLICATION_FAILURE_STATES,
- object_type="application",
- action="created",
- countdown=self.waiter_countdown,
- check_interval_seconds=self.waiter_check_interval_seconds,
- )
+ attempt = 0
+ waiter = self.hook.get_waiter("serverless_app_created")
+ while attempt < self.waiter_max_attempts:
Review Comment:
What do you mean by making them verbose? We can't make any changes to the waiters because they are in the botocore repo.
--
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] vincbeck commented on a diff in pull request #30463: Add custom waiters to EMR Serverless
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #30463:
URL: https://github.com/apache/airflow/pull/30463#discussion_r1213207936
##########
airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -901,36 +930,63 @@ def execute(self, context: Context) -> str | None:
raise AirflowException(f"Application Creation failed: {response}")
self.log.info("EMR serverless application created: %s", application_id)
-
- # This should be replaced with a boto waiter when available.
- waiter(
- get_state_callable=self.hook.conn.get_application,
- get_state_args={"applicationId": application_id},
- parse_response=["application", "state"],
- desired_state={"CREATED"},
- failure_states=EmrServerlessHook.APPLICATION_FAILURE_STATES,
- object_type="application",
- action="created",
- countdown=self.waiter_countdown,
- check_interval_seconds=self.waiter_check_interval_seconds,
- )
+ attempt = 0
+ waiter = self.hook.get_waiter("serverless_app_created")
+ while attempt < self.waiter_max_attempts:
Review Comment:
I see. I agree with you, we definitely need these logs and I am just sad this is the only way to get these. Especially because it worsen the code. My understanding is the waiter is pushing logs but at the DEBUG level right? Did you try to increase the specific logger that is sending these logs? Then, instead of pushing these logs to DEBUG level, it would do it, say, at INFO level? See [here](https://github.com/boto/boto3/issues/521) more informations.
--
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 #30463: Add custom waiters to EMR Serverless
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas merged PR #30463:
URL: https://github.com/apache/airflow/pull/30463
--
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] vincbeck commented on a diff in pull request #30463: Add custom waiters to EMR Serverless
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #30463:
URL: https://github.com/apache/airflow/pull/30463#discussion_r1228700420
##########
airflow/providers/amazon/aws/utils/waiter_with_logging.py:
##########
@@ -0,0 +1,77 @@
+# 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
+
+from botocore.exceptions import WaiterError
+
+from airflow.exceptions import AirflowException
+
+
+def wait(
Review Comment:
It is definitely way cleaner on the operator side! Thanks for doing it. However (and sorry to insist ...), what you are doing here is totally achievable with `tenacity`. You can add logs to every attempt or to every failed attempt, [see documentation here](https://tenacity.readthedocs.io/en/latest/#before-and-after-retry-and-logging). But ... I dont want to hold this PR any longer and I think I expressed myself enough on that issue, at the end of the day it is only question of style :) So LGTM :)
--
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 #30463: Add custom waiters to EMR Serverless
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #30463:
URL: https://github.com/apache/airflow/pull/30463#discussion_r1228765387
##########
airflow/providers/amazon/aws/utils/waiter_with_logging.py:
##########
@@ -0,0 +1,77 @@
+# 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
+
+from botocore.exceptions import WaiterError
+
+from airflow.exceptions import AirflowException
+
+
+def wait(
Review Comment:
Thanks! I tried quite a few different things to get it working with tenacity, but my issue was with debugging. I found that using tenacity made the debugging very difficult. It also added complications in the unit tests (which I was able to resolve, but it took some playing around).
In hindsight, I think using tenacity to implement a basic retry like this is overkill - tenacity seems to be able to implement fairly complex retry algorithms whereas we were using it as a `while` loop. I would have kept it if things were working smoothly, but I wasn't able to figure out how debugging worked, so I decided to stop before I went further down the rabbit hole :D.
--
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 #30463: Add custom waiters to EMR Serverless
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #30463:
URL: https://github.com/apache/airflow/pull/30463#discussion_r1164619929
##########
airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -809,10 +810,13 @@ class EmrServerlessCreateApplicationOperator(BaseOperator):
Its value must be unique for each request.
:param config: Optional dictionary for arbitrary parameters to the boto API create_application call.
:param aws_conn_id: AWS connection to use
- :param waiter_countdown: Total amount of time, in seconds, the operator will wait for
+ :param waiter_countdown: (deprecated) Total amount of time, in seconds, the operator will wait for
the application to start. Defaults to 25 minutes.
- :param waiter_check_interval_seconds: Number of seconds between polling the state of the application.
- Defaults to 60 seconds.
+ :param waiter_check_interval_seconds: (deprecated) Number of seconds between polling the state
Review Comment:
I think consistent naming is a worthy effort because it leads to more maintainable code. In this case, there is no reason for the user to change their code at this point. The way it is written, if a user does not specify any waiter related parameters, they will not see any difference, including the deprecation message. Only if the user is specifically using the old parameters will the message be shown and they will not see any difference in behavior - the new waiter parameters are computed from the old ones. I think its important to update variable names as code changes, otherwise it will lead to a disconnect between variable names, and what they represent.
--
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] vincbeck commented on a diff in pull request #30463: Add custom waiters to EMR Serverless
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #30463:
URL: https://github.com/apache/airflow/pull/30463#discussion_r1212336666
##########
airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -901,36 +930,63 @@ def execute(self, context: Context) -> str | None:
raise AirflowException(f"Application Creation failed: {response}")
self.log.info("EMR serverless application created: %s", application_id)
-
- # This should be replaced with a boto waiter when available.
- waiter(
- get_state_callable=self.hook.conn.get_application,
- get_state_args={"applicationId": application_id},
- parse_response=["application", "state"],
- desired_state={"CREATED"},
- failure_states=EmrServerlessHook.APPLICATION_FAILURE_STATES,
- object_type="application",
- action="created",
- countdown=self.waiter_countdown,
- check_interval_seconds=self.waiter_check_interval_seconds,
- )
+ attempt = 0
+ waiter = self.hook.get_waiter("serverless_app_created")
+ while attempt < self.waiter_max_attempts:
Review Comment:
I see ... I just feel we are going backwards. We initially decided to start using boto3 custom waiters to get rid of these while loop and to have a code cleaner. Now, we want to keep using boto3 customer waiter but wrap them around the exact same code we tried to remove initially. I now I dont provide any solution here but should we try to find another solution? If I am the only one thinking that way, I am fine
--
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] vincbeck commented on a diff in pull request #30463: Add custom waiters to EMR Serverless
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #30463:
URL: https://github.com/apache/airflow/pull/30463#discussion_r1212329796
##########
airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -901,36 +930,63 @@ def execute(self, context: Context) -> str | None:
raise AirflowException(f"Application Creation failed: {response}")
self.log.info("EMR serverless application created: %s", application_id)
-
- # This should be replaced with a boto waiter when available.
- waiter(
- get_state_callable=self.hook.conn.get_application,
- get_state_args={"applicationId": application_id},
- parse_response=["application", "state"],
- desired_state={"CREATED"},
- failure_states=EmrServerlessHook.APPLICATION_FAILURE_STATES,
- object_type="application",
- action="created",
- countdown=self.waiter_countdown,
- check_interval_seconds=self.waiter_check_interval_seconds,
- )
+ attempt = 0
+ waiter = self.hook.get_waiter("serverless_app_created")
+ while attempt < self.waiter_max_attempts:
Review Comment:
Oh! Sorry, I thought you were using custom waiters
--
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] vincbeck commented on a diff in pull request #30463: Add custom waiters to EMR Serverless
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #30463:
URL: https://github.com/apache/airflow/pull/30463#discussion_r1212276610
##########
airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -901,36 +930,63 @@ def execute(self, context: Context) -> str | None:
raise AirflowException(f"Application Creation failed: {response}")
self.log.info("EMR serverless application created: %s", application_id)
-
- # This should be replaced with a boto waiter when available.
- waiter(
- get_state_callable=self.hook.conn.get_application,
- get_state_args={"applicationId": application_id},
- parse_response=["application", "state"],
- desired_state={"CREATED"},
- failure_states=EmrServerlessHook.APPLICATION_FAILURE_STATES,
- object_type="application",
- action="created",
- countdown=self.waiter_countdown,
- check_interval_seconds=self.waiter_check_interval_seconds,
- )
+ attempt = 0
+ waiter = self.hook.get_waiter("serverless_app_created")
+ while attempt < self.waiter_max_attempts:
Review Comment:
I would make them verbose then. Either by default or by adding an option to specify a callback to call while waiting
--
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 #30463: Add custom waiters to EMR Serverless
Posted by "vandonr-amz (via GitHub)" <gi...@apache.org>.
vandonr-amz commented on code in PR #30463:
URL: https://github.com/apache/airflow/pull/30463#discussion_r1212272002
##########
airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -901,36 +930,63 @@ def execute(self, context: Context) -> str | None:
raise AirflowException(f"Application Creation failed: {response}")
self.log.info("EMR serverless application created: %s", application_id)
-
- # This should be replaced with a boto waiter when available.
- waiter(
- get_state_callable=self.hook.conn.get_application,
- get_state_args={"applicationId": application_id},
- parse_response=["application", "state"],
- desired_state={"CREATED"},
- failure_states=EmrServerlessHook.APPLICATION_FAILURE_STATES,
- object_type="application",
- action="created",
- countdown=self.waiter_countdown,
- check_interval_seconds=self.waiter_check_interval_seconds,
- )
+ attempt = 0
+ waiter = self.hook.get_waiter("serverless_app_created")
+ while attempt < self.waiter_max_attempts:
Review Comment:
it's the workaround we agreed on to be able to log the status changes while waiting. Otherwise, the waiters are completely silent until they succeed/fail.
--
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] vincbeck commented on a diff in pull request #30463: Add custom waiters to EMR Serverless
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #30463:
URL: https://github.com/apache/airflow/pull/30463#discussion_r1212258460
##########
airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -901,36 +930,63 @@ def execute(self, context: Context) -> str | None:
raise AirflowException(f"Application Creation failed: {response}")
self.log.info("EMR serverless application created: %s", application_id)
-
- # This should be replaced with a boto waiter when available.
- waiter(
- get_state_callable=self.hook.conn.get_application,
- get_state_args={"applicationId": application_id},
- parse_response=["application", "state"],
- desired_state={"CREATED"},
- failure_states=EmrServerlessHook.APPLICATION_FAILURE_STATES,
- object_type="application",
- action="created",
- countdown=self.waiter_countdown,
- check_interval_seconds=self.waiter_check_interval_seconds,
- )
+ attempt = 0
+ waiter = self.hook.get_waiter("serverless_app_created")
+ while attempt < self.waiter_max_attempts:
Review Comment:
Is not the purpose of a waiter to handle these different attempts? Why are you calling `waiter.wait(` with `"MaxAttempts": 1,` `self.waiter_max_attempts` times?
--
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] vincbeck commented on a diff in pull request #30463: Add custom waiters to EMR Serverless
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #30463:
URL: https://github.com/apache/airflow/pull/30463#discussion_r1212336666
##########
airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -901,36 +930,63 @@ def execute(self, context: Context) -> str | None:
raise AirflowException(f"Application Creation failed: {response}")
self.log.info("EMR serverless application created: %s", application_id)
-
- # This should be replaced with a boto waiter when available.
- waiter(
- get_state_callable=self.hook.conn.get_application,
- get_state_args={"applicationId": application_id},
- parse_response=["application", "state"],
- desired_state={"CREATED"},
- failure_states=EmrServerlessHook.APPLICATION_FAILURE_STATES,
- object_type="application",
- action="created",
- countdown=self.waiter_countdown,
- check_interval_seconds=self.waiter_check_interval_seconds,
- )
+ attempt = 0
+ waiter = self.hook.get_waiter("serverless_app_created")
+ while attempt < self.waiter_max_attempts:
Review Comment:
I see ... I just feel we are going backwards. We initially decided to start using boto3 custom waiters to get rid of these while loops and to make the code cleaner. Now, we want to keep using boto3 custom waiter but wrap them around the exact same code we tried to remove initially. I know I dont provide any solution here but should we try to find another solution? If I am the only one thinking that way, I am fine
--
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 #30463: Add custom waiters to EMR Serverless
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #30463:
URL: https://github.com/apache/airflow/pull/30463#discussion_r1235613954
##########
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:
Is there a reason why it would be better to separate the 2 parameters. One benefit of keeping them together is that I can package them into `EMR_SERVERLESS_APPLICATION_STATUS`, which gets used several times. I think this would be the case for multiple operators
--
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