Fix bug in task logs when using AWS CloudWatch. Do not set start_time (#33673)

* Fix bug in task logs when using AWS CloudWatch. Do not set `start_time`

* Fix tests

Author: vincbeck

airflow/providers/amazon/aws/log/cloudwatch_task_handler.py

@@ -114,9 +114,6 @@ def get_cloudwatch_logs(self, stream_name: str, task_instance: TaskInstance) ->
         :param task_instance: the task instance to get logs about
         :return: string of all logs from the given log stream
         """
-        start_time = (
-            0 if task_instance.start_date is None else datetime_to_epoch_utc_ms(task_instance.start_date)
-        )
         # If there is an end_date to the task instance, fetch logs until that date + 30 seconds
         # 30 seconds is an arbitrary buffer so that we don't miss any logs that were emitted
         end_time = (
@@ -127,7 +124,6 @@ def get_cloudwatch_logs(self, stream_name: str, task_instance: TaskInstance) ->
         events = self.hook.get_log_events(
             log_group=self.log_group,
             log_stream_name=stream_name,
-            start_time=start_time,
             end_time=end_time,
         )
         return "\n".join(self._event_to_str(event) for event in events)

tests/providers/amazon/aws/log/test_cloudwatch_task_handler.py

@@ -155,35 +155,19 @@ def test_read(self):
         )
 
     @pytest.mark.parametrize(
-        "start_date, end_date, expected_start_time, expected_end_time",
+        "end_date, expected_end_time",
         [
-            (None, None, 0, None),
-            (datetime(2020, 1, 1), None, datetime_to_epoch_utc_ms(datetime(2020, 1, 1)), None),
-            (
-                None,
-                datetime(2020, 1, 2),
-                0,
-                datetime_to_epoch_utc_ms(datetime(2020, 1, 2) + timedelta(seconds=30)),
-            ),
-            (
-                datetime(2020, 1, 1),
-                datetime(2020, 1, 2),
-                datetime_to_epoch_utc_ms(datetime(2020, 1, 1)),
-                datetime_to_epoch_utc_ms(datetime(2020, 1, 2) + timedelta(seconds=30)),
-            ),
+            (None, None),
+            (datetime(2020, 1, 2), datetime_to_epoch_utc_ms(datetime(2020, 1, 2) + timedelta(seconds=30))),
         ],
     )
     @mock.patch.object(AwsLogsHook, "get_log_events")
-    def test_get_cloudwatch_logs(
-        self, mock_get_log_events, start_date, end_date, expected_start_time, expected_end_time
-    ):
-        self.ti.start_date = start_date
+    def test_get_cloudwatch_logs(self, mock_get_log_events, end_date, expected_end_time):
         self.ti.end_date = end_date
         self.cloudwatch_task_handler.get_cloudwatch_logs(self.remote_log_stream, self.ti)
         mock_get_log_events.assert_called_once_with(
             log_group=self.remote_log_group,
             log_stream_name=self.remote_log_stream,
-            start_time=expected_start_time,
             end_time=expected_end_time,
         )