readdb

Documentation

Index

• Variables
• type ReadDB
  • func NewReadDB(ctx context.Context, logger *zap.Logger, dataDir string, e *etcd.Store, ...) (*ReadDB, error)
  • func (r *ReadDB) Do(ctx context.Context, f func(tx *db.Tx) error) error
  • func (r *ReadDB) GetActiveRuns(tx *db.Tx, groups []string, lastRun bool, phaseFilter []types.RunPhase, ...) ([]*RunData, error)
  • func (r *ReadDB) GetChangeGroupsUpdateTokens(tx *db.Tx, groups []string) (*types.ChangeGroupsUpdateToken, error)
  • func (r *ReadDB) GetChangeGroupsUpdateTokensOST(tx *db.Tx, groups []string) (*datamanager.ChangeGroupsUpdateToken, error)
  • func (r *ReadDB) GetCommittedWalSequenceOST(tx *db.Tx) (string, error)
  • func (r *ReadDB) GetRevision(ctx context.Context) (int64, error)
  • func (r *ReadDB) GetRun(tx *db.Tx, runID string) (*types.Run, error)
  • func (r *ReadDB) GetRunCounterOST(tx *db.Tx, group string) (uint64, error)
  • func (r *ReadDB) GetRunCountersOST(tx *db.Tx, start string, limit int) ([]*types.RunCounter, error)
  • func (r *ReadDB) GetRuns(tx *db.Tx, groups []string, lastRun bool, phaseFilter []types.RunPhase, ...) ([]*types.Run, error)
  • func (r *ReadDB) GetRunsFilteredOST(tx *db.Tx, groups []string, lastRun bool, phaseFilter []types.RunPhase, ...) ([]*RunData, error)
  • func (r *ReadDB) Initialize(ctx context.Context) error
  • func (r *ReadDB) IsInitialized() bool
  • func (r *ReadDB) ResetDB(ctx context.Context) error
  • func (r *ReadDB) Run(ctx context.Context) error
  • func (r *ReadDB) SetInitialized(initialized bool)
  • func (r *ReadDB) SyncFromDump(ctx context.Context) (string, error)
  • func (r *ReadDB) SyncFromWals(ctx context.Context, startWalSeq, endWalSeq string) (string, error)
  • func (r *ReadDB) SyncObjectStorage(ctx context.Context) error
  • func (r *ReadDB) SyncRDB(ctx context.Context) error
• type RunData

Variables

var Stmts = []string{

	"create table revision (revision bigint, PRIMARY KEY(revision))",

	"create table run (id varchar, grouppath varchar, phase varchar, result varchar, PRIMARY KEY (id, grouppath, phase))",

	"create table rundata (id varchar, data bytea, PRIMARY KEY (id))",

	"create table runevent (sequence varchar, data bytea, PRIMARY KEY (sequence))",

	"create table changegrouprevision (id varchar, revision varchar, PRIMARY KEY (id, revision))",

	"create table revision_ost (revision bigint, PRIMARY KEY(revision))",

	"create table committedwalsequence_ost (seq varchar, PRIMARY KEY (seq))",

	"create table changegrouprevision_ost (id varchar, revision varchar, PRIMARY KEY (id, revision))",

	"create table run_ost (id varchar, grouppath varchar, phase varchar, result varchar, PRIMARY KEY (id, grouppath, phase))",

	"create table rundata_ost (id varchar, data bytea, PRIMARY KEY (id))",

	"create table runcounter_ost (groupid varchar, counter bigint, PRIMARY KEY (groupid))",
}

Types

type ReadDB

type ReadDB struct {
	Initialized bool
	// contains filtered or unexported fields
}

func NewReadDB

func NewReadDB(ctx context.Context, logger *zap.Logger, dataDir string, e *etcd.Store, ost *objectstorage.ObjStorage, dm *datamanager.DataManager) (*ReadDB, error)

func (*ReadDB) Do

func (r *ReadDB) Do(ctx context.Context, f func(tx *db.Tx) error) error

func (*ReadDB) GetActiveRuns

func (r *ReadDB) GetActiveRuns(tx *db.Tx, groups []string, lastRun bool, phaseFilter []types.RunPhase, resultFilter []types.RunResult, startRunID string, limit int, sortOrder types.SortOrder) ([]*RunData, error)

func (*ReadDB) GetChangeGroupsUpdateTokens

func (r *ReadDB) GetChangeGroupsUpdateTokens(tx *db.Tx, groups []string) (*types.ChangeGroupsUpdateToken, error)

func (*ReadDB) GetChangeGroupsUpdateTokensOST

func (r *ReadDB) GetChangeGroupsUpdateTokensOST(tx *db.Tx, groups []string) (*datamanager.ChangeGroupsUpdateToken, error)

func (*ReadDB) GetCommittedWalSequenceOST

func (r *ReadDB) GetCommittedWalSequenceOST(tx *db.Tx) (string, error)

func (*ReadDB) GetRevision

func (r *ReadDB) GetRevision(ctx context.Context) (int64, error)

func (*ReadDB) GetRun

func (r *ReadDB) GetRun(tx *db.Tx, runID string) (*types.Run, error)

func (*ReadDB) GetRunCounterOST

func (r *ReadDB) GetRunCounterOST(tx *db.Tx, group string) (uint64, error)

func (*ReadDB) GetRunCountersOST

func (r *ReadDB) GetRunCountersOST(tx *db.Tx, start string, limit int) ([]*types.RunCounter, error)

func (*ReadDB) GetRuns

func (r *ReadDB) GetRuns(tx *db.Tx, groups []string, lastRun bool, phaseFilter []types.RunPhase, resultFilter []types.RunResult, startRunID string, limit int, sortOrder types.SortOrder) ([]*types.Run, error)

func (*ReadDB) GetRunsFilteredOST

func (r *ReadDB) GetRunsFilteredOST(tx *db.Tx, groups []string, lastRun bool, phaseFilter []types.RunPhase, resultFilter []types.RunResult, startRunID string, limit int, sortOrder types.SortOrder) ([]*RunData, error)

func (*ReadDB) Initialize

func (r *ReadDB) Initialize(ctx context.Context) error

Initialize populates the readdb with the current etcd data and save the revision to then feed it with the etcd events

func (*ReadDB) IsInitialized

func (r *ReadDB) IsInitialized() bool

func (*ReadDB) ResetDB

func (r *ReadDB) ResetDB(ctx context.Context) error

func (*ReadDB) Run

func (r *ReadDB) Run(ctx context.Context) error

func (*ReadDB) SetInitialized

func (r *ReadDB) SetInitialized(initialized bool)

func (*ReadDB) SyncFromDump

func (r *ReadDB) SyncFromDump(ctx context.Context) (string, error)

func (*ReadDB) SyncFromWals

func (r *ReadDB) SyncFromWals(ctx context.Context, startWalSeq, endWalSeq string) (string, error)

func (*ReadDB) SyncObjectStorage

func (r *ReadDB) SyncObjectStorage(ctx context.Context) error

func (*ReadDB) SyncRDB

func (r *ReadDB) SyncRDB(ctx context.Context) error

type RunData

type RunData struct {
	ID        string
	GroupPath string
	Phase     string
	Run       *types.Run
}