provider.py (6356B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\packages\content\client\flatbuffer\provider.py 2 import logging 3 import zlib 4 from abc import ABCMeta 5 from content.client.flatbuffer.errors import GenericBadRequestError, GenericUnauthorizedError, GenericForbiddenError, GenericNotFoundError, GenericInternalServerError, GenericError, ServiceUnavailableError 6 from content.client.provider import BaseVersionedContentProvider 7 from eve.common.script.util.retry import retry_with_exponential_backoff 8 from stackless_response_router.exceptions import TimeoutException, UnpackException 9 from semantic_version import SemanticVersion 10 from httplib import OK, BAD_REQUEST, UNAUTHORIZED, FORBIDDEN, NOT_FOUND, INTERNAL_SERVER_ERROR, responses as httplib_responses, HTTPSConnection 11 logger = logging.getLogger('flat_buffer_provider') 12 13 def separate_host_and_request(url): 14 split = url.split('/') 15 host = split[2] 16 split = split[3:] 17 request = '' 18 for host_segment in split: 19 request += '/' + host_segment 20 21 return (host, request) 22 23 24 class FlatbufferContentProvider(BaseVersionedContentProvider): 25 __metaclass__ = ABCMeta 26 request_type = None 27 response_type = None 28 29 def __init__(self, public_gateway = None): 30 self._public_gateway = public_gateway 31 32 @property 33 def public_gateway(self): 34 return self._public_gateway 35 36 def fetch(self, local_version, latest_version = None): 37 request = self.construct_request(local_version) 38 info_log = 'Request: FlatbufferContentProvider: fetch: {request}\n'.format(request=request) 39 response_code, response = self._send_request(request, self.response_type, 'FlatbufferContentProvider: fetch') 40 info_log += 'Response: FlatbufferContentProvider: fetch: {response}'.format(response=response) 41 logger.info(info_log) 42 if response: 43 url, version, checksum = self.parse_response(response) 44 if local_version and local_version == version: 45 return (None, None) 46 cdn_response = self.retrieve_data_from_cdn(url, checksum) 47 return (version, cdn_response) 48 return (None, None) 49 50 def construct_request(self, version): 51 request = self.request_type() 52 if version: 53 request.current_version.major = version.major 54 request.current_version.minor = version.minor 55 request.current_version.patch = version.patch 56 for tag in version.prerelease_tags: 57 request.current_version.prerelease_tags.append(tag) 58 59 for tag in version.build_tags: 60 request.current_version.build_tags.append(tag) 61 62 else: 63 request.no_local_version_available = True 64 return request 65 66 def parse_response(self, response): 67 if hasattr(response, 'checkpoint'): 68 url = response.checkpoint.url 69 else: 70 url = None 71 if hasattr(response.checkpoint, 'version'): 72 version = SemanticVersion.from_proto(response.checkpoint.version) 73 else: 74 version = SemanticVersion(0, 0, 0, [], []) 75 if hasattr(response.checkpoint, 'crc'): 76 checksum = response.checkpoint.crc.crc 77 else: 78 checksum = None 79 return (url, version, checksum) 80 81 def retrieve_data_from_cdn(self, url, crc, timeout_seconds = 10, max_attempts = 2, retry_delay_in_seconds = 1): 82 host, request = separate_host_and_request(url) 83 84 def send_request(): 85 data = self._send_http_request(host, request, timeout_seconds) 86 return data 87 88 def should_retry(result_or_exception): 89 if isinstance(result_or_exception, Exception): 90 return True 91 data = result_or_exception 92 if crc is not None and crc > 0: 93 computed_crc = zlib.crc32(data) 94 if computed_crc != crc: 95 logger.warning('CRC mismatch: expected %s, got %s', crc, computed_crc) 96 return True 97 return False 98 99 return retry_with_exponential_backoff(func=send_request, max_attempts=max_attempts, retry_delay_in_seconds=retry_delay_in_seconds, jitter=True, should_retry=should_retry) 100 101 def _send_http_request(self, host, request, timeout_seconds = 10): 102 c = HTTPSConnection(host, port=443) 103 c.putrequest('GET', request) 104 c.endheaders() 105 response = c.getresponse() 106 data = response.read() 107 c.close() 108 return data 109 110 def _send_request(self, request, response_class, info_log, timeout_seconds = 10, max_attempts = 2): 111 try: 112 response_primitive, response = self._public_gateway.send_blocking_user_request_and_receive_response(request_payload=request, expected_response_class=response_class, timeout_seconds=timeout_seconds, max_attempts=max_attempts, retry_delay_in_seconds=1) 113 if response_primitive.status_code != OK: 114 self._raise_generic_error(info_log, response_primitive.status_code) 115 except (TimeoutException, UnpackException) as exc: 116 error_info = 'ERROR: %s' % exc.__class__.__name__ 117 info_log += error_info 118 logger.info(info_log) 119 request_cls = request.__class__.__name__ 120 raise ServiceUnavailableError('%s, Failed to receive response from %s: %s' % (error_info, request_cls, exc)) 121 122 return (response_primitive.status_code, response) 123 124 @staticmethod 125 def _raise_generic_error(error_text, status_code): 126 error_text += 'Request for flatbuffer data raises %s(%s)\n' % (httplib_responses[status_code], status_code) 127 logger.info(error_text) 128 if status_code == BAD_REQUEST: 129 raise GenericBadRequestError(error_text) 130 if status_code == UNAUTHORIZED: 131 raise GenericUnauthorizedError(error_text) 132 if status_code == FORBIDDEN: 133 raise GenericForbiddenError(error_text) 134 if status_code == NOT_FOUND: 135 raise GenericNotFoundError(error_text) 136 if status_code == INTERNAL_SERVER_ERROR: 137 raise GenericInternalServerError(error_text) 138 raise GenericError(error_text) 139 140 @staticmethod 141 def _raise_custom_error(info_log, error): 142 info_log += '\nRaising %s' % error.__str__() 143 logger.info(info_log) 144 raise error