diff --git a/ops/lib/recommend_instance/bedrock.py b/ops/lib/recommend_instance/bedrock.py index 64b2fce..44a8e19 100644 --- a/ops/lib/recommend_instance/bedrock.py +++ b/ops/lib/recommend_instance/bedrock.py @@ -213,6 +213,33 @@ def _matches(fm: dict) -> bool: # Pricing (AWS Price List API — best-effort) # # --------------------------------------------------------------------------- # +# Max plausible USD per token. Even the priciest frontier models sit well under +# $100 / 1M tokens (1e-4/token); anything above this is a unit-scale error (e.g. a +# per-1M dimension mis-scaled as per-1K), so reject it rather than bake a value +# ~1000x too high into the committed CR. +_MAX_PLAUSIBLE_PER_TOKEN = 1e-3 + + +def _tokens_per_unit(unit: str, desc: str) -> int | None: + """Tokens represented by one Price List unit (1_000 or 1_000_000), read from + the dimension's unit/description — or None when it's ambiguous. + + Structured, not a guess: AWS increasingly prices per 1,000,000 tokens with a + bare unit like 'tokens', so defaulting an unlabeled unit to per-1K makes a + per-1M price 1000x too high. Return a scale ONLY when the text clearly says + 1K or 1M; otherwise None, so the caller skips the dimension (and falls back to + --input-cost/--output-cost or LiteLLM's map) instead of recording a wrong price. + """ + t = f"{unit} {desc}".lower().replace(",", "").replace(" ", "") + # Check 1M before 1K: "1000000" contains "1000". Match the numeric magnitude + # (survives words between the number and "tokens", e.g. "per 1000 input tokens"). + if "1000000" in t or "1mtoken" in t or "per1m" in t or "/1m" in t or "million" in t: + return 1_000_000 + if "1000" in t or "1ktoken" in t or "per1k" in t or "/1k" in t: + return 1_000 + return None + + def token_prices(region: str, fm: dict) -> tuple[float | None, float | None]: """Best-effort (input, output) USD-per-token from the AWS Price List API. @@ -257,9 +284,12 @@ def token_prices(region: str, fm: dict) -> tuple[float | None, float | None]: per_unit = float(usd) if per_unit <= 0: continue - # Bedrock token dims are typically per 1,000 tokens. - divisor = 1_000_000.0 if ("1m" in unit or "million" in desc) else 1_000.0 - per_token = per_unit / divisor + scale = _tokens_per_unit(unit, desc) + if scale is None: + continue # ambiguous unit — don't guess (would risk a 1000x error) + per_token = per_unit / scale + if per_token > _MAX_PLAUSIBLE_PER_TOKEN: + continue # unit-scale sanity clamp if "input" in desc and in_price is None: in_price = per_token elif "output" in desc and out_price is None: diff --git a/platform/services/litellm-sync/scripts/litellm_sync.py b/platform/services/litellm-sync/scripts/litellm_sync.py index 2eebd5b..47d97a9 100644 --- a/platform/services/litellm-sync/scripts/litellm_sync.py +++ b/platform/services/litellm-sync/scripts/litellm_sync.py @@ -149,30 +149,66 @@ def _litellm_request(method: str, path: str, body: dict | None = None) -> dict | return None -def list_db_model_ids() -> dict[str, list[str]] | None: - """Return {model_name: [model_id, ...]} for DB-registered models only. - - Unlike a name->id map, this preserves EVERY DB row for a name, so duplicate - registrations of the same model (multiple rows sharing one model_name) are - visible and can be cleaned up. Static config-file models (db_model == False) - are excluded so they can never be selected for deletion. Returns None if - LiteLLM is unreachable. +def list_db_model_ids() -> dict[str, list[tuple[str, dict]]] | None: + """Return {model_name: [(model_id, litellm_params), ...]} for DB-registered + models only. + + Preserves EVERY DB row for a name (so duplicate registrations are visible and + can be collapsed) and carries each row's live litellm_params (so register_model + can detect when a CR's params have drifted from what's registered). Static + config-file models (db_model == False) are excluded so they can never be + selected for deletion. Returns None if LiteLLM is unreachable. """ info = _litellm_request("GET", "/model/info") if info is None: return None - result: dict[str, list[str]] = {} + result: dict[str, list[tuple[str, dict]]] = {} for entry in info.get("data", []) or []: model_info = entry.get("model_info") or {} if not model_info.get("db_model"): continue name = entry.get("model_name") model_id = model_info.get("id") + params = entry.get("litellm_params") or {} if name and model_id: - result.setdefault(name, []).append(model_id) + result.setdefault(name, []).append((model_id, params)) return result +# litellm_params keys this controller sets and can meaningfully diff. api_key is +# deliberately excluded: LiteLLM redacts it in /model/info, so comparing it would +# always report drift and churn. +_DRIFT_KEYS = ("model", "api_base", "aws_bedrock_runtime_endpoint", + "input_cost_per_token", "output_cost_per_token") + + +def _params_drifted(current: dict, desired: dict) -> bool: + """True when a param the CR now specifies differs from what LiteLLM has. + + Conservative by design so it can never churn: a key that LiteLLM does NOT + echo back in `current` is treated as "can't compare -> not drift" (so if a + given LiteLLM build omits e.g. cost fields from /model/info, that edit simply + isn't hot-applied — the pre-existing documented behavior — rather than + re-registering every reconcile). Only a key present in BOTH and differing + counts. Costs compared with a small relative tolerance; other keys exact. + """ + current = current or {} + for k in _DRIFT_KEYS: + if k not in desired or k not in current: + continue + dv, cv = desired[k], current[k] + if k in ("input_cost_per_token", "output_cost_per_token"): + try: + if abs(float(cv) - float(dv)) > abs(float(dv)) * 1e-6 + 1e-15: + return True + except (TypeError, ValueError): + if str(cv) != str(dv): + return True + elif cv != dv: + return True + return False + + def register_model(name: str, litellm_params: dict) -> bool: """Ensure `name` is registered exactly once in LiteLLM. Duplicate-safe. @@ -180,31 +216,48 @@ def register_model(name: str, litellm_params: dict) -> bool: for the serving tiers, or a native bedrock/ upstream for BedrockModels). Held under REGISTRY_LOCK so a concurrent watch + reconcile can't both add the - same model. Crucially, this is *monotonic*: it only ADDS when there is no DB - row for the name, and otherwise only REMOVES extra rows (keep the first, - delete the rest). It never adds while a row already exists. + same model. It is *near-monotonic*: with unchanged params it only ADDS when + there is no DB row for the name and otherwise only REMOVES extra rows (keep + the first, delete the rest); the sole exception is a real params change, which + re-registers the one kept row (see "Param drift" below). That monotonicity is what makes it robust against LiteLLM's eventually- consistent /model/info: an earlier delete-all-then-add design could act on a stale read (see fewer rows than really exist, delete those, add one) and so - *grow* duplicates under startup churn. Because this version never adds when a - row is present, repeated passes can only shrink the row count — so it always - converges to exactly one, even if reads lag writes. - - NOTE: because it never re-adds an existing row, it does NOT push param changes - (e.g. a refreshed Bedrock price) onto an already-registered model — undeploy + - redeploy the CR to apply changed litellm_params. + *grow* duplicates under startup churn. Because this version never adds while a + row is present EXCEPT to apply a real params change, repeated passes with + unchanged params can only shrink the row count — so it always converges to + exactly one. + + Param drift: when a row already exists but the CR's litellm_params have + changed (e.g. an edited Bedrock price/endpoint), we re-register (delete the + kept row + add with the new params) under the lock so LiteLLM reflects the + edit. Drift is detected conservatively (see _params_drifted) — it can only + trigger on a genuine change, never on steady state — so this does not + reintroduce churn. """ with REGISTRY_LOCK: db_models = list_db_model_ids() if db_models is None: return False - ids = db_models.get(name, []) - if ids: - for extra_id in ids[1:]: + rows = db_models.get(name, []) + if rows: + keep_id, keep_params = rows[0] + for extra_id, _ in rows[1:]: _litellm_request("POST", "/model/delete", {"id": extra_id}) - if len(ids) > 1: - log.info("register %s: removed %d duplicate DB row(s)", name, len(ids) - 1) + if len(rows) > 1: + log.info("register %s: removed %d duplicate DB row(s)", name, len(rows) - 1) + if _params_drifted(keep_params, litellm_params): + # CR spec changed — re-register so the edit takes effect. Rare (an + # apply), so delete+add here can't churn on steady state. + _litellm_request("POST", "/model/delete", {"id": keep_id}) + resp = _litellm_request("POST", "/model/new", { + "model_name": name, + "litellm_params": litellm_params, + }) + if resp is None: + return False + log.info("re-registered model %s — CR litellm_params changed", name) return True resp = _litellm_request("POST", "/model/new", { "model_name": name, @@ -228,16 +281,16 @@ def deregister_model(name: str) -> bool: if db_models is None: log.warning("deregister %s: LiteLLM unreachable — will retry on reconcile", name) return False - ids = db_models.get(name, []) - if not ids: + rows = db_models.get(name, []) + if not rows: log.info("deregister %s: not a DB-registered model (already gone or static) — skipping", name) return True ok = True - for model_id in ids: + for model_id, _ in rows: if _litellm_request("POST", "/model/delete", {"id": model_id}) is None: ok = False if ok: - log.info("deregistered model %s (%d row(s)) from LiteLLM", name, len(ids)) + log.info("deregistered model %s (%d row(s)) from LiteLLM", name, len(rows)) return ok @@ -442,7 +495,9 @@ def reconcile_loop() -> None: # --------------------------------------------------------------------------- def _health_server() -> None: - """Tiny :8080 server — 200 when the K8s API is reachable, else 503.""" + """Tiny :8080 server — 200 when the K8s API is reachable and listable (or the + watched CRD is simply absent: 404); 503 on RBAC denial (403/401) or transport + failure, so a controller that can reconcile nothing never reports healthy.""" import http.server class Handler(http.server.BaseHTTPRequestHandler): @@ -451,13 +506,20 @@ def do_GET(self) -> None: # noqa: N802 client.CustomObjectsApi().list_cluster_custom_object( group=SERVING_GROUP, version=CR_VERSION, plural="vllmendpoints", limit=1, ) - except ApiException: - # API server reachable, but the CRD may be absent (a Bedrock-only - # / kro=false install has no vllmendpoints CRD -> 404) or the list - # may be RBAC-scoped (403). Either way the API is up and the - # watch/reconcile loops handle missing CRDs by backoff, so we're - # healthy. Only a transport failure (below) is unhealthy. - pass + except ApiException as e: + # ONLY 404 is benign: the CRD isn't installed (a Bedrock-only / + # kro=false install has no vllmendpoints CRD), and the watch/ + # reconcile loops back off until it appears. Any OTHER status is a + # real fault the probe MUST surface — in particular 403/401 means + # the ServiceAccount can't list the CRs it exists to reconcile + # (broken/rolled-back RBAC), so the controller registers nothing; + # reporting healthy there would hide a dead controller behind a + # green probe. + if e.status != 404: + self.send_response(503) + self.end_headers() + self.wfile.write(f"kubernetes API error {e.status}: {e.reason}".encode()) + return except Exception as e: # noqa: BLE001 self.send_response(503) self.end_headers() diff --git a/tests/test_bedrock_pricing.py b/tests/test_bedrock_pricing.py new file mode 100644 index 0000000..ddb77f6 --- /dev/null +++ b/tests/test_bedrock_pricing.py @@ -0,0 +1,51 @@ +"""Tests for the Bedrock pricing unit-scale resolver and id/partition helpers. + +The pricing path must never GUESS per-1K vs per-1M from free text: a per-1M +dimension with a bare 'tokens' unit, scaled as per-1K, records a per-token cost +1000x too high into the committed CR. `_tokens_per_unit` returns None on an +ambiguous unit so the caller falls back rather than recording a wrong price. +""" + +import pytest + +from recommend_instance import bedrock + + +class TestTokensPerUnit: + @pytest.mark.parametrize("unit,desc,expected", [ + ("1K tokens", "", 1_000), + ("1M tokens", "", 1_000_000), + ("tokens", "Price per 1000 input tokens", 1_000), + ("tokens", "USD per 1,000,000 output tokens", 1_000_000), + ("", "per 1M tokens", 1_000_000), + # Ambiguous — must be None (do NOT default to per-1K). + ("tokens", "Input tokens", None), + ("tokens", "", None), + ("", "", None), + ]) + def test_scale(self, unit, desc, expected): + assert bedrock._tokens_per_unit(unit, desc) == expected + + def test_ambiguous_never_defaults_to_1k(self): + # The exact regression: a bare 'tokens' unit must not be assumed per-1K. + assert bedrock._tokens_per_unit("tokens", "Input tokens") is None + + def test_clamp_threshold_is_sane(self): + # $100/1M = 1e-4/token is plausible (below clamp); $1000/1M = 1e-3 is the + # cutoff for a unit-scale error. + assert 1e-4 < bedrock._MAX_PLAUSIBLE_PER_TOKEN <= 1e-3 + + +class TestBedrockHelpers: + def test_default_alias(self): + assert bedrock.default_alias("amazon.nova-lite-v1:0") == "nova-lite" + + def test_partition_and_endpoint(self): + assert bedrock.partition_for("eu-central-1") == "aws" + assert bedrock.partition_for("eusc-de-east-1") == "aws-eusc" + assert bedrock.runtime_endpoint("eusc-de-east-1", "aws-eusc").endswith("amazonaws.eu") + + def test_region_geo(self): + assert bedrock._region_geo("us-east-1") == "us" + assert bedrock._region_geo("eu-central-1") == "eu" + assert bedrock._region_geo("ap-south-1") == "apac"