collection

package
v0.41.8 Latest Latest
Warning

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

 Go to latest Published: Jul 31, 2024 License: Apache-2.0 Imports: 31 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 ProcessingWorkflowName = "processing-workflow"

Name of the collection processing workflow.

Variables

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

Functions

func BulkWorkflow added in v0.26.0 

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

BulkWorkflow is a Temporal workflow that performs bulk operations.

func InitProcessingWorkflow 

func InitProcessingWorkflow(ctx context.Context, tr trace.Tracer, c temporalsdk_client.Client, taskQueue string, req *ProcessingWorkflowRequest) error

func NewService 

func NewService(logger logr.Logger, db *sql.DB, cc temporalsdk_client.Client, taskQueue string, 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 []byte `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

	// Pipelines that are available for processing. The workflow will choose one
	// (randomly picked for now). If the slice is empty, it will be
	// automatically populated from the list of all configured pipelines.
	PipelineNames []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 validating the transfer.
	ValidationConfig validation.Config

	// Processing configuration name.
	ProcessingConfig string

	// Whether we reject duplicates based on name (key).
	RejectDuplicates bool

	// Whether we exclude hidden files from submission.
	ExcludeHiddenFiles bool

	// Transfer type.
	TransferType string

	// Configuration for metadata management.
	MetadataConfig metadata.Config
}

type Service 

type Service interface {
	// Goa returns an implementation of the goacollection Service.
	Goa() goacollection.Service
	Create(context.Context, *Collection) error
	CheckDuplicate(ctx context.Context, id uint) (bool, 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
}

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.