fix(mcp): set Datadog log status from JSON levels and quiet noisy hosted loggers

airbyte/mcp/_logging.py

@@ -37,7 +37,25 @@
 
 # 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,
+}
+
+_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:
@@ -51,12 +69,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
 
 
@@ -116,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.
 
@@ -135,9 +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
+    apply_json_logger_overrides()
 
     return {"log_config": None}

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

tests/unit_tests/test_mcp_logging.py

@@ -9,35 +9,49 @@
 from collections.abc import Iterator
 from typing import Any
 
+import ddtrace  # noqa: F401
 import pytest
 
 from airbyte.mcp import _logging, http_main
 from airbyte.mcp._logging import (
     LOG_FORMAT_ENV,
     _build_json_formatter,
+    apply_json_logger_overrides,
     configure_logging,
     resolve_log_format,
 )
 
 
 @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,
+        *_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
+    }
+    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, 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:
@@ -92,6 +106,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 +127,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 +154,82 @@ 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_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}
+    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") == {}
@@ -160,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(