kprom

README

kprom

kprom is a plug-in package to provide prometheus metrics through a kgo.Hook.

This package tracks the following metrics under the following names, all metrics being counter vecs:

#{ns}_connects_total{node_id="#{node}"}
#{ns}_connect_errors_total{node_id="#{node}"}
#{ns}_write_errors_total{node_id="#{node}"}
#{ns}_write_bytes_total{node_id="#{node}"}
#{ns}_read_errors_total{node_id="#{node}"}
#{ns}_read_bytes_total{node_id="#{node}"}
#{ns}_produce_bytes_total{node_id="#{node}",topic="#{topic}"}
#{ns}_fetch_bytes_total{node_id="#{node}",topic="#{topic}"}
#{ns}_buffered_produce_records_total
#{ns}_buffered_fetch_records_total

The above metrics can be expanded considerably with options in this package, allowing timings, uncompressed and compressed bytes, and different labels.

Note that seed brokers use broker IDs prefixed with "seed_", with the number corresponding to which seed it is.

To use,

metrics := kprom.NewMetrics("namespace")
cl, err := kgo.NewClient(
	kgo.WithHooks(metrics),
	// ...other opts
)

You can use your own prometheus registry, as well as a few other options. See the package documentation for more info!

Documentation

Overview

Package kprom provides prometheus plug-in metrics for a kgo client.

This package tracks the following metrics under the following names, all metrics being counter vecs:

#{ns}_connects_total{node_id="#{node}"}
#{ns}_connect_errors_total{node_id="#{node}"}
#{ns}_write_errors_total{node_id="#{node}"}
#{ns}_write_bytes_total{node_id="#{node}"}
#{ns}_read_errors_total{node_id="#{node}"}
#{ns}_read_bytes_total{node_id="#{node}"}
#{ns}_produce_bytes_total{node_id="#{node}",topic="#{topic}"}
#{ns}_fetch_bytes_total{node_id="#{node}",topic="#{topic}"}
#{ns}_buffered_produce_records_total
#{ns}_buffered_fetch_records_total

The above metrics can be expanded considerably with options in this package, allowing timings, uncompressed and compressed bytes, and different labels.

This can be used in a client like so:

m := kprom.NewMetrics("my_namespace")
cl, err := kgo.NewClient(
        kgo.WithHooks(m),
        // ...other opts
)

More examples are linked in the main project readme: https://github.com/twmb/franz-go/#metrics--logging

By default, metrics are installed under the a new prometheus registry, but this can be overridden with the Registry option.

Note that seed brokers use broker IDs prefixed with "seed_", with the number corresponding to which seed it is.

Index

• Variables
• type Detail
• type Histogram
• type HistogramOpts
• type Metrics
  • func NewMetrics(namespace string, opts ...Opt) *Metrics
  • func (m *Metrics) Handler() http.Handler
  • func (m *Metrics) OnBrokerConnect(meta kgo.BrokerMetadata, _ time.Duration, _ net.Conn, err error)
  • func (m *Metrics) OnBrokerDisconnect(meta kgo.BrokerMetadata, _ net.Conn)
  • func (m *Metrics) OnBrokerE2E(meta kgo.BrokerMetadata, _ int16, e2e kgo.BrokerE2E)
  • func (m *Metrics) OnBrokerRead(meta kgo.BrokerMetadata, _ int16, bytesRead int, _, _ time.Duration, err error)
  • func (m *Metrics) OnBrokerThrottle(meta kgo.BrokerMetadata, throttleInterval time.Duration, _ bool)
  • func (m *Metrics) OnBrokerWrite(meta kgo.BrokerMetadata, _ int16, bytesWritten int, _, _ time.Duration, ...)
  • func (m *Metrics) OnClientClosed(*kgo.Client)
  • func (m *Metrics) OnFetchBatchRead(meta kgo.BrokerMetadata, topic string, _ int32, metrics kgo.FetchBatchMetrics)
  • func (m *Metrics) OnNewClient(client *kgo.Client)
  • func (m *Metrics) OnProduceBatchWritten(meta kgo.BrokerMetadata, topic string, _ int32, ...)
  • func (m *Metrics) Registry() prometheus.Registerer
• type Opt
  • func Buckets(buckets []float64) Opt
  • func FetchAndProduceDetail(details ...Detail) Opt
  • func Gatherer(gatherer prometheus.Gatherer) Opt
  • func GoCollectors() Opt
  • func HandlerOpts(opts promhttp.HandlerOpts) Opt
  • func Histograms(hs ...Histogram) Opt
  • func HistogramsFromOpts(hs ...HistogramOpts) Opt
  • func Registerer(reg prometheus.Registerer) Opt
  • func Registry(rg RegistererGatherer) Opt
  • func Subsystem(ss string) Opt
  • func WithClientLabel() Opt
• type RegistererGatherer

Variables

var DefBuckets = []float64{0.001, 0.002, 0.004, 0.008, 0.016, 0.032, 0.064, 0.128, 0.256, 0.512, 1.024, 2.048}

DefBuckets are the default Histogram buckets. The default buckets are tailored to broadly measure the kafka timings (in seconds).

Types

type Detail

type Detail uint8

A Detail is a label that can be set on fetch/produce metrics

const (
	ByNode            Detail = iota // Include label "node_id" for fetch and produce metrics.
	ByTopic                         // Include label "topic" for fetch and produce metrics.
	Batches                         // Report number of fetched and produced batches.
	Records                         // Report the number of fetched and produced records.
	CompressedBytes                 // Report the number of fetched and produced compressed bytes.
	UncompressedBytes               // Report the number of fetched and produced uncompressed bytes.
	ConsistentNaming                // Renames {fetch,produce}_bytes_total to {fetch,produce}_uncompressed_bytes_total, making the names consistent with the CompressedBytes detail.
)

type Histogram

type Histogram uint8

A Histogram is an identifier for a kprom histogram that can be enabled

const (
	ReadWait           Histogram = iota // Enables {ns}_{ss}_read_wait_seconds.
	ReadTime                            // Enables {ns}_{ss}_read_time_seconds.
	WriteWait                           // Enables {ns}_{ss}_write_wait_seconds.
	WriteTime                           // Enables {ns}_{ss}_write_time_seconds.
	RequestDurationE2E                  // Enables {ns}_{ss}_request_durationE2E_seconds.
	RequestThrottled                    // Enables {ns}_{ss}_request_throttled_seconds.
)

type HistogramOpts

type HistogramOpts struct {
	Enable  Histogram
	Buckets []float64
}

HistogramOpts allows histograms to be enabled with custom buckets

type Metrics

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

Metrics provides prometheus metrics

func NewMetrics

func NewMetrics(namespace string, opts ...Opt) *Metrics

NewMetrics returns a new Metrics that adds prometheus metrics to the registry under the given namespace.

func (*Metrics) Handler

func (m *Metrics) Handler() http.Handler

Handler returns an http.Handler providing prometheus metrics.

func (*Metrics) OnBrokerConnect

func (m *Metrics) OnBrokerConnect(meta kgo.BrokerMetadata, _ time.Duration, _ net.Conn, err error)

OnBrokerConnect implements the HookBrokerConnect interface for metrics gathering. This method is meant to be called by the hook system and not by the user

func (*Metrics) OnBrokerDisconnect

func (m *Metrics) OnBrokerDisconnect(meta kgo.BrokerMetadata, _ net.Conn)

OnBrokerDisconnect implements the HookBrokerDisconnect interface for metrics gathering. This method is meant to be called by the hook system and not by the user

func (*Metrics) OnBrokerE2E

func (m *Metrics) OnBrokerE2E(meta kgo.BrokerMetadata, _ int16, e2e kgo.BrokerE2E)

OnBrokerE2E implements the HookBrokerE2E interface for metrics gathering This method is meant to be called by the hook system and not by the user

func (*Metrics) OnBrokerRead

func (m *Metrics) OnBrokerRead(meta kgo.BrokerMetadata, _ int16, bytesRead int, _, _ time.Duration, err error)

// Nop hook for compat, logic moved to OnBrokerE2E

func (*Metrics) OnBrokerThrottle

func (m *Metrics) OnBrokerThrottle(meta kgo.BrokerMetadata, throttleInterval time.Duration, _ bool)

OnBrokerThrottle implements the HookBrokerThrottle interface for metrics gathering. This method is meant to be called by the hook system and not by the user

func (*Metrics) OnBrokerWrite

func (m *Metrics) OnBrokerWrite(meta kgo.BrokerMetadata, _ int16, bytesWritten int, _, _ time.Duration, err error)

Nop hook for compat, logic moved to OnBrokerE2E

func (*Metrics) OnClientClosed

func (m *Metrics) OnClientClosed(*kgo.Client)

OnClientClosed will unregister kprom metrics from kprom registerer

func (*Metrics) OnFetchBatchRead

func (m *Metrics) OnFetchBatchRead(meta kgo.BrokerMetadata, topic string, _ int32, metrics kgo.FetchBatchMetrics)

OnFetchBatchRead implements the HookFetchBatchRead interface for metrics gathering. This method is meant to be called by the hook system and not by the user

func (*Metrics) OnNewClient

func (m *Metrics) OnNewClient(client *kgo.Client)

OnNewClient implements the HookNewClient interface for metrics gathering. This method is meant to be called by the hook system and not by the user

func (*Metrics) OnProduceBatchWritten

func (m *Metrics) OnProduceBatchWritten(meta kgo.BrokerMetadata, topic string, _ int32, metrics kgo.ProduceBatchMetrics)

OnProduceBatchWritten implements the HookProduceBatchWritten interface for metrics gathering. This method is meant to be called by the hook system and not by the user

func (*Metrics) Registry

func (m *Metrics) Registry() prometheus.Registerer

Registry returns the prometheus registry that metrics were added to.

This is useful if you want the Metrics type to create its own registry for you to add additional metrics to.

type Opt

type Opt interface {
	// contains filtered or unexported methods
}

Opt is an option to configure Metrics.

func Buckets

func Buckets(buckets []float64) Opt

Buckets sets the buckets to be used with Histograms, overriding the default of kprom.DefBuckets. If custom buckets per histogram is needed, HistogramOpts can be used.

func FetchAndProduceDetail

func FetchAndProduceDetail(details ...Detail) Opt

FetchAndProduceDetail determines details for fetch/produce metrics, overriding the default of (UncompressedBytes, ByTopic, ByNode).

func Gatherer

func Gatherer(gatherer prometheus.Gatherer) Opt

Gatherer sets the gatherer to add gather to, rather than a new registry.

func GoCollectors

func GoCollectors() Opt

GoCollectors adds the prometheus.NewProcessCollector and prometheus.NewGoCollector collectors the the Metric's registry.

func HandlerOpts

func HandlerOpts(opts promhttp.HandlerOpts) Opt

HandlerOpts sets handler options to use if you wish you use the Metrics.Handler function.

This is only useful if you both (a) do not want to provide your own registry and (b) want to override the default handler options.

func Histograms

func Histograms(hs ...Histogram) Opt

Histograms sets the histograms to be enabled for kprom, overiding the default of disabling all histograms.

metrics, _ := kprom.NewMetrics(
	kprom.Histograms(
		kprom.RequestDurationE2E,
	),
)

func HistogramsFromOpts

func HistogramsFromOpts(hs ...HistogramOpts) Opt

HistogramsFromOpts allows the user full control of what histograms to enable and define buckets to be used with each histogram.

metrics, _ := kprom.NewMetrics(
 kprom.HistogramsFromOpts(
 	kprom.HistogramOpts{
 		Enable:  kprom.ReadWait,
 		Buckets: prometheus.LinearBuckets(10, 10, 8),
 	},
 	kprom.HistogramOpts{
 		Enable: kprom.ReadeTime,
 		// kprom default bucket will be used
 	},
 ),
)

func Registerer

func Registerer(reg prometheus.Registerer) Opt

Registerer sets the registerer to add register to, rather than a new registry.

func Registry

func Registry(rg RegistererGatherer) Opt

Registry sets the registerer and gatherer to add metrics to, rather than a new registry. Use this option if you want to configure both Gatherer and Registerer with the same object.

func Subsystem

func Subsystem(ss string) Opt

Subsystem sets the subsystem for the kprom metrics, overriding the default empty string.

func WithClientLabel

func WithClientLabel() Opt

WithClientLabel adds a "cliend_id" label to all metrics.

type RegistererGatherer

type RegistererGatherer interface {
	prometheus.Registerer
	prometheus.Gatherer
}