Data Flow

End-to-end trace of how data moves through LogSys from source to dashboard.

---

Complete Data Flow

%%{init: {'theme':'default','themeVariables':{'primaryColor':'#1a73e8','secondaryColor':'#e8f0fe','tertiaryColor':'#f8f9fa','lineColor':'#5f6368','fontFamily':'Inter','fontSize':'13px'}}}%%
flowchart TD
  subgraph SRC["📥 Sources"]
    S1[GitLab Webhook<br/>POST /webhooks/gitlab]
    S2[File Upload<br/>POST /api/sources/ephemeral]
    S3[Jira Polling<br/>REST API v3]
    S4[Prometheus Poll<br/>Instant Query]
    S5[Azure DevOps<br/>WIQL Query]
  end

  subgraph ROUTE["🔀 Routing"]
    V1[Vector :8686<br/>ephemeral_http]
    V2[Vector :8687<br/>webhook_http]
    C1[Collectors :8080<br/>webhook_receiver]
  end

  subgraph KAFKA["📨 Kafka Topics"]
    K1[(raw_logs)]
    K2[(raw_metrics)]
    K3[(raw_traces)]
    K4[(raw_events)]
    K5[(canonical-events)]
    K6[(dlq)]
  end

  subgraph PIPE["⚙️ Pipeline Stages"]
    P1[Parse<br/>Drain3]
    P2[Normalize<br/>Pydantic + PII]
    P3[Dedup<br/>Redis Lua]
    P4[Features<br/>Polars]
    P5[Classify<br/>RoBERTa/Rule]
    P6[Store<br/>executemany INSERT]
  end

  subgraph STORE["💾 Storage"]
    PG[(PostgreSQL<br/>events table<br/>partitioned)]
    TP[(template_priority<br/>cache)]
    DLQ[(dlq_events)]
  end

  subgraph SERVE["🌐 API Layer"]
    API[FastAPI<br/>REST + WS]
    REDIS[(Redis<br/>KPI cache)]
    MV[OLAP Views<br/>mv_dashboard_aggregates<br/>mv_timeseries_hourly<br/>mv_top_templates]
  end

  subgraph UI["🎨 Frontend"]
    DASH[Dashboard<br/>KPIs, Charts]
    MON[Monitoring<br/>Log Table]
    INS[AI Insights<br/>RCA, Anomalies]
  end

  %% Flow
  S1 -->|HTTP| C1
  S2 -->|HTTP| V1
  S3 -->|HTTP| C1
  S4 -->|HTTP| C1
  S5 -->|HTTP| C1

  C1 -->|Produce| K4
  V1 -->|Produce| K5
  V2 -->|Produce| K5

  K1 -.->|Consume| PIPE
  K2 -.->|Consume| PIPE
  K3 -.->|Consume| PIPE
  K4 -.->|Consume| PIPE

  PIPE --> P1 --> P2 --> P3 --> P4 --> P5 --> P6
  P2 -.->|Reject| K6
  P3 -.->|Duplicate| SKIP[Skip]

  P6 -->|Write| PG
  P6 -->|Write| TP
  P2 -.->|DLQ| DLQ

  PG -.->|Refresh 30s| MV
  PG -.->|Read| API
  MV -.->|Read| API
  REDIS -.->|Cache| API
  API -.->|WS Push| UI
  UI -.->|Upload| V1

---

Stage-by-Stage Detail

1. Ingestion

Source	Protocol	Auth	Frequency	Output Topic
GitLab Webhook	HTTP POST	HMAC SHA-256	Event-driven	raw_events
File Upload	HTTP POST (multipart)	JWT Bearer	On-demand	canonical-events (via Vector)
Jira	REST GET	Basic + API Token	10 min	raw_events
Prometheus	HTTP GET (PromQL)	None	5 min	raw_metrics
Azure DevOps	REST POST (WIQL)	PAT	10 min	raw_events

Collector Implementation (services/collectors/): - Single process: APScheduler + FastAPI webhook receiver - State persisted in Redis (polling:cursor:{source}) - Retry with exponential backoff (tenacity) - Publish via shared Kafka client (acks=all, LZ4)

2. Vector HTTP Router

# services/vector/vector.yaml
sources:
  ephemeral_http:
    type: http
    address: "0.0.0.0:8686"
  webhook_http:
    type: http
    address: "0.0.0.0:8687"

transforms:
  ephemeral_enrich:
    type: remap
    inputs: [ephemeral_http]
  webhook_enrich:
    type: remap
    inputs: [webhook_http]

sinks:
  kafka:
    type: kafka
    inputs: [ephemeral_enrich, webhook_enrich]
    bootstrap_servers: ${KAFKA_BROKERS}
    topic: canonical-events
    encoding: json

• Port 8686: Ephemeral file uploads (from API)
• Port 8687: Generic webhooks
• Transforms via VRL (Vector Remap Language)
• Single Kafka sink (JSON encoding, no compression)

3. Kafka Topics

Topic	Partitions	Retention	Key	Purpose
raw_logs	1	1 day	source	Raw log lines from collectors
raw_metrics	1	1 day	source	Prometheus metric samples
raw_traces	1	1 day	trace_id	Distributed trace spans
raw_events	1	1 day	source	Jira, Azure, GitLab events
canonical-events	1	1 day	dedup_key	Normalized, ready for pipeline
dlq	1	30 days	—	Failed processing (never lose data)

Retention: Kafka is a transport buffer — durable storage is PostgreSQL (events partitions). 1-day retention (set in docker-compose.yml via KAFKA_LOG_RETENTION_HOURS=24 and per-topic retention.ms=86400000) is safe as long as the consumer keeps up; if the pipeline is down > 1 day, unconsumed messages in Kafka are purged.

Production: Increase partitions to match pipeline replicas for parallel consumption.

4. Pipeline Stages (Consumer Pipeline)

flowchart LR
  BATCH[Kafka Poll<br/>100 msgs / 0.5s] --> FILTER[Noise Filter<br/>is_noise()]
  FILTER --> PARSE[1. Parse<br/>Drain3]
  PARSE --> NORM[2. Normalize<br/>Pydantic + PII]
  NORM -->|Invalid| DLQ[DLQ]
  NORM --> DEDUP[3. Dedup<br/>Redis Lua]
  DEDUP -->|Duplicate| SKIP
  DEDUP --> FEAT[4. Features<br/>Polars]
  FEAT --> CLASS[5. Classify<br/>Cache → RoBERTa → Rule]
  CLASS --> STORE[6. Store<br/>executemany INSERT]
  STORE --> PG[(PostgreSQL)]
  STORE -->|Notify| REDIS[Redis Pub/Sub]

Stage Details

Stage	Input	Output	Key Logic	Perf
Noise Filter	Raw Kafka msg	Filtered msg	Drop heartbeats, health checks	O(1)
Parse (Drain3)	Message text	template_id, template_text	Online clustering, sim_th=0.4	~2K eps
Normalize	Raw + template	CanonicalEvent	PII mask, severity regex, Pydantic validate	~2.2K eps
Dedup	CanonicalEvent	Unique events	SHA-256(key) + Redis Lua INCR+EXPIRE	~2.1K eps
Features	Event	Enriched event	Polars vectorized: severity_rank, counts	~2.3K eps
Classify	Event + template	Priority P1–P4	Cache → RoBERTa → Rule fallback	~2K eps
Store	Enriched batch	PG rows	executemany INSERT, intra-batch merge	~2K eps

5. Storage & Indexing

Table: events (partitioned by month, PK (id, timestamp))

-- Key indexes
CREATE INDEX ix_events_timestamp ON events (timestamp DESC);
CREATE INDEX ix_events_source_sev_ts ON events (source, severity, timestamp DESC);
CREATE INDEX ix_events_priority_ts ON events (priority, timestamp DESC) WHERE priority IS NOT NULL;
CREATE INDEX ix_events_template_id ON events (template_id);
CREATE INDEX ix_events_message_trgm ON events USING GIN (message gin_trgm_ops);
-- Deduplication backstop (partition-aware)
CREATE UNIQUE INDEX ux_events_dedup_key ON events (dedup_key, timestamp) WHERE dedup_key IS NOT NULL;

Materialized Views (refreshed every 30s concurrently):

View	Purpose	Refresh
mv_dashboard_aggregates	KPI totals, rates, top errors	30s
mv_timeseries_hourly	Hourly buckets by severity/priority	30s
mv_top_templates	Top templates by volume	30s

6. API Layer

Query Patterns:

Endpoint	Query Strategy	Cache
GET /api/logs	Keyset pagination (cursor)	Redis 30s
GET /api/logs/meta/timeseries	MV mv_timeseries_hourly	Redis 30s
GET /api/kpis	MV mv_dashboard_aggregates	Redis 30s
GET /api/ai-insights/engine	On-demand compute	Redis 30s

Real-time: WebSocket /ws/kpis → Redis pub/sub → broadcast

---

Failure Paths

flowchart TD
  PROD[Producer] -->|Network fail| RETRY[Retry + Backoff]
  RETRY -->|Max retries| DLQ1[DLQ Topic]

  PIPE[Pipeline] -->|Parse fail| DLQ2[DLQ Topic + dlq_events table]
  PIPE -->|Normalize fail| DLQ2
  PIPE -->|Storage fail| RETRY2[Retry batch]
  RETRY2 -->|Persist fail| DLQ2

  API[API] -->|DB timeout| CACHE[Serve from Redis]
  API -->|Cache miss| ERROR[500 + alert]

  DLQ1 & DLQ2 --> REPLAY[Manual replay via admin UI]

---

Latency Budget (P99)

Segment	Budget	Typical
Source → Vector	10 ms	2 ms
Vector → Kafka	5 ms	1 ms
Kafka → Pipeline	50 ms	20 ms
Pipeline (all stages)	200 ms	80 ms
Pipeline → PostgreSQL	50 ms	15 ms
PostgreSQL → API	30 ms	10 ms
API → WebSocket	20 ms	5 ms
Total E2E	< 500 ms	~133 ms