collection

package
v0.34.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Feb 1, 2022 License: Apache-2.0 Imports: 28 Imported by: 0

Documentation

Overview

Package collection manages persistency and lifecycle of collections.

Index

Constants

View Source 
const (
	BulkWorkflowName              = "collection-bulk-workflow"
	BulkWorkflowID                = "collection-bulk-workflow"
	BulkWorkflowStateQueryHandler = "collection-bulk-state"
	BulkActivityName              = "collection-bulk-activity"
)
View Source 
const (
	// EventBufferSize is the buffer size of the channel for each subscription.
	EventBufferSize = 16

	EventTypeCollectionCreated = "collection:created"
	EventTypeCollectionUpdated = "collection:updated"
	EventTypeCollectionDeleted = "collection:deleted"
)
View Source 
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.
	// Ten years is the timeout we also use in activities (policies.go).
	ProcessingWorkflowStartToCloseTimeout = time.Hour * 24 * 365 * 10
)

Variables

View Source 
var ErrBulkStatusUnavailable = errors.New("bulk status unavailable")

Functions

func BulkWorkflow added in v0.26.0 

func BulkWorkflow(ctx workflow.Context, params BulkWorkflowInput) error

BulkWorkflow is a Cadence workflow that performs bulk operations.

func InitProcessingWorkflow 

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

func NewService 

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

Types

type BulkActivity added in v0.26.0 

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

func NewBulkActivity added in v0.26.0 

func NewBulkActivity(colsvc Service) *BulkActivity

func (*BulkActivity) Execute added in v0.26.0 

func (a *BulkActivity) Execute(ctx context.Context, params BulkWorkflowInput) error

func (*BulkActivity) Retry added in v0.26.0 

func (a *BulkActivity) Retry(ctx context.Context, ID uint) error

type BulkProgress added in v0.26.0 

type BulkProgress struct {
	CurrentID uint
	Count     uint
	Max       uint
}

BulkProgress reports bulk operation status - delivered as heartbeats.

type BulkWorkflowInput added in v0.26.0 

type BulkWorkflowInput struct {
	// Status of collections where bulk is performed.
	Status Status

	// Type of operation that is performed, e.g. "retry", "cancel"...
	Operation BulkWorkflowOperation

	// Max. number of collections affected. Zero means no cap established.
	Size uint
}

type BulkWorkflowOperation added in v0.26.0 

type BulkWorkflowOperation string
const (
	BulkWorkflowOperationRetry   BulkWorkflowOperation = "retry"
	BulkWorkflowOperationCancel  BulkWorkflowOperation = "cancel"
	BulkWorkflowOperationAbandon BulkWorkflowOperation = "abandon"
)

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 EventService added in v0.34.0 

type EventService interface {
	// Publishes an event to a user's event listeners.
	// If the user is not currently subscribed then this is a no-op.
	PublishEvent(event *goacollection.EnduroMonitorUpdate)

	// Creates a subscription. Caller must call Subscription.Close() when done
	// with the subscription.
	Subscribe(ctx context.Context) (Subscription, error)
}

EventService represents a service for managing event dispatch and event listeners (aka subscriptions).

func NopEventService added in v0.34.0 

func NopEventService() EventService

NopEventService returns an event service that does nothing.

type EventServiceImpl added in v0.34.0 

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

EventService represents a service for managing events in the system.

func NewEventService added in v0.34.0 

func NewEventService() *EventServiceImpl

NewEventService returns a new instance of EventService.

func (*EventServiceImpl) PublishEvent added in v0.34.0 

func (s *EventServiceImpl) PublishEvent(event *goacollection.EnduroMonitorUpdate)

PublishEvent publishes event to all of a user's subscriptions.

If user's channel is full then the user is disconnected. This is to prevent slow users from blocking progress.

func (*EventServiceImpl) Subscribe added in v0.34.0 

func (s *EventServiceImpl) Subscribe(ctx context.Context) (Subscription, error)

Subscribe creates a new subscription.

func (*EventServiceImpl) Unsubscribe added in v0.34.0 

func (s *EventServiceImpl) Unsubscribe(sub *SubscriptionImpl)

Unsubscribe disconnects sub from the service.

type ProcessingWorkflowRequest added in v0.4.0 

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

	// Name of the watcher that received this blob.
	WatcherName string

	// Pipeline name.
	PipelineName string

	// Period of time to schedule the deletion of the original blob from the
	// watched data source. nil means no deletion.
	RetentionPeriod *time.Duration

	// Directory where the transfer is moved to once processing has completed
	// successfully.
	CompletedDir string

	// Whether the top-level directory is meant to be stripped.
	StripTopLevelDir bool

	// Key of the blob.
	Key string

	// Whether the blob is a directory (fs watcher)
	IsDir bool

	// Batch directory that contains the blob.
	BatchDir string

	// Configuration for the validating the transfer.
	ValidationConfig validation.Config

	// Processing configuration name.
	ProcessingConfig string
}

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

type Subscription added in v0.34.0 

type Subscription interface {
	// Event stream for all user's event.
	C() <-chan *goacollection.EnduroMonitorUpdate

	// Closes the event stream channel and disconnects from the event service.
	Close() error
}

Subscription represents a stream of events for a single user.

type SubscriptionImpl added in v0.34.0 

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

SubscriptionImpl represents a stream of user-related events.

func (*SubscriptionImpl) C added in v0.34.0 

func (s *SubscriptionImpl) C() <-chan *goacollection.EnduroMonitorUpdate

C returns a receive-only channel of user-related events.

func (*SubscriptionImpl) Close added in v0.34.0 

func (s *SubscriptionImpl) Close() error

Close disconnects the subscription from the service it was created from.

Source Files

View all Source files

Directories

Path Synopsis
Package fake is a generated GoMock package.
 Package fake is a generated GoMock package.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.