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"`
	Status     Status `db:"status"`

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

	// Nullable and only 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() 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

	// 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

const (
	StatusNew Status = iota
	StatusInProgress
	StatusDone
	StatusError
	StatusUnknown
)

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