diff --git a/core/git/git.go b/core/git/git.go index e18c8482..596fd03c 100644 --- a/core/git/git.go +++ b/core/git/git.go @@ -89,6 +89,40 @@ func New(directory string, logger *zap.Logger) Interface { } } +// RestoreWorktree discards tracked, untracked, and ignored changes from a Git +// worktree and restores all initialized submodules to their recorded commits. +func RestoreWorktree(ctx context.Context, directory string) error { + return restoreWorktree(ctx, directory, &osExecRunner{}) +} + +func restoreWorktree(ctx context.Context, directory string, runner commandRunner) error { + ctx, cancel := context.WithTimeout(ctx, _gitTimeout) + defer cancel() + + statusArgs := []string{"status", "--porcelain", "--untracked-files=all", "--ignored", "--ignore-submodules=none"} + status, err := runner.output(ctx, directory, "git", statusArgs...) + if err != nil { + return wrapError(ctx, statusArgs, err) + } + if len(bytes.TrimSpace(status)) == 0 { + return nil + } + + commands := [][]string{ + {"reset", "--hard", "HEAD"}, + {"clean", "-ffdx"}, + {"submodule", "foreach", "--recursive", "git", "reset", "--hard", "HEAD"}, + {"submodule", "foreach", "--recursive", "git", "clean", "-ffdx"}, + {"submodule", "update", "--init", "--recursive", "--force"}, + } + for _, args := range commands { + if err := runner.run(ctx, directory, "git", args...); err != nil { + return wrapError(ctx, args, err) + } + } + return nil +} + // wrapError wraps a non-nil err with the failing git command's arguments, // additionally wrapping ErrFatal if the command exited with a fatal (128) or // usage (129) exit code, as opposed to a non-fatal, conditional exit code diff --git a/core/git/git_test.go b/core/git/git_test.go index a4105bef..d26b1fd2 100644 --- a/core/git/git_test.go +++ b/core/git/git_test.go @@ -220,6 +220,67 @@ func TestSubmoduleUpdate_usesRunnerWithDirAndArgs(t *testing.T) { assert.EqualValues(t, []string{"submodule", "update", "--init", "--recursive"}, c.args) } +func TestRestoreWorktree_usesDestructiveCleanupSequence(t *testing.T) { + m := &mockRunner{out: []byte(" M tracked.txt\n")} + require.NoError(t, restoreWorktree(context.Background(), "/repo", m)) + + require.Len(t, m.calls, 6) + assert.Equal(t, "output", m.calls[0].kind) + assert.EqualValues(t, []string{"status", "--porcelain", "--untracked-files=all", "--ignored", "--ignore-submodules=none"}, m.calls[0].args) + assert.EqualValues(t, []string{"reset", "--hard", "HEAD"}, m.calls[1].args) + assert.EqualValues(t, []string{"clean", "-ffdx"}, m.calls[2].args) + assert.EqualValues(t, []string{"submodule", "foreach", "--recursive", "git", "reset", "--hard", "HEAD"}, m.calls[3].args) + assert.EqualValues(t, []string{"submodule", "foreach", "--recursive", "git", "clean", "-ffdx"}, m.calls[4].args) + assert.EqualValues(t, []string{"submodule", "update", "--init", "--recursive", "--force"}, m.calls[5].args) + for _, call := range m.calls[1:] { + assert.Equal(t, "run", call.kind) + assert.Equal(t, "/repo", call.dir) + assert.Equal(t, "git", call.name) + } +} + +func TestRestoreWorktree_skipsCleanupWhenAlreadyClean(t *testing.T) { + m := &mockRunner{} + require.NoError(t, restoreWorktree(context.Background(), "/repo", m)) + + require.Len(t, m.calls, 1) + assert.Equal(t, "output", m.calls[0].kind) + assert.Equal(t, "/repo", m.calls[0].dir) + assert.Equal(t, "git", m.calls[0].name) + assert.EqualValues(t, []string{"status", "--porcelain", "--untracked-files=all", "--ignored", "--ignore-submodules=none"}, m.calls[0].args) +} + +func TestRestoreWorktree_removesTrackedAndUntrackedChanges(t *testing.T) { + repoDir := t.TempDir() + runGit(t, repoDir, "init") + runGit(t, repoDir, "config", "user.email", "test@example.com") + runGit(t, repoDir, "config", "user.name", "Test User") + runGit(t, repoDir, "config", "commit.gpgsign", "false") + require.NoError(t, os.WriteFile(filepath.Join(repoDir, ".gitignore"), []byte("ignored.txt\n"), 0o644)) + require.NoError(t, os.WriteFile(filepath.Join(repoDir, "tracked.txt"), []byte("original\n"), 0o644)) + runGit(t, repoDir, "add", ".gitignore", "tracked.txt") + runGit(t, repoDir, "commit", "-m", "initial") + + require.NoError(t, os.WriteFile(filepath.Join(repoDir, "tracked.txt"), []byte("dirty\n"), 0o644)) + require.NoError(t, os.WriteFile(filepath.Join(repoDir, "untracked.txt"), []byte("untracked\n"), 0o644)) + require.NoError(t, os.WriteFile(filepath.Join(repoDir, "ignored.txt"), []byte("ignored\n"), 0o644)) + runGit(t, repoDir, "add", "tracked.txt") + + require.NoError(t, RestoreWorktree(context.Background(), repoDir)) + + content, err := os.ReadFile(filepath.Join(repoDir, "tracked.txt")) + require.NoError(t, err) + assert.Equal(t, "original\n", string(content)) + assert.NoFileExists(t, filepath.Join(repoDir, "untracked.txt")) + assert.NoFileExists(t, filepath.Join(repoDir, "ignored.txt")) + + cmd := exec.Command("git", "status", "--porcelain", "--untracked-files=all") + cmd.Dir = repoDir + output, err := cmd.Output() + require.NoError(t, err) + assert.Empty(t, output) +} + func TestDiffWithStatus_parsesNameStatusOutput(t *testing.T) { tests := []struct { name string diff --git a/core/repomanager/BUILD.bazel b/core/repomanager/BUILD.bazel index 6414cb6c..9e8e2e1c 100644 --- a/core/repomanager/BUILD.bazel +++ b/core/repomanager/BUILD.bazel @@ -24,7 +24,9 @@ go_test( srcs = ["repo_manager_test.go"], embed = [":repomanager"], deps = [ + "//core/git", "//core/git/gitmock", + "//core/workspace", "//entity", "@com_github_stretchr_testify//assert", "@com_github_stretchr_testify//require", diff --git a/core/repomanager/metrics.go b/core/repomanager/metrics.go index 60b21db7..5383f59f 100644 --- a/core/repomanager/metrics.go +++ b/core/repomanager/metrics.go @@ -25,9 +25,10 @@ import ( const ( _opLease = "lease" - _stepEnsureOrigin = "ensure_origin_duration" - _stepWaitSlot = "wait_slot_duration" - _stepCreateWorker = "create_worker_duration" + _stepEnsureOrigin = "ensure_origin_duration" + _stepWaitSlot = "wait_slot_duration" + _stepCreateWorker = "create_worker_duration" + _stepRestoreWorker = "restore_worker_duration" ) func recordStep(e *metrics.Emitter, name string, start time.Time, buckets tally.DurationBuckets) { diff --git a/core/repomanager/repo_manager.go b/core/repomanager/repo_manager.go index 2423e850..a2e26c7e 100644 --- a/core/repomanager/repo_manager.go +++ b/core/repomanager/repo_manager.go @@ -50,6 +50,7 @@ type repoManager struct { logger *zap.Logger emitter *metrics.Emitter poolSize int + restoreWorker func(context.Context, string) error mu sync.Mutex pools map[string]*workerPool @@ -109,6 +110,7 @@ func NewRepoManager(appCtx context.Context, p Params) (RepoManager, error) { logger: p.Logger, emitter: metrics.New(p.Scope).SubScope("repo_manager"), poolSize: p.PoolSize, + restoreWorker: git.RestoreWorktree, pools: make(map[string]*workerPool), appCtx: appCtx, }, nil @@ -177,13 +179,31 @@ func (r *repoManager) Lease(ctx context.Context, desc entity.BuildDescription) ( return nil, fmt.Errorf("pool for repo %s: %w", repo, waitErr) } - // Lazily create the worker clone on first use + // Restore reused workers before handing them to another request. If the + // worktree cannot be verified clean, quarantine it by recreating the clone. + var restoreErr error + if slot.created { + restoreStart := time.Now() + restoreErr = r.restoreWorker(ctx, slot.dir) + recordStep(e, _stepRestoreWorker, restoreStart, metrics.FastDurationBuckets) + if restoreErr != nil { + r.logger.Warn("failed to restore worker; recreating it", + zap.String("directory", slot.dir), + zap.Error(restoreErr)) + slot.created = false + } + } + + // Lazily create new workers and synchronously recreate quarantined workers. if !slot.created { createStart := time.Now() err := r.createWorker(ctx, pool.originDir, slot.dir) recordStep(e, _stepCreateWorker, createStart, metrics.FastDurationBuckets) if err != nil { pool.avail <- slot // return slot so others can retry + if restoreErr != nil { + return nil, fmt.Errorf("restore worker failed (%v); recreate worker: %w", restoreErr, err) + } return nil, fmt.Errorf("create worker: %w", err) } slot.created = true @@ -227,7 +247,9 @@ func (p *workerPool) ensureOrigin(ctx context.Context, g git.Interface, remote s // createWorker creates a worker by cloning the origin with --local // (fast and space-efficient). func (r *repoManager) createWorker(ctx context.Context, originDir, workerDir string) error { - os.RemoveAll(workerDir) // clean up any partial/corrupted previous state + if err := os.RemoveAll(workerDir); err != nil { + return fmt.Errorf("remove previous worker: %w", err) + } if err := os.MkdirAll(filepath.Dir(workerDir), 0o755); err != nil { return err } diff --git a/core/repomanager/repo_manager_test.go b/core/repomanager/repo_manager_test.go index 5c2819a7..f0487b70 100644 --- a/core/repomanager/repo_manager_test.go +++ b/core/repomanager/repo_manager_test.go @@ -19,25 +19,43 @@ import ( "errors" "fmt" "os" + "os/exec" "path/filepath" "testing" "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + tangogit "github.com/uber/tango/core/git" gitmock "github.com/uber/tango/core/git/gitmock" + "github.com/uber/tango/core/workspace" "github.com/uber/tango/entity" "go.uber.org/mock/gomock" "go.uber.org/zap" ) +type requestFunc func(context.Context) error + +func (f requestFunc) Apply(ctx context.Context) error { + return f(ctx) +} + func newTestRepoManager(t *testing.T, appCtx context.Context, p Params) RepoManager { t.Helper() rm, err := NewRepoManager(appCtx, p) require.NoError(t, err) + rm.(*repoManager).restoreWorker = func(context.Context, string) error { return nil } return rm } +func runRepoGit(t *testing.T, directory string, args ...string) { + t.Helper() + cmd := exec.CommandContext(t.Context(), "git", args...) + cmd.Dir = directory + output, err := cmd.CombinedOutput() + require.NoError(t, err, "git %v: %s", args, output) +} + func TestNewRepoManager_InvalidPoolSize(t *testing.T) { t.Parallel() _, err := NewRepoManager(context.Background(), Params{ @@ -117,6 +135,91 @@ func TestLease_ReusesWorker_AfterRelease(t *testing.T) { require.NoError(t, ws2.Release()) } +func TestLease_RestoresWorkerAfterFailedMaterialization(t *testing.T) { + sourceDir := filepath.Join(t.TempDir(), "source") + require.NoError(t, os.MkdirAll(sourceDir, 0o755)) + runRepoGit(t, sourceDir, "init") + runRepoGit(t, sourceDir, "config", "user.email", "test@example.com") + runRepoGit(t, sourceDir, "config", "user.name", "Test User") + runRepoGit(t, sourceDir, "config", "commit.gpgsign", "false") + require.NoError(t, os.WriteFile(filepath.Join(sourceDir, ".gitignore"), []byte("generated.txt\n"), 0o644)) + require.NoError(t, os.WriteFile(filepath.Join(sourceDir, "tracked.txt"), []byte("original\n"), 0o644)) + runRepoGit(t, sourceDir, "add", ".gitignore", "tracked.txt") + runRepoGit(t, sourceDir, "commit", "-m", "initial") + + root := t.TempDir() + rm, err := NewRepoManager(context.Background(), Params{ + Git: tangogit.New(t.TempDir(), zap.NewNop()), + Logger: zap.NewNop(), + RepoManagerClonePath: root, + PoolSize: 1, + }) + require.NoError(t, err) + + ws1, err := rm.Lease(context.Background(), entity.BuildDescription{Remote: sourceDir}) + require.NoError(t, err) + workerDir := ws1.Path() + reuseMarker := filepath.Join(workerDir, ".git", "reuse-marker") + require.NoError(t, os.WriteFile(reuseMarker, []byte("keep"), 0o644)) + + materializeErr := ws1.ApplyRequests(context.Background(), []workspace.Request{ + requestFunc(func(context.Context) error { + require.NoError(t, os.WriteFile(filepath.Join(workerDir, "tracked.txt"), []byte("dirty\n"), 0o644)) + require.NoError(t, os.WriteFile(filepath.Join(workerDir, "untracked.txt"), []byte("untracked\n"), 0o644)) + require.NoError(t, os.WriteFile(filepath.Join(workerDir, "generated.txt"), []byte("generated\n"), 0o644)) + runRepoGit(t, workerDir, "add", "tracked.txt") + return assert.AnError + }), + }) + require.ErrorIs(t, materializeErr, assert.AnError) + require.NoError(t, ws1.Release()) + + ws2, err := rm.Lease(context.Background(), entity.BuildDescription{Remote: sourceDir}) + require.NoError(t, err) + assert.Equal(t, workerDir, ws2.Path()) + assert.FileExists(t, reuseMarker, "a successfully restored worker should be reused rather than recloned") + + content, err := os.ReadFile(filepath.Join(workerDir, "tracked.txt")) + require.NoError(t, err) + assert.Equal(t, "original\n", string(content)) + assert.NoFileExists(t, filepath.Join(workerDir, "untracked.txt")) + assert.NoFileExists(t, filepath.Join(workerDir, "generated.txt")) + + cmd := exec.Command("git", "status", "--porcelain", "--untracked-files=all") + cmd.Dir = workerDir + output, err := cmd.Output() + require.NoError(t, err) + assert.Empty(t, output) + require.NoError(t, ws2.Release()) +} + +func TestLease_RecreatesWorker_WhenRestoreFails(t *testing.T) { + t.Parallel() + ctrl := gomock.NewController(t) + g := gitmock.NewMockInterface(ctrl) + + root := t.TempDir() + remote := "git@github.com:org/repo" + originDir := filepath.Join(root, "org/repo") + workerDir := filepath.Join(root, ".workers", "org/repo", "worker-1") + + g.EXPECT().Clone(gomock.Any(), remote, originDir, "-c", "gc.auto=0").Return(nil) + g.EXPECT().Clone(gomock.Any(), originDir, workerDir, "--local", "-c", "gc.auto=0").Return(nil).Times(2) + + rm := newTestRepoManager(t, context.Background(), Params{Git: g, Logger: zap.NewNop(), RepoManagerClonePath: root, PoolSize: 1}) + manager := rm.(*repoManager) + manager.restoreWorker = func(context.Context, string) error { return assert.AnError } + + ws1, err := rm.Lease(context.Background(), entity.BuildDescription{Remote: remote}) + require.NoError(t, err) + require.NoError(t, ws1.Release()) + + ws2, err := rm.Lease(context.Background(), entity.BuildDescription{Remote: remote}) + require.NoError(t, err) + assert.Equal(t, workerDir, ws2.Path()) + require.NoError(t, ws2.Release()) +} + func TestLease_CreatesMultipleWorkers(t *testing.T) { t.Parallel() ctrl := gomock.NewController(t)