repository.py (5527B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\packages\content\client\flatbuffer\repository.py 2 from abc import ABCMeta, abstractmethod 3 import uthread2 4 import blue 5 import const 6 import logging 7 import socket 8 from content.client.flatbuffer.errors import ServiceUnavailableError 9 from content.client.flatbuffer.provider import FlatbufferContentProvider 10 from content.client.repository import BaseContentRepository 11 from monolithsentry.client.error_spike_key_consts import FLATBUFFER_GET_UPDATE_CONNECTION_RESET, FLATBUFFER_GET_UPDATE_TIMEOUT 12 from monolithsentry.client.local_error_spike_report import record_handled_failure 13 if False: 14 from content.client.consumer import BaseContentConsumer 15 from typing import Type 16 from caching.client.storage.base_storage import BaseVersionedStorage 17 from caching.client.storage.base_storage import BaseVersionCollectionStorage 18 logger = logging.getLogger('flatbuffer_repository') 19 20 class VersionCachedFlatbufferRepository(BaseContentRepository): 21 __metaclass__ = ABCMeta 22 provider_type = None 23 disk_storage_type = None 24 in_memory_storage_type = None 25 in_memory_ttl = 60 26 27 def __init__(self, **kwargs): 28 self._provider = self.provider_type(**kwargs) 29 self._disk_storage = self.disk_storage_type(self.content_name) 30 self._in_memory_cache = self.in_memory_storage_type(self.content_name) 31 self._request_in_progress = False 32 self._last_timestamp = 0 33 super(VersionCachedFlatbufferRepository, self).__init__(**kwargs) 34 35 @property 36 def request_in_progress(self): 37 return self._request_in_progress 38 39 def refresh(self, refresh_interval): 40 if blue.os.GetWallclockTime() - self._last_timestamp < refresh_interval * const.SEC: 41 self._request_in_progress = False 42 return 43 self._start_request_update_task(self._in_memory_cache.get_version(), None) 44 45 def _start_request_update_task(self, local_version, latest_version = None): 46 if self._request_in_progress: 47 logger.info('update request is already in progress, skipping') 48 return 49 uthread2.start_tasklet(self._request_update_task, local_version, latest_version) 50 51 def _prime_in_memory_cache(self): 52 if self._request_in_progress: 53 return 54 version_in_memory = self._in_memory_cache.get_version() 55 if version_in_memory and blue.os.GetWallclockTime() - self._last_timestamp < self.in_memory_ttl * const.SEC: 56 return 57 if version_in_memory: 58 local_version = version_in_memory 59 else: 60 local_version = self._disk_storage.get_version() 61 cached_data = self._disk_storage.get() 62 local_data = self._construct_data_from_binary(cached_data) if cached_data is not None else None 63 if local_version is not None and local_data is not None: 64 self._in_memory_cache.set_version(local_version) 65 else: 66 local_version = None 67 local_data = self._get_default_data() 68 self._in_memory_cache.set_all(local_data) 69 self._start_request_update_task(local_version, None) 70 71 def _request_update_task(self, current_version, latest_version = None): 72 info_message = 'updating flatbuffer repository, current version is %s...' % current_version 73 if latest_version is not None: 74 info_message += ', latest version is %s...' % latest_version 75 self._request_in_progress = True 76 blue.pyos.BeNice() 77 try: 78 version, data = self._provider.fetch(current_version, latest_version) 79 if version and data: 80 info_message += 'new version found: %s! updating disk and in-memory storage.' % version 81 self._last_timestamp = blue.os.GetWallclockTime() 82 self._in_memory_cache.set_version(version) 83 self._in_memory_cache.set_all(self._construct_data_from_binary(data)) 84 self._disk_storage.set(data) 85 self._disk_storage.set_version(version) 86 self.on_content_updated(self.content_name) 87 elif self._in_memory_cache.get_version() is None: 88 storage_version = self._disk_storage.get_version() 89 info_message += 'no new version found, but in-memory cache is empty, priming with storage version %s.' % storage_version 90 self._in_memory_cache.set_version(storage_version) 91 self._in_memory_cache.set_all(self._construct_data_from_binary(self._disk_storage.get())) 92 self.on_content_updated(self.content_name) 93 else: 94 self.on_no_content_updated(self.content_name) 95 except ServiceUnavailableError as exc: 96 record_handled_failure(FLATBUFFER_GET_UPDATE_TIMEOUT, '%s: flatbuffer update unavailable: %s' % (self.content_name, exc)) 97 except Exception as exc: 98 if socket is not None and isinstance(exc, socket.error) and getattr(exc, 'errno', None) == 10054: 99 record_handled_failure(FLATBUFFER_GET_UPDATE_CONNECTION_RESET, '%s: flatbuffer update connection reset by remote host' % self.content_name) 100 else: 101 raise 102 finally: 103 logger.info('%s: %s', self.content_name, info_message) 104 self._request_in_progress = False 105 106 @abstractmethod 107 def _construct_data_from_binary(self, binary_data): 108 pass 109 110 @abstractmethod 111 def _get_default_data(self): 112 pass