Documentation
¶
Index ¶
- type GetSyncLimit
- type Key
- type LevelType
- type LockName
- type Manager
- func (sm *Manager) CheckWorkflowExistence(ctx context.Context)
- func (sm *Manager) Initialize(ctx context.Context, wfs []wfv1.Workflow) ([]StaleHold, error)
- func (sm *Manager) LockMetrics(ctx context.Context) []wfmetrics.LockGaugeSample
- func (sm *Manager) Release(ctx context.Context, wf *wfv1.Workflow, nodeName string, ...)
- func (sm *Manager) ReleaseAll(ctx context.Context, wf *wfv1.Workflow) bool
- func (sm *Manager) TryAcquire(ctx context.Context, wf *wfv1.Workflow, nodeName string, ...) (bool, bool, string, string, error)
- func (sm *Manager) WithMetrics(ctx context.Context, m *wfmetrics.Metrics) *Manager
- type NextWorkflow
- type QueueFunc
- type StaleHold
- type Throttler
- type WorkflowExists
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type LockName ¶
type LockName interface {
GetNamespace() string
GetResourceName() string
GetKey() string
String(ctx context.Context) string
// contains filtered or unexported methods
}
LockName represents a decoded lock name with its components.
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, workflowExists WorkflowExists, ensureDBConnection bool) (*Manager, error)
NewLockManager creates a new lock manager
func (*Manager) CheckWorkflowExistence ¶
func (*Manager) Initialize ¶
Initialize re-establishes, in the in-memory lock map, the holds that Running workflows record in their status.
It fails closed only when a holder is genuinely unrecoverable (see initFailureFatal): an undecodable lock name, or a database-backed hold with no database session. Those return an error the controller treats as fatal, because we can neither poison the lock nor prove the spec re-acquires it, so continuing risks a silent double-acquire.
Recoverable failures never crashloop: a lock that cannot be built (transient ConfigMap/DB read) or whose holder key is unresolvable is poisoned (lock-scoped, clears on restart). A database-backed hold that the database no longer records is returned as a StaleHold (at most one per workflow) for the controller to fail the workflow.
func (*Manager) LockMetrics ¶
func (sm *Manager) LockMetrics(ctx context.Context) []wfmetrics.LockGaugeSample
LockMetrics returns a point-in-time snapshot of the locks this controller currently participates in, for the observable locks_held / locks_pending gauges. It is called at metric scrape time.
In-memory locks (ConfigMap semaphores and Mutexes) are read directly from memory. Database-backed locks are read with a single controller-scoped aggregate query rather than one query per lock, so each controller reports only its own contribution; `sum by (lock_name)` across controllers yields the true global picture without double-counting.
func (*Manager) ReleaseAll ¶
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 StaleHold ¶ added in v4.0.6
StaleHold records a workflow whose recorded hold on a database-backed lock could not be verified against the database during Initialize. The database is the single source of truth for such locks, so the workflow is running on a hold the database no longer backs (e.g. it was expired while the controller was down and may since have been acquired by another holder). The controller fails these workflows; their teardown releases any locks they still hold.
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)
// UpdateNamespaceParallelismDefault updates the controller-config default limit for namespaces
// without an explicit Namespace label override.
UpdateNamespaceParallelismDefault(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.
Source Files
Directories