Normalizer

The Normalizer validates, enriches, and sanitizes raw events into strict CanonicalEvent objects. It runs after parsing and before deduplication.

---

Responsibilities

1. PII Masking — Remove sensitive data from messages
2. Severity Detection — Infer severity from message content
3. Schema Validation — Enforce CanonicalEvent contract (Pydantic v2)
4. Rejection — Invalid events → DLQ (never silently dropped)

---

Processing Flow

flowchart TD
  RAW[Raw Event + Template] --> PII[PII Masking]
  PII --> SEV[Severity Detection]
  SEV --> VAL[Pydantic Validation]
  VAL -->|Valid| CANON[CanonicalEvent]
  VAL -->|Invalid| DLQ[DLQ + dlq_events table]

---

1. PII Masking

Applied to message field before any other processing.

Pattern	Regex	Replacement
Email	\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z]{2,}\b	<EMAIL>
IPv4	\b(?:\d{1,3}\.){3}\d{1,3}\b	<IP>
IPv6	(?:[0-9a-fA-F]{1,4}:){7}[0-9a-fA-F]{1,4}	<IP>
Token	(Bearer|Token|API_KEY|api_key)\s*[=:]\s*[\w\-]{20,}	<TOKEN>
UUID	\b[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\b	<UUID>
Credit Card	\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b	<CARD>

# normalizer.py - mask_pii()
def mask_pii(text: str) -> str:
    text = EMAIL_RE.sub("<EMAIL>", text)
    text = IPV4_RE.sub("<IP>", text)
    text = TOKEN_RE.sub("<TOKEN>", text)
    # ...
    return text

---

2. Severity Detection

If source doesn't provide severity, infer from message:

# Priority order (first match wins)
SEVERITY_PATTERNS = [
    (CRITICAL, [r"\b(fatal|panic|emergency|crash|abort|segfault)\b"]),
    (ERROR,    [r"\b(error|exception|fail|failed|timeout|denied|refused)\b"]),
    (WARNING,  [r"\b(warn|warning|retry|degraded|slow|latency)\b"]),
    (INFO,     [r"\b(info|started|stopped|completed|success)\b"]),
    (DEBUG,    [r"\b(debug|trace|verbose)\b"]),
]

Trusted Sources: If source in PIPELINE_TRUSTED_SOURCES, use source-provided severity.

---

3. CanonicalEvent Schema (Pydantic v2)

# services/libs/common/schemas/canonical_event.py
class CanonicalEvent(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)

    schema_version: str = "1.0"
    timestamp: datetime  # UTC enforced
    source: str
    message: str  # non-empty, stripped
    summary: str | None = None
    template_id: str | None = None
    environment: Environment = Environment.prod
    type: EventType = EventType.log
    severity: Severity
    host: str | None = None
    pod_name: str | None = None
    trace_id: str | None = None
    span_id: str | None = None
    occurrence_count: int = 1
    tags: dict[str, str] = {}
    dedup_key: str | None = None  # SHA-256 hash

    # Enriched by pipeline
    severity_rank: int | None = None
    is_error_or_worse: bool | None = None
    priority: Priority | None = None
    priority_confidence: float | None = None
    priority_source: PrioritySource | None = None
    model_version: str | None = None

Validation Rules: - extra="forbid" — No unknown fields allowed - timestamp → converted to UTC - message → stripped, min_length=1 - dedup_key → SHA-256 of message|type|template_id|host|source

---

4. Rejection → DLQ

Any validation error creates a NormalizationError:

class NormalizationError(Exception):
    def __init__(self, reason: str, raw_payload: dict):
        self.reason = reason
        self.raw_payload = raw_payload

DLQ Output (Kafka dlq topic + PostgreSQL dlq_events):

{
  "reason": "validation_error: message field is required",
  "raw_payload": { ...original message... },
  "received_at": "2026-01-15T10:30:00Z"
}

---

Trusted Sources

Configure in .env:

PIPELINE_TRUSTED_SOURCES=prometheus,vector

Sources in this list keep their original severity (no regex override).

---

Troubleshooting

Symptom	Check
High DLQ rate	Inspect dlq_events table for reason patterns
Severity wrong	Check regex patterns; add source to trusted
PII not masked	Verify regex matches your log format
Validation error	Log shows field + expected type

# View DLQ
docker compose exec kafka-broker \
  kafka-console-consumer --bootstrap-server localhost:9092 \
  --topic dlq --from-beginning

# Query DLQ table
docker compose exec postgres psql -U aiops -d aiops -c \
  "SELECT reason, received_at FROM dlq_events ORDER BY received_at DESC LIMIT 10;"