consumer

Documentation

Index

• type AllowLicense
• type BlockOrganization
• type FlowData
• type KafkaConsumer
  • func NewKafkaConsumer(config KakfaConsumerConfig) (kc *KafkaConsumer, err error)
  • func (kc *KafkaConsumer) Close()
  • func (kc *KafkaConsumer) ConsumeLimits() (chan Message, chan string)
  • func (kc *KafkaConsumer) ConsumeNetflow() (chan FlowData, chan string)
• type KakfaConsumerConfig
• type Message
• type NetflowConsumer
• type RdKafkaConsumer
• type ResetSensors

Types

type AllowLicense

type AllowLicense struct {
	License string
}

AllowLicense reports a valid license.

type BlockOrganization

type BlockOrganization string

BlockOrganization identifies the organization that reached the limit

type FlowData

type FlowData struct {
	IP   uint32
	Data []byte
}

FlowData contains the IP address of the Netflow exporter and the flow itself

type KafkaConsumer

type KafkaConsumer struct {
	KakfaConsumerConfig
	// contains filtered or unexported fields
}

KafkaConsumer implements "Consumer" and consumes messages from a Kafka broker

func NewKafkaConsumer

func NewKafkaConsumer(config KakfaConsumerConfig) (kc *KafkaConsumer, err error)

NewKafkaConsumer creates a new instance of a Kafka consumer and subscribes to the provided topics

func (*KafkaConsumer) Close

func (kc *KafkaConsumer) Close()

Close terminates the rdkafka consumer

func (*KafkaConsumer) ConsumeLimits

func (kc *KafkaConsumer) ConsumeLimits() (chan Message, chan string)

ConsumeLimits receives limits messages from the kafka broker. "messages" channel receives actual messages and "info" channel receives notifications from the Kafka broker.

• "limit_reached": All sensors belonging to an organization are blocked.
• "allowed_licenses": All sensors are blocked and the only the sensors with a valid license are allowed.

func (*KafkaConsumer) ConsumeNetflow

func (kc *KafkaConsumer) ConsumeNetflow() (chan FlowData, chan string)

ConsumeNetflow receives netflow from the kafka broker. "messages" channel receives actual messages and "info" channel receives notifications from the Kafka broker.

type KakfaConsumerConfig

type KakfaConsumerConfig struct {
	NetflowConsumer RdKafkaConsumer
	LimitsConsumer  RdKafkaConsumer
	NetflowTopics   []string
	LimitsTopics    []string
}

KakfaConsumerConfig contains the configuration for a Kafka Consumer.

type Message

type Message interface{}

Message can be either an UUID to be blocked or a ResetSignal

type NetflowConsumer

type NetflowConsumer interface {
	ConsumeNetflow() (messages chan FlowData, events chan string)
	ConsumeLimits() (messages chan Message, events chan string)
}

NetflowConsumer gets an IP address and Netflow data from a resource

type RdKafkaConsumer

type RdKafkaConsumer interface {
	SubscribeTopics([]string, kafka.RebalanceCb) error
	Events() chan kafka.Event
	Assign(partitions []kafka.TopicPartition) error
	Unassign() error
	Close() error
}

RdKafkaConsumer is an interface for rdkafka consumer. Used for mocking purposes.

type ResetSensors

type ResetSensors struct{}

ResetSensors notifies that sensors from a given organization should be unblocked.