workflow: add transfer validation capability

Author: replaceafill

enduro.toml

@@ -49,6 +49,9 @@ processingConfig = "automated"
 storageServiceURL = "http://test:test@127.0.0.1:62081"
 capacity = 3
 
+[validation]
+checksumsCheckEnabled = false
+
 [[hooks."hari"]]
 baseURL = "" # E.g.: "https://192.168.1.50:8080/api"
 mock = true

go.mod

@@ -24,6 +24,7 @@ require (
 	github.com/google/uuid v1.1.1
 	github.com/google/wire v0.4.0 // indirect
 	github.com/gorilla/schema v1.1.0
+	github.com/hashicorp/go-multierror v1.1.0
 	github.com/jmoiron/sqlx v1.2.0
 	github.com/mholt/archiver v3.1.1+incompatible
 	github.com/mitchellh/mapstructure v1.3.2 // indirect

internal/collection/workflow.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	cce "github.com/artefactual-labs/enduro/internal/cadence"
+	"github.com/artefactual-labs/enduro/internal/validation"
 	"github.com/artefactual-labs/enduro/internal/watcher"
 
 	"github.com/google/uuid"
@@ -32,12 +33,15 @@ type ProcessingWorkflowRequest struct {
 	// Captured by the watcher, the event contains information about the
 	// incoming dataset.
 	Event *watcher.BlobEvent
+
+	ValidationConfig validation.Config
 }
 
-func InitProcessingWorkflow(ctx context.Context, c client.Client, event *watcher.BlobEvent) error {
+func InitProcessingWorkflow(ctx context.Context, c client.Client, event *watcher.BlobEvent, validationConfig validation.Config) error {
 	req := &ProcessingWorkflowRequest{
-		WorkflowID: fmt.Sprintf("processing-workflow-%s", uuid.New().String()),
-		Event:      event,
+		WorkflowID:       fmt.Sprintf("processing-workflow-%s", uuid.New().String()),
+		Event:            event,
+		ValidationConfig: validationConfig,
 	}
 
 	return TriggerProcessingWorkflow(ctx, c, req)

internal/validation/config.go

@@ -0,0 +1,9 @@
+package validation
+
+type Config struct {
+	ChecksumsCheckEnabled bool
+}
+
+func (c Config) IsEnabled() bool {
+	return c.ChecksumsCheckEnabled
+}

internal/validation/validate.go

@@ -0,0 +1,56 @@
+package validation
+
+import (
+	"fmt"
+	"os"
+	"path"
+
+	"github.com/hashicorp/go-multierror"
+)
+
+var checksumFiles = [4]string{
+	"checksum.md5",
+	"checksum.sha1",
+	"checksum.sha256",
+	"checksum.sha512",
+}
+
+func ValidateTransfer(c Config, path string) error {
+	var result error
+
+	if c.ChecksumsCheckEnabled {
+		v := ChecksumExistsValidator{path: path}
+		if err := v.Valid(); err != nil {
+			result = multierror.Append(result, err)
+		}
+	}
+
+	return result
+}
+
+// Validator is the interface that all validators must implement.
+type Validator interface {
+	Valid() error
+}
+
+// ChecksumExistsValidator is a Validator that checks...
+type ChecksumExistsValidator struct {
+	path string
+}
+
+func (v ChecksumExistsValidator) Valid() error {
+	for _, checksum := range checksumFiles {
+		if fileExists(path.Join(v.path, "metadata", checksum)) {
+			return nil
+		}
+	}
+	return fmt.Errorf("Transfer does not contain checksums (path=%s)", v.path)
+}
+
+func fileExists(name string) bool {
+	stat, err := os.Stat(name)
+	if err != nil {
+		return false
+	}
+	return !stat.IsDir()
+}

internal/validation/validate_test.go

@@ -0,0 +1,68 @@
+package validation
+
+import (
+	"testing"
+
+	"gotest.tools/v3/assert"
+	"gotest.tools/v3/fs"
+)
+
+func TestChecksumExistsValidator(t *testing.T) {
+	tests := map[string]struct {
+		dirOpts      []fs.PathOp
+		errorMessage string
+	}{
+		"Validates when checksum.md5 file exists": {
+			dirOpts: []fs.PathOp{
+				fs.WithDir("metadata",
+					fs.WithFile("checksum.md5", ""),
+				),
+			},
+		},
+		"Validates when checksum.sha1 file exists": {
+			dirOpts: []fs.PathOp{
+				fs.WithDir("metadata",
+					fs.WithFile("checksum.sha1", ""),
+				),
+			},
+		},
+		"Validates when checksum.sha256 file exists": {
+			dirOpts: []fs.PathOp{
+				fs.WithDir("metadata",
+					fs.WithFile("checksum.sha256", ""),
+				),
+			},
+		},
+		"Validates when checksum.sha512 file exists": {
+			dirOpts: []fs.PathOp{
+				fs.WithDir("metadata",
+					fs.WithFile("checksum.sha512", ""),
+				),
+			},
+		},
+		"Fails when expected files do not exist": {
+			dirOpts: []fs.PathOp{
+				fs.WithDir("metadata",
+					fs.WithFile("notchecksum.txt", ""),
+				),
+			},
+			errorMessage: "Transfer does not contain checksums",
+		},
+	}
+	for name, tc := range tests {
+		name, tc := name, tc
+		t.Run(name, func(t *testing.T) {
+			tmpDir := fs.NewDir(t, "transfer", tc.dirOpts...)
+			defer tmpDir.Remove()
+
+			validator := ChecksumExistsValidator{path: tmpDir.Path()}
+			err := validator.Valid()
+
+			if tc.errorMessage == "" {
+				assert.NilError(t, err)
+			} else {
+				assert.ErrorContains(t, err, tc.errorMessage)
+			}
+		})
+	}
+}

internal/workflow/activities/activities.go

@@ -2,13 +2,14 @@
 package activities
 
 const (
-	AcquirePipelineActivityName = "acquire-pipeline-activity"
-	DownloadActivityName        = "download-activity"
-	BundleActivityName          = "bundle-activity"
-	TransferActivityName        = "transfer-activity"
-	PollTransferActivityName    = "poll-transfer-activity"
-	PollIngestActivityName      = "poll-ingest-activity"
-	CleanUpActivityName         = "clean-up-activity"
-	HidePackageActivityName     = "hide-package-activity"
-	DeleteOriginalActivityName  = "delete-original-activity"
+	AcquirePipelineActivityName  = "acquire-pipeline-activity"
+	DownloadActivityName         = "download-activity"
+	BundleActivityName           = "bundle-activity"
+	TransferActivityName         = "transfer-activity"
+	PollTransferActivityName     = "poll-transfer-activity"
+	PollIngestActivityName       = "poll-ingest-activity"
+	CleanUpActivityName          = "clean-up-activity"
+	HidePackageActivityName      = "hide-package-activity"
+	DeleteOriginalActivityName   = "delete-original-activity"
+	ValidateTransferActivityName = "validate-transfer-activity"
 )

internal/workflow/activities/validate.go

@@ -0,0 +1,22 @@
+package activities
+
+import (
+	"context"
+
+	"github.com/artefactual-labs/enduro/internal/validation"
+)
+
+type ValidateTransferActivity struct{}
+
+func NewValidateTransferActivity() *ValidateTransferActivity {
+	return &ValidateTransferActivity{}
+}
+
+type ValidateTransferActivityParams struct {
+	Config validation.Config
+	Path   string
+}
+
+func (a *ValidateTransferActivity) Execute(ctx context.Context, params *ValidateTransferActivityParams) error {
+	return validation.ValidateTransfer(params.Config, params.Path)
+}

internal/workflow/processing.go

@@ -15,6 +15,7 @@ import (
 	"github.com/artefactual-labs/enduro/internal/nha"
 	nha_activities "github.com/artefactual-labs/enduro/internal/nha/activities"
 	"github.com/artefactual-labs/enduro/internal/pipeline"
+	"github.com/artefactual-labs/enduro/internal/validation"
 	"github.com/artefactual-labs/enduro/internal/watcher"
 	"github.com/artefactual-labs/enduro/internal/workflow/activities"
 	"github.com/artefactual-labs/enduro/internal/workflow/manager"
@@ -208,7 +209,7 @@ func (w *ProcessingWorkflow) Execute(ctx workflow.Context, req *collection.Proce
 				return fmt.Errorf("error creating session: %v", err)
 			}
 
-			sessErr = w.SessionHandler(sessCtx, attempt, tinfo, nameInfo)
+			sessErr = w.SessionHandler(sessCtx, attempt, tinfo, nameInfo, req.ValidationConfig)
 
 			// We want to retry the session if it has been canceled as a result
 			// of losing the worker but not otherwise. This scenario seems to be
@@ -283,7 +284,7 @@ func (w *ProcessingWorkflow) Execute(ctx workflow.Context, req *collection.Proce
 }
 
 // SessionHandler runs activities that belong to the same session.
-func (w *ProcessingWorkflow) SessionHandler(sessCtx workflow.Context, attempt int, tinfo *TransferInfo, nameInfo nha.NameInfo) error {
+func (w *ProcessingWorkflow) SessionHandler(sessCtx workflow.Context, attempt int, tinfo *TransferInfo, nameInfo nha.NameInfo, validationConfig validation.Config) error {
 	defer func() {
 		_ = releasePipeline(sessCtx, w.manager, tinfo.Event.PipelineName)
 		workflow.CompleteSession(sessCtx)
@@ -328,6 +329,23 @@ func (w *ProcessingWorkflow) SessionHandler(sessCtx workflow.Context, attempt in
 		}
 	}
 
+	// Validate transfer.
+	{
+		if validationConfig.IsEnabled() && tinfo.Bundle != (activities.BundleActivityResult{}) {
+			activityOpts := workflow.WithActivityOptions(sessCtx, workflow.ActivityOptions{
+				ScheduleToStartTimeout: forever,
+				StartToCloseTimeout:    time.Minute * 5,
+			})
+			err := workflow.ExecuteActivity(activityOpts, activities.ValidateTransferActivityName, &activities.ValidateTransferActivityParams{
+				Config: validationConfig,
+				Path:   tinfo.Bundle.FullPath,
+			}).Get(activityOpts, nil)
+			if err != nil {
+				return err
+			}
+		}
+	}
+
 	// Transfer.
 	{
 		if tinfo.TransferID == "" {

main.go

@@ -19,6 +19,7 @@ import (
 	"github.com/artefactual-labs/enduro/internal/db"
 	nha_activities "github.com/artefactual-labs/enduro/internal/nha/activities"
 	"github.com/artefactual-labs/enduro/internal/pipeline"
+	"github.com/artefactual-labs/enduro/internal/validation"
 	"github.com/artefactual-labs/enduro/internal/watcher"
 	"github.com/artefactual-labs/enduro/internal/workflow"
 	"github.com/artefactual-labs/enduro/internal/workflow/activities"
@@ -191,7 +192,7 @@ func main() {
 							}
 							logger.V(1).Info("Starting new workflow", "watcher", event.WatcherName, "bucket", event.Bucket, "key", event.Key)
 							go func() {
-								if err := collection.InitProcessingWorkflow(ctx, workflowClient, event); err != nil {
+								if err := collection.InitProcessingWorkflow(ctx, workflowClient, event, config.Validation); err != nil {
 									logger.Error(err, "Error initializing processing workflow.")
 								}
 							}()
@@ -217,6 +218,7 @@ func main() {
 		cadence.RegisterActivity(activities.NewAcquirePipelineActivity(m).Execute, activities.AcquirePipelineActivityName)
 		cadence.RegisterActivity(activities.NewDownloadActivity(m).Execute, activities.DownloadActivityName)
 		cadence.RegisterActivity(activities.NewBundleActivity().Execute, activities.BundleActivityName)
+		cadence.RegisterActivity(activities.NewValidateTransferActivity().Execute, activities.ValidateTransferActivityName)
 		cadence.RegisterActivity(activities.NewTransferActivity(m).Execute, activities.TransferActivityName)
 		cadence.RegisterActivity(activities.NewPollTransferActivity(m).Execute, activities.PollTransferActivityName)
 		cadence.RegisterActivity(activities.NewPollIngestActivity(m).Execute, activities.PollIngestActivityName)
@@ -326,6 +328,7 @@ type configuration struct {
 	Cadence     cadence.Config
 	Watcher     watcher.Config
 	Pipeline    []pipeline.Config
+	Validation  validation.Config
 
 	// This is a workaround for client-specific functionality.
 	// Simple mechanism to support an arbitrary number of hooks and parameters.