搭建基于goflow2+kafka+clickhouse+grafana架构的跨省流量监控

本次使用的系统及环境为：

软件/系统	版本
Ubuntu	22.04.5
GoFlow2	v2.2.6
Kafka	v4.1.1
ClickHouse	v25.12.2.54
Grafana	v12.3.1

实现流程#

交换机通过sFlow/NetFlow协议将流量数据发送到GoFlow2采集器，GoFlow2解析后以JSON格式写入Kafka消息队列，ClickHouse从Kafka消费数据并存储到列式数据库，最后Grafana通过SQL查询ClickHouse进行可视化展示和监控。

1
交换机(sFlow/NetFlow)
2
    ↓ (UDP 6343/2055端口)
3
GoFlow2(协议解析)
4
    ↓ (JSON格式)
5
Kafka(消息缓冲)
6
    ↓ (消费者拉取)
7
ClickHouse(列式存储)
8
    ↓ (SQL查询)
9
Grafana(可视化监控)

下载并部署kafka#

安装java#

kafka需要本地java环境为≥17，先安装java

1
apt update && apt install -y openjdk-17-jdk

下载kafka并修改环境变量#

1
wget https://dlcdn.apache.org/kafka/4.1.1/kafka_2.13-4.1.1.tgz
2
tar -xzf kafka_2.13-4.1.1.tgz
3
mv kafka_2.13-4.1.1 /opt/
4
sudo ln -s /opt/kafka_2.13-4.1.1 /opt/kafka
5
#修改环境变量
6
cat << 'EOF' >> /etc/profile
7
export KAFKA_HOME=/opt/kafka
8
export PATH=$PATH:$KAFKA_HOME/bin
9
EOF
10
#使环境变量立即生效
11
source /etc/profile
12
#查看一下kafka版本，判断环境变量和kafka是否正常
13
kafka-topics.sh --version
14
#4.1.1

初始化kafka#

1
# 创建数据目录
2
sudo mkdir -p /var/lib/kafka/data
3
sudo mkdir -p /var/lib/zookeeper/data
4
#每个Kafka集群需要一个全局唯一ID，先随机生成 Kafka 的集群 UUID
5
KAFKA_CLUSTER_ID="$(/opt/kafka/bin/kafka-storage.sh random-uuid)"
6
echo "Cluster UUID: $KAFKA_CLUSTER_ID"
7
#Cluster UUID: tStfHNWWSJ6fK555k9ILGg
8
# 创建持久化数据目录
9
sudo mkdir -p /var/lib/kafka/kraft-combined-logs
10
# 设置权限
11
sudo chmod -R 755 /var/lib/kafka

编辑Kafka配置文件#

每个参数的含义可以参考这个链接

1
nano /opt/kafka/config/server.properties

1
############################# Server Basics #############################
2

3
# KRaft 模式角色配置
4
# 因为我是单台部署，服务器同时担任数据存储和元数据管理角色，正式生产环境建议分离
5
process.roles=broker,controller
6
# 设置节点id
7
node.id=1
8
# 以下端口均可自由修改，但需要同时修改goflow和clickhouse的配置
9
controller.quorum.bootstrap.servers=localhost:9093
10

11
############################# Socket Server Settings #############################
12

13
# 监听器配置
14
listeners=PLAINTEXT://:9092,CONTROLLER://:9093
15
inter.broker.listener.name=PLAINTEXT
16
advertised.listeners=PLAINTEXT://localhost:9092,CONTROLLER://localhost:9093
17
controller.listener.names=CONTROLLER
18
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
19

20
# 网络线程配置
21
num.network.threads=3
22
num.io.threads=8
23

24
# Socket 缓冲区配置
25
socket.send.buffer.bytes=102400
26
socket.receive.buffer.bytes=102400
27
socket.request.max.bytes=104857600
28

29
############################# Log Basics #############################
30

31
# 日志目录（持久化存储）
32
log.dirs=/var/lib/kafka/kraft-combined-logs
33

34
# 分区配置（提高并行性）
35
num.partitions=3
36
num.recovery.threads.per.data.dir=1
37

38
############################# Internal Topic Settings #############################
39

40
# 内部主题副本因子
41
offsets.topic.replication.factor=1
42
share.coordinator.state.topic.replication.factor=1
43
share.coordinator.state.topic.min.isr=1
44
transaction.state.log.replication.factor=1
45
transaction.state.log.min.isr=1
46

47
############################# Log Retention Policy #############################
48

49
# 日志保留策略（7天）
50
log.retention.hours=168
51
log.segment.bytes=1073741824
52
log.retention.check.interval.ms=300000
53

54
############################# Performance Tuning #############################
55

56
# 压缩（节省存储和带宽）
57
compression.type=snappy
58

59
# 批处理优化
60
batch.size=16384
61
linger.ms=10
62

63
# 网络缓冲
64
replica.socket.receive.buffer.bytes=65536

创建服务#

1
nano /etc/systemd/system/kafka.service

1
[Unit]
2
Description=Apache Kafka Server (KRaft Mode)
3
Documentation=https://kafka.apache.org/documentation/
4
After=network.target
5
Wants=network-online.target
6

7
[Service]
8
Type=simple
9
User=root
10
Group=root
11

12
# Java 堆内存配置（根据你的服务器内存调整）
13
Environment="KAFKA_HEAP_OPTS=-Xmx1G -Xms1G"
14
Environment="LOG_DIR=/var/lib/kafka/logs"
15
Environment="KAFKA_HOME=/opt/kafka"
16

17
# 工作目录
18
WorkingDirectory=/opt/kafka
19

20
# 启动命令
21
ExecStart=/opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/server.properties
22

23
# 停止命令
24
ExecStop=/opt/kafka/bin/kafka-server-stop.sh
25

26
# 重启策略
27
Restart=on-failure
28
RestartSec=10
29
StartLimitBurst=3
30
StartLimitInterval=120
31

32
# 日志
33
StandardOutput=journal
34
StandardError=journal
35
SyslogIdentifier=kafka
36

37
# 安全和资源限制
38
LimitNOFILE=65536
39
LimitNPROC=4096
40

41
[Install]
42
WantedBy=multi-user.target

启动kafka服务#

1
systemctl daemon-reload
2
systemctl enable kafka
3
systemctl start kafka
4
systemctl status kafka

1
# 创建 flow-messages topic（用于 NetFlow/IPFIX/sFlow 数据）
2
/opt/kafka/bin/kafka-topics.sh --create   --topic flow-messages   --bootstrap-server localhost:9092   --partitions 3   --replication-factor 1   --config retention.ms=86400000   --config compression.type=lz4
3
/opt/kafka/bin/kafka-topics.sh --list   --bootstrap-server localhost:9092

下载并部署goflow2#

1
wget https://github.com/netsampler/goflow2/releases/download/v2.2.6/goflow2-2.2.6-linux-amd64
2
chmod +x goflow2-2.2.6-linux-amd64
3
mv goflow2-2.2.6-linux-amd64 /usr/local/bin/goflow2

创建工作目录#

1
mkdir -p /var/log/goflow2
2
nano /etc/systemd/system/goflow2.service

编辑服务#

1
[Unit]
2
Description=GoFlow2 NetFlow/IPFIX/sFlow Collector
3
Documentation=https://github.com/netsampler/goflow2
4
After=network.target kafka.service
5
Wants=kafka.service
6

7
[Service]
8
Type=simple
9
User=root
10
Group=root
11
WorkingDirectory=/var/log/goflow2
12

13
# 将数据转发到kafka
14
ExecStart=/usr/local/bin/goflow2 \
15
  -transport=kafka \
16
  -transport.kafka.brokers=localhost:9092 \
17
  -transport.kafka.topic=flow-messages \
18
  -listen=sflow://:6343,netflow://:2055 \
19
  -format=json \
20
  -loglevel=info \
21
  -addr=:8080
22

23
Restart=on-failure
24
RestartSec=10
25
StandardOutput=journal
26
StandardError=journal
27
SyslogIdentifier=goflow2
28
LimitNOFILE=65536
29

30
[Install]
31
WantedBy=multi-user.target

启动Goflow2服务#

1
systemctl daemon-reload
2
systemctl enable goflow2
3
systemctl start goflow2
4
systemctl status goflow2

在交换机上配置sflow#

警告
以下操作因涉及到交换机配置，如果您不了解每个配置的含义，建议您停止操作，以免导致损失。
以下操作因涉及到交换机配置，如果您不了解每个配置的含义，建议您停止操作，以免导致损失。
以下操作因涉及到交换机配置，如果您不了解每个配置的含义，建议您停止操作，以免导致损失。

S系列：#

以华为S5720为例,不同设备命令不同，可通过这个链接查询设备对应信息

1
#在system上配置
2
sflow collector <1-2> ip <监控服务器ip> port 6343
3
sflow agent ip <交换机ip>
4

5
#进入需要监控的网口进行配置
6
interface <你要配置的网口>
7
sflow flow-sampling collector <你设定的collector id>
8
sflow flow-sampling rate 2000 #采集比例 1:2000，根据自身流量大小进行修改，建议参考官方文档
9
undo sflow flow-sampling inbound #只采集上行流量，不写默认监控上行和下行

CE系列#

以华为CE12804为例,不同设备命令不同，可通过这个链接查询设备对应信息

1
#在system上配置
2
sflow collector <1-2> ip <监控服务器ip> udp-port 6343
3
sflow agent ip <交换机ip>
4

5
#进入需要监控的网口进行配置
6
interface <你要配置的网口>
7
sflow sampling collector <你设定的collector id>
8
sflow sampling rate 32768  #采集比例 1:32768，根据自身流量大小进行修改，建议参考官方文档
9
sflow sampling outbound #只采集上行流量

检查采样包#

查看这个接口当前是否在采样，有没有采样包。

1
dis sflow statistics interface <你配置的网口>

如果没有包的话，返回服务器内查看goflow状态和6343端口是否正常，或检查交换机配置是否正常。

如果有采集包，在服务器内输入

1
kafka-console-consumer.sh --bootstrap-server localhost:9092   --topic flow-messages --from-beginning

来查看goflow2到kafka数据传输是否正常，正常情况下会收到以下内容：

1
{"type":"SFLOW_5","time_received_ns":1768279339983732918,"sequence_num":4,"sampling_rate":2048,"sampler_address":"192.168.1.201","time_flow_start_ns":1768279339983732918,"time_flow_end_ns":1768279339983732918,"bytes":68,"packets":1,"src_addr":"100.64.32.105","dst_addr":"112.17.34.247","etype":"IPv4","proto":"TCP","src_port":60534,"dst_port":80,"in_if":54,"out_if":21,"src_mac":"00:90:27:e1:12:d8","dst_mac":"a0:f4:79:59:d9:c5","src_vlan":100,"dst_vlan":100,"vlan_id":100,"ip_tos":0,"forwarding_status":0,"ip_ttl":64,"ip_flags":2,"tcp_flags":4,"icmp_type":0,"icmp_code":0,"ipv6_flow_label":0,"fragment_id":0,"fragment_offset":0,"src_as":0,"dst_as":0,"next_hop":"","next_hop_as":0,"src_net":"0.0.0.0/0","dst_net":"0.0.0.0/0","bgp_next_hop":"","bgp_communities":[],"as_path":[],"mpls_ttl":[],"mpls_label":[],"mpls_ip":[],"observation_domain_id":0,"observation_point_id":0,"layer_stack":["Ethernet","Dot1Q","IPv4","TCP"],"layer_size":[14,4,20,20],"ipv6_routing_header_addresses":[],"ipv6_routing_header_seg_left":0}

安装clickhouse#

1
sudo apt-get install -y apt-transport-https ca-certificates curl gnupg
2
curl -fsSL 'https://packages.clickhouse.com/rpm/lts/repodata/repomd.xml.key' | sudo gpg --dearmor -o /usr/share/keyrings/clickhouse-keyring.gpg
3
ARCH=$(dpkg --print-architecture)
4
echo "deb [signed-by=/usr/share/keyrings/clickhouse-keyring.gpg arch=${ARCH}] https://packages.clickhouse.com/deb stable main" | sudo tee /etc/apt/sources.list.d/clickhouse.list
5
sudo apt-get update
6
sudo apt-get install -y clickhouse-server clickhouse-client
7
#这里网络不好的话大概能摸鱼1h-2h左右（大概

等看到提示Set up the password for the default user: 直接回车即可

启动clickhouse服务#

1
systemctl enable clickhouse-server.service
2
systemctl start clickhouse-server.service
3
systemctl status clickhouse-server.service

创建数据库#

根据goflow2传到kafka的json参数，创建数据库及用得到的表先进入clickhouse并创建数据库

1
clickhouse-client
2
CREATE DATABASE netflow;

创建netflow.flows表#

此表作为MergeTree表，是最终存储和处理数据的地方，提供高效的查询性能。

1
-- 创建数据
2
CREATE TABLE netflow.flows
3
(
4
    `type` String,
5
    `time_received_ns` UInt64,
6
    `sequence_num` UInt32,
7
    `sampling_rate` UInt32,
8
    `sampler_address` String,
9
    `time_flow_start_ns` UInt64,
10
    `time_flow_end_ns` UInt64,
11
    `bytes` UInt64,
12
    `packets` UInt64,
13
    `src_addr` String,
14
    `dst_addr` String,
15
    `etype` String,
16
    `proto` String,
17
    `src_port` UInt16,
18
    `dst_port` UInt16,
19
    `in_if` UInt32,
20
    `out_if` UInt32,
21
    `src_mac` String,
22
    `dst_mac` String,
23
    `src_vlan` UInt16,
24
    `dst_vlan` UInt16,
25
    `vlan_id` UInt16,
26
    `ip_tos` UInt8,
27
    `forwarding_status` UInt8,
28
    `ip_ttl` UInt8,
29
    `ip_flags` UInt8,
30
    `tcp_flags` UInt8,
31
    `icmp_type` UInt8,
32
    `icmp_code` UInt8,
33
    `ipv6_flow_label` UInt32,
34
    `fragment_id` UInt32,
35
    `fragment_offset` UInt16,
36
    `src_as` UInt32,
37
    `dst_as` UInt32,
38
    `next_hop` String,
39
    `next_hop_as` UInt32,
40
    `src_net` String,
41
    `dst_net` String,
42
    `bgp_next_hop` String,
43
    `observation_domain_id` UInt32,
44
    `observation_point_id` UInt32,
45
    -- sflow传过来的数据为纳秒时间戳，转换为秒级时间戳
46
    `timestamp` DateTime DEFAULT toDateTime(time_received_ns / 1000000000)
47
)
48
ENGINE = MergeTree
49
PARTITION BY toYYYYMM(timestamp)
50
ORDER BY (timestamp, sampler_address, src_addr, dst_addr)
51
TTL timestamp + toIntervalDay(30)
52
SETTINGS index_granularity = 8192;

创建netflow.flows_kafka表#

此表作为Kafka引擎表，从Kafka主题中消费数据，是数据进入ClickHouse的桥梁。

1
-- 创建Kafka引擎表，从kafka的topic中拉取数据
2
CREATE TABLE netflow.flows_kafka
3
(
4
    `type` String,
5
    `time_received_ns` UInt64,
6
    `sequence_num` UInt32,
7
    `sampling_rate` UInt32,
8
    `sampler_address` String,
9
    `time_flow_start_ns` UInt64,
10
    `time_flow_end_ns` UInt64,
11
    `bytes` UInt64,
12
    `packets` UInt64,
13
    `src_addr` String,
14
    `dst_addr` String,
15
    `etype` String,
16
    `proto` String,
17
    `src_port` UInt16,
18
    `dst_port` UInt16,
19
    `in_if` UInt32,
20
    `out_if` UInt32,
21
    `src_mac` String,
22
    `dst_mac` String,
23
    `src_vlan` UInt16,
24
    `dst_vlan` UInt16,
25
    `vlan_id` UInt16,
26
    `ip_tos` UInt8,
27
    `forwarding_status` UInt8,
28
    `ip_ttl` UInt8,
29
    `ip_flags` UInt8,
30
    `tcp_flags` UInt8,
31
    `icmp_type` UInt8,
32
    `icmp_code` UInt8,
33
    `ipv6_flow_label` UInt32,
34
    `fragment_id` UInt32,
35
    `fragment_offset` UInt16,
36
    `src_as` UInt32,
37
    `dst_as` UInt32,
38
    `next_hop` String,
39
    `next_hop_as` UInt32,
40
    `src_net` String,
41
    `dst_net` String,
42
    `bgp_next_hop` String,
43
    `observation_domain_id` UInt32,
44
    `observation_point_id` UInt32
45
)
46
ENGINE = Kafka
47
SETTINGS
48
    kafka_broker_list = 'localhost:9092',
49
    kafka_topic_list = 'flow-messages',
50
    kafka_group_name = 'clickhouse_netflow_consumer',
51
    kafka_format = 'JSONEachRow',
52
    kafka_num_consumers = 3,
53
    kafka_max_block_size = 65536;

创建netflow.flows_mv#

此表作为物化视图，监听flows_kafka，将数据自动转存到flows表，相当于数据传输的管道。

1
-- 创建物化视图，把flows_kafka拉取的数据自动转存到 netflow.flows
2
CREATE MATERIALIZED VIEW netflow.flows_mv TO netflow.flows
3
(
4
    `type` String,
5
    `time_received_ns` UInt64,
6
    `sequence_num` UInt32,
7
    `sampling_rate` UInt32,
8
    `sampler_address` String,
9
    `time_flow_start_ns` UInt64,
10
    `time_flow_end_ns` UInt64,
11
    `bytes` UInt64,
12
    `packets` UInt64,
13
    `src_addr` String,
14
    `dst_addr` String,
15
    `etype` String,
16
    `proto` String,
17
    `src_port` UInt16,
18
    `dst_port` UInt16,
19
    `in_if` UInt32,
20
    `out_if` UInt32,
21
    `src_mac` String,
22
    `dst_mac` String,
23
    `src_vlan` UInt16,
24
    `dst_vlan` UInt16,
25
    `vlan_id` UInt16,
26
    `ip_tos` UInt8,
27
    `forwarding_status` UInt8,
28
    `ip_ttl` UInt8,
29
    `ip_flags` UInt8,
30
    `tcp_flags` UInt8,
31
    `icmp_type` UInt8,
32
    `icmp_code` UInt8,
33
    `ipv6_flow_label` UInt32,
34
    `fragment_id` UInt32,
35
    `fragment_offset` UInt16,
36
    `src_as` UInt32,
37
    `dst_as` UInt32,
38
    `next_hop` String,
39
    `next_hop_as` UInt32,
40
    `src_net` String,
41
    `dst_net` String,
42
    `bgp_next_hop` String,
43
    `observation_domain_id` UInt32,
44
    `observation_point_id` UInt32
45
)
46
AS SELECT *
47
FROM netflow.flows_kafka;

配置完成后检查下clickhouse能否从kafka拉取数据

1
-- 检查数据是否为0
2
SELECT count() FROM netflow.flows;
3

4
-- 检查最近的几条数据
5
SELECT *
6
FROM netflow.flows
7
ORDER BY timestamp DESC
8
LIMIT 5;

正常来说你应该能看到

1
goflow :) SELECT count() FROM netflow.flows;
2

3
SELECT count()
4
FROM netflow.flows
5

6
Query id: 357f75ee-5db4-4e9c-bc9a-141fe0674240
7

8
   ┌─count()─┐
9
1. │    1300 │
10
   └─────────┘
11

12
1 row in set. Elapsed: 0.003 sec.
13

14
goflow :) SELECT *
15
FROM netflow.flows
16
ORDER BY timestamp DESC
17
LIMIT 5;
18

19
Query id: 666e4e49-1a34-4795-b577-2347e4ca7534
20

21
Row 1:
22
──────
23
type:                  SFLOW_5
24
time_received_ns:      1768289585360343652
25
sequence_num:          1092
26
sampling_rate:         2000
27
sampler_address:       192.168.1.201
28
.....
29
timestamp:             2026-01-13 07:33:05
30
.....

安装grafana#

1
sudo apt-get install -y apt-transport-https wget
2
sudo mkdir -p /etc/apt/keyrings/
3
wget -q -O - https://apt.grafana.com/gpg.key | gpg --dearmor | sudo tee /etc/apt/keyrings/grafana.gpg > /dev/null
4
# Updates the list of available packages
5
sudo apt-get update
6
# Installs the latest OSS release:
7
sudo apt-get install grafana
8
#好欸，又能摸鱼半小时

启动grafana服务#

1
systemctl daemon-reload
2
systemctl enable grafana-server.service
3
systemctl start grafana-server.service

安装插件#

访问http://<ip>:3000即可打开grafana 默认账号密码均为admin 进入平台后选择administration>Plugins and data>Plugins搜索clickhouse alt text 点击Install进行安装

添加数据源#

安装完成后点击Add new data source链接到数据库

1
Server address
2
127.0.0.1
3
Server port
4
9000

其他保持默认，点击Save & test 不出意外的话会提示

1
Data source is working
2
Next, you can start to visualize data by building a dashboard , or by querying data in the Explore view .

然后刷新一下界面，在Default DB and table Default database栏内输入netflow，再次点击点击Save & test就连接完成了

新建监控面板#

点击Dashboards>Create dashboard>Add visualization数据源选择刚刚配置的clickhouse，修改下方A栏目内的Editor Type为SQL Editor alt text 在下方代码框内输入

1
SELECT
2
    src_addr AS source_ip,
3
    dst_addr AS dest_ip,
4
    formatReadableSize(sum(bytes) * 2000) AS total_traffic, --根据采集比例自行调整
5
    sum(packets) * 2000 AS total_packets,                   --根据采集比例自行调整
6
    min(timestamp) AS first_seen,
7
    max(timestamp) AS last_seen,
8
    dateDiff('second', min(timestamp), max(timestamp)) AS duration_seconds,
9
    count() AS flow_count
10
FROM netflow.flows
11
WHERE $__timeFilter(timestamp)
12
GROUP BY src_addr, dst_addr
13
ORDER BY sum(bytes) DESC
14
LIMIT 100

点击代码框右上方Run Query就出图啦，因为我们是要看详细信息，所以说在右侧栏内，将Time series修改为Table就可以了 alt text 改成table后：点击右上方save dashboard，按需修改下内容保存退出可在退出后的界面手动调整图标大小

最终效果#

alt text

先挖个坑在这儿：后续更新：对接ip数据库，显示ip地址地理信息