utils

README

go.viam.com/utils

This is a set of go utilities you can use via importing go.viam.com/utils.

Examples

This library includes examples that demonstrate grpc functionality for a variety of contexts - see links for more information:

• echo

As a convenience, you can run the make recipes for these examples from the root of this repository via:

make example-{name}/{recipe}

For example, try running a simple echo server with:

make example-echo/run-server

Windows Support

Windows 10 22H2 and up.

Development Dependencies

• bash (from https://gitforwindows.org/ is good)

Support is not well tested yet.

Known Issues

• rpc: ICE between local connections found via ICE mDNS appear to be flaky in the establishment phase.

License check

See https://github.com/viamrobotics/rdk#license-check for instructions on how update our license policy.

License

Copyright 2021-2024 Viam Inc.

Apache 2.0 - See LICENSE file

Documentation

Overview

Package utils TODO

Package utils contains all utility functions that currently have no better home than here.

Index

• Variables
• func ContextMainIterFunc(ctx context.Context) func()
• func ContextMainQuitSignal(ctx context.Context) <-chan os.Signal
• func ContextMainReadyFunc(ctx context.Context) func()
• func ContextWithIterFunc(ctx context.Context, f func()) context.Context
• func ContextWithQuitSignal(ctx context.Context, c <-chan os.Signal) context.Context
• func ContextWithReadyFunc(ctx context.Context, c chan<- struct{}) context.Context
• func ContextualMain[L ILogger](main func(ctx context.Context, args []string, logger L) error, logger L)
• func ContextualMainQuit[L ILogger](main func(ctx context.Context, args []string, logger L) error, logger L)
• func FilterOutError(err, target error) error
• func FindGoroutineLeaks(options ...goleak.Option) error
• func GetAllLocalIPv4s() ([]string, error)
• func ManagedGo(f, onComplete func())
• func MergeContext(ctx, otherCtx context.Context) (context.Context, func())
• func MergeContextWithDeadline(ctx, otherCtx context.Context, deadline time.Time) (context.Context, func())
• func MergeContextWithTimeout(ctx, otherCtx context.Context, timeout time.Duration) (context.Context, func())
• func NewConfigValidationError(path string, err error) error
• func NewConfigValidationFieldRequiredError(path, field string) error
• func NewPlainTextHTTP2Server(handler http.Handler) (*http.Server, error)
• func NewPossiblySecureHTTPServer(handler http.Handler, opts HTTPServerOptions) (*http.Server, error)
• func NewPossiblySecureTCPListenerFromConfig(address string, tlsConfig *tls.Config) (net.Listener, bool, error)
• func NewPossiblySecureTCPListenerFromFile(address, tlsCertFile, tlsKeyFile string) (net.Listener, bool, error)
• func NewPossiblySecureTCPListenerFromMemory(address string, tlsCertPEM, tlsKeyPEM []byte) (net.Listener, bool, error)
• func PanicCapturingGo(f func())
• func PanicCapturingGoWithCallback(f func(), callback func(err interface{}))
• func ParseFlags(args []string, into interface{}) error
• func RandomAlphaString(size int) string
• func RawBytesFromSlice(val interface{}) []byte
• func ReadBytes(ctx context.Context, r io.Reader, toRead int) ([]byte, error)
• func SelectContextOrWait(ctx context.Context, dur time.Duration) bool
• func SelectContextOrWaitChan[T any](ctx context.Context, c <-chan T) bool
• func SlowGoroutineWatcher(dur time.Duration, slowMsg string, logger ZapCompatibleLogger) (<-chan struct{}, func())
• func SlowGoroutineWatcherAfterContext(ctx context.Context, dur time.Duration, slowMsg string, ...) (<-chan struct{}, func())
• func StringSliceRemove(from []string, at int) []string
• func TryReserveRandomPort() (port int, err error)
• func UncheckedError(err error)
• func UncheckedErrorFunc(f func() error)
• func UnmarshalFlags(flagSet *flag.FlagSet, into interface{}) error
• type HTTP2Server
  • func NewHTTP2Server() (*HTTP2Server, error)
  • func (srv *HTTP2Server) Close() error
• type HTTPServerOptions
• type ILogger
• type NetPortFlag
  • func (npf *NetPortFlag) Get() interface{}
  • func (npf *NetPortFlag) Set(val string) error
  • func (npf *NetPortFlag) String() string
• type RefCountedValue
  • func NewRefCountedValue(val interface{}) RefCountedValue
• type StoppableWorkers
  • func NewBackgroundStoppableWorkers(workers ...func(context.Context)) *StoppableWorkers
  • func NewStoppableWorkerWithTicker(tickRate time.Duration, workFn func(context.Context)) *StoppableWorkers
  • func NewStoppableWorkers(ctx context.Context) *StoppableWorkers
  • func (sw *StoppableWorkers) Add(worker func(context.Context))
  • func (sw *StoppableWorkers) Context() context.Context
  • func (sw *StoppableWorkers) Stop()
• type StringSet
  • func NewStringSet(values ...string) StringSet
  • func (ss StringSet) Add(value string)
  • func (ss StringSet) Remove(value string)
  • func (ss StringSet) ToList() []string
• type ZapCompatibleLogger
  • func AddFieldsToLogger(inp ZapCompatibleLogger, args ...interface{}) (loggerRet ZapCompatibleLogger)
  • func Sublogger(inp ZapCompatibleLogger, subname string) (loggerRet ZapCompatibleLogger)

Variables

var Debug = false

Debug is helpful to turn on when the library isn't working quite right.

var ErrInsufficientX509KeyPair = errors.New("must provide both cert and key of an X509 key pair, not just one part")

ErrInsufficientX509KeyPair is returned when an incomplete X509 key pair is used.

var Logger = golog.Global()

Logger is used various parts of the package for informational/debugging purposes.

Functions

func ContextMainIterFunc

func ContextMainIterFunc(ctx context.Context) func()

ContextMainIterFunc returns a function for indicating an iteration of the program has completed.

func ContextMainQuitSignal

func ContextMainQuitSignal(ctx context.Context) <-chan os.Signal

ContextMainQuitSignal returns a signal channel for quits. It may be nil if the value was never set.

func ContextMainReadyFunc

func ContextMainReadyFunc(ctx context.Context) func()

ContextMainReadyFunc returns a function for indicating readiness. This is intended for main functions that block forever (e.g. daemons).

func ContextWithIterFunc

func ContextWithIterFunc(ctx context.Context, f func()) context.Context

ContextWithIterFunc attaches an iteration func to the given context.

func ContextWithQuitSignal

func ContextWithQuitSignal(ctx context.Context, c <-chan os.Signal) context.Context

ContextWithQuitSignal attaches a quit signaler to the given context.

func ContextWithReadyFunc

func ContextWithReadyFunc(ctx context.Context, c chan<- struct{}) context.Context

ContextWithReadyFunc attaches a ready signaler to the given context.

func ContextualMain

func ContextualMain[L ILogger](main func(ctx context.Context, args []string, logger L) error, logger L)

ContextualMain calls a main entry point function with a cancellable context via SIGTERM. This should be called once per process so as to not clobber the signals from Notify.

func ContextualMainQuit

func ContextualMainQuit[L ILogger](main func(ctx context.Context, args []string, logger L) error, logger L)

ContextualMainQuit is the same as ContextualMain but catches quit signals into the provided context accessed via ContextMainQuitSignal.

func FilterOutError

func FilterOutError(err, target error) error

FilterOutError filters out an error based on the given target. For example, if err was context.Canceled and so was the target, this would return nil. Furthermore, if err was a multierr containing a context.Canceled, it would also be filtered out from a new multierr.

func FindGoroutineLeaks

func FindGoroutineLeaks(options ...goleak.Option) error

FindGoroutineLeaks finds any goroutine leaks after a program is done running. This should be used at the end of a main test run or a top-level process run.

func GetAllLocalIPv4s

func GetAllLocalIPv4s() ([]string, error)

GetAllLocalIPv4s finds all the local ips from all interfaces It only returns IPv4 addresses, and tries not to return any loopback addresses.

func ManagedGo

func ManagedGo(f, onComplete func())

ManagedGo keeps the given function alive in the background until it terminates normally.

func MergeContext

func MergeContext(ctx, otherCtx context.Context) (context.Context, func())

MergeContext merges the two given contexts together and returns a new "child" context parented by the first context that will be cancelled either by the returned cancel function or when either of the two initial contexts are canceled. Note: This implies that the values will only come from the first argument's context.

func MergeContextWithDeadline

func MergeContextWithDeadline(ctx, otherCtx context.Context, deadline time.Time) (context.Context, func())

MergeContextWithDeadline merges the two given contexts together and returns a new "child" context parented by the first context that will be cancelled either by the returned cancel function, when either of the two initial contexts are canceled, or when the given deadline lapses. Note: This implies that the values will only come from the first argument's context.

func MergeContextWithTimeout

func MergeContextWithTimeout(ctx, otherCtx context.Context, timeout time.Duration) (context.Context, func())

MergeContextWithTimeout merges the two given contexts together and returns a new "child" context parented by the first context that will be cancelled either by the returned cancel function, when either of the two initial contexts are canceled, or when the given timeout elapses. Note: This implies that the values will only come from the first argument's context.

func NewConfigValidationError

func NewConfigValidationError(path string, err error) error

NewConfigValidationError returns a config validation error occurring at a given path.

func NewConfigValidationFieldRequiredError

func NewConfigValidationFieldRequiredError(path, field string) error

NewConfigValidationFieldRequiredError returns a config validation error for a field missing at a given path.

func NewPlainTextHTTP2Server

func NewPlainTextHTTP2Server(handler http.Handler) (*http.Server, error)

NewPlainTextHTTP2Server returns an http.Server capable of handling HTTP/2 over plaintext via h2c for the given handler.

func NewPossiblySecureHTTPServer

func NewPossiblySecureHTTPServer(handler http.Handler, opts HTTPServerOptions) (*http.Server, error)

NewPossiblySecureHTTPServer returns an http.Server capable of handling HTTP/2 requests that may be plaintext or secured depending on the options.

func NewPossiblySecureTCPListenerFromConfig

func NewPossiblySecureTCPListenerFromConfig(address string, tlsConfig *tls.Config) (net.Listener, bool, error)

NewPossiblySecureTCPListenerFromConfig returns a TCP listener at the given address that is either insecure or TLS based listener depending on presence of certificates in the given TLS Config. If no address is specified, the listener will bind to localhost IPV4 on a random port.

func NewPossiblySecureTCPListenerFromFile

func NewPossiblySecureTCPListenerFromFile(address, tlsCertFile, tlsKeyFile string) (net.Listener, bool, error)

NewPossiblySecureTCPListenerFromFile returns a TCP listener at the given address that is either insecure or TLS based listener depending on presence of the tlsCertFile and tlsKeyFile which are expected to be an X509 key pair. If no address is specified, the listener will bind to localhost IPV4 on a random port.

func NewPossiblySecureTCPListenerFromMemory

func NewPossiblySecureTCPListenerFromMemory(address string, tlsCertPEM, tlsKeyPEM []byte) (net.Listener, bool, error)

NewPossiblySecureTCPListenerFromMemory returns a TCP listener at the given address that is either insecure or TLS based listener depending on presence of the tlsCertPEM and tlsKeyPEM which are expected to be an X509 key pair. If no address is specified, the listener will bind to localhost IPV4 on a random port.

func PanicCapturingGo

func PanicCapturingGo(f func())

PanicCapturingGo spawns a goroutine to run the given function and captures any panic that occurs and logs it.

func PanicCapturingGoWithCallback

func PanicCapturingGoWithCallback(f func(), callback func(err interface{}))

PanicCapturingGoWithCallback spawns a goroutine to run the given function and captures any panic that occurs, logs it, and calls the given callback. The callback can be used for restart functionality.

func ParseFlags

func ParseFlags(args []string, into interface{}) error

ParseFlags parses arguments derived from and into the given into struct.

func RandomAlphaString

func RandomAlphaString(size int) string

RandomAlphaString returns a random alphabetic string of the given size. Note(erd): all random strings are subject to modulus bias; hope that does not matter to you.

func RawBytesFromSlice

func RawBytesFromSlice(val interface{}) []byte

RawBytesFromSlice returns a view of the given slice value. It is valid as long as the given value stays within GC.

func ReadBytes

func ReadBytes(ctx context.Context, r io.Reader, toRead int) ([]byte, error)

ReadBytes ensures that all bytes requested to be read are read into a slice unless an error occurs. If the reader never returns the amount of bytes requested, this will block until the given context is done.

func SelectContextOrWait

func SelectContextOrWait(ctx context.Context, dur time.Duration) bool

SelectContextOrWait either terminates because the given context is done or the given duration elapses. It returns true if the duration elapsed.

func SelectContextOrWaitChan

func SelectContextOrWaitChan[T any](ctx context.Context, c <-chan T) bool

SelectContextOrWaitChan either terminates because the given context is done or the given time channel is received on. It returns true if the channel was received on.

func SlowGoroutineWatcher

func SlowGoroutineWatcher(
	dur time.Duration,
	slowMsg string,
	logger ZapCompatibleLogger,
) (<-chan struct{}, func())

SlowGoroutineWatcher is used to monitor if a goroutine is going "slow". It will first kick off a wait based on 'dur'. If 'dur' elapses before the cancel func is called, then currently running goroutines will be dumped. The returned channel can be used to wait for this background goroutine to finish.

func SlowGoroutineWatcherAfterContext

func SlowGoroutineWatcherAfterContext(
	ctx context.Context,
	dur time.Duration,
	slowMsg string,
	logger ZapCompatibleLogger,
) (<-chan struct{}, func())

SlowGoroutineWatcherAfterContext is used to monitor if a goroutine is going "slow". It will first wait for the given context to be done and then kick off a wait based on 'dur'. If 'dur' elapses before the cancel func is called, then currently running goroutines will be dumped. The returned channel can be used to wait for this background goroutine to finish.

func StringSliceRemove

func StringSliceRemove(from []string, at int) []string

StringSliceRemove removes an element from the slice at the given position.

func TryReserveRandomPort

func TryReserveRandomPort() (port int, err error)

TryReserveRandomPort attempts to "reserve" a random port for later use. It works by listening on a TCP port and immediately closing that listener. In most contexts this is reliable if the port is immediately used after and there is not much port churn. Typically an OS will monotonically increase the port numbers it assigns.

func UncheckedError

func UncheckedError(err error)

UncheckedError is used in places where we really do not care about an error but we want to at least report it. Never use this for closing writers.

func UncheckedErrorFunc

func UncheckedErrorFunc(f func() error)

UncheckedErrorFunc is used in places where we really do not care about an error but we want to at least report it. Never use this for closing writers.

func UnmarshalFlags

func UnmarshalFlags(flagSet *flag.FlagSet, into interface{}) error

UnmarshalFlags unmarshals parsed flags into the given value.

Types

type HTTP2Server

type HTTP2Server struct {
	HTTP1 *http.Server
	HTTP2 *http2.Server
}

HTTP2Server provides dual access to HTTP/2 via a preconfigured HTTP/1 server and a direct access HTTP/2 server.

func NewHTTP2Server

func NewHTTP2Server() (*HTTP2Server, error)

NewHTTP2Server returns an HTTP/2 server. The returned struct contains the http2.Server itself as well as a http.Server that can be used to serve TLS based connections and is also used to gracefully shutdown the HTTP/2 server itself since it does not provide a proper shutdown method.

func (*HTTP2Server) Close

func (srv *HTTP2Server) Close() error

Close shuts down the HTTP/1 server which in turn triggers the HTTP/2 server to shutdown (albeit not immediately).

type HTTPServerOptions

type HTTPServerOptions struct {
	Secure         bool
	TLSAuth        bool
	MaxHeaderBytes int
	Addr           string
}

HTTPServerOptions define options to use when calling NewPossiblySecureHTTPServer.

type ILogger

type ILogger interface {
	Debug(...interface{})
	Info(...interface{})
	Warn(...interface{})
	Fatal(...interface{})
}

ILogger is a basic logging interface for ContextualMain.

type NetPortFlag

type NetPortFlag int

NetPortFlag is used to correctly set and validate a network port.

func (*NetPortFlag) Get

func (npf *NetPortFlag) Get() interface{}

Get returns the value as an integer.

func (*NetPortFlag) Set

func (npf *NetPortFlag) Set(val string) error

Set attempts to set the value as a network port.

func (*NetPortFlag) String

func (npf *NetPortFlag) String() string

String returns the set value.

type RefCountedValue

type RefCountedValue interface {
	// Ref increments the reference count and returns the value.
	Ref() interface{}

	// Deref decrements the reference count and returns if this
	// dereference resulted in the value being unreferenced.
	Deref() (unreferenced bool)
}

RefCountedValue is a utility to "reference count" values in order to destruct them once no one references them. If you don't require that kind of logic, just rely on golang's garbage collection.

func NewRefCountedValue

func NewRefCountedValue(val interface{}) RefCountedValue

NewRefCountedValue returns a new reference counted value for the given value. Its reference count starts at zero but is not released. It is assumed the caller of this will reference it at least once.

type StoppableWorkers

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

StoppableWorkers is a collection of goroutines that can be stopped at a later time.

func NewBackgroundStoppableWorkers

func NewBackgroundStoppableWorkers(workers ...func(context.Context)) *StoppableWorkers

NewBackgroundStoppableWorkers creates a new StoppableWorkers instance. The instance's context will be derived from `context.Background()`. The passed in workers will be `Add`ed. Workers:

• MUST respond appropriately to errors on the context parameter.
• MUST NOT add more workers to the `StoppableWorkers` group to which they belong.

Any `panic`s from workers will be `recover`ed and logged.

func NewStoppableWorkerWithTicker

func NewStoppableWorkerWithTicker(tickRate time.Duration, workFn func(context.Context)) *StoppableWorkers

NewStoppableWorkerWithTicker creates a `StoppableWorkers` object with a single worker that gets called every `tickRate`. Calls to the input `worker` function are serialized. I.e: a slow "work" iteration will just slow down when the next one is called.

func NewStoppableWorkers

func NewStoppableWorkers(ctx context.Context) *StoppableWorkers

NewStoppableWorkers creates a new StoppableWorkers instance. The instance's context will be derived from passed in context.

func (*StoppableWorkers) Add

func (sw *StoppableWorkers) Add(worker func(context.Context))

Add starts up a goroutine for the passed-in function. Workers must respond appropriately to errors on the context parameter.

The worker will not be added if the StoppableWorkers instance has already been stopped. Any `panic`s from workers will be `recover`ed and logged.

func (*StoppableWorkers) Context

func (sw *StoppableWorkers) Context() context.Context

Context gets the context of the StoppableWorkers instance. Using this function is expected to be rare: usually you shouldn't need to interact with the context directly.

func (*StoppableWorkers) Stop

func (sw *StoppableWorkers) Stop()

Stop idempotently shuts down all the goroutines we started up.

type StringSet

type StringSet map[string]struct{}

StringSet represents a mathematical set of string.

func NewStringSet

func NewStringSet(values ...string) StringSet

NewStringSet returns a new string set from the given series of values where duplicates are okay.

func (StringSet) Add

func (ss StringSet) Add(value string)

Add adds the value to the StringSet if the value already exists it will be a no-op.

func (StringSet) Remove

func (ss StringSet) Remove(value string)

Remove removes the value from the StringSet If the value doesn't exist it will be a no-op.

func (StringSet) ToList

func (ss StringSet) ToList() []string

ToList converts the set into a list of strings.

type ZapCompatibleLogger

type ZapCompatibleLogger interface {
	Desugar() *zap.Logger

	Debug(args ...interface{})
	Debugf(template string, args ...interface{})
	Debugw(msg string, keysAndValues ...interface{})

	Info(args ...interface{})
	Infof(template string, args ...interface{})
	Infow(msg string, keysAndValues ...interface{})

	Warn(args ...interface{})
	Warnf(template string, args ...interface{})
	Warnw(msg string, keysAndValues ...interface{})

	Error(args ...interface{})
	Errorf(template string, args ...interface{})
	Errorw(msg string, keysAndValues ...interface{})

	Fatal(args ...interface{})
	Fatalf(template string, args ...interface{})
	Fatalw(msg string, keysAndValues ...interface{})
}

ZapCompatibleLogger is a basic logging interface for golog.Logger (alias for *zap.SugaredLogger) and RDK loggers.

func AddFieldsToLogger

func AddFieldsToLogger(inp ZapCompatibleLogger, args ...interface{}) (loggerRet ZapCompatibleLogger)

AddFieldsToLogger attempts to add fields for logging to a given ZapCompatibleLogger instance. This function uses reflection to dynamically add fields to the provided logger by calling its `WithFields` method if it is an RDK logger. If the logger is not an RDK logger, it logs a debug message and returns the original logger. Args is expected to be a list of key-value pair(s).

func Sublogger

func Sublogger(inp ZapCompatibleLogger, subname string) (loggerRet ZapCompatibleLogger)

Sublogger creates a sublogger from the given ZapCompatibleLogger instance. This function uses reflection to dynamically create a sublogger from the provided logger by calling its `Sublogger` method if it is an RDK logger, or its `Named` method if it is a Zap logger. If neither method is available, it logs a debug message and returns the original logger.