common

Documentation

Index

• Constants
• Variables
• func ConfigureLogger()
• func DoWithLabels(ctx context.Context, labels map[string]string, f func())
• func GetPeer(ctx context.Context) string
• func Memoize[T any](provider func() T, cacheTime time.Duration) func() T
• func NewBackOff(ctx context.Context) backoff.BackOff
• func NewBackOffWithInitialInterval(ctx context.Context, initialInterval time.Duration) backoff.BackOff
• func ParseLogLevel(levelStr string) (slog.Level, error)
• func RunProcess(startProcess func() (io.Closer, error))
• func RunProfiling() io.Closer
• func WaitUntilSignal(closers ...io.Closer)
• func Xxh332(key string) uint32
• type ClientPool
  • func NewClientPool(tlsConf *tls.Config) ClientPool
• type Clock
• type ConditionContext
  • func NewConditionContext(locker sync.Locker) ConditionContext
• type MockedClock
  • func (c *MockedClock) Now() time.Time
  • func (c *MockedClock) Set(currentTime int64)
• type RefCount
  • func NewRefCount[T io.Closer](t T) RefCount[T]
• type Set
  • func NewSet[T constraints.Ordered]() Set[T]
  • func NewSetFrom[T constraints.Ordered](i []T) Set[T]
• type Shard
  • func GenerateShards(baseId int64, numShards uint32) []Shard
• type WaitGroup
  • func NewWaitGroup(parties int) WaitGroup

Constants

const (
	MetadataNamespace = "namespace"
	MetadataShardId   = "shard-id"
	DefaultNamespace  = "default"

	DefaultPublicPort   = 6648
	DefaultInternalPort = 6649
	DefaultMetricsPort  = 8080

	MaxSessionTimeout = 5 * time.Minute
	MinSessionTimeout = 2 * time.Second
)

const (
	CodeNotInitialized         codes.Code = 100
	CodeInvalidTerm            codes.Code = 101
	CodeInvalidStatus          codes.Code = 102
	CodeCancelled              codes.Code = 103
	CodeAlreadyClosed          codes.Code = 104
	CodeLeaderAlreadyConnected codes.Code = 105
	CodeNodeIsNotLeader        codes.Code = 106
	CodeNodeIsNotFollower      codes.Code = 107
	CodeInvalidSession         codes.Code = 108
	CodeInvalidSessionTimeout  codes.Code = 109
	CodeNamespaceNotFound      codes.Code = 110
)

const DefaultLogLevel = slog.LevelInfo

const DefaultRpcTimeout = 30 * time.Second

const InternalKeyPrefix = "__oxia/"

InternalKeyPrefix is the prefix of keys used by oxia.

Variables

var (
	ErrorNotInitialized         = status.Error(CodeNotInitialized, "oxia: server not initialized yet")
	ErrorCancelled              = status.Error(CodeCancelled, "oxia: operation was cancelled")
	ErrorInvalidTerm            = status.Error(CodeInvalidTerm, "oxia: invalid term")
	ErrorInvalidStatus          = status.Error(CodeInvalidStatus, "oxia: invalid status")
	ErrorLeaderAlreadyConnected = status.Error(CodeLeaderAlreadyConnected, "oxia: leader is already connected")
	ErrorAlreadyClosed          = status.Error(CodeAlreadyClosed, "oxia: node is shutting down")
	ErrorNodeIsNotLeader        = status.Error(CodeNodeIsNotLeader, "oxia: node is not leader for shard")
	ErrorNodeIsNotFollower      = status.Error(CodeNodeIsNotFollower, "oxia: node is not follower for shard")
	ErrorInvalidSession         = status.Error(CodeInvalidSession, "oxia: session not found")
	ErrorInvalidSessionTimeout  = status.Error(CodeInvalidSessionTimeout, "oxia: invalid session timeout")
	ErrorNamespaceNotFound      = status.Error(CodeNamespaceNotFound, "oxia: namespace not found")
)

var (
	// LogLevel Used for flags.
	LogLevel slog.Level
	// LogJSON Used for flags.
	LogJSON bool
)

var (
	PprofEnable      bool
	PprofBindAddress string
)

var SystemClock = &systemClock{}

Functions

func ConfigureLogger

func ConfigureLogger()

func DoWithLabels

func DoWithLabels(ctx context.Context, labels map[string]string, f func())

DoWithLabels attaches the labels to the current go-routine Pprof context, for the duration of the call to f.

func GetPeer

func GetPeer(ctx context.Context) string

func Memoize

func Memoize[T any](provider func() T, cacheTime time.Duration) func() T

Memoize is used to cache the result of the invocation of a function for a certain amount of time.

func NewBackOff

func NewBackOff(ctx context.Context) backoff.BackOff

func NewBackOffWithInitialInterval

func NewBackOffWithInitialInterval(ctx context.Context, initialInterval time.Duration) backoff.BackOff

func ParseLogLevel

func ParseLogLevel(levelStr string) (slog.Level, error)

ParseLogLevel will convert the slog level configuration to slog.Level values.

func RunProcess

func RunProcess(startProcess func() (io.Closer, error))

func RunProfiling

func RunProfiling() io.Closer

func WaitUntilSignal

func WaitUntilSignal(closers ...io.Closer)

func Xxh332

func Xxh332(key string) uint32

Types

type ClientPool

type ClientPool interface {
	io.Closer
	GetClientRpc(target string) (proto.OxiaClientClient, error)
	GetHealthRpc(target string) (grpc_health_v1.HealthClient, error)
	GetCoordinationRpc(target string) (proto.OxiaCoordinationClient, error)
	GetReplicationRpc(target string) (proto.OxiaLogReplicationClient, error)
}

func NewClientPool

func NewClientPool(tlsConf *tls.Config) ClientPool

type Clock

type Clock interface {
	Now() time.Time
}

type ConditionContext

type ConditionContext interface {
	// Wait atomically unlocks the locker and suspends execution
	// of the calling goroutine. After later resuming execution,
	// Wait locks c.L before returning. Unlike in other systems,
	// Wait cannot return unless awoken by Broadcast or Signal.
	//
	// Because c.L is not locked when Wait first resumes, the caller
	// typically cannot assume that the condition is true when
	// Wait returns. Instead, the caller should Wait in a loop:
	//
	//	lock.Lock()
	//	for !condition() {
	//	    c.Wait(ctx)
	//	}
	//	... make use of condition ...
	//	lock.Unlock()
	Wait(ctx context.Context) error

	// Signal wakes one goroutine waiting on c, if there is any.
	//
	// It is allowed but not required for the caller to hold c.L
	// during the call.
	//
	// Signal() does not affect goroutine scheduling priority; if other goroutines
	// are attempting to lock c.L, they may be awoken before a "waiting" goroutine.
	Signal()

	// Broadcast wakes all goroutines waiting on c.
	//
	// It is allowed but not required for the caller to hold c.L
	// during the call.
	Broadcast()
}

ConditionContext implements a condition variable, a rendezvous point for goroutines waiting for or announcing the occurrence of an event.

This version of condition takes a `context.Context` in the `Wait()` method, to allow for timeouts and cancellations of the operation.

func NewConditionContext

func NewConditionContext(locker sync.Locker) ConditionContext

type MockedClock

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

func (*MockedClock) Now

func (c *MockedClock) Now() time.Time

func (*MockedClock) Set

func (c *MockedClock) Set(currentTime int64)

type RefCount

type RefCount[T io.Closer] interface {
	io.Closer

	Acquire() RefCount[T]

	RefCnt() int32

	Get() T
}

func NewRefCount

func NewRefCount[T io.Closer](t T) RefCount[T]

type Set

type Set[T constraints.Ordered] interface {
	Add(t T)
	Remove(t T)
	Contains(t T) bool
	Count() int
	IsEmpty() bool
	GetSorted() []T
	Complement(other Set[T]) Set[T]
}

func NewSet

func NewSet[T constraints.Ordered]() Set[T]

func NewSetFrom

func NewSetFrom[T constraints.Ordered](i []T) Set[T]

type Shard

type Shard struct {
	Id  int64
	Min uint32
	Max uint32
}

func GenerateShards

func GenerateShards(baseId int64, numShards uint32) []Shard

type WaitGroup

type WaitGroup interface {

	// Wait until all the parties in the group are either done or if there is any failure
	// You should only call wait once
	Wait(ctx context.Context) error

	// Done Signals that one party in the group is done
	Done()

	// Fail Signal that one party has failed in the operation
	Fail(err error)
}

WaitGroup is similar to sync.WaitGroup but adds 2 capabilities:

1. Returning an error if any operation fails
2. Accept a context to cancel the Wait

func NewWaitGroup

func NewWaitGroup(parties int) WaitGroup