From 42166c13b5c5857d1b92ff71bc28b1544a6c1612 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Sat, 10 Oct 2026 08:31:34 +0000 Subject: [PATCH 1/2] fix(mcp): improve hosted JSON log levels --- airbyte/mcp/_logging.py | 21 ++++++--- tests/unit_tests/test_mcp_logging.py | 66 ++++++++++++++++++++++------ 2 files changed, 68 insertions(+), 19 deletions(-) diff --git a/airbyte/mcp/_logging.py b/airbyte/mcp/_logging.py index 9abaf52f8..cfb572671 100644 --- a/airbyte/mcp/_logging.py +++ b/airbyte/mcp/_logging.py @@ -37,7 +37,13 @@ # Loggers that install their own handlers with `propagate = False`, which would # bypass the root JSON handler. -_SELF_HANDLED_LOGGERS = ("fastmcp",) +_SELF_HANDLED_LOGGERS = ("fastmcp", "ddtrace") + +_JSON_LOGGER_LEVELS = { + "mcp.server.streamable_http": logging.WARNING, + "ddtrace.contrib.internal.grpc.aio_client_interceptor": logging.ERROR, + "ddtrace.llmobs._llmobs": logging.ERROR, +} def resolve_log_format() -> LogFormat: @@ -51,12 +57,14 @@ def resolve_log_format() -> LogFormat: def _add_severity(_: WrappedLogger, __: str, event_dict: EventDict) -> EventDict: - """Add the stdlib level name as `severity`. + """Add stdlib level names for Datadog and Cloud Logging. - Datadog's status remapper and Cloud Logging both read `severity`, and the - stdlib names (`DEBUG` ... `CRITICAL`) are valid values for each. + Datadog's Python pipeline sets status from `levelname`, and Cloud Logging + reads `severity`. """ - event_dict["severity"] = event_dict["_record"].levelname + levelname = event_dict["_record"].levelname + event_dict["levelname"] = levelname + event_dict["severity"] = levelname return event_dict @@ -140,4 +148,7 @@ def configure_logging(log_format: LogFormat) -> dict[str, Any]: logger.handlers.clear() logger.propagate = True + for name, level in _JSON_LOGGER_LEVELS.items(): + logging.getLogger(name).setLevel(level) + return {"log_config": None} diff --git a/tests/unit_tests/test_mcp_logging.py b/tests/unit_tests/test_mcp_logging.py index 92cb080a0..4d2bb8685 100644 --- a/tests/unit_tests/test_mcp_logging.py +++ b/tests/unit_tests/test_mcp_logging.py @@ -9,6 +9,7 @@ from collections.abc import Iterator from typing import Any +import ddtrace # noqa: F401 import pytest from airbyte.mcp import _logging, http_main @@ -22,22 +23,27 @@ @pytest.fixture def restore_logging() -> Iterator[None]: - """Restore the root and fastmcp logger state mutated by `configure_logging`.""" + """Restore the logger state mutated by `configure_logging`.""" root = logging.getLogger() - fastmcp_logger = logging.getLogger("fastmcp") - saved = ( - root.handlers[:], - root.level, - fastmcp_logger.handlers[:], - fastmcp_logger.propagate, - ) + logger_names = ("fastmcp", "ddtrace", *_logging._JSON_LOGGER_LEVELS) + logger_states = { + name: ( + logging.getLogger(name).handlers[:], + logging.getLogger(name).level, + logging.getLogger(name).propagate, + ) + for name in logger_names + } + root_handlers = root.handlers[:] + root_level = root.level yield - ( - root.handlers[:], - root.level, - fastmcp_logger.handlers[:], - fastmcp_logger.propagate, - ) = saved + root.handlers[:] = root_handlers + root.setLevel(root_level) + for name, (handlers, level, propagate) in logger_states.items(): + logger = logging.getLogger(name) + logger.handlers[:] = handlers + logger.setLevel(level) + logger.propagate = propagate def _record(**kwargs: Any) -> logging.LogRecord: @@ -92,6 +98,7 @@ def test_json_formatter_renders_standard_fields() -> None: assert payload["message"] == "hello world" assert payload["severity"] == "WARNING" + assert payload["levelname"] == payload["severity"] assert payload["logger"] == {"name": "airbyte.mcp.test"} assert payload["dd.trace_id"] == "abc123" assert payload["dd.span_id"] == "456" @@ -112,6 +119,7 @@ def test_json_formatter_keeps_traceback_in_one_line() -> None: assert "\n" not in output assert payload["severity"] == "ERROR" + assert payload["levelname"] == payload["severity"] assert payload["error"]["kind"] == "RuntimeError" assert payload["error"]["message"] == "boom" assert "Traceback" in payload["error"]["stack"] @@ -138,6 +146,36 @@ def test_configure_json_logging_routes_everything_to_root( assert messages == ["from fastmcp", "from uvicorn"] +@pytest.mark.usefixtures("restore_logging") +def test_configure_json_logging_clears_ddtrace_handler() -> None: + ddtrace_logger = logging.getLogger("ddtrace") + ddtrace_logger.addHandler(logging.StreamHandler()) + ddtrace_logger.propagate = False + + configure_logging("json") + + assert ddtrace_logger.handlers == [] + assert ddtrace_logger.propagate is True + + +@pytest.mark.usefixtures("restore_logging") +def test_quiet_logger_levels_apply_only_in_json_mode() -> None: + loggers = {name: logging.getLogger(name) for name in _logging._JSON_LOGGER_LEVELS} + for logger in loggers.values(): + logger.setLevel(logging.DEBUG) + + configure_logging("json") + + assert { + name: logger.level for name, logger in loggers.items() + } == _logging._JSON_LOGGER_LEVELS + + for logger in loggers.values(): + logger.setLevel(logging.DEBUG) + assert configure_logging("text") == {} + assert all(logger.level == logging.DEBUG for logger in loggers.values()) + + @pytest.mark.usefixtures("restore_logging") def test_configure_text_logging_leaves_uvicorn_defaults() -> None: assert configure_logging("text") == {} From e56872d39f611059e4d2c79005533cf61aba312d Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Sat, 10 Oct 2026 09:46:56 +0000 Subject: [PATCH 2/2] fix(mcp): keep actionable LLMObs warnings and re-apply logger overrides after tracing install --- airbyte/mcp/_logging.py | 40 ++++++++++--- airbyte/mcp/http_main.py | 11 +++- tests/unit_tests/test_mcp_logging.py | 85 +++++++++++++++++++++++++++- 3 files changed, 124 insertions(+), 12 deletions(-) diff --git a/airbyte/mcp/_logging.py b/airbyte/mcp/_logging.py index cfb572671..fc5a16ea4 100644 --- a/airbyte/mcp/_logging.py +++ b/airbyte/mcp/_logging.py @@ -42,9 +42,21 @@ _JSON_LOGGER_LEVELS = { "mcp.server.streamable_http": logging.WARNING, "ddtrace.contrib.internal.grpc.aio_client_interceptor": logging.ERROR, - "ddtrace.llmobs._llmobs": logging.ERROR, } +_JSON_DROPPED_MESSAGES = { + "ddtrace.llmobs._llmobs": frozenset({"Failed to extract trace/span ID from request headers."}), +} + + +class _DroppedMessageFilter(logging.Filter): + def __init__(self, messages: frozenset[str]) -> None: + super().__init__() + self.messages = messages + + def filter(self, record: logging.LogRecord) -> bool: + return record.msg not in self.messages + def resolve_log_format() -> LogFormat: """Return the log format selected by `AIRBYTE_MCP_LOG_FORMAT`.""" @@ -124,6 +136,24 @@ def _build_json_formatter() -> logging.Formatter: ) +def apply_json_logger_overrides() -> None: + """Must also run after tracing install because importing ddtrace reinstalls its handler.""" + for name in _SELF_HANDLED_LOGGERS: + logger = logging.getLogger(name) + logger.handlers.clear() + logger.propagate = True + + for name, level in _JSON_LOGGER_LEVELS.items(): + logging.getLogger(name).setLevel(level) + + for name, messages in _JSON_DROPPED_MESSAGES.items(): + logger = logging.getLogger(name) + if not any( + isinstance(logger_filter, _DroppedMessageFilter) for logger_filter in logger.filters + ): + logger.addFilter(_DroppedMessageFilter(messages)) + + def configure_logging(log_format: LogFormat) -> dict[str, Any]: """Configure process-wide logging and return matching uvicorn overrides. @@ -143,12 +173,6 @@ def configure_logging(log_format: LogFormat) -> dict[str, Any]: root.handlers[:] = [handler] root.setLevel(logging.INFO) - for name in _SELF_HANDLED_LOGGERS: - logger = logging.getLogger(name) - logger.handlers.clear() - logger.propagate = True - - for name, level in _JSON_LOGGER_LEVELS.items(): - logging.getLogger(name).setLevel(level) + apply_json_logger_overrides() return {"log_config": None} diff --git a/airbyte/mcp/http_main.py b/airbyte/mcp/http_main.py index 7e3cb8321..100cc1b22 100644 --- a/airbyte/mcp/http_main.py +++ b/airbyte/mcp/http_main.py @@ -157,7 +157,11 @@ client_credentials_enabled, wrap_if_enabled, ) -from airbyte.mcp._logging import configure_logging, resolve_log_format +from airbyte.mcp._logging import ( + apply_json_logger_overrides, + configure_logging, + resolve_log_format, +) from airbyte.mcp._telemetry import McpRequestTelemetryMiddleware from airbyte.mcp._transport_security import ( HTTP_HOST_ENV, @@ -284,10 +288,13 @@ def main() -> None: """Start the Airbyte MCP server with HTTP transport.""" from airbyte.mcp._otel import SessionIdHeaderDigest, install # noqa: PLC0415 - uvicorn_config = configure_logging(resolve_log_format()) + log_format = resolve_log_format() + uvicorn_config = configure_logging(log_format) set_hosted_mcp_mode() app.instructions = build_mcp_server_instructions(hosted=True) install(app) + if log_format == "json": + apply_json_logger_overrides() # When deployed behind a path-stripping LB (MCP_SERVER_URL has a path # component like /cloud-mcp), serve the MCP endpoint at root so the diff --git a/tests/unit_tests/test_mcp_logging.py b/tests/unit_tests/test_mcp_logging.py index 4d2bb8685..de1fd2afa 100644 --- a/tests/unit_tests/test_mcp_logging.py +++ b/tests/unit_tests/test_mcp_logging.py @@ -16,6 +16,7 @@ from airbyte.mcp._logging import ( LOG_FORMAT_ENV, _build_json_formatter, + apply_json_logger_overrides, configure_logging, resolve_log_format, ) @@ -25,12 +26,18 @@ def restore_logging() -> Iterator[None]: """Restore the logger state mutated by `configure_logging`.""" root = logging.getLogger() - logger_names = ("fastmcp", "ddtrace", *_logging._JSON_LOGGER_LEVELS) + logger_names = ( + "fastmcp", + "ddtrace", + *_logging._JSON_LOGGER_LEVELS, + *_logging._JSON_DROPPED_MESSAGES, + ) logger_states = { name: ( logging.getLogger(name).handlers[:], logging.getLogger(name).level, logging.getLogger(name).propagate, + logging.getLogger(name).filters[:], ) for name in logger_names } @@ -39,11 +46,12 @@ def restore_logging() -> Iterator[None]: yield root.handlers[:] = root_handlers root.setLevel(root_level) - for name, (handlers, level, propagate) in logger_states.items(): + for name, (handlers, level, propagate, filters) in logger_states.items(): logger = logging.getLogger(name) logger.handlers[:] = handlers logger.setLevel(level) logger.propagate = propagate + logger.filters[:] = filters def _record(**kwargs: Any) -> logging.LogRecord: @@ -158,6 +166,52 @@ def test_configure_json_logging_clears_ddtrace_handler() -> None: assert ddtrace_logger.propagate is True +@pytest.mark.usefixtures("restore_logging") +def test_json_logger_filter_drops_only_missing_header_warning( + capsys: pytest.CaptureFixture[str], +) -> None: + logger = logging.getLogger("ddtrace.llmobs._llmobs") + logger.setLevel(logging.WARNING) + + configure_logging("json") + + logger.warning("Failed to extract trace/span ID from request headers.") + logger.warning("Failed to flush LLMObs spans and evaluation metrics.") + + lines = capsys.readouterr().out.splitlines() + assert len(lines) == 1 + payload = json.loads(lines[0]) + assert payload["message"] == "Failed to flush LLMObs spans and evaluation metrics." + assert payload["levelname"] == "WARNING" + + +@pytest.mark.usefixtures("restore_logging") +def test_json_logger_overrides_are_idempotent_and_text_mode_does_not_add_filter() -> ( + None +): + logger = logging.getLogger("ddtrace.llmobs._llmobs") + assert not any( + isinstance(logger_filter, _logging._DroppedMessageFilter) + for logger_filter in logger.filters + ) + + assert configure_logging("text") == {} + assert not any( + isinstance(logger_filter, _logging._DroppedMessageFilter) + for logger_filter in logger.filters + ) + + apply_json_logger_overrides() + apply_json_logger_overrides() + assert ( + sum( + isinstance(logger_filter, _logging._DroppedMessageFilter) + for logger_filter in logger.filters + ) + == 1 + ) + + @pytest.mark.usefixtures("restore_logging") def test_quiet_logger_levels_apply_only_in_json_mode() -> None: loggers = {name: logging.getLogger(name) for name in _logging._JSON_LOGGER_LEVELS} @@ -198,6 +252,33 @@ def test_main_passes_uvicorn_config(monkeypatch: pytest.MonkeyPatch) -> None: assert captured["uvicorn_config"] is sentinel +@pytest.mark.usefixtures("restore_logging") +def test_main_reapplies_json_logger_overrides_after_install( + monkeypatch: pytest.MonkeyPatch, +) -> None: + captured: dict[str, Any] = {} + ddtrace_logger = logging.getLogger("ddtrace") + monkeypatch.setenv(LOG_FORMAT_ENV, "json") + monkeypatch.setattr(http_main.app, "instructions", http_main.app.instructions) + monkeypatch.setattr(http_main.app, "middleware", list(http_main.app.middleware)) + monkeypatch.setattr(http_main, "set_hosted_mcp_mode", lambda: None) + + def install_with_ddtrace_handler(app: Any) -> None: + ddtrace_logger.addHandler(logging.StreamHandler()) + ddtrace_logger.propagate = False + + monkeypatch.setattr("airbyte.mcp._otel.install", install_with_ddtrace_handler) + monkeypatch.setattr( + http_main, "run_mcp_http_server", lambda app, **kwargs: captured.update(kwargs) + ) + + http_main.main() + + assert captured["uvicorn_config"] == {"log_config": None} + assert ddtrace_logger.handlers == [] + assert ddtrace_logger.propagate is True + + def test_main_fails_fast_on_invalid_format(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setenv(LOG_FORMAT_ENV, "yaml") monkeypatch.setattr(