feat: Add shared file data code with reload, retry, and polling

Adds ldclient.impl.integrations.files.filedata, the file reading, parsing,
and merging logic shared by the components that load flag and segment data
from local files. A document that starts with an opening brace is parsed as
JSON and any other document as YAML, so JSON files no longer depend on the
YAML parser. Definitions are decoded into the flag and segment models while
the file is read, so an invalid definition fails that load instead of failing
later inside the store. Files are merged in the configured order with a
duplicate keys handling of fail or ignore.

The Reloader owns the reload cycle: it serializes reloads, debounces change
signals with a settle window, keeps the last good data by not applying a
failed load, retries a failed load after a bounded delay, reports an identical
failure once, and skips an application whose file contents did not change. It
can treat a configured file that does not exist as a file with no content.

The Poller detects changes by comparing modification time and size on an
interval, including files that appear or disappear. The Watcher uses the
watchdog package, watches the directory of each file so an absent file is
picked up when it appears, matches the destination of a move so a file written
by rename is detected, and retries a directory that does not exist yet.

Both file data sources are rebuilt on this module. Their files must still
exist and duplicate keys still fail the load. The FDv2 synchronizer now keeps
the last good data across a failed load, reports an interrupted state, and
retries, where before a failed initial load ended the synchronizer. The
expansion of a flagValues entry is now an off flag serving its single
variation, which is the form the Go SDK uses, so the evaluation reason kind
for such a flag is OFF.

Author: kinyoklion

ldclient/impl/integrations/files/file_data_source.py

@@ -1,10 +1,18 @@
-import json
-import os
 import time
 import traceback
-from typing import Optional
-
-from ldclient.impl.repeating_task import RepeatingTask
+from typing import Optional, Union
+
+from ldclient.impl.integrations.files.filedata import (
+    DEFAULT_DEBOUNCE_DELAY,
+    DEFAULT_RETRY_DELAY,
+    DuplicateKeysHandling,
+    MergeResult,
+    Poller,
+    Reloader,
+    Watcher,
+    abs_file_paths,
+    have_watchdog
+)
 from ldclient.impl.util import log
 from ldclient.interfaces import (
     DataSourceErrorInfo,
@@ -15,43 +23,29 @@
 )
 from ldclient.versioned_data_kind import FEATURES, SEGMENTS
 
-have_yaml = False
-try:
-    import yaml
-
-    have_yaml = True
-except ImportError:
-    pass
-
-have_watchdog = False
-try:
-    import watchdog
-    import watchdog.events
-    import watchdog.observers
-
-    have_watchdog = True
-except ImportError:
-    pass
-
-
-def _sanitize_json_item(item):
-    if not ('version' in item):
-        item['version'] = 1
-
 
 class _FileDataSource(UpdateProcessor):
     def __init__(self, store, data_source_update_sink: Optional[DataSourceUpdateSink], ready, paths, auto_update, poll_interval, force_polling):
         self._store = store
         self._data_source_update_sink = data_source_update_sink
         self._ready = ready
         self._inited = False
-        self._paths = paths
-        if isinstance(self._paths, str):
-            self._paths = [self._paths]
+        self._paths = abs_file_paths(paths if isinstance(paths, list) else [paths])
         self._auto_update = auto_update
-        self._auto_updater = None
+        self._auto_updater: Optional[Union[Poller, Watcher]] = None
         self._poll_interval = poll_interval
         self._force_polling = force_polling
+        # Debouncing and automatic retries only matter when something can trigger further
+        # reloads. A source without auto update loads exactly once.
+        self._reloader = Reloader(
+            self._paths,
+            DuplicateKeysHandling.FAIL,
+            apply=self._apply,
+            on_error=self._handle_error,
+            debounce_delay=DEFAULT_DEBOUNCE_DELAY if auto_update else 0.0,
+            retry_delay=DEFAULT_RETRY_DELAY if auto_update else 0.0,
+            skip_unchanged=True,
+        )
 
     def _sink_or_store(self):
         """
@@ -71,7 +65,7 @@ def _sink_or_store(self):
         return self._data_source_update_sink
 
     def start(self):
-        self._load_all()
+        self._reloader.reload_now()
 
         if self._auto_update:
             self._auto_updater = self._start_auto_updater()
@@ -81,23 +75,19 @@ def start(self):
         self._ready.set()
 
     def stop(self):
-        if self._auto_updater:
-            self._auto_updater.stop()
+        if self._auto_updater is not None:
+            self._auto_updater.close()
+            self._auto_updater = None
+        self._reloader.close()
 
     def initialized(self):
         return self._inited
 
-    def _load_all(self):
-        all_data = {FEATURES: {}, SEGMENTS: {}}
-        for path in self._paths:
-            try:
-                self._load_file(path, all_data)
-            except Exception as e:
-                log.error('Unable to load flag data from "%s": %s' % (path, repr(e)))
-                traceback.print_exc()
-                if self._data_source_update_sink is not None:
-                    self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, DataSourceErrorInfo(DataSourceErrorKind.INVALID_DATA, 0, time.time, str(e)))
-                return
+    def _apply(self, merged: MergeResult):
+        all_data = {
+            FEATURES: {key: flag.to_json_dict() for key, flag in merged.flags.items()},
+            SEGMENTS: {key: segment.to_json_dict() for key, segment in merged.segments.items()},
+        }
         try:
             self._sink_or_store().init(all_data)
             self._inited = True
@@ -107,104 +97,15 @@ def _load_all(self):
             log.error('Unable to store data: %s' % repr(e))
             traceback.print_exc()
             if self._data_source_update_sink is not None:
-                self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, DataSourceErrorInfo(DataSourceErrorKind.UNKNOWN, 0, time.time, str(e)))
-
-    def _load_file(self, path, all_data):
-        content = None
-        with open(path, 'r') as f:
-            content = f.read()
-        parsed = self._parse_content(content)
-        for key, flag in parsed.get('flags', {}).items():
-            _sanitize_json_item(flag)
-            self._add_item(all_data, FEATURES, flag)
-        for key, value in parsed.get('flagValues', {}).items():
-            self._add_item(all_data, FEATURES, self._make_flag_with_value(key, value))
-        for key, segment in parsed.get('segments', {}).items():
-            _sanitize_json_item(segment)
-            self._add_item(all_data, SEGMENTS, segment)
-
-    def _parse_content(self, content):
-        if have_yaml:
-            return yaml.safe_load(content)  # pyyaml correctly parses JSON too
-        return json.loads(content)
+                self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, DataSourceErrorInfo(DataSourceErrorKind.UNKNOWN, 0, time.time(), str(e)))
 
-    def _add_item(self, all_data, kind, item):
-        items = all_data[kind]
-        key = item.get('key')
-        if items.get(key) is None:
-            items[key] = item
-        else:
-            raise Exception('In %s, key "%s" was used more than once' % (kind.namespace, key))
+    def _handle_error(self, error: Exception):
+        if self._data_source_update_sink is not None:
+            self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, DataSourceErrorInfo(DataSourceErrorKind.INVALID_DATA, 0, time.time(), str(error)))
 
-    def _make_flag_with_value(self, key, value):
-        return {'key': key, 'version': 1, 'on': True, 'fallthrough': {'variation': 0}, 'variations': [value]}
-
-    def _start_auto_updater(self):
-        resolved_paths = []
-        for path in self._paths:
-            try:
-                resolved_paths.append(os.path.realpath(path))
-            except Exception:
-                log.warning('Cannot watch for changes to data file "%s" because it is an invalid path' % path)
+    def _start_auto_updater(self) -> Union[Poller, Watcher]:
         if have_watchdog and not self._force_polling:
-            return _FileDataSource.WatchdogAutoUpdater(resolved_paths, self._load_all)
-        else:
-            return _FileDataSource.PollingAutoUpdater(resolved_paths, self._load_all, self._poll_interval)
-
-    # Watch for changes to data files using the watchdog package. This uses native OS filesystem notifications
-    # if available for the current platform.
-    class WatchdogAutoUpdater:
-        def __init__(self, resolved_paths, reloader):
-            watched_files = set(resolved_paths)
-
-            class LDWatchdogHandler(watchdog.events.FileSystemEventHandler):
-                def on_any_event(self, event):
-                    if event.src_path in watched_files:
-                        reloader()
-
-            dir_paths = set()
-            for path in resolved_paths:
-                dir_paths.add(os.path.dirname(path))
-
-            self._observer = watchdog.observers.Observer()
-            handler = LDWatchdogHandler()
-            for path in dir_paths:
-                self._observer.schedule(handler, path)
-            self._observer.start()
-
-        def stop(self):
-            self._observer.stop()
-            self._observer.join()
-
-    # Watch for changes to data files by polling their modification times. This is used if auto-update is
-    # on but the watchdog package is not installed.
-    class PollingAutoUpdater:
-        def __init__(self, resolved_paths, reloader, interval):
-            self._paths = resolved_paths
-            self._reloader = reloader
-            self._file_times = self._check_file_times()
-            self._timer = RepeatingTask.at_interval("ldclient.datasource.file.poll", interval, interval, self._poll)
-            self._timer.start()
-
-        def stop(self):
-            self._timer.stop()
-
-        def _poll(self):
-            new_times = self._check_file_times()
-            changed = False
-            for file_path, file_time in self._file_times.items():
-                if new_times.get(file_path) is not None and new_times.get(file_path) != file_time:
-                    changed = True
-                    break
-            self._file_times = new_times
-            if changed:
-                self._reloader()
-
-        def _check_file_times(self):
-            ret = {}
-            for path in self._paths:
-                try:
-                    ret[path] = os.path.getmtime(path)
-                except Exception:
-                    ret[path] = None
-            return ret
+            return Watcher(self._paths, self._reloader.trigger)
+        poller = Poller(self._paths, self._poll_interval, self._reloader.trigger)
+        poller.start()
+        return poller