Fixed scheduling ID of pipelines

scheduler/scheduler.go

@@ -7,6 +7,7 @@ import (
 	"os/exec"
 	"path/filepath"
 	"strconv"
+	"sync"
 	"time"
 
 	"github.com/gaia-pipeline/gaia"
@@ -243,9 +244,21 @@ func (s *Scheduler) schedule() {
 	}
 }
 
+var schedulerLock = sync.RWMutex{}
+
 // SchedulePipeline schedules a pipeline. We create a new schedule object
 // and save it in our store. The scheduler will later pick this up and will continue the work.
 func (s *Scheduler) SchedulePipeline(p *gaia.Pipeline, args []gaia.Argument) (*gaia.PipelineRun, error) {
+
+	// Introduce a semaphore locking here because this function can be called
+	// in parallel if multiple users happen to trigger a pipeline run at the same time.
+	// (or someone is just simply eager and presses (Start Pipeline) in quick successions).
+	// This means that one of the calls will take slightly longer (a couple of nanoseconds)
+	// while the other finishes to save the pipelinerun.
+	// This is to ensure that the highest ID for the next pipeline is calculated properly.
+	schedulerLock.Lock()
+	defer schedulerLock.Unlock()
+
 	// Get highest public id used for this pipeline
 	highestID, err := s.storeService.PipelineGetRunHighestID(p)
 	if err != nil {

scheduler/scheduler_test.go

@@ -7,6 +7,7 @@ import (
 	"io/ioutil"
 	"os/exec"
 	"path/filepath"
+	"sync"
 	"testing"
 
 	"github.com/gaia-pipeline/gaia"
@@ -260,6 +261,59 @@ func TestSchedulePipeline(t *testing.T) {
 	}
 }
 
+func TestSchedulePipelineParallel(t *testing.T) {
+	gaia.Cfg = &gaia.Config{}
+	storeInstance := store.NewBoltStore()
+	tmp, _ := ioutil.TempDir("", "TestSchedulePipeline")
+	gaia.Cfg.DataPath = tmp
+	gaia.Cfg.WorkspacePath = filepath.Join(tmp, "tmp")
+	gaia.Cfg.Bolt.Mode = 0600
+	gaia.Cfg.Logger = hclog.New(&hclog.LoggerOptions{
+		Level:  hclog.Trace,
+		Output: hclog.DefaultOutput,
+		Name:   "Gaia",
+	})
+	gaia.Cfg.Worker = "2"
+	if err := storeInstance.Init(); err != nil {
+		t.Fatal(err)
+	}
+	p1 := gaia.Pipeline{
+		ID:   0,
+		Name: "Test Pipeline 1",
+		Type: gaia.PTypeGolang,
+		Jobs: prepareJobs(),
+	}
+	p2 := gaia.Pipeline{
+		ID:   1,
+		Name: "Test Pipeline 2",
+		Type: gaia.PTypeGolang,
+		Jobs: prepareJobs(),
+	}
+	storeInstance.PipelinePut(&p1)
+	storeInstance.PipelinePut(&p2)
+	s := NewScheduler(storeInstance, &PluginFakeFailed{}, &CAFake{}, &VaultFake{})
+	err := s.Init()
+	if err != nil {
+		t.Fatal(err)
+	}
+	var run1 *gaia.PipelineRun
+	var run2 *gaia.PipelineRun
+	var wg sync.WaitGroup
+	wg.Add(2)
+	go func() {
+		run1, _ = s.SchedulePipeline(&p1, prepareArgs())
+		wg.Done()
+	}()
+	go func() {
+		run2, _ = s.SchedulePipeline(&p2, prepareArgs())
+		wg.Done()
+	}()
+	wg.Wait()
+	if run1.ID == run2.ID {
+		t.Fatal("the two run jobs id should not have equalled. was: ", run1.ID, run2.ID)
+	}
+}
+
 func TestSchedule(t *testing.T) {
 	gaia.Cfg = &gaia.Config{}
 	storeInstance := store.NewBoltStore()