[AIRFLOW-6944] Allow AWS DataSync to "catch up" when Task is already … by baolsen · Pull Request #7585 · apache/airflow

baolsen · 2020-02-28T12:23:33Z

…running

Make sure to mark the boxes below before creating PR: [x]

Description above provides context of the change
Commit message/PR title starts with [AIRFLOW-NNNN]. AIRFLOW-NNNN = JIRA ID^*
Unit tests coverage for changes (not needed for documentation changes)
Commits follow "How to write a good git commit message"
Relevant documentation is updated including usage instructions.
I will engage committers as explained in Contribution Workflow Example.

^* For document-only changes commit message can start with [AIRFLOW-XXXX].

In case of fundamental code change, Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in UPDATING.md.
Read the Pull Request Guidelines for more information.

codecov-io · 2020-02-28T13:53:19Z

Codecov Report

Merging #7585 into master will decrease coverage by 0.32%.
The diff coverage is 23.25%.

@@            Coverage Diff             @@
##           master    #7585      +/-   ##
==========================================
- Coverage   86.85%   86.52%   -0.33%     
==========================================
  Files         896      896              
  Lines       42622    42657      +35     
==========================================
- Hits        37021    36911     -110     
- Misses       5601     5746     +145

Impacted Files	Coverage Δ
airflow/providers/amazon/aws/hooks/datasync.py	`16.78% <0%> (ø)`	⬆️
airflow/providers/amazon/aws/operators/datasync.py	`32.63% <23.8%> (-1.57%)`	⬇️
airflow/kubernetes/volume_mount.py	`44.44% <0%> (-55.56%)`	⬇️
airflow/kubernetes/volume.py	`52.94% <0%> (-47.06%)`	⬇️
airflow/kubernetes/pod_launcher.py	`47.18% <0%> (-45.08%)`	⬇️
...viders/cncf/kubernetes/operators/kubernetes_pod.py	`69.69% <0%> (-25.26%)`	⬇️
airflow/kubernetes/refresh_config.py	`50.98% <0%> (-23.53%)`	⬇️

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update bbfc18e...f7c5571. Read the comment docs.

…running

baolsen · 2020-03-16T11:45:43Z

Hi @feluelle / @nuclearpinguin , would appreciate a review :)

One thing I am aware of - currently there are 2 wait loops:

Wait for task to be able to be started - implemented in the DataSync Operator.
Wait for task execution to be completed - implemented in the DataSync hook. (Task execution is what you get from AWS after triggering a Task to start)

What is your recommendation to standardise these?
I have no preference... Just don't want a Sensor because the Operator is currently self-contained in terms of creating/deleting/running tasks. I'd like to move both wait loops either into the hook or the operator, not sure where is typically the best place.

turbaszek · 2020-03-16T19:36:20Z

airflow/providers/amazon/aws/operators/datasync.py

-    def _execute_datasync_task(self):
-        """Create and monitor an AWSDataSync TaskExecution for a Task."""
-        hook = self.get_hook()
+    def _wait_get_status_before_start(


Suggested change

def _wait_get_status_before_start(

def _wait_for_status(

turbaszek · 2020-03-16T19:41:06Z

Wait for task to be able to be started - implemented in the DataSync Operator.

Wait for task execution to be completed - implemented in the DataSync hook. (Task execution is what you get from AWS after triggering a Task to start)

What is your recommendation to standardise these?

I would say that we should not wait for task to be able to start. If something has to happen before the task we should use sensor or wait for the completion in the upstream operator. For example in case of CreateResource >> UseResource the CreateResource op should check if the resource was successfully created. This is the approach we use in GCP operators/hooks.

feluelle · 2020-03-21T11:30:59Z

airflow/providers/amazon/aws/operators/datasync.py

+            self.log.info(
+                'The Task will be started because its status is in %s.',
+                self.TASK_STATUS_START)
+            # Start the DataSync Task


Suggested change

# Start the DataSync Task

feluelle · 2020-03-21T11:33:52Z

airflow/providers/amazon/aws/operators/datasync.py

+            raise AirflowException(
+                'Task cannot be started because its status is in %s.'
+                % self.TASK_STATUS_FAIL
+            )


Suggested change

raise AirflowException(

'Task cannot be started because its status is in %s.'

% self.TASK_STATUS_FAIL

)

raise AirflowException(

f'Task cannot be started because its status is in {self.TASK_STATUS_FAIL}.'

)

Please use f-strings :)

feluelle · 2020-03-21T11:34:48Z

airflow/providers/amazon/aws/operators/datasync.py

+                % self.TASK_STATUS_FAIL
+            )
+        else:
+            raise AirflowException('Unexpected task status %s.' % self.task_status)


Suggested change

raise AirflowException('Unexpected task status %s.' % self.task_status)

raise AirflowException(f'Unexpected task status {self.task_status}')

feluelle · 2020-03-21T11:50:34Z

airflow/providers/amazon/aws/operators/datasync.py

+        hook = self.get_hook()
+        task_status = hook.get_task_description(self.task_arn)['Status']
+        iteration = 0
+        while task_status in self.TASK_STATUS_WAIT_BEFORE_START:
+            self.log.info(
+                'Task status is %s.'
+                ' Waiting for it to not be %s.'
+                ' Iteration %s/%s.',
+                task_status,
+                self.TASK_STATUS_WAIT_BEFORE_START,
+                iteration,
+                max_iterations)
+            time.sleep(self.wait_interval_seconds)
+            task_status = hook.get_task_description(self.task_arn)['Status']
+            iteration = iteration + 1
+            if iteration >= max_iterations:
+                break
+
+        return task_status


Suggested change

hook = self.get_hook()

task_status = hook.get_task_description(self.task_arn)['Status']

iteration = 0

while task_status in self.TASK_STATUS_WAIT_BEFORE_START:

self.log.info(

'Task status is %s.'

' Waiting for it to not be %s.'

' Iteration %s/%s.',

task_status,

self.TASK_STATUS_WAIT_BEFORE_START,

iteration,

max_iterations)

time.sleep(self.wait_interval_seconds)

task_status = hook.get_task_description(self.task_arn)['Status']

iteration = iteration + 1

if iteration >= max_iterations:

break

return task_status

hook = self.get_hook()

for iteration in range(max_iterations):

task_status = hook.get_task_description(self.task_arn)['Status']

self.log.info(

'Task status is %s.'

' Waiting for it to not be %s.'

' Iteration %s/%s.',

task_status,

self.TASK_STATUS_WAIT_BEFORE_START,

iteration,

max_iterations)

if task_status not in self.TASK_STATUS_WAIT_BEFORE_START:

break

time.sleep(self.wait_interval_seconds)

return task_status

WDYT?

feluelle · 2020-03-21T11:56:11Z

airflow/providers/amazon/aws/operators/datasync.py

-        hook = self.get_hook()
+    def _wait_get_status_before_start(
+            self,
+            max_iterations=12 * 180):  # wait_interval_seconds*12*180=180 minutes by default


Suggested change

max_iterations=12 * 180): # wait_interval_seconds*12*180=180 minutes by default

max_iterations : int = 12 * 180) -> str:

You already have that comment in the docs below. Even better it would be to add :param and add the comment there.

feluelle · 2020-03-21T11:56:56Z

airflow/providers/amazon/aws/operators/datasync.py

+        The Task can be started when its Status is not in TASK_STATUS_WAIT_BEFORE_START
+        Uses wait_interval_seconds (which is also used while waiting for TaskExecution)
+        So, max_iterations=12*180 gives 180 minutes wait by default.
+        """


Please also add :returns: here.

feluelle · 2020-03-21T12:02:16Z

Wait for task to be able to be started - implemented in the DataSync Operator.

Wait for task execution to be completed - implemented in the DataSync hook. (Task execution is what you get from AWS after triggering a Task to start)

What is your recommendation to standardise these?

I would say that we should not wait for task to be able to start. If something has to happen before the task we should use sensor or wait for the completion in the upstream operator. For example in case of CreateResource >> UseResource the CreateResource op should check if the resource was successfully created. This is the approach we use in GCP operators/hooks.

I think both options are useful to have. To #7410 (Tableau Integration) I added a blocking parameter which in case is set to True enables a sensor task. If set to False it does not wait (=async).
See: https://github.com/apache/airflow/pull/7410/files#diff-4f65033904d8c2b6c064e959d218d513R71

feluelle · 2020-04-08T10:43:33Z

@baolsen gentle ping :)

stale · 2020-05-23T21:30:21Z

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.

boring-cyborg bot added the provider:amazon AWS/Amazon - related issues label Feb 28, 2020

[AIRFLOW-6944] Allow AWS DataSync to "catch up" when Task is already …

a9fb7ca

…running

turbaszek requested a review from feluelle March 16, 2020 12:23

turbaszek reviewed Mar 16, 2020

View reviewed changes

feluelle requested changes Mar 21, 2020

View reviewed changes

stale bot added the stale Stale PRs per the .github/workflows/stale.yml policy file label May 23, 2020

stale bot closed this May 30, 2020

	raise AirflowException('Unexpected task status %s.' % self.task_status)
	raise AirflowException(f'Unexpected task status {self.task_status}')

	max_iterations=12 * 180): # wait_interval_seconds12180=180 minutes by default
	max_iterations : int = 12 * 180) -> str:

Conversation

baolsen commented Feb 28, 2020 • edited by boring-cyborg bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

codecov-io commented Feb 28, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

baolsen commented Mar 16, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

turbaszek Mar 16, 2020

Choose a reason for hiding this comment

Uh oh!

turbaszek commented Mar 16, 2020

Uh oh!

feluelle Mar 21, 2020

Choose a reason for hiding this comment

Uh oh!

feluelle Mar 21, 2020

Choose a reason for hiding this comment

Uh oh!

feluelle Mar 21, 2020

Choose a reason for hiding this comment

Uh oh!

feluelle Mar 21, 2020

Choose a reason for hiding this comment

Uh oh!

feluelle Mar 21, 2020

Choose a reason for hiding this comment

Uh oh!

feluelle Mar 21, 2020

Choose a reason for hiding this comment

Uh oh!

feluelle commented Mar 21, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

feluelle commented Apr 8, 2020

Uh oh!

stale bot commented May 23, 2020

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

baolsen commented Feb 28, 2020 •

edited by boring-cyborg bot

Loading

codecov-io commented Feb 28, 2020 •

edited

Loading

baolsen commented Mar 16, 2020 •

edited

Loading

feluelle commented Mar 21, 2020 •

edited

Loading