Merge pull request #361 from jzding/cleanup-configmap

OCPBUGS-42840: cleanup configmap when event apiVersion change

cmd/main.go

@@ -139,7 +139,7 @@ func main() {
 	}
 	if namespace != "" && nodeName != "" && scConfig.TransportHost.Type == common.HTTP {
 		// if consumer doesn't pass namespace then this will default to empty dir
-		if e := client.InitConfigMap(scConfig.StorePath, nodeName, namespace); e != nil {
+		if e := client.InitConfigMap(scConfig.APIVersion, scConfig.StorePath, nodeName, namespace); e != nil {
 			log.Errorf("failed to initlialize configmap, subcription will be stored in empty dir %s", e.Error())
 		} else {
 			scConfig.StorageType = storageClient.ConfigMap

pkg/storage/kubernetes/client.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"os"
+	"strings"
 
 	"github.com/golang/glog"
 	"github.com/redhat-cne/sdk-go/pkg/channel"
@@ -64,9 +65,13 @@ func NewClient() (*Client, error) {
 }
 
 // CreateConfigMap ... create configmap
-func (sClient *Client) CreateConfigMap(ctx context.Context, nodeName, namespace string) (cm *corev1.ConfigMap, err error) {
+func (sClient *Client) CreateConfigMap(ctx context.Context, apiVersion, nodeName, namespace string) (cm *corev1.ConfigMap, err error) {
 	cm, err = sClient.GetConfigMap(ctx, nodeName, namespace)
 	if err == nil {
+		// clean up configMap if apiVersion changed
+		if apiVersion != "" && !validateConfigMap(apiVersion, cm) {
+			return sClient.cleanupConfigMap(ctx, cm, namespace)
+		}
 		return cm, nil
 	}
 
@@ -108,7 +113,7 @@ func (sClient *Client) UpdateConfigMap(ctx context.Context, data []subscriber.Su
 	var err error
 	cm, err = sClient.GetConfigMap(ctx, nodeName, namespace)
 	if err != nil {
-		if cm, err = sClient.CreateConfigMap(ctx, nodeName, namespace); err != nil {
+		if cm, err = sClient.CreateConfigMap(ctx, "", nodeName, namespace); err != nil {
 			log.Errorf("Error fetching configmap %s", err.Error())
 			return err
 		}
@@ -148,10 +153,10 @@ func (sClient *Client) UpdateConfigMap(ctx context.Context, data []subscriber.Su
 }
 
 // InitConfigMap ... using configmap
-func (sClient *Client) InitConfigMap(storePath, nodeName, namespace string) error {
+func (sClient *Client) InitConfigMap(apiVersion, storePath, nodeName, namespace string) error {
 	var err error
 	var cm *corev1.ConfigMap
-	if cm, err = sClient.CreateConfigMap(context.Background(), nodeName, namespace); err == nil {
+	if cm, err = sClient.CreateConfigMap(context.Background(), apiVersion, nodeName, namespace); err == nil {
 		for clientID, subscriberData := range cm.Data {
 			var newSubscriberBytes []byte
 			var subscriberErr error
@@ -179,3 +184,49 @@ func (sClient *Client) InitConfigMap(storePath, nodeName, namespace string) erro
 	}
 	return nil
 }
+
+func (sClient *Client) cleanupConfigMap(ctx context.Context, cm *corev1.ConfigMap, namespace string) (*corev1.ConfigMap, error) {
+	cm.Data = make(map[string]string)
+	_, err := sClient.clientSet.CoreV1().ConfigMaps(namespace).Update(ctx, cm, metav1.UpdateOptions{})
+	if err != nil {
+		log.Errorf("error updating configmap %s", err.Error())
+		return cm, err
+	}
+	log.Info("configmap cleaned up")
+	return cm, nil
+}
+
+func validateSubscriberVersion(apiVersion string, sub subscriber.Subscriber) bool {
+	if sub.SubStore == nil || len(sub.SubStore.Store) == 0 {
+		return true
+	}
+
+	for _, v := range sub.SubStore.Store {
+		if !isVersionsCompatible(v.GetVersion(), apiVersion) {
+			return false
+		}
+	}
+	return true
+}
+
+func validateConfigMap(apiVersion string, cm *corev1.ConfigMap) bool {
+	for _, subscriberData := range cm.Data {
+		if subscriberData == "" {
+			continue
+		}
+		var subscriberErr error
+		subscriber := subscriber.Subscriber{}
+		if err := json.Unmarshal([]byte(subscriberData), &subscriber); err == nil {
+			_, subscriberErr = json.MarshalIndent(&subscriber, "", " ")
+			if subscriberErr != nil || !validateSubscriberVersion(apiVersion, subscriber) {
+				return false
+			}
+		}
+	}
+	return true
+}
+
+// isVersionsCompatible compares major versions assuming inputs are valid
+func isVersionsCompatible(ver1, ver2 string) bool {
+	return strings.Split(ver1, ".")[0] == strings.Split(ver2, ".")[0]
+}

pkg/storage/kubernetes/client_test.go

@@ -40,7 +40,7 @@ func Subscriptions() *v1.ConfigMap {
 
 func TestClient_InitConfigMap(t *testing.T) {
 	setupClient()
-	err := clients.InitConfigMap(".", nodeName, metav1.NamespaceSystem)
+	err := clients.InitConfigMap("1.0", ".", nodeName, metav1.NamespaceSystem)
 	assert.Nil(t, err)
 	cm, e := clients.GetConfigMap(context.Background(), nodeName, metav1.NamespaceSystem)
 	assert.Nil(t, e)
@@ -49,7 +49,7 @@ func TestClient_InitConfigMap(t *testing.T) {
 
 func TestClient_GetConfigMap(t *testing.T) {
 	setupClient()
-	err := clients.InitConfigMap(".", nodeName, metav1.NamespaceSystem)
+	err := clients.InitConfigMap("1.0", ".", nodeName, metav1.NamespaceSystem)
 	assert.Nil(t, err)
 	cm, e := clients.GetConfigMap(context.Background(), nodeName, metav1.NamespaceSystem)
 	assert.Nil(t, e)
@@ -59,7 +59,7 @@ func TestClient_GetConfigMap(t *testing.T) {
 
 func Test_LoadingSubscriptionFromFileToCache(t *testing.T) {
 	setupClient()
-	err := clients.InitConfigMap(".", nodeName, metav1.NamespaceSystem)
+	err := clients.InitConfigMap("1.0", ".", nodeName, metav1.NamespaceSystem)
 	assert.Nil(t, err)
 	cm, e := clients.GetConfigMap(context.Background(), nodeName, metav1.NamespaceSystem)
 	assert.Nil(t, e)