mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-12 08:42:18 +00:00
fix(alerts): keep the deployment-wide run every five minutes
This commit is contained in:
parent
02128eaa92
commit
5492b671e5
2 changed files with 16 additions and 2 deletions
|
|
@ -151,7 +151,7 @@ _LIMITS: Final = f"""SELECT set_config('statement_timeout', '{_STATEMENT_TIMEOUT
|
|||
_CLAIM: Final = """INSERT INTO "LiteLLM_PerformanceAlertState" (id, owner, next_run_at, updated_at)
|
||||
VALUES (1, $1, now() + $2::interval, now())
|
||||
ON CONFLICT (id) DO UPDATE SET owner = EXCLUDED.owner, next_run_at = EXCLUDED.next_run_at, updated_at = now()
|
||||
WHERE "LiteLLM_PerformanceAlertState".next_run_at <= now()
|
||||
WHERE "LiteLLM_PerformanceAlertState".next_run_at <= now() + interval '30 seconds'
|
||||
RETURNING day, scope, resume_after AS cursor, wrapped, checked, skipped, completed_at IS NOT NULL AS completed"""
|
||||
_SAVE: Final = """UPDATE "LiteLLM_PerformanceAlertState"
|
||||
SET day = $2, scope = $3, resume_after = $4, wrapped = $5, checked = $6, skipped = $7,
|
||||
|
|
@ -179,7 +179,9 @@ class PerformanceAlertRepository:
|
|||
yield transaction
|
||||
|
||||
async def claim(self, interval: timedelta) -> Claim | None:
|
||||
"""Take this deployment's next run, at most one per `interval` across every worker and pod."""
|
||||
"""Take this deployment's next run, at most one per `interval` across every worker and pod. A worker whose
|
||||
timer fires up to 30 seconds early still takes it, so the claim's own delay never pushes a run back a whole
|
||||
interval."""
|
||||
owner: Final = uuid.uuid4().hex
|
||||
async with self._transaction() as transaction:
|
||||
rows: Final = _STATE_ROWS.validate_python(
|
||||
|
|
|
|||
|
|
@ -197,3 +197,15 @@ async def test_a_resource_that_cannot_be_read_is_skipped_without_blocking_the_re
|
|||
async def test_only_one_of_many_concurrent_workers_gets_each_run(fixture: Fixture) -> None:
|
||||
claims: Final = await asyncio.gather(*(fixture.repository.claim(timedelta(minutes=5)) for _ in range(8)))
|
||||
assert sum(claim is not None for claim in claims) == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("seconds_until_due", "claimed"), ((20, True), (60, False)))
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_timer_firing_just_before_the_run_is_due_still_takes_it(
|
||||
fixture: Fixture, seconds_until_due: int, claimed: bool
|
||||
) -> None:
|
||||
write_rows(
|
||||
'INSERT INTO "LiteLLM_PerformanceAlertState" (id, owner, next_run_at) VALUES (1, %s, now() + %s::interval)',
|
||||
("earlier-worker", f"{seconds_until_due} seconds"),
|
||||
)
|
||||
assert (await fixture.repository.claim(timedelta(minutes=5)) is not None) is claimed
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue