mongo

Documentation

Index

• Constants
• Variables
• func Migrate(ctx context.Context, db string, version string, client *mongo.Client, ...) error
• func NewClient(_ context.Context, c config.Reader) (*mongo.Client, error)
• type DataStoreMongo
  • func NewDataStoreWithClient(client *mongo.Client, c config.Reader) *DataStoreMongo
  • func SetupDataStore(automigrate bool) (*DataStoreMongo, error)
  • func (db *DataStoreMongo) Close()
  • func (db *DataStoreMongo) GetAllJobs(ctx context.Context, page int64, perPage int64) ([]model.Job, int64, error)
  • func (db *DataStoreMongo) GetJobByID(ctx context.Context, ID string) (*model.Job, error)
  • func (db *DataStoreMongo) GetJobByNameAndID(ctx context.Context, name string, ID string) (*model.Job, error)
  • func (db *DataStoreMongo) GetWorkflowByName(ctx context.Context, workflowName string, version string) (*model.Workflow, error)
  • func (db *DataStoreMongo) GetWorkflows(ctx context.Context) []model.Workflow
  • func (db *DataStoreMongo) InsertWorkflows(ctx context.Context, workflows ...model.Workflow) (int, error)
  • func (db *DataStoreMongo) LoadWorkflows(ctx context.Context, l *log.Logger) error
  • func (db *DataStoreMongo) Ping(ctx context.Context) error
  • func (db *DataStoreMongo) UpdateJobAddResult(ctx context.Context, job *model.Job, result *model.TaskResult) error
  • func (db *DataStoreMongo) UpdateJobStatus(ctx context.Context, job *model.Job, status int32) error
  • func (db *DataStoreMongo) UpsertJob(ctx context.Context, job *model.Job) (*model.Job, error)

Constants

const (
	// JobsCollectionName refers to the collection of finished or
	// jobs in progress.
	JobsCollectionName = "jobs"

	// WorkflowCollectionName refers to the collection of stored workflows
	WorkflowCollectionName = "workflows"
)

const (
	// DbVersion is the current schema version
	DbVersion = "1.0.0"
)

Variables

var (
	ErrNoSuchWorkflowByVersion = errors.New("Workflow of the given version not found")
)

Functions

func Migrate

func Migrate(ctx context.Context,
	db string,
	version string,
	client *mongo.Client,
	automigrate bool) error

Migrate applies migrations to the database

func NewClient

func NewClient(_ context.Context, c config.Reader) (*mongo.Client, error)

NewClient returns a mongo client

Types

type DataStoreMongo

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

DataStoreMongo is the data storage service

func NewDataStoreWithClient

func NewDataStoreWithClient(client *mongo.Client, c config.Reader) *DataStoreMongo

NewDataStoreWithClient initializes a DataStore object

func SetupDataStore

func SetupDataStore(automigrate bool) (*DataStoreMongo, error)

SetupDataStore returns the mongo data store and optionally runs migrations

func (*DataStoreMongo) Close

func (db *DataStoreMongo) Close()

Close disconnects the client

func (*DataStoreMongo) GetAllJobs

func (db *DataStoreMongo) GetAllJobs(
	ctx context.Context, page int64, perPage int64) ([]model.Job, int64, error)

func (*DataStoreMongo) GetJobByID

func (db *DataStoreMongo) GetJobByID(
	ctx context.Context, ID string) (*model.Job, error)

GetJobByID get the task execution status for a job by ID

func (*DataStoreMongo) GetJobByNameAndID

func (db *DataStoreMongo) GetJobByNameAndID(
	ctx context.Context, name string, ID string) (*model.Job, error)

GetJobByNameAndID get the task execution status for a job by workflow name and ID

func (*DataStoreMongo) GetWorkflowByName

func (db *DataStoreMongo) GetWorkflowByName(
	ctx context.Context,
	workflowName string,
	version string,
) (*model.Workflow, error)

GetWorkflowByName gets the workflow with the given name - either from the cache, or searches the database if the workflow is not cached.

func (*DataStoreMongo) GetWorkflows

func (db *DataStoreMongo) GetWorkflows(ctx context.Context) []model.Workflow

GetWorkflows gets all workflows from the cache as a list (should only be used by the server process)

func (*DataStoreMongo) InsertWorkflows

func (db *DataStoreMongo) InsertWorkflows(
	ctx context.Context,
	workflows ...model.Workflow,
) (int, error)

InsertWorkflows inserts a workflow to the database and cache and returns the number of inserted elements or an error for the first error generated.

func (*DataStoreMongo) LoadWorkflows

func (db *DataStoreMongo) LoadWorkflows(ctx context.Context, l *log.Logger) error

LoadWorkflows from filesystem if the workflowsPath setting is provided

func (*DataStoreMongo) Ping

func (db *DataStoreMongo) Ping(ctx context.Context) error

func (*DataStoreMongo) UpdateJobAddResult

func (db *DataStoreMongo) UpdateJobAddResult(ctx context.Context,
	job *model.Job, result *model.TaskResult) error

UpdateJobAddResult add a task execution result to a job status

func (*DataStoreMongo) UpdateJobStatus

func (db *DataStoreMongo) UpdateJobStatus(
	ctx context.Context, job *model.Job, status int32) error

UpdateJobStatus set the task execution status for a job status

func (*DataStoreMongo) UpsertJob

func (db *DataStoreMongo) UpsertJob(
	ctx context.Context, job *model.Job) (*model.Job, error)

UpsertJob inserts the job in the queue