Processing 10 Million Logs Per Second: Migration from RabbitMQ to Kafka

In Turkey, Law 5651 mandates that ISPs and large organizations must collect and store user activity(wan outgate) logs for security and legal compliance. At a major logistic company, we faced the challenge of processing firewall logs from multiple NAS devices that could spike up to 10 million logs per second during peak hours.

Initial Architecture:

Rsyslog → RabbitMQ → Go Consumer → Elasticsearch

This setup worked fine for normal loads but struggled dramatically during peak times, threatening our compliance requirements. If you are going to run your services on medium sized companies, this will be work fine.

Chapter 1: Elasticsearch Bulk Optimization Attempts

Before considering a complete architecture change, we tried optimizing our Elasticsearch setup.

First Attempt: Bulk Size Tuning

We experimented with different bulk configurations:

biCfg := esutil.BulkIndexerConfig{
    Client:        es,
    NumWorkers:    8,
    FlushBytes:    5 * 1024 * 1024,
    FlushInterval: 3 * time.Second,
}

Increasing workers from 8 to 32 and flush size to 50MB showed improvements:

• Before: 20K docs/s
• After: 50K docs/s

That’s hude to be honest. And This situation raised questions in our minds. We also reviewed the application again. ElasticSearch was really good for fuzzy search and log discovery. However, we were writing logs more than reading them. After that, we exported them at regular intervals and signed them with a timestamp.

Second Attempt: Index Optimization

We tuned Elasticsearch index settings:

{
  "settings": {
    "index": {
      "refresh_interval": "30s",
      "number_of_shards": 10,
      "number_of_replicas": 0,
      "translog.durability": "async"
    }
  }
}

This pushed us to 80K docs/s, but still nowhere near our 10M target. The fundamental issue was that Elasticsearch, designed for search, wasn’t optimal for high-volume time-series data ingestion.

Chapter 2: The Kafka Decision

RabbitMQ Limitations

Our RabbitMQ metrics showed clear bottlenecks:

• Average throughput: 50K msg/s
• Peak capability: 100K msg/s (unstable)
• Memory usage: 16GB RAM just for queues
• Persistent messages caused severe disk I/O bottlenecks

Why Kafka?

1. Designed for high throughput: LinkedIn built it for log aggregation
2. Horizontal scaling: Add brokers for linear performance increase
3. Compression: Native support for LZ4, reducing network load by 70%. that was perfect for us. There were too many nas located at different zones. Distance required too many firewall routings.
4. Durability: Replication ensures no data loss for compliance

Migration Strategy

We implemented a gradual migration using parallel consumers:

type DualConsumer struct {
    rabbitConsumer *RabbitConsumer
    kafkaConsumer  *KafkaConsumer
    switchPercent  int32
}

func (d *DualConsumer) Start() {
    go d.rabbitConsumer.Consume()
    go d.kafkaConsumer.Consume()

    ticker := time.NewTicker(5 * time.Minute)
    for range ticker.C {
        current := atomic.LoadInt32(&d.switchPercent)
        if current < 100 {
            atomic.AddInt32(&d.switchPercent, 10)
        }
    }
}

This approach allowed us to gradually shift traffic from RabbitMQ to Kafka over several hours, with the ability to roll back instantly if issues arose.

Chapter 3: ClickHouse Revolution

The Case for ClickHouse

Migation to new service made us nervous ofc. But we kept testing. After solving the ingestion problem with Kafka, we still faced storage and query performance issues with Elasticsearch. ClickHouse, designed specifically for time-series data, offered compelling advantages:

MetricElasticsearchClickHouseWrite Speed50K/s per node2M+/s per nodeCompression Ratio20%90%1B Row Aggregation30-60 seconds0.5-2 secondsStorage Cost$0.10/GB/month$0.02/GB/month

Schema Design

CREATE TABLE logs (
    timestamp DateTime,
    brand String,
    source_ip IPv4,
    dest_ip IPv4,
    action Enum8('allow' = 1, 'deny' = 2),
    bytes UInt32,
    INDEX idx_timestamp timestamp TYPE minmax GRANULARITY 8192
) ENGINE = MergeTree()
PARTITION BY toYYYYMMDD(timestamp)
ORDER BY (brand, timestamp)
TTL timestamp + INTERVAL 90 DAY TO DISK 'cold_storage';

The TTL feature was crucial for Law 5651 compliance, automatically moving older logs to cheaper storage while keeping them queryable. Before that on elasticsearch, we cron a job for every night to export logs and if there are exported logs drop them. TTL was perfect!

Chapter 4: Go Consumer Optimization

Memory-Efficient Processing

We implemented object pooling to reduce garbage collection pressure:

var logPool = sync.Pool{
    New: func() interface{} {
        return &ParsedLog{
            Fields: make(map[string]string, 20),
        }
    },
}

func parseLogEfficient(raw []byte) *ParsedLog {
    log := logPool.Get().(*ParsedLog)
    defer logPool.Put(log)

    log.Timestamp = binary.BigEndian.Uint64(raw[0:8])
    log.SourceIP = net.IP(raw[8:12])

    for k := range log.Fields {
        delete(log.Fields, k)
    }

    return log
}

This reduced GC pauses from 100ms to under 10ms during peak loads.

Pipeline Architecture

We structured the processing as a multi-stage pipeline:

type Pipeline struct {
    stages []Stage
    metrics *Metrics
}

func (p *Pipeline) Run(ctx context.Context) {
    channels := make([]chan Message, len(p.stages)+1)

    for i := range channels {
        channels[i] = make(chan Message, 10000)
    }

    for i, stage := range p.stages {
        for w := 0; w < stage.Workers(); w++ {
            go stage.Process(ctx, channels[i], channels[i+1])
        }
    }
}

Each stage ran in separate goroutine pools:

• Stage 1: Kafka consumption (16 workers)
• Stage 2: Log parsing (32 workers)
• Stage 3: Enrichment (16 workers)
• Stage 4: ClickHouse writing (64 workers)

Chapter 5: Production Results

Kafka Configuration

Key Kafka settings for optimal performance:

num.network.threads=16
num.io.threads=16
socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576
compression.type=lz4
batch.size=1048576
linger.ms=10

Real-World Metrics

Before Migration:

• Throughput: 50K msg/s (unstable during peaks)
• End-to-end latency: p99 = 500ms
• Storage: 10TB/month
• Monthly cost: $8,000

After Migration:

• Throughput: 5M msg/s (stable, with headroom for 10M)
• End-to-end latency: p99 = 50ms
• Storage: 1.2TB/month (after compression)
• Monthly cost: $1,200

Compliance Benefits

The new architecture enhanced our Law 5651 compliance:

• Guaranteed log retention for required 2-year period(for regulation 6 month enough)
• Sub-second query response for legal requests
• Cryptographic checksums for tamper detection
• Automated data lifecycle management

Chapter 6: Lessons Learned

1. Right Tool for the Job

Elasticsearch excels at search but struggles with high-volume time-series ingestion. ClickHouse, purpose-built for this use case, delivered 100x better performance. When choosing a microservice, we got stuck with Elasticsearch. If we had focused more on comparisons earlier, we could have delivered the project sooner.

2. Compression Matters

Network bandwidth became our first bottleneck. LZ4 compression in Kafka reduced network load by 70% with minimal CPU impact.

3. Object Pooling in Go

For high-throughput systems, garbage collection can become a major bottleneck. Object pooling and zero-allocation techniques are essential.

4. Law 5651 Considerations

When building log systems in Turkey:

• Plan for 2-year retention from day one
• Implement tamper-proof mechanisms
• Ensure fast query capabilities for compliance requests
• Consider cold storage for cost optimization

Most importantly, we transformed from a system struggling with compliance requirements to one that easily handles current loads with significant headroom for growth.

The combination of Kafka for ingestion, ClickHouse for storage, and optimized Go consumers proved to be the perfect stack for high-volume log processing in a regulatory environment.

End of the story, there are similar regulations and log streams in everywhere. I belive this little tricks might be helpfull.