fix log filtering

Author: ulixius9

ingestion/src/metadata/ingestion/source/database/databricks/connection.py

@@ -13,7 +13,6 @@
 Source connection handler
 """
 
-import logging
 from copy import deepcopy
 from functools import partial
 from typing import Optional
@@ -44,6 +43,9 @@
 )
 from metadata.ingestion.ometa.ometa_api import OpenMetadata
 from metadata.ingestion.source.database.databricks.auth import get_auth_config
+from metadata.ingestion.source.database.databricks.log_filters import (
+    suppress_user_agent_entry_deprecation_log,
+)
 from metadata.ingestion.source.database.databricks.queries import (
     DATABRICKS_GET_CATALOGS,
     DATABRICKS_SQL_STATEMENT_TEST,
@@ -60,9 +62,7 @@
 
 logger = ingestion_logger()
 
-# Suppress noisy deprecation warning from databricks-sqlalchemy using
-# the deprecated '_user_agent_entry' parameter internally
-logging.getLogger("databricks.sql.session").setLevel(logging.ERROR)
+suppress_user_agent_entry_deprecation_log()
 
 
 class DatabricksEngineWrapper:

ingestion/src/metadata/ingestion/source/database/databricks/log_filters.py

@@ -0,0 +1,43 @@
+#  Copyright 2025 Collate
+#  Licensed under the Collate Community License, Version 1.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#  https://github.com/open-metadata/OpenMetadata/blob/main/ingestion/LICENSE
+#  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.
+
+"""
+Logging filters for Databricks SQL connector noise.
+"""
+
+import logging
+
+_DATABRICKS_SESSION_LOGGER = "databricks.sql.session"
+_DEPRECATED_PARAM_FRAGMENT = "_user_agent_entry"
+_FILTER_INSTALLED_FLAG = "_om_user_agent_entry_filter_installed"
+
+
+class _UserAgentEntryDeprecationFilter(logging.Filter):
+    def filter(self, record: logging.LogRecord) -> bool:
+        try:
+            message = record.getMessage()
+        except Exception:
+            return True
+        return _DEPRECATED_PARAM_FRAGMENT not in message
+
+
+def suppress_user_agent_entry_deprecation_log() -> None:
+    """
+    Drop the `_user_agent_entry` deprecation log emitted by databricks-sqlalchemy
+    without changing the level of the `databricks.sql.session` logger, so
+    user-configured logging is preserved and other records flow through normally.
+    Idempotent: safe to call from multiple connector modules at import time.
+    """
+    target_logger = logging.getLogger(_DATABRICKS_SESSION_LOGGER)
+    if getattr(target_logger, _FILTER_INSTALLED_FLAG, False):
+        return
+    target_logger.addFilter(_UserAgentEntryDeprecationFilter())
+    setattr(target_logger, _FILTER_INSTALLED_FLAG, True)

ingestion/src/metadata/ingestion/source/database/unitycatalog/connection.py

@@ -13,7 +13,6 @@
 Source connection handler
 """
 
-import logging
 from copy import deepcopy
 from functools import partial
 from typing import Optional
@@ -50,6 +49,9 @@
 from metadata.ingestion.connections.test_connections import test_connection_steps
 from metadata.ingestion.ometa.ometa_api import OpenMetadata
 from metadata.ingestion.source.database.databricks.auth import get_auth_config
+from metadata.ingestion.source.database.databricks.log_filters import (
+    suppress_user_agent_entry_deprecation_log,
+)
 from metadata.ingestion.source.database.unitycatalog.models import DatabricksTable
 from metadata.ingestion.source.database.unitycatalog.queries import (
     UNITY_CATALOG_GET_ALL_SCHEMA_TAGS,
@@ -65,9 +67,7 @@
 
 logger = ingestion_logger()
 
-# Suppress noisy deprecation warning from databricks-sqlalchemy using
-# the deprecated '_user_agent_entry' parameter internally
-logging.getLogger("databricks.sql.session").setLevel(logging.ERROR)
+suppress_user_agent_entry_deprecation_log()
 
 
 def get_connection_url(connection: UnityCatalogConnection) -> str:

ingestion/src/metadata/ingestion/source/pipeline/databrickspipeline/connection.py

@@ -13,7 +13,6 @@
 Source connection handler
 """
 
-import logging
 from typing import Optional
 
 from metadata.generated.schema.entity.automations.workflow import (
@@ -33,11 +32,12 @@
 from metadata.ingestion.connections.test_connections import test_connection_steps
 from metadata.ingestion.ometa.ometa_api import OpenMetadata
 from metadata.ingestion.source.database.databricks.client import DatabricksClient
+from metadata.ingestion.source.database.databricks.log_filters import (
+    suppress_user_agent_entry_deprecation_log,
+)
 from metadata.utils.constants import THREE_MIN
 
-# Suppress noisy deprecation warning from databricks-sqlalchemy using
-# the deprecated '_user_agent_entry' parameter internally
-logging.getLogger("databricks.sql.session").setLevel(logging.ERROR)
+suppress_user_agent_entry_deprecation_log()
 
 
 def get_connection_url(connection: DatabricksPipelineConnection) -> str:

ingestion/tests/unit/topology/database/test_databricks_log_filters.py

@@ -0,0 +1,88 @@
+#  Copyright 2025 Collate
+#  Licensed under the Collate Community License, Version 1.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#  https://github.com/open-metadata/OpenMetadata/blob/main/ingestion/LICENSE
+#  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.
+
+"""
+Tests for the databricks.sql.session log filter helper.
+"""
+
+import logging
+
+import pytest
+
+from metadata.ingestion.source.database.databricks import log_filters
+from metadata.ingestion.source.database.databricks.log_filters import (
+    suppress_user_agent_entry_deprecation_log,
+)
+
+DATABRICKS_SESSION_LOGGER = "databricks.sql.session"
+
+
+@pytest.fixture
+def clean_logger():
+    target = logging.getLogger(DATABRICKS_SESSION_LOGGER)
+    original_filters = list(target.filters)
+    original_level = target.level
+    had_flag = hasattr(target, log_filters._FILTER_INSTALLED_FLAG)
+    flag_value = getattr(target, log_filters._FILTER_INSTALLED_FLAG, None)
+
+    target.filters = []
+    if had_flag:
+        delattr(target, log_filters._FILTER_INSTALLED_FLAG)
+
+    yield target
+
+    target.filters = original_filters
+    target.setLevel(original_level)
+    if had_flag:
+        setattr(target, log_filters._FILTER_INSTALLED_FLAG, flag_value)
+    elif hasattr(target, log_filters._FILTER_INSTALLED_FLAG):
+        delattr(target, log_filters._FILTER_INSTALLED_FLAG)
+
+
+def _emit(logger: logging.Logger, message: str) -> logging.LogRecord:
+    record = logger.makeRecord(logger.name, logging.WARNING, __file__, 0, message, None, None)
+    return record
+
+
+def test_filters_user_agent_entry_message(clean_logger):
+    suppress_user_agent_entry_deprecation_log()
+
+    record = _emit(
+        clean_logger,
+        "Parameter '_user_agent_entry' is deprecated, use 'user_agent_entry' instead",
+    )
+
+    assert clean_logger.filters, "Expected the suppression filter to be installed"
+    assert all(f.filter(record) is False for f in clean_logger.filters)
+
+
+def test_unrelated_warning_passes_through(clean_logger):
+    suppress_user_agent_entry_deprecation_log()
+
+    record = _emit(clean_logger, "Connection retry: attempt 2 of 3")
+
+    assert all(f.filter(record) is True for f in clean_logger.filters)
+
+
+def test_logger_level_is_not_modified(clean_logger):
+    clean_logger.setLevel(logging.DEBUG)
+
+    suppress_user_agent_entry_deprecation_log()
+
+    assert clean_logger.level == logging.DEBUG
+
+
+def test_helper_is_idempotent(clean_logger):
+    suppress_user_agent_entry_deprecation_log()
+    suppress_user_agent_entry_deprecation_log()
+    suppress_user_agent_entry_deprecation_log()
+
+    assert len(clean_logger.filters) == 1