Implemented ticker which periodically checks the pipeline folder for new pipelines. New pipelines are added to a global slice

Author: michelvocks

gaia.go

@@ -11,6 +11,9 @@ import (
 type PipelineType string
 
 const (
+	// UNKNOWN plugin type
+	UNKNOWN PipelineType = "unknown"
+
 	// GOLANG plugin type
 	GOLANG PipelineType = "golang"
 )
@@ -76,3 +79,8 @@ type Config struct {
 		Mode os.FileMode
 	}
 }
+
+// String returns a pipeline type string back
+func (p PipelineType) String() string {
+	return string(p)
+}

pipeline/build_golang.go

@@ -71,7 +71,7 @@ func (b *BuildPipelineGolang) ExecuteBuild(p *gaia.CreatePipeline) error {
 	args = []string{
 		"build",
 		"-o",
-		p.Pipeline.Name,
+		appendTypeToName(p.Pipeline.Name, p.Pipeline.Type),
 	}
 
 	// Execute and wait until finish or timeout
@@ -104,8 +104,8 @@ func executeCmd(path string, args []string, env []string, dir string) ([]byte, e
 // destination folder.
 func (b *BuildPipelineGolang) CopyBinary(p *gaia.CreatePipeline) error {
 	// Define src and destination
-	src := p.Pipeline.Repo.LocalDest + string(os.PathSeparator) + p.Pipeline.Name
-	dest := gaia.Cfg.PipelinePath + string(os.PathSeparator) + p.Pipeline.Name
+	src := p.Pipeline.Repo.LocalDest + string(os.PathSeparator) + appendTypeToName(p.Pipeline.Name, p.Pipeline.Type)
+	dest := gaia.Cfg.PipelinePath + string(os.PathSeparator) + appendTypeToName(p.Pipeline.Name, p.Pipeline.Type)
 
 	return copyFileContents(src, dest)
 }

pipeline/pipeline.go

@@ -1,6 +1,9 @@
 package pipeline
 
 import (
+	"fmt"
+	"sync"
+
 	"github.com/gaia-pipeline/gaia"
 )
 
@@ -20,12 +23,30 @@ type BuildPipeline interface {
 	CopyBinary(*gaia.CreatePipeline) error
 }
 
+// ActivePipelines holds all active pipelines.
+// ActivePipelines can be safely shared between goroutines.
+type ActivePipelines struct {
+	sync.RWMutex
+
+	// All active pipelines
+	Pipelines []gaia.Pipeline
+}
+
 const (
 	// Temp folder where we store our temp files during build pipeline.
 	tmpFolder = "tmp"
 
 	// Max minutes until the build process will be interrupted and marked as failed
 	maxTimeoutMinutes = 60
+
+	// typeDelimiter defines the delimiter in the file name to define
+	// the pipeline type.
+	typeDelimiter = "_"
+)
+
+var (
+	// GlobalActivePipelines holds globally all current active pipleines.
+	GlobalActivePipelines *ActivePipelines
 )
 
 // NewBuildPipeline creates a new build pipeline for the given
@@ -43,3 +64,54 @@ func NewBuildPipeline(t gaia.PipelineType) BuildPipeline {
 
 	return bP
 }
+
+// NewActivePipelines creates a new instance of ActivePipelines
+func NewActivePipelines() *ActivePipelines {
+	ap := &ActivePipelines{
+		Pipelines: make([]gaia.Pipeline, 0),
+	}
+
+	return ap
+}
+
+// Append appends a new pipeline to ActivePipelines.
+func (ap *ActivePipelines) Append(p gaia.Pipeline) {
+	ap.Lock()
+	defer ap.Unlock()
+
+	ap.Pipelines = append(ap.Pipelines, p)
+}
+
+// Iter iterates over the pipelines in the concurrent slice.
+func (ap *ActivePipelines) Iter() <-chan gaia.Pipeline {
+	c := make(chan gaia.Pipeline)
+
+	go func() {
+		ap.Lock()
+		defer ap.Unlock()
+		for _, pipeline := range ap.Pipelines {
+			c <- pipeline
+		}
+		close(c)
+	}()
+
+	return c
+}
+
+// Contains checks if the given pipeline name has been already appended
+// to the given ActivePipelines instance.
+func (ap *ActivePipelines) Contains(n string) bool {
+	for pipeline := range ap.Iter() {
+		if pipeline.Name == n {
+			return true
+		}
+	}
+
+	return false
+}
+
+// appendTypeToName appends the type to the output binary name.
+// This allows us later to define the pipeline type by the name.
+func appendTypeToName(n string, pType gaia.PipelineType) string {
+	return fmt.Sprintf("%s%s%s", n, typeDelimiter, pType.String())
+}

pipeline/ticker.go

@@ -1,43 +1,109 @@
 package pipeline
 
 import (
+	"errors"
 	"io/ioutil"
+	"strings"
 	"time"
 
 	"github.com/gaia-pipeline/gaia"
 )
 
 const (
+	// tickerIntervalSeconds defines how often the ticker will tick.
+	// Definition in seconds.
 	tickerIntervalSeconds = 5
 )
 
+var (
+	// errMissingType is the error thrown when a pipeline is missing the type
+	// in the file name.
+	errMissingType = errors.New("couldnt find pipeline type definition")
+)
+
 // InitTicker inititates the pipeline ticker.
 // This periodic job will check for new pipelines.
 func InitTicker() {
+	// Init global active pipelines slice
+	GlobalActivePipelines = NewActivePipelines()
+
+	// Check immediately to make sure we fill the list as fast as possible.
+	checkActivePipelines()
+
 	// Create ticker
 	ticker := time.NewTicker(tickerIntervalSeconds * time.Second)
-	quit := make(chan struct{})
-
-	// Actual ticker implementation
 	go func() {
 		for {
 			select {
 			case <-ticker.C:
-				files, err := ioutil.ReadDir(gaia.Cfg.PipelinePath)
-				if err != nil {
-					gaia.Cfg.Logger.Error("cannot read pipelines folder", "error", err.Error(), "path", gaia.Cfg.PipelinePath)
-				} else {
-					// Iterate all found pipeline
-					for _, file := range files {
-						// TODO: Create for every file a pipeline object
-						// and store it in a global pipeline array
-						gaia.Cfg.Logger.Debug("pipeline found", "name", file.Name())
-					}
-				}
-			case <-quit:
-				ticker.Stop()
-				return
+				checkActivePipelines()
 			}
 		}
 	}()
 }
+
+// checkActivePipelines looks up all files in the pipeline folder.
+// Every file will be handled as an active pipeline and therefore
+// saved in the global active pipelines slice.
+func checkActivePipelines() {
+	files, err := ioutil.ReadDir(gaia.Cfg.PipelinePath)
+	if err != nil {
+		gaia.Cfg.Logger.Error("cannot read pipelines folder", "error", err.Error(), "path", gaia.Cfg.PipelinePath)
+	} else {
+		// Iterate all found pipelines
+		for _, file := range files {
+			n := strings.TrimSpace(strings.ToLower(file.Name()))
+
+			// Get pipeline type
+			pType, err := getPipelineType(n)
+			if err != nil {
+				gaia.Cfg.Logger.Debug("at least one pipeline in pipeline folder is missing the type definition")
+				gaia.Cfg.Logger.Debug("Info", "name", n)
+				gaia.Cfg.Logger.Error("error thrown", "error", err.Error())
+				continue
+			}
+
+			// Get real pipeline name and check if the global active pipelines slice
+			// already contains it.
+			pName := getRealPipelineName(n, pType)
+			if GlobalActivePipelines.Contains(pName) {
+				continue
+			}
+
+			// Create pipeline object and fill it with information
+			p := gaia.Pipeline{
+				Name:    pName,
+				Type:    pType,
+				Created: time.Now(),
+			}
+
+			// Append new pipeline
+			GlobalActivePipelines.Append(p)
+		}
+	}
+}
+
+// getPipelineType looks up for specific suffix on the given file name.
+// If found, returns the pipeline type.
+func getPipelineType(n string) (gaia.PipelineType, error) {
+	s := strings.Split(n, typeDelimiter)
+
+	// Length must be higher than one
+	if len(s) < 2 {
+		return gaia.UNKNOWN, errMissingType
+	}
+
+	// Get last element and look for type
+	t := s[len(s)-1]
+	switch t {
+	case gaia.GOLANG.String():
+		return gaia.GOLANG, nil
+	}
+
+	return gaia.UNKNOWN, errMissingType
+}
+
+// getRealPipelineName removes the suffix from the pipeline name.
+func getRealPipelineName(n string, pType gaia.PipelineType) string {
+	return strings.TrimSuffix(n, typeDelimiter+pType.String())
+}