dstandish commented on code in PR #27758: URL: https://github.com/apache/airflow/pull/27758#discussion_r1059952811
########## airflow/jobs/triggerer_job.py: ########## @@ -17,26 +17,175 @@ from __future__ import annotations import asyncio +import logging import os import signal import sys import threading import time +import warnings from collections import deque -from typing import Deque +from queue import SimpleQueue +from typing import TYPE_CHECKING, Deque from sqlalchemy import func from airflow.configuration import conf from airflow.jobs.base_job import BaseJob from airflow.models.trigger import Trigger +from airflow.settings import DONOT_MODIFY_HANDLERS from airflow.stats import Stats from airflow.triggers.base import BaseTrigger, TriggerEvent from airflow.typing_compat import TypedDict +from airflow.utils.log.file_task_handler import FileTaskHandler from airflow.utils.log.logging_mixin import LoggingMixin +from airflow.utils.log.trigger_handler import ( + DropTriggerLogsFilter, + LocalQueueHandler, + TriggererHandlerWrapper, + TriggerMetadataFilter, + ctx_close_handler, + ctx_indiv_trigger, + ctx_task_instance, + ctx_trigger_id, +) from airflow.utils.module_loading import import_string from airflow.utils.session import provide_session +if TYPE_CHECKING: + from airflow.models import TaskInstance + +USING_TRIGGERER_HANDLER_WRAPPER = False +""" +If this value is true, trigger logging is configured to use TriggerHandlerWrapper + +:meta :private +""" + +logger = logging.getLogger(__name__) + + +def configure_trigger_log_handler(): + """ + Configure logging such that each trigger logs to its own file and + can be exposed through the airflow webserver. + + Generally speaking, we take the log handler configured for logger ``airflow.task``, + wrap it with TriggerHandlerWrapper, and set it as the handler for root logger. + + If there already is a handler configured for the root logger + and it supports triggers, we wrap it instead. + + :meta private: + """ + global USING_TRIGGERER_HANDLER_WRAPPER + + def supports_triggerer(handler): + return getattr(handler, "supports_triggerer", False) + + def get_task_handler_from_logger(logger_): + for h in logger_.handlers: + if isinstance(h, FileTaskHandler) and not supports_triggerer(h): + warnings.warn( + f"Handler {h.__class__.__name__} does not support " + "individual trigger logging. Please check the release notes " + "for your provider to see if a newer version supports " + "individual trigger logging." + ) + if supports_triggerer(h): + return h + + def find_suitable_task_handler(): + # check root logger then check airflow.task to see if a handler + # suitable for use with TriggerHandlerWrapper (has supports_triggerer + # attr, likely inherits from FileTaskHandler) + h = get_task_handler_from_logger(root_logger) + if not h: + # try to use handler configured from airflow task + logger.debug("No task logger configured for root logger; trying `airflow.task`.") + h = get_task_handler_from_logger(logging.getLogger("airflow.task")) + if h: + logger.debug("Using logging configuration from `airflow.task`") + if not h: + warnings.warn("Could not find log handler suitable for individual trigger logging.") + return None + return h + + def filter_trigger_logs_from_other_root_handlers(new_hdlr): + # we add context vars to log records emitted for individual triggerer logging + # we want these records to be processed by our special trigger handler wrapper + # but not by any other handlers, so we filter out these messages from + # other handlers by adding DropTriggerLogsFilter + # we could consider only adding this filter to the default console logger + # so as to leave other custom handlers alone + for h in root_logger.handlers: + if h is not new_hdlr: + h.addFilter(DropTriggerLogsFilter()) + + def add_handler_wrapper_to_root(base_handler): + # first make sure we remove from root logger if it happens to be there + # it could have come from root or airflow.task, but we only need + # to make sure we remove from root, since messages will not flow + # through airflow.task + if base_handler in root_logger.handlers: + root_logger.removeHandler(base_handler) + + h = TriggererHandlerWrapper(base_handler=base_handler, level=base_handler.level) + # just extra cautious, checking if user manually configured it there + if h not in root_logger.handlers: + root_logger.addHandler(h) + return h + + root_logger = logging.getLogger() + task_handler = find_suitable_task_handler() + if not task_handler: + return None + if TYPE_CHECKING: + assert isinstance(task_handler, FileTaskHandler) + wrapper_handler = add_handler_wrapper_to_root(task_handler) + filter_trigger_logs_from_other_root_handlers(wrapper_handler) Review Comment: Another reason I went with separate tabs is, the handlers, in their reader capacity, support streaming, reading from offset. It gets a little weird to think about what that means and how that should work when you have two different streams/files. And then add in that those files could be local, served, or remote. After the process is done it's streigforward enough. Another part is timestamps... Since we don't necessarily have control over user log format, we might parse incorrectly. Neither is necessarily a deal breaker but caused me to explore separate tabs and I was happy with the result so stuck with it. -- 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-unsubscr...@airflow.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org