pipeline: create pipeline type

Author: sevein

internal/pipeline/pipeline.go

@@ -1,8 +1,6 @@
 package pipeline
 
 import (
-	"errors"
-	"fmt"
 	"io/ioutil"
 	"net"
 	"net/http"
@@ -15,8 +13,6 @@ import (
 	"github.com/artefactual-labs/enduro/internal/amclient"
 )
 
-// TODO: PipelineRegistry should return Pipeline objects!
-
 type Config struct {
 	Name               string
 	BaseURL            string
@@ -28,68 +24,36 @@ type Config struct {
 	ProcessingConfig   string
 }
 
-// PipelineRegistry is a collection of known pipelines.
-type PipelineRegistry struct {
-	pipelines map[string]Config
+type Pipeline struct {
+	config *Config
+	client *http.Client
 }
 
-func NewPipelineRegistry(configs []Config) *PipelineRegistry {
-	pipelines := map[string]Config{}
-	for _, cfg := range configs {
-		cfg.TransferDir = expandPath(cfg.TransferDir)
-		pipelines[cfg.Name] = cfg
+func NewPipeline(config *Config) *Pipeline {
+	config.TransferDir = expandPath(config.TransferDir)
+	config.ProcessingDir = expandPath(config.ProcessingDir)
 
-	}
-	return &PipelineRegistry{
-		pipelines: pipelines,
+	return &Pipeline{
+		config: config,
+		client: httpClient(),
 	}
 }
 
-func (p PipelineRegistry) Config(name string) (*Config, error) {
-	cfg, ok := p.pipelines[name]
-	if !ok {
-		return nil, errors.New("client not found")
-	}
-
-	return &cfg, nil
+// Client returns the Archivematica API client ready for use.
+func (p Pipeline) Client() *amclient.Client {
+	return amclient.NewClient(p.client, p.config.BaseURL, p.config.User, p.config.Key)
 }
 
-func (p PipelineRegistry) Client(name string) (*amclient.Client, error) {
-	cfg, err := p.Config(name)
-	if err != nil {
-		return nil, fmt.Errorf("error fetching pipeline configuration: %w", err)
+// TempFile creates a temporary file in the processing directory.
+func (p Pipeline) TempFile(pattern string) (*os.File, error) {
+	if pattern == "" {
+		pattern = "blob-*"
 	}
-
-	client, err := amclient.New(httpClient(), cfg.BaseURL, cfg.User, cfg.Key)
-	if err != nil {
-		return nil, fmt.Errorf("error creating Archivematica API client: %w", err)
-	}
-
-	return client, nil
+	return ioutil.TempFile(p.config.ProcessingDir, pattern)
 }
 
-// TempFile returns a new temporary file inside the processing directory of the
-// given pipeline.
-func (p PipelineRegistry) TempFile(name string) (*os.File, error) {
-	cfg, err := p.Config(name)
-	if err != nil {
-		return nil, fmt.Errorf("error fetching pipeline configuration: %w", err)
-	}
-
-	return ioutil.TempFile(cfg.ProcessingDir, "blob-*")
-}
-
-func expandPath(path string) string {
-	usr, _ := user.Current()
-	dir := usr.HomeDir
-
-	if path == "~" {
-		path = dir
-	} else if strings.HasPrefix(path, "~/") {
-		path = filepath.Join(dir, path[2:])
-	}
-
-	return path
+func (p Pipeline) Config() *Config {
+	return p.config
 }
 
 func httpClient() *http.Client {
@@ -110,3 +74,16 @@ func httpClient() *http.Client {
 		Transport: transport,
 	}
 }
+
+func expandPath(path string) string {
+	usr, _ := user.Current()
+	dir := usr.HomeDir
+
+	if path == "~" {
+		path = dir
+	} else if strings.HasPrefix(path, "~/") {
+		path = filepath.Join(dir, path[2:])
+	}
+
+	return path
+}

internal/pipeline/registry.go

@@ -0,0 +1,36 @@
+package pipeline
+
+import (
+	"errors"
+	"sync"
+)
+
+var ErrUnknownPipeline = errors.New("unknown pipeline")
+
+// Registry is a collection of known pipelines.
+type Registry struct {
+	pipelines map[string]*Pipeline
+	mu        sync.Mutex
+}
+
+func NewPipelineRegistry(configs []Config) *Registry {
+	pipelines := map[string]*Pipeline{}
+	for _, config := range configs {
+		pipelines[config.Name] = NewPipeline(&config)
+	}
+	return &Registry{
+		pipelines: pipelines,
+	}
+}
+
+func (r *Registry) Pipeline(name string) (*Pipeline, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	pipeline, ok := r.pipelines[name]
+	if !ok {
+		return nil, ErrUnknownPipeline
+	}
+
+	return pipeline, nil
+}

internal/workflow/download_activity.go

@@ -21,8 +21,12 @@ func (a *DownloadActivity) Execute(ctx context.Context, event *watcher.BlobEvent
 	if event == nil {
 		return "", nonRetryableError(errors.New("error reading parameters"))
 	}
+	p, err := a.manager.Pipelines.Pipeline(event.PipelineName)
+	if err != nil {
+		return "", nonRetryableError(err)
+	}
 
-	file, err := a.manager.Pipelines.TempFile(event.PipelineName)
+	file, err := p.TempFile("blob-*")
 	if err != nil {
 		return "", nonRetryableError(fmt.Errorf("error creating temporary file in processing directory: %v", err))
 	}

internal/workflow/hide_package_activity.go

@@ -14,10 +14,11 @@ func NewHidePackageActivity(m *Manager) *HidePackageActivity {
 }
 
 func (a *HidePackageActivity) Execute(ctx context.Context, unitID, unitType, pipelineName string) error {
-	amc, err := a.manager.Pipelines.Client(pipelineName)
+	p, err := a.manager.Pipelines.Pipeline(pipelineName)
 	if err != nil {
-		return nonRetryableError(fmt.Errorf("error looking up pipeline config: %v", err))
+		return nonRetryableError(err)
 	}
+	amc := p.Client()
 
 	if unitType != "transfer" && unitType != "ingest" {
 		return nonRetryableError(fmt.Errorf("unexpected unit type: %s", unitType))

internal/workflow/manager.go

@@ -15,12 +15,12 @@ type Manager struct {
 	Logger     logr.Logger
 	Collection collection.Service
 	Watcher    watcher.Service
-	Pipelines  *pipeline.PipelineRegistry
+	Pipelines  *pipeline.Registry
 	Hooks      map[string]map[string]interface{}
 }
 
 // NewManager returns a pointer to a new Manager.
-func NewManager(logger logr.Logger, colsvc collection.Service, wsvc watcher.Service, pipelines *pipeline.PipelineRegistry, hooks map[string]map[string]interface{}) *Manager {
+func NewManager(logger logr.Logger, colsvc collection.Service, wsvc watcher.Service, pipelines *pipeline.Registry, hooks map[string]map[string]interface{}) *Manager {
 	return &Manager{
 		Logger:     logger,
 		Collection: colsvc,

internal/workflow/poll_ingest_activity.go

@@ -18,11 +18,17 @@ func NewPollIngestActivity(m *Manager) *PollIngestActivity {
 	return &PollIngestActivity{manager: m}
 }
 
-func (a *PollIngestActivity) Execute(ctx context.Context, tinfo *TransferInfo) (time.Time, error) {
-	amc, err := a.manager.Pipelines.Client(tinfo.Event.PipelineName)
+type PollIngestActivityParams struct {
+	PipelineName string
+	SIPID        string
+}
+
+func (a *PollIngestActivity) Execute(ctx context.Context, params *PollIngestActivityParams) (time.Time, error) {
+	p, err := a.manager.Pipelines.Pipeline(params.PipelineName)
 	if err != nil {
-		return time.Time{}, err
+		return time.Time{}, nonRetryableError(err)
 	}
+	amc := p.Client()
 
 	var backoffStrategy = backoff.WithContext(backoff.NewConstantBackOff(time.Second*5), ctx)
 
@@ -31,7 +37,7 @@ func (a *PollIngestActivity) Execute(ctx context.Context, tinfo *TransferInfo) (
 			ctx, cancel := context.WithTimeout(ctx, time.Second*2)
 			defer cancel()
 
-			err = pipeline.IngestStatus(ctx, amc, tinfo.SIPID)
+			err = pipeline.IngestStatus(ctx, amc, params.SIPID)
 			if errors.Is(err, pipeline.ErrStatusNonRetryable) {
 				return backoff.Permanent(nonRetryableError(err))
 			}

internal/workflow/poll_transfer_activity.go

@@ -18,11 +18,17 @@ func NewPollTransferActivity(m *Manager) *PollTransferActivity {
 	return &PollTransferActivity{manager: m}
 }
 
-func (a *PollTransferActivity) Execute(ctx context.Context, tinfo *TransferInfo) (string, error) {
-	amc, err := a.manager.Pipelines.Client(tinfo.Event.PipelineName)
+type PollTransferActivityParams struct {
+	PipelineName string
+	TransferID   string
+}
+
+func (a *PollTransferActivity) Execute(ctx context.Context, params *PollTransferActivityParams) (string, error) {
+	p, err := a.manager.Pipelines.Pipeline(params.PipelineName)
 	if err != nil {
 		return "", err
 	}
+	amc := p.Client()
 
 	var sipID string
 	var backoffStrategy = backoff.WithContext(backoff.NewConstantBackOff(time.Second*5), ctx)
@@ -32,7 +38,7 @@ func (a *PollTransferActivity) Execute(ctx context.Context, tinfo *TransferInfo)
 			ctx, cancel := context.WithTimeout(ctx, time.Second*2)
 			defer cancel()
 
-			sipID, err = pipeline.TransferStatus(ctx, amc, tinfo.TransferID)
+			sipID, err = pipeline.TransferStatus(ctx, amc, params.TransferID)
 			if errors.Is(err, pipeline.ErrStatusNonRetryable) {
 				return backoff.Permanent(nonRetryableError(err))
 			}

internal/workflow/processing.go

@@ -264,7 +264,14 @@ func (w *ProcessingWorkflow) SessionHandler(ctx workflow.Context, sessCtx workfl
 
 	// Transfer.
 	activityOpts = withActivityOptsForRequest(sessCtx)
-	err = workflow.ExecuteActivity(activityOpts, TransferActivityName, tinfo).Get(activityOpts, &tinfo.TransferID)
+	err = workflow.ExecuteActivity(activityOpts, TransferActivityName, &TransferActivityParams{
+		PipelineName:       tinfo.Event.PipelineName,
+		TransferLocationID: tinfo.PipelineConfig.TransferLocationID,
+		RelPath:            tinfo.Bundle.RelPath,
+		Name:               tinfo.Bundle.Name,
+		ProcessingConfig:   tinfo.PipelineConfig.ProcessingConfig,
+		AutoApprove:        true,
+	}).Get(activityOpts, &tinfo.TransferID)
 	if err != nil {
 		return err
 	}
@@ -275,14 +282,20 @@ func (w *ProcessingWorkflow) SessionHandler(ctx workflow.Context, sessCtx workfl
 
 	// Poll transfer.
 	activityOpts = withActivityOptsForHeartbeatedRequest(sessCtx, time.Minute)
-	err = workflow.ExecuteActivity(activityOpts, PollTransferActivityName, tinfo).Get(activityOpts, &tinfo.SIPID)
+	err = workflow.ExecuteActivity(activityOpts, PollTransferActivityName, &PollTransferActivityParams{
+		PipelineName: tinfo.Event.PipelineName,
+		TransferID:   tinfo.TransferID,
+	}).Get(activityOpts, &tinfo.SIPID)
 	if err != nil {
 		return err
 	}
 
 	// Poll ingest.
 	activityOpts = withActivityOptsForHeartbeatedRequest(sessCtx, time.Minute)
-	err = workflow.ExecuteActivity(activityOpts, PollIngestActivityName, tinfo).Get(activityOpts, &tinfo.StoredAt)
+	err = workflow.ExecuteActivity(activityOpts, PollIngestActivityName, &PollIngestActivityParams{
+		PipelineName: tinfo.Event.PipelineName,
+		SIPID:        tinfo.SIPID,
+	}).Get(activityOpts, &tinfo.StoredAt)
 	if err != nil {
 		return err
 	}
@@ -327,11 +340,12 @@ func updatePackageStatusLocalActivity(ctx context.Context, colsvc collection.Ser
 }
 
 func loadConfigLocalActivity(ctx context.Context, m *Manager, pipeline string, tinfo *TransferInfo) (*TransferInfo, error) {
-	pipelineConfig, err := m.Pipelines.Config(tinfo.Event.PipelineName)
+	p, err := m.Pipelines.Pipeline(pipeline)
 	if err != nil {
-		return nil, fmt.Errorf("error loading configuration of pipeline %s: %v", pipeline, err)
+		return nil, err
 	}
-	tinfo.PipelineConfig = pipelineConfig
+
+	tinfo.PipelineConfig = p.Config()
 	tinfo.Hooks = m.Hooks
 
 	return tinfo, nil

internal/workflow/transfer_activity.go

@@ -20,24 +20,33 @@ func NewTransferActivity(m *Manager) *TransferActivity {
 	return &TransferActivity{manager: m}
 }
 
-func (a *TransferActivity) Execute(ctx context.Context, tinfo *TransferInfo) (string, error) {
-	amc, err := a.manager.Pipelines.Client(tinfo.PipelineConfig.Name)
+type TransferActivityParams struct {
+	PipelineName       string
+	TransferLocationID string
+	RelPath            string
+	Name               string
+	ProcessingConfig   string
+	AutoApprove        bool
+}
+
+func (a *TransferActivity) Execute(ctx context.Context, params *TransferActivityParams) (string, error) {
+	p, err := a.manager.Pipelines.Pipeline(params.PipelineName)
 	if err != nil {
-		return "", err
+		return "", nonRetryableError(err)
 	}
+	amc := p.Client()
 
 	// Transfer path should include the location UUID if defined.
-	var path = tinfo.Bundle.RelPath
-	if tinfo.PipelineConfig.TransferLocationID != "" {
-		path = fmt.Sprintf("%s:%s", tinfo.PipelineConfig.TransferLocationID, path)
+	var path = params.RelPath
+	if params.TransferLocationID != "" {
+		path = fmt.Sprintf("%s:%s", params.TransferLocationID, path)
 	}
 
-	var autoApprove = true
 	resp, httpResp, err := amc.Package.Create(ctx, &amclient.PackageCreateRequest{
-		Name:             tinfo.Bundle.Name,
+		Name:             params.Name,
 		Path:             path,
-		ProcessingConfig: tinfo.PipelineConfig.ProcessingConfig,
-		AutoApprove:      &autoApprove,
+		ProcessingConfig: params.ProcessingConfig,
+		AutoApprove:      &params.AutoApprove,
 	})
 	if err != nil {
 		if httpResp != nil {