13214360144

Committed 08 Feb 2025 09:25AM UTC coverage: 13.534%. First build

Build # 13214360144

Build Type

push

github

Committed by

web-flow

Commit Message

feat: added features for user, added suit tests (#2)

* chore: updated github actions

* feat: added user recocile, added suit tests

* fix: upgrade x/net for high cve

* fix: update test command

* chore: updated github actions

* chore: updated docker file

Run Details

108 of 501 new or added lines in 31 files covered. (21.56%)

3048 of 22521 relevant lines covered (13.53%)

0.15 hits per line

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

20.13

/internal/controller/user_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"
        "reflect"
        "strings"
        "time"

        "github.com/chideat/valkey-operator/api/core"
        "github.com/chideat/valkey-operator/api/v1alpha1"
        "github.com/chideat/valkey-operator/internal/builder"
        "github.com/chideat/valkey-operator/internal/config"
        "github.com/chideat/valkey-operator/internal/controller/user"
        "github.com/chideat/valkey-operator/internal/util"
        "github.com/chideat/valkey-operator/pkg/kubernetes"
        security "github.com/chideat/valkey-operator/pkg/security/password"

        corev1 "k8s.io/api/core/v1"
        "k8s.io/apimachinery/pkg/api/errors"
        "k8s.io/apimachinery/pkg/runtime"
        "k8s.io/apimachinery/pkg/types"
        "k8s.io/client-go/tools/record"
        "k8s.io/client-go/util/retry"
        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/event"
        "sigs.k8s.io/controller-runtime/pkg/log"
        "sigs.k8s.io/controller-runtime/pkg/predicate"
        "sigs.k8s.io/controller-runtime/pkg/reconcile"
)

const (
        UserFinalizer = "buf.red/user-finalizer"
)

// UserReconciler reconciles a User object
type UserReconciler struct {
        client.Client
        Scheme *runtime.Scheme

        K8sClient kubernetes.ClientSet
        Record    record.EventRecorder
        Handler   *user.UserHandler
}

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

        instance := v1alpha1.User{}
        err := r.Client.Get(ctx, req.NamespacedName, &instance)
        if err != nil {
                logger.Error(err, "get redis user failed")
                if errors.IsNotFound(err) {
                        return reconcile.Result{}, nil
                }
                return reconcile.Result{}, err
        }

        isMarkedToBeDeleted := instance.GetDeletionTimestamp() != nil
        if isMarkedToBeDeleted {
                logger.Info("remove finalizer")
                controllerutil.RemoveFinalizer(&instance, UserFinalizer)
                if err := r.Update(ctx, &instance); err != nil {
                        return ctrl.Result{}, err
                }
                return ctrl.Result{}, nil
        }

        vkName := instance.Spec.InstanceName
        switch instance.Spec.Arch {
        case core.ValkeySentinel, core.ValkeyReplica:
                rf := &v1alpha1.Failover{}
                if err := r.Get(ctx, types.NamespacedName{Namespace: instance.Namespace, Name: vkName}, rf); err != nil {
                        if errors.IsNotFound(err) {
                                logger.Error(err, "instance not found", "name", vkName)
                                return ctrl.Result{Requeue: true}, r.Delete(ctx, &instance)
                        }
                        logger.Error(err, "get failover failed", "name", instance.Name)
                        return ctrl.Result{}, err
                }
        case core.ValkeyCluster:
                cluster := &v1alpha1.Cluster{}
                if err := r.Get(ctx, types.NamespacedName{Namespace: instance.Namespace, Name: vkName}, cluster); err != nil {
                        if errors.IsNotFound(err) {
                                logger.Error(err, "instance not found", "name", vkName)
                                return ctrl.Result{Requeue: true}, r.Delete(ctx, &instance)
                        }
                        logger.Error(err, "get cluster instance failed", "name", instance.Name)
                        return ctrl.Result{}, err
                }
        }

        // verify user password
        for _, name := range instance.Spec.PasswordSecrets {
                if name == "" {
                        continue
                }
                secret := &corev1.Secret{}
                if err := r.Get(ctx, types.NamespacedName{
                        Namespace: instance.Namespace,
                        Name:      name,
                }, secret); err != nil {
                        logger.Error(err, "get secret failed", "secret name", name)
                        instance.Status.Message = err.Error()
                        instance.Status.Phase = v1alpha1.UserFail
                        if e := r.Client.Status().Update(ctx, &instance); e != nil {
                                logger.Error(e, "update User status to Fail failed")
                        }
                        return ctrl.Result{}, err
                } else if err := security.PasswordValidate(string(secret.Data["password"]), 8, 32); err != nil {
                        if instance.Spec.AccountType != v1alpha1.System {
                                instance.Status.Message = err.Error()
                                instance.Status.Phase = v1alpha1.UserFail
                                if e := r.Client.Status().Update(ctx, &instance); e != nil {
                                        logger.Error(e, "update User status to Fail failed")
                                }
                                return ctrl.Result{}, err
                        }
                }

                if secret.GetLabels() == nil {
                        secret.SetLabels(map[string]string{})
                }
                if secret.Labels[builder.InstanceNameLabelKey] != vkName ||
                        len(secret.GetOwnerReferences()) == 0 || secret.OwnerReferences[0].UID != instance.GetUID() {

                        secret.Labels[builder.ManagedByLabelKey] = config.AppName
                        secret.Labels[builder.InstanceNameLabelKey] = vkName
                        secret.OwnerReferences = util.BuildOwnerReferences(&instance)
                        if err = retry.RetryOnConflict(retry.DefaultRetry, func() error {
                                return r.K8sClient.UpdateSecret(ctx, instance.Namespace, secret)
                        }); err != nil {
                                logger.Error(err, "update secret owner failed", "secret", secret.Name)
                                instance.Status.Message = err.Error()
                                instance.Status.Phase = v1alpha1.UserFail
                                return ctrl.Result{RequeueAfter: time.Second * 5}, r.Client.Status().Update(ctx, &instance)
                        }
                }
        }

        if err := r.Handler.Do(ctx, instance, logger); err != nil {
                if strings.Contains(err.Error(), "instance is not ready") ||
                        strings.Contains(err.Error(), "node not ready") ||
                        strings.Contains(err.Error(), "user not operator") ||
                        strings.Contains(err.Error(), "ERR unknown command `ACL`") {
                        logger.V(3).Info("instance is not ready", "instance", vkName)
                        instance.Status.Message = err.Error()
                        instance.Status.Phase = v1alpha1.UserPending
                        if err := r.updateUserStatus(ctx, &instance); err != nil {
                                logger.Error(err, "update User status to Pending failed")
                        }
                        return ctrl.Result{RequeueAfter: time.Second * 15}, nil
                }

                instance.Status.Message = err.Error()
                instance.Status.Phase = v1alpha1.UserFail
                logger.Error(err, "user reconcile failed")
                if err := r.updateUserStatus(ctx, &instance); err != nil {
                        logger.Error(err, "update User status to Fail failed")
                }
                return reconcile.Result{RequeueAfter: time.Second * 10}, nil
        }
        instance.Status.Phase = v1alpha1.UserSuccess
        instance.Status.Message = ""
        logger.V(3).Info("user reconcile success")
        if err := r.updateUserStatus(ctx, &instance); err != nil {
                logger.Error(err, "update User status to Success failed")
                return reconcile.Result{RequeueAfter: time.Second * 10}, err
        }
        if !controllerutil.ContainsFinalizer(&instance, UserFinalizer) {
                controllerutil.AddFinalizer(&instance, UserFinalizer)
                if err := r.updateUser(ctx, &instance); err != nil {
                        logger.Error(err, "update finalizer user failed")
                        return ctrl.Result{}, err
                }
        }

        return ctrl.Result{}, nil
}

func (r *UserReconciler) updateUserStatus(ctx context.Context, inst *v1alpha1.User) error {
        return retry.RetryOnConflict(retry.DefaultRetry, func() error {
                var oldUser v1alpha1.User
                if err := r.Get(ctx, types.NamespacedName{Namespace: inst.Namespace, Name: inst.Name}, &oldUser); err != nil {
                        return err
                }
                inst.ResourceVersion = oldUser.ResourceVersion
                return r.Status().Update(ctx, inst)
        })
}

func (r *UserReconciler) updateUser(ctx context.Context, inst *v1alpha1.User) error {
        return retry.RetryOnConflict(retry.DefaultRetry, func() error {
                var oldUser v1alpha1.User
                if err := r.Get(ctx, types.NamespacedName{Namespace: inst.Namespace, Name: inst.Name}, &oldUser); err != nil {
                        return err
                }
                inst.ResourceVersion = oldUser.ResourceVersion
                return r.Update(ctx, inst)
        })
}

// SetupWithManager sets up the controller with the Manager.
func (r *UserReconciler) SetupWithManager(mgr ctrl.Manager) error {
        return ctrl.NewControllerManagedBy(mgr).
                For(&v1alpha1.User{}).
                Owns(&corev1.Secret{}).
                WithEventFilter(generationChangedFilter()).
                WithOptions(controller.Options{MaxConcurrentReconciles: 8}).
                Complete(r)
}

func generationChangedFilter() predicate.Predicate {
        return predicate.Funcs{
                UpdateFunc: func(e event.UpdateEvent) bool {
                        if e.ObjectOld == nil {
                                return false
                        }
                        if e.ObjectNew == nil {
                                return false
                        }
                        if reflect.TypeOf(e.ObjectNew).String() == reflect.TypeOf(&corev1.Secret{}).String() {
                                return true
                        }
                        // Ignore updates to CR status in which case metadata.Generation does not change
                        return e.ObjectOld.GetGeneration() != e.ObjectNew.GetGeneration()
                },
        }
}

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	"reflect"
22	"strings"
23	"time"
24
25	"github.com/chideat/valkey-operator/api/core"
26	"github.com/chideat/valkey-operator/api/v1alpha1"
27	"github.com/chideat/valkey-operator/internal/builder"
28	"github.com/chideat/valkey-operator/internal/config"
29	"github.com/chideat/valkey-operator/internal/controller/user"
30	"github.com/chideat/valkey-operator/internal/util"
31	"github.com/chideat/valkey-operator/pkg/kubernetes"
32	security "github.com/chideat/valkey-operator/pkg/security/password"
33
34	corev1 "k8s.io/api/core/v1"
35	"k8s.io/apimachinery/pkg/api/errors"
36	"k8s.io/apimachinery/pkg/runtime"
37	"k8s.io/apimachinery/pkg/types"
38	"k8s.io/client-go/tools/record"
39	"k8s.io/client-go/util/retry"
40	ctrl "sigs.k8s.io/controller-runtime"
41	"sigs.k8s.io/controller-runtime/pkg/client"
42	"sigs.k8s.io/controller-runtime/pkg/controller"
43	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
44	"sigs.k8s.io/controller-runtime/pkg/event"
45	"sigs.k8s.io/controller-runtime/pkg/log"
46	"sigs.k8s.io/controller-runtime/pkg/predicate"
47	"sigs.k8s.io/controller-runtime/pkg/reconcile"
48	)
49
50	const (
51	UserFinalizer = "buf.red/user-finalizer"
52	)
53
54	// UserReconciler reconciles a User object
55	type UserReconciler struct {
56	client.Client
57	Scheme *runtime.Scheme
58
59	K8sClient kubernetes.ClientSet
60	Record record.EventRecorder
61	Handler *user.UserHandler
62	}
63
64	// +kubebuilder:rbac:groups=valkey.buf.red,resources=users,verbs=get;list;watch;create;update;patch;delete
65	// +kubebuilder:rbac:groups=valkey.buf.red,resources=users/status,verbs=get;update;patch
66	// +kubebuilder:rbac:groups=valkey.buf.red,resources=users/finalizers,verbs=update
67
68	// Reconcile is part of the main kubernetes reconciliation loop which aims to
69	// move the current state of the cluster closer to the desired state.
70	func (r *UserReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {	1✔
71	logger := log.FromContext(ctx).WithName("User").WithValues("target", req.String())	1✔
72		1✔
73	instance := v1alpha1.User{}	1✔
74	err := r.Client.Get(ctx, req.NamespacedName, &instance)	1✔
75	if err != nil {	1✔
NEW 76	logger.Error(err, "get redis user failed")	×
NEW 77	if errors.IsNotFound(err) {	×
NEW 78	return reconcile.Result{}, nil	×
NEW 79	}	×
NEW 80	return reconcile.Result{}, err	×
81	}
82
83	isMarkedToBeDeleted := instance.GetDeletionTimestamp() != nil	1✔
84	if isMarkedToBeDeleted {	1✔
NEW 85	logger.Info("remove finalizer")	×
NEW 86	controllerutil.RemoveFinalizer(&instance, UserFinalizer)	×
NEW 87	if err := r.Update(ctx, &instance); err != nil {	×
NEW 88	return ctrl.Result{}, err	×
NEW 89	}	×
NEW 90	return ctrl.Result{}, nil	×
91	}
92
93	vkName := instance.Spec.InstanceName	1✔
94	switch instance.Spec.Arch {	1✔
NEW 95	case core.ValkeySentinel, core.ValkeyReplica:	×
NEW 96	rf := &v1alpha1.Failover{}	×
NEW 97	if err := r.Get(ctx, types.NamespacedName{Namespace: instance.Namespace, Name: vkName}, rf); err != nil {	×
NEW 98	if errors.IsNotFound(err) {	×
NEW 99	logger.Error(err, "instance not found", "name", vkName)	×
NEW 100	return ctrl.Result{Requeue: true}, r.Delete(ctx, &instance)	×
NEW 101	}	×
NEW 102	logger.Error(err, "get failover failed", "name", instance.Name)	×
NEW 103	return ctrl.Result{}, err	×
104	}
105	case core.ValkeyCluster:	1✔
106	cluster := &v1alpha1.Cluster{}	1✔
107	if err := r.Get(ctx, types.NamespacedName{Namespace: instance.Namespace, Name: vkName}, cluster); err != nil {	1✔
NEW 108	if errors.IsNotFound(err) {	×
NEW 109	logger.Error(err, "instance not found", "name", vkName)	×
NEW 110	return ctrl.Result{Requeue: true}, r.Delete(ctx, &instance)	×
NEW 111	}	×
NEW 112	logger.Error(err, "get cluster instance failed", "name", instance.Name)	×
NEW 113	return ctrl.Result{}, err	×
114	}
115	}
116
117	// verify user password
118	for _, name := range instance.Spec.PasswordSecrets {	1✔
NEW 119	if name == "" {	×
NEW 120	continue	×
121	}
NEW 122	secret := &corev1.Secret{}	×
NEW 123	if err := r.Get(ctx, types.NamespacedName{	×
NEW 124	Namespace: instance.Namespace,	×
NEW 125	Name: name,	×
NEW 126	}, secret); err != nil {	×
NEW 127	logger.Error(err, "get secret failed", "secret name", name)	×
NEW 128	instance.Status.Message = err.Error()	×
NEW 129	instance.Status.Phase = v1alpha1.UserFail	×
NEW 130	if e := r.Client.Status().Update(ctx, &instance); e != nil {	×
NEW 131	logger.Error(e, "update User status to Fail failed")	×
NEW 132	}	×
NEW 133	return ctrl.Result{}, err	×
NEW 134	} else if err := security.PasswordValidate(string(secret.Data["password"]), 8, 32); err != nil {	×
NEW 135	if instance.Spec.AccountType != v1alpha1.System {	×
NEW 136	instance.Status.Message = err.Error()	×
NEW 137	instance.Status.Phase = v1alpha1.UserFail	×
NEW 138	if e := r.Client.Status().Update(ctx, &instance); e != nil {	×
NEW 139	logger.Error(e, "update User status to Fail failed")	×
NEW 140	}	×
NEW 141	return ctrl.Result{}, err	×
142	}
143	}
144
NEW 145	if secret.GetLabels() == nil {	×
NEW 146	secret.SetLabels(map[string]string{})	×
NEW 147	}	×
NEW 148	if secret.Labels[builder.InstanceNameLabelKey] != vkName \|\|	×
NEW 149	len(secret.GetOwnerReferences()) == 0 \|\| secret.OwnerReferences[0].UID != instance.GetUID() {	×
150		×
NEW 151	secret.Labels[builder.ManagedByLabelKey] = config.AppName	×
NEW 152	secret.Labels[builder.InstanceNameLabelKey] = vkName	×
NEW 153	secret.OwnerReferences = util.BuildOwnerReferences(&instance)	×
NEW 154	if err = retry.RetryOnConflict(retry.DefaultRetry, func() error {	×
NEW 155	return r.K8sClient.UpdateSecret(ctx, instance.Namespace, secret)	×
NEW 156	}); err != nil {	×
NEW 157	logger.Error(err, "update secret owner failed", "secret", secret.Name)	×
NEW 158	instance.Status.Message = err.Error()	×
NEW 159	instance.Status.Phase = v1alpha1.UserFail	×
NEW 160	return ctrl.Result{RequeueAfter: time.Second * 5}, r.Client.Status().Update(ctx, &instance)	×
NEW 161	}	×
162	}
163	}
164
165	if err := r.Handler.Do(ctx, instance, logger); err != nil {	2✔
166	if strings.Contains(err.Error(), "instance is not ready") \|\|	1✔
167	strings.Contains(err.Error(), "node not ready") \|\|	1✔
168	strings.Contains(err.Error(), "user not operator") \|\|	1✔
169	strings.Contains(err.Error(), "ERR unknown command `ACL`") {	1✔
NEW 170	logger.V(3).Info("instance is not ready", "instance", vkName)	×
NEW 171	instance.Status.Message = err.Error()	×
NEW 172	instance.Status.Phase = v1alpha1.UserPending	×
NEW 173	if err := r.updateUserStatus(ctx, &instance); err != nil {	×
NEW 174	logger.Error(err, "update User status to Pending failed")	×
NEW 175	}	×
NEW 176	return ctrl.Result{RequeueAfter: time.Second * 15}, nil	×
177	}
178
179	instance.Status.Message = err.Error()	1✔
180	instance.Status.Phase = v1alpha1.UserFail	1✔
181	logger.Error(err, "user reconcile failed")	1✔
182	if err := r.updateUserStatus(ctx, &instance); err != nil {	1✔
NEW 183	logger.Error(err, "update User status to Fail failed")	×
NEW 184	}	×
185	return reconcile.Result{RequeueAfter: time.Second * 10}, nil	1✔
186	}
NEW 187	instance.Status.Phase = v1alpha1.UserSuccess	×
NEW 188	instance.Status.Message = ""	×
NEW 189	logger.V(3).Info("user reconcile success")	×
NEW 190	if err := r.updateUserStatus(ctx, &instance); err != nil {	×
NEW 191	logger.Error(err, "update User status to Success failed")	×
NEW 192	return reconcile.Result{RequeueAfter: time.Second * 10}, err	×
NEW 193	}	×
NEW 194	if !controllerutil.ContainsFinalizer(&instance, UserFinalizer) {	×
NEW 195	controllerutil.AddFinalizer(&instance, UserFinalizer)	×
NEW 196	if err := r.updateUser(ctx, &instance); err != nil {	×
NEW 197	logger.Error(err, "update finalizer user failed")	×
NEW 198	return ctrl.Result{}, err	×
NEW 199	}	×
200	}
201
202	return ctrl.Result{}, nil	×
203	}
204
205	func (r UserReconciler) updateUserStatus(ctx context.Context, inst v1alpha1.User) error {	1✔
206	return retry.RetryOnConflict(retry.DefaultRetry, func() error {	2✔
207	var oldUser v1alpha1.User	1✔
208	if err := r.Get(ctx, types.NamespacedName{Namespace: inst.Namespace, Name: inst.Name}, &oldUser); err != nil {	1✔
NEW 209	return err	×
NEW 210	}	×
211	inst.ResourceVersion = oldUser.ResourceVersion	1✔
212	return r.Status().Update(ctx, inst)	1✔
213	})
214	}
215
NEW 216	func (r UserReconciler) updateUser(ctx context.Context, inst v1alpha1.User) error {	×
NEW 217	return retry.RetryOnConflict(retry.DefaultRetry, func() error {	×
NEW 218	var oldUser v1alpha1.User	×
NEW 219	if err := r.Get(ctx, types.NamespacedName{Namespace: inst.Namespace, Name: inst.Name}, &oldUser); err != nil {	×
NEW 220	return err	×
NEW 221	}	×
NEW 222	inst.ResourceVersion = oldUser.ResourceVersion	×
NEW 223	return r.Update(ctx, inst)	×
224	})
225	}
226
227	// SetupWithManager sets up the controller with the Manager.
228	func (r *UserReconciler) SetupWithManager(mgr ctrl.Manager) error {	×
229	return ctrl.NewControllerManagedBy(mgr).	×
NEW 230	For(&v1alpha1.User{}).	×
NEW 231	Owns(&corev1.Secret{}).	×
NEW 232	WithEventFilter(generationChangedFilter()).	×
NEW 233	WithOptions(controller.Options{MaxConcurrentReconciles: 8}).	×
234	Complete(r)	×
235	}	×
236
NEW 237	func generationChangedFilter() predicate.Predicate {	×
NEW 238	return predicate.Funcs{	×
NEW 239	UpdateFunc: func(e event.UpdateEvent) bool {	×
NEW 240	if e.ObjectOld == nil {	×
NEW 241	return false	×
NEW 242	}	×
NEW 243	if e.ObjectNew == nil {	×
NEW 244	return false	×
NEW 245	}	×
NEW 246	if reflect.TypeOf(e.ObjectNew).String() == reflect.TypeOf(&corev1.Secret{}).String() {	×
NEW 247	return true	×
NEW 248	}	×
249	// Ignore updates to CR status in which case metadata.Generation does not change
NEW 250	return e.ObjectOld.GetGeneration() != e.ObjectNew.GetGeneration()	×
251	},
252	}
253	}

chideat / valkey-operator / 13214360144

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