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

13.41

/internal/controller/cluster_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"
        "time"

        appsv1 "k8s.io/api/apps/v1"
        corev1 "k8s.io/api/core/v1"
        "k8s.io/apimachinery/pkg/api/errors"
        "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/clusterbuilder"
        "github.com/chideat/valkey-operator/internal/config"
        "github.com/chideat/valkey-operator/internal/ops"
)

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

// +kubebuilder:rbac:groups=valkey.buf.red,resources=clusters,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=valkey.buf.red,resources=clusters/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=valkey.buf.red,resources=clusters/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 *ClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
        logger := log.FromContext(ctx).WithValues("target", req.String())

        var instance v1alpha1.Cluster
        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 all pods is shutdown
                        labels := clusterbuilder.GenerateClusterLabels(instance.GetName(), nil)
                        stsList := appsv1.StatefulSetList{}
                        if err := r.List(ctx, &stsList, client.InNamespace(instance.Namespace), client.MatchingLabels(labels)); err != nil {
                                logger.Error(err, "get cluster 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 cluster statefulset failed", "name", sts.GetName())
                                        }
                                        needRequeue = true
                                }
                        }
                        if needRequeue {
                                return ctrl.Result{RequeueAfter: time.Second * 5}, nil
                        }

                        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
                        } else if len(podList.Items) > 0 {
                                // still has pods running, wait for them to shutdown
                                logger.Info("instance is deleting, but still has pods running, wait for them to shutdown")
                                return ctrl.Result{RequeueAfter: time.Second * 5}, nil
                        }

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

        if !controllerutil.ContainsFinalizer(&instance, builder.ResourceCleanFinalizer) {
                // add finalizer
                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
        }

        // update default status
        if instance.Status.Phase == "" {
                // update status to creating
                r.EventRecorder.Eventf(&instance, corev1.EventTypeNormal, "Creating", "new instance request")

                instance.Status.Phase = v1alpha1.ClusterPhaseCreating
                if err := r.Status().Update(ctx, &instance); err != nil {
                        logger.Error(err, "update instance status failed")
                        if errors.IsNotFound(err) {
                                return ctrl.Result{}, nil
                        }
                        return ctrl.Result{}, err
                }
        }

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

        // ================ setup default ===================

        // _ = instance.Default()

        // ================ setup default end ===================

        return r.Engine.Run(ctx, &instance)
}

// SetupWithManager sets up the controller with the Manager.
func (r *ClusterReconciler) SetupWithManager(mgr ctrl.Manager) error {
        return ctrl.NewControllerManagedBy(mgr).
                For(&v1alpha1.Cluster{}).
                WithOptions(controller.Options{MaxConcurrentReconciles: 8}).
                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	"time"
22
23	appsv1 "k8s.io/api/apps/v1"
24	corev1 "k8s.io/api/core/v1"
25	"k8s.io/apimachinery/pkg/api/errors"
26	"k8s.io/apimachinery/pkg/runtime"
27	"k8s.io/client-go/tools/record"
28	ctrl "sigs.k8s.io/controller-runtime"
29	"sigs.k8s.io/controller-runtime/pkg/client"
30	"sigs.k8s.io/controller-runtime/pkg/controller"
31	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
32	"sigs.k8s.io/controller-runtime/pkg/log"
33
34	"github.com/chideat/valkey-operator/api/v1alpha1"
35	"github.com/chideat/valkey-operator/internal/builder"
36	"github.com/chideat/valkey-operator/internal/builder/clusterbuilder"
37	"github.com/chideat/valkey-operator/internal/config"
38	"github.com/chideat/valkey-operator/internal/ops"
39	)
40
41	// ClusterReconciler reconciles a Cluster object
42	type ClusterReconciler struct {
43	client.Client
44	Scheme *runtime.Scheme
45	EventRecorder record.EventRecorder
46	Engine *ops.OpEngine
47	}
48
49	// +kubebuilder:rbac:groups=valkey.buf.red,resources=clusters,verbs=get;list;watch;create;update;patch;delete
50	// +kubebuilder:rbac:groups=valkey.buf.red,resources=clusters/status,verbs=get;update;patch
51	// +kubebuilder:rbac:groups=valkey.buf.red,resources=clusters/finalizers,verbs=update
52
53	// Reconcile is part of the main kubernetes reconciliation loop which aims to
54	// move the current state of the cluster closer to the desired state.
55	func (r *ClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {	1✔
56	logger := log.FromContext(ctx).WithValues("target", req.String())	1✔
57		1✔
58	var instance v1alpha1.Cluster	1✔
59	if err := r.Get(ctx, req.NamespacedName, &instance); err != nil {	1✔
60	logger.Error(err, "get resource failed")	×
NEW 61	return ctrl.Result{}, client.IgnoreNotFound(err)	×
62	} else if instance.GetDeletionTimestamp() != nil {	1✔
NEW 63	if controllerutil.ContainsFinalizer(&instance, builder.ResourceCleanFinalizer) {	×
NEW 64	// check if all pods is shutdown	×
NEW 65	labels := clusterbuilder.GenerateClusterLabels(instance.GetName(), nil)	×
NEW 66	stsList := appsv1.StatefulSetList{}	×
NEW 67	if err := r.List(ctx, &stsList, client.InNamespace(instance.Namespace), client.MatchingLabels(labels)); err != nil {	×
NEW 68	logger.Error(err, "get cluster statefulsets failed", "labels", labels)	×
NEW 69	return ctrl.Result{}, err	×
NEW 70	}	×
71
NEW 72	needRequeue := false	×
NEW 73	for _, sts := range stsList.Items {	×
NEW 74	if sts.GetDeletionTimestamp() == nil {	×
NEW 75	if err := r.Delete(ctx, &sts); err != nil {	×
NEW 76	logger.Error(err, "delete cluster statefulset failed", "name", sts.GetName())	×
NEW 77	}	×
NEW 78	needRequeue = true	×
79	}
80	}
NEW 81	if needRequeue {	×
NEW 82	return ctrl.Result{RequeueAfter: time.Second * 5}, nil	×
NEW 83	}	×
84
NEW 85	podList := corev1.PodList{}	×
NEW 86	if err := r.List(ctx, &podList, client.InNamespace(instance.Namespace), client.MatchingLabels(labels)); err != nil {	×
NEW 87	logger.Error(err, "list pods failed", "namespace", instance.Namespace, "labels", labels)	×
NEW 88	return ctrl.Result{}, err	×
NEW 89	} else if len(podList.Items) > 0 {	×
NEW 90	// still has pods running, wait for them to shutdown	×
NEW 91	logger.Info("instance is deleting, but still has pods running, wait for them to shutdown")	×
NEW 92	return ctrl.Result{RequeueAfter: time.Second * 5}, nil	×
NEW 93	}	×
94
95	// all pods is shutdown, remove finalizer
NEW 96	controllerutil.RemoveFinalizer(&instance, builder.ResourceCleanFinalizer)	×
NEW 97	if err := r.Update(ctx, &instance); err != nil {	×
NEW 98	logger.Error(err, "remove finalizer failed", "instance", instance.GetName())	×
NEW 99	return ctrl.Result{}, err	×
NEW 100	}	×
101	}
NEW 102	return ctrl.Result{}, nil	×
103	}
104
105	if !controllerutil.ContainsFinalizer(&instance, builder.ResourceCleanFinalizer) {	2✔
106	// add finalizer	1✔
107	controllerutil.AddFinalizer(&instance, builder.ResourceCleanFinalizer)	1✔
108	if err := r.Update(ctx, &instance); err != nil {	1✔
NEW 109	logger.Error(err, "add finalizer failed", "instance", instance.GetName())	×
NEW 110	return ctrl.Result{}, err	×
NEW 111	}	×
112	return ctrl.Result{RequeueAfter: time.Second}, nil	1✔
113	}
114
115	// update default status
UNCOV 116	if instance.Status.Phase == "" {	×
UNCOV 117	// update status to creating	×
UNCOV 118	r.EventRecorder.Eventf(&instance, corev1.EventTypeNormal, "Creating", "new instance request")	×
UNCOV 119		×
UNCOV 120	instance.Status.Phase = v1alpha1.ClusterPhaseCreating	×
UNCOV 121	if err := r.Status().Update(ctx, &instance); err != nil {	×
122	logger.Error(err, "update instance status failed")	×
123	if errors.IsNotFound(err) {	×
124	return ctrl.Result{}, nil	×
125	}	×
126	return ctrl.Result{}, err	×
127	}
128	}
129
UNCOV 130	crVersion := instance.Annotations[builder.CRVersionKey]	×
UNCOV 131	if crVersion == "" {	×
132	if config.GetOperatorVersion() != "" {	×
133	if instance.Annotations == nil {	×
134	instance.Annotations = make(map[string]string)	×
135	}	×
136	// update crVersion to instance
137	instance.Annotations[builder.CRVersionKey] = config.GetOperatorVersion()	×
138	if err := r.Update(ctx, &instance); err != nil {	×
139	logger.Error(err, "update instance actor version failed")	×
140	}	×
141	return ctrl.Result{RequeueAfter: time.Second}, nil	×
142	}
143	}
144
145	// ================ setup default ===================
146
147	// _ = instance.Default()
148
149	// ================ setup default end ===================
150
UNCOV 151	return r.Engine.Run(ctx, &instance)	×
152	}
153
154	// SetupWithManager sets up the controller with the Manager.
155	func (r *ClusterReconciler) SetupWithManager(mgr ctrl.Manager) error {	×
156	return ctrl.NewControllerManagedBy(mgr).	×
157	For(&v1alpha1.Cluster{}).	×
158	WithOptions(controller.Options{MaxConcurrentReconciles: 8}).	×
159	Owns(&appsv1.StatefulSet{}).	×
160	Owns(&corev1.Service{}).	×
161	Owns(&corev1.ConfigMap{}).	×
162	Owns(&corev1.Secret{}).	×
163	Complete(r)	×
164	}	×

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