feat: add automatic logging config to support debug logging

google/api_core/client_logging.py

@@ -0,0 +1,143 @@
+import logging
+import json
+import os
+
+from typing import List, Optional
+
+_LOGGING_INITIALIZED = False
+_BASE_LOGGER_NAME = "google"
+
+# Fields to be included in the StructuredLogFormatter.
+#
+# TODO(https://github.com/googleapis/python-api-core/issues/761): Update this list to support additional logging fields.
+_recognized_logging_fields = [
+    "httpRequest",
+    "rpcName",
+    "serviceName",
+    "credentialsType",
+    "credentialInfo",
+    "universeDomain",
+    "request",
+    "response",
+    "metadata",
+    "retryAttempt",
+]  # Additional fields to be Logged.
+
+
+def logger_configured(logger) -> bool:
+    """Determines whether `logger` has non-default configuration
+
+    Args:
+      logger: The logger to check.
+
+    Returns:
+      bool: Whether the logger has any non-default configuration.
+    """
+    return (
+        logger.handlers != [] or logger.level != logging.NOTSET or not logger.propagate
+    )
+
+
+def initialize_logging():
+    """Initializes "google" loggers, partly based on the environment variable
+
+    Initializes the "google" logger and any loggers (at the "google"
+    level or lower) specified by the environment variable
+    GOOGLE_SDK_PYTHON_LOGGING_SCOPE, as long as none of these loggers
+    were previously configured. If any such loggers (including the
+    "google" logger) are initialized, they are set to NOT propagate
+    log events up to their parent loggers.
+
+    This initialization is executed only once, and hence the
+    environment variable is only processed the first time this
+    function is called.
+    """
+    global _LOGGING_INITIALIZED
+    if _LOGGING_INITIALIZED:
+        return
+    scopes = os.getenv("GOOGLE_SDK_PYTHON_LOGGING_SCOPE", "")
+    setup_logging(scopes)
+    _LOGGING_INITIALIZED = True
+
+
+def parse_logging_scopes(scopes: Optional[str] = None) -> List[str]:
+    """Returns a list of logger names.
+
+    Splits the single string of comma-separated logger names into a list of individual logger name strings.
+
+    Args:
+      scopes: The name of a single logger. (In the future, this will be a comma-separated list of multiple loggers.)
+
+    Returns:
+      A list of all the logger names in scopes.
+    """
+    if not scopes:
+        return []
+    # TODO(https://github.com/googleapis/python-api-core/issues/759): check if the namespace is a valid namespace.
+    # TODO(b/380481951): Support logging multiple scopes.
+    # TODO(b/380483756): Raise or log a warning for an invalid scope.
+    namespaces = [scopes]
+    return namespaces
+
+
+def configure_defaults(logger):
+    """Configures `logger` to emit structured info to stdout."""
+    if not logger_configured(logger):
+        console_handler = logging.StreamHandler()
+        logger.setLevel("DEBUG")
+        logger.propagate = False
+        formatter = StructuredLogFormatter()
+        console_handler.setFormatter(formatter)
+        logger.addHandler(console_handler)
+
+
+def setup_logging(scopes: str = ""):
+    """Sets up logging for the specified `scopes`.
+
+    If the loggers specified in `scopes` have not been previously
+    configured, this will configure them to emit structured log
+    entries to stdout, and to not propagate their log events to their
+    parent loggers. Additionally, if the "google" logger (whether it
+    was specified in `scopes` or not) was not previously configured,
+    it will also configure it to not propagate log events to the root
+    logger.
+
+    Args:
+      scopes: The name of a single logger. (In the future, this will be a comma-separated list of multiple loggers.)
+
+    """
+
+    # only returns valid logger scopes (namespaces)
+    # this list has at most one element.
+    logger_names = parse_logging_scopes(scopes)
+
+    for namespace in logger_names:
+        # This will either create a module level logger or get the reference of the base logger instantiated above.
+        logger = logging.getLogger(namespace)
+
+        # Configure default settings.
+        configure_defaults(logger)
+
+    # disable log propagation at base logger level to the root logger only if a base logger is not already configured via code changes.
+    base_logger = logging.getLogger(_BASE_LOGGER_NAME)
+    if not logger_configured(base_logger):
+        base_logger.propagate = False
+
+
+# TODO(https://github.com/googleapis/python-api-core/issues/763): Expand documentation.
+class StructuredLogFormatter(logging.Formatter):
+    # TODO(https://github.com/googleapis/python-api-core/issues/761): ensure that additional fields such as
+    # function name, file name, and line no. appear in a log output.
+    def format(self, record: logging.LogRecord):
+        log_obj = {
+            "timestamp": self.formatTime(record),
+            "severity": record.levelname,
+            "name": record.name,
+            "message": record.getMessage(),
+        }
+
+        for field_name in _recognized_logging_fields:
+            value = getattr(record, field_name, None)
+            if value is not None:
+                log_obj[field_name] = value
+        return json.dumps(log_obj)

tests/unit/test_client_logging.py

@@ -0,0 +1,140 @@
+import json
+import logging
+from unittest import mock
+
+from google.api_core.client_logging import (
+    setup_logging,
+    initialize_logging,
+    StructuredLogFormatter,
+)
+
+
+def reset_logger(scope):
+    logger = logging.getLogger(scope)
+    logger.handlers = []
+    logger.setLevel(logging.NOTSET)
+    logger.propagate = True
+
+
+def test_setup_logging_w_no_scopes():
+    with mock.patch("google.api_core.client_logging._BASE_LOGGER_NAME", "foogle"):
+        setup_logging()
+        base_logger = logging.getLogger("foogle")
+        assert base_logger.handlers == []
+        assert not base_logger.propagate
+        assert base_logger.level == logging.NOTSET
+
+    reset_logger("foogle")
+
+
+def test_setup_logging_w_base_scope():
+    with mock.patch("google.api_core.client_logging._BASE_LOGGER_NAME", "foogle"):
+        setup_logging("foogle")
+        base_logger = logging.getLogger("foogle")
+        assert isinstance(base_logger.handlers[0], logging.StreamHandler)
+        assert not base_logger.propagate
+        assert base_logger.level == logging.DEBUG
+
+    reset_logger("foogle")
+
+
+def test_setup_logging_w_configured_scope():
+    with mock.patch("google.api_core.client_logging._BASE_LOGGER_NAME", "foogle"):
+        base_logger = logging.getLogger("foogle")
+        base_logger.propagate = False
+        setup_logging("foogle")
+        assert base_logger.handlers == []
+        assert not base_logger.propagate
+        assert base_logger.level == logging.NOTSET
+
+    reset_logger("foogle")
+
+
+def test_setup_logging_w_module_scope():
+    with mock.patch("google.api_core.client_logging._BASE_LOGGER_NAME", "foogle"):
+        setup_logging("foogle.bar")
+
+        base_logger = logging.getLogger("foogle")
+        assert base_logger.handlers == []
+        assert not base_logger.propagate
+        assert base_logger.level == logging.NOTSET
+
+        module_logger = logging.getLogger("foogle.bar")
+        assert isinstance(module_logger.handlers[0], logging.StreamHandler)
+        assert not module_logger.propagate
+        assert module_logger.level == logging.DEBUG
+
+    reset_logger("foogle")
+    reset_logger("foogle.bar")
+
+
+def test_setup_logging_w_incorrect_scope():
+    with mock.patch("google.api_core.client_logging._BASE_LOGGER_NAME", "foogle"):
+        setup_logging("abc")
+
+        base_logger = logging.getLogger("foogle")
+        assert base_logger.handlers == []
+        assert not base_logger.propagate
+        assert base_logger.level == logging.NOTSET
+
+        # TODO(https://github.com/googleapis/python-api-core/issues/759): update test once we add logic to ignore an incorrect scope.
+        logger = logging.getLogger("abc")
+        assert isinstance(logger.handlers[0], logging.StreamHandler)
+        assert not logger.propagate
+        assert logger.level == logging.DEBUG
+
+    reset_logger("foogle")
+    reset_logger("abc")
+
+
+def test_initialize_logging():
+
+    with mock.patch("os.getenv", return_value="foogle.bar"):
+        with mock.patch("google.api_core.client_logging._BASE_LOGGER_NAME", "foogle"):
+            initialize_logging()
+
+            base_logger = logging.getLogger("foogle")
+            assert base_logger.handlers == []
+            assert not base_logger.propagate
+            assert base_logger.level == logging.NOTSET
+
+            module_logger = logging.getLogger("foogle.bar")
+            assert isinstance(module_logger.handlers[0], logging.StreamHandler)
+            assert not module_logger.propagate
+            assert module_logger.level == logging.DEBUG
+
+            # Check that `initialize_logging()` is a no-op after the first time by verifying that user-set configs are not modified:
+            base_logger.propagate = True
+            module_logger.propagate = True
+
+            initialize_logging()
+
+            assert base_logger.propagate
+            assert module_logger.propagate
+
+    reset_logger("foogle")
+    reset_logger("foogle.bar")
+
+
+def test_structured_log_formatter():
+    # TODO(https://github.com/googleapis/python-api-core/issues/761): Test additional fields when implemented.
+    record = logging.LogRecord(
+        name="Appelation",
+        level=logging.DEBUG,
+        msg="This is a test message.",
+        pathname="some/path",
+        lineno=25,
+        args=None,
+        exc_info=None,
+    )
+
+    # Extra fields:
+    record.rpcName = "bar"
+
+    formatted_msg = StructuredLogFormatter().format(record)
+    parsed_msg = json.loads(formatted_msg)
+
+    assert parsed_msg["name"] == "Appelation"
+    assert parsed_msg["severity"] == "DEBUG"
+    assert parsed_msg["message"] == "This is a test message."
+    assert parsed_msg["rpcName"] == "bar"