kafka

Documentation

Index

• Constants
• type Duration
  • func (d *Duration) MarshalText() ([]byte, error)
  • func (d *Duration) UnmarshalText(text []byte) error
• type Kafka
  • func (rcv *Kafka) InitPrometheus(reg prometheus.Registerer)
  • func (rcv *Kafka) Stat(send helper.StatCallback)
  • func (rcv *Kafka) Stop()
• type Offset
  • func (o *Offset) MarshalText() ([]byte, error)
  • func (o *Offset) UnmarshalText(text []byte) error
• type Options
  • func NewOptions() *Options
• type Protocol
  • func (p *Protocol) MarshalText() ([]byte, error)
  • func (p *Protocol) ToString() string
  • func (p *Protocol) UnmarshalText(text []byte) error

Constants

const (
	// OffsetOldest represents oldest offset in Kafka
	OffsetOldest Offset = -1
	// OffsetNewest represents newest offset in Kafka
	OffsetNewest = -2

	// ProtocolPlain represents graphite line protocol
	ProtocolPlain Protocol = 0
	// ProtocolProtobuf represents protobuf messages
	ProtocolProtobuf = 1
	// ProtocolPickle represents pickled messages
	ProtocolPickle = 2
)

Types

type Duration

type Duration struct {
	time.Duration
}

Duration wrapper time.Duration for TOML

func (*Duration) MarshalText

func (d *Duration) MarshalText() ([]byte, error)

MarshalText encode text with TOML format

func (*Duration) UnmarshalText

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

UnmarshalText from TOML

type Kafka

type Kafka struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

Kafka receive metrics in protobuf or graphite line format from Kafka partitions

func (*Kafka) InitPrometheus

func (rcv *Kafka) InitPrometheus(reg prometheus.Registerer)

InitPrometheus is a stub for the receiver prom metrics. Required to satisfy Receiver interface.

func (*Kafka) Stat

func (rcv *Kafka) Stat(send helper.StatCallback)

Stat sends kafka receiver's internal stats to specified callback

func (*Kafka) Stop

func (rcv *Kafka) Stop()

Stop stops kafka receiver. It will return when all goroutines finish their work.

type Offset

type Offset int64

Offset is a special type to define kafka offsets. It's used to create custom marshal/unmarshal functions for configs.

func (*Offset) MarshalText

func (o *Offset) MarshalText() ([]byte, error)

MarshalText marshals offset. It's used to handle two special cases "newest" and "oldest".

func (*Offset) UnmarshalText

func (o *Offset) UnmarshalText(text []byte) error

UnmarshalText unmarshals text to offset. It handles "newest" and "oldest", oterwise fallbacks to time.ParseDuration.

type Options

type Options struct {
	Brokers           []string  `toml:"brokers"`
	Topic             string    `toml:"topic"`
	Partition         int32     `toml:"partition"`
	Protocol          Protocol  `toml:"parse-protocol"`
	StateFile         string    `toml:"state-file"`
	InitialOffset     Offset    `toml:"initial-offset"`
	StateSaveInterval *Duration `toml:"state-save-interval"`
	ReconnectInterval *Duration `toml:"reconnect-interval"`
	FetchInterval     *Duration `toml:"fetch-interval"`
	KafkaVersion      string    `toml:"kafka-version"`
}

Options contains all receiver's options that can be changed by user

func NewOptions

func NewOptions() *Options

NewOptions returns Options struct filled with default values.

type Protocol

type Protocol int

Protocol is a special type to allow user to define wire protocol in Config file as a simple text.

func (*Protocol) MarshalText

func (p *Protocol) MarshalText() ([]byte, error)

MarshalText converts internal enum-like representation of protocol to a text

func (*Protocol) ToString

func (p *Protocol) ToString() string

ToString returns text representation of current protocol

func (*Protocol) UnmarshalText

func (p *Protocol) UnmarshalText(text []byte) error

UnmarshalText converts text from config file to a enum.