owl

Documentation

Index

• type Broker
• type ClusterInfo
• type ConsumerGroupLag
  • func (c *ConsumerGroupLag) GetTopicLag(topicName string) *TopicLag
• type ConsumerGroupOverview
• type GroupMemberAssignment
• type GroupMemberDescription
• type PartitionLag
• type Service
  • func NewService(kafkaSvc *kafka.Service, logger *zap.Logger) *Service
  • func (s *Service) GetClusterInfo(ctx context.Context) (*ClusterInfo, error)
  • func (s *Service) GetConsumerGroupsOverview(ctx context.Context) ([]*ConsumerGroupOverview, error)
  • func (s *Service) GetTopicConfigs(topicName string, configNames []string) (*TopicConfigs, error)
  • func (s *Service) GetTopicsConfigs(topicNames []string, configNames []string) (map[string]*TopicConfigs, error)
  • func (s *Service) GetTopicsOverview() ([]*TopicOverview, error)
  • func (s *Service) ListTopicConsumers(ctx context.Context, topicName string) ([]*TopicConsumerGroup, error)
  • func (s *Service) ListTopicPartitions(topicName string) ([]TopicPartition, error)
• type TopicConfigEntry
• type TopicConfigs
  • func (t *TopicConfigs) GetConfigEntryByName(configName string) *TopicConfigEntry
• type TopicConsumerGroup
• type TopicLag
• type TopicOverview
• type TopicPartition

Types

type Broker

type Broker struct {
	BrokerID   int32  `json:"brokerId"`
	LogDirSize int64  `json:"logDirSize"`
	Address    string `json:"address"`
	Rack       string `json:"rack"`
}

Broker described by some basic broker properties

type ClusterInfo

type ClusterInfo struct {
	ControllerID int32     `json:"controllerId"`
	Brokers      []*Broker `json:"brokers"`
}

ClusterInfo describes the brokers in a cluster

type ConsumerGroupLag

type ConsumerGroupLag struct {
	GroupID   string      `json:"groupId"`
	TopicLags []*TopicLag `json:"topicLags"`
}

ConsumerGroupLag describes the kafka lag for all topics/partitions for a single consumer group

func (*ConsumerGroupLag) GetTopicLag

func (c *ConsumerGroupLag) GetTopicLag(topicName string) *TopicLag

GetTopicLag returns the group's topic lag or nil if the group has no group offsets on that topic

type ConsumerGroupOverview

type ConsumerGroupOverview struct {
	GroupID       string                    `json:"groupId"`
	State         string                    `json:"state"`
	ProtocolType  string                    `json:"protocolType"`
	Protocol      string                    `json:"-"`
	Members       []*GroupMemberDescription `json:"members"`
	CoordinatorID int32                     `json:"coordinatorId"`
	Lags          *ConsumerGroupLag         `json:"lag"`
}

ConsumerGroupOverview for a Kafka Consumer Group

type GroupMemberAssignment

type GroupMemberAssignment struct {
	TopicName    string  `json:"topicName"`
	PartitionIDs []int32 `json:"partitionIds"`
}

GroupMemberAssignment represents a partition assignment for a group member

type GroupMemberDescription

type GroupMemberDescription struct {
	ID          string                   `json:"id"`
	ClientID    string                   `json:"clientId"`
	ClientHost  string                   `json:"clientHost"`
	Assignments []*GroupMemberAssignment `json:"assignments"`
}

GroupMemberDescription is a member (e. g. connected host) of a Consumer Group

type PartitionLag

type PartitionLag struct {
	PartitionID int32 `json:"partitionId"`
	Lag         int64 `json:"lag"`
}

PartitionLag describes the kafka lag for a partition for a single consumer group

type Service

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

Service offers all methods to serve the responses for the REST API. This usually only involves fetching serveral responses from Kafka concurrently and constructing them so, that they are

func NewService

func NewService(kafkaSvc *kafka.Service, logger *zap.Logger) *Service

NewService for the Owl package

func (*Service) GetClusterInfo

func (s *Service) GetClusterInfo(ctx context.Context) (*ClusterInfo, error)

GetClusterInfo returns generic information about all brokers in a Kafka cluster and returns them

func (*Service) GetConsumerGroupsOverview

func (s *Service) GetConsumerGroupsOverview(ctx context.Context) ([]*ConsumerGroupOverview, error)

GetConsumerGroupsOverview returns a ConsumerGroupOverview for all available consumer groups

func (*Service) GetTopicConfigs

func (s *Service) GetTopicConfigs(topicName string, configNames []string) (*TopicConfigs, error)

GetTopicConfigs calls GetTopicsConfigs for a single Topic and returns a single response

func (*Service) GetTopicsConfigs

func (s *Service) GetTopicsConfigs(topicNames []string, configNames []string) (map[string]*TopicConfigs, error)

GetTopicsConfigs fetches all topic config options for the given set of topic names and config names and converts that information so that it is handy to use. Provide an empty array for configNames to describe all config entries.

func (*Service) GetTopicsOverview

func (s *Service) GetTopicsOverview() ([]*TopicOverview, error)

GetTopicsOverview returns a TopicOverview for all Kafka Topics

func (*Service) ListTopicConsumers

func (s *Service) ListTopicConsumers(ctx context.Context, topicName string) ([]*TopicConsumerGroup, error)

ListTopicConsumers returns all consumer group names along with their accumulated lag across all partitions which have at least one active offset on the given topic.

func (*Service) ListTopicPartitions

func (s *Service) ListTopicPartitions(topicName string) ([]TopicPartition, error)

ListTopicPartitions returns the partition in the topic along with their watermarks

type TopicConfigEntry

type TopicConfigEntry struct {
	Name      string `json:"name"`
	Value     string `json:"value"`
	IsDefault bool   `json:"isDefault"`
}

TopicConfigEntry is a key value pair of a config property with it's value

type TopicConfigs

type TopicConfigs struct {
	TopicName     string              `json:"topicName"`
	ConfigEntries []*TopicConfigEntry `json:"configEntries"`
}

TopicConfigs is a TopicName along with all it's config entries

func (*TopicConfigs) GetConfigEntryByName

func (t *TopicConfigs) GetConfigEntryByName(configName string) *TopicConfigEntry

GetConfigEntryByName returns the TopicConfigEntry for a given config name (e. g. "cleanup.policy") or nil if no config with that name could be found.

type TopicConsumerGroup

type TopicConsumerGroup struct {
	GroupID   string `json:"groupId"`
	SummedLag int64  `json:"summedLag"`
}

TopicConsumersGroup is a group along with it's accumulated topic log for a given topic

type TopicLag

type TopicLag struct {
	Topic                 string         `json:"topic"`
	SummedLag             int64          `json:"summedLag"`             // Sums all partition lags (non consumed partitions are not considered)
	ConsumesAllPartitions bool           `json:"consumesAllPartitions"` // Whether the consumer group has at least one active offset for all partitions or not
	PartitionLags         []PartitionLag `json:"partitionLags"`
}

TopicLag describes the kafka lag for a single topic and it's partitions for a single consumer group

type TopicOverview

type TopicOverview struct {
	TopicName         string `json:"topicName"`
	IsInternal        bool   `json:"isInternal"`
	PartitionCount    int    `json:"partitionCount"`
	ReplicationFactor int    `json:"replicationFactor"`
	CleanupPolicy     string `json:"cleanupPolicy"`
	LogDirSize        int64  `json:"logDirSize"`
}

TopicOverview is all information we get when listing Kafka topics

type TopicPartition

type TopicPartition struct {
	ID            int32 `json:"id"`
	WaterMarkLow  int64 `json:"waterMarkLow"`
	WaterMarkHigh int64 `json:"waterMarkHigh"`
}

TopicPartition consists of some (not all) information about a partition of a topic. Only data relevant to the 'partition table' in the frontend is included.