store

Documentation

Index

• Constants
• type DBStore
  • func NewDBStore(ds sqlutil.DataSource, clock clockwork.Clock) *DBStore
  • func (d *DBStore) Add(ctx context.Context, state *WorkflowExecution) error
  • func (d *DBStore) Get(ctx context.Context, executionID string) (WorkflowExecution, error)
  • func (d *DBStore) GetUnfinished(ctx context.Context, offset, limit int) ([]WorkflowExecution, error)
  • func (d *DBStore) UpdateStatus(ctx context.Context, executionID string, status string) error
  • func (d *DBStore) UpsertStep(ctx context.Context, stepState *WorkflowExecutionStep) (WorkflowExecution, error)
• type InMemoryStore
  • func NewInMemoryStore() *InMemoryStore
  • func (s *InMemoryStore) Add(ctx context.Context, state *WorkflowExecution) error
  • func (s *InMemoryStore) Get(ctx context.Context, executionID string) (WorkflowExecution, error)
  • func (s *InMemoryStore) GetUnfinished(ctx context.Context, offset, limit int) ([]WorkflowExecution, error)
  • func (s *InMemoryStore) UpdateStatus(ctx context.Context, executionID string, status string) error
  • func (s *InMemoryStore) UpsertStep(ctx context.Context, step *WorkflowExecutionStep) (WorkflowExecution, error)
• type StepOutput
• type Store
• type WorkflowExecution
• type WorkflowExecutionStep

Constants

const (
	StatusStarted            = "started"
	StatusErrored            = "errored"
	StatusTimeout            = "timeout"
	StatusCompleted          = "completed"
	StatusCompletedEarlyExit = "completed_early_exit"
)

Types

type DBStore

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

`DBStore` is a postgres-backed data store that persists workflow progress.

func NewDBStore

func NewDBStore(ds sqlutil.DataSource, clock clockwork.Clock) *DBStore

func (*DBStore) Add

func (d *DBStore) Add(ctx context.Context, state *WorkflowExecution) error

`Add` creates the relevant workflow_execution and workflow_step entries to persist the passed in ExecutionState.

func (*DBStore) Get

func (d *DBStore) Get(ctx context.Context, executionID string) (WorkflowExecution, error)

`Get` fetches the ExecutionState from the database.

func (*DBStore) GetUnfinished

func (d *DBStore) GetUnfinished(ctx context.Context, offset, limit int) ([]WorkflowExecution, error)

func (*DBStore) UpdateStatus

func (d *DBStore) UpdateStatus(ctx context.Context, executionID string, status string) error

`UpdateStatus` updates the status of the given workflow execution

func (*DBStore) UpsertStep

func (d *DBStore) UpsertStep(ctx context.Context, stepState *WorkflowExecutionStep) (WorkflowExecution, error)

`UpsertStep` updates the given step. This will correspond to an insert, or an update depending on whether a step with the ref already exists.

type InMemoryStore

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

`InMemoryStore` is a temporary in-memory equivalent of the database table that should persist workflow progress.

func NewInMemoryStore

func NewInMemoryStore() *InMemoryStore

func (*InMemoryStore) Add

func (s *InMemoryStore) Add(ctx context.Context, state *WorkflowExecution) error

Add adds a new execution state under the given executionID

func (*InMemoryStore) Get

func (s *InMemoryStore) Get(ctx context.Context, executionID string) (WorkflowExecution, error)

Get gets the state for the given executionID

func (*InMemoryStore) GetUnfinished

func (s *InMemoryStore) GetUnfinished(ctx context.Context, offset, limit int) ([]WorkflowExecution, error)

GetUnfinished gets the states for execution that are in a started state Offset and limit are ignored for the in-memory store.

func (*InMemoryStore) UpdateStatus

func (s *InMemoryStore) UpdateStatus(ctx context.Context, executionID string, status string) error

UpdateStatus updates the status for the given executionID

func (*InMemoryStore) UpsertStep

func (s *InMemoryStore) UpsertStep(ctx context.Context, step *WorkflowExecutionStep) (WorkflowExecution, error)

UpsertStep updates a step for the given executionID

type StepOutput

type StepOutput struct {
	Err   error
	Value values.Value
}

type Store

type Store interface {
	Add(ctx context.Context, state *WorkflowExecution) error
	UpsertStep(ctx context.Context, step *WorkflowExecutionStep) (WorkflowExecution, error)
	UpdateStatus(ctx context.Context, executionID string, status string) error
	Get(ctx context.Context, executionID string) (WorkflowExecution, error)
	GetUnfinished(ctx context.Context, offset, limit int) ([]WorkflowExecution, error)
}

type WorkflowExecution

type WorkflowExecution struct {
	Steps       map[string]*WorkflowExecutionStep
	ExecutionID string
	WorkflowID  string

	Status     string
	CreatedAt  *time.Time
	UpdatedAt  *time.Time
	FinishedAt *time.Time
}

type WorkflowExecutionStep

type WorkflowExecutionStep struct {
	ExecutionID string
	Ref         string
	Status      string

	Inputs  *values.Map
	Outputs StepOutput

	UpdatedAt *time.Time
}