reduce number of configmaps update
Update configmap only if it's changed to reduce number of unneeded updates and etcd load
MODIFY
pkg/controller/pxc/backup.go
pkg/controller/pxc/backup.go
@@ -4,6 +4,7 @@ import (
"container/heap"
"context"
"fmt"
+ "reflect"
"github.com/pkg/errors"
batchv1beta1 "k8s.io/api/batch/v1beta1"
@@ -45,19 +46,23 @@ func (r *ReconcilePerconaXtraDBCluster) reconcileBackups(cr *api.PerconaXtraDBCl
// Check if this Job already exists
currentBcpJob := new(batchv1beta1.CronJob)
- err = r.client.Get(context.TODO(), types.NamespacedName{Name: bcpjob.Name, Namespace: bcpjob.Namespace}, currentBcpJob)
- if err != nil && k8serrors.IsNotFound(err) {
- // reqLogger.Info("Creating a new backup job", "Namespace", bcpjob.Namespace, "Name", bcpjob.Name)
+ err = r.client.Get(context.TODO(), types.NamespacedName{
+ Name: bcpjob.Name,
+ Namespace: bcpjob.Namespace,
+ }, currentBcpJob)
+ if err != nil && !k8serrors.IsNotFound(err) {
+ return errors.Wrapf(err, "create scheduled backup %s", bcp.Name)
+ }
+
+ if k8serrors.IsNotFound(err) {
err = r.client.Create(context.TODO(), bcpjob)
if err != nil {
- return fmt.Errorf("create scheduled backup '%s': %v", bcp.Name, err)
+ return errors.Wrapf(err, "create scheduled backup %s", bcp.Name)
}
- } else if err != nil {
- return fmt.Errorf("create scheduled backup '%s': %v", bcp.Name, err)
- } else {
+ } else if !reflect.DeepEqual(currentBcpJob.Spec, bcpjob.Spec) {
err = r.client.Update(context.TODO(), bcpjob)
if err != nil {
- return fmt.Errorf("update backup schedule '%s': %v", bcp.Name, err)
+ return errors.Wrapf(err, "update backup schedule %s", bcp.Name)
}
}
}

MODIFY
pkg/controller/pxc/controller.go
pkg/controller/pxc/controller.go
@@ -4,6 +4,7 @@ import (
"context"
"crypto/md5"
"fmt"
+ "reflect"
"strings"
"sync"
"sync/atomic"
@@ -11,6 +12,7 @@ import (
"github.com/pkg/errors"
"github.com/robfig/cron/v3"
+ appsv1 "k8s.io/api/apps/v1"
corev1 "k8s.io/api/core/v1"
policyv1beta1 "k8s.io/api/policy/v1beta1"
k8serrors "k8s.io/apimachinery/pkg/api/errors"
@@ -286,8 +288,7 @@ func (r *ReconcilePerconaXtraDBCluster) Reconcile(request reconcile.Request) (re
pxc.MergeTemplateAnnotations(pxcSet.StatefulSet(), pxcAnnotations)
err = r.updatePod(pxcSet, o.Spec.PXC, o, inits)
if err != nil {
- err = fmt.Errorf("pxc upgrade error: %v", err)
- return reconcile.Result{}, err
+ return reconcile.Result{}, errors.Wrap(err, "pxc upgrade error")
}
for _, pxcService := range []*corev1.Service{pxc.NewServicePXC(o), pxc.NewServicePXCUnready(o)} {
@@ -297,12 +298,15 @@ func (r *ReconcilePerconaXtraDBCluster) Reconcile(request reconcile.Request) (re
return reconcile.Result{}, errors.Wrap(err, "failed to get current PXC service")
}
+ if reflect.DeepEqual(currentService.Spec.Ports, pxcService.Spec.Ports) {
+ continue
+ }
+
currentService.Spec.Ports = pxcService.Spec.Ports
err = r.client.Update(context.TODO(), currentService)
if err != nil {
- err = fmt.Errorf("PXC service upgrade error: %v", err)
- return reconcile.Result{}, err
+ return reconcile.Result{}, errors.Wrap(err, "PXC service upgrade error")
}
}
@@ -312,15 +316,13 @@ func (r *ReconcilePerconaXtraDBCluster) Reconcile(request reconcile.Request) (re
if o.Spec.HAProxy != nil && o.Spec.HAProxy.Enabled {
err = r.updatePod(haProxySet, o.Spec.HAProxy, o, nil)
if err != nil {
- err = fmt.Errorf("HAProxy upgrade error: %v", err)
- return reconcile.Result{}, err
+ return reconcile.Result{}, errors.Wrap(err, "HAProxy upgrade error")
}
currentService := &corev1.Service{}
err := r.client.Get(context.TODO(), types.NamespacedName{Name: haProxyService.Name, Namespace: haProxyService.Namespace}, currentService)
if err != nil {
- err = fmt.Errorf("failed to get HAProxy service: %v", err)
- return reconcile.Result{}, err
+ return reconcile.Result{}, errors.Wrap(err, "failed to get HAProxy service")
}
ports := []corev1.ServicePort{
@@ -376,8 +378,7 @@ func (r *ReconcilePerconaXtraDBCluster) Reconcile(request reconcile.Request) (re
currentServiceReplicas := &corev1.Service{}
err = r.client.Get(context.TODO(), types.NamespacedName{Name: haProxyServiceReplicas.Name, Namespace: haProxyServiceReplicas.Namespace}, currentServiceReplicas)
if err != nil {
- err = fmt.Errorf("failed to get HAProxyReplicas service: %v", err)
- return reconcile.Result{}, err
+ return reconcile.Result{}, errors.Wrap(err, "failed to get HAProxyReplicas service")
}
replicaPorts := []corev1.ServicePort{
@@ -409,18 +410,17 @@ func (r *ReconcilePerconaXtraDBCluster) Reconcile(request reconcile.Request) (re
err = r.client.Update(context.TODO(), currentServiceReplicas)
if err != nil {
- err = fmt.Errorf("HAProxyReplicas service upgrade error: %v", err)
- return reconcile.Result{}, err
+ return reconcile.Result{}, errors.Wrap(err, "HAProxyReplicas service upgrade error")
}
} else {
err = r.deleteStatefulSet(o.Namespace, haProxySet, false)
if err != nil {
- return reconcile.Result{}, err
+ return reconcile.Result{}, errors.Wrap(err, "delete HAProxy stateful set")
}
haProxyReplicasService := pxc.NewServiceHAProxyReplicas(o)
err = r.deleteServices([]*corev1.Service{haProxyService, haProxyReplicasService})
if err != nil {
- return reconcile.Result{}, err
+ return reconcile.Result{}, errors.Wrap(err, "delete HAProxy replica service")
}
}
@@ -787,22 +787,18 @@ func (r *ReconcilePerconaXtraDBCluster) reconcileConfigMap(cr *api.PerconaXtraDB
if cr.CompareVersionWith("1.3.0") >= 0 {
if len(limitMemory) > 0 || len(requestMemory) > 0 {
- autoConfigMap, err := config.NewAutoTuneConfigMap(cr, "auto-"+ls["app.kubernetes.io/instance"]+"-"+ls["app.kubernetes.io/component"])
+ configMap, err := config.NewAutoTuneConfigMap(cr, "auto-"+ls["app.kubernetes.io/instance"]+"-"+ls["app.kubernetes.io/component"])
if err != nil {
return errors.Wrap(err, "new auto-config map")
}
- err = setControllerReference(cr, autoConfigMap, r.scheme)
+ err = setControllerReference(cr, configMap, r.scheme)
if err != nil {
return errors.Wrap(err, "set auto-config controller ref")
}
- err = r.client.Create(context.TODO(), autoConfigMap)
- if err != nil && k8serrors.IsAlreadyExists(err) {
- err = r.client.Update(context.TODO(), autoConfigMap)
+
+ err = createOrUpdateConfigmap(r.client, configMap)
if err != nil {
- return errors.Wrap(err, "update AutoConfigMap")
- }
- } else if err != nil {
- return errors.Wrap(err, "create AutoConfigMap")
+ return errors.Wrap(err, "auto-config config map")
}
}
}
@@ -813,48 +809,36 @@ func (r *ReconcilePerconaXtraDBCluster) reconcileConfigMap(cr *api.PerconaXtraDB
if err != nil {
return errors.Wrap(err, "set controller ref")
}
- err = r.client.Create(context.TODO(), configMap)
- if err != nil && k8serrors.IsAlreadyExists(err) {
- err = r.client.Update(context.TODO(), configMap)
+
+ err = createOrUpdateConfigmap(r.client, configMap)
if err != nil {
- return errors.Wrap(err, "update ConfigMap")
- }
- } else if err != nil {
- return errors.Wrap(err, "create ConfigMap")
+ return errors.Wrap(err, "pxc config map")
}
}
- if cr.Spec.ProxySQL != nil && cr.Spec.ProxySQL.Configuration != "" {
+ if cr.Spec.ProxySQL != nil && cr.Spec.ProxySQL.Enabled && cr.Spec.ProxySQL.Configuration != "" {
configMap := config.NewConfigMap(cr, ls["app.kubernetes.io/instance"]+"-proxysql", "proxysql.cnf", cr.Spec.ProxySQL.Configuration)
err := setControllerReference(cr, configMap, r.scheme)
if err != nil {
return errors.Wrap(err, "set controller ref ProxySQL")
}
- err = r.client.Create(context.TODO(), configMap)
- if err != nil && k8serrors.IsAlreadyExists(err) {
- err = r.client.Update(context.TODO(), configMap)
+
+ err = createOrUpdateConfigmap(r.client, configMap)
if err != nil {
- return errors.Wrap(err, "update ConfigMap HAProxy")
- }
- } else if err != nil {
- return errors.Wrap(err, "create ConfigMap HAProxy")
+ return errors.Wrap(err, "proxysql config map")
}
}
- if cr.Spec.HAProxy != nil && cr.Spec.HAProxy.Configuration != "" {
+ if cr.Spec.HAProxy != nil && cr.Spec.HAProxy.Enabled && cr.Spec.HAProxy.Configuration != "" {
configMap := config.NewConfigMap(cr, ls["app.kubernetes.io/instance"]+"-haproxy", "haproxy-global.cfg", cr.Spec.HAProxy.Configuration)
err := setControllerReference(cr, configMap, r.scheme)
if err != nil {
return errors.Wrap(err, "set controller ref HAProxy")
}
- err = r.client.Create(context.TODO(), configMap)
- if err != nil && k8serrors.IsAlreadyExists(err) {
- err = r.client.Update(context.TODO(), configMap)
+
+ err = createOrUpdateConfigmap(r.client, configMap)
if err != nil {
- return errors.Wrap(err, "update ConfigMap HAProxy")
- }
- } else if err != nil {
- return errors.Wrap(err, "create ConfigMap HAProxy")
+ return errors.Wrap(err, "haproxy config map")
}
}
@@ -943,14 +927,26 @@ func (r *ReconcilePerconaXtraDBCluster) deleteStatefulSetPods(namespace string,
}
func (r *ReconcilePerconaXtraDBCluster) deleteStatefulSet(namespace string, sfs api.StatefulApp, deletePVC bool) error {
- err := r.client.Delete(context.TODO(), sfs.StatefulSet())
+ err := r.client.Get(context.TODO(), types.NamespacedName{
+ Name: sfs.StatefulSet().Name,
+ Namespace: namespace,
+ }, &appsv1.StatefulSet{})
+ if err != nil && !k8serrors.IsNotFound(err) {
+ return errors.Wrapf(err, "get statefulset: %s", sfs.StatefulSet().Name)
+ }
+
+ if k8serrors.IsNotFound(err) {
+ return nil
+ }
+
+ err = r.client.Delete(context.TODO(), sfs.StatefulSet())
if err != nil && !k8serrors.IsNotFound(err) {
- return fmt.Errorf("delete proxysql: %v", err)
+ return errors.Wrapf(err, "delete statefulset: %s", sfs.StatefulSet().Name)
}
if deletePVC {
err = r.deletePVC(namespace, sfs.Labels())
if err != nil {
- return fmt.Errorf("delete proxysql pvc: %v", err)
+ return errors.Wrapf(err, "delete pvc: %s", sfs.StatefulSet().Name)
}
}
@@ -959,9 +955,21 @@ func (r *ReconcilePerconaXtraDBCluster) deleteStatefulSet(namespace string, sfs
func (r *ReconcilePerconaXtraDBCluster) deleteServices(svcs []*corev1.Service) error {
for _, s := range svcs {
- err := r.client.Delete(context.TODO(), s)
+ err := r.client.Get(context.TODO(), types.NamespacedName{
+ Name: s.Name,
+ Namespace: s.Namespace,
+ }, &corev1.Service{})
if err != nil && !k8serrors.IsNotFound(err) {
- return fmt.Errorf("delete service: %v", err)
+ return errors.Wrapf(err, "get service: %s", s.Name)
+ }
+
+ if k8serrors.IsNotFound(err) {
+ continue
+ }
+
+ err = r.client.Delete(context.TODO(), s)
+ if err != nil {
+ return errors.Wrapf(err, "delete service: %s", s.Name)
}
}
return nil
@@ -1035,3 +1043,24 @@ func (r *ReconcilePerconaXtraDBCluster) resyncPXCUsersWithProxySQL(cr *api.Perco
atomic.StoreInt32(&r.syncUsersState, stateFree)
}()
}
+
+func createOrUpdateConfigmap(cl client.Client, configMap *corev1.ConfigMap) error {
+ currMap := &corev1.ConfigMap{}
+ err := cl.Get(context.TODO(), types.NamespacedName{
+ Namespace: configMap.Namespace,
+ Name: configMap.Name,
+ }, currMap)
+ if err != nil && !k8serrors.IsNotFound(err) {
+ return errors.Wrap(err, "get current configmap")
+ }
+
+ if k8serrors.IsNotFound(err) {
+ return cl.Create(context.TODO(), configMap)
+ }
+
+ if !reflect.DeepEqual(currMap.Data, configMap.Data) {
+ return cl.Update(context.TODO(), configMap)
+ }
+
+ return nil
+}

K8SPXC-546: reduce number of configmaps update (#683)

Update configmap only if it's changed to reduce number of unneeded updates and etcd load