fix google batch logs

Author: d4l3k

torchx/schedulers/gcp_batch_scheduler.py

@@ -412,8 +412,10 @@ def log_iter(
             raise ValueError(f"app not found: {app_id}")
 
         job_uid = job.uid
-        filters = [f"labels.job_uid={job_uid}"]
-        filters.append(f"resource.labels.task_id:task/{job_uid}-group0-{k}")
+        filters = [
+            f"labels.job_uid={job_uid}",
+            f"labels.task_id:{job_uid}-group0-{k}",
+        ]
 
         if since is not None:
             filters.append(f'timestamp>="{str(since.isoformat())}"')
@@ -434,7 +436,7 @@ def _batch_log_iter(self, filter: str) -> Iterable[str]:
 
         logger = logging.Client().logger(BATCH_LOGGER_NAME)
         for entry in logger.list_entries(filter_=filter):
-            yield entry.payload
+            yield entry.payload + "\n"
 
     def _job_full_name_to_app_id(self, job_full_name: str) -> str:
         """

torchx/schedulers/test/gcp_batch_scheduler_test.py

@@ -245,7 +245,7 @@ def test_batch_log_iter(self, mock_log_client: MagicMock) -> None:
         filter = "labels.job_uid=j-82c8495f-8cc9-443c-9e33-4904fcb1test"
         lines = scheduler._batch_log_iter(filter=filter)
         for l in lines:
-            self.assertEqual(l, "log line")
+            self.assertEqual(l, "log line\n")
         mock_log_client.assert_called()
         mock_logging_client.logger.assert_called_once_with("batch_task_logs")
         mock_logger.list_entries.assert_called_once_with(filter_=filter)
@@ -433,7 +433,7 @@ def test_log_iter_calls(self) -> None:
         )
         scheduler._batch_log_iter.assert_called_once_with(
             f'labels.job_uid=j-82c8495f-8cc9-443c-9e33-4904fcb1test \
-AND resource.labels.task_id:task/j-82c8495f-8cc9-443c-9e33-4904fcb1test-group0-1 \
+AND labels.task_id:j-82c8495f-8cc9-443c-9e33-4904fcb1test-group0-1 \
 AND timestamp>="{str(datetime.fromtimestamp(0).isoformat())}" \
 AND textPayload =~ "foo.*"'
         )
@@ -444,7 +444,7 @@ def test_log_iter_values(self) -> None:
         logs = scheduler.log_iter(app_id)
         scheduler._batch_log_iter.assert_called_once_with(
             f'labels.job_uid=j-82c8495f-8cc9-443c-9e33-4904fcb1test \
-AND resource.labels.task_id:task/j-82c8495f-8cc9-443c-9e33-4904fcb1test-group0-0 \
+AND labels.task_id:j-82c8495f-8cc9-443c-9e33-4904fcb1test-group0-0 \
 AND timestamp>="{str(datetime.fromtimestamp(0).isoformat())}"'
         )
         self.assertEqual(