|
1 | 1 | import logging |
2 | 2 | import time |
| 3 | +import json |
3 | 4 |
|
4 | 5 | from splitio.api import APIException |
5 | 6 | from splitio.api.commons import FetchOptions |
6 | 7 | from splitio.tasks.util import workerpool |
7 | 8 | from splitio.models import segments |
8 | 9 | from splitio.util.backoff import Backoff |
9 | | - |
| 10 | +from splitio.sync import util |
10 | 11 |
|
11 | 12 | _LOGGER = logging.getLogger(__name__) |
12 | 13 |
|
@@ -174,13 +175,153 @@ def synchronize_segments(self, segment_names = None, dont_wait = False): |
174 | 175 | """ |
175 | 176 | if segment_names is None: |
176 | 177 | segment_names = self._split_storage.get_segment_names() |
177 | | - |
| 178 | + |
178 | 179 | for segment_name in segment_names: |
179 | 180 | self._worker_pool.submit_work(segment_name) |
180 | 181 | if (dont_wait): |
181 | 182 | return True |
182 | 183 | return not self._worker_pool.wait_for_completion() |
183 | | - |
| 184 | + |
| 185 | + def segment_exist_in_storage(self, segment_name): |
| 186 | + """ |
| 187 | + Check if a segment exists in the storage |
| 188 | +
|
| 189 | + :param segment_name: Name of the segment |
| 190 | + :type segment_name: str |
| 191 | +
|
| 192 | + :return: True if segment exist. False otherwise. |
| 193 | + :rtype: bool |
| 194 | + """ |
| 195 | + return self._segment_storage.get(segment_name) != None |
| 196 | + |
| 197 | +class LocalSegmentSynchronizer(object): |
| 198 | + """Localhost mode segment synchronizer.""" |
| 199 | + |
| 200 | + _DEFAULT_SEGMENT_TILL = -1 |
| 201 | + |
| 202 | + def __init__(self, segment_folder, split_storage, segment_storage): |
| 203 | + """ |
| 204 | + Class constructor. |
| 205 | +
|
| 206 | + :param segment_folder: patch to the segment folder |
| 207 | + :type segment_folder: str |
| 208 | +
|
| 209 | + :param split_storage: Split Storage. |
| 210 | + :type split_storage: splitio.storage.InMemorySplitStorage |
| 211 | +
|
| 212 | + :param segment_storage: Segment storage reference. |
| 213 | + :type segment_storage: splitio.storage.SegmentStorage |
| 214 | +
|
| 215 | + """ |
| 216 | + self._segment_folder = segment_folder |
| 217 | + self._split_storage = split_storage |
| 218 | + self._segment_storage = segment_storage |
| 219 | + self._segment_sha = {} |
| 220 | + |
| 221 | + def synchronize_segments(self, segment_names = None): |
| 222 | + """ |
| 223 | + Loop through given segment names and synchronize each one. |
| 224 | +
|
| 225 | + :param segment_names: Optional, array of segment names to update. |
| 226 | + :type segment_name: {str} |
| 227 | +
|
| 228 | + :return: True if no error occurs. False otherwise. |
| 229 | + :rtype: bool |
| 230 | + """ |
| 231 | + _LOGGER.info('Synchronizing segments now.') |
| 232 | + if segment_names is None: |
| 233 | + segment_names = self._split_storage.get_segment_names() |
| 234 | + |
| 235 | + return_flag = True |
| 236 | + for segment_name in segment_names: |
| 237 | + if not self.synchronize_segment(segment_name): |
| 238 | + return_flag = False |
| 239 | + |
| 240 | + return return_flag |
| 241 | + |
| 242 | + def synchronize_segment(self, segment_name, till=None): |
| 243 | + """ |
| 244 | + Update a segment from queue |
| 245 | +
|
| 246 | + :param segment_name: Name of the segment to update. |
| 247 | + :type segment_name: str |
| 248 | +
|
| 249 | + :param till: ChangeNumber received. |
| 250 | + :type till: int |
| 251 | +
|
| 252 | + :return: True if no error occurs. False otherwise. |
| 253 | + :rtype: bool |
| 254 | + """ |
| 255 | + try: |
| 256 | + fetched = self._read_segment_from_json_file(segment_name) |
| 257 | + if fetched is None: |
| 258 | + return False |
| 259 | + fetched_sha = util._get_sha(json.dumps(fetched)) |
| 260 | + if not self.segment_exist_in_storage(segment_name): |
| 261 | + self._segment_sha[segment_name] = fetched_sha |
| 262 | + self._segment_storage.put(segments.from_raw(fetched)) |
| 263 | + _LOGGER.debug("segment %s is added to storage", segment_name) |
| 264 | + else: |
| 265 | + if fetched_sha != self._segment_sha[segment_name]: |
| 266 | + self._segment_sha[segment_name] = fetched_sha |
| 267 | + if self._segment_storage.get_change_number(segment_name) <= fetched['till'] or fetched['till'] == self._DEFAULT_SEGMENT_TILL: |
| 268 | + self._segment_storage.update( |
| 269 | + segment_name, |
| 270 | + fetched['added'], |
| 271 | + fetched['removed'], |
| 272 | + fetched['till'] |
| 273 | + ) |
| 274 | + _LOGGER.debug("segment %s is updated", segment_name) |
| 275 | + except Exception as e: |
| 276 | + _LOGGER.error("Could not fetch segment: %s \n" + str(e), segment_name) |
| 277 | + return False |
| 278 | + |
| 279 | + return True |
| 280 | + |
| 281 | + def _read_segment_from_json_file(self, filename): |
| 282 | + """ |
| 283 | + Parse a segment and store in segment storage. |
| 284 | +
|
| 285 | + :param filename: Path of the file containing split |
| 286 | + :type filename: str. |
| 287 | +
|
| 288 | + :return: Sanitized segment structure |
| 289 | + :rtype: Dict |
| 290 | + """ |
| 291 | + try: |
| 292 | + with open(self._segment_folder + '/' + filename + '.json', 'r') as flo: |
| 293 | + parsed = json.load(flo) |
| 294 | + santitized_segment = self._sanitize_segment(parsed) |
| 295 | + return santitized_segment |
| 296 | + except Exception as exc: |
| 297 | + raise ValueError("Error parsing file %s. Make sure it's readable." % filename) from exc |
| 298 | + |
| 299 | + def _sanitize_segment(self, parsed): |
| 300 | + """ |
| 301 | + Sanitize json elements. |
| 302 | +
|
| 303 | + :param parsed: segment dict |
| 304 | + :type parsed: Dict |
| 305 | +
|
| 306 | + :return: sanitized segment structure dict |
| 307 | + :rtype: Dict |
| 308 | + """ |
| 309 | + if 'name' not in parsed or parsed['name'] is None: |
| 310 | + _LOGGER.warning("Segment does not have [name] element, skipping") |
| 311 | + return None |
| 312 | + if parsed['name'].strip() == '': |
| 313 | + _LOGGER.warning("Segment [name] element is blank, skipping") |
| 314 | + return None |
| 315 | + |
| 316 | + for element in [('till', -1, -1, None, None, [0]), |
| 317 | + ('added', [], None, None, None, None), |
| 318 | + ('removed', [], None, None, None, None) |
| 319 | + ]: |
| 320 | + parsed = util._sanitize_object_element(parsed, 'segment', element[0], element[1], lower_value=element[2], upper_value=element[3], in_list=None, not_in_list=element[5]) |
| 321 | + parsed = util._sanitize_object_element(parsed, 'segment', 'since', parsed['till'], -1, parsed['till'], None, [0]) |
| 322 | + |
| 323 | + return parsed |
| 324 | + |
184 | 325 | def segment_exist_in_storage(self, segment_name): |
185 | 326 | """ |
186 | 327 | Check if a segment exists in the storage |
|
0 commit comments