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

1# Copyright 2016 Google LLC 

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"""Common logging helpers.""" 

16 

17import logging 

18 

19from datetime import datetime 

20from datetime import timedelta 

21from datetime import timezone 

22 

23import requests 

24 

25from google.cloud.logging_v2.entries import LogEntry 

26from google.cloud.logging_v2.entries import ProtobufEntry 

27from google.cloud.logging_v2.entries import StructEntry 

28from google.cloud.logging_v2.entries import TextEntry 

29 

30try: 

31 from google.cloud.logging_v2.types import LogSeverity 

32except ImportError: # pragma: NO COVER 

33 

34 class LogSeverity(object): 

35 """Map severities for non-GAPIC usage.""" 

36 

37 DEFAULT = 0 

38 DEBUG = 100 

39 INFO = 200 

40 NOTICE = 300 

41 WARNING = 400 

42 ERROR = 500 

43 CRITICAL = 600 

44 ALERT = 700 

45 EMERGENCY = 800 

46 

47 

48_NORMALIZED_SEVERITIES = { 

49 logging.CRITICAL: LogSeverity.CRITICAL, 

50 logging.ERROR: LogSeverity.ERROR, 

51 logging.WARNING: LogSeverity.WARNING, 

52 logging.INFO: LogSeverity.INFO, 

53 logging.DEBUG: LogSeverity.DEBUG, 

54 logging.NOTSET: LogSeverity.DEFAULT, 

55} 

56 

57_TIME_FORMAT = "%Y-%m-%dT%H:%M:%S.%f%z" 

58"""Time format for timestamps used in API""" 

59 

60METADATA_URL = "http://metadata.google.internal./computeMetadata/v1/" 

61METADATA_HEADERS = {"Metadata-Flavor": "Google"} 

62 

63 

64def entry_from_resource(resource, client, loggers): 

65 """Detect correct entry type from resource and instantiate. 

66 

67 Args: 

68 resource (dict): One entry resource from API response. 

69 client (~logging_v2.client.Client): 

70 Client that owns the log entry. 

71 loggers (dict): 

72 A mapping of logger fullnames -> loggers. If the logger 

73 that owns the entry is not in ``loggers``, the entry 

74 will have a newly-created logger. 

75 

76 Returns: 

77 google.cloud.logging_v2.entries._BaseEntry: 

78 The entry instance, constructed via the resource 

79 """ 

80 if "textPayload" in resource: 

81 return TextEntry.from_api_repr(resource, client, loggers=loggers) 

82 

83 if "jsonPayload" in resource: 

84 return StructEntry.from_api_repr(resource, client, loggers=loggers) 

85 

86 if "protoPayload" in resource: 

87 return ProtobufEntry.from_api_repr(resource, client, loggers=loggers) 

88 

89 return LogEntry.from_api_repr(resource, client, loggers=loggers) 

90 

91 

92def retrieve_metadata_server(metadata_key, timeout=5): 

93 """Retrieve the metadata key in the metadata server. 

94 

95 See: https://cloud.google.com/compute/docs/storing-retrieving-metadata 

96 

97 Args: 

98 metadata_key (str): 

99 Key of the metadata which will form the url. You can 

100 also supply query parameters after the metadata key. 

101 e.g. "tags?alt=json" 

102 timeout (number): number of seconds to wait for the HTTP request 

103 

104 Returns: 

105 str: The value of the metadata key returned by the metadata server. 

106 """ 

107 url = METADATA_URL + metadata_key 

108 

109 try: 

110 response = requests.get(url, headers=METADATA_HEADERS, timeout=timeout) 

111 

112 if response.status_code == requests.codes.ok: 

113 return response.text 

114 

115 except requests.exceptions.RequestException: 

116 # Ignore the exception, connection failed means the attribute does not 

117 # exist in the metadata server. 

118 pass 

119 

120 return None 

121 

122 

123def _normalize_severity(stdlib_level): 

124 """Normalize a Python stdlib severity to LogSeverity enum. 

125 

126 Args: 

127 stdlib_level (int): 'levelno' from a :class:`logging.LogRecord` 

128 

129 Returns: 

130 int: Corresponding Stackdriver severity. 

131 """ 

132 return _NORMALIZED_SEVERITIES.get(stdlib_level, stdlib_level) 

133 

134 

135def _add_defaults_to_filter(filter_): 

136 """Modify the input filter expression to add sensible defaults. 

137 

138 Args: 

139 filter_ (str): The original filter expression 

140 

141 Returns: 

142 str: sensible default filter string 

143 """ 

144 

145 # By default, requests should only return logs in the last 24 hours 

146 yesterday = datetime.now(timezone.utc) - timedelta(days=1) 

147 time_filter = f'timestamp>="{yesterday.strftime(_TIME_FORMAT)}"' 

148 if filter_ is None: 

149 filter_ = time_filter 

150 elif "timestamp" not in filter_.lower(): 

151 filter_ = f"{filter_} AND {time_filter}" 

152 return filter_