db

package
v4.1.0-rc1 Latest Latest
Warning

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

Go to latest
Published: Jul 23, 2026 License: Apache-2.0 Imports: 7 Imported by: 0

Documentation

Index

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 MigrateChanges(config *Config) []sqldb.Change

func SecondsToDurationWithDefault

func SecondsToDurationWithDefault(value *int, defaultSeconds int) time.Duration

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 ControllerHealthRecord struct {
	Controller string    `db:"controller"` // controller where the workflow is running
	Time       time.Time `db:"time"`       // timestamp of creation or last update
}

type Info

type Info struct {
	Config       Config
	SessionProxy *sqldb.SessionProxy
}

func (*Info) Migrate

func (d *Info) Migrate(ctx context.Context)

type LimitRecord

type LimitRecord struct {
	Name      string `db:"name"`
	SizeLimit int    `db:"sizelimit"`
}

LimitRecord is a record type for database synchronization limit operations.

type LockRecord

type LockRecord struct {
	Name       string    `db:"name"`       // semaphore name identifier
	Controller string    `db:"controller"` // controller where the workflow is running
	Time       time.Time `db:"time"`       // timestamp of creation
}

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

Directories

Path Synopsis

Jump to

Keyboard shortcuts

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