Skip to content

Canonical Event

The CanonicalEvent is the single schema contract shared by every service in the pipeline. All sources (Jira, Prometheus, Azure DevOps, webhooks, uploaded files) are normalized into this shape before entering Kafka.

Why a canonical shape?

Problem Solution
Each source emits a different format Single Pydantic model in services/libs/common
New connectors are expensive to add Connectors only map → canonical, pipeline is untouched
Breaking schema changes break consumers schema_version field + backward-compatible validation

Structure

# services/libs/common/schemas/canonical_event.py
class Environment(str, Enum):
    PROD = "prod"; PRODUCTION = "production"
    STAGING = "staging"; DEV = "dev"

class EventType(str, Enum):
    LOG = "log"; METRIC = "metric"
    TRACE = "trace"; EVENT = "event"

class Severity(str, Enum):
    DEBUG = "DEBUG"; INFO = "INFO"; WARNING = "WARNING"
    ERROR = "ERROR"; CRITICAL = "CRITICAL"

class CanonicalEvent(BaseModel):
    schema_version: int = 1
    timestamp: datetime
    source: str = Field(..., min_length=1)
    environment: Environment
    type: EventType
    severity: Severity
    message: str = Field(..., min_length=1)
    summary: str | None = None
    template_id: int | None = None   # filled after Drain3
    host: str | None = None
    pod_name: str | None = None
    trace_id: str | None = None
    span_id: str | None = None
    occurrence_count: int = Field(default=1, ge=1)
    tags: dict = Field(default_factory=dict)
    dedup_key: str | None = None
    # model_config: use_enum_values=True, extra="forbid"

Enrichment fields (severity_rank, is_error_or_worse, priority, priority_confidence, priority_source, model_version) are not part of the canonical contract — they are added by downstream pipeline stages before storage.

Field Conventions

Field Convention
timestamp ISO 8601 UTC; source timezone converted to UTC at ingest
severity Uppercase enum; mapped from source-specific levels
type One of log / metric / trace / event
message PII-stripped before any downstream processing
tags Free-form JSONB; reserved keys prefixed with correction_
dedup_key Populated by the Deduplicator
template_id int, filled after Drain3 parsing

Lifecycle

flowchart LR
  RAW[Raw event<br/>source format] --> MAP[Connector map]
  MAP --> NORM[Normalizer<br/>Pydantic validation + PII]
  NORM --> CANON[(CanonicalEvent)]
  CANON --> KAFKA[Kafka topic]
  KAFKA --> PIPE[Pipeline stages]
  PIPE --> PG[(events table)]
  1. Ingest — connector maps raw fields to canonical fields
  2. Validate — CanonicalEvent model rejects malformed events → DLQ
  3. Normalize — PII stripping, severity mapping, defaults applied
  4. Enrich — dedup key, template_id, priority added by later stages
  5. Persist — stored as a row in the events table

Example

{ "alert": "HighLatency", "labels": { "host": "api-01" }, "value": 3500 }

{
  "schema_version": 1,
  "timestamp": "2026-08-11T09:12:44Z",
  "source": "prometheus-prod",
  "environment": "production",
  "type": "metric",
  "severity": "ERROR",
  "message": "High latency detected on host api-01",
  "host": "api-01",
  "occurrence_count": 1,
  "tags": { "alert_name": "HighLatency" }
}

Validation Failures

Events that fail validation are routed to the DLQ (Kafka dlq topic + dlq_events table) with a reason explaining the failure — they are never silently dropped.