sync

Documentation

Index

• type GetSyncLimit
• type IsWorkflowDeleted
• type LockKind
• type LockName
  • func DecodeLockName(lockName string) (*LockName, error)
  • func GetLockName(sync *v1alpha1.Synchronization, namespace string) (*LockName, error)
  • func NewLockName(namespace, resourceName, lockKey string, kind LockKind) *LockName
  • func (ln *LockName) EncodeName() string
  • func (ln *LockName) Validate() error
  • func (ln *LockName) ValidateEncoding(encoding string) string
• type Manager
  • func NewLockManager(getSyncLimit GetSyncLimit, nextWorkflow NextWorkflow, ...) *Manager
  • func (cm *Manager) CheckWorkflowExistence()
  • func (cm *Manager) Initialize(wfs []wfv1.Workflow)
  • func (cm *Manager) Release(wf *wfv1.Workflow, nodeName string, syncRef *wfv1.Synchronization)
  • func (cm *Manager) ReleaseAll(wf *wfv1.Workflow) bool
  • func (cm *Manager) TryAcquire(wf *wfv1.Workflow, nodeName string, syncLockRef *wfv1.Synchronization) (bool, bool, string, error)
• type NextWorkflow
• type PriorityMutex
  • func NewMutex(name string, nextWorkflow NextWorkflow) *PriorityMutex
• type PrioritySemaphore
  • func NewSemaphore(name string, limit int, nextWorkflow NextWorkflow, lockType string) *PrioritySemaphore
• type Semaphore
• type Throttler
  • func NewThrottler(parallelism int, queue func(key string)) Throttler

Types

type GetSyncLimit

type GetSyncLimit func(string) (int, error)

type IsWorkflowDeleted

type IsWorkflowDeleted func(string) bool

type LockKind

type LockKind string

const (
	LockKindConfigMap LockKind = "ConfigMap"
	LockKindMutex     LockKind = "Mutex"
)

type LockName

type LockName struct {
	Namespace    string
	ResourceName string
	Key          string
	Kind         LockKind
}

func DecodeLockName

func DecodeLockName(lockName string) (*LockName, error)

func GetLockName

func GetLockName(sync *v1alpha1.Synchronization, namespace string) (*LockName, error)

func NewLockName

func NewLockName(namespace, resourceName, lockKey string, kind LockKind) *LockName

func (*LockName) EncodeName

func (ln *LockName) EncodeName() string

func (*LockName) Validate

func (ln *LockName) Validate() error

func (*LockName) ValidateEncoding

func (ln *LockName) ValidateEncoding(encoding string) string

type Manager

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

func NewLockManager

func NewLockManager(getSyncLimit GetSyncLimit, nextWorkflow NextWorkflow, isWFDeleted IsWorkflowDeleted) *Manager

func (*Manager) CheckWorkflowExistence

func (cm *Manager) CheckWorkflowExistence()

func (*Manager) Initialize

func (cm *Manager) Initialize(wfs []wfv1.Workflow)

func (*Manager) Release

func (cm *Manager) Release(wf *wfv1.Workflow, nodeName string, syncRef *wfv1.Synchronization)

func (*Manager) ReleaseAll

func (cm *Manager) ReleaseAll(wf *wfv1.Workflow) bool

func (*Manager) TryAcquire

func (cm *Manager) TryAcquire(wf *wfv1.Workflow, nodeName string, syncLockRef *wfv1.Synchronization) (bool, bool, string, error)

TryAcquire tries to acquire the lock from semaphore. It returns status of acquiring a lock , status of Workflow status updated, waiting message if lock is not available and any error encountered

type NextWorkflow

type NextWorkflow func(string)

type PriorityMutex

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

func NewMutex

func NewMutex(name string, nextWorkflow NextWorkflow) *PriorityMutex

NewMutex creates new mutex lock object name of the mutex callbackFunc is a release notification function.

type PrioritySemaphore

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

func NewSemaphore

func NewSemaphore(name string, limit int, nextWorkflow NextWorkflow, lockType string) *PrioritySemaphore

type Semaphore

type Semaphore interface {
	// contains filtered or unexported methods
}

type Throttler

type Throttler interface {
	Add(key string, priority int32, creationTime time.Time)
	// Admin returns if the item should be processed.
	Admit(key string) bool
	// Remove notifies throttler that item processing is no longer needed
	Remove(key string)
}

Throttler allows the controller to limit number of items it is processing in parallel. Items are processed in priority order, and one processing starts, other items (including higher-priority items) will be kept pending until the processing is complete. Implementations should be idempotent.

func NewThrottler

func NewThrottler(parallelism int, queue func(key string)) Throttler

NewThrottler returns a throttle that only runs `parallelism` items at once. When an item may need processing, `queue` is invoked.