feat: Add async event processor

ldclient/impl/aio/concurrency.py

@@ -2,7 +2,7 @@
 Async concurrency primitives used by the async data source, event processor, and
 data system. Each wraps a piece of fiddly asyncio plumbing (timeout-aware waits,
 queue exception normalization, an interval-from-start repeating task, a bounded
-task pool) that callers would otherwise inline repeatedly. The sync code uses the
+task set) that callers would otherwise inline repeatedly. The sync code uses the
 equivalent stdlib/SDK primitives (``threading.Event``/``Lock``, ``queue.Queue``,
 ``RepeatingTask``, ``FixedThreadPool``) directly, so these have no sync twin.
 """
@@ -12,7 +12,7 @@
 import time
 from queue import Empty as QueueEmpty  # noqa: F401  (shared timeout exception)
 from queue import Full as QueueFull  # noqa: F401  (shared capacity exception)
-from typing import Any, Callable, Optional, Set
+from typing import Any, Callable, Coroutine, Optional, Set
 
 from ldclient.impl.util import log
 
@@ -250,50 +250,38 @@ async def _run(self):
             pass
 
 
-class AsyncWorkerPool:
-    """A fixed-size pool of concurrent tasks that rejects jobs when its limit
-    is reached. Matches the contract of
-    ``ldclient.impl.fixed_thread_pool.FixedThreadPool``."""
+class BoundedTaskSet:
+    """Runs up to ``limit`` coroutines concurrently as background tasks. When the
+    limit is reached, ``try_run`` rejects new work (returning False) rather than
+    queuing it, so callers can apply their own backpressure. ``wait`` awaits the
+    tasks currently in flight; ``stop`` prevents any further work from being accepted."""
 
-    def __init__(self, size: int, name: str):
-        self._size = size
-        self._name = name
-        self._busy: Set[asyncio.Task] = set()
-        self._event = AsyncEvent()
-        self._stopped = False
+    def __init__(self, limit: int):
+        self._limit = limit
+        self._tasks: Set[asyncio.Task] = set()
+        self._accepting = True
 
-    def execute(self, jobFn: Callable) -> bool:
-        """Schedules a job for execution if the pool is not already at its
-        limit, and returns True if successful; returns False if all workers
-        are busy."""
-        if self._stopped or len(self._busy) >= self._size:
+    def try_run(self, job: Callable[[], Coroutine]) -> bool:
+        """Runs ``job()`` as a background task if fewer than ``limit`` tasks are
+        already running and the set is still accepting work. Returns True if the
+        task was started, or False if the set is full or has been stopped."""
+        if not self._accepting or len(self._tasks) >= self._limit:
             return False
-        task = asyncio.ensure_future(self._run_job(jobFn))
-        self._busy.add(task)
+        task = asyncio.create_task(job())
+        self._tasks.add(task)
+        task.add_done_callback(self._on_done)
         return True
 
-    async def _run_job(self, jobFn: Callable) -> None:
-        try:
-            result = jobFn()
-            if inspect.isawaitable(result):
-                await result
-        except Exception:
-            log.warning('Unhandled exception in worker thread', exc_info=True)
-        finally:
-            task = asyncio.current_task()
-            if task is not None:
-                self._busy.discard(task)
-            self._event.set()
+    def _on_done(self, task: asyncio.Task) -> None:
+        self._tasks.discard(task)
+        if not task.cancelled() and task.exception() is not None:
+            log.warning('Unhandled exception in background task', exc_info=task.exception())
 
     async def wait(self) -> None:
-        """Waits until all currently busy workers have completed their jobs."""
-        while len(self._busy) > 0:
-            self._event.clear()
-            if len(self._busy) == 0:
-                return
-            await self._event.wait()
+        """Waits for the tasks currently running to complete. This does not stop
+        new tasks from being added; call ``stop`` first to prevent further work."""
+        await asyncio.gather(*self._tasks, return_exceptions=True)
 
     def stop(self) -> None:
-        """Tells the pool to reject any further jobs; active jobs run to
-        completion."""
-        self._stopped = True
+        """Rejects any further work; tasks already running finish normally."""
+        self._accepting = False

ldclient/impl/events/async_event_processor.py

@@ -0,0 +1,300 @@
+"""
+Implementation details of the analytics event delivery component.
+"""
+
+import asyncio
+import gzip
+import json
+import queue
+import time
+import uuid
+from collections import namedtuple
+from random import Random
+from typing import Callable, Optional, Union
+
+from ldclient.async_config import AsyncConfig
+from ldclient.impl.aio.concurrency import (
+    AsyncEvent,
+    AsyncLock,
+    AsyncQueue,
+    AsyncRepeatingTask,
+    AsyncTaskRunner,
+    BoundedTaskSet
+)
+from ldclient.impl.aio.transport import AsyncHTTPTransport
+from ldclient.impl.events.diagnostics import create_diagnostic_init
+from ldclient.impl.events.event_processor_common import (
+    CURRENT_EVENT_SCHEMA,
+    EventBuffer,
+    EventDispatcherBase,
+    EventOutputFormatter
+)
+from ldclient.impl.events.types import EventInput
+from ldclient.impl.lru_cache import SimpleLRUCache
+from ldclient.impl.sampler import Sampler
+from ldclient.impl.util import (
+    _headers,
+    check_if_error_is_recoverable_and_log,
+    log
+)
+from ldclient.interfaces import AsyncEventProcessor
+
+__MAX_FLUSH_CONCURRENCY__ = 5
+
+
+EventProcessorMessage = namedtuple('EventProcessorMessage', ['type', 'param'])
+
+
+class EventPayloadSendTask:
+    def __init__(self, http: AsyncHTTPTransport, config: AsyncConfig, formatter: EventOutputFormatter, payload, response_fn: Callable):
+        self._http = http
+        self._config = config
+        self._formatter = formatter
+        self._payload = payload
+        self._response_fn = response_fn
+
+    async def run(self):
+        try:
+            output_events = self._formatter.make_output_events(self._payload.events, self._payload.summary)
+            await self._do_send(output_events)
+        except Exception:
+            log.warning('Unhandled exception in event processor. Analytics events were not processed.', exc_info=True)
+
+    async def _do_send(self, output_events):
+        # noinspection PyBroadException
+        try:
+            json_body = json.dumps(output_events, separators=(',', ':'))
+            log.debug('Sending events payload: ' + json_body)
+            payload_id = str(uuid.uuid4())
+            r = await _post_events_with_retry(self._http, self._config, self._config.events_uri, payload_id, json_body, "%d events" % len(output_events))
+            if r:
+                self._response_fn(r)
+            return r
+        except Exception as e:
+            log.warning('Unhandled exception in event processor. Analytics events were not processed. [%s]', e)
+
+
+class DiagnosticEventSendTask:
+    def __init__(self, http: AsyncHTTPTransport, config: AsyncConfig, event_body: dict):
+        self._http = http
+        self._config = config
+        self._event_body = event_body
+
+    async def run(self):
+        # noinspection PyBroadException
+        try:
+            json_body = json.dumps(self._event_body)
+            log.debug('Sending diagnostic event: ' + json_body)
+            await _post_events_with_retry(self._http, self._config, self._config.events_base_uri + '/diagnostic', None, json_body, "diagnostic event")
+        except Exception as e:
+            log.warning('Unhandled exception in event processor. Diagnostic event was not sent. [%s]', e)
+
+
+class EventDispatcher(EventDispatcherBase):
+    def __init__(self, inbox: AsyncQueue, config: AsyncConfig, http_client, diagnostic_accumulator=None):
+        self._inbox = inbox
+        self._config = config
+        # When no client is injected, the transport creates one targeting the
+        # events URI and owns it (closing it on shutdown); an injected client
+        # remains owned by the caller.
+        self._http = AsyncHTTPTransport(config, client=http_client)
+        self._disabled = False
+        self._outbox = EventBuffer(config.events_max_pending)
+        self._context_keys = SimpleLRUCache(config.context_keys_capacity)
+        self._formatter = EventOutputFormatter(config)
+        self._last_known_past_time = 0
+        self._deduplicated_contexts = 0
+        self._diagnostic_accumulator = None if config.diagnostic_opt_out else diagnostic_accumulator
+        self._sampler = Sampler(Random())
+        self._omit_anonymous_contexts = config.omit_anonymous_contexts
+
+        self._flush_workers = BoundedTaskSet(__MAX_FLUSH_CONCURRENCY__)
+        self._diagnostic_flush_workers: Optional[BoundedTaskSet] = None
+        if self._diagnostic_accumulator is not None:
+            self._diagnostic_flush_workers = BoundedTaskSet(1)
+            init_event = create_diagnostic_init(self._diagnostic_accumulator.data_since_date, self._diagnostic_accumulator.diagnostic_id, config)
+            task = DiagnosticEventSendTask(self._http, self._config, init_event)
+            self._diagnostic_flush_workers.try_run(task.run)
+
+        self._runner = AsyncTaskRunner()
+        self._runner.spawn("ldclient.events.processor", self._run_main_loop)
+
+    async def _run_main_loop(self):
+        log.info("Starting event processor")
+        while True:
+            try:
+                message = await self._inbox.get()
+                if message.type == 'event':
+                    self._process_event(message.param)
+                elif message.type == 'flush':
+                    self._trigger_flush()
+                elif message.type == 'flush_contexts':
+                    self._context_keys.clear()
+                elif message.type == 'diagnostic':
+                    self._send_and_reset_diagnostics()
+                elif message.type == 'flush_and_wait':
+                    # Ensure the buffered batch is actually handed to a worker
+                    # before reporting success; under saturation the first
+                    # trigger may leave the events buffered, so wait for a free
+                    # worker and retry.
+                    while not self._trigger_flush():
+                        await self._flush_workers.wait()
+                    await self._flush_workers.wait()
+                    message.param.set()
+                elif message.type == 'test_sync':
+                    await self._flush_workers.wait()
+                    if self._diagnostic_flush_workers is not None:
+                        await self._diagnostic_flush_workers.wait()
+                    message.param.set()
+                elif message.type == 'stop':
+                    await self._do_shutdown()
+                    message.param.set()
+                    return
+            except Exception:
+                log.error('Unhandled exception in event processor', exc_info=True)
+
+    def _trigger_flush(self) -> bool:
+        """Hands the buffered events to a flush worker. Returns True if handed
+        off (or there was nothing to flush), or False if all workers are busy
+        and the events were left buffered for a later attempt."""
+        if self._disabled:
+            return True
+        payload = self._outbox.get_payload()
+        if self._diagnostic_accumulator:
+            self._diagnostic_accumulator.record_events_in_batch(len(payload.events))
+        if len(payload.events) > 0 or not payload.summary.is_empty():
+            task = EventPayloadSendTask(self._http, self._config, self._formatter, payload, self._handle_response)
+            if self._flush_workers.try_run(task.run):
+                # The events have been handed off to a flush worker; clear them from our buffer.
+                self._outbox.clear()
+                return True
+            # We're already at our limit of concurrent flushes; leave the events in the buffer.
+            return False
+        return True
+
+    def _send_and_reset_diagnostics(self):
+        if self._diagnostic_accumulator is not None and self._diagnostic_flush_workers is not None:
+            dropped_event_count = self._outbox.get_and_clear_dropped_count()
+            stats_event = self._diagnostic_accumulator.create_event_and_reset(dropped_event_count, self._deduplicated_contexts)
+            self._deduplicated_contexts = 0
+            task = DiagnosticEventSendTask(self._http, self._config, stats_event)
+            self._diagnostic_flush_workers.try_run(task.run)
+
+    async def _do_shutdown(self):
+        self._flush_workers.stop()
+        await self._flush_workers.wait()
+
+        if self._diagnostic_flush_workers is not None:
+            self._diagnostic_flush_workers.stop()
+            await self._diagnostic_flush_workers.wait()
+
+        await self._http.close()
+
+
+class DefaultAsyncEventProcessor(AsyncEventProcessor):
+    def __init__(self, config: AsyncConfig, http=None, dispatcher_class=None, diagnostic_accumulator=None):
+        self._inbox = AsyncQueue(config.events_max_pending)
+        self._inbox_full = False
+        self._flush_timer = AsyncRepeatingTask("ldclient.events.flush", config.flush_interval, config.flush_interval, self.flush)
+        self._contexts_flush_timer = AsyncRepeatingTask("ldclient.events.context-flush", config.context_keys_flush_interval, config.context_keys_flush_interval, self._flush_contexts)
+        self._flush_timer.start()
+        self._contexts_flush_timer.start()
+        self._diagnostic_event_timer: Optional[AsyncRepeatingTask]
+        if diagnostic_accumulator is not None:
+            self._diagnostic_event_timer = AsyncRepeatingTask("ldclient.events.send-diagnostic", config.diagnostic_recording_interval, config.diagnostic_recording_interval, self._send_diagnostic)
+            self._diagnostic_event_timer.start()
+        else:
+            self._diagnostic_event_timer = None
+
+        self._close_lock = AsyncLock()
+        self._closed = False
+
+        (dispatcher_class or EventDispatcher)(self._inbox, config, http, diagnostic_accumulator)
+
+    def send_event(self, event: EventInput):
+        self._post_to_inbox(EventProcessorMessage('event', event))
+
+    def flush(self):
+        self._post_to_inbox(EventProcessorMessage('flush', None))
+
+    async def flush_and_wait(self, timeout: float) -> bool:
+        try:
+            await asyncio.wait_for(self._post_message_and_wait('flush_and_wait'), timeout)
+            return True
+        except asyncio.TimeoutError:
+            return False
+
+    async def stop(self):
+        async with self._close_lock:
+            if self._closed:
+                return
+            self._closed = True
+        self._flush_timer.stop()
+        self._contexts_flush_timer.stop()
+        if self._diagnostic_event_timer:
+            self._diagnostic_event_timer.stop()
+        self.flush()
+        # Note that here we are not calling _post_to_inbox, because we *do* want to wait if the inbox
+        # is full; an orderly shutdown can't happen unless these messages are received.
+        await self._post_message_and_wait('stop')
+
+    def _post_to_inbox(self, message: EventProcessorMessage):
+        try:
+            self._inbox.put_nowait(message)
+        except queue.Full:
+            if not self._inbox_full:
+                # possible race condition here, but it's of no real consequence - we'd just get an extra log line
+                self._inbox_full = True
+                log.warning("Events are being produced faster than they can be processed; some events will be dropped")
+
+    async def _flush_contexts(self):
+        await self._inbox.put(EventProcessorMessage('flush_contexts', None))
+
+    async def _send_diagnostic(self):
+        await self._inbox.put(EventProcessorMessage('diagnostic', None))
+
+    # Used only in tests
+    async def _wait_until_inactive(self):
+        await self._post_message_and_wait('test_sync')
+
+    async def _post_message_and_wait(self, type):
+        reply = AsyncEvent()
+        await self._inbox.put(EventProcessorMessage(type, reply))
+        await reply.wait()
+
+    # These magic methods allow use of the "with" block in tests
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, type, value, traceback):
+        await self.stop()
+
+
+async def _post_events_with_retry(http_client: AsyncHTTPTransport, config: AsyncConfig, uri: str, payload_id: Optional[str], body: str, events_description: str):
+    hdrs = _headers(config)
+    hdrs['Content-Type'] = 'application/json'
+    if config.enable_event_compression:
+        hdrs['Content-Encoding'] = 'gzip'
+
+    if payload_id:
+        hdrs['X-LaunchDarkly-Event-Schema'] = str(CURRENT_EVENT_SCHEMA)
+        hdrs['X-LaunchDarkly-Payload-ID'] = payload_id
+    can_retry = True
+    context = "posting %s" % events_description
+    data: Union[bytes, str] = gzip.compress(bytes(body, 'utf-8')) if config.enable_event_compression else body
+    while True:
+        next_action_message = "will retry" if can_retry else "some events were dropped"
+        try:
+            r = await http_client.request('POST', uri, headers=hdrs, body=data)
+            if r.status < 300:
+                return r
+            recoverable = check_if_error_is_recoverable_and_log(context, r.status, None, next_action_message)
+            if not recoverable:
+                return r
+        except Exception as e:
+            check_if_error_is_recoverable_and_log(context, None, str(e), next_action_message)
+        if not can_retry:
+            return None
+        can_retry = False
+        # fixed delay of 1 second for event retries
+        await asyncio.sleep(1)