collection

Documentation

Index

• Constants
• func InitProcessingWorkflow(ctx context.Context, c client.Client, event *watcher.BlobEvent) error
• func NewService(logger logr.Logger, db *sql.DB, cc cadenceclient.Client, ...) *collectionImpl
• func TriggerProcessingWorkflow(ctx context.Context, c client.Client, req *ProcessingWorkflowRequest) error
• type Collection
  • func (c Collection) Goa() *goacollection.EnduroStoredCollection
• type ProcessingWorkflowRequest
• type Service
• type Status
  • func NewStatus(status string) Status
  • func (p Status) MarshalJSON() ([]byte, error)
  • func (p Status) String() string
  • func (p *Status) UnmarshalJSON(b []byte) error

Constants

const (
	// Name of the collection processing workflow.
	ProcessingWorkflowName = "processing-workflow"

	// Maximum duration of the processing workflow. Cadence does not support
	// workflows with infinite duration for now, but high values are fine.
	// We consider a week more than enough.
	ProcessingWorkflowStartToCloseTimeout = time.Hour * 24 * 7
)

Functions

func InitProcessingWorkflow

func InitProcessingWorkflow(ctx context.Context, c client.Client, event *watcher.BlobEvent) error

func NewService

func NewService(logger logr.Logger, db *sql.DB, cc cadenceclient.Client, registry *pipeline.Registry) *collectionImpl

func TriggerProcessingWorkflow

func TriggerProcessingWorkflow(ctx context.Context, c client.Client, req *ProcessingWorkflowRequest) error

Types

type Collection

type Collection struct {
	ID            uint   `db:"id"`
	Name          string `db:"name"`
	WorkflowID    string `db:"workflow_id"`
	RunID         string `db:"run_id"`
	TransferID    string `db:"transfer_id"`
	AIPID         string `db:"aip_id"`
	OriginalID    string `db:"original_id"`
	PipelineID    string `db:"pipeline_id"`
	DecisionToken string `db:"decision_token"`
	Status        Status `db:"status"`

	// It defaults to CURRENT_TIMESTAMP(6) so populated as soon as possible.
	CreatedAt time.Time `db:"created_at"`

	// Nullable, populated as soon as processing starts.
	StartedAt sql.NullTime `db:"started_at"`

	// Nullable, populated as soon as ingest completes.
	CompletedAt sql.NullTime `db:"completed_at"`
}

Collection represents a collection in the collection table.

func (Collection) Goa

func (c Collection) Goa() *goacollection.EnduroStoredCollection

Goa returns the API representation of the collection.

type ProcessingWorkflowRequest

type ProcessingWorkflowRequest struct {
	WorkflowID string `json:"-"`

	// The zero value represents a new collection. It can be used to indicate
	// an existing collection in retries.
	CollectionID uint

	// Captured by the watcher, the event contains information about the
	// incoming dataset.
	Event *watcher.BlobEvent
}

type Service

type Service interface {
	// Goa returns an implementation of the goacollection Service.
	Goa() goacollection.Service

	Create(context.Context, *Collection) error
	UpdateWorkflowStatus(ctx context.Context, ID uint, name string, workflowID, runID, transferID, aipID, pipelineID string, status Status, storedAt time.Time) error
	SetStatus(ctx context.Context, ID uint, status Status) error
	SetStatusInProgress(ctx context.Context, ID uint, startedAt time.Time) error
	SetStatusPending(ctx context.Context, ID uint, taskToken []byte) error
	SetOriginalID(ctx context.Context, ID uint, originalID string) error

	// HTTPDownload returns a HTTP handler that serves the package over HTTP.
	//
	// TODO: this service is meant to be agnostic to protocols. But I haven't
	// found a way in goagen to have my service write directly to the HTTP
	// response writer. Ideally, our goacollection.Service would have a new
	// method that takes a io.Writer (e.g. http.ResponseWriter).
	HTTPDownload(mux goahttp.Muxer, dec func(r *http.Request) goahttp.Decoder) http.HandlerFunc
}

type Status

type Status uint

See https://gist.github.com/sevein/dd36c2af23fd0d9e2e2438d8eb091314.

const (
	StatusNew        Status = iota // Unused!
	StatusInProgress               // Undergoing work.
	StatusDone                     // Work has completed.
	StatusError                    // Processing failed.
	StatusUnknown                  // Unused!
	StatusQueued                   // Awaiting resource allocation.
	StatusAbandoned                // User abandoned processing.
	StatusPending                  // Awaiting user decision.
)

func NewStatus

func NewStatus(status string) Status

func (Status) MarshalJSON

func (p Status) MarshalJSON() ([]byte, error)

func (Status) String

func (p Status) String() string

func (*Status) UnmarshalJSON

func (p *Status) UnmarshalJSON(b []byte) error