Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/google/cloud/logging_v2/handlers/structured_log.py: 31%

1# Copyright 2021 Google LLC All Rights Reserved. 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); 

4# you may not use this file except in compliance with the License. 

5# You may obtain a copy of the License at 

6# 

7# http://www.apache.org/licenses/LICENSE-2.0 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14 

15"""Logging handler for printing formatted structured logs to standard output. 

16""" 

17import collections 

18import json 

19import logging 

20import logging.handlers 

21 

22from google.cloud.logging_v2.handlers.handlers import CloudLoggingFilter 

23from google.cloud.logging_v2.handlers.handlers import _format_and_parse_message 

24import google.cloud.logging_v2 

25from google.cloud.logging_v2._instrumentation import _create_diagnostic_entry 

26 

27GCP_FORMAT = ( 

28 "{%(_payload_str)s" 

29 '"severity": "%(levelname)s", ' 

30 '"logging.googleapis.com/labels": %(_labels_str)s, ' 

31 '"logging.googleapis.com/trace": "%(_trace_str)s", ' 

32 '"logging.googleapis.com/spanId": "%(_span_id_str)s", ' 

33 '"logging.googleapis.com/trace_sampled": %(_trace_sampled_str)s, ' 

34 '"logging.googleapis.com/sourceLocation": %(_source_location_str)s, ' 

35 '"httpRequest": %(_http_request_str)s ' 

36 "}" 

37) 

38 

39# reserved fields taken from Structured Logging documentation: 

40# https://cloud.google.com/logging/docs/structured-logging 

41GCP_STRUCTURED_LOGGING_FIELDS = frozenset( 

42 { 

43 "severity", 

44 "httpRequest", 

45 "time", 

46 "timestamp", 

47 "timestampSeconds", 

48 "timestampNanos", 

49 "logging.googleapis.com/insertId", 

50 "logging.googleapis.com/labels", 

51 "logging.googleapis.com/operation", 

52 "logging.googleapis.com/sourceLocation", 

53 "logging.googleapis.com/spanId", 

54 "logging.googleapis.com/trace", 

55 "logging.googleapis.com/trace_sampled", 

56 } 

57) 

58 

59 

60class StructuredLogHandler(logging.StreamHandler): 

61 """Handler to format logs into the Cloud Logging structured log format, 

62 and write them to standard output 

63 """ 

64 

65 def __init__( 

66 self, 

67 *, 

68 labels=None, 

69 stream=None, 

70 project_id=None, 

71 json_encoder_cls=None, 

72 **kwargs 

73 ): 

74 """ 

75 Args: 

76 labels (Optional[dict]): Additional labels to attach to logs. 

77 stream (Optional[IO]): Stream to be used by the handler. 

78 project (Optional[str]): Project Id associated with the logs. 

79 json_encoder_cls (Optional[Type[JSONEncoder]]): Custom JSON encoder. Defaults to json.JSONEncoder 

80 """ 

81 super(StructuredLogHandler, self).__init__(stream=stream) 

82 self.project_id = project_id 

83 

84 # add extra keys to log record 

85 log_filter = CloudLoggingFilter(project=project_id, default_labels=labels) 

86 self.addFilter(log_filter) 

87 

88 class _Formatter(logging.Formatter): 

89 """Formatter to format log message without traceback""" 

90 

91 def format(self, record): 

92 """Ignore exception info to avoid duplicating it 

93 https://github.com/googleapis/python-logging/issues/382 

94 """ 

95 record.message = record.getMessage() 

96 return self.formatMessage(record) 

97 

98 # make logs appear in GCP structured logging format 

99 self._gcp_formatter = _Formatter(GCP_FORMAT) 

100 

101 self._json_encoder_cls = json_encoder_cls or json.JSONEncoder 

102 

103 def format(self, record): 

104 """Format the message into structured log JSON. 

105 Args: 

106 record (logging.LogRecord): The log record. 

107 Returns: 

108 str: A JSON string formatted for GCP structured logging. 

109 """ 

110 payload = None 

111 message = _format_and_parse_message(record, super(StructuredLogHandler, self)) 

112 

113 if isinstance(message, collections.abc.Mapping): 

114 # remove any special fields 

115 for key in list(message.keys()): 

116 if key in GCP_STRUCTURED_LOGGING_FIELDS: 

117 del message[key] 

118 # if input is a dictionary, encode it as a json string 

119 encoded_msg = json.dumps( 

120 message, ensure_ascii=False, cls=self._json_encoder_cls 

121 ) 

122 # all json.dumps strings should start and end with parentheses 

123 # strip them out to embed these fields in the larger JSON payload 

124 if len(encoded_msg) > 2: 

125 payload = encoded_msg[1:-1] + "," 

126 elif message: 

127 # properly break any formatting in string to make it json safe 

128 encoded_message = json.dumps( 

129 message, ensure_ascii=False, cls=self._json_encoder_cls 

130 ) 

131 payload = '"message": {},'.format(encoded_message) 

132 

133 record._payload_str = payload or "" 

134 # convert to GCP structured logging format 

135 gcp_payload = self._gcp_formatter.format(record) 

136 return gcp_payload 

137 

138 def emit(self, record): 

139 if google.cloud.logging_v2._instrumentation_emitted is False: 

140 self.emit_instrumentation_info() 

141 super().emit(record) 

142 

143 def emit_instrumentation_info(self): 

144 google.cloud.logging_v2._instrumentation_emitted = True 

145 diagnostic_object = _create_diagnostic_entry() 

146 struct_logger = logging.getLogger(__name__) 

147 struct_logger.addHandler(self) 

148 struct_logger.setLevel(logging.INFO) 

149 struct_logger.info(diagnostic_object.payload) 

150 struct_logger.handlers.clear()