"""Logging. The logging library provides the handler that sends the logs to a regular HTTPS provider. The message is also formatted in a specific format (json), which includes fields that are specific to the Orchard, such as the service name, the correlation id. """ import datetime import json import logging import logging.handlers import os import socket import sys import threading import traceback from collections.abc import MutableMapping from typing import Any, Optional, Union from urllib.parse import urlparse from ddtrace.trace import tracer from pythonjsonlogger.json import JsonFormatter from owslogger import __version__, constants LEVELS = { 100: "DEBUG", 200: "INFO", 250: "NOTICE", 300: "WARNING", 400: "ERROR", 500: "CRITICAL", } try: from requests_futures.sessions import FuturesSession # type: ignore requests_session: Optional[FuturesSession] = FuturesSession() except ImportError: requests_session = None def setup( environment: str, logger_name: str, logger_level: Union[int, str], service_name: str, service_version: str, correlation_id: Optional[Union[str, int]] = None, clear_handlers: bool = True, dsn: Optional[str] = None, ) -> Union[logging.Logger, logging.LoggerAdapter[logging.Logger]]: """Set up logging. If the correlation id is provided, this will create a logger (if not already created) and an adapter. Args: environment: the application's environment. logger_name: name of the logger. logger_level: logging level of the logger. service_name: the service name. service_version: the service version. correlation_id: optional correlation id. clear_handlers: optional clear handler list on root logger. dsn: the data source name. Returns: Logger: the logger """ current_logger = logging.getLogger(logger_name) # Clear the list of handlers to override ddtrace-run behavior if clear_handlers: logging.getLogger().handlers.clear() current_logger.setLevel(logger_level) configure_handler( current_logger, environment, service_name, service_version, dsn=dsn ) if correlation_id: context = {"correlation_id": correlation_id} return OwsLoggingAdapter(current_logger, context) return current_logger def configure_handler( logger: logging.Logger, environment: str, service_name: str, service_version: str, dsn: Optional[str] = None, ) -> None: """Configure the logger based on a provided DSN. Args: logger: the logger instance. dsn: the data source name. environment: the application's environment. service_name: the service name. service_version: the service version. """ if dsn: logger.addHandler(DSNHandler(dsn, environment, service_name, service_version)) else: handler = logging.StreamHandler(sys.stdout) formatter = DDJsonFormatter(service_name, environment, service_version) handler.setFormatter(formatter) logger.addHandler(handler) class DSNHandler(logging.Handler): """Custom DSN handler. Custom DSN handler that sends a JSON payload complying with OWS1 standard. """ def __init__( self, dsn: str, environment: str, service_name: str, service_version: str ) -> None: """HTTPSHandler. Args: dsn: the full dsn to send the log to. environment: the current environment. service_name: the service name. service_version: the service version. Returns: HTTPSHandler: the created handler. """ parsed_dsn = urlparse(dsn) if parsed_dsn.scheme != "udp" and requests_session is None: raise ImportError( "The 'requests-futures' library is required. " "Please install it using: pip install requests-futures." ) logging.Handler.__init__(self) self.dsn = dsn self._parsed_dsn = parsed_dsn self.environment = environment self.service_name = service_name self.service_version = service_version def get_full_message(self, record: logging.LogRecord) -> str: """Get the full message for a record. Some of our systems might still require this data to be sent to the dsn, so instead of sending the message of the record, we send the exception information. Args: record: the record to log. Return: mixed: the string or an object (dictionary). """ if record.exc_info: return "\n".join(traceback.format_exception(*record.exc_info)) else: return record.msg def emit(self, record: logging.LogRecord) -> None: """Emit a record. From the documentation: do whatever it takes to actually log the specified logging record. Here: we send it to the provider. The payload matches Orchard format. Args: the record to log. """ try: level_name, level_number = get_standard_level_from_record(record) # Calculate the timestamp. utc_date = datetime.datetime.fromtimestamp( record.created, tz=datetime.timezone.utc ) try: correlation_id = str(record.correlation_id) # type: ignore except AttributeError: correlation_id = None payload: dict[str, Any] = { "tag": "ows1", "timestamp": utc_date.isoformat(), "level": level_number, "level_name": level_name, "correlation_id": correlation_id, "message": self.get_full_message(record), "resources": getattr(record, "resources", {}), "service": self.service_name, "service_version": self.service_version, "environment": self.environment, "meta": { "file_name": record.filename, "function_name": record.funcName, "line": record.lineno, "logger_name": record.name, }, # connect log to datadog trace "dd": tracer.get_log_correlation_context(), } payload["dd"]["version"] = ( self.service_version or payload["dd"]["version"] or "" ) if self._parsed_dsn.scheme == "udp": s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.sendto( json.dumps(payload).encode("utf-8"), (self._parsed_dsn.hostname, self._parsed_dsn.port), ) else: requests_session.post(self.dsn, json=payload) # type: ignore except (KeyboardInterrupt, SystemExit): raise except Exception: self.handleError(record) class OwsLoggingAdapter(logging.LoggerAdapter[logging.Logger]): """Custom class for Adapters. The default adapter doesn't allow passing in an extra field on logging, which is used in our case to append labels to messages. """ def __init__(self, *args: Any, **kwargs: Any) -> None: """Create a OwsLoggingAdapter. Args: args: list of arguments. kwargs: dictionary of arguments. """ super().__init__(*args, **kwargs) self.resources: dict[str, Any] = {} self.extra_message = "" def process( self, msg: Any, kwargs: MutableMapping[str, Any] ) -> tuple[Any, MutableMapping[str, Any]]: """Process the logging message and keyword arguments passed in. The main method just sets the extra fields to be equal to self.extra. So if we need to provide local data as part of the context, we can't (it will be removed as part of the override). Args: msg: the log message. kwargs: dict of keyword args """ extra = {"resources": kwargs.pop("resources", {})} extra.update(self.extra) kwargs.update(extra=extra) return msg, kwargs def get_standard_level_from_record(record: logging.LogRecord) -> tuple[str, int]: """Get standard level from a log record. This method takes the information from the LogRecord and return a two value tuple which contains the level name and level number. Args: record: the record of the level. Returns: the level name and the level number (int). """ # python levels are in 10 ... 50, we need them in the hundred. value = record.levelno * 10 if value < 100: value = 100 if value > 500: value = 500 return LEVELS.get(value, ""), value class DDJsonFormatter(JsonFormatter): """Manage fields for logging trace.""" def __init__( self, service_name: str, env: str, service_version: str, *args: Any, **kwargs: Any, ) -> None: super().__init__(*args, **kwargs) self.service_name = service_name self.env = env self.service_version = service_version def add_fields( self, log_record: dict[str, Any], record: logging.LogRecord, message_dict: dict[str, Any], ) -> None: """Update a logging record with standard fields.""" super().add_fields(log_record, record, message_dict) # Calculate the timestamp. utc_date = datetime.datetime.fromtimestamp( record.created, tz=datetime.timezone.utc ) # datadog reserved names https://docs.datadoghq.com/logs/log_configuration/attributes_naming_convention/ # noqa log_record["timestamp"] = utc_date.isoformat() log_record["level"] = record.levelname log_record["service"] = self.service_name log_record["environment"] = self.env log_record["thread_id"] = threading.get_ident() log_record["process_id"] = os.getpid() log_record["log_type"] = constants.LOG_TYPE log_record["logger_version"] = __version__ log_record["tag"] = "ows1" # https://docs.datadoghq.com/getting_started/tagging/ log_record["tags"] = [f"log_type:{constants.LOG_TYPE}"] # connect log to datadog trace dd_context = {"dd": tracer.get_log_correlation_context()} dd_context["dd"]["version"] = ( self.service_version or dd_context["dd"]["version"] or "" ) log_record.update(dd_context) # log metadata logging_context = { "logger": { "name": record.name, "pathname": record.pathname, "file_name": record.filename, "method_name": record.funcName, "lineno": record.lineno, "thread_name": threading.get_ident(), } } log_record.update(logging_context)