[WIP] Optimization for CreateOrUpdateConfigMap

pkg/controller/dependences.go

@@ -395,12 +395,12 @@ func NewDependencies(ns string, cliCfg *CLIConfig, clientset versioned.Interface
 	return deps, nil
 }
 
-func newFakeControl(kubeClientset kubernetes.Interface, informerFactory informers.SharedInformerFactory, kubeInformerFactory kubeinformers.SharedInformerFactory) Controls {
+func newFakeControl(kubeClientset kubernetes.Interface, informerFactory informers.SharedInformerFactory, kubeInformerFactory kubeinformers.SharedInformerFactory, labelFilterKubeInformerFactory kubeinformers.SharedInformerFactory) Controls {
 	genericCtrl := NewFakeGenericControl()
 	// Shared variables to construct `Dependencies` and some of its fields
 	return Controls{
 		JobControl:         NewFakeJobControl(kubeInformerFactory.Batch().V1().Jobs()),
-		ConfigMapControl:   NewFakeConfigMapControl(kubeInformerFactory.Core().V1().ConfigMaps()),
+		ConfigMapControl:   NewFakeConfigMapControl(labelFilterKubeInformerFactory.Core().V1().ConfigMaps()),
 		StatefulSetControl: NewFakeStatefulSetControl(kubeInformerFactory.Apps().V1().StatefulSets()),
 		ServiceControl:     NewFakeServiceControl(kubeInformerFactory.Core().V1().Services(), kubeInformerFactory.Core().V1().Endpoints()),
 		PVControl:          NewFakePVControl(kubeInformerFactory.Core().V1().PersistentVolumes(), kubeInformerFactory.Core().V1().PersistentVolumeClaims()),
@@ -454,6 +454,6 @@ func NewFakeDependencies() *Dependencies {
 	if err != nil {
 		klog.Fatalf("failed to create Dependencies: %s", err)
 	}
-	deps.Controls = newFakeControl(kubeCli, informerFactory, kubeInformerFactory)
+	deps.Controls = newFakeControl(kubeCli, informerFactory, kubeInformerFactory, labelFilterKubeInformerFactory)
 	return deps
 }

pkg/manager/member/config.go

@@ -15,10 +15,12 @@ package member
 
 import (
 	"fmt"
+	"reflect"
 
 	perrors "github.com/pingcap/errors"
 	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
 	"github.com/pingcap/tidb-operator/pkg/apis/util/toml"
+	"gopkg.in/yaml.v2"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	corelisters "k8s.io/client-go/listers/core/v1"
@@ -29,6 +31,8 @@ func updateConfigMap(old, new *corev1.ConfigMap) (bool, error) {
 
 	// check config
 	tomlField := []string{"config-file" /*pd,tikv,tidb */, "pump-config", "config_templ.toml" /*tiflash*/, "proxy_templ.toml" /*tiflash*/}
+	yamlField := []string{"prometheus-config", "prometheus.yml" /*prometheus*/}
+
 	for _, k := range tomlField {
 		oldData, oldOK := old.Data[k]
 		newData, newOK := new.Data[k]
@@ -53,6 +57,34 @@ func updateConfigMap(old, new *corev1.ConfigMap) (bool, error) {
 		}
 	}
 
+	for _, k := range yamlField {
+		oldData, oldOK := old.Data[k]
+		newData, newOK := new.Data[k]
+
+		if oldOK != newOK {
+			dataEqual = false
+		}
+
+		if !oldOK || !newOK {
+			continue
+		}
+		m1 := map[string]interface{}{}
+		m2 := map[string]interface{}{}
+		err := yaml.Unmarshal([]byte(oldData), &m1)
+		if err != nil {
+			dataEqual = false
+		}
+		err = yaml.Unmarshal([]byte(newData), &m2)
+		if err != nil {
+			dataEqual = false
+		}
+		if reflect.DeepEqual(m1, m2) {
+			new.Data[k] = oldData
+		} else {
+			dataEqual = false
+		}
+	}
+
 	// check startup script
 	field := "startup-script"
 	oldScript, oldExist := old.Data[field]

pkg/manager/member/config_test.go

@@ -134,6 +134,22 @@ func TestUpdateConfigMap(t *testing.T) {
 			updateKeys: []string{"pump-config", "config_templ.toml"},
 			equal:      false,
 		},
+		{
+			name: "add some yaml keys",
+			old: &corev1.ConfigMap{
+				Data: map[string]string{
+					"dashboard-config":  "foo",
+					"prometheus-config": "a = \"b\"",
+				},
+			},
+			new: &corev1.ConfigMap{
+				Data: map[string]string{
+					"prometheus.yml": "# comment \nc = \"d\"",
+				},
+			},
+			updateKeys: []string{"prometheus.yml"},
+			equal:      false,
+		},
 		{
 			name: "the data of old and new configmaps are the same",
 			old: &corev1.ConfigMap{

pkg/manager/member/dm_master_member_manager.go

@@ -371,7 +371,7 @@ func (m *masterMemberManager) syncMasterConfigMap(dc *v1alpha1.DMCluster, set *a
 	if err != nil {
 		return nil, err
 	}
-	return m.deps.TypedControl.CreateOrUpdateConfigMap(dc, newCm)
+	return CreateOrUpdateConfigMap(m.deps.ConfigMapLister, m.deps.ConfigMapControl, newCm, dc)
 }
 
 func (m *masterMemberManager) getNewMasterServiceForDMCluster(dc *v1alpha1.DMCluster) *corev1.Service {

pkg/manager/member/dm_worker_member_manager.go

@@ -321,7 +321,7 @@ func (m *workerMemberManager) syncWorkerConfigMap(dc *v1alpha1.DMCluster, set *a
 	if err != nil {
 		return nil, err
 	}
-	return m.deps.TypedControl.CreateOrUpdateConfigMap(dc, newCm)
+	return CreateOrUpdateConfigMap(m.deps.ConfigMapLister, m.deps.ConfigMapControl, newCm, dc)
 }
 
 func getNewWorkerSetForDMCluster(dc *v1alpha1.DMCluster, cm *corev1.ConfigMap) (*apps.StatefulSet, error) {

pkg/manager/member/dm_worker_member_manager_test.go

@@ -14,14 +14,14 @@
 package member
 
 import (
-	"context"
 	"fmt"
 	"strings"
 	"testing"
 
 	"github.com/pingcap/tidb-operator/pkg/apis/label"
 
 	"github.com/google/go-cmp/cmp"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/util/intstr"
 
 	. "github.com/onsi/gomega"
@@ -37,7 +37,6 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/utils/pointer"
-	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
 func TestWorkerMemberManagerSyncCreate(t *testing.T) {
@@ -80,7 +79,7 @@ func TestWorkerMemberManagerSyncCreate(t *testing.T) {
 			ctls.svc.SetCreateServiceError(errors.NewInternalError(fmt.Errorf("API server failed")), 0)
 		}
 		if test.errOnCreateCm {
-			ctls.generic.SetCreateOrUpdateError(errors.NewInternalError(fmt.Errorf("API server failed")), 0)
+			ctls.cm.SetCreateConfigMapError(errors.NewInternalError(fmt.Errorf("API server failed")), 0)
 		}
 
 		syncErr := wmm.SyncDM(dc)
@@ -94,9 +93,7 @@ func TestWorkerMemberManagerSyncCreate(t *testing.T) {
 			cmName = cmGen.Name
 			g.Expect(strings.HasPrefix(cmName, controller.DMWorkerMemberName(dcName))).To(BeTrue())
 		}
-		cm := &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: cmName}}
-		key := client.ObjectKeyFromObject(cm)
-		getCmErr := ctls.generic.FakeCli.Get(context.TODO(), key, cm)
+		cm, getCmErr := wmm.deps.ConfigMapLister.ConfigMaps(ns).Get(cmName)
 		result := result{syncErr, svc, getSvcErr, set, getStsErr, cm, getCmErr}
 		test.expectFn(g, &result)
 	}
@@ -228,7 +225,7 @@ func TestWorkerMemberManagerSyncUpdate(t *testing.T) {
 			ctls.svc.SetUpdateServiceError(errors.NewInternalError(fmt.Errorf("API server failed")), 0)
 		}
 		if test.errOnUpdateCm {
-			ctls.generic.SetCreateOrUpdateError(errors.NewInternalError(fmt.Errorf("API server failed")), 0)
+			ctls.cm.SetCreateConfigMapError(errors.NewInternalError(fmt.Errorf("API server failed")), 0)
 		}
 
 		oldCm, err := getWorkerConfigMap(dc)
@@ -240,8 +237,7 @@ func TestWorkerMemberManagerSyncUpdate(t *testing.T) {
 
 		g.Expect(indexers.set.Add(oldSet)).To(Succeed())
 		g.Expect(indexers.svc.Add(oldSvc)).To(Succeed())
-
-		g.Expect(ctls.generic.AddObject(oldCm)).To(Succeed())
+		g.Expect(indexers.cm.Add(oldCm)).To(Succeed())
 
 		if test.prepare != nil {
 			test.prepare(dc, indexers)
@@ -258,9 +254,10 @@ func TestWorkerMemberManagerSyncUpdate(t *testing.T) {
 			cmName = cmGen.Name
 			g.Expect(strings.HasPrefix(cmName, controller.DMWorkerMemberName(dcName))).To(BeTrue())
 		}
-		cm := &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: cmName}}
-		key := client.ObjectKeyFromObject(cm)
-		getCmErr := ctls.generic.FakeCli.Get(context.TODO(), key, cm)
+		cm, getCmErr := mmm.deps.ConfigMapLister.ConfigMaps(ns).Get(cmName)
+		if getCmErr != nil {
+			cm = &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: cmName}}
+		}
 		result := result{syncErr, oldSvc, svc, getSvcErr, oldSet, set, getStsErr, oldCm, cm, getCmErr, triggerDeleteWorker}
 		test.expectFn(g, &result)
 	}
@@ -593,7 +590,7 @@ func TestWorkerSyncConfigUpdate(t *testing.T) {
 		set    *appsv1.StatefulSet
 		getSet error
 		oldCm  *corev1.ConfigMap
-		cms    []corev1.ConfigMap
+		cms    []*corev1.ConfigMap
 		listCm error
 	}
 	type testcase struct {
@@ -610,7 +607,7 @@ func TestWorkerSyncConfigUpdate(t *testing.T) {
 		ns := dc.Namespace
 		dcName := dc.Name
 
-		mmm, controls, indexers, fakeMasterControl := newFakeWorkerMemberManager()
+		mmm, _, indexers, fakeMasterControl := newFakeWorkerMemberManager()
 		masterClient := controller.NewFakeMasterClient(fakeMasterControl, dc)
 		masterClient.AddReaction(dmapi.GetWorkersActionType, func(action *dmapi.Action) (interface{}, error) {
 			return test.workerInfos, nil
@@ -625,18 +622,17 @@ func TestWorkerSyncConfigUpdate(t *testing.T) {
 
 		g.Expect(indexers.set.Add(oldSet)).To(Succeed())
 		g.Expect(indexers.svc.Add(oldSvc)).To(Succeed())
-		g.Expect(controls.generic.AddObject(oldCm)).To(Succeed())
+		g.Expect(indexers.cm.Add(oldCm)).To(Succeed())
 
 		if test.prepare != nil {
 			test.prepare(dc, indexers)
 		}
 
 		syncErr := mmm.SyncDM(dc)
 		set, getStsErr := mmm.deps.StatefulSetLister.StatefulSets(ns).Get(controller.DMWorkerMemberName(dcName))
-		cmList := &corev1.ConfigMapList{}
+		cmList, listCmErr := mmm.deps.ConfigMapLister.ConfigMaps(ns).List(labels.Everything())
 		g.Expect(err).To(Succeed())
-		listCmErr := controls.generic.FakeCli.List(context.TODO(), cmList)
-		result := result{syncErr, oldSet, set, getStsErr, oldCm, cmList.Items, listCmErr}
+		result := result{syncErr, oldSet, set, getStsErr, oldCm, cmList, listCmErr}
 		test.expectFn(g, &result)
 	}
 
@@ -662,7 +658,7 @@ func TestWorkerSyncConfigUpdate(t *testing.T) {
 				for i := range r.cms {
 					cm := r.cms[i]
 					if cm.Name == using {
-						usingCm = &cm
+						usingCm = cm
 					}
 				}
 				g.Expect(usingCm).NotTo(BeNil(), "The configmap used by statefulset must be created")
@@ -685,13 +681,14 @@ func TestWorkerSyncConfigUpdate(t *testing.T) {
 type workerFakeIndexers struct {
 	svc cache.Indexer
 	set cache.Indexer
+	cm  cache.Indexer
 	pod cache.Indexer
 }
 
 type workerFakeControls struct {
-	svc     *controller.FakeServiceControl
-	set     *controller.FakeStatefulSetControl
-	generic *controller.FakeGenericControl
+	svc *controller.FakeServiceControl
+	set *controller.FakeStatefulSetControl
+	cm  *controller.FakeConfigMapControl
 }
 
 func newFakeWorkerMemberManager() (*workerMemberManager, *workerFakeControls, *workerFakeIndexers, *dmapi.FakeMasterControl) {
@@ -704,13 +701,14 @@ func newFakeWorkerMemberManager() (*workerMemberManager, *workerFakeControls, *w
 		failover: NewFakeWorkerFailover(),
 	}
 	controls := &workerFakeControls{
-		svc:     fakeDeps.ServiceControl.(*controller.FakeServiceControl),
-		set:     fakeDeps.StatefulSetControl.(*controller.FakeStatefulSetControl),
-		generic: fakeDeps.GenericControl.(*controller.FakeGenericControl),
+		svc: fakeDeps.ServiceControl.(*controller.FakeServiceControl),
+		set: fakeDeps.StatefulSetControl.(*controller.FakeStatefulSetControl),
+		cm:  fakeDeps.ConfigMapControl.(*controller.FakeConfigMapControl),
 	}
 	indexers := &workerFakeIndexers{
 		svc: fakeDeps.KubeInformerFactory.Core().V1().Services().Informer().GetIndexer(),
 		set: fakeDeps.KubeInformerFactory.Apps().V1().StatefulSets().Informer().GetIndexer(),
+		cm:  fakeDeps.LabelFilterKubeInformerFactory.Core().V1().ConfigMaps().Informer().GetIndexer(),
 		pod: fakeDeps.KubeInformerFactory.Core().V1().Pods().Informer().GetIndexer(),
 	}
 	return pmm, controls, indexers, masterControl

pkg/manager/member/pd_member_manager.go

@@ -423,7 +423,7 @@ func (m *pdMemberManager) syncPDConfigMap(tc *v1alpha1.TidbCluster, set *apps.St
 	if err != nil {
 		return nil, err
 	}
-	return m.deps.TypedControl.CreateOrUpdateConfigMap(tc, newCm)
+	return CreateOrUpdateConfigMap(m.deps.ConfigMapLister, m.deps.ConfigMapControl, newCm, tc)
 }
 
 func (m *pdMemberManager) getNewPDServiceForTidbCluster(tc *v1alpha1.TidbCluster) *corev1.Service {

pkg/manager/member/pump_member_manager.go

@@ -252,7 +252,7 @@ func (m *pumpMemberManager) syncConfigMap(tc *v1alpha1.TidbCluster, set *appsv1.
 	if err != nil {
 		return nil, err
 	}
-	return m.deps.TypedControl.CreateOrUpdateConfigMap(tc, newCm)
+	return CreateOrUpdateConfigMap(m.deps.ConfigMapLister, m.deps.ConfigMapControl, newCm, tc)
 }
 
 func getNewPumpHeadlessService(tc *v1alpha1.TidbCluster) *corev1.Service {