This is an automated email from the ASF dual-hosted git repository. ash pushed a commit to branch structlog-in-logging-mixin in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 42b9e45cc780433d2865ee593f8df5f72b3ddfac Author: Ash Berlin-Taylor <[email protected]> AuthorDate: Wed Aug 27 23:01:00 2025 +0100 respect-old-fmt --- .github/boring-cyborg.yml | 2 +- airflow-core/pyproject.toml | 6 +- airflow-core/src/airflow/_shared/logging | 1 + .../src/airflow/jobs/triggerer_job_runner.py | 3 +- airflow-core/src/airflow/logging_config.py | 16 +- airflow-core/src/airflow/utils/cli.py | 2 +- .../src/airflow/utils/log/logging_mixin.py | 2 +- devel-common/src/tests_common/test_utils/logs.py | 7 +- pyproject.toml | 9 +- .../__init__.py => shared/logging/pyproject.toml | 36 +++ .../src/airflow_shared/logging}/__init__.py | 0 .../src/airflow_shared/logging/_noncaching.py | 22 ++ .../airflow_shared/logging/percent_formatter.py | 153 +++++++++++ .../src/airflow_shared/logging}/structlog.py | 270 +++++++++++++------ shared/logging/tests/logging/test_structlog.py | 293 +++++++++++++++++++++ task-sdk/pyproject.toml | 6 +- task-sdk/src/airflow/sdk/_shared/logging | 1 + 17 files changed, 719 insertions(+), 110 deletions(-) diff --git a/.github/boring-cyborg.yml b/.github/boring-cyborg.yml index e564ec4f982..289a9231e9b 100644 --- a/.github/boring-cyborg.yml +++ b/.github/boring-cyborg.yml @@ -421,7 +421,7 @@ labelPRBasedOnFilePath: area:Logging: - airflow-core/src/airflow/config_templates/airflow_local_settings.py - - airflow-core/src/airflow/_logging/**/* + - shared/logging/**/* - airflow-core/src/airflow/utils/log/**/* - airflow-core/docs/administration-and-deployment/logging-monitoring/logging-*.rst - airflow-core/tests/unit/utils/log/**/* diff --git a/airflow-core/pyproject.toml b/airflow-core/pyproject.toml index ea303a6c2f1..5061dab295d 100644 --- a/airflow-core/pyproject.toml +++ b/airflow-core/pyproject.toml @@ -213,8 +213,9 @@ exclude = [ ] [tool.hatch.build.targets.sdist.force-include] -"../shared/timezones/src/airflow_shared/timezones" = "src/airflow/_shared/timezones" +"../shared/logging/src/airflow_shared/logging" = "src/airflow/_shared/logging" "../shared/secrets_masker/src/airflow_shared/secrets_masker" = "src/airflow/_shared/secrets_masker" +"../shared/timezones/src/airflow_shared/timezones" = "src/airflow/_shared/timezones" [tool.hatch.build.targets.custom] path = "./hatch_build.py" @@ -282,6 +283,7 @@ apache-airflow-devel-common = { workspace = true } [tool.airflow] shared_distributions = [ - "apache-airflow-shared-timezones", + "apache-airflow-shared-logging", "apache-airflow-shared-secrets-masker", + "apache-airflow-shared-timezones", ] diff --git a/airflow-core/src/airflow/_shared/logging b/airflow-core/src/airflow/_shared/logging new file mode 120000 index 00000000000..19c320ffcf9 --- /dev/null +++ b/airflow-core/src/airflow/_shared/logging @@ -0,0 +1 @@ +../../../../shared/logging/src/airflow_shared/logging \ No newline at end of file diff --git a/airflow-core/src/airflow/jobs/triggerer_job_runner.py b/airflow-core/src/airflow/jobs/triggerer_job_runner.py index ba85d00f18a..ff01cc64396 100644 --- a/airflow-core/src/airflow/jobs/triggerer_job_runner.py +++ b/airflow-core/src/airflow/jobs/triggerer_job_runner.py @@ -160,7 +160,8 @@ class TriggererJobRunner(BaseJobRunner, LoggingMixin): sys.exit(os.EX_SOFTWARE) def _execute(self) -> int | None: - self.log.info("Starting the triggerer") + logging.getLogger("airflow.foo.bar").info("Starting the triggerer") + self.log.info("Starting the triggerer", foo="bar") try: # Kick off runner sub-process without DB access self.trigger_runner = TriggerRunnerSupervisor.start( diff --git a/airflow-core/src/airflow/logging_config.py b/airflow-core/src/airflow/logging_config.py index 64e8bdd1e0a..cce8b37f684 100644 --- a/airflow-core/src/airflow/logging_config.py +++ b/airflow-core/src/airflow/logging_config.py @@ -22,7 +22,6 @@ import warnings from importlib import import_module from typing import TYPE_CHECKING, Any -from airflow._logging.structlog import configure_structlog from airflow.configuration import conf from airflow.exceptions import AirflowConfigException from airflow.utils.module_loading import import_string @@ -86,22 +85,13 @@ def load_logging_config() -> tuple[dict[str, Any], str]: def configure_logging(): + from airflow._shared.logging.structlog import configure_logging + logging_config, logging_class_path = load_logging_config() try: - # Ensure that the password masking filter is applied to the 'task' handler - # no matter what the user did. - if "filters" in logging_config and "mask_secrets" in logging_config["filters"]: - # But if they replace the logging config _entirely_, don't try to set this, it won't work - task_handler_config = logging_config["handlers"]["task"] - - task_handler_config.setdefault("filters", []) - - if "mask_secrets" not in task_handler_config["filters"]: - task_handler_config["filters"].append("mask_secrets") - level: str = conf.get_mandatory_value("logging", "LOGGING_LEVEL").upper() # Try to init logging - configure_structlog(log_level=level, stdlib_config=logging_config) + configure_logging(log_level=level, stdlib_config=logging_config) except (ValueError, KeyError) as e: log.error("Unable to load the config, contains a configuration error.") # When there is an error in the config, escalate the exception diff --git a/airflow-core/src/airflow/utils/cli.py b/airflow-core/src/airflow/utils/cli.py index 0d790c60740..bfdde70a657 100644 --- a/airflow-core/src/airflow/utils/cli.py +++ b/airflow-core/src/airflow/utils/cli.py @@ -430,7 +430,7 @@ def suppress_logs_and_warning(f: T) -> T: if args[0].verbose: f(*args, **kwargs) else: - from airflow._logging.structlog import respect_stdlib_disable + from airflow._shared.logging.structlog import respect_stdlib_disable with warnings.catch_warnings(): warnings.simplefilter("ignore") diff --git a/airflow-core/src/airflow/utils/log/logging_mixin.py b/airflow-core/src/airflow/utils/log/logging_mixin.py index c7ffbceb681..b6fa25be453 100644 --- a/airflow-core/src/airflow/utils/log/logging_mixin.py +++ b/airflow-core/src/airflow/utils/log/logging_mixin.py @@ -28,7 +28,7 @@ from typing import IO, TYPE_CHECKING, Any, TypeVar, cast import structlog if TYPE_CHECKING: - from airflow._logging.structlog import AirflowFilteringBoundLogger + from airflow._shared.logging.structlog import AirflowFilteringBoundLogger # 7-bit C1 ANSI escape sequences ANSI_ESCAPE = re.compile(r"\x1B[@-_][0-?]*[ -/]*[@-~]") diff --git a/devel-common/src/tests_common/test_utils/logs.py b/devel-common/src/tests_common/test_utils/logs.py index 29192f3de09..b4e44b75601 100644 --- a/devel-common/src/tests_common/test_utils/logs.py +++ b/devel-common/src/tests_common/test_utils/logs.py @@ -118,7 +118,7 @@ class StructlogCapture: This restores the log levels and the disabled logging levels changed by :meth:`set_level`. """ - from airflow._logging.structlog import PER_LOGGER_LEVELS + from airflow_shared.logging.structlog import PER_LOGGER_LEVELS for logger_name, level in self._initial_logger_levels.items(): logger = logging.getLogger(logger_name) @@ -163,10 +163,7 @@ class StructlogCapture: from structlog import DropEvent from structlog._log_levels import map_method_name - from airflow._logging.structlog import ( - NamedBytesLogger, - NamedWriteLogger, - ) + from airflow_shared.logging.structlog import NamedBytesLogger, NamedWriteLogger logger_name = ( event_dict.get("logger_name") diff --git a/pyproject.toml b/pyproject.toml index c3687e7a550..70247ee1176 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1264,8 +1264,9 @@ dev = [ "apache-airflow-kubernetes-tests", "apache-airflow-task-sdk", "apache-airflow-ctl", - "apache-airflow-shared-timezones", + "apache-airflow-shared-logging", "apache-airflow-shared-secrets-masker", + "apache-airflow-shared-timezones", ] # To build docs: @@ -1312,8 +1313,9 @@ apache-airflow-helm-tests = { workspace = true } apache-airflow-kubernetes-tests = { workspace = true } apache-airflow-providers = { workspace = true } apache-aurflow-docker-stack = { workspace = true } -apache-airflow-shared-timezones = { workspace = true } +apache-airflow-shared-logging = { workspace = true } apache-airflow-shared-secrets-masker = { workspace = true } +apache-airflow-shared-timezones = { workspace = true } # Automatically generated provider workspace items (update_airflow_pyproject_toml.py) apache-airflow-providers-airbyte = { workspace = true } apache-airflow-providers-alibaba = { workspace = true } @@ -1429,8 +1431,9 @@ members = [ "task-sdk", "providers-summary-docs", "docker-stack-docs", - "shared/timezones", + "shared/logging", "shared/secrets_masker", + "shared/timezones", # Automatically generated provider workspace members (update_airflow_pyproject_toml.py) "providers/airbyte", "providers/alibaba", diff --git a/airflow-core/src/airflow/_logging/__init__.py b/shared/logging/pyproject.toml similarity index 51% copy from airflow-core/src/airflow/_logging/__init__.py copy to shared/logging/pyproject.toml index 13a83393a91..ee9791e0cbd 100644 --- a/airflow-core/src/airflow/_logging/__init__.py +++ b/shared/logging/pyproject.toml @@ -14,3 +14,39 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. + +[project] +name = "apache-airflow-shared-logging" +description = "Shared logging code for Airflow distributions" +version = "0.0" +classifiers = [ + "Private :: Do Not Upload", +] + +dependencies = [ + "structlog>=25.4.0", +] + +[dependency-groups] +dev = [ + "apache-airflow-devel-common", +] + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[tool.hatch.build.targets.wheel] +packages = ["src/airflow_shared"] + +[tool.ruff] +extend = "../../pyproject.toml" +src = ["src"] + +[tool.ruff.lint.per-file-ignores] +# Ignore Doc rules et al for anything outside of tests +"!src/*" = ["D", "S101", "TRY002"] + +[tool.ruff.lint.flake8-tidy-imports] +# Override the workspace level default +ban-relative-imports = "parents" diff --git a/airflow-core/src/airflow/_logging/__init__.py b/shared/logging/src/airflow_shared/logging/__init__.py similarity index 100% copy from airflow-core/src/airflow/_logging/__init__.py copy to shared/logging/src/airflow_shared/logging/__init__.py diff --git a/airflow-core/src/airflow/_logging/__init__.py b/shared/logging/src/airflow_shared/logging/_noncaching.py similarity index 55% rename from airflow-core/src/airflow/_logging/__init__.py rename to shared/logging/src/airflow_shared/logging/_noncaching.py index 13a83393a91..7a8e21fbe38 100644 --- a/airflow-core/src/airflow/_logging/__init__.py +++ b/shared/logging/src/airflow_shared/logging/_noncaching.py @@ -14,3 +14,25 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. +from __future__ import annotations + +import os +from typing import BinaryIO, TextIO, TypeVar + +__all__ = [ + "make_file_io_non_caching", +] + +_IO = TypeVar("_IO", TextIO, BinaryIO) + + +def make_file_io_non_caching(io: _IO) -> _IO: + try: + fd = io.fileno() + os.posix_fadvise(fd, 0, 0, os.POSIX_FADV_DONTNEED) + except Exception: + # in case either file descriptor cannot be retrieved or fadvise is not available + # we should simply return the wrapper retrieved by FileHandler's open method + # the advice to the kernel is just an advice and if we cannot give it, we won't + pass + return io diff --git a/shared/logging/src/airflow_shared/logging/percent_formatter.py b/shared/logging/src/airflow_shared/logging/percent_formatter.py new file mode 100644 index 00000000000..f78880702a1 --- /dev/null +++ b/shared/logging/src/airflow_shared/logging/percent_formatter.py @@ -0,0 +1,153 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +import collections.abc +import datetime +import os +import sys +from io import StringIO +from typing import TYPE_CHECKING + +import structlog.dev +from structlog.dev import ConsoleRenderer, Styles + +if TYPE_CHECKING: + from structlog.typing import ( + EventDict, + WrappedLogger, + ) + + +class _LazyLogRecordDict(collections.abc.Mapping): + __slots__ = ("event", "styles", "level_styles", "method_name", "no_colors") + + def __init__(self, event: EventDict, method_name: str, level_styles: dict[str, str], styles: Styles): + self.event = event + self.method_name = method_name + self.level_styles = level_styles + self.styles = styles + self.no_colors = self.styles.reset == "" + + def __getitem__(self, key): + # Roughly compatible with names from https://github.com/python/cpython/blob/v3.13.7/Lib/logging/__init__.py#L571 + # Plus with ColoredLog added in + if key == "name": + return self.event.get("logger") or self.event.get("logger_name") + if key == "levelname": + return self.event.get("level", self.method_name).upper() + if key == "pathname": + return self.event.get("pathname", "?") + if key == "filename": + path = self.event.get("pathname", "?") + return os.path.basename(path) + if key == "module": + path = self.event.get("pathname", "?") + return os.path.splitext(os.path.basename(path))[0] + if key == "lineno": + return self.event.get("lineno", 0) + if key == "asctime" or key == "created": + return ( + self.event.get("timestamp", None) + or datetime.datetime.now(tz=datetime.timezone.utc).isoformat() + ) + if key == "message": + return self.event["event"] + if key == "process": + return os.getpid() + + if key in ("red", "green", "yellow", "blue", "purple", "cyan"): + if self.no_colors: + return "" + return getattr(structlog.dev, key.upper(), "") + if key == "reset": + return self.styles.reset + if key == "log_color": + if self.no_colors: + return "" + return self.level_styles.get(self.event.get("level", self.method_name), "") + + return self.event[key] + + def __iter__(self): + return self.event.__iter__() + + def __len__(self): + return len(self.event) + + +class PercentFormatRender(ConsoleRenderer): + """A Structlog processor that uses a stdlib-like percent based format string.""" + + _fmt: str + + special_keys = { + "pathname", + "lineno", + "func_name", + "event", + "name", + "logger", + "logger_name", + "timestamp", + "level", + } + + def __init__(self, fmt: str, **kwargs): + super().__init__(**kwargs) + self._fmt = fmt + + def __call__(self, logger: WrappedLogger, method_name: str, event_dict: EventDict): + exc = event_dict.pop("exception", None) + exc_info = event_dict.pop("exc_info", None) + stack = event_dict.pop("stack", None) + params = _LazyLogRecordDict( + event_dict, + method_name, + ConsoleRenderer.get_default_level_styles(), + self._styles, + ) + + sio = StringIO() + sio.write(self._fmt % params) + + sio.write( + "".join( + " " + self._default_column_formatter(key, val) + for key, val in event_dict.items() + if key not in self.special_keys + ).rstrip(" ") + ) + + if stack is not None: + sio.write("\n" + stack) + if exc_info or exc is not None: + sio.write("\n\n" + "=" * 79 + "\n") + + if exc_info: + if isinstance(exc_info, BaseException): + exc_info = (exc_info.__class__, exc_info, exc_info.__traceback__) + if not isinstance(exc_info, tuple): + if (exc_info := sys.exc_info()) == (None, None, None): + exc_info = None + if exc_info: + self._exception_formatter(sio, exc_info) + elif exc is not None: + sio.write("\n" + exc) + + return sio.getvalue() diff --git a/airflow-core/src/airflow/_logging/structlog.py b/shared/logging/src/airflow_shared/logging/structlog.py similarity index 60% rename from airflow-core/src/airflow/_logging/structlog.py rename to shared/logging/src/airflow_shared/logging/structlog.py index 9ff0583db52..7c2e56ba81a 100644 --- a/airflow-core/src/airflow/_logging/structlog.py +++ b/shared/logging/src/airflow_shared/logging/structlog.py @@ -17,14 +17,19 @@ # under the License. from __future__ import annotations +import codecs import logging import os +import re +import sys import warnings -from functools import cache -from typing import Any, BinaryIO, Callable, Generic, Protocol, TextIO, TypeVar, Union, cast +from collections.abc import Callable +from functools import cache, partial +from typing import Any, BinaryIO, Generic, Protocol, TextIO, TypeVar, cast import pygtrie import structlog +import structlog.processors from structlog.processors import NAME_TO_LEVEL from structlog.typing import ( BindableLogger, @@ -33,8 +38,16 @@ from structlog.typing import ( WrappedLogger, ) +from ._noncaching import make_file_io_non_caching +from .percent_formatter import PercentFormatRender + log = logging.getLogger(__name__) +__all__ = [ + "configure_logging", + "structlog_processors", +] + class AirflowFilteringBoundLogger(FilteringBoundLogger, Protocol): def isEnabledFor(self, level: int): ... @@ -90,7 +103,7 @@ AirflowBoundLoggerFilteringAtCritical = _make_airflow_structlogger(NAME_TO_LEVEL PER_LOGGER_LEVELS = pygtrie.StringTrie(separator=".") PER_LOGGER_LEVELS.update( { - # Top level logging default - changed to respect config in `configure_structlog` + # Top level logging default - changed to respect config in `configure_logging` "": NAME_TO_LEVEL["info"], } ) @@ -119,6 +132,8 @@ class NamedBytesLogger(structlog.BytesLogger): def __init__(self, name: str | None = None, file: BinaryIO | None = None): self.name = name + if file is not None: + file = make_file_io_non_caching(file) super().__init__(file) @@ -127,10 +142,12 @@ class NamedWriteLogger(structlog.WriteLogger): def __init__(self, name: str | None = None, file: TextIO | None = None): self.name = name + if file is not None: + file = make_file_io_non_caching(file) super().__init__(file) -LogOutputType = TypeVar("LogOutputType", bound=Union[TextIO, BinaryIO]) +LogOutputType = TypeVar("LogOutputType", bound=TextIO | BinaryIO) class LoggerFactory(Generic[LogOutputType]): @@ -177,13 +194,26 @@ def respect_stdlib_disable(logger: Any, method_name: Any, event_dict: EventDict) @cache def structlog_processors( json_output: bool, + log_format: str = "", + colors: bool = True, ): + """ + Create the correct list of structlog processors for the given config. + + Return value is a tuple of three elements: + + 1. A list of processors shared for structlgo and stblib + 2. The final processor/renderer (one that outputs a string) for use with structlog.stdlib.ProcessorFormatter + + :meta private: + """ if json_output: timestamper = structlog.processors.MaybeTimeStamper(fmt="iso") else: timestamper = structlog.processors.MaybeTimeStamper(fmt="%Y-%m-%d %H:%M:%S.%f") - processors: list[structlog.typing.Processor] = [ + # Processors shared between stdlib handlers and structlog processors + shared_processors: list[structlog.typing.Processor] = [ respect_stdlib_disable, timestamper, structlog.contextvars.merge_contextvars, @@ -191,6 +221,15 @@ def structlog_processors( structlog.stdlib.PositionalArgumentsFormatter(), logger_name, redact_jwt, + structlog.processors.CallsiteParameterAdder( + [ + # TODO: look at what is is in the log fmt, and only include what we need. + # TODO: provide an option to specify what fields to include in JSON output too + structlog.processors.CallsiteParameter.PATHNAME, + structlog.processors.CallsiteParameter.LINENO, + structlog.processors.CallsiteParameter.FUNC_NAME, + ], + ), structlog.processors.StackInfoRenderer(), ] @@ -231,108 +270,149 @@ def structlog_processors( } return msgspec.json.encode(msg, enc_hook=default) - def json_processor(logger: Any, method_name: Any, event_dict: EventDict) -> str: - # Stdlib logging doesn't need the re-ordering, it's fine as it is - return msgspec.json.encode(event_dict).decode("utf-8") - json = structlog.processors.JSONRenderer(serializer=json_dumps) - processors.extend( + def json_processor(logger: Any, method_name: Any, event_dict: EventDict) -> str: + return json(logger, method_name, event_dict).decode("utf-8") + + shared_processors.extend( ( dict_tracebacks, structlog.processors.UnicodeDecoder(), - json, ), ) - return processors, { - "timestamper": timestamper, - "dict_tracebacks": dict_tracebacks, - "json": json_processor, - } - rich_exc_formatter = structlog.dev.RichTracebackFormatter( - # These values are picked somewhat arbitrarily to produce useful-but-compact tracebacks. If - # we ever need to change these then they should be configurable. - extra_lines=0, - max_frames=30, - indent_guides=False, - suppress=suppress, - ) + return shared_processors, json_processor, json + + if os.getenv("DEV", "") != "": + # Only use Rich in dev -- optherwise for "production" deployments it makes the logs harder to read as + # it uses lots of ANSI escapes and non ASCII characters. Simpler is better for non-dev non-JSON + exc_formatter = structlog.dev.RichTracebackFormatter( + # These values are picked somewhat arbitrarily to produce useful-but-compact tracebacks. If + # we ever need to change these then they should be configurable. + extra_lines=0, + max_frames=30, + indent_guides=False, + suppress=suppress, + ) + else: + exc_formatter = structlog.dev.plain_traceback + my_styles = structlog.dev.ConsoleRenderer.get_default_level_styles() my_styles["debug"] = structlog.dev.CYAN - console = structlog.dev.ConsoleRenderer(exception_formatter=rich_exc_formatter, level_styles=my_styles) - processors.append(console) - return processors, { - "timestamper": timestamper, - "console": console, - } + if log_format: + console = PercentFormatRender( + fmt=log_format, + exception_formatter=exc_formatter, + level_styles=my_styles, + colors=colors, + ) + else: + console = structlog.dev.ConsoleRenderer( + exception_formatter=exc_formatter, + level_styles=my_styles, + colors=colors, + ) + + return shared_processors, console, console -def configure_structlog( +def configure_logging( + *, json_output: bool = False, log_level: str = "DEBUG", - cache_logger_on_first_use: bool = True, + log_format: str | None = None, stdlib_config: dict | None = None, + colors: bool | None = None, + output: LogOutputType | None = None, + log_levels: str | dict[str, str] | None = None, + cache_logger_on_first_use: bool = True, ): + """ + Configure structlog (and stbilb's logging to send via structlog processors too). + + If percent_log_format is passed then it will be handled in a similar mode to stdlib, including + interpolations such as ``%(asctime)s`` etc. + + :param json_output: Set to true to write all logs as JSON (one per line) + :param log_level: The default log level to use for most logs + :param log_format: A percent-style log format to write non JSON logs with. + :param output: Where to write the logs too. If ``json_output`` is true this must be a binary stream + :param colors: Whether to use colors for non-JSON logs. If `None` is passed, then colors are used + if standard out is a TTY (that is, an interactive session). + + It's possible to override this behavior by setting two standard environment variables to any value + except an empty string: + + * ``FORCE_COLOR`` activates colors, regardless of where output is going. + * ``NO_COLOR`` disables colors, regardless of where the output is going and regardless the value of + ``FORCE_COLOR``. Please note that ``NO_COLOR`` disables all styling, including bold and italics. + + :param log_levels: Levels of extra loggers to configure. + + To make this easier to use, this can be a string consisting of pairs of ``<logger>=<level>`` (either + string, or space delimited) which will set the level for that specific logger. + + For example:: + + ``sqlalchemy=INFO sqlalchemy.engine=DEBUG`` + """ if "fatal" not in NAME_TO_LEVEL: NAME_TO_LEVEL["fatal"] = NAME_TO_LEVEL["critical"] + if colors is None: + colors = os.environ.get("NO_COLOR", "") == "" and ( + os.environ.get("FORCE_COLOR", "") != "" + or (sys.stdout is not None and hasattr(sys.stdout, "isatty") and sys.stdout.isatty()) + ) + stdlib_config = stdlib_config or {} - """Set up struct logging and stdlib logging config.""" PER_LOGGER_LEVELS[""] = NAME_TO_LEVEL[log_level.lower()] - if json_output: - formatter = "plain" - else: - formatter = "colored" - processors, named = structlog_processors(json_output) - timestamper = named["timestamper"] - - pre_chain: list[structlog.typing.Processor] = [ - # Add the log level and a timestamp to the event_dict if the log entry - # is not from structlog. - structlog.stdlib.add_log_level, - structlog.stdlib.add_logger_name, - timestamper, - structlog.contextvars.merge_contextvars, - redact_jwt, - ] + # Extract per-logger-tree levels and set them + if isinstance(log_levels, str): + log_from_level = partial(re.compile(r"\s*=\s*").split, maxsplit=2) + log_levels = {log: level for log, level in map(log_from_level, re.split(r"[\s,]+", log_levels))} + if log_levels: + for log, level in log_levels.items(): + try: + loglevel = NAME_TO_LEVEL[level.lower()] + except KeyError: + raise ValueError(f"Invalid log level for logger {log!r}: {level!r}") from None + else: + PER_LOGGER_LEVELS[log] = loglevel + + shared_pre_chain, for_stdlib, for_structlog = structlog_processors( + json_output, log_format=log_format, colors=colors + ) + pre_chain: list[structlog.typing.Processor] = [structlog.stdlib.add_logger_name] + shared_pre_chain # Don't cache the loggers during tests, it make it hard to capture them if "PYTEST_VERSION" in os.environ: cache_logger_on_first_use = False - color_formatter: list[structlog.typing.Processor] = [ - structlog.stdlib.ProcessorFormatter.remove_processors_meta, - drop_positional_args, - ] std_lib_formatter: list[structlog.typing.Processor] = [ + # TODO: Don't include this if we are using PercentFormatter -- it'll delete something we + # just have to recerated! structlog.stdlib.ProcessorFormatter.remove_processors_meta, drop_positional_args, + for_stdlib, ] wrapper_class = cast("type[BindableLogger]", make_filtering_logger()) if json_output: - structlog.configure( - processors=processors, - cache_logger_on_first_use=cache_logger_on_first_use, - wrapper_class=wrapper_class, - logger_factory=LoggerFactory(NamedBytesLogger), # type: ignore[type-var] - ) - - pre_chain.append(named["dict_tracebacks"]) - color_formatter.append(named["json"]) - std_lib_formatter.append(named["json"]) + logger_factory = LoggerFactory(NamedBytesLogger, io=output) else: - structlog.configure( - processors=processors, - cache_logger_on_first_use=cache_logger_on_first_use, - wrapper_class=wrapper_class, - logger_factory=LoggerFactory(NamedWriteLogger), # type: ignore[type-var] - ) - color_formatter.append(named["console"]) + logger_factory = LoggerFactory(NamedWriteLogger, io=output) + + structlog.configure( + processors=shared_pre_chain + [for_structlog], + cache_logger_on_first_use=cache_logger_on_first_use, + wrapper_class=wrapper_class, + logger_factory=logger_factory, + ) global _warnings_showwarning @@ -344,23 +424,20 @@ def configure_structlog( import logging.config config = {**stdlib_config} - config["formatters"] = {**config["formatters"]} - config["handlers"] = {**config["handlers"]} - config["loggers"] = {**config["loggers"]} + config.setdefault("version", 1) + config.setdefault("disable_existing_loggers", False) + config["formatters"] = {**config.get("formatters", {})} + config["handlers"] = {**config.get("handlers", {})} + config["loggers"] = {**config.get("loggers", {})} config["formatters"].update( { - "plain": { + "structlog": { "()": structlog.stdlib.ProcessorFormatter, + "use_get_message": False, "processors": std_lib_formatter, "foreign_pre_chain": pre_chain, "pass_foreign_args": True, }, - "colored": { - "()": structlog.stdlib.ProcessorFormatter, - "processors": color_formatter, - "foreign_pre_chain": pre_chain, - "pass_foreign_args": True, - }, } ) for section in (config["loggers"], config["handlers"]): @@ -370,12 +447,17 @@ def configure_structlog( log_config.pop("formatter", None) # log_config.pop("handlers", None) + if output and not hasattr(output, "encoding"): + # This is a BinaryIO, we need to give logging.StreamHandler a TextIO + output = codecs.lookup("utf-8").streamwriter(output) # type: ignore + config["handlers"].update( { "default": { "level": log_level.upper(), "class": "logging.StreamHandler", - "formatter": formatter, + "formatter": "structlog", + "stream": output, }, } ) @@ -423,3 +505,29 @@ def _showwarning( else: log = structlog.get_logger("py.warnings") log.warning(str(message), category=category.__name__, filename=filename, lineno=lineno) + + +if __name__ == "__main__": + configure_logging( + # json_output=True, + log_format="[%(blue)s%(asctime)s%(reset)s] {%(blue)s%(filename)s:%(reset)s%(lineno)d} %(log_color)s%(levelname)s%(reset)s - %(log_color)s%(message)s%(reset)s", + ) + log = logging.getLogger("testing.stlib") + log2 = structlog.get_logger(logger_name="testing.structlog") + + def raises(): + try: + 1 / 0 + except ZeroDivisionError: + log.exception("str") + try: + 1 / 0 + except ZeroDivisionError: + log2.exception("std") + + def main(): + log.info("in main") + log2.info("in main", key="value") + raises() + + main() diff --git a/shared/logging/tests/logging/test_structlog.py b/shared/logging/tests/logging/test_structlog.py new file mode 100644 index 00000000000..6efd206043c --- /dev/null +++ b/shared/logging/tests/logging/test_structlog.py @@ -0,0 +1,293 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +import contextlib +import io +import json +import logging +import sys +import textwrap +from datetime import UTC, datetime + +import pytest +import structlog +from structlog.dev import BLUE, BRIGHT, CYAN, DIM, GREEN, MAGENTA, RESET_ALL as RESET + +from airflow_shared.logging.structlog import configure_logging + +# We don't want to use the caplog fixture in this test, as the main purpose of this file is to capture the +# _rendered_ output of the tests to make sure it is correct. + + [email protected](autouse=True) +def set_time(time_machine): + time_machine.move_to(datetime(1985, 10, 26, tzinfo=UTC), tick=False) + + [email protected] +def structlog_config(): + @contextlib.contextmanager + def configurer(**kwargs): + prev_config = structlog.get_config() + + try: + if kwargs.get("json_output"): + buff = io.BytesIO() + else: + buff = io.StringIO() + + configure_logging(**kwargs, output=buff) + yield buff + buff.seek(0) + finally: + structlog.configure(**prev_config) + + return configurer + + [email protected]( + ("get_logger", "extra_kwargs", "extra_output"), + [ + pytest.param( + structlog.get_logger, + {"key1": "value1"}, + f" {CYAN}key1{RESET}={MAGENTA}value1{RESET}", + id="structlog", + ), + pytest.param( + logging.getLogger, + {}, + "", + id="stdlib", + ), + ], +) +def test_colorful(structlog_config, get_logger, extra_kwargs, extra_output): + with structlog_config(colors=True) as sio: + logger = get_logger("my.logger") + # Test that interoplations work too + x = "world" + logger.info("Hello %s", x, **extra_kwargs) + + lineno = sys._getframe().f_lineno - 2 + + written = sio.getvalue() + # This _might_ be a little bit too specific to structlog's ConsoleRender format + assert ( + written == f"{DIM}1985-10-26 00:00:00.000000{RESET} [{GREEN}{BRIGHT}info {RESET}]" + f" {BRIGHT}Hello world {RESET}" + f" [{RESET}{BRIGHT}{BLUE}my.logger{RESET}]{RESET}" + f" {CYAN}func_name{RESET}={MAGENTA}test_colorful{RESET}" + + extra_output + + f" {CYAN}lineno{RESET}={MAGENTA}{lineno}{RESET}" + f" {CYAN}pathname{RESET}={MAGENTA}{__file__}{RESET}" + "\n" + ) + + [email protected]( + ("get_logger", "extra_kwargs", "extra_output"), + [ + pytest.param( + structlog.get_logger, + {"key1": "value1"}, + f" {CYAN}key1{RESET}={MAGENTA}value1{RESET}", + id="structlog", + ), + pytest.param( + logging.getLogger, + {}, + "", + id="stdlib", + ), + ], +) +def test_precent_fmt(structlog_config, get_logger, extra_kwargs, extra_output): + with structlog_config(colors=True, log_format="%(blue)s[%(asctime)s]%(reset)s %(message)s") as sio: + logger = get_logger("my.logger") + logger.info("Hello", **extra_kwargs) + + written = sio.getvalue() + print(written) + assert written == f"{BLUE}[1985-10-26 00:00:00.000000]{RESET} Hello" + extra_output + "\n" + + +def test_precent_fmt_force_no_colors( + structlog_config, +): + with structlog_config( + colors=False, + log_format="%(blue)s[%(asctime)s]%(reset)s %(log_color)s%(levelname)s - %(message)s", + ) as sio: + logger = structlog.get_logger("my.logger") + logger.info("Hello", key1="value1") + + written = sio.getvalue() + assert written == "[1985-10-26 00:00:00.000000] INFO - Hello key1=value1\n" + + [email protected]( + ("get_logger", "extra_kwargs"), + [ + pytest.param( + structlog.get_logger, + {"key1": "value1"}, + id="structlog", + ), + pytest.param( + logging.getLogger, + {}, + id="stdlib", + ), + ], +) +def test_json(structlog_config, get_logger, extra_kwargs): + with structlog_config(json_output=True) as bio: + logger = get_logger("my.logger") + logger.info("Hello", **extra_kwargs) + + lineno = sys._getframe().f_lineno - 2 + + written = json.load(bio) + assert written == { + "event": "Hello", + "func_name": "test_json", + "pathname": __file__, + "lineno": lineno, + "level": "info", + **extra_kwargs, + "logger": "my.logger", + "timestamp": "1985-10-26T00:00:00Z", + } + + [email protected]( + ("get_logger"), + [ + pytest.param( + structlog.get_logger, + id="structlog", + ), + pytest.param( + logging.getLogger, + id="stdlib", + ), + ], +) +def test_precent_fmt_exc(structlog_config, get_logger, monkeypatch): + monkeypatch.setenv("DEV", "") + with structlog_config( + log_format="%(message)s", + colors=False, + ) as sio: + lineno = sys._getframe().f_lineno + 2 + try: + 1 / 0 + except ZeroDivisionError: + get_logger("logger").exception("Error") + written = sio.getvalue() + + assert written == textwrap.dedent(f"""\ + Error + Traceback (most recent call last): + File "{__file__}", line {lineno}, in test_precent_fmt_exc + 1 / 0 + ~~^~~ + ZeroDivisionError: division by zero + """) + + [email protected]( + ("get_logger"), + [ + pytest.param( + structlog.get_logger, + id="structlog", + ), + pytest.param( + logging.getLogger, + id="stdlib", + ), + ], +) +def test_json_exc(structlog_config, get_logger, monkeypatch): + with structlog_config(json_output=True) as bio: + lineno = sys._getframe().f_lineno + 2 + try: + 1 / 0 + except ZeroDivisionError: + get_logger("logger").exception("Error") + written = bio.getvalue() + + written = json.load(bio) + assert written == { + "event": "Error", + "exception": [ + { + "exc_notes": [], + "exc_type": "ZeroDivisionError", + "exc_value": "division by zero", + "exceptions": [], + "frames": [ + { + "filename": __file__, + "lineno": lineno, + "name": "test_json_exc", + }, + ], + "is_cause": False, + "is_group": False, + "syntax_error": None, + }, + ], + "func_name": "test_json_exc", + "level": "error", + "pathname": __file__, + "lineno": lineno + 2, # The line no where we logged from + "logger": "logger", + "timestamp": "1985-10-26T00:00:00Z", + } + + [email protected]( + ("levels",), + ( + pytest.param("my.logger=warn", id="str"), + pytest.param({"my.logger": "warn"}, id="dict"), + ), +) +def test_logger_filtering(structlog_config, levels): + with structlog_config( + colors=False, + log_format="[%(name)s] %(message)s", + log_level="DEBUG", + log_levels=levels, + ) as sio: + structlog.get_logger("my").info("Hello", key1="value1") + structlog.get_logger("my.logger").info("Hello", key1="value2") + structlog.get_logger("my.logger.sub").info("Hello", key1="value3") + structlog.get_logger("other.logger").info("Hello", key1="value4") + structlog.get_logger("my.logger.sub").warning("Hello", key1="value5") + + written = sio.getvalue() + assert written == textwrap.dedent("""\ + [my] Hello key1=value1 + [other.logger] Hello key1=value4 + [my.logger.sub] Hello key1=value5 + """) diff --git a/task-sdk/pyproject.toml b/task-sdk/pyproject.toml index 8140b787434..9ee7e5098fe 100644 --- a/task-sdk/pyproject.toml +++ b/task-sdk/pyproject.toml @@ -92,8 +92,9 @@ build-backend = "hatchling.build" path = "src/airflow/sdk/__init__.py" [tool.hatch.build.targets.sdist.force-include] -"../shared/timezones/src/airflow_shared/timezones" = "src/airflow/sdk/_shared/timezones" +"../shared/logging/src/airflow_shared/logging" = "src/airflow/sdk/_shared/logging" "../shared/secrets_masker/src/airflow_shared/secrets_masker" = "src/airflow/sdk/_shared/secrets_masker" +"../shared/timezones/src/airflow_shared/timezones" = "src/airflow/sdk/_shared/timezones" [tool.hatch.build.targets.wheel] packages = ["src/airflow"] @@ -235,6 +236,7 @@ tmp_path_retention_policy = "failed" [tool.airflow] shared_distributions = [ - "apache-airflow-shared-timezones", + "apache-airflow-shared-logging", "apache-airflow-shared-secrets-masker", + "apache-airflow-shared-timezones", ] diff --git a/task-sdk/src/airflow/sdk/_shared/logging b/task-sdk/src/airflow/sdk/_shared/logging new file mode 120000 index 00000000000..fda343f9a7e --- /dev/null +++ b/task-sdk/src/airflow/sdk/_shared/logging @@ -0,0 +1 @@ +../../../../../shared/logging/src/airflow_shared/logging/ \ No newline at end of file
