Designing a Scalable CTI Stealer Logs Pipeline: Handling High-Volume Security Logs with FastAPI, Kafka, and Open Table Formats

In Cyber Threat Intelligence (CTI) operations, handling incoming data streams is a massive data engineering challenge. When analyzing massive streams of unstructured security logs or data feeds, security engineering teams face a critical bottleneck: how to ingest millions of unstructured payloads, validate them against threat intelligence tools, deduplicate them, and index them for rapid search without crashing the system.

If you try to process these payloads synchronously, your API will choke. If you don’t deduplicate early, your storage costs will skyrocket.

Below is an architectural breakdown of an enterprise-grade, decoupled ingestion pipeline designed specifically for high-throughput security log analysis using a modern stack: FastAPI, Telethon, Apache Kafka, Elasticsearch, PostgreSQL, and Iceberg.

The High-Level Architecture

To maintain performance, a production-grade pipeline must separate Ingestion from Heavy Processing.

Step 1: Lightweight Ingestion Layer (FastAPI & Telethon)

The first line of defense is ensuring your collection nodes never do heavy lifting. Whether you are using Telethon to stream events asynchronously from chat channels or exposed API gateways, the ingestion layer should have exactly one job: accept the data, write it to a queue, and return a 202 Accepted status.

Using FastAPI’s asynchronous endpoints allows the gateway to handle thousands of concurrent connections without blocking.

Python Implementation

from fastapi import FastAPI, BackgroundTasks, HTTPException
from pydantic import BaseModel
from aiokafka import AIOKafkaProducer
import json

app = FastAPI()
producer = AIOKafkaProducer(bootstrap_servers='localhost:9092')

class SecurityLogPayload(BaseModel):
    source_id: str
    title: str
    raw_content: str

@app.on_event("startup")
async def startup_event():
    await producer.start()

@app.post("/v1/ingest", status_code=202)
async def ingest_log(payload: SecurityLogPayload):
    # Immediate hand-off to Kafka to keep API response times under < 50ms
    payload_bytes = json.dumps(payload.dict()).encode('utf-8')
    await producer.send_and_wait("raw-security-logs", payload_bytes)
    return {"status": "accepted"}

Step 2: The Claim Check Pattern & Deduplication (Kafka + Redis)

Security logs frequently exceed the standard message size constraints of distributed brokers like Apache Kafka. Pushing multi-megabyte text structures or file attachments directly into Kafka topics degrades performance.

Instead, we employ the Claim Check Pattern:

1. Payload Extraction: The raw file or massive log chunk is pushed straight into scalable Object Storage (S3/MinIO).
2. Reference Message: A lightweight reference containing only the object_url, metadata, and content hash is produced to the Kafka topic.

Before any worker pulls the reference to parse it, the system checks a distributed Bloom Filter or high-performance cache like Redis using a calculated SHA-256 hash of the title or log content. If the hash exists, the message is instantly discarded or linked to an existing record, saving massive downstream database overhead.

Step 3: Asynchronous Enrichment & API Throttling

Once a worker picks up a unique log reference, it needs to verify file reputations (e.g., querying platforms like VirusTotal).

Because external threat intelligence APIs enforce strict rate limits, workers must never make synchronous, blocking HTTP requests inside the main thread. A distributed worker pool handles this by checking an internal cache of previously scanned hashes first. If a lookup is required, it utilizes a token-bucket rate limiter to smoothly query external APIs without exhausting quotas.

Step 4: Tiered Storage (Elasticsearch, Postgres, and Apache Iceberg)

To extract real value from CTI data, storage must be split based on the access pattern:

• PostgreSQL (Operational State): Holds structural entities, processing states, metadata records, and configuration controls.
• Elasticsearch / OpenSearch (Hot Search Layer): Unstructured text bodies and log entries are indexed into Elasticsearch. This allows threat analysts to execute fast wildcard queries, regex matches, and text aggregations.
• Apache Iceberg / Lake Keeper (Cold Analytics Layer): Keeping petabytes of data in Elasticsearch is cost-prohibitive. As logs age past 30–90 days, processing workers format and commit the data into open table formats like Apache Iceberg. This enables data scientists to run massive, low-cost analytical queries using engines like Trino or Spark.