lookoutdb

Documentation

Index

• type DB
  • func New(db *pgxpool.Pool, m *metrics.Metrics) *DB
  • func (l *DB) CreateJobRunContainers(ctx context.Context, instructions []*model.CreateJobRunContainerInstruction)
  • func (l *DB) CreateJobRunContainersBatch(ctx context.Context, instructions []*model.CreateJobRunContainerInstruction) error
  • func (l *DB) CreateJobRunContainersScalar(ctx context.Context, instructions []*model.CreateJobRunContainerInstruction)
  • func (l *DB) CreateJobRuns(ctx context.Context, instructions []*model.CreateJobRunInstruction)
  • func (l *DB) CreateJobRunsBatch(ctx context.Context, instructions []*model.CreateJobRunInstruction) error
  • func (l *DB) CreateJobRunsScalar(ctx context.Context, instructions []*model.CreateJobRunInstruction)
  • func (l *DB) CreateJobs(ctx context.Context, instructions []*model.CreateJobInstruction)
  • func (l *DB) CreateJobsBatch(ctx context.Context, instructions []*model.CreateJobInstruction) error
  • func (l *DB) CreateJobsScalar(ctx context.Context, instructions []*model.CreateJobInstruction)
  • func (l *DB) CreateUserAnnotations(ctx context.Context, instructions []*model.CreateUserAnnotationInstruction)
  • func (l *DB) CreateUserAnnotationsBatch(ctx context.Context, instructions []*model.CreateUserAnnotationInstruction) error
  • func (l *DB) CreateUserAnnotationsScalar(ctx context.Context, instructions []*model.CreateUserAnnotationInstruction)
  • func (l *DB) ProcessUpdates(ctx context.Context, db *pgxpool.Pool, msgs chan *model.InstructionSet, ...) chan []*pulsarutils.ConsumerMessageId
  • func (l *DB) Update(ctx context.Context, instructions *model.InstructionSet)
  • func (l *DB) UpdateJobRuns(ctx context.Context, instructions []*model.UpdateJobRunInstruction)
  • func (l *DB) UpdateJobRunsBatch(ctx context.Context, instructions []*model.UpdateJobRunInstruction) error
  • func (l *DB) UpdateJobRunsScalar(ctx context.Context, instructions []*model.UpdateJobRunInstruction)
  • func (l *DB) UpdateJobs(ctx context.Context, instructions []*model.UpdateJobInstruction)
  • func (l *DB) UpdateJobsBatch(ctx context.Context, instructions []*model.UpdateJobInstruction) error
  • func (l *DB) UpdateJobsScalar(ctx context.Context, instructions []*model.UpdateJobInstruction)

Types

type DB

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

func New

func New(db *pgxpool.Pool, m *metrics.Metrics) *DB

func (*DB) CreateJobRunContainers

func (l *DB) CreateJobRunContainers(ctx context.Context, instructions []*model.CreateJobRunContainerInstruction)

func (*DB) CreateJobRunContainersBatch

func (l *DB) CreateJobRunContainersBatch(ctx context.Context, instructions []*model.CreateJobRunContainerInstruction) error

func (*DB) CreateJobRunContainersScalar

func (l *DB) CreateJobRunContainersScalar(ctx context.Context, instructions []*model.CreateJobRunContainerInstruction)

func (*DB) CreateJobRuns

func (l *DB) CreateJobRuns(ctx context.Context, instructions []*model.CreateJobRunInstruction)

func (*DB) CreateJobRunsBatch

func (l *DB) CreateJobRunsBatch(ctx context.Context, instructions []*model.CreateJobRunInstruction) error

func (*DB) CreateJobRunsScalar

func (l *DB) CreateJobRunsScalar(ctx context.Context, instructions []*model.CreateJobRunInstruction)

func (*DB) CreateJobs

func (l *DB) CreateJobs(ctx context.Context, instructions []*model.CreateJobInstruction)

func (*DB) CreateJobsBatch

func (l *DB) CreateJobsBatch(ctx context.Context, instructions []*model.CreateJobInstruction) error

func (*DB) CreateJobsScalar

func (l *DB) CreateJobsScalar(ctx context.Context, instructions []*model.CreateJobInstruction)

CreateJobsScalar will insert jobs one by one into the database

func (*DB) CreateUserAnnotations

func (l *DB) CreateUserAnnotations(ctx context.Context, instructions []*model.CreateUserAnnotationInstruction)

func (*DB) CreateUserAnnotationsBatch

func (l *DB) CreateUserAnnotationsBatch(ctx context.Context, instructions []*model.CreateUserAnnotationInstruction) error

func (*DB) CreateUserAnnotationsScalar

func (l *DB) CreateUserAnnotationsScalar(ctx context.Context, instructions []*model.CreateUserAnnotationInstruction)

func (*DB) ProcessUpdates

func (l *DB) ProcessUpdates(ctx context.Context, db *pgxpool.Pool, msgs chan *model.InstructionSet, bufferSize int) chan []*pulsarutils.ConsumerMessageId

ProcessUpdates will update the lookout database according to the incoming channel of instructions. It returns a channel containing all the message ids that have been successfully processed.

func (*DB) Update

func (l *DB) Update(ctx context.Context, instructions *model.InstructionSet)

Update updates the lookout database according to the supplied InstructionSet. The updates are applied in the following order: * New Job Creations * Job Updates, New Job Creations, New User Annotations * Job Run Updates, New Job Containers In each case we first try to bach insert the rows using the postgres copy protocol. If this fails then we try a slower, serial insert and discard any rows that cannot be inserted.

func (*DB) UpdateJobRuns

func (l *DB) UpdateJobRuns(ctx context.Context, instructions []*model.UpdateJobRunInstruction)

func (*DB) UpdateJobRunsBatch

func (l *DB) UpdateJobRunsBatch(ctx context.Context, instructions []*model.UpdateJobRunInstruction) error

func (*DB) UpdateJobRunsScalar

func (l *DB) UpdateJobRunsScalar(ctx context.Context, instructions []*model.UpdateJobRunInstruction)

func (*DB) UpdateJobs

func (l *DB) UpdateJobs(ctx context.Context, instructions []*model.UpdateJobInstruction)

func (*DB) UpdateJobsBatch

func (l *DB) UpdateJobsBatch(ctx context.Context, instructions []*model.UpdateJobInstruction) error

func (*DB) UpdateJobsScalar

func (l *DB) UpdateJobsScalar(ctx context.Context, instructions []*model.UpdateJobInstruction)