Patch FileTaskHandler._read to accommodate pre-migration timezone naive logs.

Co-authored-by: Cooper Stimson <cooper.stimson@cloverhealth.com>
Co-authored-by: Christian Miller <christian.miller@cloverhealth.com>

Author: Cooper Stimson

airflow/utils/log/file_task_handler.py

@@ -121,7 +121,7 @@ def _read(self, ti, try_number, metadata=None):
                 except (AirflowConfigException, ValueError):
                     pass
 
-                response = requests.get(url, timeout=timeout)
+                response, log = self._try_both_log_locations(url, timeout, log)
 
                 # Check if the resource was properly fetched
                 response.raise_for_status()
@@ -205,3 +205,29 @@ def _init_file(self, ti):
             os.chmod(full_path, 0o666)
 
         return full_path
+
+    def _try_both_log_locations(self, url, timeout, log):
+        """If a logfile cannot be found, try the timezone naive url format."""
+        response = requests.get(url, timeout=timeout)
+
+        try:
+            response.raise_for_status()
+        except requests.HTTPError as _:
+            log += "*** (clover patch) Cannot find log at filename {url}\n".format(url=url)
+            old_url = self._convert_to_old_url(url)
+            try:
+                response = requests.get(old_url, timeout=timeout)
+                response.raise_for_status()
+            except requests.HTTPError as _:
+                log += "*** (clover patch) Cannot find log at filename {old_url}\n".format(old_url=old_url)
+
+        return response, log
+
+    @staticmethod
+    def _convert_to_old_url(url):
+        """Convert a logfile url from timezone aware to timezone naive.
+
+        ..NOTE ::
+            This assumes that all timestamps involved are UTC.
+        """
+        return url.replace('+00:00', '')