Method run

vflow/sflow.go:90–158

()

}

func (s *SFlow) run() {
	var err error
	// exit if the sflow is disabled
	if !opts.SFlowEnabled {
		logger.Println("sflow has been disabled")
		return
	}

	s.pool = make(chan chan struct{}, maxWorkers)

	hostPort := net.JoinHostPort(s.addr, strconv.Itoa(s.port))
	udpAddr, _ := net.ResolveUDPAddr("udp", hostPort)

	s.conn, err = net.ListenUDP("udp", udpAddr)
	if err != nil {
		logger.Fatal(err)
	}

	atomic.AddInt32(&s.stats.Workers, int32(s.workers))
	for i := 0; i < s.workers; i++ {
		go func() {
			wQuit := make(chan struct{})
			s.pool <- wQuit
			s.sFlowWorker(wQuit)
		}()
	}

	go mirrorSFlowDispatcher(sFlowMCh)

	logger.Printf("sFlow is running (UDP: listening on [::]:%d workers#: %d)", s.port, s.workers)

	go func() {
		if !opts.ProducerEnabled {
			return
		}

		p := producer.NewProducer(opts.MQName)
		p.MQConfigFile = path.Join(opts.VFlowConfigPath, opts.MQConfigFile)
		p.MQErrorCount = &s.stats.MQErrorCount
		p.Logger = logger
		p.Chan = sFlowMQCh
		p.Topic = opts.SFlowTopic

		if err := p.Run(); err != nil {
			logger.Fatal(err)
		}
	}()

	go func() {
		if !opts.DynWorkers {
			logger.Println("sFlow dynamic worker disabled")
			return
		}

		s.dynWorkers()
	}()

	for !s.stop {