Distributed Logging (eg: Sumologc, Datadog, Splunk cloud)

This is a multi-tenant distributed version of Sumo Logic, Splunk Cloud, Datadog Logs, or Grafana Loki, where applications running across thousands of machines and multiple regions continuously stream logs to a central platform for indexing, searching, alerting, and analytics.
Example: Netskope sending(Event logs, application logs, transaction logs, service logs, alerts), CompanyX(general logs), Domain Controller(sends syslogs)
Purpose: Distributed logging system will store the logs and provide dashboard for monitoring.

Requirements

Functional

1. Log Collection: Tenants/Customers/Applications can send(Application logs, System logs, Audit logs, Security logs, Kubernetes logs, Container logs, Network device logs. Support multiple protocols(HTTP HTTPS gRPC TCP UDP (Syslog) Kafka)
2. Highly scalable: Millions of logs/sec
3. Near Real-Time Search(within 2-5 seconds) Formats:
Text Search: (status=500, user=amit, host=server12, requestId=abc123, service=auth)
Time Range Query: (Last 5 minutes, Last 1 hour, Yesterday, Last 30 days)
Aggregation: (count(), avg(), sum(), percentile(), top(), histogram(), group by())
4. Send Alerts: If ERROR > 100/sec, Send Slack, PagerDuty, Email, Webhook
5. Multitenant: 1000s of customers
6. Different Rentention polcies: 7 days, 30 days, 180 days, 1 year

Non-functional Requirements

High availability (99.99%+), Horizontal scalability, Fault tolerance, Low ingestion latency, High write throughput, Eventual consistency for search, Secure transport (TLS), Encryption at rest, RBAC, Cost-efficient storage, Multi-region deployment

HLD

There are 2 independent systems:
1. Storage Pipeline (Write Path): optimized for ingesting millions of events/sec. Where tenants write the logs and those are stored
2. Query Pipeline (Read Path): optimized for searching petabytes of historical data. Dashboards where data is searched looked upon for investigation/forensics

Storage Pipeline (Write Path)

flowchart LR
    subgraph TENANT_NS1["Tenant Netskope"]
        APP1[POP1
nsproxy]
        subgraph Agent Client1
            AGENT1[Fluent Bit
Collector
Agent]
            COMPRESS1[gzip / zstd]
        end
    end
    subgraph TENANT_NS1["Tenant Netskope"]
        APP2[POP2
nsproxy]
        subgraph Agent Client2
            AGENT2[Collector
Agent]
            COMPRESS2[gzip / zstd]
        end
    end

    subgraph Tenant XYZ 
        APP3[Server]
        subgraph AWS
            AGENT3[Open Telemetry
Collector
Agent]
            COMPRESS3[gzip / zstd]
        end
    end

    APP1 -->|transaction logs
Alerts
Event logs

Log Format1| AGENT1
    AGENT1 --> COMPRESS1
    APP2 --> AGENT2
    AGENT2 --> COMPRESS2
    APP3 -->|General logs

Log Format2| AGENT3
    AGENT3 --> COMPRESS3

    COMPRESS1 -->|"1000 log bundle 
HTTP
Header:metadata(tenantid,Timestamp..)
Body:500 log batch logs.zip"| GLB[Global
Load
Balancer]
    COMPRESS2 -->|"500 log bundle 
HTTP
Header:metadata
Body:logs.zip"| GLB[Global
Load
Balancer]
    COMPRESS3 -->|"100 log bundle 
HTTP
Header:metadata
Body:logs.zip"| GLB[Global
Load
Balancer]
    GLB --> RLB[Regional
Load
Balancer]
    RLB --> APIGW[API
Gateway]

    subgraph Sumologic Datacenter 
    APIGW --> AUTH[Tenant Validation Cache]

    APIGW --> INGRESS[Kubernetes Ingress
Load Balancer]

    INGRESS --> INGEST1[Ingestion
Service Pod1]
    INGRESS --> INGEST2[Ingestion
Service Pod2]
    INGEST1 -.- Note["1.Add Meta Data to logs
2.Create Kafka Topic"]
    style Note fill:#fff9c4,stroke:#fbc02d,stroke-width:1px

    INGEST1 -->|validate request| CACHE[(Authentication
CACHE)]
    CACHE -->|Validated| INGEST1

    INGEST1 -->|Topic=Log
Msg=logs.zip+MetaData| RETRY[Retry Queue]

    RETRY --> KAFKA[(Kafka)]

    INGEST1 -->|Topic=Log
Msg=logs.zip+MetaData| DLQ[(Dead Letter Queue)]

    KAFKA --> Processing_Worker[Workers]
    end
    Processing_Worker -->|logs.zip| Hot[(Hot Storage
7-30 days
SSSD)]
    Processing_Worker -->|logs.zip| Cold[(Cold Storage
1,5,10 years
Amazon S3)]
    Processing_Worker -->|Metadata| Meta[(Metadata)]

    INGEST1 --> METRICS[Prometheus Metrics]

Collector Agent (Installed on Log Server). Eg: Opentelemetry Agent, Fluent Bit Agent

Suppose one server generates: 100,000 logs/sec. It does not send: 100,000 HTTP requests/sec. Instead it buffers logs. like
500 logs ->One HTTP request
Now authentication will happen on this 1 HTTP request

API Gateway

TLS termination, DDoS, Rate limiting, API key/JWT validation, Routing

Ingestion Service

Validate tenant exists, Validate source type, Attach metadata, send to kafka

Authentication

Imagine Netskope signs up for your logging platform. During onboarding, they receive:

Tenant ID: tenant-3421
API Key: abcXYZ...
Secret: xxxxxxxx
Region: us-east
Retention: 90 days

Their log agent (for example, Fluent Bit or OpenTelemetry Collector) is configured with these credentials. Every request includes headers such as:

POST /v1/logs
Authorization: Bearer eyJhb...
X-Tenant-ID: tenant-3421
X-Source: kubernetes
Content-Encoding: gzip
Content-Type: application/json

Authentication will happen on every batched request. See Collector agent sends logs in batch of 500,100 logs
At every HTTP request following things can be checked from local cache tenant, rentention days, quota, region, allowed sources

Pods Scaling

if traffic doubles. Scales itself(HPA. On CPU, memory, request rate, or queue depth). 6 Pods -> 40 Pods -> 100 Pods

Proccessing at Ingestion Service

Ingestion service will not unzip every log.zip, if Ingestion service keep unzipping, it will become bottleneck.
With every log message customer also sends Meta data(whose schema is same for all customers)
Different customers will have different log formats, but same schema


Log format
Company A {"user":"amit","error":"timeout"}
Company B {"employee":"123","status":"failed"}

Meta data schema
Tenant ID, Timestamp, Host, Region, Source, Log Type, Service, Namespace, Cluster, Severity, Trace ID, Span ID, Request ID, Compression Type, Retention Class, Encryption Key ID, Schema Version

Sequence Diagram

sequenceDiagram
    autonumber

    box Customer Premises POP
        participant App as Server Generating Log
        participant Collector
    end

    participant API as API Gateway

    box Datacenter
        participant Ingest as Ingestion Service
        participant Cache as Tenant Cache
        participant MetaDB as Metadata DB Batch Level
        participant Kafka
    end

    App->>Collector: Generate log events

    Collector->>Collector: Buffer 500 logs / 1 MB / 1 sec

    Note over Collector: Compress batch gzip/zstd
500 logs per HTTP POST

    Collector->>API: HTTPS POST JWT + Tenant ID
Body logs.zip

    API->>API: TLS Rate Limit JWT Validation

    API->>Ingest: Forward authenticated request

    Ingest->>Cache: Get tenant data
    Cache-->>Ingest: Retention Quota Allowed Sources

    Note over Ingest: Validate tenant and sources

    Note over Ingest: Add metadata
Tenant ID,Region
Source,Log Type
Service,Host
Namespace,Cluster
Retention,Encryption Key ID
Schema Version,Compression Type
Ingestion Timestamp

    Ingest->>MetaDB: Store Batch Metadata
(BatchID, TenantID, Time Range, Region,
Source, Log Count, Compression,
Retention, Object ID*)

    MetaDB-->>Ingest: Metadata stored

    Ingest->>Kafka: Publish internal event batch

    Kafka-->>Ingest: ACK

    Ingest-->>Collector: HTTP 200 OK

Query Pipeline (Read Path)

Downstream workers: Messages are consumed by downstream workers, transformed, indexed, and stored.
Dashboard: queries the Search Index (OpenSearch, ClickHouse, Loki, etc.) and retrieves raw logs from Object Storage only when necessary

flowchart LR

    K1[(Kafka
raw-logs)]

    P[Parser Service
multipod]
    K2[(Kafka
normalized-logs)]

    E[Enrichment Service
multipod]
    K3[(Kafka
enriched-logs)]

    IX[Indexer Service
multipod

consumer-group=indexer-group]
    CW[Chunk Writer Service
multipod

consumer-group=chunker-group]
    MW[Metadata Writer Service
multipod

consumer-group=metadata-group]

    SI[(Search Index
OpenSearch / ClickHouse)]
    OS[(Object Storage
S3 / Blob)]
    MDB[(Metadata DB)]

    K1 --> P
    P --> K2

    K2 --> E
    E --> K3

    K3 --> IX
    K3 --> CW
    K3 --> MW

    IX --> SI
    CW --> OS
    MW --> MDB

Service Work

Parser Servic

Parser Service converts customer-specific formats into a normalized representation

Suppose Kafka contains:

{
  "tenantId": "abc company",
  "source": "kubernetes",
  "payload": {
    "time": "10:31:12",
    "level": "ERROR",
    "msg": "Redis timeout"
  }
}

Normalized representation (ie its own internal representation):

{
  "tenantId": "abc company",
  "timestamp": "2026-08-08T10:31:12Z",       <<<<<
  "severity": "ERROR",              <<<<<
  "message": "Redis timeout",       <<<<<
  "service": "policy-engine",
  "host": "server-101"
}

Enrichment Service

Adds additional information(recieved from Collector Agent) to logs

Cloud region
Kubernetes cluster
Namespace
Pod
Container
Availability zone
Geo information
Security classification

{
  "tenantId": "abc company",
  "timestamp": "...",
  "severity": "ERROR",
  "service": "policy-engine",
  "host": "server-101",
  "region": "us-east-1",
  "cluster": "prod-cluster-01",
  "namespace": "proxy"
}

Indexer Service

Creates the inverted indexes


ERROR (present in following documents)
  ├── document 123
  ├── document 145
  └── document 991

policy-engine
  ├── document 123
  ├── document 145

Chunk Writer Create batch of logs. logs -> Create batch(100MB or 5 sec) -> compress -> encrypt -> S3

Metadata Writer

Maintains the catalog telling us where the actual data lives

Chunk ID: chunk-12345
Tenant: abc
Region: us-east-1
Start Time: 10:00
End Time: 10:05
Object: s3://.../chunk-12345.zst
Log Count: 4,500,000
Compression: zstd
Encryption Key: key-42

Logs on Dashboard

flowchart LR

    USER[User]

    UI[Logging Dashboard]

    API[Query API Service]

    AUTH[Authentication &
Authorization]

    QC[Query Coordinator]

    QS[ClickHouse Query
Service]

    CH[(ClickHouse Shard1
Distributed Tables)]
    CH1[(ClickHouse Shard2
Distributed Tables)]
    CH2[(ClickHouse Shard3
Distributed Tables)]

    CACHE[Query Cache
Optional]

    FORMAT[Result Formatter
Pagination / Sorting / Aggregation]

    USER --> UI

    UI --> API

    API --> AUTH

    AUTH --> QC

    QC --> CACHE

    CACHE -->|Cache Miss| QS

    QS <--> CH
    QS <--> CH1
    QS <--> CH2

    QS --> QC

    QC --> FORMAT

    FORMAT --> API

    API --> UI

    UI --> USER

Sequence Diagram

Optimization is that (tenant_id, timestamp) should be part of the physical data layout/partitioning strategy, so ClickHouse doesn't scan the entire log dataset.
Search from S3/Object Store:When the requested time range has moved to cold storage or when you deliberately keep only a subset of fields in ClickHouse.

sequenceDiagram

    participant User
    participant UI as Dashboard/Browser
    participant API as Query API
    participant Auth as Auth/IDP
    participant QC as Query Coordinator
    participant Cache as Query Cache
    participant QSvc as ClickHouse Query Service
    participant CH as ClickHouse Cluster
    participant S1 as Shard 1
    participant S2 as Shard 2
    participant S3 as Shard 3

    User->>UI: Selects
Tenant: abc
Service: Policy Engine
Severity: ERROR
Time: Last 1 hour

    UI->>API: GET /api/v1/logs?
service=policy-engine
&severity=ERROR
&from=10:00
&to=11:00

    API->>Auth: Authenticate + authorize user

    Auth-->>API: Authorized for tenant=abc

    API->>QC: Execute log query

    QC->>QC: Validate query
Apply tenant scope
Determine time range
Apply query limits

    QC->>Cache: Check query cache

    alt Cache Hit

        Cache-->>QC: Cached results

    else Cache Miss

        QC->>QSvc: Execute ClickHouse query

        Note over QSvc:SELECT timestamp, service,
host,severity,
message,trace_id
FROM logs
WHERE tenant_id = 'abc'
AND service = 'policy-engine'
AND severity = 'ERROR'
AND timestamp >= now() - INTERVAL 1 HOUR
ORDER BY timestamp DESC
LIMIT 100;
        QSvc->>CH: Distributed query

        CH->>S1: Query relevant partitions
        CH->>S2: Query relevant partitions
        CH->>S3: Query relevant partitions

        S1-->>CH: Matching rows
        S2-->>CH: Matching rows
        S3-->>CH: Matching rows

        CH->>CH: Merge shard results
        CH->>CH: Sort by timestamp
        CH->>CH: Apply LIMIT / aggregation

        CH-->>QSvc: Query results

        QSvc-->>QC: Query results

        QC->>Cache: Store result (optional)

    end

    QC->>QC: Merge / sort / paginate

    QC-->>API: Final result set

    API-->>UI: JSON response

    UI->>UI: Render log table

    UI-->>User: Display logs