Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1import json
2import logging
3import os
4from collections.abc import Mapping
5from typing import List, Optional
7_LOGGING_INITIALIZED = False
8_BASE_LOGGER_NAME = "google"
10# Fields to be included in the StructuredLogFormatter.
11#
12# TODO(https://github.com/googleapis/python-api-core/issues/761): Update this list to support additional logging fields.
13_recognized_logging_fields = [
14 # Application payload debugging fields:
15 "httpRequest",
16 "rpcName",
17 "serviceName",
18 "credentialsType",
19 "credentialsInfo",
20 "universeDomain",
21 "request",
22 "response",
23 "metadata",
24 "retryAttempt",
25 "httpResponse",
26 # Diagnostic error and trace correlation fields:
27 "url.full",
28 "rpc.method",
29 "exception.type",
30 "exception.message",
31 "exception.stacktrace",
32 "rpc.system.name",
33 "rpc.response.status_code",
34 "error.type",
35 "status.message",
36 "gcp.errors.domain",
37 "gcp.errors.metadata",
38 "trace_id",
39 "span_id",
40 "logging.googleapis.com/trace",
41 "logging.googleapis.com/spanId",
42 "logging.googleapis.com/trace_sampled",
43 # Dynamic metadata prefixes:
44 "gcp.errors.metadata.*",
45] # Additional fields to be Logged.
47# Pre-compiled allowlists for O(1) exact lookups and C-level prefix matching.
48_EXACT_FIELDS = frozenset(f for f in _recognized_logging_fields if not f.endswith("*"))
49_PREFIX_FIELDS = tuple(f[:-1] for f in _recognized_logging_fields if f.endswith("*"))
52def logger_configured(logger) -> bool:
53 """Determines whether `logger` has non-default configuration
55 Args:
56 logger: The logger to check.
58 Returns:
59 bool: Whether the logger has any non-default configuration.
60 """
61 return (
62 logger.handlers != [] or logger.level != logging.NOTSET or not logger.propagate
63 )
66def initialize_logging():
67 """Initializes "google" loggers, partly based on the environment variable
69 Initializes the "google" logger and any loggers (at the "google"
70 level or lower) specified by the environment variable
71 GOOGLE_SDK_PYTHON_LOGGING_SCOPE, as long as none of these loggers
72 were previously configured. If any such loggers (including the
73 "google" logger) are initialized, they are set to NOT propagate
74 log events up to their parent loggers.
76 This initialization is executed only once, and hence the
77 environment variable is only processed the first time this
78 function is called.
79 """
80 global _LOGGING_INITIALIZED
81 if _LOGGING_INITIALIZED:
82 return
83 scopes = os.getenv("GOOGLE_SDK_PYTHON_LOGGING_SCOPE", "")
84 setup_logging(scopes)
85 _LOGGING_INITIALIZED = True
88def parse_logging_scopes(scopes: Optional[str] = None) -> List[str]:
89 """Returns a list of logger names.
91 Splits the single string of comma-separated logger names into a list of individual logger name strings.
93 Args:
94 scopes: The name of a single logger. (In the future, this will be a comma-separated list of multiple loggers.)
96 Returns:
97 A list of all the logger names in scopes.
98 """
99 if not scopes:
100 return []
101 # TODO(https://github.com/googleapis/python-api-core/issues/759): check if the namespace is a valid namespace.
102 # TODO(b/380481951): Support logging multiple scopes.
103 # TODO(b/380483756): Raise or log a warning for an invalid scope.
104 namespaces = [scopes]
105 return namespaces
108def configure_defaults(logger):
109 """Configures `logger` to emit structured info to stdout."""
110 if not logger_configured(logger):
111 console_handler = logging.StreamHandler()
112 logger.setLevel("DEBUG")
113 logger.propagate = False
114 formatter = StructuredLogFormatter()
115 console_handler.setFormatter(formatter)
116 logger.addHandler(console_handler)
119def setup_logging(scopes: str = ""):
120 """Sets up logging for the specified `scopes`.
122 If the loggers specified in `scopes` have not been previously
123 configured, this will configure them to emit structured log
124 entries to stdout, and to not propagate their log events to their
125 parent loggers. Additionally, if the "google" logger (whether it
126 was specified in `scopes` or not) was not previously configured,
127 it will also configure it to not propagate log events to the root
128 logger.
130 Args:
131 scopes: The name of a single logger. (In the future, this will be a comma-separated list of multiple loggers.)
133 """
135 # only returns valid logger scopes (namespaces)
136 # this list has at most one element.
137 logger_names = parse_logging_scopes(scopes)
139 for namespace in logger_names:
140 # This will either create a module level logger or get the reference of the base logger instantiated above.
141 logger = logging.getLogger(namespace)
143 # Configure default settings.
144 configure_defaults(logger)
146 # disable log propagation at base logger level to the root logger only if a base logger is not already configured via code changes.
147 base_logger = logging.getLogger(_BASE_LOGGER_NAME)
148 if not logger_configured(base_logger):
149 base_logger.propagate = False
152# TODO(https://github.com/googleapis/python-api-core/issues/763): Expand documentation.
153class StructuredLogFormatter(logging.Formatter):
154 # TODO(https://github.com/googleapis/python-api-core/issues/761): ensure that additional fields such as
155 # function name, file name, and line no. appear in a log output.
156 def format(self, record: logging.LogRecord):
157 log_obj = {
158 "timestamp": self.formatTime(record),
159 "severity": record.levelname,
160 "name": record.name,
161 "message": record.getMessage(),
162 }
164 # Extract exact allowed fields via getattr() (supports @property and class-level attributes)
165 for key in _EXACT_FIELDS:
166 value = getattr(record, key, None)
167 if value is not None:
168 log_obj[key] = value
170 # Extract dynamic wildcard prefix fields (e.g. gcp.errors.metadata.*) from instance __dict__
171 for key, value in vars(record).items():
172 if value is not None and key.startswith(_PREFIX_FIELDS):
173 log_obj[key] = value
175 # Unpack record.json_fields last so explicit user-provided structured fields
176 # take precedence over default LogRecord attributes (consistent with google-cloud-logging).
177 json_fields = getattr(record, "json_fields", None)
178 if isinstance(json_fields, Mapping):
179 for key, value in json_fields.items():
180 if value is not None:
181 log_obj[key] = value
182 return json.dumps(log_obj)