Implemented pipeline integrity check during pipeline creation process (#96)

* Implemented pipeline integrity check during pipeline creation process

* Fixed tests. Pipelines will be now removed if integrity check fails

* Added one more test

Author: michelvocks

frontend/client/views/overview/index.vue

@@ -1,7 +1,7 @@
 <template>
   <div class="columns is-multiline">
     <template v-for="(pipeline, index) in pipelines">
-      <div class="column is-one-third" :key="index">
+      <div class="column is-one-third" :key="index" v-if="!pipeline.p.notvalid">
         <div class="pipeline-box notification content-article">
           <div class="status-display-success" v-if="pipeline.r.status === 'success'"></div>
           <div class="status-display-fail" v-else-if="pipeline.r.status === 'failed'"></div>

frontend/client/views/pipeline/create.vue

@@ -369,11 +369,16 @@ export default {
                   this.historyRows[i].output += 'Starting build process. This may take some time...\n'
                 }
 
-                if (this.historyRows[i].status >= 75) {
+                if (this.historyRows[i].status >= 50) {
                   this.historyRows[i].output += 'Pipeline has been successfully compiled.\n'
                   this.historyRows[i].output += 'Copy binary to pipelines folder...\n'
                 }
 
+                if (this.historyRows[i].status >= 75) {
+                  this.historyRows[i].output += 'Pipeline binary has been copied to pipelines folder.\n'
+                  this.historyRows[i].output += 'Starting integrity check...\n'
+                }
+
                 if (this.historyRows[i].status === 100) {
                   this.historyRows[i].output += 'Copy was successful.\n'
                   this.historyRows[i].output += 'Finished.\n'

gaia.go

@@ -101,15 +101,16 @@ type User struct {
 
 // Pipeline represents a single pipeline
 type Pipeline struct {
-	ID        int          `json:"id,omitempty"`
-	Name      string       `json:"name,omitempty"`
-	Repo      GitRepo      `json:"repo,omitempty"`
-	Type      PipelineType `json:"type,omitempty"`
-	ExecPath  string       `json:"execpath,omitempty"`
-	SHA256Sum []byte       `json:"sha256sum,omitempty"`
-	Jobs      []Job        `json:"jobs,omitempty"`
-	Created   time.Time    `json:"created,omitempty"`
-	UUID      string       `json:"uuid,omitempty"`
+	ID         int          `json:"id,omitempty"`
+	Name       string       `json:"name,omitempty"`
+	Repo       GitRepo      `json:"repo,omitempty"`
+	Type       PipelineType `json:"type,omitempty"`
+	ExecPath   string       `json:"execpath,omitempty"`
+	SHA256Sum  []byte       `json:"sha256sum,omitempty"`
+	Jobs       []Job        `json:"jobs,omitempty"`
+	Created    time.Time    `json:"created,omitempty"`
+	UUID       string       `json:"uuid,omitempty"`
+	IsNotValid bool         `json:"notvalid,omitempty"`
 }
 
 // GitRepo represents a single git repository

pipeline/create_pipeline.go

@@ -2,6 +2,7 @@ package pipeline
 
 import (
 	"fmt"
+	"path/filepath"
 
 	"github.com/gaia-pipeline/gaia"
 	"github.com/gaia-pipeline/gaia/services"
@@ -12,7 +13,10 @@ const (
 	pipelineCloneStatus = 25
 
 	// Percent of pipeline creation progress after compile process done
-	pipelineCompileStatus = 75
+	pipelineCompileStatus = 50
+
+	// Percent of pipeline creation progress after validation binary copy
+	pipelineCopyStatus = 75
 
 	// Completed percent progress
 	pipelineCompleteStatus = 100
@@ -86,6 +90,39 @@ func CreatePipeline(p *gaia.CreatePipeline) {
 		return
 	}
 
+	// Update status of our pipeline build
+	p.Status = pipelineCopyStatus
+	err = storeService.CreatePipelinePut(p)
+	if err != nil {
+		gaia.Cfg.Logger.Error("cannot put create pipeline into store", "error", err.Error())
+		return
+	}
+
+	// Run update if needed
+	p.Pipeline.ExecPath = filepath.Join(gaia.Cfg.PipelinePath, appendTypeToName(p.Pipeline.Name, p.Pipeline.Type))
+	err = updatePipeline(&p.Pipeline)
+	if err != nil {
+		p.StatusType = gaia.CreatePipelineFailed
+		p.Output = fmt.Sprintf("cannot update pipeline: %s", err.Error())
+		storeService.CreatePipelinePut(p)
+
+		// Creation failed. Remove broken pipeline.
+		DeleteBinary(p.Pipeline)
+		return
+	}
+
+	// Try to get pipeline jobs to check if this pipeline is valid.
+	schedulerService, _ := services.SchedulerService()
+	if err = schedulerService.SetPipelineJobs(&p.Pipeline); err != nil {
+		p.StatusType = gaia.CreatePipelineFailed
+		p.Output = fmt.Sprintf("cannot validate pipeline: %s", err.Error())
+		storeService.CreatePipelinePut(p)
+
+		// Creation failed. Remove broken pipeline.
+		DeleteBinary(p.Pipeline)
+		return
+	}
+
 	// Save the generated pipeline data
 	err = bP.SavePipeline(&p.Pipeline)
 	if err != nil {

pipeline/create_pipeline_test.go

@@ -4,13 +4,12 @@ import (
 	"bytes"
 	"errors"
 	"io/ioutil"
+	"strings"
 	"testing"
 
+	"github.com/gaia-pipeline/gaia"
 	"github.com/gaia-pipeline/gaia/services"
-
 	"github.com/gaia-pipeline/gaia/store"
-
-	"github.com/gaia-pipeline/gaia"
 	"github.com/hashicorp/go-hclog"
 )
 
@@ -28,6 +27,16 @@ func (mcp *mockCreatePipelineStore) PipelinePut(p *gaia.Pipeline) error {
 	return mcp.Error
 }
 
+type mockScheduler struct {
+	Error error
+}
+
+func (ms *mockScheduler) Init() error { return nil }
+func (ms *mockScheduler) SchedulePipeline(p *gaia.Pipeline, args []gaia.Argument) (*gaia.PipelineRun, error) {
+	return nil, nil
+}
+func (ms *mockScheduler) SetPipelineJobs(p *gaia.Pipeline) error { return ms.Error }
+
 func TestCreatePipelineUnknownType(t *testing.T) {
 	tmp, _ := ioutil.TempDir("", "TestCreatePipelineUnknownType")
 	gaia.Cfg = new(gaia.Config)
@@ -101,6 +110,7 @@ func TestCreatePipeline(t *testing.T) {
 	tmp, _ := ioutil.TempDir("", "TestCreatePipeline")
 	gaia.Cfg = new(gaia.Config)
 	gaia.Cfg.HomePath = tmp
+	gaia.Cfg.PipelinePath = tmp
 	buf := new(bytes.Buffer)
 	gaia.Cfg.Logger = hclog.New(&hclog.LoggerOptions{
 		Level:  hclog.Trace,
@@ -110,11 +120,43 @@ func TestCreatePipeline(t *testing.T) {
 	mcp := new(mockCreatePipelineStore)
 	services.MockStorageService(mcp)
 	defer func() { services.MockStorageService(nil) }()
+	ms := new(mockScheduler)
+	services.MockSchedulerService(ms)
+	defer func() { services.MockSchedulerService(nil) }()
 	cp := new(gaia.CreatePipeline)
+	cp.Pipeline.Name = "test"
 	cp.Pipeline.Type = gaia.PTypeGolang
 	cp.Pipeline.Repo.URL = "https://github.com/gaia-pipeline/pipeline-test"
 	CreatePipeline(cp)
 	if cp.StatusType != gaia.CreatePipelineSuccess {
 		t.Fatal("pipeline status was not success. was: ", cp.StatusType)
 	}
 }
+
+func TestCreatePipelineSetPipelineJobsFail(t *testing.T) {
+	tmp, _ := ioutil.TempDir("", "TestCreatePipelineSetPipelineJobsFail")
+	gaia.Cfg = new(gaia.Config)
+	gaia.Cfg.HomePath = tmp
+	gaia.Cfg.PipelinePath = tmp
+	buf := new(bytes.Buffer)
+	gaia.Cfg.Logger = hclog.New(&hclog.LoggerOptions{
+		Level:  hclog.Trace,
+		Output: buf,
+		Name:   "Gaia",
+	})
+	mcp := new(mockCreatePipelineStore)
+	services.MockStorageService(mcp)
+	defer func() { services.MockStorageService(nil) }()
+	ms := new(mockScheduler)
+	ms.Error = errors.New("error")
+	services.MockSchedulerService(ms)
+	defer func() { services.MockSchedulerService(nil) }()
+	cp := new(gaia.CreatePipeline)
+	cp.Pipeline.Name = "test"
+	cp.Pipeline.Type = gaia.PTypeGolang
+	cp.Pipeline.Repo.URL = "https://github.com/gaia-pipeline/pipeline-test"
+	CreatePipeline(cp)
+	if !strings.Contains(cp.Output, "cannot validate pipeline") {
+		t.Fatalf("error thrown should contain 'cannot validate pipeline' but its %s", cp.Output)
+	}
+}

pipeline/ticker.go

@@ -111,10 +111,10 @@ func checkActivePipelines() {
 						}
 
 						// Let us try again to start the plugin and receive all implemented jobs
-						schedulerService.SetPipelineJobs(p)
-
-						// Set new pipeline hash
-						p.SHA256Sum = checksum
+						if err = schedulerService.SetPipelineJobs(p); err != nil {
+							// Mark that this pipeline is broken.
+							p.IsNotValid = true
+						}
 
 						// Replace pipeline
 						if ok := GlobalActivePipelines.Replace(*p); !ok {
@@ -157,16 +157,19 @@ func checkActivePipelines() {
 
 			// update pipeline if needed
 			if bytes.Compare(pipeline.SHA256Sum, pipelineCheckSum) != 0 {
-				pipeline.SHA256Sum = pipelineCheckSum
 				if err = updatePipeline(pipeline); err != nil {
 					storeService.PipelinePut(pipeline)
 					gaia.Cfg.Logger.Error("cannot update pipeline", "error", err.Error(), "pipeline", pipeline)
 					continue
 				}
+				storeService.PipelinePut(pipeline)
 			}
 
 			// Let us try to start the plugin and receive all implemented jobs
-			schedulerService.SetPipelineJobs(pipeline)
+			if err = schedulerService.SetPipelineJobs(pipeline); err != nil {
+				// Mark that this pipeline is broken.
+				pipeline.IsNotValid = true
+			}
 
 			// We encountered a drop-in pipeline previously. Now is the time to save it.
 			if shouldStore {

pipeline/update_pipeline.go

@@ -52,5 +52,13 @@ func updatePipeline(p *gaia.Pipeline) error {
 			return err
 		}
 	}
+
+	// Update checksum
+	checksum, err := getSHA256Sum(p.ExecPath)
+	if err != nil {
+		return err
+	}
+	p.SHA256Sum = checksum
+
 	return nil
 }

plugin/plugin.go

@@ -2,7 +2,9 @@ package plugin
 
 import (
 	"bufio"
+	"bytes"
 	"errors"
+	"fmt"
 	"io"
 	"os"
 	"os/exec"
@@ -46,8 +48,9 @@ type Plugin struct {
 	// Log file where all output is stored.
 	logFile *os.File
 
-	// Writer used to write logs from execution to file
+	// Writer used to write logs from execution to file or buffer
 	writer *bufio.Writer
+	buffer *bytes.Buffer
 
 	// CA instance used to handle certificates
 	ca security.CAAPI
@@ -85,10 +88,14 @@ func (p *Plugin) Connect(command *exec.Cmd, logPath *string) error {
 		if err != nil {
 			return err
 		}
-	}
 
-	// Create new writer
-	p.writer = bufio.NewWriter(p.logFile)
+		// Create new writer
+		p.writer = bufio.NewWriter(p.logFile)
+	} else {
+		// If no path is provided, write output to buffer
+		p.buffer = new(bytes.Buffer)
+		p.writer = bufio.NewWriter(p.buffer)
+	}
 
 	// Create and sign a new pair of certificates for the server
 	var err error
@@ -129,7 +136,8 @@ func (p *Plugin) Connect(command *exec.Cmd, logPath *string) error {
 	// Connect via gRPC
 	gRPCClient, err := p.client.Client()
 	if err != nil {
-		return err
+		p.writer.Flush()
+		return fmt.Errorf("%s\n\n--- output ---\n%s", err.Error(), p.buffer.String())
 	}
 
 	// Request the plugin

scheduler/scheduler.go

@@ -608,7 +608,6 @@ func (s *Scheduler) SetPipelineJobs(p *gaia.Pipeline) error {
 	// Get jobs
 	jobs, err := s.getPipelineJobs(p)
 	if err != nil {
-		gaia.Cfg.Logger.Debug("cannot get jobs from pipeline", "error", err.Error(), "pipeline", p)
 		return err
 	}
 	p.Jobs = jobs