Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/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, *, labels=None, stream=None, project_id=None, json_encoder_cls=None 

67 ): 

68 """ 

69 Args: 

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

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

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

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

74 """ 

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

76 self.project_id = project_id 

77 

78 # add extra keys to log record 

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

80 self.addFilter(log_filter) 

81 

82 class _Formatter(logging.Formatter): 

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

84 

85 def format(self, record): 

86 """Ignore exception info to avoid duplicating it 

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

88 """ 

89 record.message = record.getMessage() 

90 return self.formatMessage(record) 

91 

92 # make logs appear in GCP structured logging format 

93 self._gcp_formatter = _Formatter(GCP_FORMAT) 

94 

95 self._json_encoder_cls = json_encoder_cls or json.JSONEncoder 

96 

97 def format(self, record): 

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

99 Args: 

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

101 Returns: 

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

103 """ 

104 payload = None 

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

106 

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

108 # remove any special fields 

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

110 if key in GCP_STRUCTURED_LOGGING_FIELDS: 

111 del message[key] 

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

113 encoded_msg = json.dumps( 

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

115 ) 

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

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

118 if len(encoded_msg) > 2: 

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

120 elif message: 

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

122 encoded_message = json.dumps( 

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

124 ) 

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

126 

127 record._payload_str = payload or "" 

128 # convert to GCP structured logging format 

129 gcp_payload = self._gcp_formatter.format(record) 

130 return gcp_payload 

131 

132 def emit(self, record): 

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

134 self.emit_instrumentation_info() 

135 super().emit(record) 

136 

137 def emit_instrumentation_info(self): 

138 google.cloud.logging_v2._instrumentation_emitted = True 

139 diagnostic_object = _create_diagnostic_entry() 

140 struct_logger = logging.getLogger(__name__) 

141 struct_logger.addHandler(self) 

142 struct_logger.setLevel(logging.INFO) 

143 struct_logger.info(diagnostic_object.payload) 

144 struct_logger.handlers.clear()