D7net
Home
Console
Upload
information
Create File
Create Folder
About
Tools
:
/
opt
/
imh-python
/
lib
/
python3.9
/
site-packages
/
botocore
/
Filename :
monitoring.py
back
Copy
# Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"). You # may not use this file except in compliance with the License. A copy of # the License is located at # # http://aws.amazon.com/apache2.0/ # # or in the "license" file accompanying this file. This file is # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF # ANY KIND, either express or implied. See the License for the specific # language governing permissions and limitations under the License. import json import logging import re import time from botocore.compat import ensure_unicode, ensure_bytes, urlparse from botocore.retryhandler import EXCEPTION_MAP as RETRYABLE_EXCEPTIONS logger = logging.getLogger(__name__) class Monitor(object): _EVENTS_TO_REGISTER = [ 'before-parameter-build', 'request-created', 'response-received', 'after-call', 'after-call-error', ] def __init__(self, adapter, publisher): """Abstraction for monitoring clients API calls :param adapter: An adapter that takes event emitter events and produces monitor events :param publisher: A publisher for generated monitor events """ self._adapter = adapter self._publisher = publisher def register(self, event_emitter): """Register an event emitter to the monitor""" for event_to_register in self._EVENTS_TO_REGISTER: event_emitter.register_last(event_to_register, self.capture) def capture(self, event_name, **payload): """Captures an incoming event from the event emitter It will feed an event emitter event to the monitor's adaptor to create a monitor event and then publish that event to the monitor's publisher. """ try: monitor_event = self._adapter.feed(event_name, payload) if monitor_event: self._publisher.publish(monitor_event) except Exception as e: logger.debug( 'Exception %s raised by client monitor in handling event %s', e, event_name, exc_info=True) class MonitorEventAdapter(object): def __init__(self, time=time.time): """Adapts event emitter events to produce monitor events :type time: callable :param time: A callable that produces the current time """ self._time = time def feed(self, emitter_event_name, emitter_payload): """Feed an event emitter event to generate a monitor event :type emitter_event_name: str :param emitter_event_name: The name of the event emitted :type emitter_payload: dict :param emitter_payload: The payload to associated to the event emitted :rtype: BaseMonitorEvent :returns: A monitor event based on the event emitter events fired """ return self._get_handler(emitter_event_name)(**emitter_payload) def _get_handler(self, event_name): return getattr( self, '_handle_' + event_name.split('.')[0].replace('-', '_') ) def _handle_before_parameter_build(self, model, context, **kwargs): context['current_api_call_event'] = APICallEvent( service=model.service_model.service_id, operation=model.wire_name, timestamp=self._get_current_time(), ) def _handle_request_created(self, request, **kwargs): context = request.context new_attempt_event = context[ 'current_api_call_event'].new_api_call_attempt( timestamp=self._get_current_time()) new_attempt_event.request_headers = request.headers new_attempt_event.url = request.url context['current_api_call_attempt_event'] = new_attempt_event def _handle_response_received(self, parsed_response, context, exception, **kwargs): attempt_event = context.pop('current_api_call_attempt_event') attempt_event.latency = self._get_latency(attempt_event) if parsed_response is not None: attempt_event.http_status_code = parsed_response[ 'ResponseMetadata']['HTTPStatusCode'] attempt_event.response_headers = parsed_response[ 'ResponseMetadata']['HTTPHeaders'] attempt_event.parsed_error = parsed_response.get('Error') else: attempt_event.wire_exception = exception return attempt_event def _handle_after_call(self, context, parsed, **kwargs): context['current_api_call_event'].retries_exceeded = parsed[ 'ResponseMetadata'].get('MaxAttemptsReached', False) return self._complete_api_call(context) def _handle_after_call_error(self, context, exception, **kwargs): # If the after-call-error was emitted and the error being raised # was a retryable connection error, then the retries must have exceeded # for that exception as this event gets emitted **after** retries # happen. context['current_api_call_event'].retries_exceeded = \ self._is_retryable_exception(exception) return self._complete_api_call(context) def _is_retryable_exception(self, exception): return isinstance( exception, tuple(RETRYABLE_EXCEPTIONS['GENERAL_CONNECTION_ERROR'])) def _complete_api_call(self, context): call_event = context.pop('current_api_call_event') call_event.latency = self._get_latency(call_event) return call_event def _get_latency(self, event): return self._get_current_time() - event.timestamp def _get_current_time(self): return int(self._time() * 1000) class BaseMonitorEvent(object): def __init__(self, service, operation, timestamp): """Base monitor event :type service: str :param service: A string identifying the service associated to the event :type operation: str :param operation: A string identifying the operation of service associated to the event :type timestamp: int :param timestamp: Epoch time in milliseconds from when the event began """ self.service = service self.operation = operation self.timestamp = timestamp def __repr__(self): return '%s(%r)' % (self.__class__.__name__, self.__dict__) def __eq__(self, other): if isinstance(other, self.__class__): return self.__dict__ == other.__dict__ return False class APICallEvent(BaseMonitorEvent): def __init__(self, service, operation, timestamp, latency=None, attempts=None, retries_exceeded=False): """Monitor event for a single API call This event corresponds to a single client method call, which includes every HTTP requests attempt made in order to complete the client call :type service: str :param service: A string identifying the service associated to the event :type operation: str :param operation: A string identifying the operation of service associated to the event :type timestamp: int :param timestamp: Epoch time in milliseconds from when the event began :type latency: int :param latency: The time in milliseconds to complete the client call :type attempts: list :param attempts: The list of APICallAttempts associated to the APICall :type retries_exceeded: bool :param retries_exceeded: True if API call exceeded retries. False otherwise """ super(APICallEvent, self).__init__( service=service, operation=operation, timestamp=timestamp) self.latency = latency self.attempts = attempts if attempts is None: self.attempts = [] self.retries_exceeded = retries_exceeded def new_api_call_attempt(self, timestamp): """Instantiates APICallAttemptEvent associated to the APICallEvent :type timestamp: int :param timestamp: Epoch time in milliseconds to associate to the APICallAttemptEvent """ attempt_event = APICallAttemptEvent( service=self.service, operation=self.operation, timestamp=timestamp ) self.attempts.append(attempt_event) return attempt_event class APICallAttemptEvent(BaseMonitorEvent): def __init__(self, service, operation, timestamp, latency=None, url=None, http_status_code=None, request_headers=None, response_headers=None, parsed_error=None, wire_exception=None): """Monitor event for a single API call attempt This event corresponds to a single HTTP request attempt in completing the entire client method call. :type service: str :param service: A string identifying the service associated to the event :type operation: str :param operation: A string identifying the operation of service associated to the event :type timestamp: int :param timestamp: Epoch time in milliseconds from when the HTTP request started :type latency: int :param latency: The time in milliseconds to complete the HTTP request whether it succeeded or failed :type url: str :param url: The URL the attempt was sent to :type http_status_code: int :param http_status_code: The HTTP status code of the HTTP response if there was a response :type request_headers: dict :param request_headers: The HTTP headers sent in making the HTTP request :type response_headers: dict :param response_headers: The HTTP headers returned in the HTTP response if there was a response :type parsed_error: dict :param parsed_error: The error parsed if the service returned an error back :type wire_exception: Exception :param wire_exception: The exception raised in sending the HTTP request (i.e. ConnectionError) """ super(APICallAttemptEvent, self).__init__( service=service, operation=operation, timestamp=timestamp ) self.latency = latency self.url = url self.http_status_code = http_status_code self.request_headers = request_headers self.response_headers = response_headers self.parsed_error = parsed_error self.wire_exception = wire_exception class CSMSerializer(object): _MAX_CLIENT_ID_LENGTH = 255 _MAX_EXCEPTION_CLASS_LENGTH = 128 _MAX_ERROR_CODE_LENGTH = 128 _MAX_USER_AGENT_LENGTH = 256 _MAX_MESSAGE_LENGTH = 512 _RESPONSE_HEADERS_TO_EVENT_ENTRIES = { 'x-amzn-requestid': 'XAmznRequestId', 'x-amz-request-id': 'XAmzRequestId', 'x-amz-id-2': 'XAmzId2', } _AUTH_REGEXS = { 'v4': re.compile( r'AWS4-HMAC-SHA256 ' r'Credential=(?P<access_key>\w+)/\d+/' r'(?P<signing_region>[a-z0-9-]+)/' ), 's3': re.compile( r'AWS (?P<access_key>\w+):' ) } _SERIALIZEABLE_EVENT_PROPERTIES = [ 'service', 'operation', 'timestamp', 'attempts', 'latency', 'retries_exceeded', 'url', 'request_headers', 'http_status_code', 'response_headers', 'parsed_error', 'wire_exception', ] def __init__(self, csm_client_id): """Serializes monitor events to CSM (Client Side Monitoring) format :type csm_client_id: str :param csm_client_id: The application identifier to associate to the serialized events """ self._validate_client_id(csm_client_id) self.csm_client_id = csm_client_id def _validate_client_id(self, csm_client_id): if len(csm_client_id) > self._MAX_CLIENT_ID_LENGTH: raise ValueError( 'The value provided for csm_client_id: %s exceeds the ' 'maximum length of %s characters' % ( csm_client_id, self._MAX_CLIENT_ID_LENGTH) ) def serialize(self, event): """Serializes a monitor event to the CSM format :type event: BaseMonitorEvent :param event: The event to serialize to bytes :rtype: bytes :returns: The CSM serialized form of the event """ event_dict = self._get_base_event_dict(event) event_type = self._get_event_type(event) event_dict['Type'] = event_type for attr in self._SERIALIZEABLE_EVENT_PROPERTIES: value = getattr(event, attr, None) if value is not None: getattr(self, '_serialize_' + attr)( value, event_dict, event_type=event_type) return ensure_bytes( json.dumps(event_dict, separators=(',', ':'))) def _get_base_event_dict(self, event): return { 'Version': 1, 'ClientId': self.csm_client_id, } def _serialize_service(self, service, event_dict, **kwargs): event_dict['Service'] = service def _serialize_operation(self, operation, event_dict, **kwargs): event_dict['Api'] = operation def _serialize_timestamp(self, timestamp, event_dict, **kwargs): event_dict['Timestamp'] = timestamp def _serialize_attempts(self, attempts, event_dict, **kwargs): event_dict['AttemptCount'] = len(attempts) if attempts: self._add_fields_from_last_attempt(event_dict, attempts[-1]) def _add_fields_from_last_attempt(self, event_dict, last_attempt): if last_attempt.request_headers: # It does not matter which attempt to use to grab the region # for the ApiCall event, but SDKs typically do the last one. region = self._get_region(last_attempt.request_headers) if region is not None: event_dict['Region'] = region event_dict['UserAgent'] = self._get_user_agent( last_attempt.request_headers) if last_attempt.http_status_code is not None: event_dict['FinalHttpStatusCode'] = last_attempt.http_status_code if last_attempt.parsed_error is not None: self._serialize_parsed_error( last_attempt.parsed_error, event_dict, 'ApiCall') if last_attempt.wire_exception is not None: self._serialize_wire_exception( last_attempt.wire_exception, event_dict, 'ApiCall') def _serialize_latency(self, latency, event_dict, event_type): if event_type == 'ApiCall': event_dict['Latency'] = latency elif event_type == 'ApiCallAttempt': event_dict['AttemptLatency'] = latency def _serialize_retries_exceeded(self, retries_exceeded, event_dict, **kwargs): event_dict['MaxRetriesExceeded'] = (1 if retries_exceeded else 0) def _serialize_url(self, url, event_dict, **kwargs): event_dict['Fqdn'] = urlparse(url).netloc def _serialize_request_headers(self, request_headers, event_dict, **kwargs): event_dict['UserAgent'] = self._get_user_agent(request_headers) if self._is_signed(request_headers): event_dict['AccessKey'] = self._get_access_key(request_headers) region = self._get_region(request_headers) if region is not None: event_dict['Region'] = region if 'X-Amz-Security-Token' in request_headers: event_dict['SessionToken'] = request_headers[ 'X-Amz-Security-Token'] def _serialize_http_status_code(self, http_status_code, event_dict, **kwargs): event_dict['HttpStatusCode'] = http_status_code def _serialize_response_headers(self, response_headers, event_dict, **kwargs): for header, entry in self._RESPONSE_HEADERS_TO_EVENT_ENTRIES.items(): if header in response_headers: event_dict[entry] = response_headers[header] def _serialize_parsed_error(self, parsed_error, event_dict, event_type, **kwargs): field_prefix = 'Final' if event_type == 'ApiCall' else '' event_dict[field_prefix + 'AwsException'] = self._truncate( parsed_error['Code'], self._MAX_ERROR_CODE_LENGTH) event_dict[field_prefix + 'AwsExceptionMessage'] = self._truncate( parsed_error['Message'], self._MAX_MESSAGE_LENGTH) def _serialize_wire_exception(self, wire_exception, event_dict, event_type, **kwargs): field_prefix = 'Final' if event_type == 'ApiCall' else '' event_dict[field_prefix + 'SdkException'] = self._truncate( wire_exception.__class__.__name__, self._MAX_EXCEPTION_CLASS_LENGTH) event_dict[field_prefix + 'SdkExceptionMessage'] = self._truncate( str(wire_exception), self._MAX_MESSAGE_LENGTH) def _get_event_type(self, event): if isinstance(event, APICallEvent): return 'ApiCall' elif isinstance(event, APICallAttemptEvent): return 'ApiCallAttempt' def _get_access_key(self, request_headers): auth_val = self._get_auth_value(request_headers) _, auth_match = self._get_auth_match(auth_val) return auth_match.group('access_key') def _get_region(self, request_headers): if not self._is_signed(request_headers): return None auth_val = self._get_auth_value(request_headers) signature_version, auth_match = self._get_auth_match(auth_val) if signature_version != 'v4': return None return auth_match.group('signing_region') def _get_user_agent(self, request_headers): return self._truncate( ensure_unicode(request_headers.get('User-Agent', '')), self._MAX_USER_AGENT_LENGTH ) def _is_signed(self, request_headers): return 'Authorization' in request_headers def _get_auth_value(self, request_headers): return ensure_unicode(request_headers['Authorization']) def _get_auth_match(self, auth_val): for signature_version, regex in self._AUTH_REGEXS.items(): match = regex.match(auth_val) if match: return signature_version, match return None, None def _truncate(self, text, max_length): if len(text) > max_length: logger.debug( 'Truncating following value to maximum length of ' '%s: %s', text, max_length) return text[:max_length] return text class SocketPublisher(object): _MAX_MONITOR_EVENT_LENGTH = 8 * 1024 def __init__(self, socket, host, port, serializer): """Publishes monitor events to a socket :type socket: socket.socket :param socket: The socket object to use to publish events :type host: string :param host: The host to send events to :type port: integer :param port: The port on the host to send events to :param serializer: The serializer to use to serialize the event to a form that can be published to the socket. This must have a `serialize()` method that accepts a monitor event and return bytes """ self._socket = socket self._address = (host, port) self._serializer = serializer def publish(self, event): """Publishes a specified monitor event :type event: BaseMonitorEvent :param event: The monitor event to be sent over the publisher's socket to the desired address. """ serialized_event = self._serializer.serialize(event) if len(serialized_event) > self._MAX_MONITOR_EVENT_LENGTH: logger.debug( 'Serialized event of size %s exceeds the maximum length ' 'allowed: %s. Not sending event to socket.', len(serialized_event), self._MAX_MONITOR_EVENT_LENGTH ) return self._socket.sendto(serialized_event, self._address)