utils

Documentation

Index

• Variables
• func DefaultAccountCallback(name string, id int, start, end time.Time)
• func GetServiceAddresses(srv string) (addrs []string, err error)
• func UDPRoutine(name string, decodeFunc decoder.DecoderFunc, workers int, addr string, ...) error
• func UDPStoppableRoutine(stopCh <-chan struct{}, name string, decodeFunc decoder.DecoderFunc, ...) error
• type BaseMessage
• type DefaultErrorCallback
  • func (cb *DefaultErrorCallback) Callback(name string, id int, start, end time.Time, err error)
• type Formatter
• type Logger
• type ProducerConfig
  • func LoadMapping(f io.Reader) (ProducerConfig, error)
• type StateNFLegacy
  • func NewStateNFLegacy() *StateNFLegacy
  • func (s *StateNFLegacy) DecodeFlow(msg interface{}) error
  • func (s *StateNFLegacy) FlowRoutine(workers int, addr string, port int, reuseport bool) error
  • func (s *StateNFLegacy) Shutdown()
• type StateNetFlow
  • func NewStateNetFlow() *StateNetFlow
  • func (s *StateNetFlow) DecodeFlow(msg interface{}) error
  • func (s *StateNetFlow) FlowRoutine(workers int, addr string, port int, reuseport bool) error
  • func (s *StateNetFlow) Shutdown()
• type StateSFlow
  • func NewStateSFlow() *StateSFlow
  • func (s *StateSFlow) DecodeFlow(msg interface{}) error
  • func (s *StateSFlow) FlowRoutine(workers int, addr string, port int, reuseport bool) error
  • func (s *StateSFlow) Shutdown()
• type Transport

Variables

var (
	MetricTrafficBytes = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name: "flow_traffic_bytes",
			Help: "Bytes received by the application.",
		},
		[]string{"remote_ip", "local_ip", "local_port", "type"},
	)
	MetricTrafficPackets = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name: "flow_traffic_packets",
			Help: "Packets received by the application.",
		},
		[]string{"remote_ip", "local_ip", "local_port", "type"},
	)
	MetricPacketSizeSum = prometheus.NewSummaryVec(
		prometheus.SummaryOpts{
			Name:       "flow_traffic_summary_size_bytes",
			Help:       "Summary of packet size.",
			Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001},
		},
		[]string{"remote_ip", "local_ip", "local_port", "type"},
	)
	DecoderStats = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name: "flow_decoder_count",
			Help: "Decoder processed count.",
		},
		[]string{"worker", "name"},
	)
	DecoderErrors = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name: "flow_decoder_error_count",
			Help: "Decoder processed error count.",
		},
		[]string{"worker", "name"},
	)
	DecoderTime = prometheus.NewSummaryVec(
		prometheus.SummaryOpts{
			Name:       "flow_summary_decoding_time_us",
			Help:       "Decoding time summary.",
			Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001},
		},
		[]string{"name"},
	)
	DecoderProcessTime = prometheus.NewSummaryVec(
		prometheus.SummaryOpts{
			Name:       "flow_summary_processing_time_us",
			Help:       "Processing time summary.",
			Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001},
		},
		[]string{"name"},
	)
	NetFlowStats = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name: "flow_process_nf_count",
			Help: "NetFlows processed.",
		},
		[]string{"router", "version"},
	)
	NetFlowErrors = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name: "flow_process_nf_errors_count",
			Help: "NetFlows processed errors.",
		},
		[]string{"router", "error"},
	)
	NetFlowSetRecordsStatsSum = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name: "flow_process_nf_flowset_records_sum",
			Help: "NetFlows FlowSets sum of records.",
		},
		[]string{"router", "version", "type"},
	)
	NetFlowSetStatsSum = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name: "flow_process_nf_flowset_sum",
			Help: "NetFlows FlowSets sum.",
		},
		[]string{"router", "version", "type"},
	)
	NetFlowTimeStatsSum = prometheus.NewSummaryVec(
		prometheus.SummaryOpts{
			Name:       "flow_process_nf_delay_summary_seconds",
			Help:       "NetFlows time difference between time of flow and processing.",
			Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001},
		},
		[]string{"router", "version"},
	)
	NetFlowTemplatesStats = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name: "flow_process_nf_templates_count",
			Help: "NetFlows Template count.",
		},
		[]string{"router", "version", "obs_domain_id", "template_id", "type"},
	)
	SFlowStats = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name: "flow_process_sf_count",
			Help: "sFlows processed.",
		},
		[]string{"router", "agent", "version"},
	)
	SFlowErrors = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name: "flow_process_sf_errors_count",
			Help: "sFlows processed errors.",
		},
		[]string{"router", "error"},
	)
	SFlowSampleStatsSum = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name: "flow_process_sf_samples_sum",
			Help: "SFlows samples sum.",
		},
		[]string{"router", "agent", "version", "type"},
	)
	SFlowSampleRecordsStatsSum = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name: "flow_process_sf_samples_records_sum",
			Help: "SFlows samples sum of records.",
		},
		[]string{"router", "agent", "version", "type"},
	)
)

var ErrAlreadyStarted = errors.New("the routine is already started")

ErrAlreadyStarted error happens when you try to start twice a flow routine

Functions

func DefaultAccountCallback

func DefaultAccountCallback(name string, id int, start, end time.Time)

func GetServiceAddresses

func GetServiceAddresses(srv string) (addrs []string, err error)

func UDPRoutine

func UDPRoutine(name string, decodeFunc decoder.DecoderFunc, workers int, addr string, port int, sockReuse bool, logger Logger) error

func UDPStoppableRoutine

func UDPStoppableRoutine(stopCh <-chan struct{}, name string, decodeFunc decoder.DecoderFunc, workers int, addr string, port int, sockReuse bool, logger Logger) error

UDPStoppableRoutine runs a UDPRoutine that can be stopped by closing the stopCh passed as argument

Types

type BaseMessage

type BaseMessage struct {
	Src     net.IP
	Port    int
	Payload []byte

	SetTime  bool
	RecvTime time.Time
}

type DefaultErrorCallback

type DefaultErrorCallback struct {
	Logger Logger
}

type DefaultLogTransport struct { }

func (s *DefaultLogTransport) Publish(msgs []*flowmessage.FlowMessage) {
	for _, msg := range msgs {
		fmt.Printf("%v\n", FlowMessageToString(msg))
	}
}

type DefaultJSONTransport struct { }

func (s *DefaultJSONTransport) Publish(msgs []*flowmessage.FlowMessage) {
	for _, msg := range msgs {
		fmt.Printf("%v\n", FlowMessageToJSON(msg))
	}
}

func (*DefaultErrorCallback) Callback

func (cb *DefaultErrorCallback) Callback(name string, id int, start, end time.Time, err error)

type Formatter

type Formatter interface {
	Format([]*flowmessage.FlowMessage)
}

type Logger

type Logger interface {
	Printf(string, ...interface{})
	Errorf(string, ...interface{})
	Warnf(string, ...interface{})
	Warn(...interface{})
	Error(...interface{})
	Debug(...interface{})
	Debugf(string, ...interface{})
	Infof(string, ...interface{})
	Fatalf(string, ...interface{})
}

type ProducerConfig

type ProducerConfig *producer.ProducerConfig

func LoadMapping

func LoadMapping(f io.Reader) (ProducerConfig, error)

type StateNFLegacy

type StateNFLegacy struct {
	Format    format.FormatInterface
	Transport transport.TransportInterface
	Logger    Logger
	// contains filtered or unexported fields
}

func NewStateNFLegacy

func NewStateNFLegacy() *StateNFLegacy

func (*StateNFLegacy) DecodeFlow

func (s *StateNFLegacy) DecodeFlow(msg interface{}) error

func (*StateNFLegacy) FlowRoutine

func (s *StateNFLegacy) FlowRoutine(workers int, addr string, port int, reuseport bool) error

func (*StateNFLegacy) Shutdown

func (s *StateNFLegacy) Shutdown()

type StateNetFlow

type StateNetFlow struct {
	Format    format.FormatInterface
	Transport transport.TransportInterface
	Logger    Logger

	Config *producer.ProducerConfig

	TemplateSystem templates.TemplateInterface
	// contains filtered or unexported fields
}

func NewStateNetFlow

func NewStateNetFlow() *StateNetFlow

func (*StateNetFlow) DecodeFlow

func (s *StateNetFlow) DecodeFlow(msg interface{}) error

func (*StateNetFlow) FlowRoutine

func (s *StateNetFlow) FlowRoutine(workers int, addr string, port int, reuseport bool) error

func (*StateNetFlow) Shutdown

func (s *StateNetFlow) Shutdown()

type StateSFlow

type StateSFlow struct {
	Format    format.FormatInterface
	Transport transport.TransportInterface
	Logger    Logger

	Config *producer.ProducerConfig
	// contains filtered or unexported fields
}

func NewStateSFlow

func NewStateSFlow() *StateSFlow

func (*StateSFlow) DecodeFlow

func (s *StateSFlow) DecodeFlow(msg interface{}) error

func (*StateSFlow) FlowRoutine

func (s *StateSFlow) FlowRoutine(workers int, addr string, port int, reuseport bool) error

func (*StateSFlow) Shutdown

func (s *StateSFlow) Shutdown()

type Transport

type Transport interface {
	Send([]*flowmessage.FlowMessage)
}