Coverage for /usr/local/lib/python3.10/site-packages/opal_common-0.0.0-py3.10.egg/opal_common/logger.py: 89%

1import logging 

2import sys 

3 

4from loguru import logger 

5from opal_common.config import opal_common_config 

6from opal_common.http_utils import redact_url_in_text 

7from opal_common.logging_utils.filter import ModuleFilter 

8from opal_common.logging_utils.formatter import Formatter 

9from opal_common.logging_utils.intercept import InterceptHandler 

10from opal_common.logging_utils.scrubbing import CredentialScrubbingStream 

11from opal_common.logging_utils.thirdparty import hijack_uvicorn_logs 

12from opal_common.monitoring.apm import fix_ddtrace_logging 

13 

14 

15def _scrub_record_message(record): 

16 """loguru patcher: scrub embedded URL credentials from every record's 

17 message before it reaches any sink. 

18 

19 Runs once per record (order-independent across sinks), so it also protects 

20 the optional rotating file sink, which is path-based and cannot be 

21 stream-wrapped. The stream sink is *additionally* wrapped with 

22 ``CredentialScrubbingStream`` to scrub rendered exception tracebacks, which 

23 this message-only pass does not touch. 

24 """ 

25 message = record.get("message") 

26 if message: 26 ↛ exitline 26 didn't return from function '_scrub_record_message' because the condition on line 26 was always true

27 record["message"] = redact_url_in_text(message) 

28 

29 

30def configure_logs(): 

31 """Takeover process logs and create a logger with Loguru according to the 

32 configuration.""" 

33 fix_ddtrace_logging() 

34 

35 intercept_handler = InterceptHandler() 

36 formatter = Formatter(opal_common_config.LOG_FORMAT) 

37 filter = ModuleFilter( 

38 include_list=opal_common_config.LOG_MODULE_INCLUDE_LIST, 

39 exclude_list=opal_common_config.LOG_MODULE_EXCLUDE_LIST, 

40 ) 

41 logging.basicConfig(handlers=[intercept_handler], level=0, force=True) 

42 

43 if opal_common_config.LOG_PATCH_UVICORN_LOGS: 43 ↛ 47line 43 didn't jump to line 47 because the condition on line 43 was always true

44 # Monkey patch UVICORN to use our logger 

45 hijack_uvicorn_logs(intercept_handler) 

46 # Clean slate 

47 logger.remove() 

48 # Scrub credentialed URLs from every record's message, for all sinks. 

49 logger.configure(patcher=_scrub_record_message) 

50 # Logger configuration 

51 pipe = sys.stderr if opal_common_config.LOG_PIPE_TO_STDERR else sys.stdout 

52 # Wrap the stream so embedded URL credentials are scrubbed from the final 

53 # formatted record - including third-party exception tracebacks (e.g. aiohttp 

54 # rendering ``https://user:pw@host?token=...``) that the model-layer 

55 # redaction cannot reach. Message-level scrubbing for all sinks (incl. the 

56 # rotating file sink, which cannot be wrapped) is handled by the patcher above. 

57 logger.add( 

58 CredentialScrubbingStream(pipe), 

59 filter=filter.filter, 

60 format=formatter.format, 

61 level=opal_common_config.LOG_LEVEL, 

62 backtrace=opal_common_config.LOG_TRACEBACK, 

63 diagnose=opal_common_config.LOG_DIAGNOSE, 

64 colorize=opal_common_config.LOG_COLORIZE, 

65 serialize=opal_common_config.LOG_SERIALIZE, 

66 ) 

67 # log to a file 

68 if opal_common_config.LOG_TO_FILE: 68 ↛ 69line 68 didn't jump to line 69 because the condition on line 68 was never true

69 logger.add( 

70 opal_common_config.LOG_FILE_PATH, 

71 compression=opal_common_config.LOG_FILE_COMPRESSION, 

72 retention=opal_common_config.LOG_FILE_RETENTION, 

73 rotation=opal_common_config.LOG_FILE_ROTATION, 

74 serialize=opal_common_config.LOG_FILE_SERIALIZE, 

75 level=opal_common_config.LOG_FILE_LEVEL, 

76 ) 

77 

78 

79def get_logger(name=""): 

80 """Backward compatibility to old get_logger.""" 

81 return logger