Separate Deps and Event in hook triggering (#1463)

actions/event.go

@@ -13,10 +13,13 @@ const (
 	EventTypePreMerge  EventType = "pre-merge"
 )
 
+type Deps struct {
+	Source Source
+	Output OutputWriter
+}
+
 type Event struct {
 	EventID       uuid.UUID
-	Source        Source
-	Output        OutputWriter
 	EventType     EventType
 	EventTime     time.Time
 	RepositoryID  string

actions/hook.go

@@ -14,7 +14,7 @@ const (
 
 // Hook is the abstraction of the basic user-configured runnable building-stone
 type Hook interface {
-	Run(ctx context.Context, event Event) error
+	Run(ctx context.Context, event Event, writer *HookOutputWriter) error
 }
 
 type NewHookFunc func(ActionHook, *Action) (Hook, error)

actions/hook_output_writer.go

@@ -13,11 +13,11 @@ type HookOutputWriter struct {
 	Writer     OutputWriter
 }
 
-func (h *HookOutputWriter) OutputWrite(ctx context.Context, name string, reader io.Reader) error {
-	outputPath := FormatHookOutputPath(h.RunID, h.ActionName, h.HookID, name)
-	return h.Writer.OutputWrite(ctx, outputPath, reader)
+func (h *HookOutputWriter) OutputWrite(ctx context.Context, reader io.Reader, size int64) error {
+	outputPath := FormatHookOutputPath(h.RunID, h.ActionName, h.HookID)
+	return h.Writer.OutputWrite(ctx, outputPath, reader, size)
 }
 
-func FormatHookOutputPath(runID, actionName, hookID, name string) string {
-	return path.Join(runID, actionName, hookID+"_"+name)
+func FormatHookOutputPath(runID, actionName, hookID string) string {
+	return path.Join(runID, actionName, hookID)
 }

actions/hook_output_writer_test.go

@@ -13,21 +13,22 @@ import (
 
 func TestHookWriter_OutputWritePath(t *testing.T) {
 	ctx := context.Background()
-	contentReader := strings.NewReader("content")
+	content := "content"
+	contentReader := strings.NewReader(content)
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
 
-	hookOutput := actions.FormatHookOutputPath("runID", "actionName", "hookID", "name1")
+	hookOutput := actions.FormatHookOutputPath("runID", "actionName", "hookID")
 	writer := mock.NewMockOutputWriter(ctrl)
-	writer.EXPECT().OutputWrite(ctx, hookOutput, contentReader).Return(nil)
+	writer.EXPECT().OutputWrite(ctx, hookOutput, contentReader, int64(len(content))).Return(nil)
 
 	w := &actions.HookOutputWriter{
 		RunID:      "runID",
 		ActionName: "actionName",
 		HookID:     "hookID",
 		Writer:     writer,
 	}
-	err := w.OutputWrite(ctx, "name1", contentReader)
+	err := w.OutputWrite(ctx, contentReader, int64(len(content)))
 	if err != nil {
 		t.Fatalf("OutputWrite failed with err=%s", err)
 	}
@@ -39,7 +40,7 @@ func TestHookWriter_OutputWriteError(t *testing.T) {
 
 	errSomeError := errors.New("some error")
 	writer := mock.NewMockOutputWriter(ctrl)
-	writer.EXPECT().OutputWrite(gomock.Any(), gomock.Any(), gomock.Any()).Return(errSomeError)
+	writer.EXPECT().OutputWrite(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(errSomeError)
 
 	w := &actions.HookOutputWriter{
 		RunID:      "runID",
@@ -49,7 +50,7 @@ func TestHookWriter_OutputWriteError(t *testing.T) {
 	}
 	ctx := context.Background()
 	contentReader := strings.NewReader("content")
-	err := w.OutputWrite(ctx, "name1", contentReader)
+	err := w.OutputWrite(ctx, contentReader, 10)
 	if !errors.Is(err, errSomeError) {
 		t.Fatalf("OutputWrite() err=%v expected=%v", err, errSomeError)
 	}

actions/output_writer.go

@@ -6,5 +6,5 @@ import (
 )
 
 type OutputWriter interface {
-	OutputWrite(ctx context.Context, name string, reader io.Reader) error
+	OutputWrite(ctx context.Context, name string, reader io.Reader, size int64) error
 }

actions/service.go

@@ -29,9 +29,9 @@ func New(db db.Database) *Service {
 	}
 }
 
-func (s *Service) Run(ctx context.Context, event Event) error {
+func (s *Service) Run(ctx context.Context, event Event, deps Deps) error {
 	// load relevant actions
-	actions, err := s.loadMatchedActions(ctx, event.Source, MatchSpec{EventType: event.EventType, Branch: event.BranchID})
+	actions, err := s.loadMatchedActions(ctx, deps.Source, MatchSpec{EventType: event.EventType, Branch: event.BranchID})
 	if err != nil || len(actions) == 0 {
 		return err
 	}
@@ -41,7 +41,7 @@ func (s *Service) Run(ctx context.Context, event Event) error {
 	if err != nil {
 		return nil
 	}
-	return s.runTasks(ctx, tasks, event)
+	return s.runTasks(ctx, tasks, event, deps)
 }
 
 func (s *Service) loadMatchedActions(ctx context.Context, source Source, spec MatchSpec) ([]*Action, error) {
@@ -74,20 +74,17 @@ func (s *Service) allocateTasks(runID string, actions []*Action) ([]*Task, error
 	return tasks, nil
 }
 
-func (s *Service) runTasks(ctx context.Context, hooks []*Task, event Event) error {
+func (s *Service) runTasks(ctx context.Context, hooks []*Task, event Event, deps Deps) error {
 	var g multierror.Group
 	for _, h := range hooks {
 		hh := h // pinning
 		g.Go(func() error {
-			// set hook's event to have scoped writer
-			hookEvent := event
-			hookEvent.Output = &HookOutputWriter{
+			return hh.Hook.Run(ctx, event, &HookOutputWriter{
 				RunID:      hh.RunID,
 				ActionName: hh.Action.Name,
 				HookID:     hh.HookID,
-				Writer:     event.Output,
-			}
-			return hh.Hook.Run(ctx, hookEvent)
+				Writer:     deps.Output,
+			})
 		})
 	}
 	return g.Wait().ErrorOrNil()

actions/service_test.go

@@ -24,7 +24,7 @@ func TestManager_RunActions(t *testing.T) {
 
 	ctx := context.Background()
 	testOutputWriter := mock.NewMockOutputWriter(ctrl)
-	testOutputWriter.EXPECT().OutputWrite(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil)
+	testOutputWriter.EXPECT().OutputWrite(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil)
 
 	testSource := mock.NewMockSource(ctrl)
 	testSource.EXPECT().List(gomock.Any()).Return([]actions.FileRef{{Path: "act.yaml", Address: "act.addr"}}, nil)
@@ -41,8 +41,6 @@ hooks:
 
 	evtTime := time.Now()
 	evt := actions.Event{
-		Source:        testSource,
-		Output:        testOutputWriter,
 		EventType:     actions.EventTypePreCommit,
 		EventTime:     evtTime,
 		RepositoryID:  "repoID",
@@ -52,8 +50,12 @@ hooks:
 		Committer:     "committer",
 		Metadata:      map[string]string{"key": "value"},
 	}
+	deps := actions.Deps{
+		Source: testSource,
+		Output: testOutputWriter,
+	}
 	actionsManager := actions.New(nil)
-	err := actionsManager.Run(ctx, evt)
+	err := actionsManager.Run(ctx, evt, deps)
 	if err != nil {
 		t.Fatalf("Run() failed with err=%s", err)
 	}

actions/webhook.go

@@ -60,7 +60,7 @@ func NewWebhook(h ActionHook, action *Action) (Hook, error) {
 	}, nil
 }
 
-func (w *Webhook) Run(ctx context.Context, event Event) error {
+func (w *Webhook) Run(ctx context.Context, event Event, writer *HookOutputWriter) error {
 	// post event information as json to webhook endpoint
 	eventData, err := w.marshalEventInformation(event)
 	if err != nil {
@@ -85,7 +85,7 @@ func (w *Webhook) Run(ctx context.Context, event Event) error {
 
 	// log response body if needed
 	if resp.Body != nil && resp.ContentLength > 0 {
-		if err := event.Output.OutputWrite(ctx, w.ID, resp.Body); err != nil {
+		if err := writer.OutputWrite(ctx, resp.Body, resp.ContentLength); err != nil {
 			return err
 		}
 	}

catalog/actions.go

@@ -3,6 +3,7 @@ package catalog
 import (
 	"context"
 	"fmt"
+	"io"
 	"io/ioutil"
 
 	"github.com/treeverse/lakefs/actions"
@@ -59,3 +60,15 @@ func (as *actionsSource) Load(ctx context.Context, fileRef actions.FileRef) ([]b
 	}
 	return bytes, nil
 }
+
+type actionsWriter struct {
+	adapter          block.Adapter
+	storageNamespace graveler.StorageNamespace
+}
+
+func (aw *actionsWriter) OutputWrite(ctx context.Context, path string, reader io.Reader, size int64) error {
+	return aw.adapter.WithContext(ctx).Put(block.ObjectPointer{
+		StorageNamespace: aw.storageNamespace.String(),
+		Identifier:       path,
+	}, size, reader, block.PutOpts{})
+}

catalog/entry_catalog.go

@@ -82,12 +82,12 @@ type Store interface {
 	graveler.Loader
 }
 
-type ActionsClient interface {
-	Run(ctx context.Context, event actions.Event) error
+type Actions interface {
+	Run(ctx context.Context, event actions.Event, deps actions.Deps) error
 }
 
 type EntryCatalog struct {
-	Actions      ActionsClient
+	Actions      Actions
 	BlockAdapter block.Adapter
 	Store        Store
 }
@@ -101,7 +101,7 @@ type Config struct {
 	Config  *config.Config
 	DB      db.Database
 	LockDB  db.Database
-	Actions ActionsClient
+	Actions Actions
 }
 
 func NewEntryCatalog(cfg Config) (*EntryCatalog, error) {
@@ -553,23 +553,29 @@ func (e *EntryCatalog) LoadTags(ctx context.Context, repositoryID graveler.Repos
 
 func (e *EntryCatalog) PreCommitHook(ctx context.Context, eventID uuid.UUID, repositoryRecord graveler.RepositoryRecord, branch graveler.BranchID, commit graveler.Commit) error {
 	evt := actions.Event{
-		EventType: actions.EventTypePreCommit,
-		EventID:   eventID,
-		EventTime: time.Now(),
+		EventType:     actions.EventTypePreCommit,
+		EventID:       eventID,
+		EventTime:     time.Now(),
+		RepositoryID:  repositoryRecord.RepositoryID.String(),
+		BranchID:      branch.String(),
+		CommitMessage: commit.Message,
+		Committer:     commit.Committer,
+		Metadata:      commit.Metadata,
+	}
+	deps := actions.Deps{
 		Source: &actionsSource{
 			catalog:          e,
 			adapter:          e.BlockAdapter,
 			repositoryID:     repositoryRecord.RepositoryID,
 			storageNamespace: repositoryRecord.StorageNamespace,
 			ref:              branch.Ref(),
 		},
-		RepositoryID:  repositoryRecord.RepositoryID.String(),
-		BranchID:      branch.String(),
-		CommitMessage: commit.Message,
-		Committer:     commit.Committer,
-		Metadata:      commit.Metadata,
+		Output: &actionsWriter{
+			adapter:          e.BlockAdapter,
+			storageNamespace: repositoryRecord.StorageNamespace,
+		},
 	}
-	return e.Actions.Run(ctx, evt)
+	return e.Actions.Run(ctx, evt, deps)
 }
 
 func (e *EntryCatalog) PostCommitHook(ctx context.Context, eventID uuid.UUID, repositoryRecord graveler.RepositoryRecord, branch graveler.BranchID, commitRecord graveler.CommitRecord) error {
@@ -578,24 +584,30 @@ func (e *EntryCatalog) PostCommitHook(ctx context.Context, eventID uuid.UUID, re
 
 func (e *EntryCatalog) PreMergeHook(ctx context.Context, eventID uuid.UUID, repositoryRecord graveler.RepositoryRecord, destination graveler.BranchID, source graveler.Ref, commit graveler.Commit) error {
 	evt := actions.Event{
-		EventType: actions.EventTypePreMerge,
-		EventID:   eventID,
-		EventTime: time.Now(),
+		EventType:     actions.EventTypePreMerge,
+		EventID:       eventID,
+		EventTime:     time.Now(),
+		RepositoryID:  repositoryRecord.RepositoryID.String(),
+		BranchID:      destination.String(),
+		SourceRef:     source.String(),
+		CommitMessage: commit.Message,
+		Committer:     commit.Committer,
+		Metadata:      commit.Metadata,
+	}
+	deps := actions.Deps{
 		Source: &actionsSource{
 			catalog:          e,
 			adapter:          e.BlockAdapter,
 			repositoryID:     repositoryRecord.RepositoryID,
 			storageNamespace: repositoryRecord.StorageNamespace,
 			ref:              source,
 		},
-		RepositoryID:  repositoryRecord.RepositoryID.String(),
-		BranchID:      destination.String(),
-		SourceRef:     source.String(),
-		CommitMessage: commit.Message,
-		Committer:     commit.Committer,
-		Metadata:      commit.Metadata,
+		Output: &actionsWriter{
+			adapter:          e.BlockAdapter,
+			storageNamespace: repositoryRecord.StorageNamespace,
+		},
 	}
-	return e.Actions.Run(ctx, evt)
+	return e.Actions.Run(ctx, evt, deps)
 }
 
 func (e *EntryCatalog) PostMergeHook(ctx context.Context, eventID uuid.UUID, repositoryRecord graveler.RepositoryRecord, destination graveler.BranchID, source graveler.Ref, commitRecord graveler.CommitRecord) error {