Implemented md5 checksum check if set pipeline jobs functionality did not work

Author: michelvocks

gaia.go

@@ -29,12 +29,13 @@ type User struct {
 
 // Pipeline represents a single pipeline
 type Pipeline struct {
-	Name     string       `json:"name"`
-	Repo     GitRepo      `json:"repo"`
-	Type     PipelineType `json:"type"`
-	ExecPath string       `json:"execpath"`
-	Jobs     []Job        `json:"jobs"`
-	Created  time.Time    `json:"created"`
+	Name        string       `json:"name"`
+	Repo        GitRepo      `json:"repo"`
+	Type        PipelineType `json:"type"`
+	ExecPath    string       `json:"execpath"`
+	Md5Checksum []byte       `json:"md5checksum"`
+	Jobs        []Job        `json:"jobs"`
+	Created     time.Time    `json:"created"`
 }
 
 // GitRepo represents a single git repository

pipeline/pipeline.go

@@ -104,6 +104,40 @@ func (ap *ActivePipelines) Append(p gaia.Pipeline) {
 	ap.Pipelines = append(ap.Pipelines, p)
 }
 
+// Get looks up the pipeline with the given name.
+func (ap *ActivePipelines) Get(n string) *gaia.Pipeline {
+	for pipeline := range ap.Iter() {
+		if pipeline.Name == n {
+			return &pipeline
+		}
+	}
+	return nil
+}
+
+// Replace takes the given pipeline and replaces it in the ActivePipelines
+// slice. Return true when success otherwise false.
+func (ap *ActivePipelines) Replace(p gaia.Pipeline) bool {
+	ap.Lock()
+	defer ap.Unlock()
+
+	// Search for the id
+	var i = -1
+	for id, pipeline := range ap.Pipelines {
+		if pipeline.Name == p.Name {
+			i = id
+		}
+	}
+
+	// We got it?
+	if i != -1 {
+		return false
+	}
+
+	// Yes
+	ap.Pipelines[i] = p
+	return true
+}
+
 // Iter iterates over the pipelines in the concurrent slice.
 func (ap *ActivePipelines) Iter() <-chan gaia.Pipeline {
 	c := make(chan gaia.Pipeline)

pipeline/scheduler.go

@@ -8,12 +8,12 @@ import (
 // setPipelineJobs uses the plugin system to get all
 // jobs from the given pipeline.
 // This function is blocking and might take some time.
-func setPipelineJobs(p *gaia.Pipeline) {
+func setPipelineJobs(p *gaia.Pipeline) error {
 	// Create the start command for the pipeline
 	c := createPipelineCmd(p)
 	if c == nil {
 		gaia.Cfg.Logger.Debug("cannot set pipeline jobs", "error", errMissingType.Error(), "pipeline", p)
-		return
+		return errMissingType
 	}
 
 	// Create new plugin instance
@@ -22,15 +22,17 @@ func setPipelineJobs(p *gaia.Pipeline) {
 	// Connect to plugin(pipeline)
 	if err := pC.Connect(); err != nil {
 		gaia.Cfg.Logger.Debug("cannot connect to pipeline", "error", err.Error(), "pipeline", p)
-		return
+		return err
 	}
 	defer pC.Close()
 
 	// Get jobs
 	jobs, err := pC.GetJobs()
 	if err != nil {
 		gaia.Cfg.Logger.Debug("cannot get jobs from pipeline", "error", err.Error(), "pipeline", p)
-		return
+		return err
 	}
 	p.Jobs = jobs
+
+	return nil
 }

pipeline/ticker.go

@@ -1,6 +1,9 @@
 package pipeline
 
 import (
+	"bytes"
+	"crypto/md5"
+	"io"
 	"io/ioutil"
 	"os"
 	"strings"
@@ -61,6 +64,27 @@ func checkActivePipelines() {
 			// already contains it.
 			pName := getRealPipelineName(n, pType)
 			if GlobalActivePipelines.Contains(pName) {
+				// If Md5Checksum is set, we should check if pipeline has been changed.
+				p := GlobalActivePipelines.Get(pName)
+				if p != nil && p.Md5Checksum != nil {
+					// Get MD5 Checksum
+					checksum, err := getMd5Checksum(gaia.Cfg.PipelinePath + string(os.PathSeparator) + file.Name())
+					if err != nil {
+						gaia.Cfg.Logger.Debug("cannot calculate md5 checksum for pipeline", "error", err.Error(), "pipeline", p)
+						continue
+					}
+
+					// Pipeline has been changed?
+					if bytes.Compare(p.Md5Checksum, checksum) != 0 {
+						// Let us try again to start the plugin and receive all implemented jobs
+						setPipelineJobsTicker(p)
+
+						// Replace pipeline
+						GlobalActivePipelines.Replace(*p)
+					}
+				}
+
+				// Its already in the list
 				continue
 			}
 
@@ -73,7 +97,7 @@ func checkActivePipelines() {
 			}
 
 			// Let us try to start the plugin and receive all implemented jobs
-			setPipelineJobs(&p)
+			setPipelineJobsTicker(&p)
 
 			// Append new pipeline
 			GlobalActivePipelines.Append(p)
@@ -105,3 +129,36 @@ func getPipelineType(n string) (gaia.PipelineType, error) {
 func getRealPipelineName(n string, pType gaia.PipelineType) string {
 	return strings.TrimSuffix(n, typeDelimiter+pType.String())
 }
+
+func setPipelineJobsTicker(p *gaia.Pipeline) {
+	err := setPipelineJobs(p)
+	if err != nil {
+		// We were not able to get jobs from the pipeline.
+		// We set the Md5Checksum for later to try it again.
+		p.Md5Checksum, err = getMd5Checksum(p.ExecPath)
+		if err != nil {
+			gaia.Cfg.Logger.Debug("cannot calculate md5 checksum for pipeline", "error", err.Error(), "pipeline", p)
+		}
+	} else {
+		// Reset md5 checksum in case we already set it
+		p.Md5Checksum = nil
+	}
+}
+
+func getMd5Checksum(file string) ([]byte, error) {
+	// Open file
+	f, err := os.Open(file)
+	if err != nil {
+		return nil, err
+	}
+	defer f.Close()
+
+	// Create md5 obj and insert bytes
+	h := md5.New()
+	if _, err := io.Copy(h, f); err != nil {
+		return nil, err
+	}
+
+	// return md5 checksum
+	return h.Sum(nil), nil
+}