syncutil

Documentation

Index

• func BackoffWithContext(ctx context.Context, backoff wait.Backoff, condition wait.ConditionFunc) error
• func GetKeyForSyncMetrics(namespace string, name string) string
• func WaitAll(w ...Waiter) error
• type ConcurrentErrorSlice
  • func NewConcurrentErrorSlice() ConcurrentErrorSlice
  • func (c ConcurrentErrorSlice) Append(e error) ConcurrentErrorSlice
  • func (c ConcurrentErrorSlice) GetSlice() []error
  • func (c ConcurrentErrorSlice) Last() error
• type MetricsCache
  • func NewMetricsCache() *MetricsCache
  • func (c *MetricsCache) AddKind(key string)
  • func (c *MetricsCache) AddObject(key string, t Tags)
  • func (c *MetricsCache) DeleteObject(key string)
  • func (c *MetricsCache) GetTags(key string) *Tags
  • func (c *MetricsCache) HasObject(key string) bool
  • func (c *MetricsCache) ReportSync()
  • func (c *MetricsCache) ResetCache()
• type Reporter
  • func NewStatsReporter() (*Reporter, error)
  • func (r *Reporter) ReportLastSync() error
  • func (r *Reporter) ReportSync(t Tags, v int64) error
  • func (r *Reporter) ReportSyncDuration(d time.Duration) error
• type SingleRunner
  • func NewSingleRunner(errChan chan<- error) *SingleRunner
  • func (s *SingleRunner) Cancel(key string)
  • func (s *SingleRunner) Go(ctx context.Context, key string, f func(context.Context, chan<- error))
  • func (s *SingleRunner) Wait()
• type SyncBool
  • func (b *SyncBool) Get() bool
  • func (b *SyncBool) Set(v bool)
• type Tags
• type Waiter

Functions

func BackoffWithContext

func BackoffWithContext(ctx context.Context, backoff wait.Backoff, condition wait.ConditionFunc) error

BackoffWithContext repeats a condition check with exponential backoff, exiting early if the provided context is canceled.

It repeatedly checks the condition and then sleeps, using `backoff.Step()` to determine the length of the sleep and adjust Duration and Steps. Stops and returns as soon as: 1. the condition check returns true or an error, or 2. the context is canceled. In case (1) the returned error is what the condition function returned. In all other cases, wait.ErrorInterrupted is returned.

Adapted from wait.ExponentialBackoff in https://github.com/kubernetes/kubernetes/blob/master/staging/src/k8s.io/apimachinery/pkg/util/wait/wait.go

func GetKeyForSyncMetrics

func GetKeyForSyncMetrics(namespace string, name string) string

func WaitAll

func WaitAll(w ...Waiter) error

WaitAll waits (blocks) for multiple Wait()-ables and returns the first non-nil error.

Types

type ConcurrentErrorSlice

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

func NewConcurrentErrorSlice

func NewConcurrentErrorSlice() ConcurrentErrorSlice

func (ConcurrentErrorSlice) Append

func (c ConcurrentErrorSlice) Append(e error) ConcurrentErrorSlice

func (ConcurrentErrorSlice) GetSlice

func (c ConcurrentErrorSlice) GetSlice() []error

func (ConcurrentErrorSlice) Last

func (c ConcurrentErrorSlice) Last() error

type MetricsCache

type MetricsCache struct {
	KnownKinds map[string]bool
	Cache      map[string]Tags
	// contains filtered or unexported fields
}

func NewMetricsCache

func NewMetricsCache() *MetricsCache

func (*MetricsCache) AddKind

func (c *MetricsCache) AddKind(key string)

need to know encountered kinds to reset metrics for that kind this is a known memory leak footprint should naturally reset on Pod upgrade b/c the container restarts.

func (*MetricsCache) AddObject

func (c *MetricsCache) AddObject(key string, t Tags)

func (*MetricsCache) DeleteObject

func (c *MetricsCache) DeleteObject(key string)

func (*MetricsCache) GetTags

func (c *MetricsCache) GetTags(key string) *Tags

func (*MetricsCache) HasObject

func (c *MetricsCache) HasObject(key string) bool

func (*MetricsCache) ReportSync

func (c *MetricsCache) ReportSync()

func (*MetricsCache) ResetCache

func (c *MetricsCache) ResetCache()

type Reporter

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

func NewStatsReporter

func NewStatsReporter() (*Reporter, error)

NewStatsReporter creates a reporter for sync metrics.

func (*Reporter) ReportLastSync

func (r *Reporter) ReportLastSync() error

func (*Reporter) ReportSync

func (r *Reporter) ReportSync(t Tags, v int64) error

func (*Reporter) ReportSyncDuration

func (r *Reporter) ReportSyncDuration(d time.Duration) error

type SingleRunner

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

SingleRunner wraps an errgroup to run keyed goroutines as singletons. Keys are single-use and subsequent usage to schedule will be silently ignored. Goroutines can be individually canceled provided they respect the context passed to them.

func NewSingleRunner

func NewSingleRunner(errChan chan<- error) *SingleRunner

NewSingleRunner returns an initialized SingleRunner.

func (*SingleRunner) Cancel

func (s *SingleRunner) Cancel(key string)

Cancel cancels a keyed goroutine if it exists.

func (*SingleRunner) Go

func (s *SingleRunner) Go(ctx context.Context, key string, f func(context.Context, chan<- error))

Go schedules the provided function on a new goroutine if the provided key has not been used for scheduling before.

func (*SingleRunner) Wait

func (s *SingleRunner) Wait()

Wait waits for all goroutines managed by the SingleRunner to complete. Returns the first error returned from a managed goroutine, or nil.

type SyncBool

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

SyncBool represents a synchronized boolean flag. Its methods are safe to call concurrently.

func (*SyncBool) Get

func (b *SyncBool) Get() bool

Get returns the current value of the flag.

func (*SyncBool) Set

func (b *SyncBool) Set(v bool)

Set sets the values of the flag.

type Tags

type Tags struct {
	Kind   string
	Status metrics.Status
}

type Waiter

type Waiter interface {
	Wait() error
}