Merge pull request #34 from bidhan-a/issue-29

Remove deleted pipelines from global active pipeline list

pipeline/pipeline.go

@@ -87,6 +87,14 @@ func (ap *ActivePipelines) Append(p gaia.Pipeline) {
 	ap.Pipelines = append(ap.Pipelines, p)
 }
 
+// Remove removes a pipeline at the given index from ActivePipelines.
+func (ap *ActivePipelines) Remove(index int) {
+	ap.Lock()
+	defer ap.Unlock()
+
+	ap.Pipelines = append(ap.Pipelines[:index], ap.Pipelines[index+1:]...)
+}
+
 // GetByName looks up the pipeline by the given name.
 func (ap *ActivePipelines) GetByName(n string) *gaia.Pipeline {
 	var foundPipeline gaia.Pipeline
@@ -157,6 +165,29 @@ func (ap *ActivePipelines) Contains(n string) bool {
 	return foundPipeline
 }
 
+// RemoveDeletedPipelines removes the pipelines whose names are NOT
+// present in `existingPipelineNames` from the given ActivePipelines instance.
+func (ap *ActivePipelines) RemoveDeletedPipelines(existingPipelineNames []string) {
+	var deletedPipelineIndices []int
+	var index int
+	for pipeline := range ap.Iter() {
+		found := false
+		for _, name := range existingPipelineNames {
+			if pipeline.Name == name {
+				found = true
+				break
+			}
+		}
+		if !found {
+			deletedPipelineIndices = append(deletedPipelineIndices, index)
+		}
+		index++
+	}
+	for _, idx := range deletedPipelineIndices {
+		ap.Remove(idx)
+	}
+}
+
 // 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 {

pipeline/pipeline_test.go

@@ -0,0 +1,77 @@
+package pipeline
+
+import (
+	"testing"
+	"time"
+
+	"github.com/gaia-pipeline/gaia"
+)
+
+func TestRemove(t *testing.T) {
+	ap := NewActivePipelines()
+
+	p1 := gaia.Pipeline{
+		Name:    "Pipeline A",
+		Type:    gaia.PTypeGolang,
+		Created: time.Now(),
+	}
+	ap.Append(p1)
+
+	p2 := gaia.Pipeline{
+		Name:    "Pipeline B",
+		Type:    gaia.PTypeGolang,
+		Created: time.Now(),
+	}
+	ap.Append(p2)
+
+	ap.Remove(1)
+
+	count := 0
+	for pipeline := range ap.Iter() {
+		count++
+		if pipeline.Name == "Pipeline B" {
+			t.Fatalf("Pipeline B still exists. It should have been removed.")
+		}
+	}
+
+	if count != 1 {
+		t.Fatalf("Expected pipeline count to be %v. Got %v.", 1, count)
+	}
+}
+
+func TestRemoveDeletedPipelines(t *testing.T) {
+	ap := NewActivePipelines()
+
+	p1 := gaia.Pipeline{
+		Name:    "Pipeline A",
+		Type:    gaia.PTypeGolang,
+		Created: time.Now(),
+	}
+	ap.Append(p1)
+
+	p2 := gaia.Pipeline{
+		Name:    "Pipeline B",
+		Type:    gaia.PTypeGolang,
+		Created: time.Now(),
+	}
+	ap.Append(p2)
+
+	p3 := gaia.Pipeline{
+		Name:    "Pipeline C",
+		Type:    gaia.PTypeGolang,
+		Created: time.Now(),
+	}
+	ap.Append(p3)
+
+	// Let's assume Pipeline B was deleted.
+	existingPipelineNames := []string{"Pipeline A", "Pipeline C"}
+
+	ap.RemoveDeletedPipelines(existingPipelineNames)
+
+	for pipeline := range ap.Iter() {
+		if pipeline.Name == "Pipeline B" {
+			t.Fatalf("Pipeline B still exists. It should have been removed.")
+		}
+	}
+
+}

pipeline/ticker.go

@@ -58,6 +58,7 @@ func InitTicker(store *store.Store, scheduler *scheduler.Scheduler) {
 // Every file will be handled as an active pipeline and therefore
 // saved in the global active pipelines slice.
 func checkActivePipelines() {
+	var existingPipelineNames []string
 	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)
@@ -78,6 +79,8 @@ func checkActivePipelines() {
 			// Get real pipeline name and check if the global active pipelines slice
 			// already contains it.
 			pName := getRealPipelineName(n, pType)
+			// Add the real pipeline name to the slice of existing pipeline names.
+			existingPipelineNames = append(existingPipelineNames, pName)
 			if GlobalActivePipelines.Contains(pName) {
 				// If SHA256Sum is set, we should check if pipeline has been changed.
 				p := GlobalActivePipelines.GetByName(pName)
@@ -113,7 +116,7 @@ func checkActivePipelines() {
 				continue
 			}
 
-			// We couldn't finde the pipeline. Create a new one.
+			// We couldn't find the pipeline. Create a new one.
 			var shouldStore = false
 			if pipeline == nil {
 				// Create pipeline object and fill it with information
@@ -152,6 +155,7 @@ func checkActivePipelines() {
 			GlobalActivePipelines.Append(*pipeline)
 		}
 	}
+	GlobalActivePipelines.RemoveDeletedPipelines(existingPipelineNames)
 }
 
 // getPipelineType looks up for specific suffix on the given file name.