Don't use time.time() or timezone.utcnow() for duration calculations

[ashb]

time.time() - start, or timezone.utcnow() - start_dttm will work
fine in 99% of cases, but it has one fatal flaw:

They both operate on system time, and that can go backwards.

While this might be surprising, it can happen -- usually due to clocks
being adjusted.

And while it is might seem rare, for long running processes it is more
common than we might expect. Most of these durations are harmless to get
wrong (just being logs) it is better to be safe than sorry.

Also the utcnow() style I have replaced will be much lighter weight -
creating a date time object is a comparatively expensive operation, and
computing a diff between two even more so, especially when compared to
just subtracting two floats.

To make the "common" case easier of wanting to compute a duration for a
block, I have made Stats.timer() return an object that has a
duration field.

---

^ Add meaningful description above

Read the Pull Request Guidelines for more information.
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.

[ashb]

(This will likely fail tests at first. I likely messed up some calculations somewhere.)

[turbaszek]

Should we add note about this approach in https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst#coding-style-and-best-practices ?

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[kaxil]

Master was failing, I have fixed it and rebased this PR

[ashb]

Should we add note about this approach in https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst#coding-style-and-best-practices ?

Done in a7642459d

[kaxil]

One failure

self = <Task(PythonSensor): test_reschedule_handling_sensor>
context = {'conf': <airflow.configuration.AirflowConfigParser object at 0x7f383eda7860>, 'dag': <DAG: test_reschedule_handling>,...36:50.699432+00:00: scheduled__2020-11-14T12:36:50.699432+00:00, externally triggered: False>, 'ds': '2020-11-14', ...}

    def execute(self, context: Dict) -> Any:
        started_at = time.monotonic()
        try_number = 1
        log_dag_id = self.dag.dag_id if self.has_dag() else ""
        if self.reschedule:
            # If reschedule, use first start date of current try
            task_reschedules = TaskReschedule.find_for_task_instance(context['ti'])
            if task_reschedules:
                started_at = task_reschedules[0].start_date
                try_number = len(task_reschedules) + 1
        while not self.poke(context):
>           if time.monotonic() - started_at > self.timeout:
E           TypeError: unsupported operand type(s) for -: 'float' and 'FakeDatetime'

airflow/sensors/base_sensor_operator.py:210: TypeError

[ashb]

That failure was an actual bug -- I had broken rescheduled sensors. Yay for tests!

[github-actions]

The PR needs to run all tests because it modifies core of Airflow! Please rebase it to latest master or ask committer to re-run it!

[ashb]

Hopefully got the tests sorted now (just one last instance of mocking time.time when it now needs to mock time.monotonic)

[ashb]

Hmmmm, even with rebasing Kube tests are failing. I wonder what I broke.

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[ashb]

YAY GREEN FINALLY!

[turbaszek]

@ashb what exactly was the problem with the tests?

[ashb]

https://github.com/apache/airflow/pull/12353/files#diff-f0e400bd7a526279c114ce0bd87e80f1f9344885b749f2da495d4f7bb670432eR228

When statsd was actually enabled that line was failing, but we don't test with statsd on anywhere outside of the kube tests (thankfully?) -- but to get the error we had to download the kube logs artifact and trawl through the logs until I found the actual error.