metadatastore

Documentation

Index

• Constants
• type Agent
  • func (a *Agent) IsLive(threshold time.Duration) bool
• type AvailabilityZone
  • func (az AvailabilityZone) Healthy(ctx context.Context, quorum int, livenessThreshold time.Duration, ds Datastore) (bool, []*Agent, int, error)
  • func (az AvailabilityZone) String() string
• type Datastore
• type Partition
• type Topic
• type UUIDService

Constants

const (
	NoAz = AvailabilityZone("")
)

Types

type Agent

type Agent struct {
	ID               int32     `db:"id"`
	Hostname         string    `db:"hostname"`
	Port             int16     `db:"port"`
	AvailabilityZone string    `db:"availability_zone"`
	LastSeen         time.Time `db:"last_seen"`
}

func (*Agent) IsLive

func (a *Agent) IsLive(threshold time.Duration) bool

type AvailabilityZone

type AvailabilityZone string

func (AvailabilityZone) Healthy

func (az AvailabilityZone) Healthy(ctx context.Context, quorum int, livenessThreshold time.Duration, ds Datastore) (bool, []*Agent, int, error)

Healthy returns true if the availability zone has a quorum of live agents.

The agents returned by this function are considered live within the availability zone, as they will have sent a heartbeat within the defined livenessThreshold.

If the availability zone is empty (i.e. NoAz), then all agents that have sent a heartbeat within the defined livenessThreshold are considered live.

The int returned is the index of the agent that is the leader for the availability zone. In the event that there are no live agents in the availability zone, and the availability zone is not NoAz, then the leader is chosen amongst all live agents. If there is no live agents in the availability zone, and the availability zone is NoAz, then an error is returned.

func (AvailabilityZone) String

func (az AvailabilityZone) String() string

type Datastore

type Datastore interface {
	CreateAgent(ctx context.Context, agent *Agent) (int32, error)
	ReadAgent(ctx context.Context, agentID int32) (*Agent, error)
	ReadAllAgents(ctx context.Context) ([]*Agent, error)
	UpdateAgent(ctx context.Context, agent *Agent) error
	DeleteAgent(ctx context.Context, agentID int32) error
	TouchAgent(ctx context.Context, agentID int32) error

	CreateTopic(ctx context.Context, topic *Topic) error
	ReadTopic(ctx context.Context, topicID string) (*Topic, error)
	ReadAllTopics(ctx context.Context) ([]*Topic, error)
	UpdateTopic(ctx context.Context, topic *Topic) error
	DeleteTopic(ctx context.Context, topicID string) error

	CreatePartition(ctx context.Context, partition *Partition) error
	ReadPartition(ctx context.Context, partitionID string) (*Partition, error)
	ReadAllPartitions(ctx context.Context) ([]*Partition, error)
	UpdatePartition(ctx context.Context, partition *Partition) error
	DeletePartition(ctx context.Context, partitionID string) error

	RoundRobinForAz(ctx context.Context, az string, liveAgents int) (int, error)
}

type Partition

type Partition struct {
	TopicID string `db:"topic_id"`
	ID      int    `db:"partition_id"`
}

type Topic

type Topic struct {
	// The ID is a UUIDv4 to keep inline with KIP-516
	ID       string `db:"topic_id"`
	Name     string `db:"name"`
	Internal bool   `db:"internal"`
}

type UUIDService

type UUIDService interface {
	RandomUUID() (string, error)
	ParseUUID(uuid string) ([]byte, error)
	FormatUUID(uuid []byte) (string, error)
}