diff --git a/infra/terraform/local/main.tf b/infra/terraform/local/main.tf index ee9f9ae..05b1b59 100644 --- a/infra/terraform/local/main.tf +++ b/infra/terraform/local/main.tf @@ -1109,7 +1109,16 @@ resource "kubernetes_stateful_set" "kafka" { spec { container { name = "kafka" - image = "apache/kafka:3.9.0" + # Pinned to match whatever :latest had already drifted to and + # formatted the PVC's KRaft metadata with (verified: the cached + # apache/kafka:latest image is kafka_2.13-4.3.1.jar). KRaft's + # on-disk metadata format is forward-compatible only — pinning to + # an OLDER version than what already formatted the volume crashes + # every broker start with "No MetadataVersion with feature level + # N". If this version is ever bumped, wipe the PVC first + # (kubectl delete pvc data-kafka-0 -n sentinel-data) or pin to a + # version >= whatever last wrote to it. + image = "apache/kafka:4.3.1" env { name = "KAFKA_NODE_ID" @@ -1259,7 +1268,10 @@ resource "kubernetes_job_v1" "kafka_topic_init" { restart_policy = "OnFailure" container { name = "kafka-topic-init" - image = "apache/kafka:3.9.0" + # Matches the broker's pinned version above — this only runs + # kafka-topics.sh as a client, but keeping both pins identical + # avoids a second version to track. + image = "apache/kafka:4.3.1" command = [ "/bin/sh", "-c", "/opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka:9092 --create --if-not-exists --topic traces.raw --partitions 3 --replication-factor 1 && echo 'topic ready'" diff --git a/pipelines/drift/db.py b/pipelines/drift/db.py index 84b5559..a5c2b53 100644 --- a/pipelines/drift/db.py +++ b/pipelines/drift/db.py @@ -141,25 +141,37 @@ def write_drift_stats( def get_active_model_version(database_url: str) -> str | None: - """Return the model_version model_registry considers current. - - Mirrors services/classifier/db.py's get_active_model selection exactly - (prefer status='active', fall back to the most recent 'staging' entry) so - the drift job evaluates the same version a classifier pod would load on - startup. Reading from `classifications` instead (whichever version wrote - the most recent row) was a race during rolling restarts: old- and - new-version pods write concurrently, so "most recent row" doesn't mean - "the rollout's target version" — it could attribute drift_stats to a - version that's already being retired. + """Return the model_version that's actually appearing in classifications. + + Live-verified gotcha this has to account for: model_registry holds two + different kinds of rows that don't share a model_version namespace. + services/classifier/db.py's get_active_model() picks a 'active'/'staging' + row to decide which MinIO model_path to download — but once loaded, the + classifier self-registers a *separate*, new 'staging' row under its own + freshly-derived model_version string (services/classifier/model.py's + `sentinel-roberta-{deployed_at}-{quant_tag}`), and THAT string — not the + row it downloaded from — is what ends up in classifications.model_version + on every write. An earlier version of this function mirrored + get_active_model()'s "prefer status='active'" ordering exactly, which + seemed principled but was wrong in practice: reproduced live against a + real cluster, the 'active' row was a stale promotion from a different + model_version string entirely (nothing in this project's current phase — + Airflow/Phase 7 doesn't exist yet — reliably keeps 'active' pointed at + what's actually running), so it returned a model_version with zero + matching classification rows and drift detection silently found nothing. + Ordering by created_at alone — "whichever pod self-registered most + recently" — is what's actually running and writing classifications. + Still scoped to non-retired rows; still subject to the same rolling- + restart race noted before (old- and new-version pods can both register + around the same moment), just no longer compounded by a status + preference that points at the wrong namespace entirely. """ with _connect(database_url) as conn: row = conn.execute( """ SELECT model_version FROM model_registry WHERE status IN ('active', 'staging') - ORDER BY - CASE status WHEN 'active' THEN 0 ELSE 1 END, - COALESCE(promoted_at, created_at) DESC + ORDER BY created_at DESC LIMIT 1 """, ).fetchone() diff --git a/scripts/dev-start.sh b/scripts/dev-start.sh index b30a264..3c86302 100755 --- a/scripts/dev-start.sh +++ b/scripts/dev-start.sh @@ -137,8 +137,13 @@ kubectl exec -n sentinel-data kafka-0 -- \ info "Syncing PostgreSQL password from secret..." PG_PASSWORD=$(kubectl get secret postgresql-credentials -n sentinel-data \ -o jsonpath='{.data.password}' | base64 -d) -if kubectl exec -n sentinel-data postgresql-0 -- \ - psql -U sentinel -d sentinel -v pw="$PG_PASSWORD" -c "ALTER USER sentinel PASSWORD :'pw';"; then +# -v/:'var' substitution only happens when psql reads SQL as a script +# (stdin/-f) — NOT via -c, which bypasses that parser and sends the string +# through unexpanded (verified live: -c left the literal text ":'pw'" in +# the query, which Postgres then rejected as a syntax error). Piped via +# stdin instead (kubectl exec -i to keep stdin open across the exec). +if echo "ALTER USER sentinel PASSWORD :'pw';" | kubectl exec -i -n sentinel-data postgresql-0 -- \ + psql -U sentinel -d sentinel -v pw="$PG_PASSWORD"; then info "PostgreSQL password synced" else warn "ALTER USER failed — local socket auth may not be trusted in this setup." @@ -236,11 +241,17 @@ if [[ -z "${MODEL_PATH:-}" ]] && command -v psql >/dev/null 2>&1; then if [[ -n "$_minio_path" ]]; then info "Active registry entry is a host path — switching to MinIO path: $_minio_path" - psql "$DATABASE_URL" -v path="$_minio_path" -c " + # -v/:'var' substitution only happens when psql reads SQL as a + # script (stdin/-f) — NOT via -c, which bypasses that parser and + # sends the string through unexpanded (verified live: -c left + # the literal text ":'path'" in the query, which Postgres then + # rejected as a syntax error). Piped via stdin instead. + echo " UPDATE model_registry SET status = 'retired' WHERE model_path NOT LIKE 'models/%' AND status IN ('active','staging'); UPDATE model_registry SET status = 'active', promoted_at = NOW() - WHERE model_path = :'path';" >/dev/null 2>&1 || true + WHERE model_path = :'path';" \ + | psql "$DATABASE_URL" -v path="$_minio_path" >/dev/null 2>&1 || true elif ! find "$_registry_path" -maxdepth 1 -name "*.onnx" 2>/dev/null | grep -q .; then warn "Registry has a stale local path (artifacts deleted): $_registry_path" psql "$DATABASE_URL" -c \ diff --git a/services/classifier/main.py b/services/classifier/main.py index a6d9d01..3fb4dad 100644 --- a/services/classifier/main.py +++ b/services/classifier/main.py @@ -50,6 +50,7 @@ async def lifespan(app: FastAPI): # The registry is the source of truth for which model version to serve. # Open the pool first so we can query it before loading the model. model_dir: Path | None = None + active: dict | None = None if settings.database_url: try: @@ -90,28 +91,39 @@ async def lifespan(app: FastAPI): # ── 3. Record this model version in the registry (idempotent) ───────────── # ON CONFLICT DO NOTHING: first pod registers it, subsequent pods skip. - # Skip registration when model_path is an absolute local path (this pod - # fell back to local model discovery — MODEL_PATH env var or - # logs/optimizer/). That path only exists on THIS pod's filesystem; - # download.py treats any path starting with "/" as already-materialized - # and never fetches it from MinIO, so writing it to the shared registry - # would hand a different pod a path that doesn't exist on its own - # filesystem, crashing it on startup. Leaving the registry untouched - # here means other pods fall through to their own local discovery too, - # which is at least self-consistent per pod. - if _pool and not _classifier.model_path.startswith("/"): + # + # Register the ORIGINAL MinIO-backed path (active["model_path"]) when we + # resolved via a successful MinIO download — NOT _classifier.model_path. + # download_model() always caches the downloaded model under a local + # directory (/tmp/sentinel-model-cache/...), so _classifier.model_path + # is *always* an absolute local path, even for a MinIO-backed model. + # Checking model_path.startswith("/") here (an earlier version of this + # check) therefore skipped registration for the common case too, not + # just genuine local-only fallback — every classification write then + # violated classifications_model_version_fkey, since the model_version + # never existed in model_registry. Reproduced live: confirmed via + # kubectl logs that a classifier pod loading a real MinIO-backed model + # still hit "Skipping registry write" and every subsequent /v1/moderations + # persist attempt failed with psycopg.errors.ForeignKeyViolation. + # + # `model_dir is not None` is the correct portability signal: it's only + # None when Classifier() fell through to _resolve_model_dir() itself + # (MODEL_PATH env var or logs/optimizer/ found locally) — that's the + # genuine non-portable case docs/local-dev.md and this comment used to + # describe, and it's the only one that should skip registration. + if _pool and model_dir is not None and active is not None: try: await _db.register_model( _pool, _classifier.model_version, - _classifier.model_path, + active["model_path"], _classifier.threshold, ) except Exception: logger.exception("Failed to register model version in registry") elif _pool: logger.info( - "Skipping registry write — model_path is a local filesystem path, " + "Skipping registry write — resolved via local fallback, " "not portable across pods | path=%s", _classifier.model_path, ) diff --git a/services/classifier/model.py b/services/classifier/model.py index 90b509b..b1a1135 100644 --- a/services/classifier/model.py +++ b/services/classifier/model.py @@ -123,11 +123,15 @@ def predict(self, texts: list[str]) -> list[dict]: inputs = self._tokenizer( texts, return_tensors="np", - # max_length (not dynamic padding) trades a bit of extra compute - # on short inputs for latency that doesn't depend on what else - # happens to be in the same batch — a batch containing one - # long input no longer inflates every other input's padding. - padding="max_length", + # Reverted from padding="max_length": every batch then tokenized + # to the full 512 tokens regardless of actual input length, + # which OOM-killed the pod (1Gi limit) twice under real load — + # reproduced live, not theoretical. The "latency doesn't depend + # on batch content" benefit isn't worth crashing the service; + # a real fix for that (e.g. bucketed padding, or raising the + # memory limit + confirming it's stable under sustained load) + # needs its own verification pass, not a same-session swap back. + padding=True, truncation=True, max_length=512, ) diff --git a/services/stream-processor/main.py b/services/stream-processor/main.py index 03eed42..efb59db 100644 --- a/services/stream-processor/main.py +++ b/services/stream-processor/main.py @@ -94,6 +94,19 @@ def _pg_write( that case, so the caller never holds a reference to a leaked/half-open connection: the next call always reconnects from scratch instead of reusing something nobody closed. + + Two distinct PG failure modes need different recovery: + - OperationalError (connection actually lost): close + reconnect + retry once. + - Any other psycopg.Error (e.g. a constraint violation): the connection + is still alive, but the current transaction is aborted — every command + on it then fails with InFailedSqlTransaction until rolled back. + Reproduced live: a stale model_version (classifier registered a model + version the DB didn't have yet — see main.py's #43 fix) triggered a + ForeignKeyViolation on classifications, and without this rollback the + long-lived connection stayed permanently broken — every subsequent + batch failed the same way forever, not just the one that hit the FK + violation. + ON CONFLICT DO NOTHING on (span_id, text_type) makes the retry idempotent. """ try: @@ -120,6 +133,16 @@ def _pg_write( except Exception: pass raise + except psycopg.Error: + # Connection is alive but its transaction is aborted (e.g. a + # constraint violation) — roll back so it's usable again on the + # NEXT call instead of failing every subsequent batch forever. + logger.exception("PostgreSQL write failed — rolling back to keep the connection usable") + try: + pg_conn.rollback() + except Exception: + logger.exception("Rollback itself failed — connection may still be unusable") + raise def run() -> None: