remove deleted pipelines from global active pipeline list

Author: bidhan-a

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,23 @@ 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) {
+	for i, pipeline := range ap.Pipelines {
+		found := false
+		for _, name := range existingPipelineNames {
+			if pipeline.Name == name {
+				found = true
+				break
+			}
+		}
+		if !found {
+			ap.Remove(i)
+		}
+	}
+}
+
 // 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/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.