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