boltdb

Documentation

Index

• type Store
  • func NewStore(ctx context.Context, dbPath string) (*Store, error)
  • func (s *Store) AddExecutionEvent(ctx context.Context, executionID string, events ...*models.Event) error
  • func (s *Store) BeginTx(ctx context.Context) (boltdblib.TxContext, error)
  • func (s *Store) Checkpoint(ctx context.Context, name string, sequenceNumber uint64) error
  • func (s *Store) Close(ctx context.Context) error
  • func (s *Store) CreateExecution(ctx context.Context, execution models.Execution, events ...*models.Event) error
  • func (s *Store) DeleteExecution(ctx context.Context, executionID string) error
  • func (s *Store) GetCheckpoint(ctx context.Context, name string) (uint64, error)
  • func (s *Store) GetEventStore() watcher.EventStore
  • func (s *Store) GetExecution(ctx context.Context, executionID string) (*models.Execution, error)
  • func (s *Store) GetExecutionCount(ctx context.Context, state models.ExecutionStateType) (uint64, error)
  • func (s *Store) GetExecutionEvents(ctx context.Context, executionID string) ([]*models.Event, error)
  • func (s *Store) GetExecutions(ctx context.Context, jobID string) ([]*models.Execution, error)
  • func (s *Store) GetLiveExecutions(ctx context.Context) ([]*models.Execution, error)
  • func (s *Store) UpdateExecutionState(ctx context.Context, request store.UpdateExecutionRequest) 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> -> {*models.Execution}

* Execution events is stored in a bucket called `execution_events`. Each execution that has events 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 event item, ensuring they are returned in write order.

execution_events

|--> <execution-id>
          |--> <seqnum> -> {models.Event}

* 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 events 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) AddExecutionEvent

func (s *Store) AddExecutionEvent(ctx context.Context, executionID string, events ...*models.Event) error

func (*Store) BeginTx

func (s *Store) BeginTx(ctx context.Context) (boltdblib.TxContext, error)

BeginTx starts a new writable transaction for the store

func (*Store) Checkpoint

func (s *Store) Checkpoint(ctx context.Context, name string, sequenceNumber uint64) error

func (*Store) Close

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

func (*Store) CreateExecution

func (s *Store) CreateExecution(ctx context.Context, execution models.Execution, events ...*models.Event) error

func (*Store) DeleteExecution

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

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

func (*Store) GetCheckpoint

func (s *Store) GetCheckpoint(ctx context.Context, name string) (uint64, error)

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) (*models.Execution, error)

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

func (*Store) GetExecutionCount

func (s *Store) GetExecutionCount(ctx context.Context, state models.ExecutionStateType) (uint64, error)

func (*Store) GetExecutionEvents

func (s *Store) GetExecutionEvents(ctx context.Context, executionID string) ([]*models.Event, error)

GetExecutionEvents retrieves the execution events for a single execution specified by the executionID parameter.

func (*Store) GetExecutions

func (s *Store) GetExecutions(ctx context.Context, jobID string) ([]*models.Execution, 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) ([]*models.Execution, error)

func (*Store) UpdateExecutionState

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