Skip to content

fix: Use monotonic clocks for migration latency and FDv2 stream-init timing #539

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
tanderson-ld wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix: Use monotonic clocks for migration latency and FDv2 stream-init timing #539

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 28 additions & 18 deletions ldclient/impl/datasourcev2/async_streaming.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
"""

import json
from time import time
import time
from typing import AsyncGenerator, Callable, Optional, Tuple
from urllib import parse

Expand All @@ -24,7 +24,13 @@
with_fallback_signal
)
from ldclient.impl.datasystem import DiagnosticAccumulator, DiagnosticSource
from ldclient.impl.util import _LD_ENVID_HEADER, _LD_FD_FALLBACK_HEADER, log
from ldclient.impl.util import (
_LD_ENVID_HEADER,
_LD_FD_FALLBACK_HEADER,
current_time_millis,
log,
monotonic_seconds
)
from ldclient.interfaces import (
AsyncSynchronizer,
ChangeSetBuilder,
Expand Down Expand Up @@ -117,7 +123,7 @@ def __init__(
self._owned_session: Optional[aiohttp.ClientSession] = None
self._running = False
self._diagnostic_accumulator: Optional[DiagnosticAccumulator] = None
self._connection_attempt_start_time: Optional[float] = None
self._connection_attempt_started_monotonic: Optional[float] = None

def set_diagnostic_accumulator(self, diagnostic_accumulator: DiagnosticAccumulator):
self._diagnostic_accumulator = diagnostic_accumulator
Expand Down Expand Up @@ -150,7 +156,7 @@ async def sync(self, ss: SelectorStore) -> AsyncGenerator[Update, None]:

change_set_builder = ChangeSetBuilder()
self._running = True
self._connection_attempt_start_time = time()
self._connection_attempt_started_monotonic = monotonic_seconds()

envid = None
# A Start action with the X-LD-FD-Fallback: true header sets fallback_requested.
Expand All @@ -167,6 +173,7 @@ async def sync(self, ss: SelectorStore) -> AsyncGenerator[Update, None]:
# emit a fault with no-error. We can ignore this since we want
# the connection to continue.
if action.error is None:
self._connection_attempt_started_monotonic = None
continue

if action.headers is not None:
Expand All @@ -183,10 +190,12 @@ async def sync(self, ss: SelectorStore) -> AsyncGenerator[Update, None]:
break
continue

if isinstance(action, Start) and action.headers is not None:
envid = action.headers.get(_LD_ENVID_HEADER, envid)
if action.headers.get(_LD_FD_FALLBACK_HEADER) == 'true':
fallback_requested = True
if isinstance(action, Start):
self._connection_attempt_started_monotonic = monotonic_seconds()
if action.headers is not None:
envid = action.headers.get(_LD_ENVID_HEADER, envid)
if action.headers.get(_LD_FD_FALLBACK_HEADER) == 'true':
fallback_requested = True

if not isinstance(action, Event):
continue
Expand All @@ -195,7 +204,7 @@ async def sync(self, ss: SelectorStore) -> AsyncGenerator[Update, None]:
update = process_message(action, change_set_builder, envid)
if update is not None:
self._record_stream_init(False)
self._connection_attempt_start_time = None
self._connection_attempt_started_monotonic = None
if fallback_requested:
# The completed update is the natural moment to honor
# the latched directive: yield once with the signal,
Expand All @@ -219,11 +228,12 @@ async def sync(self, ss: SelectorStore) -> AsyncGenerator[Update, None]:
"Error while handling stream event; will restart stream: %s", e
)
await self._sse.interrupt()
self._connection_attempt_started_monotonic = None

yield with_fallback_signal(Update(
state=DataSourceState.INTERRUPTED,
error=DataSourceErrorInfo(
DataSourceErrorKind.UNKNOWN, 0, time(), str(e)
DataSourceErrorKind.UNKNOWN, 0, time.time(), str(e)
),
fallback_to_fdv1=False,
environment_id=envid,
Expand Down Expand Up @@ -253,10 +263,12 @@ async def _close_owned_session(self):
self._owned_session = None

def _record_stream_init(self, failed: bool):
if self._diagnostic_accumulator and self._connection_attempt_start_time:
current_time = int(time() * 1000)
elapsed = current_time - int(self._connection_attempt_start_time * 1000)
self._diagnostic_accumulator.record_stream_init(current_time, elapsed if elapsed >= 0 else 0, failed)
if self._diagnostic_accumulator and self._connection_attempt_started_monotonic is not None:
# The reported timestamp is a wall-clock value; the duration comes from
# the monotonic clock so a wall-clock step cannot distort it.
current_time = current_time_millis()
elapsed = int((monotonic_seconds() - self._connection_attempt_started_monotonic) * 1000)
self._diagnostic_accumulator.record_stream_init(current_time, elapsed, failed)

async def _handle_error(self, error: Exception, envid: Optional[str]) -> Tuple[Optional[Update], bool]:
"""
Expand All @@ -273,11 +285,9 @@ async def _handle_error(self, error: Exception, envid: Optional[str]) -> Tuple[O
if not self._running:
return (None, False) # don't retry if we've been deliberately stopped

decision = classify_stream_error(
error, self._sse.next_retry_delay, envid # type: ignore
)
decision = classify_stream_error(error, envid)
self._record_stream_init(True)
self._connection_attempt_start_time = decision.next_attempt_start
self._connection_attempt_started_monotonic = None
if decision.should_stop:
await self.stop()

Expand Down
44 changes: 26 additions & 18 deletions ldclient/impl/datasourcev2/streaming.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
"""

import json
from time import time
import time
from typing import Callable, Generator, Optional, Tuple
from urllib import parse

Expand All @@ -28,7 +28,13 @@
)
from ldclient.impl.datasystem import DiagnosticAccumulator, DiagnosticSource
from ldclient.impl.http import HTTPFactory, _base_headers
from ldclient.impl.util import _LD_ENVID_HEADER, _LD_FD_FALLBACK_HEADER, log
from ldclient.impl.util import (
_LD_ENVID_HEADER,
_LD_FD_FALLBACK_HEADER,
current_time_millis,
log,
monotonic_seconds
)
from ldclient.interfaces import (
ChangeSetBuilder,
DataSourceErrorInfo,
Expand Down Expand Up @@ -126,7 +132,7 @@ def __init__(self,
self._sse: Optional[SSEClient] = None
self._running = False
self._diagnostic_accumulator: Optional[DiagnosticAccumulator] = None
self._connection_attempt_start_time: Optional[float] = None
self._connection_attempt_started_monotonic: Optional[float] = None

def set_diagnostic_accumulator(self, diagnostic_accumulator: DiagnosticAccumulator):
self._diagnostic_accumulator = diagnostic_accumulator
Expand Down Expand Up @@ -158,7 +164,7 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]:

change_set_builder = ChangeSetBuilder()
self._running = True
self._connection_attempt_start_time = time()
self._connection_attempt_started_monotonic = monotonic_seconds()

envid = None
# fallback_requested is set when a Start action carries
Expand All @@ -176,6 +182,7 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]:
# emit a fault with no-error. We can ignore this since we want
# the connection to continue.
if action.error is None:
self._connection_attempt_started_monotonic = None
continue

if action.headers is not None:
Expand All @@ -192,10 +199,12 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]:
break
continue

if isinstance(action, Start) and action.headers is not None:
envid = action.headers.get(_LD_ENVID_HEADER, envid)
if action.headers.get(_LD_FD_FALLBACK_HEADER) == 'true':
fallback_requested = True
if isinstance(action, Start):
self._connection_attempt_started_monotonic = monotonic_seconds()
if action.headers is not None:
envid = action.headers.get(_LD_ENVID_HEADER, envid)
if action.headers.get(_LD_FD_FALLBACK_HEADER) == 'true':
fallback_requested = True

if not isinstance(action, Event):
continue
Expand All @@ -204,7 +213,7 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]:
update = process_message(action, change_set_builder, envid)
if update is not None:
self._record_stream_init(False)
self._connection_attempt_start_time = None
self._connection_attempt_started_monotonic = None
if fallback_requested:
# The completed update is the natural moment to honor
# the latched directive: yield once with the signal,
Expand All @@ -228,11 +237,12 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]:
"Error while handling stream event; will restart stream: %s", e
)
self._sse.interrupt()
self._connection_attempt_started_monotonic = None

yield with_fallback_signal(Update(
state=DataSourceState.INTERRUPTED,
error=DataSourceErrorInfo(
DataSourceErrorKind.UNKNOWN, 0, time(), str(e)
DataSourceErrorKind.UNKNOWN, 0, time.time(), str(e)
),
fallback_to_fdv1=False,
environment_id=envid,
Expand All @@ -252,10 +262,10 @@ def stop(self):
self._sse.close()

def _record_stream_init(self, failed: bool):
if self._diagnostic_accumulator and self._connection_attempt_start_time:
current_time = int(time() * 1000)
elapsed = current_time - int(self._connection_attempt_start_time * 1000)
self._diagnostic_accumulator.record_stream_init(current_time, elapsed if elapsed >= 0 else 0, failed)
if self._diagnostic_accumulator and self._connection_attempt_started_monotonic is not None:
current_time = current_time_millis()
elapsed = int((monotonic_seconds() - self._connection_attempt_started_monotonic) * 1000)
self._diagnostic_accumulator.record_stream_init(current_time, elapsed, failed)

def _handle_error(self, error: Exception, envid: Optional[str]) -> Tuple[Optional[Update], bool]:
"""
Expand All @@ -272,11 +282,9 @@ def _handle_error(self, error: Exception, envid: Optional[str]) -> Tuple[Optiona
if not self._running:
return (None, False) # don't retry if we've been deliberately stopped

decision = classify_stream_error(
error, self._sse.next_retry_delay, envid # type: ignore
)
decision = classify_stream_error(error, envid)
self._record_stream_init(True)
self._connection_attempt_start_time = decision.next_attempt_start
self._connection_attempt_started_monotonic = None
if decision.should_stop:
self.stop()

Expand Down
20 changes: 5 additions & 15 deletions ldclient/impl/datasourcev2/streaming_common.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -132,26 +132,23 @@ class StreamErrorDecision:

The values are computed without any I/O so both the sync and async
synchronizers share one decision, then apply the side effects themselves:
they record a failed stream init, set the connection attempt start time to
``next_attempt_start``, stop the synchronizer when ``should_stop`` is set,
and retry the connection when ``should_continue`` is set.
they record a failed stream init, stop the synchronizer when ``should_stop``
is set, and retry the connection when ``should_continue`` is set.
"""
update: Update
should_continue: bool
should_stop: bool
next_attempt_start: Optional[float]


def classify_stream_error(
error: Exception, next_retry_delay: float, envid: Optional[str]
error: Exception, envid: Optional[str]
) -> StreamErrorDecision:
"""
Decide how to react to a streaming connection error.

The caller must first check that the synchronizer is still running; this
function assumes it is. It returns a decision the caller applies. The caller
always records a failed stream init before it sets the connection attempt
start time to the returned ``next_attempt_start``.
function assumes it is. It returns a decision the caller applies, recording
a failed stream init before doing so.
"""
if isinstance(error, json.decoder.JSONDecodeError):
log.error("Unexpected error on stream connection: %s, will retry", error)
Expand All @@ -167,12 +164,9 @@ def classify_stream_error(
update=update,
should_continue=True,
should_stop=False,
next_attempt_start=time() + next_retry_delay,
)

if isinstance(error, HTTPStatusError):
next_attempt_start: Optional[float] = time() + next_retry_delay

error_info = DataSourceErrorInfo(
DataSourceErrorKind.ERROR_RESPONSE,
error.status,
Expand All @@ -194,7 +188,6 @@ def classify_stream_error(
update=update,
should_continue=False,
should_stop=True,
next_attempt_start=next_attempt_start,
)

http_error_message_result = http_error_message(
Expand All @@ -218,15 +211,13 @@ def classify_stream_error(
update=update,
should_continue=False,
should_stop=True,
next_attempt_start=None,
)

log.warning(http_error_message_result)
return StreamErrorDecision(
update=update,
should_continue=True,
should_stop=False,
next_attempt_start=next_attempt_start,
)

log.warning("Unexpected error on stream connection: %s, will retry", error)
Expand All @@ -244,5 +235,4 @@ def classify_stream_error(
update=update,
should_continue=True,
should_stop=False,
next_attempt_start=time() + next_retry_delay,
)
1 change: 1 addition & 0 deletions ldclient/impl/events/diagnostics.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ def __init__(self, diagnostic_id):
self._stream_inits = []

def record_stream_init(self, timestamp, duration, failed):
duration = max(0, duration)
with self._state_lock:
self._stream_inits.append({'timestamp': timestamp, 'durationMillis': duration, 'failed': failed})

Expand Down
8 changes: 4 additions & 4 deletions ldclient/migrations/async_migrator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@

import asyncio
from abc import ABC, abstractmethod
from datetime import datetime
from datetime import timedelta
from random import Random
from typing import (
TYPE_CHECKING,
Expand All @@ -15,7 +15,7 @@
)

from ldclient.impl.sampler import Sampler
from ldclient.impl.util import Result, log
from ldclient.impl.util import Result, log, monotonic_seconds
from ldclient.migrations.tracker import OpTracker
from ldclient.migrations.types import (
ExecutionOrder,
Expand Down Expand Up @@ -419,7 +419,7 @@ async def run(self) -> OperationResult:
"""
Execute the configured operation and track any available measurements.
"""
start = datetime.now()
start = monotonic_seconds()

try:
result = await self.__fn(self.__payload)
Expand All @@ -428,7 +428,7 @@ async def run(self) -> OperationResult:

# Record required tracker measurements
if self.__measure_latency:
self.__tracker.latency(self.__origin, datetime.now() - start)
self.__tracker.latency(self.__origin, timedelta(seconds=monotonic_seconds() - start))

if self.__measure_errors and not result.is_success():
self.__tracker.error(self.__origin)
Expand Down
8 changes: 4 additions & 4 deletions ldclient/migrations/migrator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,12 +2,12 @@

import concurrent.futures
from abc import ABC, abstractmethod
from datetime import datetime
from datetime import timedelta
from random import Random
from typing import TYPE_CHECKING, Any, Optional, Tuple, Union

from ldclient.impl.sampler import Sampler
from ldclient.impl.util import Result
from ldclient.impl.util import Result, monotonic_seconds
from ldclient.migrations.tracker import OpTracker
from ldclient.migrations.types import (
ExecutionOrder,
Expand Down Expand Up @@ -313,7 +313,7 @@ def run(self) -> OperationResult:
"""
Execute the configured operation and track any available measurements.
"""
start = datetime.now()
start = monotonic_seconds()

try:
result = self.__fn(self.__payload)
Expand All @@ -322,7 +322,7 @@ def run(self) -> OperationResult:

# Record required tracker measurements
if self.__measure_latency:
self.__tracker.latency(self.__origin, datetime.now() - start)
self.__tracker.latency(self.__origin, timedelta(seconds=monotonic_seconds() - start))

if self.__measure_errors and not result.is_success():
self.__tracker.error(self.__origin)
Expand Down
Loading
Loading