kafka

Documentation

Overview

Package kafka consumes oslo.messaging notifications and kicks a Neutron reconcile on each committed metadata change, so metadata refreshes within a pass instead of waiting for the periodic net.

It deliberately does NOT parse payloads or touch kernel maps. An event means only "metadata changed"; the reconciler re-reads authoritative state and applies the delta. Routing everything through that one goroutine keeps a single applier for both timer and Kafka — no lock, no parallel apply path — and makes the post-event trie identical to a cold start at the same instant.

docs/architecture/trie-construction.md#incremental-updates

Index

• type Consumer
  • func New(opts Options) *Consumer
  • func (c *Consumer) Run(ctx context.Context)
• type Metrics
  • func NewMetrics(topic string) *Metrics
  • func (m *Metrics) Collectors() []prometheus.Collector
  • func (m *Metrics) RecordConsumeError(topic string)
  • func (m *Metrics) SetLag(topic string, lag int64)
• type Options
• type Trigger

Types

type Consumer

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

Consumer reads notifications from Kafka and kicks a reconcile on every committed Neutron metadata change. Construct with New, then run Consumer.Run on a long-lived goroutine.

func New

func New(opts Options) *Consumer

New constructs a Consumer from opts.

func (*Consumer) Run

func (c *Consumer) Run(ctx context.Context)

Run consumes until ctx is cancelled. It first blocks on boot.PhaseStateRestored so a kick can't fire before the reconciler is ready to apply; if the boot aborts (or ctx is cancelled) before that phase, Run returns without consuming. The reader is closed on exit.

type Metrics

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

Metrics holds the Kafka consumer's Prometheus instruments:

• lachesis_kafka_lag_messages{topic} gauge
• lachesis_kafka_consume_errors_total{topic} counter

Metric catalogue: docs/architecture/metrics.md

func NewMetrics

func NewMetrics(topic string) *Metrics

NewMetrics constructs the bundle and seeds both series for topic, so they exist (at 0) before any traffic — including when the consumer is disabled, so dashboards read 0 rather than "No data".

func (*Metrics) Collectors

func (m *Metrics) Collectors() []prometheus.Collector

Collectors returns the underlying prometheus.Collector values for registration by the agent.

func (*Metrics) RecordConsumeError

func (m *Metrics) RecordConsumeError(topic string)

RecordConsumeError counts one failed read from topic.

func (*Metrics) SetLag

func (m *Metrics) SetLag(topic string, lag int64)

SetLag records the current consumer lag for topic.

type Options

type Options struct {
	Reader  messageReader
	Trigger Trigger
	Seq     *boot.Sequencer
	Metrics *Metrics
	Topic   string
}

Options bundles the inputs to New. Reader, Trigger, and Metrics are required; Seq is optional (nil skips the boot barrier, used by tests); Topic labels the metrics.

type Trigger

type Trigger interface {
	Kick()
}

Trigger is the reconcile kick the consumer fires on a Neutron change. *reconcile.Reconciler satisfies it via its Kick method; tests wire a recording fake.