kafka

Documentation

Overview

Package kafka is a Metricbeat module that contains MetricSets.

Kafka is organised as following

- Topic - Partition - Producer - Consumer - Consumer Groups - Broker

Notes - Topics has a list of partitions - Each partition has an offset - Topic can be across brokers - Each broker has a list of partitions

Index

• func AssetKafka() string
• func Version(version string) kafka.Version
• type Broker
  • func NewBroker(host string, settings BrokerSettings) *Broker
  • func (b *Broker) Addr() string
  • func (b *Broker) AdvertisedAddr() string
  • func (b *Broker) Close() error
  • func (b *Broker) Connect() error
  • func (b *Broker) DescribeGroups(queryGroups []string) (map[string]GroupDescription, error)
  • func (b *Broker) FetchGroupOffsets(group string, partitions map[string][]int32) (*sarama.OffsetFetchResponse, error)
  • func (b *Broker) GetMetadata(topics ...string) (*sarama.MetadataResponse, error)
  • func (b *Broker) GetTopicsMetadata(topics ...string) ([]*sarama.TopicMetadata, error)
  • func (b *Broker) ID() int32
  • func (b *Broker) ListGroups() ([]string, error)
  • func (b *Broker) PartitionOffset(replicaID int32, topic string, partition int32, time int64) (int64, error)
• type BrokerSettings
• type GroupDescription
• type MemberDescription
• type MetricSet
  • func NewMetricSet(base mb.BaseMetricSet, options MetricSetOptions) (*MetricSet, error)
  • func (m *MetricSet) Connect() (*Broker, error)
• type MetricSetOptions
• type NetInfo

Functions

func AssetKafka

func AssetKafka() string

AssetKafka returns asset data. This is the base64 encoded gzipped contents of ../metricbeat/module/kafka.

func Version

func Version(version string) kafka.Version

Version returns a kafka version from its string representation

Types

type Broker

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

Broker provides functionality for communicating with a single kafka broker

func NewBroker

func NewBroker(host string, settings BrokerSettings) *Broker

NewBroker creates a new unconnected kafka Broker connection instance.

func (*Broker) Addr

func (b *Broker) Addr() string

Addr returns the configured broker endpoint.

func (*Broker) AdvertisedAddr

func (b *Broker) AdvertisedAddr() string

AdvertisedAddr returns the advertised broker address in case of matching broker has been found.

func (*Broker) Close

func (b *Broker) Close() error

Close the broker connection

func (*Broker) Connect

func (b *Broker) Connect() error

Connect connects the broker to the configured host

func (*Broker) DescribeGroups

func (b *Broker) DescribeGroups(
	queryGroups []string,
) (map[string]GroupDescription, error)

DescribeGroups fetches group details from broker.

func (*Broker) FetchGroupOffsets

func (b *Broker) FetchGroupOffsets(group string, partitions map[string][]int32) (*sarama.OffsetFetchResponse, error)

FetchGroupOffsets fetches the consume offset of group. The partitions is a MAP mapping from topic name to partitionid array.

func (*Broker) GetMetadata

func (b *Broker) GetMetadata(topics ...string) (*sarama.MetadataResponse, error)

GetMetadata fetches most recent cluster metadata from the broker.

func (*Broker) GetTopicsMetadata

func (b *Broker) GetTopicsMetadata(topics ...string) ([]*sarama.TopicMetadata, error)

GetTopicsMetadata fetches most recent topics/partition metadata from the broker.

func (*Broker) ID

func (b *Broker) ID() int32

ID returns the broker or -1 if the broker id is unknown.

func (*Broker) ListGroups

func (b *Broker) ListGroups() ([]string, error)

ListGroups lists all groups managed by the broker. Other consumer groups might be managed by other brokers.

func (*Broker) PartitionOffset

func (b *Broker) PartitionOffset(
	replicaID int32,
	topic string,
	partition int32,
	time int64,
) (int64, error)

PartitionOffset fetches the available offset from a partition.

type BrokerSettings

type BrokerSettings struct {
	MatchID                  bool
	DialTimeout, ReadTimeout time.Duration
	ClientID                 string
	Retries                  int
	Backoff                  time.Duration
	TLS                      *tls.Config
	Username, Password       string
	Version                  kafka.Version
}

BrokerSettings defines common configurations used when connecting to a broker

type GroupDescription

type GroupDescription struct {
	Members map[string]MemberDescription
}

type MemberDescription

type MemberDescription struct {
	Err        error
	ClientID   string
	ClientHost string
	Topics     map[string][]int32
}

type MetricSet

type MetricSet struct {
	mb.BaseMetricSet
	// contains filtered or unexported fields
}

MetricSet is the base metricset for all Kafka metricsets

func NewMetricSet

func NewMetricSet(base mb.BaseMetricSet, options MetricSetOptions) (*MetricSet, error)

NewMetricSet creates a base metricset for Kafka metricsets

func (*MetricSet) Connect

func (m *MetricSet) Connect() (*Broker, error)

Connect connects with a kafka broker

type MetricSetOptions

type MetricSetOptions struct {
	Version string
}

MetricSetOptions are the options of a Kafka metricset

type NetInfo

type NetInfo interface {
	LookupIP(string) ([]net.IP, error)
	LookupAddr(string) ([]string, error)
	LocalIPAddrs() ([]net.IP, error)
	Hostname() (string, error)
}

NetInfo can be used to obtain network information