diff --git a/ldclient/impl/events/async_event_processor.py b/ldclient/impl/events/async_event_processor.py new file mode 100644 index 00000000..e7d178b8 --- /dev/null +++ b/ldclient/impl/events/async_event_processor.py @@ -0,0 +1,291 @@ +""" +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, + AsyncWorkerPool +) +from ldclient.impl.aio.transport import AsyncHTTPTransport +from ldclient.impl.events.diagnostics import create_diagnostic_init +from ldclient.impl.events.event_processor_common import ( + 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_THREADS__ = 5 +__CURRENT_EVENT_SCHEMA__ = 4 + + +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 = AsyncWorkerPool(__MAX_FLUSH_THREADS__, "ldclient.flush") + self._diagnostic_flush_workers: Optional[AsyncWorkerPool] = None + if self._diagnostic_accumulator is not None: + self._diagnostic_flush_workers = AsyncWorkerPool(1, "ldclient.events.diag_flush") + 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.execute(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': + self._trigger_flush() + 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): + if self._disabled: + return + 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.execute(task.run): + # The events have been handed off to a flush worker; clear them from our buffer. + self._outbox.clear() + else: + # We're already at our limit of concurrent flushes; leave the events in the buffer. + pass + + 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.execute(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) diff --git a/ldclient/impl/events/event_processor.py b/ldclient/impl/events/event_processor.py index 97792fc7..f9ee3ddb 100644 --- a/ldclient/impl/events/event_processor.py +++ b/ldclient/impl/events/event_processor.py @@ -7,30 +7,20 @@ import queue import time import uuid -from calendar import timegm from collections import namedtuple -from email.utils import parsedate from random import Random from threading import Event, Lock, Thread -from typing import Any, Callable, Optional import urllib3 from ldclient.config import Config -from ldclient.context import Context from ldclient.impl.events.diagnostics import create_diagnostic_init from ldclient.impl.events.event_processor_common import ( - DebugEvent, EventBuffer, - EventOutputFormatter, - IndexEvent -) -from ldclient.impl.events.types import ( - EventInput, - EventInputCustom, - EventInputEvaluation, - EventInputIdentify + EventDispatcherBase, + EventOutputFormatter ) +from ldclient.impl.events.types import EventInput from ldclient.impl.fixed_thread_pool import FixedThreadPool from ldclient.impl.http import _http_factory from ldclient.impl.lru_cache import SimpleLRUCache @@ -39,12 +29,9 @@ from ldclient.impl.util import ( _headers, check_if_error_is_recoverable_and_log, - current_time_millis, - is_http_error_recoverable, log ) from ldclient.interfaces import EventProcessor -from ldclient.migrations.tracker import MigrationOpEvent __MAX_FLUSH_THREADS__ = 5 __CURRENT_EVENT_SCHEMA__ = 4 @@ -98,7 +85,7 @@ def run(self): log.warning('Unhandled exception in event processor. Diagnostic event was not sent. [%s]', e) -class EventDispatcher: +class EventDispatcher(EventDispatcherBase): def __init__(self, inbox, config, http_client, diagnostic_accumulator=None): self._inbox = inbox self._config = config @@ -150,78 +137,6 @@ def _run_main_loop(self): except Exception as e: log.error('Unhandled exception in event processor', exc_info=True) - def _process_event(self, event: EventInput): - if self._disabled: - return - - # Decide whether to add the event to the payload. Feature events may be added twice, once for - # the event (if tracked) and once for debugging. - context = None # type: Optional[Context] - full_event = None # type: Any - debug_event = None # type: Optional[DebugEvent] - sampling_ratio = 1 if event.sampling_ratio is None else event.sampling_ratio - - if isinstance(event, EventInputEvaluation): - context = event.context - if not event.exclude_from_summaries: - self._outbox.add_to_summary(event) - if event.track_events: - full_event = event - if self._should_debug_event(event): - debug_event = DebugEvent(event) - elif isinstance(event, EventInputIdentify): - if self._omit_anonymous_contexts: - context = event.context.without_anonymous_contexts() - if not context.valid: - return - - event = EventInputIdentify(event.timestamp, context, event.sampling_ratio) - - full_event = event - elif isinstance(event, EventInputCustom): - context = event.context - full_event = event - elif isinstance(event, MigrationOpEvent): - full_event = event - - self._get_indexable_context(event, lambda c: self._outbox.add_event(IndexEvent(event.timestamp, c))) - - if full_event and self._sampler.sample(sampling_ratio): - self._outbox.add_event(full_event) - - if debug_event and self._sampler.sample(sampling_ratio): - self._outbox.add_event(debug_event) - - def _get_indexable_context(self, event: EventInput, block: Callable[[Context], None]): - if event.context is None: - return - - context = event.context - if self._omit_anonymous_contexts: - context = context.without_anonymous_contexts() - - if not context.valid: - return - - already_seen = self._context_keys.put(context.fully_qualified_key, True) - if already_seen: - self._deduplicated_contexts += 1 - return - elif isinstance(event, EventInputIdentify) or isinstance(event, MigrationOpEvent): - return - - block(context) - - def _should_debug_event(self, event: EventInputEvaluation): - if event.flag is None: - return False - debug_until = event.flag.debug_events_until_date - if debug_until is not None: - last_past = self._last_known_past_time - if debug_until > last_past and debug_until > current_time_millis(): - return True - return False - def _trigger_flush(self): if self._disabled: return @@ -237,17 +152,6 @@ def _trigger_flush(self): # We're already at our limit of concurrent flushes; leave the events in the buffer. pass - def _handle_response(self, r): - server_date_str = r.headers.get('Date') - if server_date_str is not None: - server_date = parsedate(server_date_str) - if server_date is not None: - timestamp = int(timegm(server_date) * 1000) - self._last_known_past_time = timestamp - if r.status > 299 and not is_http_error_recoverable(r.status): - self._disabled = True - return - def _send_and_reset_diagnostics(self): if self._diagnostic_accumulator is not None: dropped_event_count = self._outbox.get_and_clear_dropped_count() diff --git a/ldclient/impl/events/event_processor_common.py b/ldclient/impl/events/event_processor_common.py index 1b558aed..b8b60827 100644 --- a/ldclient/impl/events/event_processor_common.py +++ b/ldclient/impl/events/event_processor_common.py @@ -5,19 +5,29 @@ and async (async_event_processor.py) implementations. """ +from calendar import timegm from collections import namedtuple -from typing import Any, Dict, List +from email.utils import parsedate +from typing import Any, Callable, Dict, List, Optional from ldclient.config import PrivateAttributesConfig from ldclient.context import Context from ldclient.impl.events.event_context_formatter import EventContextFormatter from ldclient.impl.events.event_summarizer import EventSummarizer, EventSummary from ldclient.impl.events.types import ( + EventInput, EventInputCustom, EventInputEvaluation, EventInputIdentify ) -from ldclient.impl.util import log, timedelta_millis +from ldclient.impl.lru_cache import SimpleLRUCache +from ldclient.impl.sampler import Sampler +from ldclient.impl.util import ( + current_time_millis, + is_http_error_recoverable, + log, + timedelta_millis +) from ldclient.migrations.tracker import MigrationOpEvent # --------------------------------------------------------------------------- @@ -224,3 +234,110 @@ def _base_eval_props(self, e: EventInputEvaluation, kind: str) -> dict: if e.prereq_of is not None: out['prereqOf'] = e.prereq_of.key return out + + +# --------------------------------------------------------------------------- +# EventDispatcherBase — pure event-handling logic shared by both dispatchers +# --------------------------------------------------------------------------- + +class EventDispatcherBase: + """ + Pure event-handling methods shared by the sync and async EventDispatcher + implementations. These methods perform no I/O. + + This class does not define ``__init__``. Subclasses are responsible for + setting the following attributes, which these methods rely on: + ``_disabled``, ``_outbox`` (an :class:`EventBuffer`), ``_sampler``, + ``_context_keys``, ``_last_known_past_time``, ``_omit_anonymous_contexts``, + and ``_deduplicated_contexts``. + """ + + _disabled: bool + _outbox: EventBuffer + _sampler: Sampler + _context_keys: SimpleLRUCache + _last_known_past_time: int + _omit_anonymous_contexts: bool + _deduplicated_contexts: int + + def _process_event(self, event: EventInput): + if self._disabled: + return + + # Decide whether to add the event to the payload. Feature events may be added twice, once for + # the event (if tracked) and once for debugging. + context: Optional[Context] = None + full_event: Any = None + debug_event: Optional[DebugEvent] = None + sampling_ratio = 1 if event.sampling_ratio is None else event.sampling_ratio + + if isinstance(event, EventInputEvaluation): + context = event.context + if not event.exclude_from_summaries: + self._outbox.add_to_summary(event) + if event.track_events: + full_event = event + if self._should_debug_event(event): + debug_event = DebugEvent(event) + elif isinstance(event, EventInputIdentify): + if self._omit_anonymous_contexts: + context = event.context.without_anonymous_contexts() + if not context.valid: + return + + event = EventInputIdentify(event.timestamp, context, event.sampling_ratio) + + full_event = event + elif isinstance(event, EventInputCustom): + context = event.context + full_event = event + elif isinstance(event, MigrationOpEvent): + full_event = event + + self._get_indexable_context(event, lambda c: self._outbox.add_event(IndexEvent(event.timestamp, c))) + + if full_event and self._sampler.sample(sampling_ratio): + self._outbox.add_event(full_event) + + if debug_event and self._sampler.sample(sampling_ratio): + self._outbox.add_event(debug_event) + + def _get_indexable_context(self, event: EventInput, block: Callable[[Context], None]): + if event.context is None: + return + + context = event.context + if self._omit_anonymous_contexts: + context = context.without_anonymous_contexts() + + if not context.valid: + return + + already_seen = self._context_keys.put(context.fully_qualified_key, True) + if already_seen: + self._deduplicated_contexts += 1 + return + elif isinstance(event, EventInputIdentify) or isinstance(event, MigrationOpEvent): + return + + block(context) + + def _should_debug_event(self, event: EventInputEvaluation): + if event.flag is None: + return False + debug_until = event.flag.debug_events_until_date + if debug_until is not None: + last_past = self._last_known_past_time + if debug_until > last_past and debug_until > current_time_millis(): + return True + return False + + def _handle_response(self, r): + server_date_str = r.headers.get('Date') + if server_date_str is not None: + server_date = parsedate(server_date_str) + if server_date is not None: + timestamp = int(timegm(server_date) * 1000) + self._last_known_past_time = timestamp + if r.status > 299 and not is_http_error_recoverable(r.status): + self._disabled = True diff --git a/ldclient/testing/impl/events/test_async_event_processor.py b/ldclient/testing/impl/events/test_async_event_processor.py new file mode 100644 index 00000000..780513c7 --- /dev/null +++ b/ldclient/testing/impl/events/test_async_event_processor.py @@ -0,0 +1,461 @@ +""" +Tests for DefaultAsyncEventProcessor. + +These mirror the sync DefaultEventProcessor tests: the processor is driven +through its public API plus the ``_wait_until_inactive`` test handshake, and +HTTP delivery is captured by a mock aiohttp-style session injected through +the ``http`` constructor parameter. +""" + +import asyncio +import gzip +import json +import uuid +from contextlib import asynccontextmanager +from email.utils import formatdate +from typing import Any, List, Optional, Tuple + +import pytest + +from ldclient.async_config import AsyncConfig +from ldclient.context import Context +from ldclient.impl.events.async_event_processor import ( + DefaultAsyncEventProcessor +) +from ldclient.impl.events.diagnostics import ( + _DiagnosticAccumulator, + create_diagnostic_id +) +from ldclient.impl.events.types import ( + EventInputCustom, + EventInputEvaluation, + EventInputIdentify +) +from ldclient.testing.builders import FlagBuilder +from ldclient.testing.stub_util import _MockHTTPHeaderDict + +pytestmark = pytest.mark.asyncio + +context = Context.builder('userkey').name('Red').build() +flag = FlagBuilder('flagkey').version(2).build() +timestamp = 10000 + + +# --------------------------------------------------------------------------- +# Mock aiohttp session (mirrors stub_util.MockHttp for the aiohttp surface +# used by AsyncHTTPTransport) +# --------------------------------------------------------------------------- + +class MockAioResponse: + def __init__(self, status: int, headers: dict): + self.status = status + self.headers = _MockHTTPHeaderDict(headers) + + async def text(self, encoding: str = 'UTF-8', errors: str = 'strict') -> str: + return '' + + +class _MockRequestContext: + def __init__(self, response: MockAioResponse): + self._response = response + + async def __aenter__(self) -> MockAioResponse: + return self._response + + async def __aexit__(self, exc_type, exc_value, traceback) -> bool: + return False + + +class MockAioHttp: + def __init__(self): + self._recorded_requests: List[Tuple[Any, Any]] = [] + self._response_status = 200 + self._server_time: Optional[int] = None + + def request(self, method, uri, headers=None, data=None, timeout=None, proxy=None): + self._recorded_requests.append((headers, data)) + resp_hdr = dict() + if self._server_time is not None: + resp_hdr['date'] = formatdate(self._server_time / 1000, localtime=False, usegmt=True) + return _MockRequestContext(MockAioResponse(self._response_status, resp_hdr)) + + @property + def request_data(self): + if len(self._recorded_requests) != 0: + return self._recorded_requests[-1][1] + + @property + def request_headers(self): + if len(self._recorded_requests) != 0: + return self._recorded_requests[-1][0] + + @property + def recorded_requests(self): + return self._recorded_requests + + def set_response_status(self, status: int): + self._response_status = status + + def set_server_time(self, timestamp: int): + self._server_time = timestamp + + def reset(self): + self._recorded_requests = [] + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +@asynccontextmanager +async def make_processor(mock_http: MockAioHttp, **kwargs): + kwargs.setdefault('diagnostic_opt_out', True) + kwargs.setdefault('sdk_key', 'SDK_KEY') + config = AsyncConfig(**kwargs) + diagnostic_accumulator = _DiagnosticAccumulator(create_diagnostic_id(config)) + ep = DefaultAsyncEventProcessor(config, mock_http, diagnostic_accumulator=diagnostic_accumulator) + try: + yield ep + finally: + await ep.stop() + + +async def flush_and_get_events(ep: DefaultAsyncEventProcessor, mock_http: MockAioHttp): + ep.flush() + await ep._wait_until_inactive() + if mock_http.request_data is None: + raise AssertionError('Expected to get an HTTP request but did not get one') + return json.loads(mock_http.request_data) + + +# --------------------------------------------------------------------------- +# Event payload tests +# --------------------------------------------------------------------------- + +async def test_identify_event_is_queued(): + mock_http = MockAioHttp() + async with make_processor(mock_http) as ep: + e = EventInputIdentify(timestamp, context) + ep.send_event(e) + + output = await flush_and_get_events(ep, mock_http) + assert len(output) == 1 + assert output[0] == {'kind': 'identify', 'creationDate': timestamp, 'context': context.to_dict()} + + +async def test_individual_feature_event_is_queued_with_index_event(): + mock_http = MockAioHttp() + async with make_processor(mock_http) as ep: + e = EventInputEvaluation(timestamp, context, flag.key, flag, 1, 'value', None, 'default', None, True) + ep.send_event(e) + + output = await flush_and_get_events(ep, mock_http) + assert len(output) == 3 + assert output[0]['kind'] == 'index' + assert output[0]['context'] == context.to_dict() + assert output[1]['kind'] == 'feature' + assert output[1]['key'] == flag.key + assert output[1]['value'] == 'value' + assert output[2]['kind'] == 'summary' + + +async def test_custom_event_is_queued(): + mock_http = MockAioHttp() + async with make_processor(mock_http) as ep: + e = EventInputCustom(timestamp, context, 'eventkey', {'thing': 'stuff'}, 1.5) + ep.send_event(e) + + output = await flush_and_get_events(ep, mock_http) + assert len(output) == 2 + assert output[0]['kind'] == 'index' + assert output[1]['kind'] == 'custom' + assert output[1]['key'] == 'eventkey' + assert output[1]['data'] == {'thing': 'stuff'} + assert output[1]['metricValue'] == 1.5 + + +# --------------------------------------------------------------------------- +# flush_and_wait +# --------------------------------------------------------------------------- + +async def test_flush_and_wait_delivers_events_and_returns_true(): + mock_http = MockAioHttp() + async with make_processor(mock_http) as ep: + ep.send_event(EventInputIdentify(timestamp, context)) + + delivered = await ep.flush_and_wait(5) + + assert delivered is True + assert mock_http.request_data is not None + output = json.loads(mock_http.request_data) + assert len(output) == 1 + assert output[0]['kind'] == 'identify' + + +async def test_flush_and_wait_returns_false_on_timeout(): + mock_http = MockAioHttp() + async with make_processor(mock_http) as ep: + ep.send_event(EventInputIdentify(timestamp, context)) + + # A zero timeout can't complete the flush round-trip, so it reports False. + delivered = await ep.flush_and_wait(0) + + assert delivered is False + + +async def test_two_events_for_same_context_only_produce_one_index_event(): + mock_http = MockAioHttp() + async with make_processor(mock_http) as ep: + e0 = EventInputEvaluation(timestamp, context, flag.key, flag, 1, 'value1', None, 'default', None, True) + e1 = EventInputEvaluation(timestamp, context, flag.key, flag, 2, 'value2', None, 'default', None, True) + ep.send_event(e0) + ep.send_event(e1) + + output = await flush_and_get_events(ep, mock_http) + assert len(output) == 4 + assert output[0]['kind'] == 'index' + assert output[1]['kind'] == 'feature' + assert output[2]['kind'] == 'feature' + assert output[3]['kind'] == 'summary' + + +async def test_nontracked_events_are_summarized(): + mock_http = MockAioHttp() + async with make_processor(mock_http) as ep: + e = EventInputEvaluation(timestamp, context, flag.key, flag, 1, 'value', None, 'default', None, False) + ep.send_event(e) + + output = await flush_and_get_events(ep, mock_http) + assert len(output) == 2 + assert output[0]['kind'] == 'index' + assert output[1]['kind'] == 'summary' + assert flag.key in output[1]['features'] + + +async def test_nothing_is_sent_if_there_are_no_events(): + mock_http = MockAioHttp() + async with make_processor(mock_http) as ep: + ep.flush() + await ep._wait_until_inactive() + assert mock_http.request_data is None + + +async def test_stop_flushes_remaining_events(): + mock_http = MockAioHttp() + async with make_processor(mock_http) as ep: + ep.send_event(EventInputIdentify(timestamp, context)) + # stop() (via the context manager) triggers a final flush + assert mock_http.request_data is not None + output = json.loads(mock_http.request_data) + assert output[0]['kind'] == 'identify' + + +# --------------------------------------------------------------------------- +# Header tests +# --------------------------------------------------------------------------- + +async def test_sdk_key_is_sent(): + mock_http = MockAioHttp() + async with make_processor(mock_http, sdk_key='SDK_KEY') as ep: + ep.send_event(EventInputIdentify(timestamp, context)) + await flush_and_get_events(ep, mock_http) + + assert mock_http.request_headers.get('Authorization') == 'SDK_KEY' + + +async def test_event_schema_set_on_event_send(): + mock_http = MockAioHttp() + async with make_processor(mock_http) as ep: + ep.send_event(EventInputIdentify(timestamp, context)) + await flush_and_get_events(ep, mock_http) + + assert mock_http.request_headers.get('X-LaunchDarkly-Event-Schema') == "4" + + +async def test_event_payload_id_is_sent(): + mock_http = MockAioHttp() + async with make_processor(mock_http) as ep: + ep.send_event(EventInputIdentify(timestamp, context)) + await flush_and_get_events(ep, mock_http) + + header_val = mock_http.request_headers.get('X-LaunchDarkly-Payload-ID') + assert header_val is not None + # Throws on invalid UUID + uuid.UUID(header_val) + + +async def test_event_payload_id_changes_between_requests(): + mock_http = MockAioHttp() + async with make_processor(mock_http) as ep: + ep.send_event(EventInputIdentify(timestamp, context)) + await flush_and_get_events(ep, mock_http) + + ep.send_event(EventInputIdentify(timestamp, context)) + ep.flush() + await ep._wait_until_inactive() + + first_payload_id = mock_http.recorded_requests[0][0].get('X-LaunchDarkly-Payload-ID') + second_payload_id = mock_http.recorded_requests[1][0].get('X-LaunchDarkly-Payload-ID') + assert first_payload_id != second_payload_id + + +# --------------------------------------------------------------------------- +# Diagnostics +# --------------------------------------------------------------------------- + +async def test_init_diagnostic_event_sent(): + mock_http = MockAioHttp() + async with make_processor(mock_http, diagnostic_opt_out=False) as ep: + diag_init = await flush_and_get_events(ep, mock_http) + assert len(diag_init) == 6 + assert diag_init['kind'] == 'diagnostic-init' + + +async def test_periodic_diagnostic_includes_events_in_batch(): + mock_http = MockAioHttp() + async with make_processor(mock_http, diagnostic_opt_out=False) as ep: + # Ignore init event + await flush_and_get_events(ep, mock_http) + # Send a payload with a single event + ep.send_event(EventInputIdentify(timestamp, context)) + await flush_and_get_events(ep, mock_http) + + await ep._send_diagnostic() + diag_event = await flush_and_get_events(ep, mock_http) + assert len(diag_event) == 8 + assert diag_event['kind'] == 'diagnostic' + assert diag_event['eventsInLastBatch'] == 1 + assert diag_event['deduplicatedUsers'] == 0 + + +async def test_periodic_diagnostic_includes_deduplicated_users(): + mock_http = MockAioHttp() + async with make_processor(mock_http, diagnostic_opt_out=False) as ep: + # Ignore init event + await flush_and_get_events(ep, mock_http) + # Send two custom events with the same user to cause a user deduplication + e0 = EventInputCustom(timestamp, context, 'event1', None, None) + e1 = EventInputCustom(timestamp, context, 'event2', None, None) + ep.send_event(e0) + ep.send_event(e1) + await flush_and_get_events(ep, mock_http) + + await ep._send_diagnostic() + diag_event = await flush_and_get_events(ep, mock_http) + assert len(diag_event) == 8 + assert diag_event['kind'] == 'diagnostic' + assert diag_event['eventsInLastBatch'] == 3 + assert diag_event['deduplicatedUsers'] == 1 + + +async def test_event_schema_not_set_on_diagnostic_send(): + mock_http = MockAioHttp() + async with make_processor(mock_http, diagnostic_opt_out=False) as ep: + await ep._wait_until_inactive() + assert mock_http.request_headers.get('X-LaunchDarkly-Event-Schema') is None + + +# --------------------------------------------------------------------------- +# HTTP error handling +# --------------------------------------------------------------------------- + +async def verify_unrecoverable_http_error(status: int): + mock_http = MockAioHttp() + async with make_processor(mock_http, sdk_key='SDK_KEY') as ep: + mock_http.set_response_status(status) + ep.send_event(EventInputIdentify(timestamp, context)) + ep.flush() + await ep._wait_until_inactive() + mock_http.reset() + + ep.send_event(EventInputIdentify(timestamp, context)) + ep.flush() + await ep._wait_until_inactive() + assert mock_http.request_data is None + + +async def verify_recoverable_http_error(status: int): + mock_http = MockAioHttp() + async with make_processor(mock_http, sdk_key='SDK_KEY') as ep: + mock_http.set_response_status(status) + ep.send_event(EventInputIdentify(timestamp, context)) + ep.flush() + await ep._wait_until_inactive() + mock_http.reset() + + ep.send_event(EventInputIdentify(timestamp, context)) + ep.flush() + await ep._wait_until_inactive() + assert mock_http.request_data is not None + + +async def test_no_more_payloads_are_sent_after_401_error(): + await verify_unrecoverable_http_error(401) + + +async def test_no_more_payloads_are_sent_after_403_error(): + await verify_unrecoverable_http_error(403) + + +async def test_will_still_send_after_408_error(): + await verify_recoverable_http_error(408) + + +async def test_will_still_send_after_429_error(): + await verify_recoverable_http_error(429) + + +async def test_will_still_send_after_500_error(): + await verify_recoverable_http_error(500) + + +# --------------------------------------------------------------------------- +# Inbox capacity +# --------------------------------------------------------------------------- + +async def test_does_not_block_on_full_inbox(): + config = AsyncConfig("fake_sdk_key", events_max_pending=1) # this sets the size of both the inbox and the outbox to 1 + ep_inbox_holder = [None] + + def dispatcher_factory(inbox, config, http, diag): + ep_inbox_holder[0] = inbox # it's an array because otherwise it's hard for a closure to modify a variable + return None # the dispatcher object itself doesn't matter, we only manipulate the inbox + + async def event_consumer(): + while True: + message = await ep_inbox.get() + if message.type == 'stop': + message.param.set() + return + + mock_http = MockAioHttp() + ep = DefaultAsyncEventProcessor(config, mock_http, dispatcher_factory) + ep_inbox = ep_inbox_holder[0] + event1 = EventInputCustom(timestamp, context, 'event1') + event2 = EventInputCustom(timestamp, context, 'event2') + ep.send_event(event1) + ep.send_event(event2) # this event should be dropped - inbox is full + message1 = await ep_inbox.get(block=False) + had_no_more = ep_inbox.empty() + consumer = asyncio.ensure_future(event_consumer()) + await ep.stop() + await consumer + assert message1.param == event1 + assert had_no_more + + +# --------------------------------------------------------------------------- +# Compression +# --------------------------------------------------------------------------- + +async def test_event_payload_is_gzip_compressed_when_enabled(): + mock_http = MockAioHttp() + async with make_processor(mock_http, enable_event_compression=True) as ep: + ep.send_event(EventInputIdentify(timestamp, context)) + ep.flush() + await ep._wait_until_inactive() + + assert mock_http.request_headers.get('Content-Encoding') == 'gzip' + output = json.loads(gzip.decompress(mock_http.request_data)) + assert len(output) == 1 + assert output[0]['kind'] == 'identify'