Worker builds binary in case the binary doesn't match the worker arch

gaia.go

@@ -273,6 +273,14 @@ type Worker struct {
 	Tags         []string     `json:"tags"`
 }
 
+// SHAPair struct contains the original sha of a pipeline executable and the
+// new sha which was created when the worker had to rebuild it.
+type SHAPair struct {
+	Original   []byte `json:"original"`
+	Worker     []byte `json:"worker"`
+	PipelineID int    `json:"pipelineid"`
+}
+
 // Cfg represents the global config instance
 var Cfg = &Config{}
 

store/sha_pair.go

@@ -0,0 +1,51 @@
+package store
+
+import (
+	"encoding/json"
+
+	bolt "github.com/coreos/bbolt"
+	"github.com/gaia-pipeline/gaia"
+)
+
+// UpsertSHAPair creates or updates a record for a SHA pair of the original SHA and the
+// rebuilt Worker SHA for a pipeline.
+func (s *BoltStore) UpsertSHAPair(pair gaia.SHAPair) error {
+	return s.db.Update(func(tx *bolt.Tx) error {
+		// Get bucket
+		b := tx.Bucket(shaPairBucket)
+
+		// Marshal SHAPair struct
+		m, err := json.Marshal(pair)
+		if err != nil {
+			return err
+		}
+
+		// Put SHAPair
+		return b.Put(itob(pair.PipelineID), m)
+	})
+}
+
+// GetSHAPair returns a pair of shas for this pipeline run.
+func (s *BoltStore) GetSHAPair(pipelineID int) (ok bool, pair gaia.SHAPair, err error) {
+	return ok, pair, s.db.View(func(tx *bolt.Tx) error {
+		// Get bucket
+		b := tx.Bucket(shaPairBucket)
+
+		// Get SHAPair
+		v := b.Get(itob(pipelineID))
+
+		// Check if we found the SHAPair
+		if v == nil {
+			ok = false
+			return nil
+		}
+
+		// Unmarshal SHAPair struct
+		err := json.Unmarshal(v, &pair)
+		if err != nil {
+			return err
+		}
+		ok = true
+		return nil
+	})
+}

store/sha_pair_test.go

@@ -0,0 +1,104 @@
+package store
+
+import (
+	"bytes"
+	"io/ioutil"
+	"os"
+	"testing"
+
+	"github.com/gaia-pipeline/gaia"
+)
+
+func TestGetSHAPair(t *testing.T) {
+	// Create tmp folder
+	tmp, err := ioutil.TempDir("", "TestGetSHAPAir")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer os.RemoveAll(tmp)
+
+	store := NewBoltStore()
+	gaia.Cfg.Bolt.Mode = 0600
+	err = store.Init(tmp)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer store.Close()
+
+	pair := gaia.SHAPair{}
+	pair.PipelineID = 1
+	pair.Original = []byte("original")
+	pair.Worker = []byte("worker")
+	err = store.UpsertSHAPair(pair)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ok, p, err := store.GetSHAPair(1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !ok {
+		t.Fatalf("sha pair not found")
+	}
+
+	if p.PipelineID != pair.PipelineID {
+		t.Fatalf("pipeline id match error. want %d got %d", pair.PipelineID, p.PipelineID)
+	}
+	if !bytes.Equal(p.Worker, pair.Worker) {
+		t.Fatalf("worker sha match error. want %s got %s", pair.Worker, p.Worker)
+	}
+	if !bytes.Equal(p.Original, pair.Original) {
+		t.Fatalf("original sha match error. want %s got %s", pair.Original, p.Original)
+	}
+}
+
+func TestUpsertSHAPair(t *testing.T) {
+	// Create tmp folder
+	tmp, err := ioutil.TempDir("", "TestUpsertSHAPair")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer os.RemoveAll(tmp)
+
+	store := NewBoltStore()
+	gaia.Cfg.Bolt.Mode = 0600
+	err = store.Init(tmp)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer store.Close()
+
+	pair := gaia.SHAPair{}
+	pair.PipelineID = 1
+	pair.Original = []byte("original")
+	pair.Worker = []byte("worker")
+	err = store.UpsertSHAPair(pair)
+	if err != nil {
+		t.Fatal(err)
+	}
+	// Test is upsert overwrites existing records.
+	pair.Original = []byte("original2")
+	err = store.UpsertSHAPair(pair)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ok, p, err := store.GetSHAPair(1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !ok {
+		t.Fatalf("sha pair not found")
+	}
+
+	if p.PipelineID != pair.PipelineID {
+		t.Fatalf("pipeline id match error. want %d got %d", pair.PipelineID, p.PipelineID)
+	}
+	if !bytes.Equal(p.Worker, pair.Worker) {
+		t.Fatalf("worker sha match error. want %s got %s", pair.Worker, p.Worker)
+	}
+	if !bytes.Equal(p.Original, pair.Original) {
+		t.Fatalf("original sha match error. want %s got %s", pair.Original, p.Original)
+	}
+}

store/store.go

@@ -33,6 +33,9 @@ var (
 
 	// Name of the bucket where we store all worker.
 	workerBucket = []byte("Worker")
+
+	// SHA pair bucket.
+	shaPairBucket = []byte("SHAPair")
 )
 
 const (
@@ -85,6 +88,8 @@ type GaiaStore interface {
 	WorkerDelete(id string) error
 	WorkerDeleteAll() error
 	WorkerGet(id string) (*gaia.Worker, error)
+	UpsertSHAPair(pair gaia.SHAPair) error
+	GetSHAPair(pipelineID int) (bool, gaia.SHAPair, error)
 }
 
 // Compile time interface compliance check for BoltStore. If BoltStore
@@ -120,54 +125,48 @@ func (s *BoltStore) Close() error {
 	return s.db.Close()
 }
 
-// setupDatabase create all buckets in the db.
-// Additionally, it makes sure that the admin user exists.
-func (s *BoltStore) setupDatabase() error {
-	// Create bucket if not exists function
-	var bucketName []byte
+type setup struct {
+	bs  *BoltStore
+	err error
+}
+
+// Create bucket if not exists function
+func (s *setup) update(bucketName []byte) {
+	// update is a no-op in case there was already an error
+	if s.err != nil {
+		return
+	}
 	c := func(tx *bolt.Tx) error {
 		_, err := tx.CreateBucketIfNotExists(bucketName)
 		if err != nil {
 			return fmt.Errorf("create bucket: %s", err)
 		}
 		return nil
 	}
+	s.err = s.bs.db.Update(c)
+}
 
-	// Make sure buckets exist
-	bucketName = userBucket
-	err := s.db.Update(c)
-	if err != nil {
-		return err
-	}
-	bucketName = userPermsBucket
-	err = s.db.Update(c)
-	if err != nil {
-		return err
-	}
-	bucketName = pipelineBucket
-	err = s.db.Update(c)
-	if err != nil {
-		return err
-	}
-	bucketName = createPipelineBucket
-	err = s.db.Update(c)
-	if err != nil {
-		return err
-	}
-	bucketName = pipelineRunBucket
-	err = s.db.Update(c)
-	if err != nil {
-		return err
-	}
-	bucketName = settingsBucket
-	err = s.db.Update(c)
-	if err != nil {
-		return err
+// setupDatabase create all buckets in the db.
+// Additionally, it makes sure that the admin user exists.
+func (s *BoltStore) setupDatabase() error {
+	// Create bucket if not exists function
+	setP := &setup{
+		bs:  s,
+		err: nil,
 	}
-	bucketName = workerBucket
-	err = s.db.Update(c)
-	if err != nil {
-		return err
+
+	// Make sure buckets exist
+	setP.update(userBucket)
+	setP.update(userPermsBucket)
+	setP.update(pipelineBucket)
+	setP.update(createPipelineBucket)
+	setP.update(pipelineRunBucket)
+	setP.update(settingsBucket)
+	setP.update(workerBucket)
+	setP.update(shaPairBucket)
+
+	if setP.err != nil {
+		return setP.err
 	}
 
 	// Make sure that the user "admin" does exist
@@ -195,6 +194,9 @@ func (s *BoltStore) setupDatabase() error {
 	}
 
 	u, err := s.UserGet(autoUsername)
+	if err != nil {
+		return err
+	}
 
 	if u == nil {
 		triggerToken := security.GenerateRandomUUIDV5()