stats

Documentation

Index

• type BrokerSource
  • func (s BrokerSource) Int64() int64
  • func (s *BrokerSource) UnmarshalJSON(data []byte) error
• type BrokerState
  • func (s BrokerState) Int64() int64
  • func (s *BrokerState) UnmarshalJSON(data []byte) error
• type BrokerStats
• type BrokerTopicPartition
• type ConsumerGroupJoinState
  • func (s ConsumerGroupJoinState) Int64() int64
  • func (s *ConsumerGroupJoinState) UnmarshalJSON(data []byte) error
• type ConsumerGroupState
  • func (s ConsumerGroupState) Int64() int64
  • func (s *ConsumerGroupState) UnmarshalJSON(data []byte) error
• type ConsumerGroupStats
• type Partition
• type Stats
• type TopicStats
• type WindowStats

Types

type BrokerSource

type BrokerSource string

const (
	BrokerSourceUnknown    BrokerSource = "unknown"
	BrokerSourceLearned    BrokerSource = "learned"
	BrokerSourceConfigured BrokerSource = "configured"
	BrokerSourceInternal   BrokerSource = "internal"
	BrokerSourceLogical    BrokerSource = "logical"
)

func (BrokerSource) Int64

func (s BrokerSource) Int64() int64

func (*BrokerSource) UnmarshalJSON

func (s *BrokerSource) UnmarshalJSON(data []byte) error

type BrokerState

type BrokerState string

const (
	BrokerStateUnknown         BrokerState = "unknown"
	BrokerStateInit            BrokerState = "init"
	BrokerStateDown            BrokerState = "down"
	BrokerStateConnect         BrokerState = "connect"
	BrokerStateAuth            BrokerState = "auth"
	BrokerStateApiVersionQuery BrokerState = "apiversion-query"
	BrokerStateAuthHandshake   BrokerState = "auth-handshake"
	BrokerStateUp              BrokerState = "up"
	BrokerStateUpdate          BrokerState = "update"
)

func (BrokerState) Int64

func (s BrokerState) Int64() int64

func (*BrokerState) UnmarshalJSON

func (s *BrokerState) UnmarshalJSON(data []byte) error

type BrokerStats

type BrokerStats struct {
	// Broker hostname, port and broker id
	Name string `json:"name"`
	// Broker id (-1 for bootstraps)
	NodeID int64 `json:"nodeid"`
	// Broker hostname
	NodeName string `json:"nodename"`
	// Broker source (learned, configured, internal, logical)
	Source BrokerSource `json:"source"`
	// Broker state (INIT, DOWN, CONNECT, AUTH, APIVERSION_QUERY, AUTH_HANDSHAKE, UP, UPDATE)
	State BrokerState `json:"state"`
	// Time since last broker state change (microseconds)
	StateAge int64 `json:"stateage"`
	// Number of requests awaiting transmission to broker
	RequestsAwaitingTransmission int64 `json:"outbuf_cnt"`
	// Number of messages awaiting transmission to broker
	MessagesAwaitingTransmission int64 `json:"outbuf_msg_cnt"`
	// Number of requests in-flight to broker awaiting response
	InflightRequestsAwaitingResponse int64 `json:"waitresp_cnt"`
	// Number of messages in-flight to broker awaiting response
	InflightMessagesAwaitingResponse int64 `json:"waitresp_msg_cnt"`
	// Total number of requests sent
	RequestsSent int64 `json:"tx"`
	// Total number of bytes sent
	RequestBytesSent int64 `json:"txbytes"`
	// Total number of transmission errors
	RequestErrors int64 `json:"txerrs"`
	// Total number of request retries
	RequestRetries int64 `json:"txretries"`
	// Microseconds since last socket send (or -1 if no sends yet for current connection).
	LastSocketSend int64 `json:"txidle"`
	// Total number of requests timed out
	RequestTimeouts int64 `json:"req_timeouts"`
	// Total number of responses received
	ResponsesReceived int64 `json:"rx"`
	// Total number of bytes received
	ResponseBytesReceived int64 `json:"rxbytes"`
	// Total number of receive errors
	ResponseErrors int64 `json:"rxerrs"`
	// Microseconds since last socket receive (or -1 if no receives yet for current connection).
	LastSocketReceive int64 `json:"rxidle"`
	// Broker thread poll loop wakeups
	Wakeups int64 `json:"wakeups"`
	// Number of connection attempts, including successful and failed, and name resolution failures.
	Connects int64 `json:"connects"`
	// Number of disconnects (triggered by broker, network, load-balancer, etc.).
	Disconnects     int64                           `json:"disconnects"`
	Latency         WindowStats                     `json:"rtt"`
	Throttle        WindowStats                     `json:"throttle"`
	TopicPartitions map[string]BrokerTopicPartition `json:"toppars"`
}

type BrokerTopicPartition

type BrokerTopicPartition struct {
	Topic     string `json:"topic"`
	Partition int64  `json:"partition"`
}

type ConsumerGroupJoinState

type ConsumerGroupJoinState string

const (
	ConsumerGroupJoinStateUnknown ConsumerGroupJoinState = "unknown"
	// ConsumerGroupJoinStateInit join or rejoin, possibly with an existing assignment
	ConsumerGroupJoinStateInit ConsumerGroupJoinState = "init"
	// ConsumerGroupJoinStateWaitJoin JoinGroupRequest sent, awaiting response
	ConsumerGroupJoinStateWaitJoin ConsumerGroupJoinState = "wait-join"
	// ConsumerGroupJoinStateWaitMetadata MetadataRequest sent, awaiting response.
	ConsumerGroupJoinStateWaitMetadata ConsumerGroupJoinState = "wait-metadata"
	// ConsumerGroupJoinStateWaitSync SyncGroupRequest sent, awaiting response. Follower only.
	ConsumerGroupJoinStateWaitSync ConsumerGroupJoinState = "wait-sync"
	// ConsumerGroupJoinStateWaitAssignCall waiting for application to call *_assign()
	ConsumerGroupJoinStateWaitAssignCall ConsumerGroupJoinState = "wait-assign-call"
	// ConsumerGroupJoinStateWaitUnassignCall waiting for application to call *_unassign()
	ConsumerGroupJoinStateWaitUnassignCall ConsumerGroupJoinState = "wait-unassign-call"
	// ConsumerGroupJoinStateWaitUnassignToComplete means that the coordinator is up and manager is assigned
	ConsumerGroupJoinStateWaitUnassignToComplete ConsumerGroupJoinState = "wait-unassign-to-complete"
	// ConsumerGroupJoinStateWaitIncrementalUnassignToComplete means that the coordinator is up and manager is assigned
	ConsumerGroupJoinStateWaitIncrementalUnassignToComplete ConsumerGroupJoinState = "wait-incr-unassign-to-complete"
	// ConsumerGroupJoinStateSteady means synchronized and assigned may be an empty assignment
	ConsumerGroupJoinStateSteady ConsumerGroupJoinState = "steady"
)

func (ConsumerGroupJoinState) Int64

func (s ConsumerGroupJoinState) Int64() int64

func (*ConsumerGroupJoinState) UnmarshalJSON

func (s *ConsumerGroupJoinState) UnmarshalJSON(data []byte) error

type ConsumerGroupState

type ConsumerGroupState string

const (
	ConsumerGroupStateUnknown ConsumerGroupState = "unknown"
	// ConsumerGroupStateInit means the consumer group is initialization state
	ConsumerGroupStateInit ConsumerGroupState = "init"
	// ConsumerGroupStateTerminated means the Consumer Group has been stopped. This is a final state.
	ConsumerGroupStateTerminated ConsumerGroupState = "term"
	// ConsumerGroupStateQueryCoordinator means the consumer group is querying for group coordinator
	ConsumerGroupStateQueryCoordinator ConsumerGroupState = "query-coord"
	// ConsumerGroupStateWaitCoordinator means that the client is awaiting response for outstanding query
	ConsumerGroupStateWaitCoordinator ConsumerGroupState = "wait-coord"
	// ConsumerGroupStateWaitBroker means that the client is waiting ack from assigned consumer group manager broker thread
	ConsumerGroupStateWaitBroker ConsumerGroupState = "wait-broker"
	// ConsumerGroupStateWaitBrokerTransport means that the client is waiting for manager broker thread to connect to broker
	ConsumerGroupStateWaitBrokerTransport ConsumerGroupState = "wait-broker-transport"
	// ConsumerGroupStateUp means that the coordinator is up and manager is assigned
	ConsumerGroupStateUp ConsumerGroupState = "up"
)

func (ConsumerGroupState) Int64

func (s ConsumerGroupState) Int64() int64

func (*ConsumerGroupState) UnmarshalJSON

func (s *ConsumerGroupState) UnmarshalJSON(data []byte) error

type ConsumerGroupStats

type ConsumerGroupStats struct {
	// Local consumer group handler's state
	State ConsumerGroupState `json:"state"`
	// Time elapsed since last state change (milliseconds)
	StateAge int64 `json:"stateage"`
	// Local consumer group handler's join state
	JoinState ConsumerGroupJoinState `json:"join_state"`
	// Time elapsed since last rebalance (assign or revoke) (milliseconds)
	RebalanceAge int64 `json:"rebalance_age"`
	// Total number of rebalances (assign or revoke)
	RebalanceCount int64 `json:"rebalance_cnt"`
	// Last rebalance reason, or empty string
	RebalanceReason string `json:"rebalance_reason"`
	// Current assignment's partition count
	PartitionAssigned int64 `json:"assignment_size"`
}

type Partition

type Partition struct {
	// Partition Id (-1 for internal UA/UnAssigned partition)
	Partition int64 `json:"partition"`
	// The id of the broker that messages are currently being fetched from
	Broker int64 `json:"broker"`
	// Current leader broker id
	Leader int64 `json:"leader"`
	// Partition is explicitly desired by application
	Desired bool `json:"desired"`
	// Partition not seen in topic metadata from broker
	Unknown bool `json:"unknown"`
	// Number of messages waiting to be produced in first-level queue
	MessagesInQueue int64 `json:"msgq_cnt"`
	// Number of bytes in msgq_cnt
	MessageBytesInQueue int64 `json:"msgq_bytes"`
	// Number of messages ready to be produced in transmit queue
	MessagesReadyToTransmit int64 `json:"xmit_msgq_cnt"`
	// Number of bytes ready to be produced in transmit queue
	MessageBytesReadyToTransmit int64 `json:"xmit_msgq_bytes"`
	// Number of pre-fetched messages in fetch queue
	MessagesInFetchQueue int64 `json:"fetchq_cnt"`
	// Number of message bytes pre-fetched in fetch queue
	MessageBytesInFetchQueue int64 `json:"fetchq_size"`
	// Current/Last logical offset query
	QueryOffset int64 `json:"query_offset"`
	// Next offset to fetch
	NextOffset int64 `json:"next_offset"`
	// Offset of last message passed to application + 1
	AppOffset int64 `json:"app_offset"`
	// Offset to be committed
	StoredOffset int64 `json:"stored_offset"`
	// Last committed offset
	CommittedOffset int64 `json:"committed_offset"`
	// Last PARTITION_EOF signaled offset
	EOFOffset int64 `json:"eof_offset"`
	// Partition's low watermark offset on broker
	LowWatermarkOffset int64 `json:"lo_offset"`
	// Partition's high watermark offset on broker
	HighWatermarkOffset int64 `json:"hi_offset"`
	// Partition's last stable offset on broker, or same as HighWatermarkOffset is broker version is less than 0.11.0.0.
	LastStableOffsetOnBroker int64 `json:"ls_offset"`
	// Difference between (HighWatermarkOffset or LowWatermarkOffset) and CommittedOffset). HighWatermarkOffset is used when isolation.level=read_uncommitted, otherwise LastStableOffsetOnBroker.
	ConsumerLag int64 `json:"consumer_lag"`
	// Difference between (HighWatermarkOffset or LastStableOffsetOnBroker) and StoredOffset. See consumer_lag and StoredOffset.
	ConsumerLagStored int64 `json:"consumer_lag_stored"`
	// Total number of messages transmitted (produced)
	MessagesSent int64 `json:"txmsgs"`
	// Total number of bytes transmitted for MessagesSent
	MessageBytesSent int64 `json:"txbytes"`
	// Total number of messages consumed, not including ignored messages (due to offset, etc).
	MessagesReceived int64 `json:"rxmsgs"`
	// Total number of bytes received for MessageBytesReceived
	MessageBytesReceived int64 `json:"rxbytes"`
	// Total number of messages received (consumer, same as MessageBytesReceived), or total number of messages produced (possibly not yet transmitted) (producer).
	TotalNumOfMessages int64 `json:"msgs"`
	// Current number of messages in-flight to/from broker
	MessagesInflight int64 `json:"msginflight"`
}

type Stats

type Stats struct {
	// Handle instance name
	Name string `json:"name"`
	// The configured (or default) client.id
	ClientID string `json:"client_id"`
	// Instance type (producer or consumer)
	Type string `json:"type"`
	// Time since this client instance was created (microseconds)
	Age int64 `json:"age"`
	// Number of ops (callbacks, events, etc) waiting in queue for application to serve with rd_kafka_poll()
	ReplyQueue int64 `json:"replyq"`
	// Current number of messages in producer queues
	MessageCount int64 `json:"msg_cnt"`
	// Current total size of messages in producer queues
	MessageSize int64 `json:"msg_size"`
	// Threshold: maximum number of messages allowed on the producer queues
	MessageMax int64 `json:"msg_max"`
	// Total number of requests sent to Kafka brokers
	RequestsSent int64 `json:"tx"`
	// Total number of bytes transmitted to Kafka brokers
	RequestsBytesSent int64 `json:"tx_bytes"`
	// Total number of responses received from Kafka brokers
	RequestsReceived int64 `json:"rx"`
	// Total number of bytes received from Kafka brokers
	RequestsBytesReceived int64 `json:"rx_bytes"`
	// Total number of messages transmitted (produced) to Kafka brokers
	MessagesProduced int64 `json:"txmsgs"`
	// Total number of message bytes (including framing, such as per-Message framing and MessageSet/batch framing) transmitted to Kafka brokers
	MessagesBytesProduced int64 `json:"txmsg_bytes"`
	// Total number of messages consumed, not including ignored messages (due to offset, etc), from Kafka brokers.
	MessagesConsumed int64 `json:"rxmsgs"`
	// Total number of message bytes (including framing) received from Kafka brokers
	MessagesBytesConsumed int64 `json:"rxmsg_bytes"`
	// Number of topics in the metadata cache
	TopicsInMetadataCache int64                  `json:"metadata_cache_cnt"`
	Brokers               map[string]BrokerStats `json:"brokers"`
	Topics                map[string]TopicStats  `json:"topics"`
	ConsumerGroup         ConsumerGroupStats     `json:"cgrp"`
}

https://github.com/confluentinc/librdkafka/blob/v2.4.0/STATISTICS.md

type TopicStats

type TopicStats struct {
	// Topic name
	Topic string `json:"topic"`
	// Age of client's topic object (milliseconds)
	Age int64 `json:"age"`
	// Age of metadata from broker for this topic (milliseconds)
	MetadataAge int64 `json:"metadata_age"`
	// Batch sizes in bytes
	BatchSize WindowStats `json:"batch_size"`
	// Batch message counts
	BatchCount WindowStats `json:"batch_count"`
	// Partitions
	Partitions map[string]Partition `json:"partitions"`
}

type WindowStats

type WindowStats struct {
	// Smallest value
	Min int64 `json:"min"`
	// Largest value
	Max int64 `json:"max"`
	// Average value
	Avg int64 `json:"avg"`
	// Sum of values
	Sum int64 `json:"sum"`
	// Number of values sampled
	Count int64 `json:"count"`
	// Standard deviation (based on histogram)
	StdDev int64 `json:"stddev"`
	// Memory size of Hdr Histogram
	HdrSize int64 `json:"hdrsize"`
	// 50th percentile
	P50 int64 `json:"p50"`
	// 75th percentile
	P75 int64 `json:"p75"`
	// 90th percentile
	P90 int64 `json:"p90"`
	// 95th percentile
	P95 int64 `json:"p95"`
	// 99th percentile
	P99 int64 `json:"p99"`
	// 99.99th percentile
	P9999 int64 `json:"p99_99"`
}

WindowStats stores rolling window statistics. The values are in microseconds unless otherwise stated.