Actions: support storage ID (#8651)

* Actions: support storage ID

* Fix tests

pkg/actions/event.go

@@ -29,7 +29,7 @@ func marshalEventInformation(actionName, hookID string, record graveler.HookReco
 		EventTime:      now.UTC().Format(time.RFC3339),
 		ActionName:     actionName,
 		HookID:         hookID,
-		RepositoryID:   record.RepositoryID.String(),
+		RepositoryID:   record.Repository.RepositoryID.String(),
 		BranchID:       record.BranchID.String(),
 		SourceRef:      record.SourceRef.String(),
 		TagID:          record.TagID.String(),

pkg/actions/hook_output_writer.go

@@ -4,15 +4,17 @@ import (
 	"context"
 	"io"
 	"path"
+
+	"github.com/treeverse/lakefs/pkg/graveler"
 )
 
 type HookOutputWriter struct {
-	StorageNamespace string
-	RunID            string
-	HookRunID        string
-	ActionName       string
-	HookID           string
-	Writer           OutputWriter
+	Repository *graveler.RepositoryRecord
+	RunID      string
+	HookRunID  string
+	ActionName string
+	HookID     string
+	Writer     OutputWriter
 }
 
 const (
@@ -23,7 +25,7 @@ const (
 
 func (h *HookOutputWriter) OutputWrite(ctx context.Context, reader io.Reader, size int64) error {
 	name := FormatHookOutputPath(h.RunID, h.HookRunID)
-	return h.Writer.OutputWrite(ctx, h.StorageNamespace, name, reader, size)
+	return h.Writer.OutputWrite(ctx, h.Repository, name, reader, size)
 }
 
 func FormatHookOutputPath(runID, hookRunID string) string {

pkg/actions/hook_output_writer_test.go

@@ -26,15 +26,21 @@ func TestHookWriter_OutputWritePath(t *testing.T) {
 	runID := hooks.NewRunID()
 	hookRunID := hooks.NewRunID()
 	writer := mock.NewMockOutputWriter(ctrl)
-	writer.EXPECT().OutputWrite(ctx, storageNamespace, actions.FormatHookOutputPath(runID, hookRunID), contentReader, int64(len(content))).Return(nil)
+	repositoryRecord := &graveler.RepositoryRecord{
+		RepositoryID: "someRepo",
+		Repository: &graveler.Repository{
+			StorageNamespace: storageNamespace,
+		},
+	}
+	writer.EXPECT().OutputWrite(ctx, repositoryRecord, actions.FormatHookOutputPath(runID, hookRunID), contentReader, int64(len(content))).Return(nil)
 
 	w := &actions.HookOutputWriter{
-		StorageNamespace: storageNamespace,
-		RunID:            runID,
-		HookID:           hookID,
-		HookRunID:        hookRunID,
-		ActionName:       actionName,
-		Writer:           writer,
+		Repository: repositoryRecord,
+		RunID:      runID,
+		HookID:     hookID,
+		HookRunID:  hookRunID,
+		ActionName: actionName,
+		Writer:     writer,
 	}
 	err := w.OutputWrite(ctx, contentReader, int64(len(content)))
 	if err != nil {
@@ -50,17 +56,23 @@ func TestHookWriter_OutputWriteError(t *testing.T) {
 	hooks := graveler.HooksNoOp{}
 	runID := hooks.NewRunID()
 	hookRunID := hooks.NewRunID()
+	repositoryRecord := &graveler.RepositoryRecord{
+		RepositoryID: "someRepo",
+		Repository: &graveler.Repository{
+			StorageNamespace: "storageNamespace",
+		},
+	}
 	errSomeError := errors.New("some error")
 	writer := mock.NewMockOutputWriter(ctrl)
-	writer.EXPECT().OutputWrite(ctx, "storageNamespace", actions.FormatHookOutputPath(runID, hookRunID), gomock.Any(), gomock.Any()).Return(errSomeError)
+	writer.EXPECT().OutputWrite(ctx, repositoryRecord, actions.FormatHookOutputPath(runID, hookRunID), gomock.Any(), gomock.Any()).Return(errSomeError)
 
 	w := &actions.HookOutputWriter{
-		RunID:            runID,
-		HookRunID:        hookRunID,
-		StorageNamespace: "storageNamespace",
-		ActionName:       "actionName",
-		HookID:           "hookID",
-		Writer:           writer,
+		RunID:      runID,
+		HookRunID:  hookRunID,
+		Repository: repositoryRecord,
+		ActionName: "actionName",
+		HookID:     "hookID",
+		Writer:     writer,
 	}
 	contentReader := strings.NewReader("content")
 	err := w.OutputWrite(ctx, contentReader, 10)

pkg/actions/lua.go

@@ -54,8 +54,8 @@ func applyRecord(l *lua.State, actionName, hookID string, record graveler.HookRe
 		"branch_id":         record.BranchID.String(),
 		"source_ref":        record.SourceRef.String(),
 		"tag_id":            record.TagID.String(),
-		"repository_id":     record.RepositoryID.String(),
-		"storage_namespace": record.StorageNamespace.String(),
+		"repository_id":     record.Repository.RepositoryID.String(),
+		"storage_namespace": record.Repository.StorageNamespace.String(),
 		"commit": map[string]interface{}{
 			"message":       record.Commit.Message,
 			"meta_range_id": record.Commit.MetaRangeID.String(),
@@ -125,7 +125,7 @@ func (h *LuaHook) Run(ctx context.Context, record graveler.HookRecord, buf *byte
 			return fmt.Errorf("no endpoint configured, cannot request object: %s: %w", h.ScriptPath, ErrInvalidAction)
 		}
 		reqURL, err := url.JoinPath(apiutil.BaseURL,
-			"repositories", string(record.RepositoryID), "refs", string(record.SourceRef), "objects")
+			"repositories", string(record.Repository.RepositoryID), "refs", string(record.SourceRef), "objects")
 		if err != nil {
 			return err
 		}

pkg/actions/lua_test.go

@@ -89,12 +89,17 @@ func TestLuaRun(t *testing.T) {
 		Username:  "user1",
 	})
 	err = h.Run(ctx, graveler.HookRecord{
-		RunID:            "abc123",
-		EventType:        graveler.EventTypePreCreateBranch,
-		RepositoryID:     "example123",
-		StorageNamespace: "local://foo/bar",
-		SourceRef:        "abc123",
-		BranchID:         "my-branch",
+		RunID:     "abc123",
+		EventType: graveler.EventTypePreCreateBranch,
+		Repository: &graveler.RepositoryRecord{
+			RepositoryID: "example123",
+			Repository: &graveler.Repository{
+				StorageNamespace: "local://foo/bar",
+				CreationDate:     time.Time{},
+			},
+		},
+		SourceRef: "abc123",
+		BranchID:  "my-branch",
 		Commit: graveler.Commit{
 			Version: 1,
 		},
@@ -141,12 +146,17 @@ func TestLuaRun_NetHttpDisabled(t *testing.T) {
 		Username:  "user1",
 	})
 	err = h.Run(ctx, graveler.HookRecord{
-		RunID:            "abc123",
-		EventType:        graveler.EventTypePreCreateBranch,
-		RepositoryID:     "example123",
-		StorageNamespace: "local://foo/bar",
-		SourceRef:        "abc123",
-		BranchID:         "my-branch",
+		RunID:     "abc123",
+		EventType: graveler.EventTypePreCreateBranch,
+		Repository: &graveler.RepositoryRecord{
+			RepositoryID: "example123",
+			Repository: &graveler.Repository{
+				StorageNamespace: "local://foo/bar",
+				CreationDate:     time.Time{},
+			},
+		},
+		SourceRef: "abc123",
+		BranchID:  "my-branch",
 		Commit: graveler.Commit{
 			Version: 1,
 		},
@@ -291,12 +301,17 @@ print(code .. " " .. body .. " " .. status)
 			})
 			runID := nanoid.Must(20)
 			err = h.Run(ctx, graveler.HookRecord{
-				RunID:            runID,
-				EventType:        graveler.EventTypePreCreateBranch,
-				RepositoryID:     "example123",
-				StorageNamespace: "local://foo/bar",
-				SourceRef:        "abc123",
-				BranchID:         "my-branch",
+				RunID:     runID,
+				EventType: graveler.EventTypePreCreateBranch,
+				Repository: &graveler.RepositoryRecord{
+					RepositoryID: "example123",
+					Repository: &graveler.Repository{
+						StorageNamespace: "local://foo/bar",
+						CreationDate:     time.Time{},
+					},
+				},
+				SourceRef: "abc123",
+				BranchID:  "my-branch",
 				Commit: graveler.Commit{
 					Version: 1,
 				},
@@ -420,12 +435,17 @@ func TestLuaRunTable(t *testing.T) {
 				Username:  "user1",
 			})
 			err = h.Run(ctx, graveler.HookRecord{
-				RunID:            "abc123",
-				EventType:        graveler.EventTypePreCreateBranch,
-				RepositoryID:     "example123",
-				StorageNamespace: "local://foo/bar",
-				SourceRef:        "abc123",
-				BranchID:         "my-branch",
+				RunID:     "abc123",
+				EventType: graveler.EventTypePreCreateBranch,
+				Repository: &graveler.RepositoryRecord{
+					RepositoryID: "example123",
+					Repository: &graveler.Repository{
+						StorageNamespace: "local://foo/bar",
+						CreationDate:     time.Time{},
+					},
+				},
+				SourceRef: "abc123",
+				BranchID:  "my-branch",
 				Commit: graveler.Commit{
 					Version: 1,
 				},

pkg/actions/output_writer.go

@@ -3,8 +3,10 @@ package actions
 import (
 	"context"
 	"io"
+
+	"github.com/treeverse/lakefs/pkg/graveler"
 )
 
 type OutputWriter interface {
-	OutputWrite(ctx context.Context, storageNamespace, name string, reader io.Reader, size int64) error
+	OutputWrite(ctx context.Context, repository *graveler.RepositoryRecord, name string, reader io.Reader, size int64) error
 }

pkg/actions/service.go

@@ -12,10 +12,9 @@ import (
 	"sync"
 	"time"
 
-	"github.com/treeverse/lakefs/pkg/actions/lua/hook"
-
 	"github.com/antonmedv/expr"
 	"github.com/hashicorp/go-multierror"
+	"github.com/treeverse/lakefs/pkg/actions/lua/hook"
 	"github.com/treeverse/lakefs/pkg/auth"
 	"github.com/treeverse/lakefs/pkg/graveler"
 	"github.com/treeverse/lakefs/pkg/kv"
@@ -212,7 +211,7 @@ type TaskResultIterator interface {
 type Service interface {
 	Stop()
 	Run(ctx context.Context, record graveler.HookRecord) error
-	UpdateCommitID(ctx context.Context, repositoryID string, storageNamespace string, runID string, commitID string) error
+	UpdateCommitID(ctx context.Context, repository *graveler.RepositoryRecord, runID string, commitID string) error
 	GetRunResult(ctx context.Context, repositoryID string, runID string) (*RunResult, error)
 	GetTaskResult(ctx context.Context, repositoryID string, runID string, hookRunID string) (*TaskResult, error)
 	ListRunResults(ctx context.Context, repositoryID string, branchID, commitID string, after string) (RunResultIterator, error)
@@ -317,18 +316,18 @@ func (s *StoreService) allocateTasks(runID string, actions []*Action) ([][]*Task
 	var tasks [][]*Task
 	for actionIdx, action := range actions {
 		var actionTasks []*Task
-		for hookIdx, hook := range action.Hooks {
-			h, err := NewHook(hook, action, s.cfg, s.endpoint, s.serverAddress, s.stats)
+		for hookIdx, actionHook := range action.Hooks {
+			h, err := NewHook(actionHook, action, s.cfg, s.endpoint, s.serverAddress, s.stats)
 			if err != nil {
 				return nil, err
 			}
 			task := &Task{
 				RunID:     runID,
 				HookRunID: NewHookRunID(actionIdx, hookIdx),
 				Action:    action,
-				HookID:    hook.ID,
+				HookID:    actionHook.ID,
 				Hook:      h,
-				If:        hook.If,
+				If:        actionHook.If,
 			}
 			// append new task or chain to the last one based on the current action
 			actionTasks = append(actionTasks, task)
@@ -347,12 +346,12 @@ func (s *StoreService) runTasks(ctx context.Context, record graveler.HookRecord,
 			var actionErr error
 			for _, task := range actionTasks {
 				hookOutputWriter := &HookOutputWriter{
-					Writer:           s.Writer,
-					StorageNamespace: record.StorageNamespace.String(),
-					RunID:            task.RunID,
-					HookRunID:        task.HookRunID,
-					ActionName:       task.Action.Name,
-					HookID:           task.HookID,
+					Writer:     s.Writer,
+					Repository: record.Repository,
+					RunID:      task.RunID,
+					HookRunID:  task.HookRunID,
+					ActionName: task.Action.Name,
+					HookID:     task.HookID,
 				}
 
 				// evaluate if expression and keep error for later
@@ -431,23 +430,24 @@ func (s *StoreService) saveRunInformation(ctx context.Context, record graveler.H
 
 	manifest := buildRunManifestFromTasks(record, tasks)
 
-	err := s.saveRunManifestDB(ctx, record.RepositoryID, manifest)
+	repositoryID := record.Repository.RepositoryID
+	err := s.saveRunManifestDB(ctx, repositoryID, manifest)
 	if err != nil {
 		return fmt.Errorf("insert run information: %w", err)
 	}
 
-	return s.saveRunManifestObjectStore(ctx, manifest, record.StorageNamespace.String(), record.RunID)
+	return s.saveRunManifestObjectStore(ctx, manifest, record.Repository, record.RunID)
 }
 
-func (s *StoreService) saveRunManifestObjectStore(ctx context.Context, manifest RunManifest, storageNamespace string, runID string) error {
+func (s *StoreService) saveRunManifestObjectStore(ctx context.Context, manifest RunManifest, repository *graveler.RepositoryRecord, runID string) error {
 	manifestJSON, err := json.Marshal(manifest)
 	if err != nil {
 		return fmt.Errorf("marshal run manifest: %w", err)
 	}
 	runManifestPath := FormatRunManifestOutputPath(runID)
 	manifestReader := bytes.NewReader(manifestJSON)
 	manifestSize := int64(len(manifestJSON))
-	return s.Writer.OutputWrite(ctx, storageNamespace, runManifestPath, manifestReader, manifestSize)
+	return s.Writer.OutputWrite(ctx, repository, runManifestPath, manifestReader, manifestSize)
 }
 
 func (s *StoreService) saveRunManifestDB(ctx context.Context, repositoryID graveler.RepositoryID, manifest RunManifest) error {
@@ -495,8 +495,8 @@ func buildRunManifestFromTasks(record graveler.HookRecord, tasks [][]*Task) RunM
 }
 
 // UpdateCommitID assume record is a post event, we use the PreRunID to update the commit_id and save the run manifest again
-func (s *StoreService) UpdateCommitID(ctx context.Context, repositoryID string, storageNamespace string, runID string, commitID string) error {
-	manifest, err := s.Store.UpdateCommitID(ctx, repositoryID, runID, commitID)
+func (s *StoreService) UpdateCommitID(ctx context.Context, repository *graveler.RepositoryRecord, runID string, commitID string) error {
+	manifest, err := s.Store.UpdateCommitID(ctx, repository.RepositoryID.String(), runID, commitID)
 	if err != nil {
 		return fmt.Errorf("updating commit ID: %w", err)
 	}
@@ -505,7 +505,7 @@ func (s *StoreService) UpdateCommitID(ctx context.Context, repositoryID string,
 	}
 
 	// update manifest
-	return s.saveRunManifestObjectStore(ctx, *manifest, storageNamespace, runID)
+	return s.saveRunManifestObjectStore(ctx, *manifest, repository, runID)
 }
 
 func (s *StoreService) GetRunResult(ctx context.Context, repositoryID string, runID string) (*RunResult, error) {
@@ -530,7 +530,7 @@ func (s *StoreService) PreCommitHook(ctx context.Context, record graveler.HookRe
 
 func (s *StoreService) PostCommitHook(ctx context.Context, record graveler.HookRecord) error {
 	// update pre-commit with commit ID if needed
-	err := s.UpdateCommitID(ctx, record.RepositoryID.String(), record.StorageNamespace.String(), record.PreRunID, record.CommitID.String())
+	err := s.UpdateCommitID(ctx, record.Repository, record.PreRunID, record.CommitID.String())
 	if err != nil {
 		return err
 	}