You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "bujjibabukatta (via GitHub)" <gi...@apache.org> on 2026/07/28 17:30:17 UTC
[PR] fix(amazon): include deferred-task trigger logs when reading from Clo… [airflow]
bujjibabukatta opened a new pull request, #70635:
URL: https://github.com/apache/airflow/pull/70635
Title
fix(amazon): include deferred-task trigger logs when reading from CloudWatch (#70317)
Body
Closes #70317.
Root cause
CloudWatchRemoteLogIO.stream() (and the legacy CloudwatchTaskHandler._read_remote_logs()) only ever read the task's own CloudWatch stream. A deferred task's triggerer logs are written to a separate stream, <task log stream>.trigger.<triggerer job id>.log (FileTaskHandler.add_triggerer_suffix). While the task is DEFERRED, the UI tails those logs live from the triggerer over HTTP, which is why they're visible during the deferral — but once the task finishes, that live path goes away and nothing ever looked in the separate CloudWatch stream. The local-file backend doesn't have this problem because FileTaskHandler._read_from_local() uses glob(worker_log_path.name + "*"), which picks up attempt=1.log and attempt=1.log.trigger.*.log for free. CloudWatch has no glob equivalent, so the trigger stream has to be discovered explicitly.
Fix
Added AwsLogsHook.describe_log_streams(): a synchronous, paginated wrapper around DescribeLogStreams with a prefix filter (mirrors the existing async variant's ResourceNotFoundException → [] handling for a not-yet-existing log group).
CloudWatchRemoteLogIO.stream() now looks up any <stream>.trigger.*.log streams via that prefix scan and reads each one the same way it reads the base stream, merging them into the returned log groups. Skipped while the task is actively DEFERRED, since the live HTTP path already covers that state and the extra API call would just be repeated on every UI poll for a potentially long-running deferral.
Multiple trigger streams (a task deferred, resumed, and deferred again — even by different triggerer processes) are sorted numerically by job id, not lexicographically — job id 7 must sort before job id 200. (Caught this via test — an initial plain-string sort got it backwards.)
Applied the identical fix to the legacy CloudwatchTaskHandler._read_remote_logs() path (still registered in provider.yaml for older-style log-handler configs). Also removed a pre-existing dead call there (self.io.read(...), whose result was immediately discarded and redone manually right below it — wasting a CloudWatch API call on every read).
Trigger-stream discovery failures (e.g. a permissions issue) are reported as an extra message rather than raising, consistent with the existing error handling on the base stream.
Backward compatibility
When no trigger streams exist (the common, non-deferred case), behavior and API call count are unchanged — verified via test that the discovery call only fires when needed and returns [] cleanly for a non-existent/empty-prefix case.
Testing
12 new tests added to test_cloudwatch_task_handler.py, covering: base+trigger merge, correct numeric ordering across multiple trigger streams, the DEFERRED-state skip, graceful handling of a DescribeLogStreams failure, the legacy handler path, correct colon-replacement in the discovery prefix, and hook-level pagination (tested with 54+ streams) and missing-log-group handling.
All 42 tests in the file pass (33 existing + 12 new — some existing tests exercise overlapping paths).
ruff check, ruff format --check, and mypy all pass clean on the three changed files.
Trade-off worth flagging for reviewers
This adds one DescribeLogStreams API call per log read for tasks that are not currently DEFERRED (previously: zero). For a task that was never deferred, this returns an empty list almost instantly, but it is still an extra round-trip. Happy to gate this further (e.g. only when try_number matches a TaskInstanceHistory row known to have gone through DEFERRED) if that trade-off is a concern — went with the simpler, always-check approach first since it's the correct and easy default and mirrors what the issue reporter's own tested workaround did.
AI Disclosure
This contribution used AI assistance.
Model(s) used: Claude
AI was used for:
Helping write this PR description.
Running the test suite locally to make sure everything passes before submitting.
Reviewing the implementation from a performance perspective.
The implementation and code changes were developed and validated by me.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(amazon): include deferred-task trigger logs when reading from Clo… [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on PR #70635:
URL: https://github.com/apache/airflow/pull/70635#issuecomment-5108864482
> This adds one DescribeLogStreams API call per log read for tasks that are not currently DEFERRED
Not sure if this is an issue worth making changes, but maybe worth keeping an eye on if we go with this plan: DescribeLogStreams is a fairly low-quota API at 25 TPS per account per region. I don't know how likely a large environment is to hit that but it seems like a low quota to hit?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(amazon): include deferred-task trigger logs when reading from Clo… [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #70635:
URL: https://github.com/apache/airflow/pull/70635#discussion_r3668841147
##########
providers/amazon/src/airflow/providers/amazon/aws/log/cloudwatch_task_handler.py:
##########
@@ -417,18 +474,37 @@ def _read_remote_logs(
self, task_instance, try_number, metadata=None
) -> tuple[LogSourceInfo, LogMessages]:
stream_name = self._render_filename(task_instance, try_number)
- messages, logs = self.io.read(stream_name, task_instance)
messages = [
f"Reading remote log from Cloudwatch log_group: {self.io.log_group} log_stream: {stream_name}"
]
+ logs: LogMessages = []
try:
events = self.io.get_cloudwatch_logs(stream_name, task_instance)
- logs = ["\n".join(self._event_to_str(event) for event in events)]
+ logs.append("\n".join(self._event_to_str(event) for event in events))
except Exception as e:
- logs = []
messages.append(str(e))
+ # See CloudWatchRemoteLogIO.stream() for why deferred-task trigger logs need a
+ # separate lookup -- the same gap applies to this (legacy) handler class.
+ if getattr(task_instance, "state", None) != TaskInstanceState.DEFERRED:
+ try:
+ trigger_stream_names = self.io._get_trigger_stream_names(stream_name)
Review Comment:
Calling a "private" method from another class is a code smell, should `io._get_trigger_stream_names` be renamed to drop the leading underscore? It also looks like a genuinely useful helper that may be reused elsewhere if it wasn't private.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(amazon): include deferred-task trigger logs when reading from Clo… [airflow]
Posted by "bujjibabukatta (via GitHub)" <gi...@apache.org>.
bujjibabukatta commented on PR #70635:
URL: https://github.com/apache/airflow/pull/70635#issuecomment-5127977930
Hey @ferruzzi, thanks for digging into this so carefully — went through all four:
The re-deferral thing was a genuine bug, glad you caught it. Fixed it so we only skip the current triggerer's stream while DEFERRED, not the whole lookup — so logs from an earlier deferral don't disappear anymore. Added a test to lock that in.
Renamed _get_trigger_stream_names to get_trigger_stream_names (dropped the underscore), and did the same for the new shared helper.
Cleaned up the parametrize test too — it's just a plain test now, no more unused param.
And on the quota thing — heads up that the re-deferral fix actually makes DescribeLogStreams get called a bit more often now, not less, since it runs during DEFERRED too. Not opposed to gating it down the line (caching, or keying off TaskInstanceHistory) if it ends up mattering in practice, just didn't want to over-engineer it without real numbers.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(amazon): include deferred-task trigger logs when reading from Clo… [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #70635:
URL: https://github.com/apache/airflow/pull/70635#discussion_r3668790812
##########
providers/amazon/src/airflow/providers/amazon/aws/log/cloudwatch_task_handler.py:
##########
@@ -271,8 +272,64 @@ def stream(self, relative_path: str, ti: RuntimeTI) -> StreamingLogResponse:
except Exception as e:
messages.append(str(e))
+ # A deferred task's triggerer logs are stored under a *separate* stream
+ # (`<task log stream>.trigger.<job id>.log`) rather than the task's own stream --
+ # see FileTaskHandler.add_triggerer_suffix / TriggerLoggingFactory. The local-file
+ # reader picks these up for free via `glob(worker_log_path.name + "*")`
+ # (FileTaskHandler._read_from_local); CloudWatch has no glob equivalent, so they
+ # have to be discovered explicitly via a ListLogStreams-style prefix scan.
+ #
+ # Skip this while the task is actively DEFERRED: the UI already tails those logs
+ # live from the triggerer over HTTP in that state (FileTaskHandler
+ # ._read_from_logs_server), so the extra CloudWatch API call would be both redundant
+ # and, for a long-running deferral, repeated on every UI poll.
Review Comment:
This may be an edge case, but what happens if the task is deferred again? `_get_log_retrieval_url` builds the served-log URL from `ti.triggerer_job.id` so the UI is only tailing the logs of the current deferred task. Do we need some way to persist the logs from previous deferrals into the main logs?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(amazon): include deferred-task trigger logs when reading from Clo… [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #70635:
URL: https://github.com/apache/airflow/pull/70635#discussion_r3693687555
##########
providers/amazon/src/airflow/providers/amazon/aws/log/cloudwatch_task_handler.py:
##########
@@ -271,8 +274,98 @@ def stream(self, relative_path: str, ti: RuntimeTI) -> StreamingLogResponse:
except Exception as e:
messages.append(str(e))
+ def _read_trigger_stream(name: str) -> RawLogStream:
+ return (
+ self._parse_log_event_as_dumped_json(event) for event in self.get_cloudwatch_logs(name, ti)
+ )
+
+ extra_messages, extra_logs = self.merge_trigger_logs(relative_path, ti, _read_trigger_stream)
+ messages.extend(extra_messages)
+ logs.extend(extra_logs)
+
return messages, logs
+ def get_trigger_stream_names(self, relative_path: str) -> list[str]:
+ """
+ Return the names of any trigger log streams associated with ``relative_path``.
+
+ A task instance can be deferred and resumed more than once -- even by different
+ triggerer processes -- so there may be more than one ``.trigger.<job id>.log``
+ stream for a single attempt. Sorted numerically by job id so multiple deferrals of
+ the same attempt read back in the order they actually happened (job ids are assigned
+ from a monotonically increasing DB sequence); anything that doesn't parse as an
+ integer job id sorts after, by name, rather than raising.
+ """
+ prefix = f"{relative_path.replace(':', '_')}.trigger."
+ streams = self.hook.describe_log_streams(log_group=self.log_group, log_stream_name_prefix=prefix)
+ names = [name for s in streams if (name := s.get("logStreamName"))]
+
+ def _sort_key(name: str) -> tuple[int, int | str]:
+ tail = name[len(prefix) :]
+ job_id_str = tail[: -len(".log")] if tail.endswith(".log") else tail
+ if job_id_str.isdigit():
+ return (0, int(job_id_str))
+ return (1, name)
+
+ return sorted(names, key=_sort_key)
+
+ def merge_trigger_logs(
+ self,
+ relative_path: str,
+ ti: RuntimeTI,
+ read_stream: Callable[[str], T],
+ ) -> tuple[list[str], list[T]]:
+ """
+ Discover trigger log streams for ``relative_path`` and read each with ``read_stream``.
+
+ Shared by :meth:`stream` and the legacy :meth:`CloudwatchTaskHandler._read_remote_logs`
+ so both read/format trigger-log content in their own way (raw JSON-line generators vs.
+ joined strings) while sharing the discovery/ordering/exclusion logic. See :meth:`stream`
+ for why this lookup exists at all -- CloudWatch has no glob() equivalent to find trigger
+ streams the way the local-file reader does.
+
+ While the task instance is actively DEFERRED, the stream for the *currently active*
+ triggerer job is excluded: the UI already tails that one live from the triggerer over
+ HTTP (``FileTaskHandler._read_from_logs_server``), so re-reading it here would just be
+ redundant, and on every UI poll during a long-running deferral. Streams from any
+ *earlier* deferral of the same attempt (a task deferred, resumed, and deferred again)
+ are still included -- the live HTTP tail only ever covers the current triggerer job, so
+ without this those earlier logs would otherwise be invisible while re-deferred.
+
+ :param relative_path: The task's own (base) log stream name.
+ :param ti: The task instance being read for.
+ :param read_stream: Called with each trigger stream's name; return its content in
+ whatever form the caller needs (e.g. a generator of parsed lines, or a joined str).
+ :return: (extra messages to append, one result per successfully-read trigger stream).
+ """
+ messages: list[str] = []
+ results: list[T] = []
+ try:
+ trigger_stream_names = self.get_trigger_stream_names(relative_path)
+ except Exception as e:
+ messages.append(f"Could not list trigger log streams for {relative_path}: {e}")
+ return messages, results
+
+ if getattr(ti, "state", None) == TaskInstanceState.DEFERRED:
+ current_job_id = getattr(getattr(ti, "triggerer_job", None), "id", None)
+ if current_job_id is not None:
+ live_suffix = f".trigger.{current_job_id}.log"
+ trigger_stream_names = [
+ name for name in trigger_stream_names if not name.endswith(live_suffix)
+ ]
Review Comment:
So many stacked `getattrs` here makes me think there should be a better way. Can you try this, and see if it works? It may need a little tweaking, but should be cleaner, I think:
```python
if (ti.state == TaskInstanceState.DEFERRED) and (job := getattr(ti, "triggerer_job", None)):
live_stream = f"{relative_path.replace(':', '_')}.trigger.{job.id}.log"
trigger_stream_names = [name for name in trigger_stream_names if name != live_stream]
```
Plus, hardcoding the prefix assembly here makes me think we should likely move that builder into a shared helper at some point, but that can be done in a different PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(amazon): include deferred-task trigger logs when reading from Clo… [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #70635:
URL: https://github.com/apache/airflow/pull/70635#issuecomment-5746422896
This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(amazon): include deferred-task trigger logs when reading from Clo… [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #70635: fix(amazon): include deferred-task trigger logs when reading from Clo…
URL: https://github.com/apache/airflow/pull/70635
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(amazon): include deferred-task trigger logs when reading from Clo… [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #70635:
URL: https://github.com/apache/airflow/pull/70635#discussion_r3668196711
##########
providers/amazon/src/airflow/providers/amazon/aws/log/cloudwatch_task_handler.py:
##########
@@ -417,18 +474,37 @@ def _read_remote_logs(
self, task_instance, try_number, metadata=None
) -> tuple[LogSourceInfo, LogMessages]:
stream_name = self._render_filename(task_instance, try_number)
- messages, logs = self.io.read(stream_name, task_instance)
messages = [
f"Reading remote log from Cloudwatch log_group: {self.io.log_group} log_stream: {stream_name}"
]
+ logs: LogMessages = []
try:
events = self.io.get_cloudwatch_logs(stream_name, task_instance)
- logs = ["\n".join(self._event_to_str(event) for event in events)]
+ logs.append("\n".join(self._event_to_str(event) for event in events))
except Exception as e:
- logs = []
messages.append(str(e))
+ # See CloudWatchRemoteLogIO.stream() for why deferred-task trigger logs need a
+ # separate lookup -- the same gap applies to this (legacy) handler class.
+ if getattr(task_instance, "state", None) != TaskInstanceState.DEFERRED:
Review Comment:
This code looks very similar to the block above, any chance to factorize?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(amazon): include deferred-task trigger logs when reading from Clo… [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #70635:
URL: https://github.com/apache/airflow/pull/70635#discussion_r3669150998
##########
providers/amazon/tests/unit/amazon/aws/log/test_cloudwatch_task_handler.py:
##########
@@ -594,6 +594,168 @@ def _raise_access_denied(*args, **kwargs):
with pytest.raises(ClientError):
list(self.cloudwatch_task_handler.io.get_cloudwatch_logs(self.remote_log_stream, self.ti))
+ def _write_trigger_test_events(self, stream_name, messages):
+ """Create (idempotently) the log group and write events to a stream.
+
+ Unlike generate_log_events(), this tolerates the log group already existing --
+ several regression tests below write to multiple streams within the same test.
+ """
+ with contextlib.suppress(ClientError):
+ self.conn.create_log_group(logGroupName=self.remote_log_group)
+ self.conn.create_log_stream(logGroupName=self.remote_log_group, logStreamName=stream_name)
+ self.conn.put_log_events(
+ logGroupName=self.remote_log_group, logStreamName=stream_name, logEvents=messages
+ )
+
+ # --- Regression tests for https://github.com/apache/airflow/issues/70317 ---
+ # Deferred-task trigger logs are written to a *separate* CloudWatch stream
+ # (`<task log stream>.trigger.<job id>.log`), which .stream()/._read_remote_logs()
+ # must discover and merge in alongside the base stream -- CloudWatch has no glob()
+ # equivalent to find them automatically the way the local-file reader does.
+
+ def test_stream_includes_trigger_log_stream(self):
+ current_time = int(time.time()) * 1000
+ self._write_trigger_test_events(
+ self.remote_log_stream, [{"timestamp": current_time, "message": "base"}]
+ )
+ trigger_stream = f"{self.remote_log_stream}.trigger.42.log"
+ self._write_trigger_test_events(trigger_stream, [{"timestamp": current_time, "message": "trigger"}])
+
+ self.ti.state = State.SUCCESS
+ messages, logs = self.cloudwatch_task_handler.io.stream(self.remote_log_stream, self.ti)
+
+ assert len(logs) == 2, "expected the base stream plus the one trigger stream"
+ assert any("base" in line for line in logs[0])
+ assert any("trigger" in line for line in logs[1])
+ assert any(trigger_stream in m for m in messages)
+
+ def test_stream_orders_multiple_trigger_streams_numerically_by_job_id(self):
+ # job id 7 happened before job id 200 (a task deferred, resumed, and deferred
+ # again). A plain lexicographic sort would put "200" before "7".
+ current_time = int(time.time()) * 1000
+ self._write_trigger_test_events(
+ self.remote_log_stream, [{"timestamp": current_time, "message": "base"}]
+ )
+ self._write_trigger_test_events(
+ f"{self.remote_log_stream}.trigger.200.log",
+ [{"timestamp": current_time, "message": "second deferral"}],
+ )
+ self._write_trigger_test_events(
+ f"{self.remote_log_stream}.trigger.7.log",
+ [{"timestamp": current_time, "message": "first deferral"}],
+ )
+
+ self.ti.state = State.SUCCESS
+ names = self.cloudwatch_task_handler.io._get_trigger_stream_names(self.remote_log_stream)
+
+ assert names == [
+ f"{self.remote_log_stream}.trigger.7.log",
+ f"{self.remote_log_stream}.trigger.200.log",
+ ]
+
+ def test_stream_skips_trigger_discovery_while_deferred(self):
+ # While DEFERRED, the UI already tails trigger logs live from the triggerer over
+ # HTTP (FileTaskHandler._read_from_logs_server) -- discovery here would be both
+ # redundant and, over a long deferral, repeated on every poll.
+ current_time = int(time.time()) * 1000
+ generate_log_events(
+ self.conn,
+ self.remote_log_group,
+ self.remote_log_stream,
+ [{"timestamp": current_time, "message": "base"}],
+ )
+ self._write_trigger_test_events(
+ f"{self.remote_log_stream}.trigger.1.log",
+ [{"timestamp": current_time, "message": "live trigger log"}],
+ )
+
+ self.ti.state = State.DEFERRED
+ messages, logs = self.cloudwatch_task_handler.io.stream(self.remote_log_stream, self.ti)
+
+ assert len(logs) == 1, "trigger-stream discovery must be skipped while DEFERRED"
+
+ def test_get_trigger_stream_names_empty_when_none_exist(self):
+ current_time = int(time.time()) * 1000
+ generate_log_events(
+ self.conn,
+ self.remote_log_group,
+ self.remote_log_stream,
+ [{"timestamp": current_time, "message": "base"}],
+ )
+ assert self.cloudwatch_task_handler.io._get_trigger_stream_names(self.remote_log_stream) == []
+
+ def test_stream_tolerates_trigger_discovery_failure(self):
+ # A DescribeLogStreams failure (e.g. a permissions issue) must not take down the
+ # base stream read -- it should be reported as an extra message, not an exception.
+ self.ti.state = State.SUCCESS
+ with mock.patch.object(
+ AwsLogsHook,
+ "describe_log_streams",
+ side_effect=ClientError({"Error": {"Code": "AccessDeniedException"}}, "DescribeLogStreams"),
+ ):
+ messages, logs = self.cloudwatch_task_handler.io.stream(self.remote_log_stream, self.ti)
+
+ assert len(logs) == 1 # base stream still returned
+ assert any("Could not list trigger log streams" in m for m in messages)
+
+ def test_read_remote_logs_legacy_handler_includes_trigger_logs(self):
+ """The same fix applied to the legacy CloudwatchTaskHandler._read_remote_logs() path."""
+ current_time = int(time.time()) * 1000
+ self._write_trigger_test_events(
+ self.remote_log_stream, [{"timestamp": current_time, "message": "base"}]
+ )
+ self._write_trigger_test_events(
+ f"{self.remote_log_stream}.trigger.42.log",
+ [{"timestamp": current_time, "message": "trigger"}],
+ )
+
+ self.ti.state = State.SUCCESS
+ messages, logs = self.cloudwatch_task_handler._read_remote_logs(self.ti, self.ti.try_number)
+
+ assert len(logs) == 2
+ assert "base" in logs[0]
+ assert "trigger" in logs[1]
+
+ @mock.patch.object(AwsLogsHook, "describe_log_streams")
+ def test_get_trigger_stream_names_uses_colon_replaced_prefix(self, mock_describe):
+ # The stored stream name always has ":" replaced with "_" (CloudWatch stream names
+ # can't contain colons); the discovery prefix must match that, not the raw path.
+ mock_describe.return_value = []
+ raw_path = "dag_id=a/run_id=scheduled__2024-01-01T00:00:00+00:00/task_id=t/attempt=1.log"
+ self.cloudwatch_task_handler.io._get_trigger_stream_names(raw_path)
+ mock_describe.assert_called_once_with(
+ log_group=self.remote_log_group,
+ log_stream_name_prefix=f"{raw_path.replace(':', '_')}.trigger.",
+ )
+
+ @pytest.mark.parametrize(
+ ("stream_names", "expected_error_code"),
+ [
+ pytest.param([], "ResourceNotFoundException", id="missing_log_group_returns_empty"),
+ ],
+ )
+ def test_describe_log_streams_missing_log_group_returns_empty(self, stream_names, expected_error_code):
Review Comment:
Parameterizing with only one param case? It also doesn't look like `expected_error_code` is used int he test unless I'm missing something
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org