subhramit commented on code in PR #73004:
URL: https://github.com/apache/airflow/pull/73004#discussion_r3996452588
##########
providers/amazon/src/airflow/providers/amazon/aws/log/cloudwatch_task_handler.py:
##########
@@ -168,32 +176,70 @@ def handler(self) -> watchtower.CloudWatchLogHandler:
Rebuild only while the IO is live: once :meth:`close` has run, keep
the closed handler
so a late record is dropped instead of spawning an orphan handler and
background thread.
"""
- if self._cached_handler is None or (not self._closed and
self._cached_handler.shutting_down):
- self._cached_handler = self._build_handler()
- return self._cached_handler
+ with self._stream_lock:
+ if self._cached_handler is None or (not self._closed and
self._cached_handler.shutting_down):
+ self._cached_handler = self._build_handler()
+ return self._cached_handler
+
+ def _get_stream_handler(self, stream_name: str) ->
watchtower.CloudWatchLogHandler | None:
+ """Return the live handler for ``stream_name`` while holding
``_stream_lock``."""
+ if self._closed or self._building_stream_handler or stream_name in
self._closing_streams:
+ return None
+
+ handler = self._stream_handlers.get(stream_name)
+ if handler is not None and not handler.shutting_down:
+ return handler
+
+ self._stream_handlers.pop(stream_name, None)
+ if (
+ not self._stream_handlers
+ and self._cached_handler is not None
+ and not self._cached_handler.shutting_down
+ ):
+ handler = self._cached_handler
+ handler.log_stream_name = stream_name
+ else:
+ self._building_stream_handler = True
+ try:
+ handler = self._build_handler(stream_name)
+ finally:
+ self._building_stream_handler = False
+ self._stream_handlers[stream_name] = handler
+ self._cached_handler = handler
+ return handler
+
+ def _close_stream(self, stream_name: str) -> None:
+ with self._stream_lock:
+ handler = self._stream_handlers.pop(stream_name, None)
+ if handler is None:
Review Comment:
If `path` doesn't match `stream_name`, this will return silently
Maybe worth adding a log ?
##########
providers/amazon/src/airflow/providers/amazon/aws/log/cloudwatch_task_handler.py:
##########
@@ -168,32 +176,70 @@ def handler(self) -> watchtower.CloudWatchLogHandler:
Rebuild only while the IO is live: once :meth:`close` has run, keep
the closed handler
so a late record is dropped instead of spawning an orphan handler and
background thread.
"""
- if self._cached_handler is None or (not self._closed and
self._cached_handler.shutting_down):
- self._cached_handler = self._build_handler()
- return self._cached_handler
+ with self._stream_lock:
+ if self._cached_handler is None or (not self._closed and
self._cached_handler.shutting_down):
+ self._cached_handler = self._build_handler()
+ return self._cached_handler
+
+ def _get_stream_handler(self, stream_name: str) ->
watchtower.CloudWatchLogHandler | None:
+ """Return the live handler for ``stream_name`` while holding
``_stream_lock``."""
+ if self._closed or self._building_stream_handler or stream_name in
self._closing_streams:
+ return None
+
+ handler = self._stream_handlers.get(stream_name)
+ if handler is not None and not handler.shutting_down:
+ return handler
+
+ self._stream_handlers.pop(stream_name, None)
+ if (
+ not self._stream_handlers
+ and self._cached_handler is not None
+ and not self._cached_handler.shutting_down
+ ):
+ handler = self._cached_handler
+ handler.log_stream_name = stream_name
+ else:
+ self._building_stream_handler = True
+ try:
+ handler = self._build_handler(stream_name)
+ finally:
+ self._building_stream_handler = False
+ self._stream_handlers[stream_name] = handler
+ self._cached_handler = handler
+ return handler
+
+ def _close_stream(self, stream_name: str) -> None:
+ with self._stream_lock:
+ handler = self._stream_handlers.pop(stream_name, None)
+ if handler is None:
+ return
+ self._closing_streams.add(stream_name)
+ if self._cached_handler is handler:
+ self._cached_handler =
next(reversed(self._stream_handlers.values()), None)
+
+ try:
+ handler.close()
Review Comment:
(Question for others as well) is a synchronous blocking call fine here?
This was a cheap flush before.
Could this potentially slow down the triggerer in the reproduction
conditions mentioned?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
