Documentation
¶
Index ¶
- Constants
- func MigrateChanges(config *Config) []sqldb.Change
- func SecondsToDurationWithDefault(value *int, defaultSeconds int) time.Duration
- func SessionProxyFromConfig(ctx context.Context, kubectlConfig kubernetes.Interface, namespace string, ...) *sqldb.SessionProxy
- type Config
- type ControllerHealthRecord
- type Info
- type LimitRecord
- type LockRecord
- type StateCountRecord
- type StateRecord
- type SyncQueries
Constants ¶
View Source
const ( DefaultDBPollSeconds = 10 DefaultDBHeartbeatSeconds = 60 DefaultDBInactiveControllerSeconds = 600 )
View Source
const ( LimitNameField = "name" LimitSizeField = "sizelimit" StateNameField = "name" StateKeyField = "workflowkey" StateControllerField = "controller" StateHeldField = "held" StatePriorityField = "priority" StateTimeField = "time" ControllerNameField = "controller" ControllerTimeField = "time" LockNameField = "name" LockControllerField = "controller" )
Field name constants
Variables ¶
This section is empty.
Functions ¶
func MigrateChanges ¶
func SessionProxyFromConfig ¶
func SessionProxyFromConfig(ctx context.Context, kubectlConfig kubernetes.Interface, namespace string, config *config.SyncConfig) *sqldb.SessionProxy
Types ¶
type Config ¶
type Config struct {
LimitTable string
StateTable string
ControllerTable string
LockTable string
ControllerName string
InactiveControllerTimeout time.Duration
SkipMigration bool
}
Config holds database configuration for sync operations.
func ConfigFromConfig ¶
func ConfigFromConfig(config *config.SyncConfig) Config
type ControllerHealthRecord ¶
type Info ¶
type Info struct {
Config Config
SessionProxy *sqldb.SessionProxy
}
type LimitRecord ¶
LimitRecord is a record type for database synchronization limit operations.
type LockRecord ¶
type StateCountRecord ¶
type StateCountRecord struct {
Name string `db:"name"` // semaphore name identifier (as stored, e.g. "sem/<ns>/<resource>")
Held bool `db:"held"` // true = held, false = pending
Count int64 `db:"lock_count"` // number of rows in this (name, held) group
}
StateCountRecord is an aggregate count of state rows for a lock, grouped by held/pending.
type StateRecord ¶
type StateRecord struct {
Name string `db:"name"` // semaphore name identifier
Key string `db:"workflowkey"` // workflow key holding or waiting for the lock of the form <namespace>/<name>
Controller string `db:"controller"` // controller where the workflow is running
Held bool `db:"held"`
Priority int32 `db:"priority"` // higher number = higher priority in queue
Time time.Time `db:"time"` // timestamp of creation or last update
}
type SyncQueries ¶
type SyncQueries interface {
CreateSemaphoreLimit(ctx context.Context, name string, sizeLimit int) error
UpdateSemaphoreLimit(ctx context.Context, name string, sizeLimit int) error
DeleteSemaphoreLimit(ctx context.Context, name string) error
GetSemaphoreLimit(ctx context.Context, dbKey string) (*LimitRecord, error)
GetCurrentState(ctx context.Context, sessionProxy *sqldb.SessionProxy, semaphoreName string, held bool) ([]StateRecord, error)
GetCurrentHolders(ctx context.Context, sessionProxy *sqldb.SessionProxy, semaphoreName string) ([]StateRecord, error)
GetCurrentPending(ctx context.Context, semaphoreName string) ([]StateRecord, error)
GetStateCountsByController(ctx context.Context, controllerName string) ([]StateCountRecord, error)
GetOrderedQueue(ctx context.Context, sessionProxy *sqldb.SessionProxy, semaphoreName string, inactiveTimeout time.Duration) ([]StateRecord, error)
AddToQueue(ctx context.Context, record *StateRecord) error
RemoveFromQueue(ctx context.Context, semaphoreName, holderKey string) error
CheckQueueExists(ctx context.Context, semaphoreName, holderKey, controllerName string) ([]StateRecord, error)
UpdateStateToHeld(ctx context.Context, sessionProxy *sqldb.SessionProxy, semaphoreName, holderKey, controllerName string) error
InsertHeldState(ctx context.Context, sessionProxy *sqldb.SessionProxy, record *StateRecord) error
GetPendingInQueue(ctx context.Context, sessionProxy *sqldb.SessionProxy, semaphoreName, holderKey, controllerName string) ([]StateRecord, error)
ReleaseHeld(ctx context.Context, semaphoreName, key, controllerName string) error
GetExistingLocks(ctx context.Context, lockName, controllerName string) ([]LockRecord, error)
InsertLock(ctx context.Context, record *LockRecord) error
DeleteLock(ctx context.Context, lockName string) error
ExpireInactiveLocks(ctx context.Context, inactiveTimeout time.Duration) (int64, error)
InsertControllerHealth(ctx context.Context, record *ControllerHealthRecord) error
UpdateControllerTimestamp(ctx context.Context, controllerName string, timestamp time.Time) error
}
func NewSyncQueries ¶
func NewSyncQueries(sessionProxy *sqldb.SessionProxy, config Config) SyncQueries
NewSyncQueries creates a new syncQueries instance
Click to show internal directories.
Click to hide internal directories.