forwarder

Documentation

Index

• func StartForwarders(wg *sync.WaitGroup, c []Config, messageChan <-chan snmp.Message)
• type Base
  • func NewBase(c Config, idx int) Base
  • func (b *Base) Close()
  • func (b *Base) Done() <-chan struct{}
  • func (b *Base) Send(message snmp.Message)
• type Config
  • func (c *Config) Type() string
• type Duration
  • func (d Duration) MarshalJSON() ([]byte, error)
  • func (d *Duration) UnmarshalText(b []byte) error
• type File
  • func (f *File) Run()
• type FileConfig
• type Forwarder
  • func NewFile(c Config, idx int) Forwarder
  • func NewKafka(c Config, idx int) Forwarder
  • func NewMQTT(c Config, idx int) Forwarder
  • func NewSNMPTrap(c Config, idx int) Forwarder
  • func NewZabbixTrapper(c Config, idx int) Forwarder
• type Kafka
  • func (k *Kafka) Run()
• type KafkaConfig
• type LookupResult
• type LookupStrategy
  • func (l *LookupStrategy) String() string
  • func (l *LookupStrategy) UnmarshalText(text []byte) error
• type MQTT
  • func (m *MQTT) Run()
• type MQTTConfig
• type ProxyConf
• type QueryResult
• type SNMPTrap
  • func (s *SNMPTrap) Run()
• type SNMPTrapConfig
• type TlsConfig
• type ZSAdvancedConfig
  • func (z *ZSAdvancedConfig) GetDSN() (string, string, error)
• type ZabbixLookup
  • func NewZabbixLookup(c *ZabbixTrapperConfig, logger zerolog.Logger, ctx context.Context) *ZabbixLookup
  • func (z *ZabbixLookup) Lookup(m snmp.Message, strategy LookupStrategy) (LookupResult, error)
  • func (z *ZabbixLookup) Refresh()
• type ZabbixTrapper
  • func (z *ZabbixTrapper) Run()
• type ZabbixTrapperConfig

Functions

func StartForwarders

func StartForwarders(wg *sync.WaitGroup, c []Config, messageChan <-chan snmp.Message)

Types

type Base

type Base struct {
	// contains filtered or unexported fields
}

func NewBase

func NewBase(c Config, idx int) Base

func (*Base) Close

func (b *Base) Close()

func (*Base) Done

func (b *Base) Done() <-chan struct{}

func (*Base) Send

func (b *Base) Send(message snmp.Message)

Send snmp message to forwarder TODO: requeue message if the forwarder fails to send the messages

type Config

type Config struct {
	// ID identifies forwarder name, also used for prometheus labelling
	ID string
	// QueueSize defines the size of queue of each forwarder, when queue is full (might be caused
	// by slow forwarder) the message is dropped
	QueueSize uint `mapstructure:"queue_size"`
	// AgentAddressObjectPrefix populates Message.AgentAddress
	AgentAddressObjectPrefix string `mapstructure:"agent_address_object_prefix"`
	// TimeFormat specifies golang time format for casting time related fields to string
	TimeFormat string `mapstructure:"time_format"`
	// TimeAsTimezone will cast any time field to specified timezone
	TimeAsTimezone string `mapstructure:"time_as_timezone"`
	// Filter, JSONFormat, ValueJSONFormat utilizes maja42/goval expressions
	Filter          string
	JSONFormat      string `mapstructure:"json_format"`
	ValueJSONFormat string `mapstructure:"value_json_format"`
	ValueJSONIsFlat bool   `mapstructure:"value_json_is_flat"`
	File            *FileConfig
	Kafka           *KafkaConfig
	MQTT            *MQTTConfig
	Trap            *SNMPTrapConfig
	ZabbixTrapper   *ZabbixTrapperConfig `mapstructure:"zabbix_trapper"`
}

func (*Config) Type

func (c *Config) Type() string

type Duration

type Duration struct {
	time.Duration
}

func (Duration) MarshalJSON

func (d Duration) MarshalJSON() ([]byte, error)

func (*Duration) UnmarshalText

func (d *Duration) UnmarshalText(b []byte) error

type File

type File struct {
	Base
}

func (*File) Run

func (f *File) Run()

type FileConfig

type FileConfig struct {
	// Path for json log. Make sure the user has sufficient permission to write
	Path string
}

type Forwarder

type Forwarder interface {
	// Send will send the trap message to its corresponding forwarder.
	// Does nothing if the queue buffer is full or forwarder is already closed
	Send(message snmp.Message)
	// Close informs the forwarder to stop processing any new messages
	Close()
	// Done informs the caller if forwarder is done processing
	Done() <-chan struct{}
}

func NewFile

func NewFile(c Config, idx int) Forwarder

func NewKafka

func NewKafka(c Config, idx int) Forwarder

func NewMQTT

func NewMQTT(c Config, idx int) Forwarder

func NewSNMPTrap

func NewSNMPTrap(c Config, idx int) Forwarder

func NewZabbixTrapper

func NewZabbixTrapper(c Config, idx int) Forwarder

type Kafka

type Kafka struct {
	Base
}

func (*Kafka) Run

func (k *Kafka) Run()

type KafkaConfig

type KafkaConfig struct {
	RequiredAcks kafka.RequiredAcks `mapstructure:"required_acks"`
	KeyField     string             `mapstructure:"key_field"`
	Hosts        []string
	Topic        string
}

type LookupResult

type LookupResult struct {
	Server   *ProxyConf
	Hostname string
}

type LookupStrategy

type LookupStrategy int8

const (
	// LookupFromAgentAddress will use the agent address as hostname, or search for interface
	// ip/dns if db_url config is specified
	LookupFromAgentAddress LookupStrategy = iota
	// LookupFromSourceAddress will use the source address as hostname, or search for interface
	// ip/dns if db_url config is specified
	LookupFromSourceAddress
	// LookupFromOID will use the value of a given OID as the hostname. If db_url config is specified,
	// this will check for host existence before sending, and send it using default_hostname if it's not found
	LookupFromOID
)

func (*LookupStrategy) String

func (l *LookupStrategy) String() string

func (*LookupStrategy) UnmarshalText

func (l *LookupStrategy) UnmarshalText(text []byte) error

type MQTT

type MQTT struct {
	Base
}

func (*MQTT) Run

func (m *MQTT) Run()

type MQTTConfig

type MQTTConfig struct {
	Hosts    []string
	ClientID string `mapstructure:"client_id"`
	Username string
	Password string
	Ordered  *bool
	TLS      *TlsConfig
	Topic    string
	Qos      uint8
}

type ProxyConf

type ProxyConf struct {
	Hostname string
	Address  string
	Port     int
}

ProxyConf is the list of available proxies in a zabbix system. In case of HA zabbix server, you need to include it here with its HANodeName

type QueryResult

type QueryResult struct {
	IPOrDNS       string         `db:"ip_or_dns"`
	Hostname      string         `db:"hostname"`
	ProxyHostname sql.NullString `db:"proxy_hostname"`
}

type SNMPTrap

type SNMPTrap struct {
	Base
	// contains filtered or unexported fields
}

func (*SNMPTrap) Run

func (s *SNMPTrap) Run()

type SNMPTrapConfig

type SNMPTrapConfig struct {
	Workers      int
	EnableInform bool `mapstructure:"enable_inform"`
	Host         string
	Version      string
	Community    string
	Context      string
	User         snmp.User
}

type TlsConfig

type TlsConfig struct {
	InsecureSkipVerify bool `mapstructure:"insecure_skip_verify"`
}

type ZSAdvancedConfig

type ZSAdvancedConfig struct {
	Proxies []ProxyConf

	// for example:
	// - postgres://user:pass@127.0.0.1:5432/dbname?param1=value1&param2=value2
	// - mysql://user:pass@127.0.0.1:3306/dbname?param1=value1&param2=value2
	DBUrl             string   `mapstructure:"db_url"`
	DBRefreshInterval Duration `mapstructure:"db_refresh_interval"`
	DBQueryTimeout    Duration `mapstructure:"db_query_timeout"`
	// contains filtered or unexported fields
}

func (*ZSAdvancedConfig) GetDSN

func (z *ZSAdvancedConfig) GetDSN() (string, string, error)

type ZabbixLookup

type ZabbixLookup struct {
	// contains filtered or unexported fields
}

func NewZabbixLookup

func NewZabbixLookup(
	c *ZabbixTrapperConfig,
	logger zerolog.Logger,
	ctx context.Context,
) *ZabbixLookup

func (*ZabbixLookup) Lookup

func (z *ZabbixLookup) Lookup(m snmp.Message, strategy LookupStrategy) (LookupResult, error)

func (*ZabbixLookup) Refresh

func (z *ZabbixLookup) Refresh()

type ZabbixTrapper

type ZabbixTrapper struct {
	Base
	// contains filtered or unexported fields
}

func (*ZabbixTrapper) Run

func (z *ZabbixTrapper) Run()

type ZabbixTrapperConfig

type ZabbixTrapperConfig struct {
	// default_* is used whenever host lookup fails:
	// - no advanced config defined
	// - proxy is defined in zabbix, but not defined in configuration
	// - can't find monitored hostname
	// - lookup strategy fails
	// DefaultAddress and DefaultPort are also used in case the host
	// is monitored directly with zabbix server and zabbix server
	// is not configured as HA
	DefaultAddress         string         `mapstructure:"default_address"`
	DefaultPort            int            `mapstructure:"default_port"`
	DefaultHostname        string         `mapstructure:"default_hostname"`
	ItemKey                string         `mapstructure:"item_key"`
	HostnameLookupStrategy LookupStrategy `mapstructure:"hostname_lookup_strategy"`
	OIDLookup              string         `mapstructure:"oid_lookup"`
	// Advanced config is for systems with proxies setup
	Advanced *ZSAdvancedConfig
}