mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-29 06:33:28 +00:00
fix(proxy): compare reload requests against pod data age seeded at boot
A pod that had never reloaded kept its in-memory clock at None, and with no interval configured nothing ever set it, so every manual reload request was ignored by every pod except the one serving the click (Greptile P1 on the previous commit). Seed the per-pod timestamp at boot as the time its data was loaded and reload whenever a request or the interval is older than that, which also removes both None special cases from the due predicate. A schedule whose row has no last_run_at fires on the next tick so the first run does not wait a full interval.
This commit is contained in:
parent
bfa47535b5
commit
b7ac3e237b
4 changed files with 86 additions and 61 deletions
|
|
@ -9,8 +9,8 @@ reload job and the manual reload endpoints own the dedicated ``last_run_at`` /
|
|||
memory so the Admin UI still reports the last execution after a restart and across pods.
|
||||
``reload_requested_at`` fans a manual reload out to every pod without any pod having to
|
||||
write the flag back down, so no pod can starve another of the request. Each pod tracks
|
||||
its own last reload in memory and compares it against those timestamps to decide when to
|
||||
refresh its in-memory copy of the data.
|
||||
only when its own in-memory copy was loaded (seeded at boot) and reloads whenever that
|
||||
copy is older than the request or the configured interval.
|
||||
"""
|
||||
|
||||
from collections.abc import Mapping
|
||||
|
|
@ -127,35 +127,30 @@ def reload_schedule_status(schedule: ReloadSchedule | None) -> ReloadScheduleSta
|
|||
def pod_reload_is_due(
|
||||
*,
|
||||
schedule: ReloadSchedule,
|
||||
pod_last_reload: datetime | None,
|
||||
pod_data_loaded_at: datetime,
|
||||
current_time: datetime,
|
||||
description: str,
|
||||
) -> bool:
|
||||
"""
|
||||
Whether this pod should reload now, based on its own last reload rather than the
|
||||
persisted one, so every pod refreshes its in-memory data on the configured interval.
|
||||
A pod that has never reloaded ignores ``reload_requested_at``: whatever it booted with
|
||||
is at least as fresh as the request
|
||||
Whether this pod should reload now: when its in-memory data (loaded at boot or at its
|
||||
last reload) is older than a manual request or the configured interval. A schedule that
|
||||
has never run anywhere fires immediately so the first run is not one interval away
|
||||
"""
|
||||
if (
|
||||
schedule.reload_requested_at is not None
|
||||
and pod_last_reload is not None
|
||||
and schedule.reload_requested_at > pod_last_reload
|
||||
):
|
||||
if schedule.reload_requested_at is not None and schedule.reload_requested_at > pod_data_loaded_at:
|
||||
verbose_proxy_logger.info("%s reload triggered by manual reload request", description)
|
||||
return True
|
||||
if schedule.interval_hours is None:
|
||||
return False
|
||||
if pod_last_reload is None:
|
||||
verbose_proxy_logger.info("%s reload triggered - no previous reload time recorded", description)
|
||||
if schedule.last_run_at is None:
|
||||
verbose_proxy_logger.info("%s reload triggered - schedule has never run", description)
|
||||
return True
|
||||
hours_since_last_reload = (current_time - pod_last_reload).total_seconds() / 3600
|
||||
if hours_since_last_reload < schedule.interval_hours:
|
||||
hours_since_data_loaded = (current_time - pod_data_loaded_at).total_seconds() / 3600
|
||||
if hours_since_data_loaded < schedule.interval_hours:
|
||||
return False
|
||||
verbose_proxy_logger.info(
|
||||
"%s reload triggered by interval. Hours since last reload: %.2f, Interval: %s",
|
||||
"%s reload triggered by interval. Hours since data loaded: %.2f, Interval: %s",
|
||||
description,
|
||||
hours_since_last_reload,
|
||||
hours_since_data_loaded,
|
||||
schedule.interval_hours,
|
||||
)
|
||||
return True
|
||||
|
|
|
|||
|
|
@ -3856,8 +3856,8 @@ class ProxyConfig:
|
|||
self._last_semantic_filter_config: Optional[Dict[str, Any]] = None
|
||||
self._last_hashicorp_vault_config: Optional[Dict[str, Any]] = None
|
||||
self.worker_registry: List["WorkerRegistryEntry"] = []
|
||||
self.last_model_cost_map_reload: datetime | None = None
|
||||
self.last_anthropic_beta_headers_reload: datetime | None = None
|
||||
self.model_cost_map_loaded_at: datetime = utc_now()
|
||||
self.anthropic_beta_headers_loaded_at: datetime = utc_now()
|
||||
|
||||
def is_yaml(self, config_file_path: str) -> bool:
|
||||
if not os.path.isfile(config_file_path):
|
||||
|
|
@ -6443,7 +6443,7 @@ class ProxyConfig:
|
|||
current_time = utc_now()
|
||||
if not pod_reload_is_due(
|
||||
schedule=schedule,
|
||||
pod_last_reload=self.last_model_cost_map_reload,
|
||||
pod_data_loaded_at=self.model_cost_map_loaded_at,
|
||||
current_time=current_time,
|
||||
description="Model cost map",
|
||||
):
|
||||
|
|
@ -6451,7 +6451,7 @@ class ProxyConfig:
|
|||
|
||||
models_count = await asyncio.to_thread(_reload_model_cost_map_in_process)
|
||||
|
||||
self.last_model_cost_map_reload = current_time
|
||||
self.model_cost_map_loaded_at = current_time
|
||||
await record_reload_run(prisma_client, MODEL_COST_MAP_RELOAD_PARAM_NAME, current_time)
|
||||
|
||||
verbose_proxy_logger.info(f"Model cost map reloaded successfully. Models count: {models_count}")
|
||||
|
|
@ -6472,7 +6472,7 @@ class ProxyConfig:
|
|||
current_time = utc_now()
|
||||
if not pod_reload_is_due(
|
||||
schedule=schedule,
|
||||
pod_last_reload=self.last_anthropic_beta_headers_reload,
|
||||
pod_data_loaded_at=self.anthropic_beta_headers_loaded_at,
|
||||
current_time=current_time,
|
||||
description="Anthropic beta headers",
|
||||
):
|
||||
|
|
@ -6484,7 +6484,7 @@ class ProxyConfig:
|
|||
|
||||
new_config = await asyncio.to_thread(reload_beta_headers_config)
|
||||
|
||||
self.last_anthropic_beta_headers_reload = current_time
|
||||
self.anthropic_beta_headers_loaded_at = current_time
|
||||
await record_reload_run(prisma_client, ANTHROPIC_BETA_HEADERS_RELOAD_PARAM_NAME, current_time)
|
||||
|
||||
provider_count = sum(1 for k in new_config.keys() if k != "provider_aliases" and k != "description")
|
||||
|
|
@ -15570,7 +15570,7 @@ async def reload_model_cost_map(
|
|||
models_count = await asyncio.to_thread(_reload_model_cost_map_in_process)
|
||||
|
||||
current_time = utc_now()
|
||||
proxy_config.last_model_cost_map_reload = current_time
|
||||
proxy_config.model_cost_map_loaded_at = current_time
|
||||
|
||||
# Stamp the shared last run and the reload request so other pods reload too
|
||||
await record_manual_reload(prisma_client, MODEL_COST_MAP_RELOAD_PARAM_NAME, current_time)
|
||||
|
|
@ -15800,7 +15800,7 @@ async def reload_anthropic_beta_headers(
|
|||
new_config = await asyncio.to_thread(reload_beta_headers_config)
|
||||
|
||||
current_time = utc_now()
|
||||
proxy_config.last_anthropic_beta_headers_reload = current_time
|
||||
proxy_config.anthropic_beta_headers_loaded_at = current_time
|
||||
|
||||
# Stamp the shared last run and the reload request so other pods reload too
|
||||
await record_manual_reload(prisma_client, ANTHROPIC_BETA_HEADERS_RELOAD_PARAM_NAME, current_time)
|
||||
|
|
|
|||
|
|
@ -96,30 +96,37 @@ def test_next_run_needs_both_interval_and_last_run():
|
|||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"schedule, pod_last_reload, expected",
|
||||
"schedule, pod_data_loaded_at, expected",
|
||||
[
|
||||
(ReloadSchedule(reload_requested_at=datetime(2024, 1, 1, 11, 0, tzinfo=timezone.utc)), LAST_RUN, True),
|
||||
(
|
||||
ReloadSchedule(interval_hours=6, reload_requested_at=datetime(2024, 1, 1, 11, 30, tzinfo=timezone.utc)),
|
||||
ReloadSchedule(
|
||||
interval_hours=6,
|
||||
reload_requested_at=datetime(2024, 1, 1, 11, 30, tzinfo=timezone.utc),
|
||||
last_run_at=datetime(2024, 1, 1, 11, 0, tzinfo=timezone.utc),
|
||||
),
|
||||
datetime(2024, 1, 1, 11, 0, tzinfo=timezone.utc),
|
||||
True,
|
||||
),
|
||||
(ReloadSchedule(reload_requested_at=LAST_RUN), LAST_RUN, False),
|
||||
(ReloadSchedule(reload_requested_at=datetime(2024, 1, 1, 5, 0, tzinfo=timezone.utc)), LAST_RUN, False),
|
||||
(ReloadSchedule(interval_hours=6, reload_requested_at=LAST_RUN), LAST_RUN, True),
|
||||
(ReloadSchedule(reload_requested_at=NOW), None, False),
|
||||
(ReloadSchedule(interval_hours=6, reload_requested_at=NOW), None, True),
|
||||
(ReloadSchedule(), None, False),
|
||||
(ReloadSchedule(interval_hours=6), None, True),
|
||||
(ReloadSchedule(interval_hours=6), datetime(2024, 1, 1, 11, 0, tzinfo=timezone.utc), False),
|
||||
(ReloadSchedule(interval_hours=6), LAST_RUN, True),
|
||||
(ReloadSchedule(reload_requested_at=LAST_RUN), datetime(2024, 1, 1, 11, 0, tzinfo=timezone.utc), False),
|
||||
(ReloadSchedule(interval_hours=6, reload_requested_at=LAST_RUN, last_run_at=LAST_RUN), LAST_RUN, True),
|
||||
(ReloadSchedule(), LAST_RUN, False),
|
||||
(ReloadSchedule(interval_hours=6), datetime(2024, 1, 1, 11, 59, tzinfo=timezone.utc), True),
|
||||
(
|
||||
ReloadSchedule(interval_hours=6, last_run_at=datetime(2024, 1, 1, 6, 30, tzinfo=timezone.utc)),
|
||||
datetime(2024, 1, 1, 11, 0, tzinfo=timezone.utc),
|
||||
False,
|
||||
),
|
||||
(ReloadSchedule(interval_hours=6, last_run_at=LAST_RUN), LAST_RUN, True),
|
||||
],
|
||||
)
|
||||
def test_pod_reload_is_due(schedule, pod_last_reload, expected):
|
||||
def test_pod_reload_is_due(schedule, pod_data_loaded_at, expected):
|
||||
assert (
|
||||
pod_reload_is_due(
|
||||
schedule=schedule,
|
||||
pod_last_reload=pod_last_reload,
|
||||
pod_data_loaded_at=pod_data_loaded_at,
|
||||
current_time=NOW,
|
||||
description="test",
|
||||
)
|
||||
|
|
@ -127,16 +134,18 @@ def test_pod_reload_is_due(schedule, pod_last_reload, expected):
|
|||
)
|
||||
|
||||
|
||||
def test_manual_request_reaches_every_pod_that_has_not_seen_it():
|
||||
def test_manual_request_reaches_every_pod_with_older_data():
|
||||
"""Nothing clears reload_requested_at, so each pod reloads exactly once per request:
|
||||
due while its own last reload predates the request, not due afterwards"""
|
||||
due while its data predates the request, not due once refreshed or booted after it"""
|
||||
schedule = ReloadSchedule(reload_requested_at=datetime(2024, 1, 1, 11, 0, tzinfo=timezone.utc))
|
||||
|
||||
assert pod_reload_is_due(schedule=schedule, pod_last_reload=LAST_RUN, current_time=NOW, description="test") is True
|
||||
assert (
|
||||
pod_reload_is_due(schedule=schedule, pod_data_loaded_at=LAST_RUN, current_time=NOW, description="test") is True
|
||||
)
|
||||
assert (
|
||||
pod_reload_is_due(
|
||||
schedule=schedule,
|
||||
pod_last_reload=datetime(2024, 1, 1, 11, 0, 1, tzinfo=timezone.utc),
|
||||
pod_data_loaded_at=datetime(2024, 1, 1, 11, 0, 1, tzinfo=timezone.utc),
|
||||
current_time=NOW,
|
||||
description="test",
|
||||
)
|
||||
|
|
@ -145,12 +154,26 @@ def test_manual_request_reaches_every_pod_that_has_not_seen_it():
|
|||
|
||||
|
||||
def test_pod_reload_decision_ignores_persisted_last_run():
|
||||
"""A pod that has never reloaded must refresh its own copy even when another pod
|
||||
already stamped last_run_at within the interval"""
|
||||
"""A pod with stale data must refresh even when another pod already stamped
|
||||
last_run_at within the interval"""
|
||||
assert (
|
||||
pod_reload_is_due(
|
||||
schedule=ReloadSchedule(interval_hours=6, last_run_at=datetime(2024, 1, 1, 11, 59, tzinfo=timezone.utc)),
|
||||
pod_last_reload=None,
|
||||
pod_data_loaded_at=LAST_RUN,
|
||||
current_time=NOW,
|
||||
description="test",
|
||||
)
|
||||
is True
|
||||
)
|
||||
|
||||
|
||||
def test_schedule_that_never_ran_fires_immediately():
|
||||
"""A fresh schedule must not wait a full interval for its first run, even on a pod
|
||||
whose own data is boot-fresh"""
|
||||
assert (
|
||||
pod_reload_is_due(
|
||||
schedule=ReloadSchedule(interval_hours=6, last_run_at=None),
|
||||
pod_data_loaded_at=datetime(2024, 1, 1, 11, 59, tzinfo=timezone.utc),
|
||||
current_time=NOW,
|
||||
description="test",
|
||||
)
|
||||
|
|
|
|||
|
|
@ -3831,9 +3831,11 @@ class TestPriceDataReloadAPI:
|
|||
def test_get_model_cost_map_reload_status_admin_access(self, client_with_auth):
|
||||
"""
|
||||
Regression (LIT-4882): status is served purely from the DB row, so a restarted pod
|
||||
with no in-memory last-reload time still reports the real last/next run
|
||||
(whose in-memory clock only knows its own boot) still reports the real last/next run
|
||||
"""
|
||||
proxy_server_module.proxy_config.last_model_cost_map_reload = None
|
||||
proxy_server_module.proxy_config.model_cost_map_loaded_at = datetime(
|
||||
2030, 6, 1, tzinfo=timezone.utc
|
||||
)
|
||||
|
||||
with patch("litellm.proxy.proxy_server.prisma_client") as mock_prisma:
|
||||
mock_prisma.db.litellm_config.find_unique = AsyncMock(
|
||||
|
|
@ -3983,10 +3985,11 @@ class TestPriceDataReloadIntegration:
|
|||
mock_prisma.db.litellm_config.update_many = AsyncMock(return_value=None)
|
||||
mock_prisma.db.litellm_config.find_unique = AsyncMock(return_value=None)
|
||||
|
||||
boot_loaded_at = proxy_config.model_cost_map_loaded_at
|
||||
asyncio.run(proxy_config._check_and_reload_model_cost_map(mock_prisma))
|
||||
|
||||
mock_prisma.db.litellm_config.update_many.assert_not_called()
|
||||
assert proxy_config.last_model_cost_map_reload is None
|
||||
assert proxy_config.model_cost_map_loaded_at == boot_loaded_at
|
||||
|
||||
frozen_now = datetime(2024, 1, 1, 7, 0, tzinfo=timezone.utc)
|
||||
requested_at = datetime(2024, 1, 1, 6, 59, 30, tzinfo=timezone.utc)
|
||||
|
|
@ -3997,7 +4000,7 @@ class TestPriceDataReloadIntegration:
|
|||
last_run_at=requested_at,
|
||||
)
|
||||
)
|
||||
proxy_config.last_model_cost_map_reload = frozen_now - timedelta(minutes=1)
|
||||
proxy_config.model_cost_map_loaded_at = frozen_now - timedelta(minutes=1)
|
||||
|
||||
original_model_cost = litellm.model_cost.copy()
|
||||
try:
|
||||
|
|
@ -4016,7 +4019,7 @@ class TestPriceDataReloadIntegration:
|
|||
assert litellm.model_cost["gpt-3.5-turbo"] == {
|
||||
"input_cost_per_token": 0.001
|
||||
}
|
||||
assert proxy_config.last_model_cost_map_reload == frozen_now
|
||||
assert proxy_config.model_cost_map_loaded_at == frozen_now
|
||||
assert mock_prisma.db.litellm_config.update_many.call_args[1] == {
|
||||
"data": {"last_run_at": frozen_now},
|
||||
"where": {"param_name": "model_cost_map_reload_config"},
|
||||
|
|
@ -4044,8 +4047,8 @@ class TestPriceDataReloadIntegration:
|
|||
)
|
||||
)
|
||||
frozen_now = datetime(2024, 1, 1, 7, 0, tzinfo=timezone.utc)
|
||||
pod_last_reload = frozen_now - timedelta(minutes=1)
|
||||
proxy_config.last_model_cost_map_reload = pod_last_reload
|
||||
pod_data_loaded_at = frozen_now - timedelta(minutes=1)
|
||||
proxy_config.model_cost_map_loaded_at = pod_data_loaded_at
|
||||
|
||||
original_model_cost = litellm.model_cost.copy()
|
||||
try:
|
||||
|
|
@ -4059,15 +4062,16 @@ class TestPriceDataReloadIntegration:
|
|||
|
||||
mock_get_map.assert_not_called()
|
||||
mock_prisma.db.litellm_config.update_many.assert_not_called()
|
||||
assert proxy_config.last_model_cost_map_reload == pod_last_reload
|
||||
assert proxy_config.model_cost_map_loaded_at == pod_data_loaded_at
|
||||
finally:
|
||||
litellm.model_cost = original_model_cost
|
||||
_invalidate_model_cost_lowercase_map()
|
||||
|
||||
def test_periodic_reload_uses_pod_local_last_reload(self):
|
||||
def test_periodic_reload_uses_pod_local_data_age(self):
|
||||
"""
|
||||
Each pod decides from its own last reload, so a pod that never reloaded refreshes even
|
||||
when the shared row was just stamped, and stays put while inside the interval
|
||||
Each pod decides from the age of its own data, so a pod holding a stale copy
|
||||
refreshes even when the shared row was just stamped by another pod, and stays
|
||||
put while its copy is inside the interval
|
||||
"""
|
||||
from litellm.proxy.proxy_server import ProxyConfig
|
||||
|
||||
|
|
@ -4081,6 +4085,9 @@ class TestPriceDataReloadIntegration:
|
|||
)
|
||||
mock_prisma.db.litellm_config.update_many = AsyncMock(return_value=None)
|
||||
frozen_now = datetime(2024, 1, 1, 7, 0, tzinfo=timezone.utc)
|
||||
proxy_config.model_cost_map_loaded_at = datetime(
|
||||
2024, 1, 1, 0, 0, tzinfo=timezone.utc
|
||||
)
|
||||
|
||||
original_model_cost = litellm.model_cost.copy()
|
||||
try:
|
||||
|
|
@ -4095,20 +4102,20 @@ class TestPriceDataReloadIntegration:
|
|||
asyncio.run(proxy_config._check_and_reload_model_cost_map(mock_prisma))
|
||||
|
||||
assert litellm.model_cost["gpt-4-test"] == {"input_cost_per_token": 0.5}
|
||||
assert proxy_config.last_model_cost_map_reload == frozen_now
|
||||
assert proxy_config.model_cost_map_loaded_at == frozen_now
|
||||
assert mock_prisma.db.litellm_config.update_many.call_args[1][
|
||||
"data"
|
||||
] == {"last_run_at": frozen_now}
|
||||
|
||||
mock_get_map.reset_mock()
|
||||
mock_prisma.db.litellm_config.update_many.reset_mock()
|
||||
proxy_config.last_model_cost_map_reload = frozen_now - timedelta(hours=1)
|
||||
proxy_config.model_cost_map_loaded_at = frozen_now - timedelta(hours=1)
|
||||
|
||||
asyncio.run(proxy_config._check_and_reload_model_cost_map(mock_prisma))
|
||||
|
||||
mock_get_map.assert_not_called()
|
||||
mock_prisma.db.litellm_config.update_many.assert_not_called()
|
||||
assert proxy_config.last_model_cost_map_reload == frozen_now - timedelta(
|
||||
assert proxy_config.model_cost_map_loaded_at == frozen_now - timedelta(
|
||||
hours=1
|
||||
)
|
||||
finally:
|
||||
|
|
@ -4200,7 +4207,7 @@ class TestPriceDataReloadIntegration:
|
|||
"reload_requested_at": frozen_now,
|
||||
}
|
||||
assert (
|
||||
proxy_server_module.proxy_config.last_model_cost_map_reload
|
||||
proxy_server_module.proxy_config.model_cost_map_loaded_at
|
||||
== frozen_now
|
||||
)
|
||||
finally:
|
||||
|
|
@ -4232,7 +4239,7 @@ class TestPriceDataReloadIntegration:
|
|||
proxy_config._check_and_reload_anthropic_beta_headers(mock_prisma)
|
||||
)
|
||||
|
||||
assert proxy_config.last_anthropic_beta_headers_reload == frozen_now
|
||||
assert proxy_config.anthropic_beta_headers_loaded_at == frozen_now
|
||||
assert mock_prisma.db.litellm_config.update_many.call_args[1] == {
|
||||
"data": {"last_run_at": frozen_now},
|
||||
"where": {"param_name": "anthropic_beta_headers_reload_config"},
|
||||
|
|
@ -4280,7 +4287,7 @@ class TestPriceDataReloadIntegration:
|
|||
"reload_requested_at": frozen_now,
|
||||
}
|
||||
assert (
|
||||
proxy_server_module.proxy_config.last_anthropic_beta_headers_reload
|
||||
proxy_server_module.proxy_config.anthropic_beta_headers_loaded_at
|
||||
== frozen_now
|
||||
)
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue