marc/courgette
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor: simpler task tracker, remove ambiguous references to context

Browse source
This commit is contained in:
Marc 2024-08-17 00:18:53 -04:00
parent ab3030ac7b
commit a970cb1f5d
Signed by: marc
GPG key ID: 048E042F22B5DC79
4 changed files with 55 additions and 92 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
internal/commands/execute_workflow.go
View file

@ -42,13 +42,13 @@ func ExecuteWorkflow(configuration Configuration, workflowFile string) error {
return nil
}
for _, job := range taskResult.Context.Jobs {
for _, job := range taskResult.Children {
if job.Status == "success" {
logger.Info(logger.Green("Job %s: %s"), job.Id, job.Status)
logger.Info(logger.Green("Job %s: %s"), job.TaskId, job.Status)
} else if job.Status == "failed" {
logger.Error(logger.Red("Job %s: %s"), job.Id, job.Status)
logger.Error(logger.Red("Job %s: %s"), job.TaskId, job.Status)
}
}
return fmt.Errorf("Task %s failed with at least 1 error.", taskResult.Id)
return fmt.Errorf("Task %s failed with at least 1 error.", taskResult.TaskId)
}

42
internal/runner/runner.go
View file

@ -12,7 +12,6 @@ import (
type Runner struct {
Labels map[string]string
Driver ContainerDriver
Tasks map[string]*Task
Runs int
// Deferred tasks, in order their were scheduled.
deferred *DeferredTaskManager
@ -22,7 +21,6 @@ func NewRunner(driver ContainerDriver, labels map[string]string) Runner {
return Runner{
Driver: driver,
Labels: labels,
Tasks: map[string]*Task{},
deferred: NewDeferredTaskManager(),
}
}
@ -41,27 +39,14 @@ func (r *Runner) GetContainerName(jobId string) string {
return fmt.Sprintf("runner-%s", jobId)
}
func (r *Runner) AddTask() string {
task := NewTask()
r.Tasks[task.Id] = &task
return task.Id
}
func (r *Runner) GetTask(taskId string) *Task {
task, _ := r.Tasks[taskId]
return task
}
// Executes a workflow using the runner.
//
// This is the high-level call that will set up the container
// that the jobs will be executed in, run the jobs's steps and
// tear down the container once no longer useful.
func (r *Runner) RunWorkflow(workflow workflow.Workflow) Task {
func (r *Runner) RunWorkflow(workflow workflow.Workflow) TaskTracker {
logger.Info("Executing workflow: %s", workflow.SourcePath)
task := r.GetTask(r.AddTask())
rootTracker := NewTaskTracker(nil)
workflowContext := context.WithValue(context.Background(), "workflow", workflow)
@ -75,28 +60,24 @@ func (r *Runner) RunWorkflow(workflow workflow.Workflow) Task {
runJob := func(context context.Context) {
defer groupWait.Done()
// FIXME: Disambiguate the usage of "context" as a term.
jobContext := task.GetJobContext(task.AddJob())
jobContext.SetStatus("started")
jobTracker := NewTaskTracker(rootTracker).SetStatus("started")
runnerImage := r.GetImageUriByLabel(job.RunsOn)
containerName := r.GetContainerName(jobContext.Id)
containerName := r.GetContainerName(jobTracker.TaskId)
logger.Info("Using image %s (label: %s)", runnerImage, job.RunsOn)
if pullError := r.Driver.Pull(runnerImage); pullError != nil {
jobContext.SetStatus("failed").SetError(pullError)
jobTracker.SetStatus("failed").SetError(pullError)
return
}
if runError := r.RunJobInContainer(runnerImage, containerName, context); runError != nil {
jobContext.SetStatus("failed").SetError(runError)
jobTracker.SetStatus("failed").SetError(runError)
return
}
jobContext.SetStatus("success")
jobTracker.SetStatus("success")
r.deferred.RunDeferredTasksInScope(fmt.Sprintf("job-%s", containerName))
}
@ -108,7 +89,7 @@ func (r *Runner) RunWorkflow(workflow workflow.Workflow) Task {
r.deferred.RunAllDeferredTasks()
return *task
return *rootTracker
}
// Executes a command within the given container.
@ -146,9 +127,14 @@ func (r *Runner) RunJobInContainer(imageUri string, containerId string, jobConte
for stepIndex, step := range job.Steps {
logger.Info("Run: %s", step.Run)
logger.Info("Using working directory %s", jobContext.Value("workflow").(workflow.Workflow).GetWorkingDirectory(job.Name, stepIndex))
var stepError error
if step.Run != "" {
return r.RunCommandInContainer(containerId, step.Run)
stepError = r.RunCommandInContainer(containerId, step.Run)
}
if stepError != nil {
return stepError
}
}

85
internal/runner/task.go
View file

@ -8,70 +8,47 @@ import (
"github.com/gofrs/uuid"
)
type JobContext struct {
Id string
Status string
Error error
type TaskTracker struct {
TaskId string
Status string
Error error
Parent *TaskTracker
Children []*TaskTracker
}
type TaskContext struct {
Jobs map[string]*JobContext
}
func NewTaskTracker(parent *TaskTracker) *TaskTracker {
newTask := &TaskTracker{TaskId: uuid.Must(uuid.NewV1()).String()}
type Task struct {
Id string
Context TaskContext
}
func NewTask() Task {
return Task{
Id: uuid.Must(uuid.NewV1()).String(),
Context: TaskContext{
Jobs: map[string]*JobContext{},
},
}
}
func (t *Task) AddJob() string {
jobContext := NewJobContext()
t.Context.Jobs[jobContext.Id] = &jobContext
return jobContext.Id
}
func (t *Task) GetJobContext(jobId string) *JobContext {
ctx, exists := t.Context.Jobs[jobId]
if exists {
return ctx
if parent != nil {
parent.Children = append(parent.Children, newTask)
newTask.Parent = parent
}
return nil
return newTask
}
func (t Task) HasError() bool {
for _, job := range t.Context.Jobs {
if job.Error != nil {
func (t *TaskTracker) SetStatus(status string) *TaskTracker {
t.Status = status
return t
}
func (t *TaskTracker) SetError(err error) *TaskTracker {
t.Error = err
return t
}
func (t TaskTracker) HasError() bool {
if t.Error != nil {
return true
}
for _, child := range t.Children {
if child.HasError() {
return true
}
}
return false
}
func NewJobContext() JobContext {
return JobContext{
Id: uuid.Must(uuid.NewV1()).String(),
}
}
func (c *JobContext) SetStatus(newStatus string) *JobContext {
c.Status = newStatus
return c
}
func (c *JobContext) SetError(err error) *JobContext {
c.Error = err
return c
}

12
internal/runner/task_test.go
View file

@ -6,8 +6,8 @@ import (
)
func TestTaskHasErrorReturnsFalseIfNoUnderlyingJobHaveErrors(t *testing.T) {
task := NewTask()
task.AddJob()
task := NewTaskTracker(nil)
NewTaskTracker(task)
if task.HasError() {
t.Errorf("Expected false, got true.")
@ -15,13 +15,13 @@ func TestTaskHasErrorReturnsFalseIfNoUnderlyingJobHaveErrors(t *testing.T) {
}
func TestTaskHasErrorReturnsTrueIfAnyJobHasErrors(t *testing.T) {
task := NewTask()
task := NewTaskTracker(nil)
// Two jobs, one of which has an error.
task.AddJob()
jobId := task.AddJob()
NewTaskTracker(task)
subTask := NewTaskTracker(task)
task.GetJobContext(jobId).SetError(errors.New("test"))
subTask.SetError(errors.New("test"))
if !task.HasError() {
t.Errorf("Expected true, got false.")