kafka

README

kafka

kafka 监控采集插件，由kafka-exporter（https://github.com/davidmparrott/kafka_exporter）封装而来。

Configuration

请参考配置示例

监控大盘和告警规则

同级目录下的 dashboard.json、alerts.json 可以直接导入夜莺使用。

开源kafka-exporter 兼容说明

categraf的exporter 封装 https://github.com/davidmparrott/kafka_exporter (以下简称davidmparrott版本) davidmparrott版本 fork自 https://github.com/danielqsj/kafka_exporter (以下简称danielqsj版本)

danielqsj版本作为原始版本, github版本也相对活跃, prometheus生态使用较多
categraf kafka plugin基于davidmparrott版本, 以下配置可以对danielqsj版本做一些兼容

1. 增加metric: kafka_broker_info davidmparrott版本无此metric, 默认以增加, 无需配置
2. davidmparrott版本与danielqsj版本, 有以下metric名字不同:

   davidmparrott版本	danielqsj版本
   kafka_consumergroup_uncommit_offsets	kafka_consumergroup_lag
   kafka_consumergroup_uncommit_offsets_sum	kafka_consumergroup_lag_sum
   kafka_consumergroup_uncommitted_offsets_zookeeper	kafka_consumergroup_lag_zookeeper

如果想使用danielqsj版本的metric, 在kafka instance 配置文件中进行如下配置:

rename_uncommit_offset_to_lag = true

1. davidmparrott版本 比 danielqsj版本多以下metric 以下metric是对延迟速率进行了计算

• kafka_consumer_lag_millis
• kafka_consumer_lag_interpolation
• kafka_consumer_lag_extrapolation

计算在内存一map, 记录每个partition的offset时序数据, 时序数据的数量配置文件max_offsets来控制 较大的kafka集群会建议关闭此功能, 占用内存较多, 计算速率可以使用promeql rate进行计算. 可以在kafka instance 配置文件中进行如下配置:

disable_calculate_lag_rate = true

1. 增加参数配置offset_show_all, 默认为true, 采集所有consumer group. 配置为false的话仅采集connected consumer groups davidmparrott版本无此配置.

Documentation

Index

• type Instance
  • func (ins *Instance) Gather(slist *types.SampleList)
  • func (ins *Instance) Init() error
• type Kafka
  • func (r *Kafka) Clone() inputs.Input
  • func (r *Kafka) Drop()
  • func (r *Kafka) GetInstances() []inputs.Instance
  • func (r *Kafka) Name() string

Types

type Instance

type Instance struct {
	config.InstanceConfig

	LogLevel string `toml:"log_level"`

	// Address (host:port) of Kafka server.
	KafkaURIs []string `toml:"kafka_uris,omitempty"`

	// Connect using SASL/PLAIN
	UseSASL bool `toml:"use_sasl,omitempty"`

	// Only set this to false if using a non-Kafka SASL proxy
	UseSASLHandshake *bool `toml:"use_sasl_handshake,omitempty"`

	// SASL user name
	SASLUsername string `toml:"sasl_username,omitempty"`

	// SASL user password
	SASLPassword string `toml:"sasl_password,omitempty"`

	// The SASL SCRAM SHA algorithm sha256 or sha512 as mechanism
	SASLMechanism string `toml:"sasl_mechanism,omitempty"`

	// Connect using TLS
	UseTLS bool `toml:"use_tls,omitempty"`

	// The optional certificate authority file for TLS client authentication
	CAFile string `toml:"ca_file,omitempty"`

	// The optional certificate file for TLS client authentication
	CertFile string `toml:"cert_file,omitempty"`

	// The optional key file for TLS client authentication
	KeyFile string `toml:"key_file,omitempty"`

	// If true, the server's certificate will not be checked for validity. This will make your HTTPS connections insecure
	InsecureSkipVerify bool `toml:"insecure_skip_verify,omitempty"`

	// Kafka broker version
	KafkaVersion string `toml:"kafka_version,omitempty"`

	// if you need to use a group from zookeeper
	UseZooKeeperLag bool `toml:"use_zookeeper_lag,omitempty"`

	// Address array (hosts) of zookeeper server.
	ZookeeperURIs []string `toml:"zookeeper_uris,omitempty"`

	// Metadata refresh interval
	MetadataRefreshInterval string `toml:"metadata_refresh_interval,omitempty"`

	// Whether show the offset/lag for all consumer group, otherwise, only show connected consumer groups, default is true
	OffsetShowAll *bool `toml:"offset_show_all,omitempty"`

	// If true, all scrapes will trigger kafka operations otherwise, they will share results. WARN: This should be disabled on large clusters
	AllowConcurrent *bool `toml:"allow_concurrency,omitempty"`

	// Maximum number of offsets to store in the interpolation table for a partition
	MaxOffsets int `toml:"max_offsets,omitempty"`

	// How frequently should the interpolation table be pruned, in seconds
	PruneIntervalSeconds int `toml:"prune_interval_seconds,omitempty"`

	// Regex filter for topics to be monitored
	TopicsFilter string `toml:"topics_filter_regex,omitempty"`

	// Regex filter for consumer groups to be monitored
	GroupFilter string `toml:"groups_filter_regex,omitempty"`

	// rename metric: kafka_consumergroup_uncommitted_offsets to kafka_consumergroup_lag
	RenameUncommitOffsetsToLag bool `toml:"rename_uncommit_offset_to_lag,omitempty"`
	// disable calculate lag rate
	DisableCalculateLagRate bool `toml:"disable_calculate_lag_rate,omitempty"`
	// contains filtered or unexported fields
}

func (*Instance) Gather

func (ins *Instance) Gather(slist *types.SampleList)

func (*Instance) Init

func (ins *Instance) Init() error

type Kafka

type Kafka struct {
	config.PluginConfig
	Instances []*Instance `toml:"instances"`
}

func (*Kafka) Clone

func (r *Kafka) Clone() inputs.Input

func (*Kafka) Drop

func (r *Kafka) Drop()

func (*Kafka) GetInstances

func (r *Kafka) GetInstances() []inputs.Instance

func (*Kafka) Name

func (r *Kafka) Name() string