Chapter 57

Flow Collection at Scale

An enterprise with 200 routers and switches, each exporting NetFlow at 1000 records/second, generates 200,000 flow records per second. A naively designed collector will be overwhelmed and drop records — losing exactly the visibility you deployed NetFlow to gain. This chapter covers flow collection architectures that handle production scale without record loss.

Scenario

A telecom company has 500 edge routers each exporting IPFIX at 5000 flows/second (2.5 million flows/second total). They deploy a dedicated flow collection cluster with 4 collector nodes each handling 25% of exporters, writing to Kafka, and feeding into a ClickHouse time-series database. Total throughput: 2.5M flow records/second, queryable with sub-second response time for any 24-hour window.

Collection Architecture at Scale

  High-Scale Flow Collection Architecture
  ═══════════════════════════════════════════════════════════════════

  Exporters (routers, switches, firewalls)
  │
  ├── Group A (routers 1-50) → Collector-1 UDP 2055
  ├── Group B (routers 51-100) → Collector-2 UDP 2055
  ├── Group C (routers 101-150) → Collector-3 UDP 2055
  └── Group D (routers 151-200) → Collector-4 UDP 2055

  Each collector node:
  ├── UDP receive: nfcapd or GoFlow2 or Logstash
  ├── Template cache: stateful! Templates must reach same collector as records
  ├── Normalization: decode v5/v9/IPFIX to unified schema
  └── Output: write to Kafka

  Kafka (flow data bus):
  ├── Topic: "netflow-raw" (all flow records)
  ├── Topic: "netflow-alerts" (anomaly detections)
  └── Consumers: Elasticsearch, ClickHouse, Flink (real-time processing)

  Storage options:
  ├── ClickHouse: columnar OLAP, excellent for flow analytics
  │     Query 100B rows in <1s for count(*) WHERE src_ip='x'
  ├── Elasticsearch: good for ad-hoc text search, worse for range aggregations
  └── Apache Arrow + Parquet: long-term cold storage, batch analytics

  Query interface:
  ├── Grafana dashboards (ClickHouse/Elasticsearch sources)
  ├── nfdump CLI: for forensic queries on raw .nf files
  └── Python/pandas: for custom anomaly detection analytics

GoFlow2 — High-Performance Collector

bashgoflow2-setup.sh
#!/bin/bash

echo "=== GoFlow2: high-performance multi-format flow collector ==="
# GoFlow2 by NetSampler: handles NetFlow v5, v9, IPFIX, sFlow
# Exports to Kafka, file, or stdout as JSON
# https://github.com/netsampler/goflow2

# Download binary
GOFLOW_VERSION="v1.3.0"
wget "https://github.com/netsampler/goflow2/releases/download/${GOFLOW_VERSION}/goflow2-linux-amd64"
chmod +x goflow2-linux-amd64
mv goflow2-linux-amd64 /usr/local/bin/goflow2

echo ""
echo "=== Run GoFlow2 with Kafka output ==="
cat > /etc/goflow2/config.yaml << 'CONF'
# GoFlow2 configuration
transport:
  kafka:
    brokers: ["kafka1:9092", "kafka2:9092"]
    topic: "netflow-raw"
    flushBytes: 1048576    # 1 MB batch size
    flushFrequency: 1000   # 1 second max delay

format:
  type: json               # Output as JSON (one record per line)

listen:
  - addr: "0.0.0.0:2055"  # NetFlow v5/v9
    transport: netflow
  - addr: "0.0.0.0:4739"  # IPFIX
    transport: nflegacy
  - addr: "0.0.0.0:6343"  # sFlow
    transport: sflow
CONF

goflow2 --config /etc/goflow2/config.yaml &
echo "GoFlow2 listening on ports 2055 (NetFlow), 4739 (IPFIX), 6343 (sFlow)"

echo ""
echo "=== GoFlow2 JSON output format sample ==="
cat << 'JSON'
{
  "Type": "NETFLOW_V9",
  "TimeReceived": 1700000000,
  "SamplerAddress": "10.0.0.1",
  "SrcAddr": "10.1.2.3",
  "DstAddr": "52.1.2.3",
  "SrcPort": 45123,
  "DstPort": 443,
  "Proto": 6,
  "Bytes": 1048576,
  "Packets": 1024,
  "TimeFlowStart": 1700000000,
  "TimeFlowEnd": 1700000120,
  "TCPFlags": 27,
  "InputInt": 4,
  "OutputInt": 2,
  "NextHop": "10.0.0.254"
}
JSON

echo ""
echo "=== Consume from Kafka and write to ClickHouse ==="
# Python consumer using confluent-kafka + clickhouse-driver
cat > /opt/flow-consumer.py << 'PYTHON'
#!/usr/bin/env python3
"""Consume NetFlow records from Kafka, insert into ClickHouse."""
from confluent_kafka import Consumer
from clickhouse_driver import Client
import json

kafka = Consumer({
    'bootstrap.servers': 'kafka1:9092,kafka2:9092',
    'group.id': 'netflow-consumer',
    'auto.offset.reset': 'earliest',
})
kafka.subscribe(['netflow-raw'])

ch = Client('clickhouse1', database='netflow')
batch = []
BATCH_SIZE = 10000

while True:
    msg = kafka.poll(timeout=1.0)
    if msg is None:
        continue
    if msg.error():
        print(f"Error: {msg.error()}")
        continue

    try:
        record = json.loads(msg.value().decode('utf-8'))
        batch.append((
            record.get('TimeFlowStart', 0),
            record.get('SamplerAddress', ''),
            record.get('SrcAddr', ''),
            record.get('DstAddr', ''),
            record.get('SrcPort', 0),
            record.get('DstPort', 0),
            record.get('Proto', 0),
            record.get('Bytes', 0),
            record.get('Packets', 0),
            record.get('TCPFlags', 0),
        ))
    except Exception as e:
        print(f"Parse error: {e}")

    if len(batch) >= BATCH_SIZE:
        ch.execute(
            'INSERT INTO flows (ts,sampler,src_ip,dst_ip,src_port,dst_port,'
            'proto,bytes,packets,flags) VALUES',
            batch
        )
        batch = []
PYTHON
Why template state must stay with the same collector node

NetFlow v9 and IPFIX use a template-based encoding: before sending data records, the exporter sends a template record that defines the field layout. The collector must receive and cache the template before it can decode any data records that use that template. This creates a critical stateful constraint in distributed collection: all flow records from a given exporter must go to the same collector instance, because only that instance has the template in its cache. If you load-balance exporters across collectors by round-robin at the UDP level, template records go to one collector and data records go to another — the data records are undecodable. The correct approach is to assign each exporter to exactly one collector statically (configure the router to send to a specific IP). For high-availability: use primary/backup collector IPs on the router, not load balancing. GoFlow2 handles template state internally and is stateful per exporter address.

Q & A

Q: How do I detect when a flow exporter goes silent and stops sending records?

Silent exporter detection is critical because an attacker who controls a router might disable NetFlow export to blind the monitoring system. Detection approach: (1) Maintain a "last seen" timestamp per exporter IP in your collector. Alert if any known exporter hasn't sent a record in more than N minutes (e.g., 5 minutes for active exporters, accounting for the flow timeout settings). (2) Track record count per exporter per 5-minute window. A sudden 100% drop in records from an exporter that was previously reliable is either a device failure or tampering. (3) In nfcapd: check the nfstat output regularly — it shows per-exporter statistics including last-received timestamp. (4) In ClickHouse/Elasticsearch: SELECT sampler, max(ts) FROM flows GROUP BY sampler HAVING max(ts) < now() - INTERVAL 10 MINUTE returns exporters that have gone silent. Set this as a scheduled query that generates an alert. The alerting threshold must account for the router's flow cache timeout (flows are only exported when they close or the active timeout fires — typically every 60-300 seconds) plus transport latency.