V0.2.4

Dockerfile

@@ -13,7 +13,7 @@ ARG GIT_SHA=0000000
 
 WORKDIR /src
 
-COPY go.mod ./ go.sum ./
+COPY go.mod go.sum ./
 RUN go mod download
 
 COPY pkg ./ cmd ./ version ./

pkg/apis/redis/v1alpha1/default.go

@@ -12,9 +12,10 @@ import (
 )
 
 const (
-	minMasterSize      = 3
-	minClusterReplicas = 1
-	defaultRedisImage  = "redis:5.0.4-alpine"
+	minMasterSize       = 3
+	minClusterReplicas  = 1
+	defaultRedisImage   = "redis:5.0.4-alpine"
+	defaultMonitorImage = "oliver006/redis_exporter:latest"
 )
 
 func (in *DistributedRedisCluster) DefaultSpec(log logr.Logger) bool {
@@ -41,6 +42,11 @@ func (in *DistributedRedisCluster) DefaultSpec(log logr.Logger) bool {
 
 	mon := in.Spec.Monitor
 	if mon != nil {
+		if mon.Image == "" {
+			mon.Image = defaultMonitorImage
+			update = true
+		}
+
 		if mon.Prometheus == nil {
 			mon.Prometheus = &PrometheusSpec{}
 			update = true

pkg/apis/redis/v1alpha1/distributedrediscluster_types.go

@@ -24,16 +24,18 @@ type DistributedRedisClusterSpec struct {
 	ClusterReplicas  int32                         `json:"clusterReplicas,omitempty"`
 	ServiceName      string                        `json:"serviceName,omitempty"`
 	Config           map[string]string             `json:"config,omitempty"`
-	Affinity         *corev1.Affinity              `json:"affinity,omitempty"`
-	NodeSelector     map[string]string             `json:"nodeSelector,omitempty"`
-	ToleRations      []corev1.Toleration           `json:"toleRations,omitempty"`
-	SecurityContext  *corev1.PodSecurityContext    `json:"securityContext,omitempty"`
-	Annotations      map[string]string             `json:"annotations,omitempty"`
-	Storage          *RedisStorage                 `json:"storage,omitempty"`
-	Resources        *corev1.ResourceRequirements  `json:"resources,omitempty"`
-	PasswordSecret   *corev1.LocalObjectReference  `json:"passwordSecret,omitempty"`
-	Monitor          *AgentSpec                    `json:"monitor,omitempty"`
-	Init             *InitSpec                     `json:"init,omitempty"`
+	// Set RequiredAntiAffinity to force the master-slave node anti-affinity.
+	RequiredAntiAffinity bool                         `json:"requiredAntiAffinity,omitempty"`
+	Affinity             *corev1.Affinity             `json:"affinity,omitempty"`
+	NodeSelector         map[string]string            `json:"nodeSelector,omitempty"`
+	ToleRations          []corev1.Toleration          `json:"toleRations,omitempty"`
+	SecurityContext      *corev1.PodSecurityContext   `json:"securityContext,omitempty"`
+	Annotations          map[string]string            `json:"annotations,omitempty"`
+	Storage              *RedisStorage                `json:"storage,omitempty"`
+	Resources            *corev1.ResourceRequirements `json:"resources,omitempty"`
+	PasswordSecret       *corev1.LocalObjectReference `json:"passwordSecret,omitempty"`
+	Monitor              *AgentSpec                   `json:"monitor,omitempty"`
+	Init                 *InitSpec                    `json:"init,omitempty"`
 }
 
 type AgentSpec struct {

pkg/controller/manager/ensurer.go

@@ -119,7 +119,25 @@ func shouldUpdateRedis(cluster *redisv1alpha1.DistributedRedisCluster, sts *apps
 	if result := expectResource.Limits.Cpu().Cmp(*currentResource.Limits.Cpu()); result != 0 {
 		return true
 	}
-	return false
+	return monitorChanged(cluster, sts)
+}
+
+func monitorChanged(cluster *redisv1alpha1.DistributedRedisCluster, sts *appsv1.StatefulSet) bool {
+	if cluster.Spec.Monitor != nil {
+		for _, container := range sts.Spec.Template.Spec.Containers {
+			if container.Name == statefulsets.ExporterContainerName {
+				return false
+			}
+		}
+		return true
+	} else {
+		for _, container := range sts.Spec.Template.Spec.Containers {
+			if container.Name == statefulsets.ExporterContainerName {
+				return true
+			}
+		}
+		return false
+	}
 }
 
 func (r *realEnsureResource) ensureRedisPDB(cluster *redisv1alpha1.DistributedRedisCluster, name string, labels map[string]string) error {

pkg/controller/redisclusterbackup/sync_handler.go

@@ -267,15 +267,14 @@ func (r *ReconcileRedisClusterBackup) getBackupJob(reqLogger logr.Logger, backup
 		},
 	}
 	if backup.Spec.PodSpec != nil {
-		podSpec := job.Spec.Template.Spec
-		podSpec.NodeSelector = backup.Spec.PodSpec.NodeSelector
-		podSpec.Affinity = backup.Spec.PodSpec.Affinity
-		podSpec.SchedulerName = backup.Spec.PodSpec.SchedulerName
-		podSpec.Tolerations = backup.Spec.PodSpec.Tolerations
-		podSpec.PriorityClassName = backup.Spec.PodSpec.PriorityClassName
-		podSpec.Priority = backup.Spec.PodSpec.Priority
-		podSpec.SecurityContext = backup.Spec.PodSpec.SecurityContext
-		podSpec.ImagePullSecrets = backup.Spec.PodSpec.ImagePullSecrets
+		job.Spec.Template.Spec.NodeSelector = backup.Spec.PodSpec.NodeSelector
+		job.Spec.Template.Spec.Affinity = backup.Spec.PodSpec.Affinity
+		job.Spec.Template.Spec.SchedulerName = backup.Spec.PodSpec.SchedulerName
+		job.Spec.Template.Spec.Tolerations = backup.Spec.PodSpec.Tolerations
+		job.Spec.Template.Spec.PriorityClassName = backup.Spec.PodSpec.PriorityClassName
+		job.Spec.Template.Spec.Priority = backup.Spec.PodSpec.Priority
+		job.Spec.Template.Spec.SecurityContext = backup.Spec.PodSpec.SecurityContext
+		job.Spec.Template.Spec.ImagePullSecrets = backup.Spec.PodSpec.ImagePullSecrets
 	}
 	if backup.Spec.Backend.Local != nil {
 		job.Spec.Template.Spec.Volumes = append(job.Spec.Template.Spec.Volumes, corev1.Volume{

pkg/redisutil/admin.go

@@ -8,6 +8,8 @@ import (
 	"time"
 
 	"github.com/go-logr/logr"
+
+	"github.com/ucloud/redis-cluster-operator/pkg/utils"
 )
 
 const (
@@ -333,6 +335,24 @@ func (a *Admin) GetAllConfig(c IClient, addr string) (map[string]string, error)
 	return raw, nil
 }
 
+var parseConfigMap = map[string]int8{
+	"maxmemory":                  0,
+	"proto-max-bulk-len":         0,
+	"client-query-buffer-limit":  0,
+	"repl-backlog-size":          0,
+	"auto-aof-rewrite-min-size":  0,
+	"active-defrag-ignore-bytes": 0,
+	"hash-max-ziplist-entries":   0,
+	"hash-max-ziplist-value":     0,
+	"stream-node-max-bytes":      0,
+	"set-max-intset-entries":     0,
+	"zset-max-ziplist-entries":   0,
+	"zset-max-ziplist-value":     0,
+	"hll-sparse-max-bytes":       0,
+	// TODO parse client-output-buffer-limit
+	//"client-output-buffer-limit": 0,
+}
+
 // SetConfigIfNeed set redis config
 func (a *Admin) SetConfigIfNeed(newConfig map[string]string) error {
 	for addr, c := range a.Connections().GetAll() {
@@ -342,6 +362,14 @@ func (a *Admin) SetConfigIfNeed(newConfig map[string]string) error {
 		}
 
 		for key, value := range newConfig {
+			var err error
+			if _, ok := parseConfigMap[key]; ok {
+				value, err = utils.ParseRedisMemConf(value)
+				if err != nil {
+					a.log.Error(err, "redis config format err", "key", key, "value", value)
+					continue
+				}
+			}
 			if value != oldConfig[key] {
 				a.log.V(3).Info("CONFIG SET", key, value)
 				resp := c.Cmd("CONFIG", "SET", key, value)

pkg/resources/statefulsets/statefulset.go

@@ -24,6 +24,7 @@ const (
 	redisRestoreLocalVolumeName = "redis-local"
 	redisServerName             = "redis"
 	hostnameTopologyKey         = "kubernetes.io/hostname"
+	ExporterContainerName       = "exporter"
 
 	graceTime = 30
 
@@ -61,7 +62,7 @@ func NewStatefulSetForCR(cluster *redisv1alpha1.DistributedRedisCluster, ssName,
 				},
 				Spec: corev1.PodSpec{
 					ImagePullSecrets: cluster.Spec.ImagePullSecrets,
-					Affinity:         getAffinity(spec.Affinity, labels),
+					Affinity:         getAffinity(cluster, labels),
 					Tolerations:      spec.ToleRations,
 					SecurityContext:  spec.SecurityContext,
 					NodeSelector:     cluster.Spec.NodeSelector,
@@ -96,11 +97,37 @@ func NewStatefulSetForCR(cluster *redisv1alpha1.DistributedRedisCluster, ssName,
 	return ss, nil
 }
 
-func getAffinity(affinity *corev1.Affinity, labels map[string]string) *corev1.Affinity {
+func getAffinity(cluster *redisv1alpha1.DistributedRedisCluster, labels map[string]string) *corev1.Affinity {
+	affinity := cluster.Spec.Affinity
 	if affinity != nil {
 		return affinity
 	}
 
+	if cluster.Spec.RequiredAntiAffinity {
+		return &corev1.Affinity{
+			PodAntiAffinity: &corev1.PodAntiAffinity{
+				PreferredDuringSchedulingIgnoredDuringExecution: []corev1.WeightedPodAffinityTerm{
+					{
+						Weight: 100,
+						PodAffinityTerm: corev1.PodAffinityTerm{
+							TopologyKey: hostnameTopologyKey,
+							LabelSelector: &metav1.LabelSelector{
+								MatchLabels: map[string]string{redisv1alpha1.LabelClusterName: cluster.Name},
+							},
+						},
+					},
+				},
+				RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
+					{
+						TopologyKey: hostnameTopologyKey,
+						LabelSelector: &metav1.LabelSelector{
+							MatchLabels: labels,
+						},
+					},
+				},
+			},
+		}
+	}
 	// return a SOFT anti-affinity by default
 	return &corev1.Affinity{
 		PodAntiAffinity: &corev1.PodAntiAffinity{
@@ -110,7 +137,7 @@ func getAffinity(affinity *corev1.Affinity, labels map[string]string) *corev1.Af
 					PodAffinityTerm: corev1.PodAffinityTerm{
 						TopologyKey: hostnameTopologyKey,
 						LabelSelector: &metav1.LabelSelector{
-							MatchLabels: labels,
+							MatchLabels: map[string]string{redisv1alpha1.LabelClusterName: cluster.Name},
 						},
 					},
 				},
@@ -282,7 +309,7 @@ func redisServerContainer(cluster *redisv1alpha1.DistributedRedisCluster, passwo
 
 func redisExporterContainer(cluster *redisv1alpha1.DistributedRedisCluster, password *corev1.EnvVar) corev1.Container {
 	container := corev1.Container{
-		Name: "exporter",
+		Name: ExporterContainerName,
 		Args: append([]string{
 			fmt.Sprintf("--web.listen-address=:%v", cluster.Spec.Monitor.Prometheus.Port),
 			fmt.Sprintf("--web.telemetry-path=%v", redisv1alpha1.PrometheusExporterTelemetryPath),

pkg/utils/parse.go

@@ -0,0 +1,42 @@
+package utils
+
+import (
+	"strconv"
+	"strings"
+)
+
+func ParseRedisMemConf(p string) (string, error) {
+	var mul int64 = 1
+	u := strings.ToLower(p)
+	digits := u
+
+	if strings.HasSuffix(u, "k") {
+		digits = u[:len(u)-len("k")]
+		mul = 1000
+	} else if strings.HasSuffix(u, "kb") {
+		digits = u[:len(u)-len("kb")]
+		mul = 1024
+	} else if strings.HasSuffix(u, "m") {
+		digits = u[:len(u)-len("m")]
+		mul = 1000 * 1000
+	} else if strings.HasSuffix(u, "mb") {
+		digits = u[:len(u)-len("mb")]
+		mul = 1024 * 1024
+	} else if strings.HasSuffix(u, "g") {
+		digits = u[:len(u)-len("g")]
+		mul = 1000 * 1000 * 1000
+	} else if strings.HasSuffix(u, "gb") {
+		digits = u[:len(u)-len("gb")]
+		mul = 1024 * 1024 * 1024
+	} else if strings.HasSuffix(u, "b") {
+		digits = u[:len(u)-len("b")]
+		mul = 1
+	}
+
+	val, err := strconv.ParseInt(digits, 10, 64)
+	if err != nil {
+		return "", err
+	}
+
+	return strconv.FormatInt(val*mul, 10), nil
+}