operations

Documentation

Index

• type Job
  • func (j *Job) Execute(ctxWithTimeout context.Context, repository storage.Repository) (operationID string, err error)
• type Maintainer
  • func NewMaintainer(smCtx context.Context, repository storage.Repository, options *Settings) *Maintainer
  • func (om *Maintainer) Run()
• type OperationError
• type PoolSettings
  • func (ps *PoolSettings) Validate() error
• type Scheduler
  • func NewScheduler(smCtx context.Context, repository storage.Repository, jobTimeout time.Duration, ...) *Scheduler
  • func (ds *Scheduler) Schedule(job Job) (string, error)
• type Settings
  • func DefaultSettings() *Settings
  • func (s *Settings) Validate() error

Types

type Job

type Job struct {
	ReqCtx     context.Context
	ObjectType types.ObjectType

	Operation     *types.Operation
	OperationFunc func(ctx context.Context, repository storage.Repository) (types.Object, error)
}

Job is responsible for executing a C/U/D DB operation

func (*Job) Execute

func (j *Job) Execute(ctxWithTimeout context.Context, repository storage.Repository) (operationID string, err error)

Execute executes a C/U/D DB operation

type Maintainer

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

Maintainer ensures that operations old enough are deleted and that no orphan operations are left in the DB due to crashes/restarts of SM

func NewMaintainer

func NewMaintainer(smCtx context.Context, repository storage.Repository, options *Settings) *Maintainer

NewMaintainer constructs a Maintainer

func (*Maintainer) Run

func (om *Maintainer) Run()

Run starts the two recurring jobs responsible for cleaning up operations which are too old and deleting orphan operations

type OperationError

type OperationError struct {
	Message string `json:"message"`
}

OperationError holds an error message returned from an execution of an async job

type PoolSettings

type PoolSettings struct {
	Resource string `mapstructure:"resource" description:"name of the resource for which a worker pool is created"`
	Size     int    `mapstructure:"size" description:"size of the worker pool"`
}

PoolSettings defines the settings for a worker pool

func (*PoolSettings) Validate

func (ps *PoolSettings) Validate() error

Validate validates the Pool settings

type Scheduler

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

Scheduler is responsible for storing Operation entities in the DB and also for spawning goroutines to execute the respective DB transaction asynchronously

func NewScheduler

func NewScheduler(smCtx context.Context, repository storage.Repository, jobTimeout time.Duration, workerPoolSize int, wg *sync.WaitGroup) *Scheduler

NewScheduler constructs a Scheduler

func (*Scheduler) Schedule

func (ds *Scheduler) Schedule(job Job) (string, error)

Schedule stores the Job's Operation entity in DB and spawns a goroutine to execute the CREATE/UPDATE/DELETE DB transaction asynchronously

type Settings

type Settings struct {
	JobTimeout          time.Duration  `mapstructure:"job_timeout" description:"timeout for async operations"`
	MarkOrphansInterval time.Duration  `mapstructure:"mark_orphans_interval" description:"interval denoting how often to mark orphan operations as failed"`
	CleanupInterval     time.Duration  `mapstructure:"cleanup_interval" description:"cleanup interval of old operations"`
	ExpirationTime      time.Duration  `` /* 140-byte string literal not displayed */
	DefaultPoolSize     int            `mapstructure:"default_pool_size" description:"default worker pool size"`
	Pools               []PoolSettings `mapstructure:"pools" description:"defines the different available worker pools"`
}

Settings type to be loaded from the environment

func DefaultSettings

func DefaultSettings() *Settings

DefaultSettings returns default values for API settings

func (*Settings) Validate

func (s *Settings) Validate() error

Validate validates the Operations settings