ferruzzi commented on code in PR #68779:
URL: https://github.com/apache/airflow/pull/68779#discussion_r3455112840
##########
providers/amazon/src/airflow/providers/amazon/aws/log/cloudwatch_task_handler.py:
##########
@@ -118,6 +121,17 @@ def handler(self) -> watchtower.CloudWatchLogHandler:
json_serialize_default=_json_serialize or json_serialize_legacy,
)
+ @property
+ def handler(self) -> watchtower.CloudWatchLogHandler:
+ """Return the streaming handler, rebuilding it if dictConfig closed it
mid-task."""
+ # dictConfig's non-incremental reset closes every handler in
logging._handlerList,
+ # leaving this one with shutting_down=True (it then silently drops
every record).
+ # Rebuild only while the IO is live: once close() has run, keep the
closed handler so a
+ # late record is dropped instead of spawning an orphan handler +
background thread.
Review Comment:
Nit: Can you merge this into a proper docsting instead of a docstring and
several inlines?
##########
providers/amazon/src/airflow/providers/amazon/aws/log/cloudwatch_task_handler.py:
##########
@@ -127,16 +141,19 @@ def processors(self) -> tuple[structlog.typing.Processor,
...]:
logRecordFactory = getLogRecordFactory()
# The handler MUST be initted here, before the processor is actually
used to log anything.
# Otherwise, logging that occurs during the creation of the handler
can create infinite loops.
- _handler = self.handler
+ _ = self.handler
from airflow.sdk.log import relative_path_from_logger
def proc(logger: structlog.typing.WrappedLogger, method_name: str,
event: structlog.typing.EventDict):
if not logger or not (stream_name :=
relative_path_from_logger(logger)):
return event
+ # Resolve the handler on every record: configure_logging() may have
+ # closed the one built above, in which case ``handler`` rebuilds
it.
+ handler = self.handler
Review Comment:
I wonder if this is necessary instead of using `self.handler` inline below,
but maybe it avoids a race condition? I guess it's trivial to keep it.
##########
providers/amazon/src/airflow/providers/amazon/aws/log/cloudwatch_task_handler.py:
##########
@@ -151,20 +168,21 @@ def proc(logger: structlog.typing.WrappedLogger,
method_name: str, event: struct
ct = created.timestamp()
record.created = ct
record.msecs = int((ct - int(ct)) * 1000) + 0.0 # Copied from
stdlib logging
- _handler.handle(record)
+ handler.handle(record)
return event
return (proc,)
def close(self):
- # Use the flush method to ensure all logs are sent to CloudWatch.
- # Closing the handler sets `shutting_down` to True, which prevents any
further logs from being sent.
- # When `shutting_down` is True, means the logging system is in the
process of shutting down,
- # during which it attempts to flush the logs which are queued.
- if self.handler is None or self.handler.shutting_down:
+ # Terminal flush — only ever called from upload(). Mark the IO closed
first so `handler`
+ # stops rebuilding: a record arriving after teardown must be dropped,
not revive a fresh
+ # handler. Read the cached handler directly so we never build one just
to flush it.
Review Comment:
Similar nit to above; I see it was already done this way before, but can you
make this a proper docstring? It will have to start with an imperative to pass
CI, so maybe move the last sentence up to be the first line.
--
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]
