Enable version controlling for config map changes

pkg/apis/siddhi/v1alpha2/siddhiprocess_functions.go

@@ -60,11 +60,11 @@ func (p *MessagingSystem) Equals(q *MessagingSystem) bool {
 	return (typeEq && cidEq)
 }
 
-// Equals 
+// Equals function checks the equality of two SiddhiProcess specs
 func (p *SiddhiProcessSpec) Equals(q *SiddhiProcessSpec) bool {
 	if !EqualApps(p.Apps, q.Apps) {
 		return false
-	}
+	} 
 	if p.SiddhiConfig != q.SiddhiConfig {
 		return false
 	}
@@ -83,7 +83,7 @@ func (p *SiddhiProcessSpec) Equals(q *SiddhiProcessSpec) bool {
 	return true
 }
 
-// EqualApps 
+// EqualApps checks the equality of two app slices
 func EqualApps(p []Apps, q []Apps) bool {
 	if len(p) != len(q) {
 		return false
@@ -103,7 +103,7 @@ func EqualApps(p []Apps, q []Apps) bool {
 	return true
 }
 
-// EqualContainers 
+// EqualContainers checks the equality of two container specs
 func EqualContainers(p *corev1.Container, q *corev1.Container) bool {
 	if p.Image != q.Image {
 		return false

pkg/apis/siddhi/v1alpha2/siddhiprocess_types.go

@@ -87,7 +87,8 @@ type SiddhiProcessSpec struct {
 type SiddhiProcessStatus struct {
 	Status string   `json:"status"`
 	Ready  string   `json:"ready"`
-	ObservedGeneration int64 `json:"observedGeneration,omitempty"`
+	CurrentVersion int64 `json:"currentVersion"`
+	PreviousVersion int64 `json:"previousVersion"`
 }
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

pkg/controller/siddhiprocess/artifacts.go

@@ -75,7 +75,7 @@ func (rsp *ReconcileSiddhiProcess) CreateOrUpdateIngress(
 	sp *siddhiv1alpha2.SiddhiProcess,
 	siddhiApp SiddhiApp,
 	configs Configs,
-) (err error) {
+) (operationResult controllerutil.OperationResult, err error) {
 
 	var ingressPaths []extensionsv1beta1.HTTPIngressPath
 	for _, port := range siddhiApp.ContainerPorts {
@@ -142,7 +142,7 @@ func (rsp *ReconcileSiddhiProcess) CreateOrUpdateIngress(
 		},
 		Spec: ingressSpec,
 	}
-	_, err = controllerutil.CreateOrUpdate(
+	operationResult, err = controllerutil.CreateOrUpdate(
 		context.TODO(),
 		rsp.client,
 		ingress,

pkg/controller/siddhiprocess/artifacts_test.go

@@ -64,7 +64,7 @@ func TestCreateOrUpdateIngress(t *testing.T) {
 	cl := fake.NewFakeClient(objs...)
 	rsp := &ReconcileSiddhiProcess{client: cl, scheme: s}
 	configs := getTestConfigs(testSP)
-	err := rsp.CreateOrUpdateIngress(testSP, testSiddhiApp, configs)
+	_, err := rsp.CreateOrUpdateIngress(testSP, testSiddhiApp, configs)
 	if err != nil {
 		t.Error(err)
 	}
@@ -86,7 +86,7 @@ func TestCreateOrUpdateIngress(t *testing.T) {
 		},
 		PersistenceEnabled: true,
 	}
-	err = rsp.CreateOrUpdateIngress(testSP, sa, configs)
+	_, err = rsp.CreateOrUpdateIngress(testSP, sa, configs)
 	if err != nil {
 		t.Error(err)
 	}

pkg/controller/siddhiprocess/config.go

@@ -262,6 +262,12 @@ type SiddhiAppConfig struct {
 	Replicas                int32                    `json:"replicas"`
 }
 
+// ConfigMapListner holds the change details of a config map
+type ConfigMapListner struct {
+	SiddhiProcess string `json:"siddhiProcess"`
+	Changed       bool   `json:"changed"`
+}
+
 // Status of a Siddhi process
 type Status int
 

pkg/controller/siddhiprocess/operator.go

@@ -124,7 +124,7 @@ func (rsp *ReconcileSiddhiProcess) deployApp(
 			RollingUpdate: &rollingUpdate,
 		}
 	}
-	configMapName := siddhiApp.Name + strconv.Itoa(int(sp.ObjectMeta.Generation))
+	configMapName := siddhiApp.Name + strconv.Itoa(int(sp.Status.CurrentVersion))
 	for k, v := range siddhiApp.Apps {
 		key := k + configs.SiddhiExt
 		configMapData[key] = v
@@ -251,10 +251,10 @@ func (rsp *ReconcileSiddhiProcess) createArtifacts(
 	availableDep := 0
 	reqLogger := log.WithValues("Request.Namespace", sp.Namespace, "Request.Name", sp.Name)
 	eventType := controllerutil.OperationResultNone
-	if sp.Status.ObservedGeneration == 0 {
+	if sp.Status.CurrentVersion == 0 {
 		eventType = controllerutil.OperationResultCreated
 	} else {
-		if sp.ObjectMeta.Generation > sp.Status.ObservedGeneration {
+		if sp.Status.CurrentVersion > sp.Status.PreviousVersion {
 			eventType = controllerutil.OperationResultUpdated
 		} else {
 			eventType = controllerutil.OperationResultNone
@@ -270,19 +270,17 @@ func (rsp *ReconcileSiddhiProcess) createArtifacts(
 				sp = rsp.updateErrorStatus(sp, ER, ERROR, "AppDeploymentError", err)
 				continue
 			}
-			if (eventType != controllerutil.OperationResultNone) &&
+			if (eventType == controllerutil.OperationResultCreated) &&
 				(operationResult == controllerutil.OperationResultCreated) {
-				availableDep++
 				sp = rsp.updateRunningStatus(
 					sp,
 					ER,
 					RUNNING,
 					"DeploymentCreated",
 					(siddhiApp.Name + " deployment created successfully"),
 				)
-			} else if (eventType != controllerutil.OperationResultNone) &&
+			} else if (eventType == controllerutil.OperationResultUpdated) &&
 				(operationResult == controllerutil.OperationResultUpdated) {
-				availableDep++
 				sp = rsp.updateRunningStatus(
 					sp,
 					ER,
@@ -291,14 +289,14 @@ func (rsp *ReconcileSiddhiProcess) createArtifacts(
 					(siddhiApp.Name + " deployment updated successfully"),
 				)
 			}
-
+			availableDep++
 			if siddhiApp.ServiceEnabled {
 				operationResult, err = rsp.CreateOrUpdateService(sp, siddhiApp, configs)
 				if err != nil {
 					sp = rsp.updateErrorStatus(sp, ER, WARNING, "ServiceCreationError", err)
 					continue
 				}
-				if (eventType != controllerutil.OperationResultNone) &&
+				if (eventType == controllerutil.OperationResultCreated) &&
 					(operationResult == controllerutil.OperationResultCreated) {
 					sp = rsp.updateRunningStatus(
 						sp,
@@ -307,7 +305,7 @@ func (rsp *ReconcileSiddhiProcess) createArtifacts(
 						"ServiceCreated",
 						(siddhiApp.Name + " service created successfully"),
 					)
-				} else if (eventType != controllerutil.OperationResultNone) &&
+				} else if (eventType == controllerutil.OperationResultUpdated) &&
 					(operationResult == controllerutil.OperationResultUpdated) {
 					sp = rsp.updateRunningStatus(
 						sp,
@@ -319,20 +317,23 @@ func (rsp *ReconcileSiddhiProcess) createArtifacts(
 				}
 
 				if configs.AutoCreateIngress {
-					err := rsp.CreateOrUpdateIngress(sp, siddhiApp, configs)
+					operationResult, err = rsp.CreateOrUpdateIngress(sp, siddhiApp, configs)
 					if err != nil {
 						sp = rsp.updateErrorStatus(sp, ER, ERROR, "IngressCreationError", err)
 						continue
 					}
-					if eventType == controllerutil.OperationResultCreated ||
-						eventType == controllerutil.OperationResultUpdated {
+					if (eventType == controllerutil.OperationResultCreated) &&
+						(operationResult == controllerutil.OperationResultCreated) {
+						reqLogger.Info("Ingress created", "Ingress.Name", configs.HostName)
+					} else if (eventType == controllerutil.OperationResultUpdated) &&
+						(operationResult == controllerutil.OperationResultUpdated) {
 						reqLogger.Info("Ingress changed", "Ingress.Name", configs.HostName)
 					}
 				}
 			}
 		}
 	}
-	sp = rsp.syncGeneration(sp)
+	sp = rsp.syncVersion(sp)
 	if (eventType == controllerutil.OperationResultCreated) ||
 		(eventType == controllerutil.OperationResultUpdated) {
 		sp = rsp.updateReady(sp, availableDep, needDep)
@@ -369,7 +370,7 @@ func (rsp *ReconcileSiddhiProcess) populateSiddhiApps(
 	var siddhiApps []SiddhiApp
 	var err error
 	modified := false
-	if (sp.Status.ObservedGeneration > 0) && (sp.ObjectMeta.Generation > sp.Status.ObservedGeneration) {
+	if sp.Status.CurrentVersion > sp.Status.PreviousVersion {
 		modified = true
 	}
 	if modified {
@@ -425,6 +426,11 @@ func (rsp *ReconcileSiddhiProcess) getSiddhiApps(sp *siddhiv1alpha2.SiddhiProces
 			if err != nil {
 				return
 			}
+			cmListner := ConfigMapListner{
+				SiddhiProcess: sp.Name,
+				Changed:       false,
+			}
+			CMContainer[app.ConfigMap] = cmListner
 			for _, siddhiFileContent := range configMap.Data {
 				siddhiApps = append(siddhiApps, siddhiFileContent)
 			}
@@ -436,11 +442,18 @@ func (rsp *ReconcileSiddhiProcess) getSiddhiApps(sp *siddhiv1alpha2.SiddhiProces
 	return
 }
 
-// syncGeneration synchronize the siddhi process internal generation number
-// this simply assing ObjectMeta.Generation value to the Status.ObservedGeneration and update the sidhhi process
+// syncVersion synchronize the siddhi process internal version number
+// this simply assing Status.CurrentVersion value to the Status.PreviousVersion and update the siddhi process
 // this funtionality used for version controlling inside a siddhi process
-func (rsp *ReconcileSiddhiProcess) syncGeneration(sp *siddhiv1alpha2.SiddhiProcess) *siddhiv1alpha2.SiddhiProcess {
-	sp.Status.ObservedGeneration = sp.ObjectMeta.Generation
+func (rsp *ReconcileSiddhiProcess) syncVersion(sp *siddhiv1alpha2.SiddhiProcess) *siddhiv1alpha2.SiddhiProcess {
+	sp.Status.PreviousVersion = sp.Status.CurrentVersion
+	_ = rsp.client.Status().Update(context.TODO(), sp)
+	return sp
+}
+
+// upgradeVersion upgrade the siddhi process internal version number
+func (rsp *ReconcileSiddhiProcess) upgradeVersion(sp *siddhiv1alpha2.SiddhiProcess) *siddhiv1alpha2.SiddhiProcess {
+	sp.Status.CurrentVersion++
 	_ = rsp.client.Status().Update(context.TODO(), sp)
 	return sp
 }
@@ -513,7 +526,7 @@ func (rsp *ReconcileSiddhiProcess) cleanArtifacts(
 				}
 			}
 
-			cmName := artifactName + strconv.Itoa(int(sp.Status.ObservedGeneration))
+			cmName := artifactName + strconv.Itoa(int(sp.Status.PreviousVersion))
 			cm := &corev1.ConfigMap{}
 			er = rsp.client.Get(
 				context.TODO(),

pkg/controller/siddhiprocess/siddhiprocess_controller.go

@@ -44,6 +44,9 @@ var log = logf.Log.WithName("siddhi")
 // SPContainer holds siddhi apps
 var SPContainer map[string][]SiddhiApp
 
+// CMContainer holds the config map name along with CM listner for listen changes of the CM
+var CMContainer map[string]ConfigMapListner
+
 // ER recoder
 var ER record.EventRecorder
 
@@ -61,6 +64,7 @@ func newReconciler(mgr manager.Manager) reconcile.Reconciler {
 // add adds a new Controller to mgr with rsp as the reconcile.Reconciler
 func add(mgr manager.Manager, r reconcile.Reconciler) error {
 	SPContainer = make(map[string][]SiddhiApp)
+	CMContainer = make(map[string]ConfigMapListner)
 	ER = mgr.GetRecorder("siddhiprocess-controller")
 
 	// Create a new controller
@@ -69,7 +73,7 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error {
 		return err
 	}
 
-	pred := predicate.Funcs{
+	spPredicate := predicate.Funcs{
 		DeleteFunc: func(e event.DeleteEvent) bool {
 			if _, ok := SPContainer[e.Meta.GetName()]; ok {
 				delete(SPContainer, e.Meta.GetName())
@@ -80,22 +84,45 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error {
 			oldObject := e.ObjectOld.(*siddhiv1alpha2.SiddhiProcess)
 			newObject := e.ObjectNew.(*siddhiv1alpha2.SiddhiProcess)
 			if !oldObject.Spec.Equals(&newObject.Spec) {
+				newObject.Status.CurrentVersion++
 				return true
 			}
 			return false
 		},
 		CreateFunc: func(e event.CreateEvent) bool {
 			object := e.Object.(*siddhiv1alpha2.SiddhiProcess)
-			object.Status.ObservedGeneration = 0
+			object.Status.CurrentVersion = 0
+			object.Status.PreviousVersion = 0
 			return true
 		},
 		GenericFunc: func(e event.GenericEvent) bool {
 			return false
 		},
 	}
 
+	cmPredicate := predicate.Funcs{
+		UpdateFunc: func(e event.UpdateEvent) bool {
+			if _, ok := CMContainer[e.MetaNew.GetName()]; ok {
+				cmListner := CMContainer[e.MetaNew.GetName()]
+				cmListner.Changed = true
+				CMContainer[e.MetaNew.GetName()] = cmListner
+				return true
+			}
+			return false
+		},
+		CreateFunc: func(e event.CreateEvent) bool {
+			return false
+		},
+	}
+
 	// Watch for changes to primary resource SiddhiProcess
-	err = c.Watch(&source.Kind{Type: &siddhiv1alpha2.SiddhiProcess{}}, &handler.EnqueueRequestForObject{}, pred)
+	err = c.Watch(&source.Kind{Type: &siddhiv1alpha2.SiddhiProcess{}}, &handler.EnqueueRequestForObject{}, spPredicate)
+	if err != nil {
+		return err
+	}
+
+	// Watch for changes to secondary resource Config Map and requeue the owner SiddhiProcess
+	err = c.Watch(&source.Kind{Type: &corev1.ConfigMap{}}, &handler.EnqueueRequestForObject{}, cmPredicate)
 	if err != nil {
 		return err
 	}
@@ -144,13 +171,27 @@ type ReconcileSiddhiProcess struct {
 // and what is in the SiddhiProcess.Spec
 func (rsp *ReconcileSiddhiProcess) Reconcile(request reconcile.Request) (reconcile.Result, error) {
 	sp := &siddhiv1alpha2.SiddhiProcess{}
-	err := rsp.client.Get(context.TODO(), request.NamespacedName, sp)
+	cm := &corev1.ConfigMap{}
+	siddhiProcessName := request.NamespacedName
+	SiddhiProcessChanged := true
+	err := rsp.client.Get(context.TODO(), request.NamespacedName, cm)
+	if err == nil {
+		cmListner := CMContainer[request.NamespacedName.Name]
+		siddhiProcessName.Name = cmListner.SiddhiProcess
+		SiddhiProcessChanged = false
+	}
+
+	err = rsp.client.Get(context.TODO(), siddhiProcessName, sp)
 	if err != nil {
 		if errors.IsNotFound(err) {
 			return reconcile.Result{}, nil
 		}
 		return reconcile.Result{}, err
 	}
+	if !SiddhiProcessChanged {
+		sp = rsp.upgradeVersion(sp)
+	}
+
 	configs := rsp.Configurations(sp)
 	sp, siddhiApps, err := rsp.populateSiddhiApps(sp, configs)
 	if err != nil {
@@ -166,5 +207,10 @@ func (rsp *ReconcileSiddhiProcess) Reconcile(request reconcile.Request) (reconci
 
 	sp = rsp.createArtifacts(sp, siddhiApps, configs)
 	sp = rsp.checkDeployments(sp, siddhiApps)
+	if !SiddhiProcessChanged {
+		cmListner := CMContainer[request.NamespacedName.Name]
+		cmListner.Changed = false
+		CMContainer[request.NamespacedName.Name] = cmListner
+	}
 	return reconcile.Result{Requeue: false}, nil
 }