flow的采集与分析vflow项目分析和解读

前言

vflow这个开源项目，是我一开始做flow这块的最开始参考的项目。虽然到最后，我们没有使用这个项目，但是该项目是一个企业级的处理netflow等协议的项目，有许多地方值得去研究。

特性

• PFIX RFC7011 collector
• sFLow v5 raw header packet collector
• Netflow v9 collector
• Decoding sFlow raw header L2/L3/L4
• Produce to Apache Kafka, NSQ, NATS
• Replicate IPFIX to 3rd party collector
• Supports IPv4 and IPv6
• Monitoring with InfluxDB and OpenTSDB backend

架构

Dynamic pool

提供了work pool，而且wokrer数量是根据buff堆积消息数量动态调整的，这点之前文章介绍过实现方式。

插件架构

vFlow 支持 Kafka,NSQ,NATS三种消息队列，但是你可以很容易地将消息写到其他的消息队列，比如 RabbitMQ。

package producer
import (
"log"
"sync"
)
// Producer represents messaging queue
type Producer struct {
MQ MQueue
MQConfigFile string
MQErrorCount *uint64
Topic string
Chan chan []byte
Logger *log.Logger
}
// MQueue represents messaging queue methods
type MQueue interface {
setup(string, *log.Logger) error
inputMsg(string, chan []byte, *uint64)
}
// NewProducer constructs new Messaging Queue
func NewProducer(mqName string) *Producer {
var mqRegistered = map[string]MQueue{
"kafka": new(Kafka),
"nsq": new(NSQ),
"nats": new(NATS),
"rawSocket": new(RawSocket),
}
return &Producer{
MQ: mqRegistered[mqName],
}
}
// Run configs and tries to be ready to produce
func (p *Producer) Run() error {
var (
wg sync.WaitGroup
err error
)
err = p.MQ.setup(p.MQConfigFile, p.Logger)
if err != nil {
return err
}
wg.Add(1)
go func() {
defer wg.Done()
topic := p.Topic
p.MQ.inputMsg(topic, p.Chan, p.MQErrorCount)
}()
wg.Wait()
return nil
}
// Shutdown stops the producer
func (p *Producer) Shutdown() {
close(p.Chan)
}


提供了接口，如果扩展到其他的消息队列，实现接口即可。

扩展支持更多的协议

vflow暂时支持sflow和netflow协议，对于cflow和netstream，我们可以扩展实现。我已经扩展了netstream的模块。在vflow.go中，加入指定的flow worker。如下：

//版本太老，不再支持
netflow5 := NewNetFlowV5()
netstream5 := NewNetStreamV5()
// 主要支持四大协议
sFlow := NewSFlow()
ipfix := NewIPFIX()
netflow9 := NewNetflowV9()
netstream9 := NewNetStreamV9()
protos[0] = sFlow
protos[1] = ipfix
protos[2] = netflow9
protos[3] = netstream9
protos[4] = netflow5
protos[5] = netstream5
for _, p := range protos {
wg.Add(1)
go func(p proto) {
defer wg.Done()
p.run()
}(p)
}

支持监控

在monitor文件夹下提供了监控相关，目前支持metrics导入到influxdb和opentsdb。
提供了完善的metrics指标。

硬件要求

总结

1）udp的负载均衡
这里nginx就支持udp协议的代理。
2）关于最终存放flow采集以后的数据，vflow提供了clickhouse和memsql以及spark三种思路。
3）当然我们之所以没有采用vflow，是因为我们编写了telegraf的flow input插件，可以充分利用telegraf提供的各种输出插件以及过滤器等。