util

Documentation

Index

• Constants
• Variables
• func AuthenticateUser(on bool) middleware.Interface
• func ClientHTTPStatusAndError(err error) (int, 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 ShuffleShardExpectedInstances(shardSize, numZones int) int
• func ShuffleShardExpectedInstancesPerZone(shardSize, numZones int) int
• func ShuffleShardSeed(identifier, zone string) int64
• 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 WriteError(err error, w http.ResponseWriter)
• 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 Log
  • func (l Log) Wrap(next http.Handler) http.Handler
• type RoundTripperFunc
  • func (f RoundTripperFunc) RoundTrip(req *http.Request) (*http.Response, error)

Constants

const (
	ErrClientCanceled   = "The request was cancelled by the client."
	ErrDeadlineExceeded = "" /* 160-byte string literal not displayed */
)

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

const StatusClientClosedRequest = 499

StatusClientClosedRequest is the status code for when a client request cancellation of an http request

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 {
					WriteError(httpgrpc.Errorf(http.StatusInternalServerError, "error while processing request: %v", panicError(p)), w)
				}
			}()
			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 ClientHTTPStatusAndError

func ClientHTTPStatusAndError(err error) (int, error)

ClientHTTPStatusAndError returns error and http status that is "safe" to return to client without exposing any implementation details.

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 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 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 WriteError

func WriteError(err error, w http.ResponseWriter)

WriteError write a go error with the correct status code.

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 Log

type Log struct {
	Log                   logging.Interface
	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)