sync

package
v3.7.6 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Dec 9, 2025 License: Apache-2.0 Imports: 22 Imported by: 2

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func DecodeLockName

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

Types

type GetSyncLimit

type GetSyncLimit func(string) (int, error)

type IsWorkflowDeleted

type IsWorkflowDeleted func(string) bool

type Key added in v3.1.0

type Key = string

type Manager

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

func NewLockManager

func NewLockManager(ctx context.Context, kubectlConfig kubernetes.Interface, namespace string, config *config.SyncConfig, getSyncLimit GetSyncLimit, nextWorkflow NextWorkflow, isWFDeleted IsWorkflowDeleted) *Manager

func (*Manager) CheckWorkflowExistence

func (sm *Manager) CheckWorkflowExistence(ctx context.Context)

func (*Manager) Initialize

func (sm *Manager) Initialize(ctx context.Context, wfs []wfv1.Workflow)

func (*Manager) Release

func (sm *Manager) Release(ctx context.Context, wf *wfv1.Workflow, nodeName string, syncRef *wfv1.Synchronization)

func (*Manager) ReleaseAll

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

func (*Manager) TryAcquire

func (sm *Manager) TryAcquire(ctx context.Context, wf *wfv1.Workflow, nodeName string, syncLockRef *wfv1.Synchronization) (bool, bool, string, 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, the failed lock, and any error encountered

type NextWorkflow

type NextWorkflow func(string)

type QueueFunc added in v3.1.0

type QueueFunc func(Key)

type SyncLevelType added in v3.5.11

type SyncLevelType int
const (
	WorkflowLevel SyncLevelType = 1
	TemplateLevel SyncLevelType = 2
	ErrorLevel    SyncLevelType = 3
)

type Throttler

type Throttler interface {
	Init(wfs []wfv1.Workflow) error
	Add(key Key, priority int32, creationTime time.Time)
	// Admit returns if the item should be processed.
	Admit(key Key) bool
	// Remove notifies throttler that item processing is no longer needed
	Remove(key Key)
	// UpdateParallelism
	UpdateParallelism(limit int)
	// UpdateNamespaceParallelism updates the namespace parallelism
	UpdateNamespaceParallelism(namespace string, limit int)
	// ResetNamespaceParallelism sets the namespace parallelism to the default value
	ResetNamespaceParallelism(namespace 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 NewMultiThrottler added in v3.5.14

func NewMultiThrottler(parallelism int, namespaceParallelismLimit int, queue QueueFunc) Throttler

NewMultiThrottler creates a new multi throttler for throttling both namespace and global parallelism, a parallelism value of zero disables throttling

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL