17861088256

Committed 19 Sep 2025 02:22PM UTC coverage: 18.853% (-0.5%) from 19.354%

Build # 17861088256

Build Type

Pull #59

github

Committed by

chideat

Commit Message

chore: update release pipeline and dependencies

Pull Request Pull Request #59: Enhanced Stability, Pause Logic, and Sentinel Improvements

Run Details

88 of 1118 new or added lines in 42 files covered. (7.87%)

62 existing lines in 12 files now uncovered.

3881 of 20586 relevant lines covered (18.85%)

0.22 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

8.33

/internal/controller/failover_controller.go

/*
Copyright 2024 chideat.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package controller

import (
        "context"
        "fmt"
        "reflect"
        "time"

        appsv1 "k8s.io/api/apps/v1"
        corev1 "k8s.io/api/core/v1"
        "k8s.io/apimachinery/pkg/runtime"
        "k8s.io/client-go/tools/record"
        ctrl "sigs.k8s.io/controller-runtime"
        "sigs.k8s.io/controller-runtime/pkg/client"
        "sigs.k8s.io/controller-runtime/pkg/controller"
        "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
        "sigs.k8s.io/controller-runtime/pkg/log"

        "github.com/chideat/valkey-operator/api/v1alpha1"
        "github.com/chideat/valkey-operator/internal/builder"
        "github.com/chideat/valkey-operator/internal/builder/certbuilder"
        "github.com/chideat/valkey-operator/internal/builder/failoverbuilder"
        "github.com/chideat/valkey-operator/internal/builder/sentinelbuilder"
        "github.com/chideat/valkey-operator/internal/config"
        "github.com/chideat/valkey-operator/internal/ops"
)

// FailoverReconciler reconciles a Failover object
type FailoverReconciler struct {
        client.Client
        Scheme        *runtime.Scheme
        EventRecorder record.EventRecorder
        Engine        *ops.OpEngine
}

// +kubebuilder:rbac:groups=valkey.buf.red,resources=failovers,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=valkey.buf.red,resources=failovers/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=valkey.buf.red,resources=failovers/finalizers,verbs=update

// Reconcile is part of the main kubernetes reconciliation loop which aims to
// move the current state of the cluster closer to the desired state.
func (r *FailoverReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
        logger := log.FromContext(ctx).WithValues("target", req.String())

        var instance v1alpha1.Failover
        if err := r.Get(ctx, req.NamespacedName, &instance); err != nil {
                logger.Error(err, "get resource failed")
                return ctrl.Result{}, client.IgnoreNotFound(err)
        } else if instance.GetDeletionTimestamp() != nil {
                if controllerutil.ContainsFinalizer(&instance, builder.ResourceCleanFinalizer) {
                        // check if sts is marked for deletion
                        labels := failoverbuilder.GenerateCommonLabels(instance.GetName())
                        stsList := appsv1.StatefulSetList{}
                        if err := r.List(ctx, &stsList, client.InNamespace(instance.Namespace), client.MatchingLabels(labels)); err != nil {
                                logger.Error(err, "get failover statefulsets failed", "labels", labels)
                                return ctrl.Result{}, err
                        }

                        needRequeue := false
                        for _, sts := range stsList.Items {
                                if sts.GetDeletionTimestamp() == nil {
                                        if err := r.Delete(ctx, &sts); err != nil {
                                                logger.Error(err, "delete failover statefulset failed", "name", sts.GetName())
                                        }
                                        needRequeue = true
                                }
                        }
                        if needRequeue {
                                return ctrl.Result{RequeueAfter: time.Second * 5}, nil
                        }

                        // check if all pods is shutdown
                        podList := corev1.PodList{}
                        if err := r.List(ctx, &podList, client.InNamespace(instance.Namespace), client.MatchingLabels(labels)); err != nil {
                                logger.Error(err, "list pods failed", "namespace", instance.Namespace, "labels", labels)
                                return ctrl.Result{}, err
                        }
                        if len(podList.Items) > 0 {
                                logger.Info("failover pods is not shutdown, waiting for next reconcile", "pods", len(podList.Items))
                                return ctrl.Result{RequeueAfter: time.Second * 10}, nil
                        }

                        logger.Info("instance is deleting, remove finalizer", "name", instance.GetName())
                        controllerutil.RemoveFinalizer(&instance, builder.ResourceCleanFinalizer)
                        if err := r.Update(ctx, &instance); err != nil {
                                logger.Error(err, "update instance finalizer failed")
                                return ctrl.Result{}, err
                        }
                }
        }

        if !controllerutil.ContainsFinalizer(&instance, builder.ResourceCleanFinalizer) {
                controllerutil.AddFinalizer(&instance, builder.ResourceCleanFinalizer)
                if err := r.Update(ctx, &instance); err != nil {
                        logger.Error(err, "add finalizer failed", "instance", instance.GetName())
                        return ctrl.Result{}, err
                }
                return ctrl.Result{RequeueAfter: time.Second}, nil
        }

        if crVersion := instance.Annotations[builder.CRVersionKey]; crVersion == "" {
                if config.GetOperatorVersion() != "" {
                        if instance.Annotations == nil {
                                instance.Annotations = make(map[string]string)
                        }
                        instance.Annotations[builder.CRVersionKey] = config.GetOperatorVersion()
                        if err := r.Client.Update(ctx, &instance); err != nil {
                                logger.Error(err, "update instance actor version failed")
                        }
                        return ctrl.Result{RequeueAfter: time.Second}, nil
                }
        }

        {
                var (
                        nodes       []v1alpha1.SentinelMonitorNode
                        serviceName = sentinelbuilder.SentinelHeadlessServiceName(instance.GetName())
                        status      = &instance.Status
                        oldStatus   = instance.Status.DeepCopy()
                )
                if instance.Spec.Sentinel == nil {
                        status.Monitor = v1alpha1.MonitorStatus{
                                Policy: v1alpha1.ManualFailoverPolicy,
                        }
                } else {
                        // TODO: use DNS SRV replace config all sentinel node address, which will cause data pods restart
                        for i := range instance.Spec.Sentinel.Replicas {
                                podName := sentinelbuilder.SentinelPodServiceName(instance.GetName(), int(i))
                                srv := fmt.Sprintf("%s.%s.%s", podName, serviceName, instance.GetNamespace())
                                nodes = append(nodes, v1alpha1.SentinelMonitorNode{IP: srv, Port: 26379})
                        }

                        status.Monitor.Policy = v1alpha1.SentinelFailoverPolicy
                        if instance.Spec.Sentinel.SentinelReference == nil {
                                // HARDCODE: use mymaster as sentinel monitor name
                                status.Monitor.Name = "mymaster"

                                // append history password secrets
                                // NOTE: here recorded empty password
                                passwordSecret := instance.Spec.Sentinel.Access.DefaultPasswordSecret
                                if status.Monitor.PasswordSecret != passwordSecret {
                                        status.Monitor.OldPasswordSecret = status.Monitor.PasswordSecret
                                        status.Monitor.PasswordSecret = passwordSecret
                                }

                                if instance.Spec.Sentinel.Access.EnableTLS {
                                        if instance.Spec.Sentinel.Access.ExternalTLSSecret != "" {
                                                status.Monitor.TLSSecret = instance.Spec.Sentinel.Access.ExternalTLSSecret
                                        } else {
                                                status.Monitor.TLSSecret = certbuilder.GenerateSSLSecretName(instance.GetName())
                                        }
                                }
                                status.Monitor.Nodes = nodes
                        } else {
                                status.Monitor.Name = fmt.Sprintf("%s.%s", instance.GetNamespace(), instance.GetName())
                                status.Monitor.OldPasswordSecret = status.Monitor.PasswordSecret
                                status.Monitor.PasswordSecret = instance.Spec.Sentinel.SentinelReference.Auth.PasswordSecret
                                status.Monitor.TLSSecret = instance.Spec.Sentinel.SentinelReference.Auth.TLSSecret
                                status.Monitor.Nodes = instance.Spec.Sentinel.SentinelReference.Nodes
                        }
                }
                if !reflect.DeepEqual(status, oldStatus) {
                        if err := r.Status().Update(ctx, &instance); err != nil {
                                logger.Error(err, "update status failed")
                                return ctrl.Result{}, err
                        }
                        return ctrl.Result{RequeueAfter: time.Second}, nil
                }
        }
        return r.Engine.Run(ctx, &instance)
}

// SetupWithManager sets up the controller with the Manager.
func (r *FailoverReconciler) SetupWithManager(mgr ctrl.Manager) error {
        return ctrl.NewControllerManagedBy(mgr).
                For(&v1alpha1.Failover{}).
                WithOptions(controller.Options{MaxConcurrentReconciles: 8}).
                Owns(&v1alpha1.Sentinel{}).
                Owns(&appsv1.StatefulSet{}).
                Owns(&corev1.Service{}).
                Owns(&corev1.ConfigMap{}).
                Owns(&corev1.Secret{}).
                Complete(r)
}

1	/*
2	Copyright 2024 chideat.
3
4	Licensed under the Apache License, Version 2.0 (the "License");
5	you may not use this file except in compliance with the License.
6	You may obtain a copy of the License at
7
8	http://www.apache.org/licenses/LICENSE-2.0
9
10	Unless required by applicable law or agreed to in writing, software
11	distributed under the License is distributed on an "AS IS" BASIS,
12	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	See the License for the specific language governing permissions and
14	limitations under the License.
15	*/
16
17	package controller
18
19	import (
20	"context"
21	"fmt"
22	"reflect"
23	"time"
24
25	appsv1 "k8s.io/api/apps/v1"
26	corev1 "k8s.io/api/core/v1"
27	"k8s.io/apimachinery/pkg/runtime"
28	"k8s.io/client-go/tools/record"
29	ctrl "sigs.k8s.io/controller-runtime"
30	"sigs.k8s.io/controller-runtime/pkg/client"
31	"sigs.k8s.io/controller-runtime/pkg/controller"
32	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
33	"sigs.k8s.io/controller-runtime/pkg/log"
34
35	"github.com/chideat/valkey-operator/api/v1alpha1"
36	"github.com/chideat/valkey-operator/internal/builder"
37	"github.com/chideat/valkey-operator/internal/builder/certbuilder"
38	"github.com/chideat/valkey-operator/internal/builder/failoverbuilder"
39	"github.com/chideat/valkey-operator/internal/builder/sentinelbuilder"
40	"github.com/chideat/valkey-operator/internal/config"
41	"github.com/chideat/valkey-operator/internal/ops"
42	)
43
44	// FailoverReconciler reconciles a Failover object
45	type FailoverReconciler struct {
46	client.Client
47	Scheme *runtime.Scheme
48	EventRecorder record.EventRecorder
49	Engine *ops.OpEngine
50	}
51
52	// +kubebuilder:rbac:groups=valkey.buf.red,resources=failovers,verbs=get;list;watch;create;update;patch;delete
53	// +kubebuilder:rbac:groups=valkey.buf.red,resources=failovers/status,verbs=get;update;patch
54	// +kubebuilder:rbac:groups=valkey.buf.red,resources=failovers/finalizers,verbs=update
55
56	// Reconcile is part of the main kubernetes reconciliation loop which aims to
57	// move the current state of the cluster closer to the desired state.
58	func (r *FailoverReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {	1✔
59	logger := log.FromContext(ctx).WithValues("target", req.String())	1✔
60		1✔
61	var instance v1alpha1.Failover	1✔
62	if err := r.Get(ctx, req.NamespacedName, &instance); err != nil {	1✔
63	logger.Error(err, "get resource failed")	×
NEW 64	return ctrl.Result{}, client.IgnoreNotFound(err)	×
65	} else if instance.GetDeletionTimestamp() != nil {	1✔
NEW 66	if controllerutil.ContainsFinalizer(&instance, builder.ResourceCleanFinalizer) {	×
NEW 67	// check if sts is marked for deletion	×
NEW 68	labels := failoverbuilder.GenerateCommonLabels(instance.GetName())	×
NEW 69	stsList := appsv1.StatefulSetList{}	×
NEW 70	if err := r.List(ctx, &stsList, client.InNamespace(instance.Namespace), client.MatchingLabels(labels)); err != nil {	×
NEW 71	logger.Error(err, "get failover statefulsets failed", "labels", labels)	×
NEW 72	return ctrl.Result{}, err	×
NEW 73	}	×
74
NEW 75	needRequeue := false	×
NEW 76	for _, sts := range stsList.Items {	×
NEW 77	if sts.GetDeletionTimestamp() == nil {	×
NEW 78	if err := r.Delete(ctx, &sts); err != nil {	×
NEW 79	logger.Error(err, "delete failover statefulset failed", "name", sts.GetName())	×
NEW 80	}	×
NEW 81	needRequeue = true	×
82	}
83	}
NEW 84	if needRequeue {	×
NEW 85	return ctrl.Result{RequeueAfter: time.Second * 5}, nil	×
NEW 86	}	×
87
88	// check if all pods is shutdown
NEW 89	podList := corev1.PodList{}	×
NEW 90	if err := r.List(ctx, &podList, client.InNamespace(instance.Namespace), client.MatchingLabels(labels)); err != nil {	×
NEW 91	logger.Error(err, "list pods failed", "namespace", instance.Namespace, "labels", labels)	×
NEW 92	return ctrl.Result{}, err	×
NEW 93	}	×
NEW 94	if len(podList.Items) > 0 {	×
NEW 95	logger.Info("failover pods is not shutdown, waiting for next reconcile", "pods", len(podList.Items))	×
NEW 96	return ctrl.Result{RequeueAfter: time.Second * 10}, nil	×
NEW 97	}	×
98
NEW 99	logger.Info("instance is deleting, remove finalizer", "name", instance.GetName())	×
NEW 100	controllerutil.RemoveFinalizer(&instance, builder.ResourceCleanFinalizer)	×
NEW 101	if err := r.Update(ctx, &instance); err != nil {	×
NEW 102	logger.Error(err, "update instance finalizer failed")	×
NEW 103	return ctrl.Result{}, err	×
NEW 104	}	×
105	}
106	}
107
108	if !controllerutil.ContainsFinalizer(&instance, builder.ResourceCleanFinalizer) {	2✔
109	controllerutil.AddFinalizer(&instance, builder.ResourceCleanFinalizer)	1✔
110	if err := r.Update(ctx, &instance); err != nil {	1✔
NEW 111	logger.Error(err, "add finalizer failed", "instance", instance.GetName())	×
NEW 112	return ctrl.Result{}, err	×
NEW 113	}	×
114	return ctrl.Result{RequeueAfter: time.Second}, nil	1✔
115	}
116
UNCOV 117	if crVersion := instance.Annotations[builder.CRVersionKey]; crVersion == "" {	×
118	if config.GetOperatorVersion() != "" {	×
119	if instance.Annotations == nil {	×
120	instance.Annotations = make(map[string]string)	×
121	}	×
122	instance.Annotations[builder.CRVersionKey] = config.GetOperatorVersion()	×
123	if err := r.Client.Update(ctx, &instance); err != nil {	×
124	logger.Error(err, "update instance actor version failed")	×
125	}	×
126	return ctrl.Result{RequeueAfter: time.Second}, nil	×
127	}
128	}
129
UNCOV 130	{	×
UNCOV 131	var (	×
UNCOV 132	nodes []v1alpha1.SentinelMonitorNode	×
UNCOV 133	serviceName = sentinelbuilder.SentinelHeadlessServiceName(instance.GetName())	×
UNCOV 134	status = &instance.Status	×
UNCOV 135	oldStatus = instance.Status.DeepCopy()	×
UNCOV 136	)	×
UNCOV 137	if instance.Spec.Sentinel == nil {	×
138	status.Monitor = v1alpha1.MonitorStatus{	×
139	Policy: v1alpha1.ManualFailoverPolicy,	×
140	}	×
UNCOV 141	} else {	×
UNCOV 142	// TODO: use DNS SRV replace config all sentinel node address, which will cause data pods restart	×
UNCOV 143	for i := range instance.Spec.Sentinel.Replicas {	×
UNCOV 144	podName := sentinelbuilder.SentinelPodServiceName(instance.GetName(), int(i))	×
UNCOV 145	srv := fmt.Sprintf("%s.%s.%s", podName, serviceName, instance.GetNamespace())	×
UNCOV 146	nodes = append(nodes, v1alpha1.SentinelMonitorNode{IP: srv, Port: 26379})	×
UNCOV 147	}	×
148
UNCOV 149	status.Monitor.Policy = v1alpha1.SentinelFailoverPolicy	×
UNCOV 150	if instance.Spec.Sentinel.SentinelReference == nil {	×
UNCOV 151	// HARDCODE: use mymaster as sentinel monitor name	×
UNCOV 152	status.Monitor.Name = "mymaster"	×
UNCOV 153		×
UNCOV 154	// append history password secrets	×
UNCOV 155	// NOTE: here recorded empty password	×
UNCOV 156	passwordSecret := instance.Spec.Sentinel.Access.DefaultPasswordSecret	×
UNCOV 157	if status.Monitor.PasswordSecret != passwordSecret {	×
158	status.Monitor.OldPasswordSecret = status.Monitor.PasswordSecret	×
159	status.Monitor.PasswordSecret = passwordSecret	×
160	}	×
161
UNCOV 162	if instance.Spec.Sentinel.Access.EnableTLS {	×
163	if instance.Spec.Sentinel.Access.ExternalTLSSecret != "" {	×
164	status.Monitor.TLSSecret = instance.Spec.Sentinel.Access.ExternalTLSSecret	×
165	} else {	×
166	status.Monitor.TLSSecret = certbuilder.GenerateSSLSecretName(instance.GetName())	×
167	}	×
168	}
UNCOV 169	status.Monitor.Nodes = nodes	×
170	} else {	×
171	status.Monitor.Name = fmt.Sprintf("%s.%s", instance.GetNamespace(), instance.GetName())	×
172	status.Monitor.OldPasswordSecret = status.Monitor.PasswordSecret	×
173	status.Monitor.PasswordSecret = instance.Spec.Sentinel.SentinelReference.Auth.PasswordSecret	×
174	status.Monitor.TLSSecret = instance.Spec.Sentinel.SentinelReference.Auth.TLSSecret	×
175	status.Monitor.Nodes = instance.Spec.Sentinel.SentinelReference.Nodes	×
176	}	×
177	}
UNCOV 178	if !reflect.DeepEqual(status, oldStatus) {	×
UNCOV 179	if err := r.Status().Update(ctx, &instance); err != nil {	×
180	logger.Error(err, "update status failed")	×
181	return ctrl.Result{}, err	×
182	}	×
UNCOV 183	return ctrl.Result{RequeueAfter: time.Second}, nil	×
184	}
185	}
186	return r.Engine.Run(ctx, &instance)	×
187	}
188
189	// SetupWithManager sets up the controller with the Manager.
190	func (r *FailoverReconciler) SetupWithManager(mgr ctrl.Manager) error {	×
191	return ctrl.NewControllerManagedBy(mgr).	×
192	For(&v1alpha1.Failover{}).	×
193	WithOptions(controller.Options{MaxConcurrentReconciles: 8}).	×
194	Owns(&v1alpha1.Sentinel{}).	×
195	Owns(&appsv1.StatefulSet{}).	×
196	Owns(&corev1.Service{}).	×
197	Owns(&corev1.ConfigMap{}).	×
198	Owns(&corev1.Secret{}).	×
199	Complete(r)	×
200	}	×

chideat / valkey-operator / 17861088256

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous