Add flag to enable/disable sending messages to producer queue

This is useful for users that are not interested on sending
flow data to message queue and they simply consume the output
from console with verbose flag set, i.e.:

vflow -producer-enabled=false -verbose

If not provided, it will just work as is now.

vflow/ipfix.go

@@ -129,6 +129,10 @@ func (i *IPFIX) run() {
 	go mirrorIPFIXDispatcher(ipfixMCh)
 
 	go func() {
+		if !opts.ProducerEnabled {
+			logger.Println("Producer message queue disabled")
+			return
+		}
 		p := producer.NewProducer(opts.MQName)
 
 		p.MQConfigFile = path.Join(opts.VFlowConfigPath, opts.MQConfigFile)

vflow/netflow_v5.go

@@ -109,6 +109,10 @@ func (i *NetflowV5) run() {
 	logger.Printf("netflow v5 is running (UDP: listening on [::]:%d workers#: %d)", i.port, i.workers)
 
 	go func() {
+		if !opts.ProducerEnabled {
+			logger.Println("Producer message queue disabled")
+			return
+		}
 		p := producer.NewProducer(opts.MQName)
 
 		p.MQConfigFile = path.Join(opts.VFlowConfigPath, opts.MQConfigFile)

vflow/netflow_v9.go

@@ -114,6 +114,10 @@ func (i *NetflowV9) run() {
 	mCacheNF9 = netflow9.GetCache(opts.NetflowV9TplCacheFile)
 
 	go func() {
+		if !opts.ProducerEnabled {
+			logger.Println("Producer message queue disabled")
+			return
+		}
 		p := producer.NewProducer(opts.MQName)
 
 		p.MQConfigFile = path.Join(opts.VFlowConfigPath, opts.MQConfigFile)

vflow/options.go

@@ -102,8 +102,9 @@ type Options struct {
 	NetflowV9TplCacheFile string `yaml:"netflow9-tpl-cache-file"`
 
 	// producer
-	MQName       string `yaml:"mq-name"`
-	MQConfigFile string `yaml:"mq-config-file"`
+	ProducerEnabled bool   `yaml:producer-enabled"`
+	MQName          string `yaml:"mq-name"`
+	MQConfigFile    string `yaml:"mq-config-file"`
 
 	VFlowConfigPath string
 }
@@ -180,8 +181,9 @@ func NewOptions() *Options {
 		NetflowV9Topic:        "vflow.netflow9",
 		NetflowV9TplCacheFile: "/tmp/netflowv9.templates",
 
-		MQName:       "kafka",
-		MQConfigFile: "mq.conf",
+		ProducerEnabled: true,
+		MQName:          "kafka",
+		MQConfigFile:    "mq.conf",
 
 		VFlowConfigPath: "/etc/vflow",
 	}
@@ -359,6 +361,7 @@ func (opts *Options) flagSet() {
 	flag.StringVar(&opts.NetflowV9TplCacheFile, "netflow9-tpl-cache-file", opts.NetflowV9TplCacheFile, "Netflow version 9 template cache file")
 
 	// producer options
+	flag.BoolVar(&opts.ProducerEnabled, "producer-enabled", opts.ProducerEnabled, "enable/disable producer message queue")
 	flag.StringVar(&opts.MQName, "mqueue", opts.MQName, "producer message queue name")
 	flag.StringVar(&opts.MQConfigFile, "mqueue-conf", opts.MQConfigFile, "producer message queue configuration file")
 

vflow/sflow.go

@@ -118,6 +118,10 @@ func (s *SFlow) run() {
 
 	go func() {
 		p := producer.NewProducer(opts.MQName)
+		if !opts.ProducerEnabled {
+			logger.Println("Producer message queue disabled")
+			return
+		}
 
 		p.MQConfigFile = path.Join(opts.VFlowConfigPath, opts.MQConfigFile)
 		p.MQErrorCount = &s.stats.MQErrorCount