Dec 2, 2024 · Nov 4, 2024 · Nov 25, 2024 · Nov 25, 2024 · Nov 26, 2024 · Nov 26, 2024
diff --git a/google/api_core/client_logging.py b/google/api_core/client_logging.py
 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)
diff --git a/tests/unit/test_client_logging.py b/tests/unit/test_client_logging.py
 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"
Original file line number	Diff line number	Diff line change
		@@ -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 = [
Copy link Collaborator partheaNov 28, 2024 Choose a reason for hiding this comment The reason will be displayed to describe this comment to others.Learn more. One thing to note is that users will need to update`google-api-core` to get any additional logging fields in the future. Copy link ContributorAuthor ohmayrDec 2, 2024 Choose a reason for hiding this comment The reason will be displayed to describe this comment to others.Learn more. This is true. Any fields that we add in the future will require users to update`google-api-core`.
		"httpRequest",
		"rpcName",
		"serviceName",
ohmayr marked this conversation as resolved. Show resolvedHide resolved
		"credentialsType",
		"credentialInfo",
		"universeDomain",
		"request",
		"response",
		"metadata",
		"retryAttempt",
		] # Additional fields to be Logged.
ohmayr marked this conversation as resolved. Show resolvedHide resolved


		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():
ohmayr marked this conversation as resolved. Show resolvedHide resolved
		"""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:
parthea marked this conversation as resolved. Show resolvedHide resolved
		# 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 = {
ohmayr marked this conversation as resolved. Show resolvedHide resolved
		"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)
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1,140 @@
		import json
		import logging
		from unittest import mock

		from google.api_core.client_logging import (
		setup_logging,
		initialize_logging,
		StructuredLogFormatter,
		)

ohmayr marked this conversation as resolved. Show resolvedHide resolved

		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

ohmayr marked this conversation as resolved. Show resolvedHide resolved
		# 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"