kafka

package
v0.1.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 5, 2026 License: MIT Imports: 28 Imported by: 0

Documentation

Overview

Package kafka wraps the Kafka client and exposes the operations kafman needs.

It deliberately does not import bubbletea: everything here takes a context and returns values or errors, and internal/ui adapts that into tea.Cmd. Keeping the domain layer free of the UI framework is what makes a headless mode, and plain unit tests, possible.

Index

Constants

View Source 
const (
	GroupStable              = "Stable"
	GroupPreparingRebalance  = "PreparingRebalance"
	GroupCompletingRebalance = "CompletingRebalance"
	GroupEmpty               = "Empty"
	GroupDead                = "Dead"
)

Consumer group states as Kafka reports them.

View Source 
const (
	ScramSHA256 = "SCRAM-SHA-256"
	ScramSHA512 = "SCRAM-SHA-512"
)

SCRAM mechanisms, in the spelling Kafka uses on the wire and kafman shows.

View Source 
const DefaultConnectTimeout = 15 * time.Second

DefaultConnectTimeout bounds a connectivity check when the caller gives no deadline of its own.

View Source 
const DefaultScramIterations int32 = 4096

DefaultScramIterations is what Kafka itself defaults to. Fewer than 4096 is refused by the broker.

Variables

View Source 
var (
	// ACLResourceTypes are the resources an ACL can be attached to.
	ACLResourceTypes = []string{"Topic", "Group", "Cluster", "TransactionalID", "DelegationToken"}

	// ACLPatternTypes decide whether ResourceName is a name or a prefix.
	ACLPatternTypes = []string{"Literal", "Prefixed"}

	// ACLOperations are what a principal may do.
	ACLOperations = []string{
		"All", "Read", "Write", "Create", "Delete", "Alter", "Describe",
		"ClusterAction", "DescribeConfigs", "AlterConfigs", "IdempotentWrite",
	}

	// ACLPermissions are the two answers an ACL gives.
	ACLPermissions = []string{"Allow", "Deny"}
)

The vocabulary of an ACL, in Kafka's own spelling.

kafman carries these as strings rather than sarama's enums: they are typed by people, shown in tables and compared, and sarama declares String() on a pointer receiver, so a value in a struct field does not stringify at all.

View Source 
var CommonTopicConfigs = []struct {
	Key         string
	Description string
	Example     string
}{
	{"cleanup.policy", "delete old records, or compact by key", "delete"},
	{"retention.ms", "how long records are kept", "604800000"},
	{"retention.bytes", "how much is kept per partition, -1 for unlimited", "-1"},
	{"min.insync.replicas", "replicas that must acknowledge a write", "2"},
	{"max.message.bytes", "largest record the topic accepts", "1048588"},
	{"segment.ms", "how often a new log segment is rolled", "604800000"},
	{"compression.type", "producer, gzip, snappy, lz4, zstd or uncompressed", "producer"},
}

CommonTopicConfigs are the settings offered when creating a topic, with a short description each. Kafka has well over a hundred; these are the ones people actually set.

View Source 
var ErrNoQuorum = errors.New("the cluster does not expose a metadata quorum (not KRaft, or too old)")

ErrNoQuorum reports a cluster without a describable metadata quorum: a ZooKeeper-era cluster, or brokers too old to answer.

Functions

func BuildSaramaConfig 

func BuildSaramaConfig(c *config.Cluster) (*sarama.Config, error)

BuildSaramaConfig translates a cluster definition into a sarama configuration.

It performs no I/O beyond reading the certificate files named in the config, so it is safe to call while validating a cluster the user is still editing.

func Diagnose 

func Diagnose(err error) string

Diagnose turns a Kafka or network error into a sentence that says what to do.

sarama's errors are accurate but unhelpful to anyone who is not already familiar with them: "kafka: client has run out of available brokers to talk to" is what you get for a wrong port, a wrong password and a firewall alike.

func Explain 

func Explain(err error) string

Explain renders an error the way it should reach a person: kafman's own wording when it has one, and the error's own text when it does not.

Everything the UI shows goes through here, so a protocol code never reaches the screen on its own.

func ExportPath 

func ExportPath(dir string) string

ExportPath builds a default file name in dir, stamped with the time so repeated exports do not overwrite each other.

func WriteExport 

func WriteExport(path string, export Export) error

WriteExport writes an export to path, creating the directory if needed.

Types

type ACL 

type ACL struct {
	Principal  string
	Host       string
	Operation  string
	Permission string

	ResourceType string
	ResourceName string
	PatternType  string
}

ACL is one access control entry: a principal may or may not perform an operation on a resource, from a host.

func MockACLs 

func MockACLs() []ACL

MockACLs is the rule set NewMock reports, including one denial.

func (ACL) Denies 

func (a ACL) Denies() bool

Denies reports whether the entry is a denial, which is the row that has to stand out: a single Deny overrides every Allow.

func (ACL) Resource 

func (a ACL) Resource() string

Resource renders the resource the way Kafka names it, prefix pattern included.

type ACLFilter 

type ACLFilter struct {
	Principal    string
	ResourceType string
	ResourceName string
}

ACLFilter narrows a listing. Empty fields match anything.

type AppliedReset 

type AppliedReset struct {
	Group   string
	Changes []OffsetChange
}

AppliedReset is one reset the mock carried out.

type Batch 

type Batch struct {
	Records []Record

	// Read is how many records have been consumed in total, matched how many
	// passed the filter. Until the query language lands they are equal.
	Read    int64
	Matched int64
}

Batch is a group of records delivered together, with the running totals so the UI can show "matched / read" without counting them itself.

type Broker 

type Broker struct {
	ID   int32
	Addr string
	Rack string

	// Roles is the node's process.roles ("broker", "controller",
	// "broker,controller"), read from its static configuration. Empty on a
	// ZooKeeper-era broker, or when the configuration could not be read.
	Roles string

	// IsController means the node's own controller role on KRaft; on a
	// ZooKeeper cluster it is the elected controller. The KRaft quorum
	// *leader* is deliberately not claimed anywhere: brokers do not expose it
	// to clients — the metadata "controller id" there is a random live broker.
	IsController bool
}

Broker describes one node of the cluster.

type BrokerLoad 

type BrokerLoad struct {
	ID   int32
	Addr string
	Rack string

	// Leaders is the number of partitions this broker leads, Replicas the number
	// it stores at all. The first is what decides the traffic it takes.
	Leaders  int
	Replicas int

	// OutOfSync is the number of its replicas that are behind.
	OutOfSync int

	// UnderReplicated is the number of partitions this broker leads whose ISR
	// is missing replicas; NotPreferred the number it leads while not being the
	// preferred replica (replicas[0]). Each partition has exactly one leader,
	// so summing these over the brokers gives honest cluster totals.
	UnderReplicated int
	NotPreferred    int

	// Roles and IsController carry the same honest semantics as Broker's: on
	// KRaft the node's own role, never the metadata "controller id".
	Roles        string
	IsController bool
}

BrokerLoad is how much of the cluster one broker carries.

It comes from the cached metadata, so a collector can ask for it every poll without adding a round trip.

type Client 

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

Client is the live implementation of Facade, backed by sarama.

func New 

func New(c *config.Cluster) (*Client, error)

New connects to a cluster and returns a ready client.

Connecting is synchronous and can take up to the cluster's request-timeout; callers on the UI thread must run it off the main goroutine.

func (*Client) AddPartitions 

func (c *Client) AddPartitions(ctx context.Context, topic string, total int32) error

AddPartitions raises a topic's partition count to total.

Kafka cannot reduce it, and raising it changes which partition a key routes to, so existing per-key ordering is broken from that point on. The UI warns about this before calling.

func (*Client) ApplyOffsetReset 

func (c *Client) ApplyOffsetReset(ctx context.Context, group string, changes []OffsetChange) error

ApplyOffsetReset commits the previewed positions.

Kafka refuses a commit made from outside the group while the group has members, and that refusal is correct: moving the position under a running consumer would be worse than the error. The message says so instead of repeating the protocol's wording.

func (*Client) BrokerLoads 

func (c *Client) BrokerLoads(ctx context.Context) ([]BrokerLoad, error)

BrokerLoads describes what each broker carries, from cached metadata.

func (*Client) BrokerLogDirs 

func (c *Client) BrokerLogDirs(ctx context.Context) (map[int32][]string, error)

BrokerLogDirs returns log.dirs per broker id, split on commas.

func (*Client) BrokerSizes 

func (c *Client) BrokerSizes(ctx context.Context) (map[int32]int64, error)

BrokerSizes returns the on-disk size held by each broker.

func (*Client) Close 

func (c *Client) Close() error

Close releases the connections. Safe to call more than once.

func (*Client) Cluster 

func (c *Client) Cluster() *config.Cluster

Cluster returns the configuration this client was built from.

func (*Client) ClusterHealth 

func (c *Client) ClusterHealth(ctx context.Context) (Health, error)

ClusterHealth walks every topic's metadata and aggregates the replication state. Internal topics are included: __consumer_offsets going under-replicated matters as much as anything else.

func (*Client) ConsumerGroupsWithLag 

func (c *Client) ConsumerGroupsWithLag(ctx context.Context) ([]ConsumerGroup, error)

ConsumerGroupsWithLag returns the groups including their total lag.

Both halves are batched: the committed offsets of every group come back in one request, and the end offsets of every partition they touch in one more. That matters because the Monitoring collector calls this on a timer — the earlier version asked for one offset per partition, which on a real cluster is hundreds of round trips every few seconds.

func (*Client) CreateACL 

func (c *Client) CreateACL(ctx context.Context, acl ACL) error

CreateACL adds one entry. Kafka accepts a duplicate silently, so creating an entry that already exists is not an error here either.

func (*Client) CreateTopic 

func (c *Client) CreateTopic(ctx context.Context, spec TopicSpec) error

CreateTopic creates a topic.

func (*Client) DeleteACL 

func (c *Client) DeleteACL(ctx context.Context, acl ACL) (int, error)

DeleteACL removes the entries matching exactly the one given, and reports how many were removed.

The filter is built from every field of the entry, pattern type included: deleting "the Literal topic orders.events" must not also remove the Prefixed rule that happens to share its name.

func (*Client) DeleteConsumerGroup 

func (c *Client) DeleteConsumerGroup(ctx context.Context, name string) error

DeleteConsumerGroup removes a group. It fails on the broker side if the group still has members, which is the desired behaviour.

func (*Client) DeleteGroupOffsets 

func (c *Client) DeleteGroupOffsets(ctx context.Context, group, topic string) error

DeleteGroupOffsets removes a group's committed offsets for one topic, which makes the group start from its auto.offset.reset position again.

func (*Client) DeleteRecords 

func (c *Client) DeleteRecords(ctx context.Context, topic string, offsets map[int32]int64) error

DeleteRecords discards records before the given offset on each partition.

This is not the same as deleting a topic: the partitions stay, their start offset simply moves forward. Passing an offset beyond the end deletes everything currently retained.

func (*Client) DeleteScramUser 

func (c *Client) DeleteScramUser(ctx context.Context, name string, mechanisms []string) error

DeleteScramUser removes the given mechanisms of a user, which is what removing the user amounts to. Passing none removes every mechanism it holds.

func (*Client) DeleteTopic 

func (c *Client) DeleteTopic(ctx context.Context, name string) error

DeleteTopic deletes a topic.

The broker refuses unless delete.topic.enable is on, which is worth surfacing rather than leaving as a bare error code.

func (*Client) DescribeCluster 

func (c *Client) DescribeCluster(ctx context.Context) (ClusterInfo, error)

DescribeCluster reports the brokers and which one is the controller.

func (*Client) DescribeConsumerGroup 

func (c *Client) DescribeConsumerGroup(ctx context.Context, name string) (ConsumerGroup, []GroupOffset, error)

DescribeConsumerGroup returns the per-partition detail for one group.

func (*Client) DescribePartitions 

func (c *Client) DescribePartitions(ctx context.Context, topic string, withOffsets bool) ([]Partition, error)

DescribePartitions returns the partitions of a topic, optionally with their offset range.

withOffsets costs two requests per partition, so callers pass false unless the user is actually looking at that topic.

func (*Client) DescribeQuorum 

func (c *Client) DescribeQuorum(ctx context.Context) (QuorumInfo, error)

DescribeQuorum reports the KRaft metadata quorum: the elected leader, its epoch, and every voter and observer with how far behind it is.

func (*Client) ListACLs 

func (c *Client) ListACLs(ctx context.Context, filter ACLFilter) ([]ACL, error)

ListACLs returns the entries matching the filter.

func (*Client) ListConsumerGroups 

func (c *Client) ListConsumerGroups(ctx context.Context) ([]ConsumerGroup, error)

ListConsumerGroups returns the groups without their lag, which is the cheap call suitable for a list.

func (*Client) ListScramUsers 

func (c *Client) ListScramUsers(ctx context.Context) ([]ScramUser, error)

ListScramUsers returns every SCRAM principal known to the cluster.

func (*Client) ListTopics 

func (c *Client) ListTopics(ctx context.Context, includeInternal bool) ([]Topic, error)

ListTopics returns every topic with its replication state.

It refreshes the metadata once and then reads from sarama's cache, so the cost is one round trip regardless of how many topics exist.

func (*Client) PreviewOffsetReset 

func (c *Client) PreviewOffsetReset(ctx context.Context, spec ResetSpec) ([]OffsetChange, error)

PreviewOffsetReset works out where every partition would land, without changing anything.

func (*Client) Produce 

func (c *Client) Produce(ctx context.Context, req ProduceRequest) (ProduceResult, error)

Produce publishes one record.

The producer is created on first use and kept: building one costs a metadata round trip, and a read-only session should not pay for it.

func (*Client) ReadRecords 

func (c *Client) ReadRecords(ctx context.Context, opts ReadOptions) (*ReadSession, error)

ReadRecords starts consuming.

It assigns partitions directly rather than joining a consumer group, and never commits an offset. kafman therefore cannot disturb a production consumer no matter what the user does here — a property borrowed from yozefu and worth keeping deliberate.

func (*Client) ResetTopicConfig 

func (c *Client) ResetTopicConfig(ctx context.Context, topic, key string) error

ResetTopicConfig removes an override, returning the setting to its default.

func (*Client) SetTopicConfig 

func (c *Client) SetTopicConfig(ctx context.Context, topic, key, value string) error

SetTopicConfig changes one setting on a topic.

IncrementalAlterConfig is used rather than AlterConfig: the latter replaces the whole configuration, so anything not listed reverts to its default. That is an easy way to lose a retention setting nobody meant to touch.

func (*Client) TopicConfig 

func (c *Client) TopicConfig(ctx context.Context, topic string, includeDefaults bool) ([]ConfigEntry, error)

TopicConfig returns a topic's settings.

includeDefaults decides whether inherited broker defaults are listed; without it only explicit overrides come back, which is what someone auditing a topic usually wants to see.

func (*Client) TopicEndOffsets 

func (c *Client) TopicEndOffsets(ctx context.Context, includeInternal bool) (map[string]int64, error)

TopicEndOffsets returns the sum of every partition's end offset, per topic.

The difference between two calls is exactly how many records were written in between, which is the only throughput figure the Kafka API can give without JMX. One ListOffsets covers the whole cluster: it is batched per partition leader, so this costs one request per broker.

func (*Client) TopicSizes 

func (c *Client) TopicSizes(ctx context.Context) (map[string]int64, error)

TopicSizes returns the on-disk size of each topic, summed over the brokers.

DescribeLogDirs is one request per broker and returns every partition it hosts, which makes it far cheaper than asking per topic.

func (*Client) TopicStartOffsets added in v0.1.2 

func (c *Client) TopicStartOffsets(ctx context.Context, includeInternal bool) (map[string]int64, error)

TopicStartOffsets returns the sum of every partition's oldest offset, per topic — the retention head. The difference between two calls is how many records deletion (retention, compaction or a truncate) removed in between; end minus start is what is currently retained.

It reuses the metadata TopicEndOffsets refreshed: the collector calls the ends first, and a second refresh in the same poll would buy nothing.

func (*Client) TruncateTopic 

func (c *Client) TruncateTopic(ctx context.Context, topic string) error

TruncateTopic discards every retained record of a topic by moving each partition's start offset to its end.

func (*Client) UpsertScramUser 

func (c *Client) UpsertScramUser(ctx context.Context, spec ScramUserSpec) error

UpsertScramUser creates a user or replaces the password of an existing one. Kafka makes no distinction between the two, and neither does this.

The call returns once the credential is visible in the cluster's own listing. That is not the same as "the new password works everywhere and the old one does not": each broker applies the change to its own SASL state a moment later, and no API reports when. The UI says as much rather than promising an instant cut-off.

type ClusterInfo 

type ClusterInfo struct {
	// ClusterID is empty when the broker does not report one.
	ClusterID    string
	ControllerID int32
	Brokers      []Broker
}

ClusterInfo is the cluster-level metadata shown on the Overview tab.

type ConfigEntry 

type ConfigEntry struct {
	Name  string
	Value string

	// Default reports whether the value comes from the broker default rather
	// than an explicit override on this topic.
	Default bool

	// ReadOnly settings cannot be changed at runtime.
	ReadOnly  bool
	Sensitive bool

	// Source names where the value came from, for the details panel.
	Source string
}

ConfigEntry is one topic or broker setting.

func MockTopicConfig 

func MockTopicConfig(includeDefaults bool) []ConfigEntry

MockTopicConfig is the configuration NewMock reports for any topic.

type ConnCheck 

type ConnCheck struct {
	Cluster *config.Cluster
	Info    ClusterInfo
	Latency time.Duration

	// Err is nil on success. Diagnosis carries a human-readable explanation.
	Err       error
	Diagnosis string
}

ConnCheck is the outcome of testing a cluster connection.

func CheckConnectivity 

func CheckConnectivity(ctx context.Context, c *config.Cluster) ConnCheck

CheckConnectivity opens a throwaway connection, describes the cluster and closes it again. It never returns an error: a failure is part of the result, because the Clusters tab has to render it rather than propagate it.

func (ConnCheck) OK 

func (r ConnCheck) OK() bool

OK reports whether the cluster was reachable.

type ConsumerGroup 

type ConsumerGroup struct {
	Name    string
	State   string
	Members int
	Topics  []string

	// TotalLag is the sum over every assigned partition. It is only filled by
	// the calls that ask for it, because it costs an offset request per
	// partition.
	TotalLag    int64
	LagKnown    bool
	Coordinator int32

	// CommittedSum is the sum of committed offsets over every partition the
	// group has actually committed on — never-committed partitions are skipped,
	// not counted as zero. Its growth between two readings is the group's
	// consumption rate. CommittedParts is how many partitions contributed;
	// NoCommit how many assigned partitions carry no commit at all (0 for a
	// group with no members: there is no assignment to compare against).
	// Filled only by ConsumerGroupsWithLag.
	CommittedSum   int64
	CommittedParts int
	CommittedKnown bool
	NoCommit       int

	// Protocol is the assignment strategy the group agreed on, empty for a group
	// with no members.
	Protocol string

	// MemberList is filled only by DescribeConsumerGroup: the list view needs
	// the count, not the members.
	MemberList []GroupMember
}

ConsumerGroup is one group with its aggregate lag.

func MockGroups 

func MockGroups() []ConsumerGroup

MockGroups is the consumer group set NewMock reports, sorted by lag.

func (ConsumerGroup) Rebalancing 

func (g ConsumerGroup) Rebalancing() bool

Rebalancing reports whether the group is between assignments.

type Export 

type Export struct {
	// Query is the search that produced these records, so the export can be
	// reproduced.
	Query string `json:"query,omitempty"`

	Cluster    string    `json:"cluster"`
	Topics     []string  `json:"topics"`
	ExportedAt time.Time `json:"exported_at"`
	Count      int       `json:"count"`

	Records []ExportedRecord `json:"records"`
}

Export is the file written to disk.

func BuildExport 

func BuildExport(cluster, query string, records []Record) Export

BuildExport prepares records for writing.

type ExportedRecord 

type ExportedRecord struct {
	Topic     string `json:"topic"`
	Partition int32  `json:"partition"`
	Offset    int64  `json:"offset"`

	Timestamp int64  `json:"timestamp_ms"`
	DateTime  string `json:"date_time"`

	// AbsoluteDeltaMs is the offset in time from the first exported record,
	// RelativeDeltaMs from the previous one.
	AbsoluteDeltaMs int64 `json:"absolute_delta_ms"`
	RelativeDeltaMs int64 `json:"relative_delta_ms"`

	Key     any               `json:"key"`
	Value   any               `json:"value"`
	Headers map[string]string `json:"headers,omitempty"`

	Size int `json:"size_bytes"`
}

ExportedRecord is a record prepared for a file.

It carries more than the record itself: the deltas say how far apart events were without the reader having to subtract timestamps, and the query says how the set was selected. An export that cannot be interpreted six months later is not much of an export. Idea from yozefu.

type Facade 

type Facade interface {
	Cluster() *config.Cluster
	Close() error

	// Cluster shape and health, all derived from cached metadata.
	DescribeCluster(ctx context.Context) (ClusterInfo, error)
	ClusterHealth(ctx context.Context) (Health, error)

	// DescribeQuorum reports the KRaft metadata quorum; ErrNoQuorum on a
	// cluster that has none to describe.
	DescribeQuorum(ctx context.Context) (QuorumInfo, error)
	ListTopics(ctx context.Context, includeInternal bool) ([]Topic, error)
	DescribePartitions(ctx context.Context, topic string, withOffsets bool) ([]Partition, error)

	// Throughput sources: end offsets for msg/s, start offsets for the
	// retention head, broker loads for the distribution of partitions.
	TopicEndOffsets(ctx context.Context, includeInternal bool) (map[string]int64, error)
	TopicStartOffsets(ctx context.Context, includeInternal bool) (map[string]int64, error)
	BrokerLoads(ctx context.Context) ([]BrokerLoad, error)

	// Sizes and lag, which cost real round trips.
	TopicSizes(ctx context.Context) (map[string]int64, error)
	BrokerSizes(ctx context.Context) (map[int32]int64, error)

	// BrokerLogDirs returns each broker's log.dirs config, split on commas.
	// Cached for the life of the connection: the value only changes with a
	// restart.
	BrokerLogDirs(ctx context.Context) (map[int32][]string, error)
	ListConsumerGroups(ctx context.Context) ([]ConsumerGroup, error)
	ConsumerGroupsWithLag(ctx context.Context) ([]ConsumerGroup, error)
	DescribeConsumerGroup(ctx context.Context, name string) (ConsumerGroup, []GroupOffset, error)

	// Consumer group administration. A reset is previewed and applied as two
	// calls so the user approves the exact positions that get committed.
	PreviewOffsetReset(ctx context.Context, spec ResetSpec) ([]OffsetChange, error)
	ApplyOffsetReset(ctx context.Context, group string, changes []OffsetChange) error
	DeleteGroupOffsets(ctx context.Context, group, topic string) error
	DeleteConsumerGroup(ctx context.Context, name string) error

	// Reading records. Always by direct partition assignment, never as part of a
	// consumer group, and offsets are never committed.
	ReadRecords(ctx context.Context, opts ReadOptions) (*ReadSession, error)

	// Produce publishes one record.
	Produce(ctx context.Context, req ProduceRequest) (ProduceResult, error)

	// Topic administration.
	CreateTopic(ctx context.Context, spec TopicSpec) error
	DeleteTopic(ctx context.Context, name string) error
	AddPartitions(ctx context.Context, topic string, total int32) error
	TopicConfig(ctx context.Context, topic string, includeDefaults bool) ([]ConfigEntry, error)
	SetTopicConfig(ctx context.Context, topic, key, value string) error
	ResetTopicConfig(ctx context.Context, topic, key string) error
	TruncateTopic(ctx context.Context, topic string) error

	// SCRAM principals and access control.
	ListScramUsers(ctx context.Context) ([]ScramUser, error)
	UpsertScramUser(ctx context.Context, spec ScramUserSpec) error
	DeleteScramUser(ctx context.Context, name string, mechanisms []string) error
	ListACLs(ctx context.Context, filter ACLFilter) ([]ACL, error)
	CreateACL(ctx context.Context, acl ACL) error
	DeleteACL(ctx context.Context, acl ACL) (int, error)
}

Facade is the set of operations the UI depends on. It exists so screens can be tested against Mock instead of a live cluster; it grows as stages land.

func NewFacade 

func NewFacade(c *config.Cluster) (Facade, error)

NewFacade adapts New to the Instantiator signature.

type GroupMember 

type GroupMember struct {
	MemberID string
	ClientID string
	Host     string

	// Assignment is what the group leader handed this member, sorted by topic
	// and partition.
	Assignment []PartitionRef
}

GroupMember is one consumer in a group, with what it was assigned.

type GroupOffset 

type GroupOffset struct {
	Topic     string
	Partition int32

	// Current is the committed offset, or -1 when the group has never committed
	// for this partition. It is not zero: never-committed and committed-at-zero
	// mean very different things.
	Current int64

	OldestOffset int64
	NewestOffset int64

	// Lag is how far behind the end the group is, and Lead how far past the
	// start. Both are -1 when Current is unknown.
	Lag  int64
	Lead int64

	Host     string
	ClientID string
}

GroupOffset is one partition's position for a group.

func MockGroupOffsets 

func MockGroupOffsets() []GroupOffset

MockGroupOffsets is the per-partition detail NewMock reports for a group. 

type Header struct {
	Key   string
	Value string
}

Header is one record header.

type Health 

type Health struct {
	Topics     int
	Partitions int

	// InSync is the number of partitions whose ISR matches their replica list.
	InSync int

	// UnderReplicated partitions have a leader but are missing in-sync replicas:
	// the cluster still serves them, but has lost redundancy.
	UnderReplicated int

	// Offline partitions have no leader and cannot be read or written.
	Offline int

	// TotalReplicas and InSyncReplicas are replica counts, not partition counts,
	// and give the average replication factor.
	TotalReplicas  int
	InSyncReplicas int

	// Worst names the single most degraded partition, so the user has somewhere
	// to start rather than just a count.
	Worst *PartitionIssue
}

Health summarises the replication state of the whole cluster.

Every number here is derived from cached metadata, so refreshing the health panel costs one metadata request no matter how many topics exist.

func (Health) OK 

func (h Health) OK() bool

OK reports whether every partition is fully replicated and online.

func (Health) ReplicationFactor 

func (h Health) ReplicationFactor() float64

ReplicationFactor is the average number of replicas per partition.

type Instantiator 

type Instantiator func(*config.Cluster) (Facade, error)

Instantiator creates a Facade for a cluster. Injecting it into the root model keeps the UI testable without a broker (pattern borrowed from ktea).

func MockInstantiator 

func MockInstantiator(m *Mock) Instantiator

MockInstantiator returns an Instantiator handing out the given mock, for injecting into the root model.

type Mock 

type Mock struct {
	ClusterFunc               func() *config.Cluster
	DescribeClusterFunc       func(ctx context.Context) (ClusterInfo, error)
	ClusterHealthFunc         func(ctx context.Context) (Health, error)
	DescribeQuorumFunc        func(ctx context.Context) (QuorumInfo, error)
	ListTopicsFunc            func(ctx context.Context, includeInternal bool) ([]Topic, error)
	DescribePartitionsFunc    func(ctx context.Context, topic string, withOffsets bool) ([]Partition, error)
	TopicSizesFunc            func(ctx context.Context) (map[string]int64, error)
	TopicEndOffsetsFunc       func(ctx context.Context, includeInternal bool) (map[string]int64, error)
	TopicStartOffsetsFunc     func(ctx context.Context, includeInternal bool) (map[string]int64, error)
	BrokerLoadsFunc           func(ctx context.Context) ([]BrokerLoad, error)
	BrokerSizesFunc           func(ctx context.Context) (map[int32]int64, error)
	BrokerLogDirsFunc         func(ctx context.Context) (map[int32][]string, error)
	ListConsumerGroupsFunc    func(ctx context.Context) ([]ConsumerGroup, error)
	ConsumerGroupsWithLagFunc func(ctx context.Context) ([]ConsumerGroup, error)
	DescribeConsumerGroupFunc func(ctx context.Context, name string) (ConsumerGroup, []GroupOffset, error)
	PreviewOffsetResetFunc    func(ctx context.Context, spec ResetSpec) ([]OffsetChange, error)
	ApplyOffsetResetFunc      func(ctx context.Context, group string, changes []OffsetChange) error
	DeleteGroupOffsetsFunc    func(ctx context.Context, group, topic string) error
	DeleteConsumerGroupFunc   func(ctx context.Context, name string) error
	ReadRecordsFunc           func(ctx context.Context, opts ReadOptions) (*ReadSession, error)
	ProduceFunc               func(ctx context.Context, req ProduceRequest) (ProduceResult, error)
	CreateTopicFunc           func(ctx context.Context, spec TopicSpec) error
	DeleteTopicFunc           func(ctx context.Context, name string) error
	AddPartitionsFunc         func(ctx context.Context, topic string, total int32) error
	TopicConfigFunc           func(ctx context.Context, topic string, includeDefaults bool) ([]ConfigEntry, error)
	SetTopicConfigFunc        func(ctx context.Context, topic, key, value string) error
	ResetTopicConfigFunc      func(ctx context.Context, topic string, key string) error
	TruncateTopicFunc         func(ctx context.Context, topic string) error
	ListScramUsersFunc        func(ctx context.Context) ([]ScramUser, error)
	UpsertScramUserFunc       func(ctx context.Context, spec ScramUserSpec) error
	DeleteScramUserFunc       func(ctx context.Context, name string, mechanisms []string) error
	ListACLsFunc              func(ctx context.Context, filter ACLFilter) ([]ACL, error)
	CreateACLFunc             func(ctx context.Context, acl ACL) error
	DeleteACLFunc             func(ctx context.Context, acl ACL) (int, error)
	CloseFunc                 func() error

	// Created and Deleted record what the tab asked for, for assertions.
	Created []TopicSpec
	Deleted []string

	// Applied records the offset resets that were carried out, DeletedGroups the
	// groups removed and DeletedOffsets the group/topic pairs cleared.
	Applied        []AppliedReset
	DeletedGroups  []string
	DeletedOffsets []string

	// Upserted records the SCRAM users written and DeletedUsers those removed;
	// CreatedACLs and DeletedACLs the access rules.
	Upserted     []ScramUserSpec
	DeletedUsers []string
	CreatedACLs  []ACL
	DeletedACLs  []ACL

	// Produced records the requests passed to Produce, for assertions.
	Produced []ProduceRequest
	// contains filtered or unexported fields
}

Mock is a Facade for tests: UI screens can be driven without a broker.

Every operation is backed by a function field. Leaving one nil returns the zero value, so a test only has to set what it actually exercises. NewMock fills them with a small healthy cluster.

func NewMock 

func NewMock(cluster *config.Cluster) *Mock

NewMock returns a mock reporting a three-broker cluster with one under-replicated partition.

func (*Mock) AddPartitions 

func (m *Mock) AddPartitions(ctx context.Context, topic string, total int32) error

func (*Mock) ApplyOffsetReset 

func (m *Mock) ApplyOffsetReset(ctx context.Context, group string, changes []OffsetChange) error

func (*Mock) BrokerLoads 

func (m *Mock) BrokerLoads(ctx context.Context) ([]BrokerLoad, error)

func (*Mock) BrokerLogDirs 

func (m *Mock) BrokerLogDirs(ctx context.Context) (map[int32][]string, error)

func (*Mock) BrokerSizes 

func (m *Mock) BrokerSizes(ctx context.Context) (map[int32]int64, error)

func (*Mock) Called 

func (m *Mock) Called(name string) bool

Called reports whether the named method was invoked.

func (*Mock) Calls 

func (m *Mock) Calls() []string

Calls returns the names of the methods invoked so far, in order.

func (*Mock) Close 

func (m *Mock) Close() error

func (*Mock) Cluster 

func (m *Mock) Cluster() *config.Cluster

func (*Mock) ClusterHealth 

func (m *Mock) ClusterHealth(ctx context.Context) (Health, error)

func (*Mock) ConsumerGroupsWithLag 

func (m *Mock) ConsumerGroupsWithLag(ctx context.Context) ([]ConsumerGroup, error)

func (*Mock) CreateACL 

func (m *Mock) CreateACL(ctx context.Context, acl ACL) error

func (*Mock) CreateTopic 

func (m *Mock) CreateTopic(ctx context.Context, spec TopicSpec) error

func (*Mock) DeleteACL 

func (m *Mock) DeleteACL(ctx context.Context, acl ACL) (int, error)

func (*Mock) DeleteConsumerGroup 

func (m *Mock) DeleteConsumerGroup(ctx context.Context, name string) error

func (*Mock) DeleteGroupOffsets 

func (m *Mock) DeleteGroupOffsets(ctx context.Context, group, topic string) error

func (*Mock) DeleteScramUser 

func (m *Mock) DeleteScramUser(ctx context.Context, name string, mechanisms []string) error

func (*Mock) DeleteTopic 

func (m *Mock) DeleteTopic(ctx context.Context, name string) error

func (*Mock) DescribeCluster 

func (m *Mock) DescribeCluster(ctx context.Context) (ClusterInfo, error)

func (*Mock) DescribeConsumerGroup 

func (m *Mock) DescribeConsumerGroup(ctx context.Context, name string) (ConsumerGroup, []GroupOffset, error)

func (*Mock) DescribePartitions 

func (m *Mock) DescribePartitions(ctx context.Context, topic string, withOffsets bool) ([]Partition, error)

func (*Mock) DescribeQuorum 

func (m *Mock) DescribeQuorum(ctx context.Context) (QuorumInfo, error)

func (*Mock) ListACLs 

func (m *Mock) ListACLs(ctx context.Context, filter ACLFilter) ([]ACL, error)

func (*Mock) ListConsumerGroups 

func (m *Mock) ListConsumerGroups(ctx context.Context) ([]ConsumerGroup, error)

func (*Mock) ListScramUsers 

func (m *Mock) ListScramUsers(ctx context.Context) ([]ScramUser, error)

func (*Mock) ListTopics 

func (m *Mock) ListTopics(ctx context.Context, includeInternal bool) ([]Topic, error)

func (*Mock) PreviewOffsetReset 

func (m *Mock) PreviewOffsetReset(ctx context.Context, spec ResetSpec) ([]OffsetChange, error)

func (*Mock) Produce 

func (m *Mock) Produce(ctx context.Context, req ProduceRequest) (ProduceResult, error)

func (*Mock) ReadRecords 

func (m *Mock) ReadRecords(ctx context.Context, opts ReadOptions) (*ReadSession, error)

func (*Mock) ResetTopicConfig 

func (m *Mock) ResetTopicConfig(ctx context.Context, topic, key string) error

func (*Mock) SetTopicConfig 

func (m *Mock) SetTopicConfig(ctx context.Context, topic, key, value string) error

func (*Mock) TopicConfig 

func (m *Mock) TopicConfig(ctx context.Context, topic string, includeDefaults bool) ([]ConfigEntry, error)

func (*Mock) TopicEndOffsets 

func (m *Mock) TopicEndOffsets(ctx context.Context, includeInternal bool) (map[string]int64, error)

func (*Mock) TopicSizes 

func (m *Mock) TopicSizes(ctx context.Context) (map[string]int64, error)

func (*Mock) TopicStartOffsets added in v0.1.2 

func (m *Mock) TopicStartOffsets(ctx context.Context, includeInternal bool) (map[string]int64, error)

func (*Mock) TruncateTopic 

func (m *Mock) TruncateTopic(ctx context.Context, topic string) error

func (*Mock) UpsertScramUser 

func (m *Mock) UpsertScramUser(ctx context.Context, spec ScramUserSpec) error

type OffsetChange 

type OffsetChange struct {
	Topic     string
	Partition int32

	// Current is the committed offset, or -1 when the group has never committed
	// for this partition.
	Current int64

	// New is where the partition ends up.
	New int64

	Oldest int64
	Newest int64

	// Clamped reports that the requested position fell outside the partition and
	// was pulled back to its boundary.
	Clamped bool

	// Note explains a partition the reset could not place as asked.
	Note string
}

OffsetChange is one partition's move: what the preview shows and what the apply step commits.

Preview and apply are separate calls over the same values on purpose. What the user approved is exactly what is written, without a second look at the partition boundaries that could have moved in between and quietly turned an approved reset into a different one.

func MockOffsetChanges 

func MockOffsetChanges() []OffsetChange

MockOffsetChanges is the reset preview NewMock reports: one partition that moves back, one that is clamped to the start of the partition.

func (OffsetChange) Delta 

func (c OffsetChange) Delta() int64

Delta is how far the partition moves. It is meaningless, and reported as 0, when the group has never committed here.

func (OffsetChange) LagAfter 

func (c OffsetChange) LagAfter() int64

LagAfter is how far behind the end the group will be once applied.

func (OffsetChange) Moves 

func (c OffsetChange) Moves() bool

Moves reports whether the partition actually changes position.

type Partition 

type Partition struct {
	ID       int32
	Leader   int32
	Replicas []int32
	ISR      []int32

	// OldestOffset and NewestOffset are only filled by the calls that ask for
	// them; they cost a request per partition.
	OldestOffset int64
	NewestOffset int64
}

Partition is the detail of a single partition.

func (Partition) Messages 

func (p Partition) Messages() int64

Messages is the number of records currently retained, which is the offset range rather than everything ever written.

func (Partition) Offline 

func (p Partition) Offline() bool

Offline reports whether the partition has no leader.

func (Partition) UnderReplicated 

func (p Partition) UnderReplicated() bool

UnderReplicated reports whether the partition is missing in-sync replicas.

type PartitionIssue 

type PartitionIssue struct {
	Topic     string
	Partition int32
	ISR       int
	Replicas  int
	Offline   bool
}

PartitionIssue identifies one degraded partition.

func (PartitionIssue) String 

func (p PartitionIssue) String() string

String renders the issue for the "worst:" line.

type PartitionRef 

type PartitionRef struct {
	Topic     string
	Partition int32
}

PartitionRef names one partition.

type ProduceRequest 

type ProduceRequest struct {
	Topic   string
	Key     []byte
	Value   []byte
	Headers []Header

	// Partition is honoured only when Manual is set; otherwise the partition is
	// chosen from the key.
	Partition int32
	Manual    bool
}

ProduceRequest is one record to publish.

type ProduceResult 

type ProduceResult struct {
	Topic     string
	Partition int32
	Offset    int64
}

ProduceResult is where a published record landed.

type QuorumInfo 

type QuorumInfo struct {
	LeaderID      int32
	LeaderEpoch   int32
	HighWatermark int64

	// Voters are the controller nodes of the quorum; Observers are the nodes
	// that follow it without voting — on KRaft these are the plain brokers.
	Voters    []QuorumReplica
	Observers []QuorumReplica
}

QuorumInfo is what DescribeQuorum reports about the metadata quorum.

func (QuorumInfo) Role 

func (q QuorumInfo) Role(id int32) string

Role names a node's place in the quorum: "leader", "follower", "observer", or "" for a node the quorum does not know.

type QuorumReplica 

type QuorumReplica struct {
	ID           int32
	LogEndOffset int64

	// Lag is how far this replica's log is behind the leader's, 0 for the
	// leader itself, -1 when either end is unknown.
	Lag int64
}

QuorumReplica is one member of the metadata quorum.

type ReadOptions 

type ReadOptions struct {
	Topics []string

	// Partitions restricts the read; empty means every partition.
	Partitions []int32

	Start Start

	// Limit stops the read after this many records. Zero means no limit.
	Limit int

	// Follow keeps the consumer running once it reaches the end. When false the
	// read finishes at the high watermark it saw when it started.
	Follow bool
}

ReadOptions configures a read.

type ReadSession 

type ReadSession struct {
	Records <-chan Batch

	// Err holds the reason the session ended, if it was not a clean finish.
	// Read it only after Records has closed.
	Err func() error
	// contains filtered or unexported fields
}

ReadSession is a running consumption. Records arrives in batches; the channel closes when the read finishes, is cancelled, or hits its limit.

func MockSession 

func MockSession(ctx context.Context, records []Record) *ReadSession

MockSession delivers the given records as one batch and closes, which is what a bounded read looks like from the UI's side.

func (*ReadSession) Stop 

func (s *ReadSession) Stop()

Stop ends the session. Safe to call more than once.

type Record 

type Record struct {
	Topic     string
	Partition int32
	Offset    int64
	Key       []byte
	Value     []byte
	Headers   []Header
	Timestamp time.Time

	// Size is the payload size in bytes: key plus value.
	Size int
}

Record is one Kafka record.

func MockRecords 

func MockRecords(topics []string) []Record

MockRecords builds a handful of records covering the formats the record view has to render: JSON, plain text, an empty value and a Confluent payload.

type ResetKind 

type ResetKind int

ResetKind is how the new position of every partition is worked out. 

const (
	ResetToEarliest ResetKind = iota
	ResetToLatest
	ResetToOffset
	ResetToTime
	ResetShiftBy
)

The reset modes, matching what kafka-consumer-groups offers.

func (ResetKind) String 

func (k ResetKind) String() string

type ResetSpec 

type ResetSpec struct {
	Group string

	// Topics limits the reset. Empty means every topic the group has committed
	// offsets for.
	Topics []string

	Kind ResetKind

	// Offset is used by ResetToOffset, At by ResetToTime and Shift by
	// ResetShiftBy. Shift may be negative.
	Offset int64
	At     time.Time
	Shift  int64
}

ResetSpec describes a reset before it is turned into per-partition changes.

type ScramCredential 

type ScramCredential struct {
	Mechanism  string
	Iterations int32
}

ScramCredential is one mechanism registered for a user.

type ScramUser 

type ScramUser struct {
	Name        string
	Credentials []ScramCredential
}

ScramUser is one SASL/SCRAM principal.

Kafka has no user object beyond its credentials: a user exists exactly as long as it has at least one credential, which is why deleting one means deleting every mechanism it holds.

func MockUsers 

func MockUsers() []ScramUser

MockUsers is the SCRAM principal set NewMock reports: one user with both mechanisms, so the list has something to render beyond a single row.

func (ScramUser) Has 

func (u ScramUser) Has(mechanism string) bool

Has reports whether the user holds a credential for the mechanism.

func (ScramUser) Mechanisms 

func (u ScramUser) Mechanisms() []string

Mechanisms lists the user's mechanisms in a stable order.

type ScramUserSpec 

type ScramUserSpec struct {
	Name      string
	Mechanism string
	Password  string

	// Iterations of 0 means DefaultScramIterations.
	Iterations int32
}

ScramUserSpec creates a user or changes its password.

The password is carried by value and never logged, never stored and never put in an error message: it leaves this struct only as the salted hash sarama computes while encoding the request.

type Start 

type Start struct {
	Kind   StartKind
	N      int64
	Offset int64
	Time   time.Time
}

Start describes where to begin reading.

type StartKind 

type StartKind int

StartKind is where consumption begins. 

const (
	// StartEnd reads only what arrives from now on.
	StartEnd StartKind = iota
	// StartBeginning reads everything still retained.
	StartBeginning
	// StartLastN reads the last N records of each partition.
	StartLastN
	// StartOffset reads from an explicit offset.
	StartOffset
	// StartTime reads from the first record at or after a timestamp.
	StartTime
)

Starting positions.

type Topic 

type Topic struct {
	Name       string
	Partitions int

	// ReplicationFactor is the smallest replica count across the partitions;
	// a topic mid-reassignment can have partitions that differ.
	ReplicationFactor int

	// InSync and TotalReplicas aggregate the ISR state over all partitions, and
	// are what the list column "ISR 36/36" is built from.
	InSync        int
	TotalReplicas int

	// UnderReplicated is the number of partitions whose ISR is short of their
	// replica list.
	UnderReplicated int

	// Offline is the number of partitions with no leader at all.
	Offline int

	Internal bool
}

Topic is one topic as seen from the cluster metadata.

The fields here all come from the metadata sarama already caches, so building the whole list costs one refresh rather than a request per topic. Anything that needs a round trip — offsets, sizes, configs — is fetched separately and only for what is on screen.

func MockTopics 

func MockTopics() []Topic

MockTopics is the topic set NewMock reports: healthy, plus one topic that is deliberately under-replicated so health rendering can be exercised.

func (Topic) Healthy 

func (t Topic) Healthy() bool

Healthy reports whether every partition of the topic is fully in sync.

type TopicSpec 

type TopicSpec struct {
	Name              string
	Partitions        int32
	ReplicationFactor int16

	// Configs are topic-level overrides such as cleanup.policy or retention.ms.
	Configs map[string]string
}

TopicSpec describes a topic to create.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.