Skip to content

Commit

Permalink
fix: tolerate workflow that needs a missing job (#1595) (#1619)
Browse files Browse the repository at this point in the history
Change planner functions to return errors

This enables createStages to return `unable to build dependency graph`

Fix PlanEvent to properly report errors relating to events/workflows
  • Loading branch information
jsoref committed Feb 16, 2023
1 parent 21ea3d0 commit be4a147
Show file tree
Hide file tree
Showing 9 changed files with 238 additions and 48 deletions.
43 changes: 33 additions & 10 deletions cmd/root.go
Original file line number Diff line number Diff line change
Expand Up @@ -354,7 +354,7 @@ func newRunCommand(ctx context.Context, input *Input) func(*cobra.Command, []str
var filterPlan *model.Plan

// Determine the event name to be filtered
var filterEventName string = ""
var filterEventName string

if len(args) > 0 {
log.Debugf("Using first passed in arguments event for filtering: %s", args[0])
Expand All @@ -366,23 +366,35 @@ func newRunCommand(ctx context.Context, input *Input) func(*cobra.Command, []str
filterEventName = events[0]
}

var plannerErr error
if jobID != "" {
log.Debugf("Preparing plan with a job: %s", jobID)
filterPlan = planner.PlanJob(jobID)
filterPlan, plannerErr = planner.PlanJob(jobID)
} else if filterEventName != "" {
log.Debugf("Preparing plan for a event: %s", filterEventName)
filterPlan = planner.PlanEvent(filterEventName)
filterPlan, plannerErr = planner.PlanEvent(filterEventName)
} else {
log.Debugf("Preparing plan with all jobs")
filterPlan = planner.PlanAll()
filterPlan, plannerErr = planner.PlanAll()
}
if filterPlan == nil && plannerErr != nil {
return plannerErr
}

if list {
return printList(filterPlan)
err = printList(filterPlan)
if err != nil {
return err
}
return plannerErr
}

if graph {
return drawGraph(filterPlan)
err = drawGraph(filterPlan)
if err != nil {
return err
}
return plannerErr
}

// plan with triggered jobs
Expand Down Expand Up @@ -410,10 +422,13 @@ func newRunCommand(ctx context.Context, input *Input) func(*cobra.Command, []str
// build the plan for this run
if jobID != "" {
log.Debugf("Planning job: %s", jobID)
plan = planner.PlanJob(jobID)
plan, plannerErr = planner.PlanJob(jobID)
} else {
log.Debugf("Planning jobs for event: %s", eventName)
plan = planner.PlanEvent(eventName)
plan, plannerErr = planner.PlanEvent(eventName)
}
if plan == nil && plannerErr != nil {
return plannerErr
}

// check to see if the main branch was defined
Expand Down Expand Up @@ -501,14 +516,22 @@ func newRunCommand(ctx context.Context, input *Input) func(*cobra.Command, []str
if watch, err := cmd.Flags().GetBool("watch"); err != nil {
return err
} else if watch {
return watchAndRun(ctx, r.NewPlanExecutor(plan))
err = watchAndRun(ctx, r.NewPlanExecutor(plan))
if err != nil {
return err
}
return plannerErr
}

executor := r.NewPlanExecutor(plan).Finally(func(ctx context.Context) error {
cancel()
return nil
})
return executor(ctx)
err = executor(ctx)
if err != nil {
return err
}
return plannerErr
}
}

Expand Down
15 changes: 9 additions & 6 deletions pkg/artifacts/server_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -297,13 +297,16 @@ func runTestJobFile(ctx context.Context, t *testing.T, tjfi TestJobFileInfo) {
planner, err := model.NewWorkflowPlanner(fullWorkflowPath, true)
assert.Nil(t, err, fullWorkflowPath)

plan := planner.PlanEvent(tjfi.eventName)

err = runner.NewPlanExecutor(plan)(ctx)
if tjfi.errorMessage == "" {
assert.Nil(t, err, fullWorkflowPath)
plan, err := planner.PlanEvent(tjfi.eventName)
if err == nil {
err = runner.NewPlanExecutor(plan)(ctx)
if tjfi.errorMessage == "" {
assert.Nil(t, err, fullWorkflowPath)
} else {
assert.Error(t, err, tjfi.errorMessage)
}
} else {
assert.Error(t, err, tjfi.errorMessage)
assert.Nil(t, plan)
}

fmt.Println("::endgroup::")
Expand Down
71 changes: 53 additions & 18 deletions pkg/model/planner.go
Original file line number Diff line number Diff line change
Expand Up @@ -15,9 +15,9 @@ import (

// WorkflowPlanner contains methods for creating plans
type WorkflowPlanner interface {
PlanEvent(eventName string) *Plan
PlanJob(jobName string) *Plan
PlanAll() *Plan
PlanEvent(eventName string) (*Plan, error)
PlanJob(jobName string) (*Plan, error)
PlanAll() (*Plan, error)
GetEvents() []string
}

Expand Down Expand Up @@ -169,47 +169,76 @@ type workflowPlanner struct {
}

// PlanEvent builds a new list of runs to execute in parallel for an event name
func (wp *workflowPlanner) PlanEvent(eventName string) *Plan {
func (wp *workflowPlanner) PlanEvent(eventName string) (*Plan, error) {
plan := new(Plan)
if len(wp.workflows) == 0 {
log.Debugf("no events found for workflow: %s", eventName)
log.Debug("no workflows found by planner")
return plan, nil
}
var lastErr error

for _, w := range wp.workflows {
for _, e := range w.On() {
events := w.On()
if len(events) == 0 {
log.Debugf("no events found for workflow: %s", w.File)
continue
}

for _, e := range events {
if e == eventName {
plan.mergeStages(createStages(w, w.GetJobIDs()...))
stages, err := createStages(w, w.GetJobIDs()...)
if err != nil {
log.Warn(err)
lastErr = err
} else {
plan.mergeStages(stages)
}
}
}
}
return plan
return plan, lastErr
}

// PlanJob builds a new run to execute in parallel for a job name
func (wp *workflowPlanner) PlanJob(jobName string) *Plan {
func (wp *workflowPlanner) PlanJob(jobName string) (*Plan, error) {
plan := new(Plan)
if len(wp.workflows) == 0 {
log.Debugf("no jobs found for workflow: %s", jobName)
}
var lastErr error

for _, w := range wp.workflows {
plan.mergeStages(createStages(w, jobName))
stages, err := createStages(w, jobName)
if err != nil {
log.Warn(err)
lastErr = err
} else {
plan.mergeStages(stages)
}
}
return plan
return plan, lastErr
}

// PlanAll builds a new run to execute in parallel all
func (wp *workflowPlanner) PlanAll() *Plan {
func (wp *workflowPlanner) PlanAll() (*Plan, error) {
plan := new(Plan)
if len(wp.workflows) == 0 {
log.Debugf("no jobs found for loaded workflows")
log.Debug("no workflows found by planner")
return plan, nil
}
var lastErr error

for _, w := range wp.workflows {
plan.mergeStages(createStages(w, w.GetJobIDs()...))
stages, err := createStages(w, w.GetJobIDs()...)
if err != nil {
log.Warn(err)
lastErr = err
} else {
plan.mergeStages(stages)
}
}

return plan
return plan, lastErr
}

// GetEvents gets all the events in the workflows file
Expand Down Expand Up @@ -282,7 +311,7 @@ func (p *Plan) mergeStages(stages []*Stage) {
p.Stages = newStages
}

func createStages(w *Workflow, jobIDs ...string) []*Stage {
func createStages(w *Workflow, jobIDs ...string) ([]*Stage, error) {
// first, build a list of all the necessary jobs to run, and their dependencies
jobDependencies := make(map[string][]string)
for len(jobIDs) > 0 {
Expand All @@ -299,6 +328,8 @@ func createStages(w *Workflow, jobIDs ...string) []*Stage {
jobIDs = newJobIDs
}

var err error

// next, build an execution graph
stages := make([]*Stage, 0)
for len(jobDependencies) > 0 {
Expand All @@ -314,12 +345,16 @@ func createStages(w *Workflow, jobIDs ...string) []*Stage {
}
}
if len(stage.Runs) == 0 {
log.Fatalf("Unable to build dependency graph!")
return nil, fmt.Errorf("unable to build dependency graph for %s (%s)", w.Name, w.File)
}
stages = append(stages, stage)
}

return stages
if len(stages) == 0 && err != nil {
return nil, err
}

return stages, nil
}

// return true iff all strings in srcList exist in at least one of the stages
Expand Down
3 changes: 2 additions & 1 deletion pkg/model/workflow_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -241,7 +241,8 @@ func TestReadWorkflow_Strategy(t *testing.T) {
w, err := NewWorkflowPlanner("testdata/strategy/push.yml", true)
assert.NoError(t, err)

p := w.PlanJob("strategy-only-max-parallel")
p, err := w.PlanJob("strategy-only-max-parallel")
assert.NoError(t, err)

assert.Equal(t, len(p.Stages), 1)
assert.Equal(t, len(p.Stages[0].Runs), 1)
Expand Down
5 changes: 4 additions & 1 deletion pkg/runner/reusable_workflow.go
Original file line number Diff line number Diff line change
Expand Up @@ -74,7 +74,10 @@ func newReusableWorkflowExecutor(rc *RunContext, directory string, workflow stri
return err
}

plan := planner.PlanEvent("workflow_call")
plan, err := planner.PlanEvent("workflow_call")
if err != nil {
return err
}

runner, err := NewReusableWorkflowRunner(rc)
if err != nil {
Expand Down
Loading

0 comments on commit be4a147

Please sign in to comment.