util

Documentation

Index

• Constants
• func AbsDiff[T constraints.Unsigned](x, y T) T
• func AddHandler[T any, R any](logger *zap.Logger, mux *http.ServeMux, endpoint string, method string, ...)
• func AtomicMax[A AtomicInt[I], I constraints.Integer](a A, i I) I
• func MakePPROF(addr string) *http.Server
• func NewCondChannelPair() (CondChannelSender, CondChannelReceiver)
• func NewSingleSignalPair[T any]() (SignalSender[T], SignalReceiver[T])
• func PodCompleted(pod *corev1.Pod) bool
• func PodNameFields(pod *corev1.Pod) zap.Field
• func PodPreferredAZIfPresent(pod *corev1.Pod) string
• func PodReady(pod *corev1.Pod) bool
• func PodStartedBefore(p, q *corev1.Pod) bool
• func RegisterMetric[P prometheus.Collector](reg prometheus.Registerer, collector P) P
• func RootError(err error) error
• func SaturatingSub[T constraints.Unsigned](x, y T) T
• func StartPrometheusMetricsServer(ctx context.Context, logger *zap.Logger, port uint16, reg *prometheus.Registry) error
• func VMNameFields(vm *vmapi.VirtualMachine) zap.Field
• type AtomicInt
• type BroadcastReceiver
  • func (r *BroadcastReceiver) Awake()
  • func (r *BroadcastReceiver) Wait() <-chan struct{}
• type Broadcaster
  • func NewBroadcaster() *Broadcaster
  • func (b *Broadcaster) Broadcast()
  • func (b *Broadcaster) NewReceiver() BroadcastReceiver
• type ChanMutex
  • func NewChanMutex() ChanMutex
  • func (m *ChanMutex) DeadlockChecker(timeout, delay time.Duration) func(ctx context.Context)
  • func (m *ChanMutex) Lock()
  • func (m *ChanMutex) TryLock(ctx context.Context) error
  • func (m *ChanMutex) Unlock()
  • func (m *ChanMutex) WaitLock() <-chan struct{}
• type CondChannelReceiver
  • func (c *CondChannelReceiver) Consume()
  • func (c *CondChannelReceiver) Recv() <-chan struct{}
• type CondChannelSender
  • func (c *CondChannelSender) Send()
  • func (c *CondChannelSender) Unsend() bool
• type NamespacedName
  • func GetNamespacedName(obj metav1.ObjectMetaAccessor) NamespacedName
  • func TryPodOwnerVirtualMachine(pod *corev1.Pod) *NamespacedName
  • func TryPodOwnerVirtualMachineMigration(pod *corev1.Pod) *NamespacedName
  • func (n NamespacedName) Format(state fmt.State, verb rune)
  • func (n NamespacedName) MarshalLogObject(enc zapcore.ObjectEncoder) error
• type RecentCounter
  • func NewRecentCounter(interval time.Duration) *RecentCounter
  • func (rc *RecentCounter) Get() uint
  • func (rc *RecentCounter) Inc()
• type SignalReceiver
  • func (s SignalReceiver[T]) Close()
  • func (s SignalReceiver[T]) Recv() <-chan T
• type SignalSender
  • func (s SignalSender[T]) Send(data T)
• type TimeRange
  • func NewTimeRange(units time.Duration, minTime, maxTime int) *TimeRange
  • func (r TimeRange) Random() time.Duration

Constants

const Separator = '/'

Functions

func AbsDiff

func AbsDiff[T constraints.Unsigned](x, y T) T

AbsDiff returns the absolute value of the difference between x and y

func AddHandler

func AddHandler[T any, R any](
	logger *zap.Logger,
	mux *http.ServeMux,
	endpoint string,
	method string,
	reqTypeName string,
	handle func(context.Context, *zap.Logger, *T) (_ *R, statusCode int, _ error),
)

AddHandler is a helper function to wrap the handle function with JSON [de]serialization and check that the HTTP method is correct

The provided logPrefix is prepended to every log line emitted by the wrapped handler function, to offer distinction where that's useful.

func AtomicMax

func AtomicMax[A AtomicInt[I], I constraints.Integer](a A, i I) I

AtomicMax atomically sets a to the maximum of *a and i, returning the old value at a.

On ISAs without atomic maximum/minimum instructions, a fallback is typically implemented as the Load + CompareAndSwap loop that this function uses. At time of writing (Go 1.20), the Go standard library does not include atomic maximum/minimum functions.

This function is lock-free but not wait-free.

func MakePPROF

func MakePPROF(addr string) *http.Server

func NewCondChannelPair

func NewCondChannelPair() (CondChannelSender, CondChannelReceiver)

NewCondChannelPair creates a sender/receiver pair for a sync.Cond-like interface

The differences from sync.Cond are that receiving is exposed through a channel (so it can be select-ed) and there is no equivalent to (*Cond).Broadcast()

func NewSingleSignalPair

func NewSingleSignalPair[T any]() (SignalSender[T], SignalReceiver[T])

func PodCompleted

func PodCompleted(pod *corev1.Pod) bool

PodCompleted returns true iff all of the Pod's containers have stopped and will not be restarted

func PodNameFields

func PodNameFields(pod *corev1.Pod) zap.Field

func PodPreferredAZIfPresent

func PodPreferredAZIfPresent(pod *corev1.Pod) string

PodPreferredAZIfPresent returns the desired availability zone of the Pod, if it has one

func PodReady

func PodReady(pod *corev1.Pod) bool

PodReady returns true iff the pod is marked as ready (as determined by the pod's Status.Conditions)

func PodStartedBefore

func PodStartedBefore(p, q *corev1.Pod) bool

PodStartedBefore returns true iff Pod p started before Pod q

func RegisterMetric

func RegisterMetric[P prometheus.Collector](reg prometheus.Registerer, collector P) P

func RootError

func RootError(err error) error

RootError returns the root cause of the error, calling errors.Unwrap until it returns nil

func SaturatingSub

func SaturatingSub[T constraints.Unsigned](x, y T) T

SaturatingSub returns x - y if x >= y, otherwise zero

func StartPrometheusMetricsServer

func StartPrometheusMetricsServer(ctx context.Context, logger *zap.Logger, port uint16, reg *prometheus.Registry) error

Starts the prometheus server in a background thread. Returns error if binding on the port fails.

func VMNameFields

func VMNameFields(vm *vmapi.VirtualMachine) zap.Field

Types

type AtomicInt

type AtomicInt[I any] interface {
	Add(delta I) (new I)                      //nolint:predeclared // same var names as methods
	CompareAndSwap(old, new I) (swapped bool) //nolint:predeclared // same var names as methods
	Load() I
	Store(val I)
	Swap(new I) (old I) //nolint:predeclared // same var names as methods
}

AtomicInt represents the shared interface provided by various atomic.<NAME> integers

This interface type is primarily used by AtomicMax.

type BroadcastReceiver

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

func (*BroadcastReceiver) Awake

func (r *BroadcastReceiver) Awake()

Awake marks the most recent broadcast event as received, so that the next call to Wait returns a channel that will only be closed once there's been a new event after this call to Awake.

func (*BroadcastReceiver) Wait

func (r *BroadcastReceiver) Wait() <-chan struct{}

Wait returns a channel that will be closed once there has been an event broadcasted since the BroadcastReceiver was created, or the last call to Awake().

Typical usage of Wait will involve selecting on the channel returned and calling Awake immediately in the branch handling the event, for example:

select {
case <-ctx.Done():
    return
case <-receiver.Wait():
    receiver.Awake()
    ...
}

type Broadcaster

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

func NewBroadcaster

func NewBroadcaster() *Broadcaster

func (*Broadcaster) Broadcast

func (b *Broadcaster) Broadcast()

Broadcast sends a signal to all receivers

func (*Broadcaster) NewReceiver

func (b *Broadcaster) NewReceiver() BroadcastReceiver

NewReceiver creates a new BroadcastReceiver that will receive only future broadcasted events.

It's generally not recommended to call (*BroadcastReceiver).Wait() on a single BroadcastReceiver from more than one thread at a time, although it *is* thread-safe.

type ChanMutex

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

ChanMutex is a select-able mutex

It is fair if and only if receiving on a channel is fair. As of Go 1.19/2022-01-17, receiving on a channel appears to be fair. However: this is a runtime implementation detail, and so it may change without notice in the future.

Unlike sync.Mutex, ChanMutex requires initialization before use because it's basically just a channel.

Also unlike sync.Mutex, a ChanMutex may be copied without issue (again, because it's just a channel).

func NewChanMutex

func NewChanMutex() ChanMutex

NewChanMutex creates a new ChanMutex

func (*ChanMutex) DeadlockChecker

func (m *ChanMutex) DeadlockChecker(timeout, delay time.Duration) func(ctx context.Context)

DeadlockChecker creates a function that, when called, periodically attempts to acquire the lock, panicking if it fails

The returned function exits when the context is done.

func (*ChanMutex) Lock

func (m *ChanMutex) Lock()

Lock locks m

This method is semantically equivalent to sync.Mutex.Lock

func (*ChanMutex) TryLock

func (m *ChanMutex) TryLock(ctx context.Context) error

TryLock blocks until locking m succeeds or the context is cancelled

If the context is cancelled while waiting to lock m, the lock will be left unchanged and ctx.Err() will be returned.

func (*ChanMutex) Unlock

func (m *ChanMutex) Unlock()

Unlock unlocks m

This method is semantically equivalent to sync.Mutex.Unlock

func (*ChanMutex) WaitLock

func (m *ChanMutex) WaitLock() <-chan struct{}

WaitLock is like Lock, but instead returns a channel

If receiving on the channel succeeds, the caller "holds" the lock and must now be responsible for Unlock-ing it.

type CondChannelReceiver

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

CondChannelReceiver is the receiving half of a sync.Cond-like interface

func (*CondChannelReceiver) Consume

func (c *CondChannelReceiver) Consume()

Consume removes any existing signal created by Send, requiring an additional Send to be made before the receiving on Recv will unblock

This method is non-blocking.

func (*CondChannelReceiver) Recv

func (c *CondChannelReceiver) Recv() <-chan struct{}

Recv returns a channel for which receiving will complete either (a) immediately, if Send has been called without Consume or another receive since; or (b) as soon as Send is next called

This method is non-blocking but receiving on the returned channel may block.

type CondChannelSender

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

CondChannelSender is the sending half of a sync.Cond-like interface

func (*CondChannelSender) Send

func (c *CondChannelSender) Send()

Send performs a non-blocking notify of the associated CondChannelReceiver

If there is currently a receiver waiting via Recv, then this will immediately wake them. Otherwise, the next receive on the channel returned by Recv will complete immediately.

func (*CondChannelSender) Unsend

func (c *CondChannelSender) Unsend() bool

Unsend cancels an existing signal that has been sent but not yet received.

It returns whether there was a signal to be cancelled.

type NamespacedName

type NamespacedName struct {
	Namespace string `json:"namespace"`
	Name      string `json:"name"`
}

NamespacedName represents a resource name with the namespace it's in.

When printed with '%v', NamespacedName is rendered as "<namespace>/<name>". Printing with '%+v' or '%#v' renders as it would normally.

func GetNamespacedName

func GetNamespacedName(obj metav1.ObjectMetaAccessor) NamespacedName

func TryPodOwnerVirtualMachine

func TryPodOwnerVirtualMachine(pod *corev1.Pod) *NamespacedName

TryPodOwnerVirtualMachine returns the name of the VirtualMachine that owns the pod, if there is one that does. Otherwise returns nil.

func TryPodOwnerVirtualMachineMigration

func TryPodOwnerVirtualMachineMigration(pod *corev1.Pod) *NamespacedName

TryPodOwnerVirtualMachineMigration returns the name of the VirtualMachineMigration that owns the pod, if there is one. Otherwise returns nil.

func (NamespacedName) Format

func (n NamespacedName) Format(state fmt.State, verb rune)

func (NamespacedName) MarshalLogObject

func (n NamespacedName) MarshalLogObject(enc zapcore.ObjectEncoder) error

MarshalLogObject implements zapcore.ObjectMarshaler, so that NamespacedName can be used with zap.Object

type RecentCounter

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

RecentCounter is a struct that keeps track of recent timestamps within a given interval.

func NewRecentCounter

func NewRecentCounter(interval time.Duration) *RecentCounter

func (*RecentCounter) Get

func (rc *RecentCounter) Get() uint

Get returns the number of recent timestamps stored in the RecentCounter.

func (*RecentCounter) Inc

func (rc *RecentCounter) Inc()

Inc increments the counter and adds the current timestamp to the list of timestamps.

type SignalReceiver

type SignalReceiver[T any] struct {
	// contains filtered or unexported fields
}

func (SignalReceiver[T]) Close

func (s SignalReceiver[T]) Close()

func (SignalReceiver[T]) Recv

func (s SignalReceiver[T]) Recv() <-chan T

type SignalSender

type SignalSender[T any] struct {
	// contains filtered or unexported fields
}

func (SignalSender[T]) Send

func (s SignalSender[T]) Send(data T)

type TimeRange

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

func NewTimeRange

func NewTimeRange(units time.Duration, minTime, maxTime int) *TimeRange

func (TimeRange) Random

func (r TimeRange) Random() time.Duration

Random returns a random time.Duration within the range