boltdb

Documentation

Index

• type Store
  • func NewStore(ctx context.Context, dbPath string) (*Store, error)
  • func (s *Store) Close(ctx context.Context) error
  • func (s *Store) CreateExecution(ctx context.Context, execution store.LocalExecutionState) error
  • func (s *Store) DeleteExecution(ctx context.Context, executionID string) error
  • func (s *Store) GetEventStore() watcher.EventStore
  • func (s *Store) GetExecution(ctx context.Context, executionID string) (store.LocalExecutionState, error)
  • func (s *Store) GetExecutionCount(ctx context.Context, state store.LocalExecutionStateType) (uint64, error)
  • func (s *Store) GetExecutionHistory(ctx context.Context, executionID string) ([]store.LocalStateHistory, error)
  • func (s *Store) GetExecutions(ctx context.Context, jobID string) ([]store.LocalExecutionState, error)
  • func (s *Store) GetLiveExecutions(ctx context.Context) ([]store.LocalExecutionState, error)
  • func (s *Store) UpdateExecutionState(ctx context.Context, request store.UpdateExecutionStateRequest) error

Types

type Store

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

Store represents an execution store that is backed by a boltdb database on disk. The structure of the store is organized into boltdb buckets and sub-buckets for efficient data retrieval and organization.

The schema (<key> -> {json-value}) looks like the following, where <> represents keys, {} represents values, and undecorated values are boltdb buckets:

* Executions are stored in a bucket called `execution` where each key is an execution ID and the value is the JSON representation of the execution.

executions

|--> <execution-id> -> {store.LocalExecutionState}

* Execution history is stored in a bucket called `execution-history`. Each execution that has history is stored in a sub-bucket, whose name is the execution ID. Within the execution ID bucket, each key is a sequential number for the history item, ensuring they are returned in write order.

execution_history

|--> <execution-id>
          |--> <seqnum> -> {store.LocalStateHistory}

* The job index is stored in a bucket called `idx:executions-by-jobid` where each job is represented by a sub-bucket, named after the job ID. Within that job bucket, each execution is represented by a key which is the ID of that execution, with a nil value.

idx_executions_by_job_id

|--> <job-id>
          |--> <execution-id> -> nil

* The state index is stored in a bucket called `idx:executions-by-state` where each execution state is represented by a sub-bucket, named after the state. Within each state bucket, executions are indexed by their ID with a nil value.

idx_executions_by_state

|--> <state>
          |--> <execution-id> -> nil

* Additional buckets for event storage:

• `events`: Stores event data
• `checkpoints`: Stores checkpoint information for the event system

This structure allows for efficient querying of executions by ID, job, and state, as well as maintaining a complete history of execution state changes.

func NewStore

func NewStore(ctx context.Context, dbPath string) (*Store, error)

NewStore creates a new store backed by a boltdb database at the file location provided by the caller. During initialisation the primary buckets are created, but they are not stored in the struct as they are tied to the transaction where they are referenced and it would mean later transactions will fail unless they obtain their own reference to the bucket.

func (*Store) Close

func (s *Store) Close(ctx context.Context) error

func (*Store) CreateExecution

func (s *Store) CreateExecution(ctx context.Context, execution store.LocalExecutionState) error

func (*Store) DeleteExecution

func (s *Store) DeleteExecution(ctx context.Context, executionID string) error

DeleteExecution delete the execution, removes its history and removes it from the job index (along with the job indexes bucket)

func (*Store) GetEventStore

func (s *Store) GetEventStore() watcher.EventStore

GetEventStore returns the event store for the execution store

func (*Store) GetExecution

func (s *Store) GetExecution(ctx context.Context, executionID string) (store.LocalExecutionState, error)

GetExecution returns the stored LocalExecutionState structure for the provided execution ID.

func (*Store) GetExecutionCount

func (s *Store) GetExecutionCount(ctx context.Context, state store.LocalExecutionStateType) (uint64, error)

func (*Store) GetExecutionHistory

func (s *Store) GetExecutionHistory(ctx context.Context, executionID string) ([]store.LocalStateHistory, error)

GetExecutionHistory retrieves the execution history for a single execution specified by the executionID parameter.

func (*Store) GetExecutions

func (s *Store) GetExecutions(ctx context.Context, jobID string) ([]store.LocalExecutionState, error)

GetExecutions retrieves akk if the executions from the job-index bucket for the provided Job ID.

func (*Store) GetLiveExecutions

func (s *Store) GetLiveExecutions(ctx context.Context) ([]store.LocalExecutionState, error)

func (*Store) UpdateExecutionState

func (s *Store) UpdateExecutionState(ctx context.Context, request store.UpdateExecutionStateRequest) error