pikers
/
piker
3
0
Fork 0
Code Issues 4 Pull Requests 6 Packages Projects Releases Wiki Activity

Add "no-tsdb-found" history load length defaults

pre_flow
Tyler Goodlet 2022-05-15 13:36:08 -04:00
parent 7555cb4318
commit 54ae86544a
1 changed files with 28 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
piker/data/feed.py
View File

@ -227,7 +227,7 @@ def diff_history(
# the + 1 is because ``last_tsdb_dt`` is pulled from # the + 1 is because ``last_tsdb_dt`` is pulled from
# the last row entry for the ``'time'`` field retreived # the last row entry for the ``'time'`` field retreived
# from the tsdb. # from the tsdb.
to_push = array[abs(s_diff)+1:] to_push = array[abs(s_diff) + 1:]
else: else:
# pass back only the portion of the array that is # pass back only the portion of the array that is
@ -250,6 +250,7 @@ async def start_backfill(
last_tsdb_dt: Optional[datetime] = None, last_tsdb_dt: Optional[datetime] = None,
storage: Optional[Storage] = None, storage: Optional[Storage] = None,
write_tsdb: bool = True, write_tsdb: bool = True,
tsdb_is_up: bool = False,
task_status: TaskStatus[trio.CancelScope] = trio.TASK_STATUS_IGNORED, task_status: TaskStatus[trio.CancelScope] = trio.TASK_STATUS_IGNORED,
@ -265,8 +266,8 @@ async def start_backfill(
# sample period step size in seconds # sample period step size in seconds
step_size_s = ( step_size_s = (
pendulum.from_timestamp(times[-1]) - pendulum.from_timestamp(times[-1])
pendulum.from_timestamp(times[-2]) - pendulum.from_timestamp(times[-2])
).seconds ).seconds
# "frame"'s worth of sample period steps in seconds # "frame"'s worth of sample period steps in seconds
@ -291,25 +292,33 @@ async def start_backfill(
# let caller unblock and deliver latest history frame # let caller unblock and deliver latest history frame
task_status.started((shm, start_dt, end_dt, bf_done)) task_status.started((shm, start_dt, end_dt, bf_done))
# based on the sample step size, maybe load a certain amount history
if last_tsdb_dt is None: if last_tsdb_dt is None:
# maybe a better default (they don't seem to define epoch?!) if step_size_s not in (1, 60):
# based on the sample step size load a certain amount
# history
if step_size_s == 1:
last_tsdb_dt = pendulum.now().subtract(days=2)
elif step_size_s == 60:
last_tsdb_dt = pendulum.now().subtract(years=2)
else:
raise ValueError( raise ValueError(
'`piker` only needs to support 1m and 1s sampling ' '`piker` only needs to support 1m and 1s sampling '
'but ur api is trying to deliver a longer ' 'but ur api is trying to deliver a longer '
f'timeframe of {step_size_s} ' 'seconds.. so ye, dun ' f'timeframe of {step_size_s} ' 'seconds.. so ye, dun '
'do dat bruh.' 'do dat brudder.'
) )
# when no tsdb "last datum" is provided, we just load
# some near-term history.
periods = {
1: {'days': 1},
60: {'days': 14},
}
if tsdb_is_up:
# do a decently sized backfill and load it into storage.
periods = {
1: {'days': 6},
60: {'years': 2},
}
kwargs = periods[step_size_s]
last_tsdb_dt = start_dt.subtract(**kwargs)
# configure async query throttling # configure async query throttling
erlangs = config.get('erlangs', 1) erlangs = config.get('erlangs', 1)
rate = config.get('rate', 1) rate = config.get('rate', 1)