Graveler pre-commit/merge hooks (#1407)

catalog/action.go

@@ -0,0 +1,49 @@
+package catalog
+
+import (
+	"fmt"
+	"regexp"
+)
+
+type Action struct {
+	Name        string `yaml:"name"`
+	Description string `yaml:"description"`
+	On          struct {
+		PreMerge  *ActionOn `yaml:"pre-merge"`
+		PreCommit *ActionOn `yaml:"pre-commit"`
+	} `yaml:"on"`
+	Hooks []ActionHook `yaml:"hooks"`
+}
+
+type ActionOn struct {
+	Branches []string `yaml:"branches"`
+}
+
+type ActionHook struct {
+	ID          string            `yaml:"id"`
+	Type        string            `yaml:"type"`
+	Description string            `yaml:"description"`
+	Properties  map[string]string `yaml:"properties"`
+}
+
+var reHookID = regexp.MustCompile(`^[_a-zA-Z][\-_a-zA-Z0-9]{1,255}$`)
+
+func (a *Action) Validate() error {
+	if a.On.PreMerge == nil && a.On.PreCommit == nil {
+		return fmt.Errorf("%w 'on' is required", ErrInvalidAction)
+	}
+	ids := make(map[string]struct{})
+	for i, hook := range a.Hooks {
+		if !reHookID.MatchString(hook.ID) {
+			return fmt.Errorf("hook[%d] missing ID: %w", i, ErrInvalidAction)
+		}
+		if _, found := ids[hook.ID]; found {
+			return fmt.Errorf("hook[%d] duplicate ID '%s': %w", i, hook.ID, ErrInvalidAction)
+		}
+		ids[hook.ID] = struct{}{}
+		if hook.Type != "webhook" {
+			return fmt.Errorf("hook[%d] '%s' unknown type: %w", i, hook.ID, ErrInvalidAction)
+		}
+	}
+	return nil
+}

catalog/action_test.go

@@ -0,0 +1,41 @@
+package catalog
+
+import (
+	"errors"
+	"io/ioutil"
+	"path"
+	"testing"
+
+	"gopkg.in/yaml.v3"
+)
+
+func TestAction_Validate(t *testing.T) {
+	tests := []struct {
+		name     string
+		filename string
+		wantErr  error
+	}{
+		{name: "full", filename: "action_full.yaml", wantErr: nil},
+		{name: "required", filename: "action_required.yaml", wantErr: nil},
+		{name: "duplicate id", filename: "action_duplicate_id.yaml", wantErr: ErrInvalidAction},
+		{name: "invalid id", filename: "action_invalid_id.yaml", wantErr: ErrInvalidAction},
+		{name: "invalid hook type", filename: "action_invalid_type.yaml", wantErr: ErrInvalidAction},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			actionData, err := ioutil.ReadFile(path.Join("testdata", tt.filename))
+			if err != nil {
+				t.Fatalf("Failed to load testdata %s, err=%s", tt.filename, err)
+			}
+			var act Action
+			err = yaml.Unmarshal(actionData, &act)
+			if err != nil {
+				t.Fatalf("Unmarshal action err=%s", err)
+			}
+			err = act.Validate()
+			if !errors.Is(err, tt.wantErr) {
+				t.Errorf("Validate() error = %v, wantErr %v", err, tt.wantErr)
+			}
+		})
+	}
+}

catalog/cataloger.go

@@ -96,35 +96,10 @@ type Cataloger interface {
 
 	Merge(ctx context.Context, repository, destinationBranch, sourceRef, committer, message string, metadata Metadata) (*MergeResult, error)
 
-	Hooks() *CatalogerHooks
-
 	// dump/load metadata
 	DumpCommits(ctx context.Context, repositoryID string) (string, error)
 	DumpBranches(ctx context.Context, repositoryID string) (string, error)
 	DumpTags(ctx context.Context, repositoryID string) (string, error)
 
 	io.Closer
 }
-
-type PostCommitFunc func(ctx context.Context, repo, branch string, commitLog CommitLog) error
-type PostMergeFunc func(ctx context.Context, repo, branch string, mergeResult MergeResult) error
-
-// CatalogerHooks describes the hooks available for some operations on the catalog.  Hooks are
-// called after the transaction ends; if they return an error they do not affect commit/merge.
-type CatalogerHooks struct {
-	// PostCommit hooks are called at the end of a commit.
-	PostCommit []PostCommitFunc
-
-	// PostMerge hooks are called at the end of a merge.
-	PostMerge []PostMergeFunc
-}
-
-func (h *CatalogerHooks) AddPostCommit(f PostCommitFunc) *CatalogerHooks {
-	h.PostCommit = append(h.PostCommit, f)
-	return h
-}
-
-func (h *CatalogerHooks) AddPostMerge(f PostMergeFunc) *CatalogerHooks {
-	h.PostMerge = append(h.PostMerge, f)
-	return h
-}

catalog/entry_catalog.go

@@ -128,9 +128,11 @@ func NewEntryCatalog(cfg *config.Config, db db.Database) (*EntryCatalog, error)
 	stagingManager := staging.NewManager(db)
 	refManager := ref.NewPGRefManager(db, ident.NewHexAddressProvider())
 	branchLocker := ref.NewBranchLocker(db)
-	return &EntryCatalog{
-		Store: graveler.NewGraveler(branchLocker, committedManager, stagingManager, refManager),
-	}, nil
+	store := graveler.NewGraveler(branchLocker, committedManager, stagingManager, refManager)
+	entryCatalog := &EntryCatalog{Store: store}
+	store.SetPreCommitHook(entryCatalog.preCommitHook)
+	store.SetPreMergeHook(entryCatalog.preMergeHook)
+	return entryCatalog, nil
 }
 
 func (e *EntryCatalog) AddCommitToBranchHead(ctx context.Context, repositoryID graveler.RepositoryID, branchID graveler.BranchID, commit graveler.Commit) (graveler.CommitID, error) {
@@ -501,3 +503,11 @@ func (e *EntryCatalog) DumpBranches(ctx context.Context, repositoryID graveler.R
 func (e *EntryCatalog) DumpTags(ctx context.Context, repositoryID graveler.RepositoryID) (*graveler.MetaRangeID, error) {
 	return e.Store.DumpTags(ctx, repositoryID)
 }
+
+func (e *EntryCatalog) preCommitHook(ctx context.Context, repositoryID graveler.RepositoryID, branchID graveler.BranchID, commit graveler.Commit) error {
+	return nil
+}
+
+func (e *EntryCatalog) preMergeHook(ctx context.Context, repositoryID graveler.RepositoryID, destination graveler.BranchID, source graveler.Ref, commit graveler.Commit) error {
+	return nil
+}

catalog/errors.go

@@ -8,29 +8,14 @@ import (
 )
 
 var (
-	ErrInvalidReference            = errors.New("invalid reference")
-	ErrInvalidMetadataSrcFormat    = errors.New("invalid metadata src format")
-	ErrExpired                     = errors.New("expired from storage")
-	ErrByteSliceTypeAssertion      = errors.New("type assertion to []byte failed")
-	ErrFeatureNotSupported         = errors.New("feature not supported")
-	ErrBranchNotFound              = fmt.Errorf("branch %w", db.ErrNotFound)
-	ErrCommitNotFound              = fmt.Errorf("commit %w", db.ErrNotFound)
-	ErrRepositoryNotFound          = fmt.Errorf("repository %w", db.ErrNotFound)
-	ErrEntryNotFound               = fmt.Errorf("entry %w", db.ErrNotFound)
-	ErrUnexpected                  = errors.New("unexpected error")
-	ErrReadEntryTimeout            = errors.New("read entry timeout")
-	ErrInvalidValue                = errors.New("invalid value")
-	ErrNonDirectNotSupported       = errors.New("non direct diff not supported")
-	ErrSameBranchMergeNotSupported = errors.New("same branch merge not supported")
-	ErrLineageCorrupted            = errors.New("lineage corrupted")
-	ErrOperationNotPermitted       = errors.New("operation not permitted")
-	ErrNothingToCommit             = errors.New("nothing to commit")
-	ErrInvalidLockValue            = errors.New("invalid lock value")
-	ErrNoDifferenceWasFound        = errors.New("no difference was found")
-	ErrConflictFound               = errors.New("conflict found")
-	ErrUnsupportedRelation         = errors.New("unsupported relation")
-	ErrUnsupportedDelimiter        = errors.New("unsupported delimiter")
-	ErrBadTypeConversion           = errors.New("bad type")
-	ErrExportFailed                = errors.New("export failed")
-	ErrRollbackWithActiveBranch    = fmt.Errorf("%w: rollback with active branch", ErrFeatureNotSupported)
+	ErrInvalidMetadataSrcFormat = errors.New("invalid metadata src format")
+	ErrExpired                  = errors.New("expired from storage")
+	ErrFeatureNotSupported      = errors.New("feature not supported")
+	ErrBranchNotFound           = fmt.Errorf("branch %w", db.ErrNotFound)
+	ErrRepositoryNotFound       = fmt.Errorf("repository %w", db.ErrNotFound)
+	ErrInvalidValue             = errors.New("invalid value")
+	ErrNoDifferenceWasFound     = errors.New("no difference was found")
+	ErrConflictFound            = errors.New("conflict found")
+	ErrUnsupportedRelation      = errors.New("unsupported relation")
+	ErrInvalidAction            = errors.New("invalid action")
 )

catalog/fake_graveler_test.go

@@ -16,6 +16,8 @@ type FakeGraveler struct {
 	RepositoryIteratorFactory func() graveler.RepositoryIterator
 	BranchIteratorFactory     func() graveler.BranchIterator
 	TagIteratorFactory        func() graveler.TagIterator
+	preCommitHook             graveler.PreCommitFunc
+	preMergeHook              graveler.PreMergeFunc
 }
 
 func (g *FakeGraveler) DumpCommits(ctx context.Context, repositoryID graveler.RepositoryID) (*graveler.MetaRangeID, error) {
@@ -200,6 +202,22 @@ func (g *FakeGraveler) Compare(_ context.Context, _ graveler.RepositoryID, _, _
 	return g.DiffIteratorFactory(), nil
 }
 
+func (g *FakeGraveler) PreCommitHook() graveler.PreCommitFunc {
+	return g.preCommitHook
+}
+
+func (g *FakeGraveler) SetPreCommitHook(fn graveler.PreCommitFunc) {
+	g.preCommitHook = fn
+}
+
+func (g *FakeGraveler) PreMergeHook() graveler.PreMergeFunc {
+	return g.preMergeHook
+}
+
+func (g *FakeGraveler) SetPreMergeHook(fn graveler.PreMergeFunc) {
+	g.preMergeHook = fn
+}
+
 func (g *FakeGraveler) AddCommitToBranchHead(ctx context.Context, repositoryID graveler.RepositoryID, branchID graveler.BranchID, commit graveler.Commit) (graveler.CommitID, error) {
 	panic("implement me")
 }

catalog/rocks_cataloger.go

@@ -16,7 +16,6 @@ import (
 type cataloger struct {
 	EntryCatalog *EntryCatalog
 	log          logging.Logger
-	hooks        CatalogerHooks
 }
 
 const (
@@ -37,7 +36,6 @@ func NewCataloger(db db.Database, cfg *config.Config) (Cataloger, error) {
 	return &cataloger{
 		EntryCatalog: entryCatalog,
 		log:          logging.Default(),
-		hooks:        CatalogerHooks{},
 	}, nil
 }
 
@@ -631,10 +629,6 @@ func (c *cataloger) Merge(ctx context.Context, repository string, destinationBra
 	}, nil
 }
 
-func (c *cataloger) Hooks() *CatalogerHooks {
-	return &c.hooks
-}
-
 func (c *cataloger) DumpCommits(ctx context.Context, repositoryID string) (string, error) {
 	metaRangeID, err := c.EntryCatalog.DumpCommits(ctx, graveler.RepositoryID(repositoryID))
 	if err != nil {

catalog/testdata/action_duplicate_id.yaml

@@ -0,0 +1,13 @@
+on:
+  pre-merge:
+    branches:
+      - master
+hooks:
+  - id: hook1
+    type: webhook
+    properties:
+      url: "https://api.lakefs.io/webhook1?t=1za2PbkZK1bd4prMuTDr6BeEQwWYcX2R"
+  - id: hook1
+    type: webhook
+    properties:
+      url: "https://api.lakefs.io/webhook1?t=1za2PbkZK1bd4prMuTDr6BeEQwWYcX2R"

catalog/testdata/action_full.yaml

@@ -0,0 +1,21 @@
+name: Good merge
+description: set of checks to verify that branch is good
+on:
+  pre-merge:
+    branches:
+      - master
+      - stage
+  pre-commit:
+    branches:
+      - feature-*
+hooks:
+  - id: no_temp
+    type: webhook
+    description: checking no temporary files found
+    properties:
+      url: "https://api.lakefs.io/webhook1?t=1za2PbkZK1bd4prMuTDr6BeEQwWYcX2R"
+  - id: no_freeze
+    type: webhook
+    description: check production is not in dev freeze
+    properties:
+      url: "https://api.lakefs.io/webhook2?t=1za2PbkZK1bd4prMuTDr6BeEQwWYcX2R"

catalog/testdata/action_invalid_id.yaml

@@ -0,0 +1,9 @@
+on:
+  pre-merge:
+    branches:
+      - master
+hooks:
+  - id: not valid to use space
+    type: webhook
+    properties:
+      url: "https://api.lakefs.io/webhook1?t=1za2PbkZK1bd4prMuTDr6BeEQwWYcX2R"

catalog/testdata/action_invalid_type.yaml

@@ -0,0 +1,7 @@
+on:
+  pre-merge:
+    branches:
+      - master
+hooks:
+  - id: no_temp
+    type: command

catalog/testdata/action_required.yaml

@@ -0,0 +1,9 @@
+on:
+  pre-merge:
+    branches:
+      - master
+hooks:
+  - id: no_temp
+    type: webhook
+    properties:
+      url: "https://api.lakefs.io/webhook1?t=1za2PbkZK1bd4prMuTDr6BeEQwWYcX2R"

db/errors.go

@@ -11,5 +11,4 @@ var (
 	ErrNotFound      = fmt.Errorf("not found: %w", pgx.ErrNoRows)
 	ErrAlreadyExists = errors.New("already exists")
 	ErrSerialization = errors.New("serialization error")
-	ErrNotASlice     = errors.New("results must be a pointer to a slice")
 )

go.mod

@@ -46,7 +46,6 @@ require (
 	github.com/jackc/pgconn v1.8.0
 	github.com/jackc/pgerrcode v0.0.0-20201024163028-a0d42d470451
 	github.com/jackc/pgproto3/v2 v2.0.6
-	github.com/jackc/pgtype v1.6.2 // indirect
 	github.com/jackc/pgx/v4 v4.10.1
 	github.com/jamiealquiza/tachymeter v2.0.0+incompatible
 	github.com/jedib0t/go-pretty v4.3.0+incompatible
@@ -89,5 +88,6 @@ require (
 	google.golang.org/api v0.36.0
 	google.golang.org/protobuf v1.25.0
 	gopkg.in/dgrijalva/jwt-go.v3 v3.2.0
+	gopkg.in/yaml.v3 v3.0.0-20200615113413-eeeca48fe776
 	pgregory.net/rapid v0.4.0 // indirect
 )

graveler/errors.go

@@ -43,6 +43,7 @@ var (
 	ErrAddCommitNoParent       = errors.New("added commit must have a parent")
 	ErrMultipleParents         = errors.New("cannot have more than a single parent")
 	ErrRevertParentOutOfRange  = errors.New("given commit does not have the given parent number")
+	ErrAbortedByHook           = errors.New("aborted by hook")
 )
 
 // wrappedError is an error for wrapping another error while ignoring its message.