util

Documentation

Overview

CopyFile copies the contents of the file named src to the file named by dst. The file will be created if it does not already exist. Attempt to do a symlink first if it fails. The file mode will be copied from the source and the copied data is synced/flushed to stable storage. If the destination file exists, all it's contents will be replaced by the contents of the source file.

Index

• Constants
• Variables
• func AuthenticateUser(on bool) middleware.Interface
• func CopyDir(src string, dst string) (err error)
• func CopyFile(src, dst string) (err error)
• func DiffConfig(defaultConfig, actualConfig map[interface{}]interface{}) (map[interface{}]interface{}, error)
• func DurationWithJitter(input time.Duration, variancePerc float64) time.Duration
• func DurationWithNegativeJitter(input time.Duration, variancePerc float64) time.Duration
• func DurationWithPositiveJitter(input time.Duration, variancePerc float64) time.Duration
• func FormatTimeMillis(ms int64) string
• func FormatTimeModel(t model.Time) string
• func GetFirstAddressOf(names []string) (string, error)
• func InstrumentedHTTPClient() *http.Client
• func IsWSHandshakeRequest(req *http.Request) bool
• func LoggerWithContext(ctx context.Context, l log.Logger) log.Logger
• func LoggerWithTraceID(traceID string, l log.Logger) log.Logger
• func LoggerWithUserID(tenantID string, l log.Logger) log.Logger
• func LoggerWithUserIDs(tenantIDs []string, l log.Logger) log.Logger
• func NewDisableableTicker(interval time.Duration) (func(), <-chan time.Time)
• func NewHTTPMetricMiddleware(mux *mux.Router, namespace string, reg prometheus.Registerer) (middleware.Interface, error)
• func NewLogInterceptor(logger log.Logger) connect.UnaryInterceptorFunc
• func ParseTime(s string) (int64, error)
• func RecoverPanic(f func() error) func() error
• func RegisterOrGet[T prometheus.Collector](reg prometheus.Registerer, c T) T
• func RenderHTTPResponse(w http.ResponseWriter, v interface{}, t *template.Template, r *http.Request)
• func ShuffleShardExpectedInstances(shardSize, numZones int) int
• func ShuffleShardExpectedInstancesPerZone(shardSize, numZones int) int
• func ShuffleShardSeed(identifier, zone string) int64
• func SplitTimeRangeByResolution(start, end time.Time, resolutions []time.Duration, fn func(TimeRange))
• func StableHash(ls labels.Labels) uint64
• func StringsContain(values []string, search string) bool
• func StringsMap(values []string) map[string]bool
• func TimeFromMillis(ms int64) time.Time
• func TimeToMillis(t time.Time) int64
• func UnsafeGetBytes(s string) []byte
• func UnsafeGetString(buf []byte) string
• func WithSourceIPs(sourceIPs string, l log.Logger) log.Logger
• func WithTimeout(timeout time.Duration) connect.Interceptor
• func WrapWithInstrumentedHTTPTransport(next http.RoundTripper) http.RoundTripper
• func WriteHTMLResponse(w http.ResponseWriter, message string)
• func WriteJSONResponse(w http.ResponseWriter, v interface{})
• func WriteTextResponse(w http.ResponseWriter, message string)
• func WriteYAMLResponse(w http.ResponseWriter, v interface{})
• func YAMLMarshalUnmarshal(in interface{}) (map[interface{}]interface{}, error)
• func YoloBuf(s string) []byte
• type ActiveUsers
  • func NewActiveUsers() *ActiveUsers
  • func (m *ActiveUsers) PurgeInactiveUsers(deadline int64) []string
  • func (m *ActiveUsers) UpdateUserTimestamp(userID string, ts int64)
• type ActiveUsersCleanupService
  • func NewActiveUsersCleanupService(cleanupInterval, inactiveTimeout time.Duration, cleanupFn func(string)) *ActiveUsersCleanupService
  • func NewActiveUsersCleanupWithDefaultValues(cleanupFn func(string)) *ActiveUsersCleanupService
  • func (s *ActiveUsersCleanupService) UpdateUserTimestamp(user string, now time.Time)
• type CommonRingConfig
  • func (cfg *CommonRingConfig) RegisterFlags(flagPrefix, kvStorePrefix, componentPlural string, f *flag.FlagSet, ...)
  • func (cfg *CommonRingConfig) ToRingConfig() ring.Config
• type Log
  • func (l Log) Wrap(next http.Handler) http.Handler
• type RoundTripperFunc
  • func (f RoundTripperFunc) RoundTrip(req *http.Request) (*http.Response, error)
• type TimeRange

Constants

const (
	// Sharding strategies.
	ShardingStrategyDefault = "default"
	ShardingStrategyShuffle = "shuffle-sharding"
)

Variables

var (
	RecoveryHTTPMiddleware = middleware.Func(func(next http.Handler) http.Handler {
		return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
			defer func() {
				if p := recover(); p != nil {
					httputil.Error(w, httpgrpc.Errorf(http.StatusInternalServerError, "error while processing request: %v", panicError(p)))
				}
			}()
			next.ServeHTTP(w, req)
		})
	})
	RecoveryGRPCStreamInterceptor = grpc_recovery.StreamServerInterceptor(grpc_recovery.WithRecoveryHandler(panicError))
	RecoveryGRPCUnaryInterceptor  = grpc_recovery.UnaryServerInterceptor(grpc_recovery.WithRecoveryHandler(panicError))
)

var Logger = log.NewNopLogger()

Logger is a global logger to use only where you cannot inject a logger.

Functions

func AuthenticateUser

func AuthenticateUser(on bool) middleware.Interface

AuthenticateUser propagates the user ID from HTTP headers back to the request's context. If on is false, it will inject the default tenant ID.

func CopyDir

func CopyDir(src string, dst string) (err error)

CopyDir recursively copies a directory tree, attempting to preserve permissions. Source directory must exist, destination directory must *not* exist. Symlinks are ignored and skipped.

func CopyFile

func CopyFile(src, dst string) (err error)

func DiffConfig

func DiffConfig(defaultConfig, actualConfig map[interface{}]interface{}) (map[interface{}]interface{}, error)

DiffConfig utility function that returns the diff between two config map objects

func DurationWithJitter

func DurationWithJitter(input time.Duration, variancePerc float64) time.Duration

DurationWithJitter returns random duration from "input - input*variance" to "input + input*variance" interval.

func DurationWithNegativeJitter

func DurationWithNegativeJitter(input time.Duration, variancePerc float64) time.Duration

DurationWithNegativeJitter returns random duration from "input - input*variance" to "input" interval.

func DurationWithPositiveJitter

func DurationWithPositiveJitter(input time.Duration, variancePerc float64) time.Duration

DurationWithPositiveJitter returns random duration from "input" to "input + input*variance" interval.

func FormatTimeMillis

func FormatTimeMillis(ms int64) string

FormatTimeMillis returns a human readable version of the input time (in milliseconds).

func FormatTimeModel

func FormatTimeModel(t model.Time) string

FormatTimeModel returns a human readable version of the input time.

func GetFirstAddressOf

func GetFirstAddressOf(names []string) (string, error)

GetFirstAddressOf returns the first IPv4 address of the supplied interface names, omitting any 169.254.x.x automatic private IPs if possible.

func InstrumentedHTTPClient

func InstrumentedHTTPClient() *http.Client

InstrumentedHTTPClient returns a HTTP client with tracing instrumented default RoundTripper.

func IsWSHandshakeRequest

func IsWSHandshakeRequest(req *http.Request) bool

IsWSHandshakeRequest returns true if the given request is a websocket handshake request.

func LoggerWithContext

func LoggerWithContext(ctx context.Context, l log.Logger) log.Logger

LoggerWithContext returns a Logger that has information about the current user or users and trace in its details.

e.g.

log = util.WithContext(ctx, log)
# level=error tenant=user-1|user-2 traceID=123abc msg="Could not chunk chunks" err="an error"
level.Error(log).Log("msg", "Could not chunk chunks", "err", err)

func LoggerWithTraceID

func LoggerWithTraceID(traceID string, l log.Logger) log.Logger

LoggerWithTraceID returns a Logger that has information about the traceID in its details.

func LoggerWithUserID

func LoggerWithUserID(tenantID string, l log.Logger) log.Logger

LoggerWithUserID returns a Logger that has information about the current user in its details.

func LoggerWithUserIDs

func LoggerWithUserIDs(tenantIDs []string, l log.Logger) log.Logger

LoggerWithUserIDs returns a Logger that has information about the current user or users (separated by "|") in its details.

func NewDisableableTicker

func NewDisableableTicker(interval time.Duration) (func(), <-chan time.Time)

NewDisableableTicker essentially wraps NewTicker but allows the ticker to be disabled by passing zero duration as the interval. Returns a function for stopping the ticker, and the ticker channel.

func NewHTTPMetricMiddleware

func NewHTTPMetricMiddleware(mux *mux.Router, namespace string, reg prometheus.Registerer) (middleware.Interface, error)

NewHTTPMetricMiddleware creates a new middleware that automatically instruments HTTP requests from the given router.

func NewLogInterceptor

func NewLogInterceptor(logger log.Logger) connect.UnaryInterceptorFunc

LogRequest logs the request parameters. It logs all kinds of requests.

func ParseTime

func ParseTime(s string) (int64, error)

ParseTime parses the string into an int64, milliseconds since epoch.

func RecoverPanic

func RecoverPanic(f func() error) func() error

RecoverPanic is a helper function to recover from panic and return an error.

func RegisterOrGet

func RegisterOrGet[T prometheus.Collector](reg prometheus.Registerer, c T) T

RegisterOrGet registers the collector c with the provided registerer. If the registerer is nil, the collector is returned without registration. If the collector is already registered, the existing collector is returned.

func RenderHTTPResponse

func RenderHTTPResponse(w http.ResponseWriter, v interface{}, t *template.Template, r *http.Request)

RenderHTTPResponse either responds with JSON or a rendered HTML page using the passed in template by checking the Accepts header.

func ShuffleShardExpectedInstances

func ShuffleShardExpectedInstances(shardSize, numZones int) int

ShuffleShardExpectedInstances returns the total number of instances that should be selected for a given tenant. If zone-aware replication is disabled, the input numZones should be 1.

func ShuffleShardExpectedInstancesPerZone

func ShuffleShardExpectedInstancesPerZone(shardSize, numZones int) int

ShuffleShardExpectedInstancesPerZone returns the number of instances that should be selected for each zone when zone-aware replication is enabled. The algorithm expects the shard size to be divisible by the number of zones, in order to have nodes balanced across zones. If it's not, we do round up.

func ShuffleShardSeed

func ShuffleShardSeed(identifier, zone string) int64

ShuffleShardSeed returns seed for random number generator, computed from provided identifier.

func SplitTimeRangeByResolution

func SplitTimeRangeByResolution(start, end time.Time, resolutions []time.Duration, fn func(TimeRange))

SplitTimeRangeByResolution splits the given time range into the minimal number of non-overlapping sub-ranges aligned with resolutions. All ranges have inclusive start and end; one millisecond step.

func StableHash

func StableHash(ls labels.Labels) uint64

StableHash is a labels hashing implementation which is guaranteed to not change over time. This function should be used whenever labels hashing backward compatibility must be guaranteed.

func StringsContain

func StringsContain(values []string, search string) bool

StringsContain returns true if the search value is within the list of input values.

func StringsMap

func StringsMap(values []string) map[string]bool

StringsMap returns a map where keys are input values.

func TimeFromMillis

func TimeFromMillis(ms int64) time.Time

TimeFromMillis is a helper to turn milliseconds -> time.Time

func TimeToMillis

func TimeToMillis(t time.Time) int64

func UnsafeGetBytes

func UnsafeGetBytes(s string) []byte

func UnsafeGetString

func UnsafeGetString(buf []byte) string

func WithSourceIPs

func WithSourceIPs(sourceIPs string, l log.Logger) log.Logger

WithSourceIPs returns a Logger that has information about the source IPs in its details.

func WithTimeout

func WithTimeout(timeout time.Duration) connect.Interceptor

NewTimeoutInterceptor returns a new timeout interceptor.

func WrapWithInstrumentedHTTPTransport

func WrapWithInstrumentedHTTPTransport(next http.RoundTripper) http.RoundTripper

WrapWithInstrumentedHTTPTransport wraps the given RoundTripper with an tracing instrumented one.

func WriteHTMLResponse

func WriteHTMLResponse(w http.ResponseWriter, message string)

WriteHTMLResponse sends message as text/html response with 200 status code.

func WriteJSONResponse

func WriteJSONResponse(w http.ResponseWriter, v interface{})

WriteJSONResponse writes some JSON as a HTTP response.

func WriteTextResponse

func WriteTextResponse(w http.ResponseWriter, message string)

WriteTextResponse sends message as text/plain response with 200 status code.

func WriteYAMLResponse

func WriteYAMLResponse(w http.ResponseWriter, v interface{})

WriteYAMLResponse writes some YAML as a HTTP response.

func YAMLMarshalUnmarshal

func YAMLMarshalUnmarshal(in interface{}) (map[interface{}]interface{}, error)

YAMLMarshalUnmarshal utility function that converts a YAML interface in a map doing marshal and unmarshal of the parameter

func YoloBuf

func YoloBuf(s string) []byte

Types

type ActiveUsers

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

ActiveUsers keeps track of latest user's activity timestamp, and allows purging users that are no longer active.

func NewActiveUsers

func NewActiveUsers() *ActiveUsers

func (*ActiveUsers) PurgeInactiveUsers

func (m *ActiveUsers) PurgeInactiveUsers(deadline int64) []string

PurgeInactiveUsers removes users that were last active before given deadline, and returns removed users.

func (*ActiveUsers) UpdateUserTimestamp

func (m *ActiveUsers) UpdateUserTimestamp(userID string, ts int64)

type ActiveUsersCleanupService

type ActiveUsersCleanupService struct {
	services.Service
	// contains filtered or unexported fields
}

ActiveUsersCleanupService tracks active users, and periodically purges inactive ones while running.

func NewActiveUsersCleanupService

func NewActiveUsersCleanupService(cleanupInterval, inactiveTimeout time.Duration, cleanupFn func(string)) *ActiveUsersCleanupService

func NewActiveUsersCleanupWithDefaultValues

func NewActiveUsersCleanupWithDefaultValues(cleanupFn func(string)) *ActiveUsersCleanupService

func (*ActiveUsersCleanupService) UpdateUserTimestamp

func (s *ActiveUsersCleanupService) UpdateUserTimestamp(user string, now time.Time)

type CommonRingConfig

type CommonRingConfig struct {
	// KV store details
	KVStore          kv.Config     `yaml:"kvstore" doc:"description=The key-value store used to share the hash ring across multiple instances."`
	HeartbeatPeriod  time.Duration `yaml:"heartbeat_period" category:"advanced"`
	HeartbeatTimeout time.Duration `yaml:"heartbeat_timeout" category:"advanced"`

	// Instance details
	InstanceID             string   `yaml:"instance_id" doc:"default=<hostname>" category:"advanced"`
	InstanceInterfaceNames []string `yaml:"instance_interface_names" doc:"default=[<private network interfaces>]"`
	InstancePort           int      `yaml:"instance_port" category:"advanced"`
	InstanceAddr           string   `yaml:"instance_addr" category:"advanced"`
	EnableIPv6             bool     `yaml:"instance_enable_ipv6" category:"advanced"`

	// Injected internally
	ListenPort int `yaml:"-"`
}

CommonRingConfig is the configuration commonly used by components that use a ring for various coordination tasks such as sharding or service discovery.

func (*CommonRingConfig) RegisterFlags

func (cfg *CommonRingConfig) RegisterFlags(flagPrefix, kvStorePrefix, componentPlural string, f *flag.FlagSet, logger log.Logger)

RegisterFlags adds the flags required to config this to the given FlagSet

func (*CommonRingConfig) ToRingConfig

func (cfg *CommonRingConfig) ToRingConfig() ring.Config

type Log

type Log struct {
	Log                   log.Logger
	LogRequestHeaders     bool // LogRequestHeaders true -> dump http headers at debug log level
	LogRequestAtInfoLevel bool // LogRequestAtInfoLevel true -> log requests at info log level
	SourceIPs             *middleware.SourceIPExtractor
}

Log middleware logs http requests

func (Log) Wrap

func (l Log) Wrap(next http.Handler) http.Handler

Wrap implements Middleware

type RoundTripperFunc

type RoundTripperFunc func(req *http.Request) (*http.Response, error)

func (RoundTripperFunc) RoundTrip

func (f RoundTripperFunc) RoundTrip(req *http.Request) (*http.Response, error)

type TimeRange

type TimeRange struct {
	Start      time.Time
	End        time.Time
	Resolution time.Duration
}