587 lines
20 KiB
Python
587 lines
20 KiB
Python
# 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_bytes, ensure_unicode, urlparse
|
|
from botocore.retryhandler import EXCEPTION_MAP as RETRYABLE_EXCEPTIONS
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class Monitor:
|
|
_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:
|
|
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:
|
|
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 f'{self.__class__.__name__}({self.__dict__!r})'
|
|
|
|
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().__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().__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:
|
|
_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(
|
|
f'The value provided for csm_client_id: {csm_client_id} exceeds '
|
|
f'the maximum length of {self._MAX_CLIENT_ID_LENGTH} characters'
|
|
)
|
|
|
|
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:
|
|
_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)
|