DSPM classifier · async python · kafka · kubernetes
A Python microservice that consumes object-storage events from Kafka, classifies content for sensitive data (PII, secrets, financial, health) through a tiered regex → NER → LLM pipeline, persists findings to Postgres, and emits to a downstream Kafka topic. It ships as two Kubernetes deployments built from one image so reads and writes scale independently.
The service exercises a demanding production skill set: async Python, Kafka at scale, Kubernetes microservices, AWS-compatible storage, and Gen-AI integration. The domain mirrors a real DSPM (data security posture management) product surface.
S3 events flow through Kafka into the worker pool, which fetches objects, runs the tiered pipeline, persists findings to Postgres, and emits to the findings topic. The api reads from Postgres only. Retry and DLQ topics handle transient and permanent failures so the live path stays clean.
Partition key is tenant_id, so all of one tenant's events stay ordered and isolated to a single partition. The wire guarantee is at-least-once; the idempotency table makes it effectively-once end-to-end.
Three tiers, ordered cheapest first. Each stage emits a confidence score; the pipeline decides whether to escalate. Merge and dedupe happen after the final tier so overlapping spans collapse into one finding.
Deterministic, high-precision pass for known-shape secrets and identifiers.
Detects
Escalates when
Never escalates further on its own — always passes through to NER so it can contribute entities to the merged span set.
Example
Card 4111 1111 1111 1111, key AKIAIOSFODNN7EXAMPLE
Presidio + spaCy named-entity recognition for PII the regex tier cannot model.
Detects
Escalates when
Any returned entity score falls below NER_CONFIDENCE_THRESHOLD (default 0.6), or the sensitive-document heuristic fires while regex/NER were silent.
Example
Contact Alice at alice@example.com or 415-555-1212.
Tiered LLM pass invoked only on ambiguous content, with structured JSON output.
Detects
Escalates when
Final tier — never escalates further. On parse failure or 5xx after retries, pipeline persists the finding with llm_failed=true using regex + NER results only.
Example
"Re: signed NDA — please share the new contract on the shared drive."
scripts/classify_one.py wires the engine for a manual end-to-end test against a single S3 object.
$ python scripts/classify_one.py s3://acme-uploads/contracts/q2.pdf
[regex] matched 1 (FINANCIAL.CREDIT_CARD) t=2.1ms
[ner] matched 3 (PII.PERSON, PII.EMAIL, PII.PHONE) conf=0.92 t=87ms
[llm] skipped (NER confidence above threshold)
─────────────────────────────────────────────
finding tenant=acme-corp event=evt_01HZABCXYZ sensitivity=HIGH
categories=[PII, FINANCIAL] match_count=4
pipeline_version=1 llm_failed=false
persisted to findings(tenant_id=acme-corp, event_id=evt_01HZABCXYZ)The failure matrix classifies every per-stage error so the worker knows what to do without inventing policy at runtime.
| Failure | Class | Action |
|---|---|---|
| Idempotency check shows already-processed | expected | Skip, commit offset, increment duplicate_total. |
| S3 fetch: 404 / NoSuchKey | permanent | Emit object_missing finding, persist, commit. |
| S3 fetch: 5xx / network | transient | Send to retry topic with attempt counter. |
| Object > MAX_OBJECT_BYTES | permanent | Persist too_large finding, no classification, commit. |
| Regex / NER pass raises | unexpected | DLQ with traceback; do not retry (treat as a code bug). |
| LLM: timeout / 5xx / 429 | transient | Send to retry topic. |
| LLM: malformed JSON after N parse retries | permanent | Persist finding with llm_failed=true using regex + NER results; commit. |
| Postgres: connection error | transient | Back off; do not commit offset; partition pauses naturally via backpressure. |
| Kafka produce to findings fails | transient | Retry the produce; do not commit consume offset. |
processed_messages(event_id PK, …, pipeline_version) is inserted in the same transaction as the findings upsert. A duplicate event is a PK violation that rolls the transaction back cleanly — the offset still commits. Bumping pipeline_version is how an intentional re-classification pass reprocesses everything.
BoundedWorkPool owns an asyncio.Semaphore(N) plus the set of paused TopicPartitions. It pauses at the high-water mark and resumes at the low-water mark, so a slow LLM tier never unbounded-queues messages. The consumer loop only asks “can I dispatch?” — it doesn't know how backpressure works.
One Docker image, two Deployments. The worker stays at or below the partition count; the api scales on HTTP load. Both pods share ConfigMap and sealed Secret; migrations run as a pre-deploy Job.
The AWS positioning is intentionally cheap: aioboto3 targets MinIO locally and real AWS S3 in deploy via the S3_ENDPOINT_URL env var. Same SDK, same IAM patterns, zero code change. Real cloud bills only show up if and when this actually deploys.
Each metric maps to an alert intent. Cardinality is bounded — labels are tenant_id, sensitivity, stage, and partition, not anything user-supplied.
findings_totalMeaning · Counter of classifications produced, partitioned per tenant and sensitivity.
Alert · Volume sanity — sudden drops mean the pipeline is starved or stuck.
pipeline_stage_latency_secondsMeaning · Per-tier latency histogram (regex / ner / llm).
Alert · LLM-tier slow path: alert on p95 sustained above SLO.
kafka_consumer_lag_secondsMeaning · End-to-end backlog measured from producer timestamp to consumer commit.
Alert · Feeds the HPA; pages oncall above a hard ceiling.
in_flight_workMeaning · Gauge of concurrent classifications currently held by the work pool.
Alert · Backpressure visibility — sustained pinning means the LLM tier is the bottleneck.
partitions_pausedMeaning · Gauge of TopicPartitions currently paused by the BoundedWorkPool.
Alert · Sustained pause across partitions = consumer is overwhelmed, scale or shed load.
dlq_totalMeaning · Counter of permanent failures sent to the dead-letter topic.
Alert · Alertable on any non-zero rate; reason label tells you which path failed.
duplicate_totalMeaning · Counter of events rejected by the idempotency table.
Alert · Replay health — sustained increase signals upstream is replaying without intent.
Async + perf
Backpressure via BoundedWorkPool (semaphore + partition pause/resume), tiered classifier so the cheap path stays cheap, asyncio throughout.
Kafka at scale
Tenant-keyed partitioning, consumer group, manual commits + idempotency table for effectively-once, retry topic with exponential backoff, DLQ, graceful shutdown without redelivery storms.
K8s / microservices
Shared-image two-deployment topology, HPA on consumer lag via prometheus-adapter, sealed-secret pattern, Alembic migration Job, ServiceMonitor for scrape.
AWS
aioboto3 against real S3 in deploy and MinIO locally; same SDK, standard credential chain, IAM patterns unchanged.
Gen-AI
Structured-output LLM pass with Pydantic schema, parse/retry loop, cost-aware tiering, model swap by env var.
DSPM domain
PII / SECRETS / FINANCIAL / HEALTH categories, sensitivity levels, tenant-scoped findings, pipeline_version for re-classification passes.
Production rigor
Idempotency, DLQ, structured logs with bound contextvars, Prometheus metrics mapped to alerts, graceful shutdown respecting the K8s lifecycle.