14173979426

Committed 31 Mar 2025 02:26PM UTC coverage: 10.247% (-3.3%) from 13.534%

Build # 14173979426

Build Type

push

github

Committed by

web-flow

Commit Message

feat: added webhooks

Run Details

168 of 2201 new or added lines in 67 files covered. (7.63%)

1254 existing lines in 61 files now uncovered.

2074 of 20241 relevant lines covered (10.25%)

0.12 hits per line

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

43.9

/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/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/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/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); errors.IsNotFound(err) {
                return ctrl.Result{}, nil
        } else if err != nil {
                logger.Error(err, "get resource failed")
                return ctrl.Result{}, err
        }

        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/api/errors"
28	"k8s.io/apimachinery/pkg/runtime"
29	"k8s.io/client-go/tools/record"
30	ctrl "sigs.k8s.io/controller-runtime"
31	"sigs.k8s.io/controller-runtime/pkg/client"
32	"sigs.k8s.io/controller-runtime/pkg/controller"
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/sentinelbuilder"
39	"github.com/chideat/valkey-operator/internal/config"
40	"github.com/chideat/valkey-operator/internal/ops"
41	)
42
43	// FailoverReconciler reconciles a Failover object
44	type FailoverReconciler struct {
45	client.Client
46	Scheme *runtime.Scheme
47	EventRecorder record.EventRecorder
48	Engine *ops.OpEngine
49	}
50
51	// +kubebuilder:rbac:groups=valkey.buf.red,resources=failovers,verbs=get;list;watch;create;update;patch;delete
52	// +kubebuilder:rbac:groups=valkey.buf.red,resources=failovers/status,verbs=get;update;patch
53	// +kubebuilder:rbac:groups=valkey.buf.red,resources=failovers/finalizers,verbs=update
54
55	// Reconcile is part of the main kubernetes reconciliation loop which aims to
56	// move the current state of the cluster closer to the desired state.
57	func (r *FailoverReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {	1✔
58	logger := log.FromContext(ctx).WithValues("target", req.String())	1✔
59		1✔
60	var instance v1alpha1.Failover	1✔
61	if err := r.Get(ctx, req.NamespacedName, &instance); errors.IsNotFound(err) {	1✔
62	return ctrl.Result{}, nil	×
63	} else if err != nil {	1✔
64	logger.Error(err, "get resource failed")	×
65	return ctrl.Result{}, err	×
66	}	×
67
68	if crVersion := instance.Annotations[builder.CRVersionKey]; crVersion == "" {	1✔
UNCOV 69	if config.GetOperatorVersion() != "" {	×
UNCOV 70	if instance.Annotations == nil {	×
UNCOV 71	instance.Annotations = make(map[string]string)	×
UNCOV 72	}	×
NEW 73	instance.Annotations[builder.CRVersionKey] = config.GetOperatorVersion()	×
UNCOV 74	if err := r.Client.Update(ctx, &instance); err != nil {	×
75	logger.Error(err, "update instance actor version failed")	×
76	}	×
UNCOV 77	return ctrl.Result{RequeueAfter: time.Second}, nil	×
78	}
79	}
80
81	{	1✔
82	var (	1✔
83	nodes []v1alpha1.SentinelMonitorNode	1✔
84	serviceName = sentinelbuilder.SentinelHeadlessServiceName(instance.GetName())	1✔
85	status = &instance.Status	1✔
86	oldStatus = instance.Status.DeepCopy()	1✔
87	)	1✔
88	if instance.Spec.Sentinel == nil {	1✔
89	status.Monitor = v1alpha1.MonitorStatus{	×
90	Policy: v1alpha1.ManualFailoverPolicy,	×
91	}	×
92	} else {	1✔
93	// TODO: use DNS SRV replace config all sentinel node address, which will cause data pods restart	1✔
94	for i := range instance.Spec.Sentinel.Replicas {	2✔
95	podName := sentinelbuilder.SentinelPodServiceName(instance.GetName(), int(i))	1✔
96	srv := fmt.Sprintf("%s.%s.%s", podName, serviceName, instance.GetNamespace())	1✔
97	nodes = append(nodes, v1alpha1.SentinelMonitorNode{IP: srv, Port: 26379})	1✔
98	}	1✔
99
100	status.Monitor.Policy = v1alpha1.SentinelFailoverPolicy	1✔
101	if instance.Spec.Sentinel.SentinelReference == nil {	2✔
102	// HARDCODE: use mymaster as sentinel monitor name	1✔
103	status.Monitor.Name = "mymaster"	1✔
104		1✔
105	// append history password secrets	1✔
106	// NOTE: here recorded empty password	1✔
107	passwordSecret := instance.Spec.Sentinel.Access.DefaultPasswordSecret	1✔
108	if status.Monitor.PasswordSecret != passwordSecret {	1✔
109	status.Monitor.OldPasswordSecret = status.Monitor.PasswordSecret	×
110	status.Monitor.PasswordSecret = passwordSecret	×
111	}	×
112
113	if instance.Spec.Sentinel.Access.EnableTLS {	1✔
114	if instance.Spec.Sentinel.Access.ExternalTLSSecret != "" {	×
115	status.Monitor.TLSSecret = instance.Spec.Sentinel.Access.ExternalTLSSecret	×
116	} else {	×
117	status.Monitor.TLSSecret = certbuilder.GenerateSSLSecretName(instance.GetName())	×
118	}	×
119	}
120	status.Monitor.Nodes = nodes	1✔
121	} else {	×
122	status.Monitor.Name = fmt.Sprintf("%s.%s", instance.GetNamespace(), instance.GetName())	×
123	status.Monitor.OldPasswordSecret = status.Monitor.PasswordSecret	×
124	status.Monitor.PasswordSecret = instance.Spec.Sentinel.SentinelReference.Auth.PasswordSecret	×
125	status.Monitor.TLSSecret = instance.Spec.Sentinel.SentinelReference.Auth.TLSSecret	×
126	status.Monitor.Nodes = instance.Spec.Sentinel.SentinelReference.Nodes	×
127	}	×
128	}
129	if !reflect.DeepEqual(status, oldStatus) {	2✔
130	if err := r.Status().Update(ctx, &instance); err != nil {	1✔
131	logger.Error(err, "update status failed")	×
132	return ctrl.Result{}, err	×
133	}	×
134	return ctrl.Result{RequeueAfter: time.Second}, nil	1✔
135	}
136	}
137	return r.Engine.Run(ctx, &instance)	×
138	}
139
140	// SetupWithManager sets up the controller with the Manager.
141	func (r *FailoverReconciler) SetupWithManager(mgr ctrl.Manager) error {	×
142	return ctrl.NewControllerManagedBy(mgr).	×
143	For(&v1alpha1.Failover{}).	×
144	WithOptions(controller.Options{MaxConcurrentReconciles: 8}).	×
145	Owns(&v1alpha1.Sentinel{}).	×
146	Owns(&appsv1.StatefulSet{}).	×
147	Owns(&corev1.Service{}).	×
148	Owns(&corev1.ConfigMap{}).	×
149	Owns(&corev1.Secret{}).	×
150	Complete(r)	×
151	}	×

chideat / valkey-operator / 14173979426

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