Run pipeline

Author: boutros

project.go

@@ -4,9 +4,15 @@ import "time"
 
 // A Project represents a software project that has a number of jobs associated with it.
 type Project struct {
-	ID     uint64
-	Name   string
-	JobIDs []uint64
+	ID   uint64
+	Name string
+
+	// The pipeline is a list of jobs that will be run sequentially, stopping if any
+	// of the jobs ends with exit status other than 0.
+	Pipeline []uint64
+
+	// OneOffJobs is for jobs that is not part of the Pipeline, but can be run individually.
+	OneOffJobs []uint64
 }
 
 // A Job is a job that can be executed.

runner.go

@@ -30,7 +30,9 @@ type task struct {
 // NewRunner returns a new Runner, using the given Storer.
 func NewRunner(s Storer) *Runner {
 	return &Runner{
-		Storer: s,
+		Storer:       s,
+		runningJobs:  make(map[uint64]bool),
+		runningProjs: make(map[uint64]bool),
 	}
 }
 
@@ -49,6 +51,7 @@ func (r *Runner) ScheduleProject(id uint64) error {
 }
 
 func (r *Runner) runJob(id uint64, inProjectPipeline bool) (Run, error) {
+	// TODO why return Run struct?
 	defer r.doneJob(id, inProjectPipeline)
 	var run Run
 
@@ -62,7 +65,7 @@ func (r *Runner) runJob(id uint64, inProjectPipeline bool) (Run, error) {
 		return run, errors.Wrap(err, "cannot assign run ID")
 	}
 
-	if err := createAndChdir(job.Workspace); err != nil {
+	if err = createAndChdir(job.Workspace); err != nil {
 		return run, errors.Wrap(err, "failed to create/change workspace")
 	}
 
@@ -99,6 +102,30 @@ func (r *Runner) runJob(id uint64, inProjectPipeline bool) (Run, error) {
 	return run, nil
 }
 
+func (r *Runner) runPipe(id uint64) error {
+	proj, err := r.GetProject(id)
+	if err != nil {
+		return errors.Wrapf(err, "cannot get project %d", id)
+	}
+	defer r.donePipeline(id)
+
+	for _, jobID := range proj.Pipeline {
+		r.mu.Lock()
+		// TODO check if job is running first
+		r.runningJobs[jobID] = true
+		r.mu.Unlock()
+		run, err := r.runJob(jobID, true)
+		if err != nil {
+			return err
+		}
+
+		if !run.Success {
+			break
+		}
+	}
+	return nil
+}
+
 func (r *Runner) doneJob(id uint64, inProjectPipeline bool) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
@@ -119,12 +146,31 @@ func (r *Runner) doneJob(id uint64, inProjectPipeline bool) {
 					}
 				}()
 			} else {
-				//go r.runPipe(task.pipe)
+				go r.runPipe(task.proj)
 			}
 		}
 	}
 }
 
+func (r *Runner) donePipeline(id uint64) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	// Mark the Project as not running
+	delete(r.runningProjs, id)
+
+	// If there are enqueued jobs/pipelines  - run it
+	if len(r.scheduled) > 0 {
+		task := r.scheduled[0]
+		r.scheduled = r.scheduled[1:]
+		if task.job != 0 {
+			go r.runJob(task.job, false)
+		} else {
+			go r.runPipe(task.proj)
+		}
+	}
+}
+
 func createAndChdir(path string) error {
 	if path == "" {
 		return nil

runner_test.go

@@ -108,7 +108,7 @@ func TestRunOutputsForJob(t *testing.T) {
 
 	const numRuns = 10
 	for i := 0; i <= numRuns; i++ {
-		r.runJob(job.ID, false)
+		_, _ = r.runJob(job.ID, false)
 	}
 
 	runs, err := r.GetNRunsForJob(job.ID, numRuns)
@@ -133,5 +133,88 @@ func TestRunOutputsForJob(t *testing.T) {
 	if !sort.IntsAreSorted(times) {
 		t.Fatalf("runs not run sequentially")
 	}
+}
+
+// Verify that a Project can be created and it's Pipeline of Jobs can run.
+func TestPipeline(t *testing.T) {
+	db := newMemDB()
+	r := NewRunner(db)
+
+	job1, err := db.NewJob(Job{
+		Cmd: "echo 'hello' > goodbye.txt",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	job2, err := db.NewJob(Job{
+		Cmd: "cat goodbye.txt",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
 
+	copyOfjob2, err := db.NewJob(Job{
+		Cmd: "cat goodbye.txt",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	job3, err := db.NewJob(Job{
+		Cmd: "rm goodbye.txt",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	job4, err := db.NewJob(Job{
+		Cmd: "echo 'I should not run'",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	proj, err := db.NewProject(Project{
+		Pipeline: []uint64{job1.ID, job2.ID, job3.ID, copyOfjob2.ID, job4.ID},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if err := r.runPipe(proj.ID); err != nil {
+		t.Fatal(err)
+	}
+
+	want := []struct {
+		success bool
+		output  string
+	}{
+		{true, ""},
+		{true, "hello\n"},
+		{true, ""},
+		{false, "cat: goodbye.txt: No such file or directory\nexit status 1"},
+	}
+
+	var runs []Run
+	for _, id := range proj.Pipeline {
+		run, err := db.GetNRunsForJob(id, 100)
+		if err != nil {
+			t.Fatal(err)
+		}
+		runs = append(runs, run...)
+	}
+
+	if len(runs) != len(want) {
+		t.Fatalf("pipeline got %d runs; want %d", len(runs), len(want))
+	}
+
+	for i, run := range runs {
+		if run.Success != want[i].success {
+			t.Errorf("pipelined job success=%v; want %v", run.Success, want[i].success)
+		}
+		if run.Output != want[i].output {
+			t.Errorf("pipelined job output=%v; want %v", run.Output, want[i].output)
+		}
+	}
 }