diff --git a/server/events/command_runner_test.go b/server/events/command_runner_test.go index f4f0e59a4c..5c2c539aac 100644 --- a/server/events/command_runner_test.go +++ b/server/events/command_runner_test.go @@ -68,6 +68,7 @@ var postWorkflowHooksCommandRunner events.PostWorkflowHooksCommandRunner var cancellationTracker *mocks.MockCancellationTracker type TestConfig struct { + planRunnerWrapper func(events.ProjectCommandRunner) events.ProjectPlanCommandRunner parallelPoolSize int SilenceNoProjects bool silenceVCSStatusNoPlans bool @@ -177,6 +178,10 @@ func setup(t *testing.T, options ...func(testConfig *TestConfig)) *vcsmocks.Mock if workingDirLocker == nil { workingDirLocker = events.NewDefaultWorkingDirLocker() } + var planRunner events.ProjectPlanCommandRunner = projectCommandRunner + if testConfig.planRunnerWrapper != nil { + planRunner = testConfig.planRunnerWrapper(projectCommandRunner) + } planCommandRunner = events.NewPlanCommandRunner( testConfig.silenceVCSStatusNoPlans, testConfig.silenceVCSStatusNoProjects, @@ -186,7 +191,7 @@ func setup(t *testing.T, options ...func(testConfig *TestConfig)) *vcsmocks.Mock workingDirLocker, commitUpdater, projectCommandBuilder, - projectCommandRunner, + planRunner, cancellationTracker, dbUpdater, pullUpdater, diff --git a/server/events/instrumented_project_command_runner.go b/server/events/instrumented_project_command_runner.go index 05f1ba5ae1..f31170bc27 100644 --- a/server/events/instrumented_project_command_runner.go +++ b/server/events/instrumented_project_command_runner.go @@ -103,3 +103,9 @@ func RunAndEmitStats(ctx command.ProjectContext, execute func(ctx command.Projec return result } + +func (p *InstrumentedProjectCommandRunner) PublishDeferredPlanStatuses(projectCmds []command.ProjectContext, result command.Result, status models.CommitStatus) { + if publisher, ok := p.projectCommandRunner.(DeferredPlanStatusPublisher); ok { + publisher.PublishDeferredPlanStatuses(projectCmds, result, status) + } +} diff --git a/server/events/plan_command_runner.go b/server/events/plan_command_runner.go index a3357da35f..3dd67acf03 100644 --- a/server/events/plan_command_runner.go +++ b/server/events/plan_command_runner.go @@ -187,13 +187,14 @@ func (p *PlanCommandRunner) runAutoplan(ctx *command.Context) { result.PlansDeleted = true } - p.pullUpdater.updatePull(ctx, AutoplanCommand{}, result) - pullStatus, err := p.dbUpdater.updateDB(ctx, ctx.Pull, result.ProjectResults) if err != nil { - ctx.Log.Err("writing results: %s", err) + p.planPersistenceFailed(ctx, AutoplanCommand{}, projectCmds, result, err) + return } + p.publishPlanStatuses(projectCmds, result, models.SuccessCommitStatus) + p.pullUpdater.updatePull(ctx, AutoplanCommand{}, result) p.updateCommitStatus(ctx, pullStatus, command.Plan) p.updateCommitStatus(ctx, pullStatus, command.Apply) @@ -334,11 +335,6 @@ func (p *PlanCommandRunner) run(ctx *command.Context, cmd *CommentCommand) { result.PlansDeleted = true } - p.pullUpdater.updatePull( - ctx, - cmd, - result) - var pullStatus models.PullStatus if noProjectPullStatus != nil { pullStatus = *noProjectPullStatus @@ -348,10 +344,12 @@ func (p *PlanCommandRunner) run(ctx *command.Context, cmd *CommentCommand) { pullStatus, err = p.dbUpdater.updateDB(ctx, pull, result.ProjectResults) } if err != nil { - ctx.Log.Err("writing results: %s", err) + p.planPersistenceFailed(ctx, cmd, projectCmds, result, err) return } + p.publishPlanStatuses(projectCmds, result, models.SuccessCommitStatus) + p.pullUpdater.updatePull(ctx, cmd, result) p.updateCommitStatus(ctx, pullStatus, command.Plan) p.updateCommitStatus(ctx, pullStatus, command.Apply) @@ -604,3 +602,22 @@ func (p *PlanCommandRunner) partitionProjectCmds( func (p *PlanCommandRunner) isParallelEnabled(projectCmds []command.ProjectContext) bool { return len(projectCmds) > 0 && projectCmds[0].ParallelPlanEnabled } + +// Successful project checks are deferred until the command's results are durable. +func (p *PlanCommandRunner) publishPlanStatuses(projectCmds []command.ProjectContext, result command.Result, status models.CommitStatus) { + if publisher, ok := p.prjCmdRunner.(DeferredPlanStatusPublisher); ok { + publisher.PublishDeferredPlanStatuses(projectCmds, result, status) + } +} + +func (p *PlanCommandRunner) planPersistenceFailed(ctx *command.Context, cmd PullCommand, projectCmds []command.ProjectContext, result command.Result, err error) { + ctx.CommandHasErrors = true + result.Error = fmt.Errorf("persisting plan results: %w; restore database connectivity and run `atlantis plan` again before applying", err) + p.publishPlanStatuses(projectCmds, result, models.FailedCommitStatus) + for _, name := range []command.Name{command.Plan, command.Apply} { + if statusErr := p.commitStatusUpdater.UpdateCombined(ctx.Log, ctx.Pull.BaseRepo, ctx.Pull, models.FailedCommitStatus, name); statusErr != nil { + ctx.Log.Warn("unable to update commit status: %s", statusErr) + } + } + p.pullUpdater.updatePull(ctx, cmd, result) +} diff --git a/server/events/plan_command_runner_test.go b/server/events/plan_command_runner_test.go index 4fec6357cf..a2ed13595d 100644 --- a/server/events/plan_command_runner_test.go +++ b/server/events/plan_command_runner_test.go @@ -5,8 +5,12 @@ package events_test import ( "errors" + "strings" "testing" + "github.com/runatlantis/atlantis/server/core/db" + "github.com/runatlantis/atlantis/server/events/mocks" + "github.com/google/go-github/v88/github" . "github.com/petergtz/pegomock/v4" "github.com/runatlantis/atlantis/server/core/boltdb" @@ -1147,3 +1151,88 @@ func TestPlanCommandRunner_PendingApplyStatus(t *testing.T) { }) } } + +// A callback at the durable write boundary makes ordering deterministic without +// racing a VCS poll against the plan command. +type observingPlanDatabase struct { + db.Database + beforeWrite func() + writeErr error + persisted bool +} + +func (d *observingPlanDatabase) UpdatePullWithResults(pull models.PullRequest, results []command.ProjectResult) (models.PullStatus, error) { + d.beforeWrite() + if d.writeErr != nil { + return models.PullStatus{}, d.writeErr + } + status, err := d.Database.UpdatePullWithResults(pull, results) + d.persisted = err == nil + return status, err +} + +func TestPlanCommandRunner_PersistenceBeforePublication(t *testing.T) { + for _, auto := range []bool{false, true} { + for _, fail := range []bool{false, true} { + name := "manual" + if auto { + name = "autoplan" + } + if fail { + name += "_write_failure" + } + t.Run(name, func(t *testing.T) { + RegisterMockTestingT(t) + storage, err := boltdb.New(t.TempDir()) + Ok(t, err) + t.Cleanup(func() { storage.Close() }) + database := &observingPlanDatabase{Database: storage} + if fail { + database.writeErr = errors.New("disk full") + } + setter := mocks.NewMockJobURLSetter() + messages := mocks.NewMockJobMessageSender() + vcsClient := setup(t, func(tc *TestConfig) { + tc.database = database + tc.planRunnerWrapper = func(r events.ProjectCommandRunner) events.ProjectPlanCommandRunner { + return events.NewInstrumentedProjectCommandRunner(metricstest.NewLoggingScope(t, logging.NewNoopLogger(t), "atlantis"), &events.ProjectOutputWrapper{ProjectCommandRunner: r, JobURLSetter: setter, JobMessageSender: messages}) + } + }) + ctx := &command.Context{Log: logging.NewNoopLogger(t), Pull: testdata.Pull, HeadRepo: testdata.GithubRepo, Scope: metricstest.NewLoggingScope(t, logging.NewNoopLogger(t), "atlantis")} + if auto { + ctx.Trigger = command.AutoTrigger + } else { + ctx.Trigger = command.CommentTrigger + } + cmd := &events.CommentCommand{Name: command.Plan, RepoRelDir: "project"} + project := command.ProjectContext{CommandName: command.Plan, RepoRelDir: "project", Workspace: "default", Log: ctx.Log} + When(projectCommandBuilder.BuildPlanCommands(ctx, cmd)).ThenReturn([]command.ProjectContext{project}, nil) + When(projectCommandBuilder.BuildAutoplanCommands(ctx)).ThenReturn([]command.ProjectContext{project}, nil) + When(projectCommandRunner.Plan(Any[command.ProjectContext]())).ThenReturn(command.ProjectCommandOutput{PlanSuccess: &models.PlanSuccess{}}) + database.beforeWrite = func() { + setter.VerifyWasCalled(Never()).SetJobURLWithStatus(Any[command.ProjectContext](), Eq(command.Plan), Eq(models.SuccessCommitStatus), Any[*command.ProjectCommandOutput]()) + vcsClient.VerifyWasCalled(Never()).CreateComment(Any[logging.SimpleLogging](), Any[models.Repo](), Any[int](), Any[string](), Any[string]()) + commitUpdater.VerifyWasCalled(Never()).UpdateCombinedCount(Any[logging.SimpleLogging](), Any[models.Repo](), Any[models.PullRequest](), Eq(models.SuccessCommitStatus), Eq(command.Plan), Any[models.ProjectCounts]()) + } + When(setter.SetJobURLWithStatus(Any[command.ProjectContext](), Eq(command.Plan), Eq(models.SuccessCommitStatus), Any[*command.ProjectCommandOutput]())).Then(func([]Param) ReturnValues { + Assert(t, database.persisted, "project success must follow persistence") + return ReturnValues{nil} + }) + planCommandRunner.Run(ctx, cmd) + Equals(t, !fail, database.persisted) + if fail { + Assert(t, ctx.CommandHasErrors, "persistence failure must fail the command") + setter.VerifyWasCalledOnce().SetJobURLWithStatus(Any[command.ProjectContext](), Eq(command.Plan), Eq(models.FailedCommitStatus), Any[*command.ProjectCommandOutput]()) + setter.VerifyWasCalled(Never()).SetJobURLWithStatus(Any[command.ProjectContext](), Eq(command.Plan), Eq(models.SuccessCommitStatus), Any[*command.ProjectCommandOutput]()) + for _, name := range []command.Name{command.Plan, command.Apply} { + commitUpdater.VerifyWasCalledOnce().UpdateCombined(Any[logging.SimpleLogging](), Any[models.Repo](), Any[models.PullRequest](), Eq(models.FailedCommitStatus), Eq(name)) + } + _, _, _, comment, _ := vcsClient.VerifyWasCalledOnce().CreateComment(Any[logging.SimpleLogging](), Any[models.Repo](), Any[int](), Any[string](), Any[string]()).GetCapturedArguments() + Assert(t, strings.Contains(comment, "disk full") && strings.Contains(comment, "atlantis plan"), "expected actionable persistence error: %s", comment) + } else { + setter.VerifyWasCalledOnce().SetJobURLWithStatus(Any[command.ProjectContext](), Eq(command.Plan), Eq(models.SuccessCommitStatus), Any[*command.ProjectCommandOutput]()) + } + }) + } + } +} diff --git a/server/events/project_command_runner.go b/server/events/project_command_runner.go index 00e7cc7c04..7ed30a0ce2 100644 --- a/server/events/project_command_runner.go +++ b/server/events/project_command_runner.go @@ -162,6 +162,10 @@ type JobURLSetter interface { SetJobURLWithStatus(ctx command.ProjectContext, cmdName command.Name, status models.CommitStatus, res *command.ProjectCommandOutput) error } +type DeferredPlanStatusPublisher interface { + PublishDeferredPlanStatuses([]command.ProjectContext, command.Result, models.CommitStatus) +} + type DeferredApplyStatusPublisher interface { PublishDeferredApplyStatuses(projectCmds []command.ProjectContext, result command.Result, status models.CommitStatus) } @@ -221,7 +225,7 @@ func (p *ProjectOutputWrapper) updateProjectPRStatus(commandName command.Name, c return result } - if commandName == command.Apply { + if commandName == command.Apply || (commandName == command.Plan && !ctx.API) { return result } @@ -1223,3 +1227,24 @@ func getMissingPolicySetNames(policySets []valid.PolicySet, receivedCount int) [ func requiresManagedPlanFileForApply(ctx command.ProjectContext) bool { return ctx.RequiresAtlantisManagedPlanFile || hasAtlantisManagedApplyStep(ctx.Steps) } + +func (p *ProjectOutputWrapper) PublishDeferredPlanStatuses(projectCmds []command.ProjectContext, result command.Result, status models.CommitStatus) { + for _, res := range result.ProjectResults { + if res.Command != command.Plan || res.PlanSuccess == nil || res.Error != nil || res.Failure != "" { + continue + } + for _, ctx := range projectCmds { + if ctx.CommandName != command.Plan || ctx.RepoRelDir != res.RepoRelDir || ctx.Workspace != res.Workspace || ctx.ProjectName != res.ProjectName || ctx.SuppressVCSStatus { + continue + } + output := res.ProjectCommandOutput + if result.Error != nil { + output = command.ProjectCommandOutput{Error: result.Error} + } + if err := p.JobURLSetter.SetJobURLWithStatus(ctx, command.Plan, status, &output); err != nil { + ctx.Log.Err("updating project PR status: %s", err) + } + break + } + } +} diff --git a/server/events/project_command_runner_test.go b/server/events/project_command_runner_test.go index 0ea555abd1..39aab7f995 100644 --- a/server/events/project_command_runner_test.go +++ b/server/events/project_command_runner_test.go @@ -356,7 +356,7 @@ func TestProjectOutputWrapper(t *testing.T) { } mockJobURLSetter.VerifyWasCalled(Once()).SetJobURLWithStatus(ctx, c.CommandName, models.PendingCommitStatus, nil) - if c.CommandName == command.Apply && c.Success { + if c.Success { mockJobURLSetter.VerifyWasCalled(Never()).SetJobURLWithStatus(ctx, c.CommandName, models.SuccessCommitStatus, &prjResult) } else { mockJobURLSetter.VerifyWasCalled(Once()).SetJobURLWithStatus(ctx, c.CommandName, expCommitStatus, &prjResult)