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

0.0

/internal/controller/user/handler.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 user

import (
        "context"
        "encoding/json"
        "fmt"

        "github.com/chideat/valkey-operator/pkg/types"
        "github.com/chideat/valkey-operator/pkg/types/user"
        "github.com/go-logr/logr"
        "k8s.io/apimachinery/pkg/api/errors"
        "k8s.io/client-go/tools/record"

        "github.com/chideat/valkey-operator/api/core"
        "github.com/chideat/valkey-operator/api/v1alpha1"
        "github.com/chideat/valkey-operator/internal/builder/aclbuilder"
        "github.com/chideat/valkey-operator/internal/valkey/cluster"
        "github.com/chideat/valkey-operator/internal/valkey/failover"
        "github.com/chideat/valkey-operator/pkg/kubernetes"
)

type UserHandler struct {
        k8sClient     kubernetes.ClientSet
        eventRecorder record.EventRecorder
        logger        logr.Logger
}

func NewUserHandler(k8sservice kubernetes.ClientSet, eventRecorder record.EventRecorder, logger logr.Logger) *UserHandler {
        return &UserHandler{
                k8sClient:     k8sservice,
                eventRecorder: eventRecorder,
                logger:        logger.WithName("UserHandler"),
        }
}

func (r *UserHandler) Delete(ctx context.Context, inst v1alpha1.User, logger logr.Logger) error {
        logger.V(3).Info("delete user", "user instance name", inst.Name,
                "instance", inst.Spec.InstanceName, "type", inst.Spec.Arch)
        if inst.Spec.Username == user.DefaultUserName || inst.Spec.Username == user.DefaultOperatorUserName {
                return nil
        }

        vkName := inst.Spec.InstanceName
        cmName := aclbuilder.GenerateACLConfigMapName(inst.Spec.Arch, vkName)
        if configMap, err := r.k8sClient.GetConfigMap(ctx, inst.Namespace, cmName); err != nil {
                if !errors.IsNotFound(err) {
                        logger.Error(err, "delete user from configmap failed")
                        return err
                }
        } else if _, ok := configMap.Data[inst.Spec.Username]; ok {
                delete(configMap.Data, inst.Spec.Username)
                if err := r.k8sClient.UpdateConfigMap(ctx, inst.Namespace, configMap); err != nil {
                        logger.Error(err, "delete user from configmap failed", "configmap", cmName)
                        return err
                }
        }

        switch inst.Spec.Arch {
        case core.ValkeyCluster:
                logger.V(3).Info("cluster", "instance", vkName)
                rc, err := r.k8sClient.GetCluster(ctx, inst.Namespace, vkName)
                if errors.IsNotFound(err) {
                        return nil
                } else if err != nil {
                        return err
                }

                rcm, err := cluster.NewCluster(ctx, r.k8sClient, r.eventRecorder, rc, logger)
                if err != nil {
                        return err
                }
                if !rcm.IsReady() {
                        logger.V(3).Info("instance is not ready", "instance", vkName)
                        return fmt.Errorf("instance is not ready")
                }

                for _, node := range rcm.Nodes() {
                        err := node.Setup(ctx, []any{"ACL", "DELUSER", inst.Spec.Username})
                        if err != nil {
                                logger.Error(err, "acl del user failed", "node", node.GetName())
                                return err
                        }
                        logger.V(3).Info("acl del user success", "node", node.GetName())
                }
        case core.ValkeyFailover, core.ValkeyReplica:
                logger.V(3).Info("sentinel", "instane", vkName)
                rf, err := r.k8sClient.GetFailover(ctx, inst.Namespace, vkName)
                if errors.IsNotFound(err) {
                        return nil
                } else if err != nil {
                        return err
                }

                rfm, err := failover.NewFailover(ctx, r.k8sClient, r.eventRecorder, rf, logger)
                if err != nil {
                        return err
                }
                if !rfm.IsReady() {
                        logger.V(3).Info("instance is not ready", "instance", vkName)
                        return fmt.Errorf("instance is not ready")
                }
                for _, node := range rfm.Nodes() {
                        err := node.Setup(ctx, []any{"ACL", "DELUSER", inst.Spec.Username})
                        if err != nil {
                                logger.Error(err, "acl del user failed", "node", node.GetName())
                                return err
                        }
                        logger.V(3).Info("acl del user success", "node", node.GetName())
                }
        }
        return nil
}

func (r *UserHandler) Do(ctx context.Context, inst v1alpha1.User, logger logr.Logger) error {
        if inst.Annotations == nil {
                inst.Annotations = map[string]string{}
        }

        passwords := []string{}
        userPassword := &user.Password{}
        for _, secretName := range inst.Spec.PasswordSecrets {
                secret, err := r.k8sClient.GetSecret(ctx, inst.Namespace, secretName)
                if err != nil {
                        return err
                }
                passwords = append(passwords, string(secret.Data["password"]))
                userPassword = &user.Password{
                        SecretName: secretName,
                }
        }

        logger.V(3).Info("reconcile user", "user name", inst.Name, "type", inst.Spec.Arch)
        vkName := inst.Spec.InstanceName
        cmName := aclbuilder.GenerateACLConfigMapName(inst.Spec.Arch, vkName)

        switch inst.Spec.Arch {
        case core.ValkeyCluster:
                logger.V(3).Info("cluster", "instance", vkName)
                rc, err := r.k8sClient.GetCluster(ctx, inst.Namespace, vkName)
                if err != nil {
                        return err
                }

                rcm, err := cluster.NewCluster(ctx, r.k8sClient, r.eventRecorder, rc, logger)
                if err != nil {
                        return err
                }
                if !rcm.IsReady() {
                        logger.V(3).Info("instance is not ready", "instance", vkName)
                        return fmt.Errorf("instance is not ready")
                }

                aclRules := inst.Spec.AclRules
                rule, err := user.NewRule(inst.Spec.AclRules)
                if err != nil {
                        logger.V(3).Info("rule parse failed", "rule", inst.Spec.AclRules)
                        return err
                }
                rule = types.PatchClusterClientRequiredRules(rule)
                aclRules = rule.Encode()

                userObj, err := types.NewUserFromValkeyUser(inst.Spec.Username, aclRules, userPassword)
                if err != nil {
                        return err
                }
                info, err := json.Marshal(userObj)
                if err != nil {
                        return err
                }

                configmap, err := r.k8sClient.GetConfigMap(ctx, inst.Namespace, cmName)
                if err != nil {
                        return err
                }
                configmap.Data[inst.Spec.Username] = string(info)

                if inst.Spec.AccountType != v1alpha1.SystemAccount {
                        for _, node := range rcm.Nodes() {
                                _, err := node.SetACLUser(ctx, inst.Spec.Username, passwords, aclRules)
                                if err != nil {
                                        logger.Error(err, "acl set user failed", "node", node.GetName())
                                        return err
                                }
                                logger.V(3).Info("acl set user success", "node", node.GetName())
                        }
                } else {
                        logger.V(3).Info("skip system account online update", "username", inst.Spec.Username)
                }

                if err := r.k8sClient.UpdateConfigMap(ctx, inst.Namespace, configmap); err != nil {
                        logger.Error(err, "update configmap failed", "configmap", configmap.Name)
                        return err
                }
        case core.ValkeyFailover, core.ValkeyReplica:
                logger.V(3).Info("sentinel", "instance", vkName)
                rf, err := r.k8sClient.GetFailover(ctx, inst.Namespace, vkName)
                if err != nil {
                        return err
                }
                rfm, err := failover.NewFailover(ctx, r.k8sClient, r.eventRecorder, rf, logger)
                if err != nil {
                        return err
                }
                if !rfm.IsReady() {
                        logger.V(3).Info("instance is not ready", "instance", vkName)
                        return fmt.Errorf("instance is not ready")
                }
                configmap, err := r.k8sClient.GetConfigMap(ctx, inst.Namespace, cmName)
                if err != nil {
                        return err
                }
                userObj, err := types.NewUserFromValkeyUser(inst.Spec.Username, inst.Spec.AclRules, userPassword)
                if err != nil {
                        return err
                }
                info, err := json.Marshal(userObj)
                if err != nil {
                        return err
                }
                configmap.Data[inst.Spec.Username] = string(info)

                if inst.Spec.AccountType != v1alpha1.SystemAccount {
                        for _, node := range rfm.Nodes() {
                                _, err := node.SetACLUser(ctx, inst.Spec.Username, passwords, inst.Spec.AclRules)
                                if err != nil {
                                        logger.Error(err, "acl set user failed", "node", node.GetName())
                                        return err
                                }
                                logger.V(3).Info("acl set user success", "node", node.GetName())
                        }
                } else {
                        logger.V(3).Info("skip system account online update", "username", inst.Spec.Username)
                }

                if err := r.k8sClient.UpdateConfigMap(ctx, inst.Namespace, configmap); err != nil {
                        logger.Error(err, "update configmap failed", "configmap", configmap.Name)
                        return err
                }
        }
        return nil
}

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 user
18
19	import (
20	"context"
21	"encoding/json"
22	"fmt"
23
24	"github.com/chideat/valkey-operator/pkg/types"
25	"github.com/chideat/valkey-operator/pkg/types/user"
26	"github.com/go-logr/logr"
27	"k8s.io/apimachinery/pkg/api/errors"
28	"k8s.io/client-go/tools/record"
29
30	"github.com/chideat/valkey-operator/api/core"
31	"github.com/chideat/valkey-operator/api/v1alpha1"
32	"github.com/chideat/valkey-operator/internal/builder/aclbuilder"
33	"github.com/chideat/valkey-operator/internal/valkey/cluster"
34	"github.com/chideat/valkey-operator/internal/valkey/failover"
35	"github.com/chideat/valkey-operator/pkg/kubernetes"
36	)
37
38	type UserHandler struct {
39	k8sClient kubernetes.ClientSet
40	eventRecorder record.EventRecorder
41	logger logr.Logger
42	}
43
UNCOV 44	func NewUserHandler(k8sservice kubernetes.ClientSet, eventRecorder record.EventRecorder, logger logr.Logger) *UserHandler {	×
UNCOV 45	return &UserHandler{	×
UNCOV 46	k8sClient: k8sservice,	×
UNCOV 47	eventRecorder: eventRecorder,	×
UNCOV 48	logger: logger.WithName("UserHandler"),	×
UNCOV 49	}	×
UNCOV 50	}	×
51
52	func (r *UserHandler) Delete(ctx context.Context, inst v1alpha1.User, logger logr.Logger) error {	×
53	logger.V(3).Info("delete user", "user instance name", inst.Name,	×
54	"instance", inst.Spec.InstanceName, "type", inst.Spec.Arch)	×
55	if inst.Spec.Username == user.DefaultUserName \|\| inst.Spec.Username == user.DefaultOperatorUserName {	×
56	return nil	×
57	}	×
58
59	vkName := inst.Spec.InstanceName	×
60	cmName := aclbuilder.GenerateACLConfigMapName(inst.Spec.Arch, vkName)	×
61	if configMap, err := r.k8sClient.GetConfigMap(ctx, inst.Namespace, cmName); err != nil {	×
62	if !errors.IsNotFound(err) {	×
63	logger.Error(err, "delete user from configmap failed")	×
64	return err	×
65	}	×
66	} else if _, ok := configMap.Data[inst.Spec.Username]; ok {	×
67	delete(configMap.Data, inst.Spec.Username)	×
68	if err := r.k8sClient.UpdateConfigMap(ctx, inst.Namespace, configMap); err != nil {	×
69	logger.Error(err, "delete user from configmap failed", "configmap", cmName)	×
70	return err	×
71	}	×
72	}
73
74	switch inst.Spec.Arch {	×
75	case core.ValkeyCluster:	×
76	logger.V(3).Info("cluster", "instance", vkName)	×
77	rc, err := r.k8sClient.GetCluster(ctx, inst.Namespace, vkName)	×
78	if errors.IsNotFound(err) {	×
79	return nil	×
80	} else if err != nil {	×
81	return err	×
82	}	×
83
84	rcm, err := cluster.NewCluster(ctx, r.k8sClient, r.eventRecorder, rc, logger)	×
85	if err != nil {	×
86	return err	×
87	}	×
88	if !rcm.IsReady() {	×
89	logger.V(3).Info("instance is not ready", "instance", vkName)	×
90	return fmt.Errorf("instance is not ready")	×
91	}	×
92
93	for _, node := range rcm.Nodes() {	×
NEW 94	err := node.Setup(ctx, []any{"ACL", "DELUSER", inst.Spec.Username})	×
95	if err != nil {	×
96	logger.Error(err, "acl del user failed", "node", node.GetName())	×
97	return err	×
98	}	×
99	logger.V(3).Info("acl del user success", "node", node.GetName())	×
100	}
101	case core.ValkeyFailover, core.ValkeyReplica:	×
102	logger.V(3).Info("sentinel", "instane", vkName)	×
103	rf, err := r.k8sClient.GetFailover(ctx, inst.Namespace, vkName)	×
104	if errors.IsNotFound(err) {	×
105	return nil	×
106	} else if err != nil {	×
107	return err	×
108	}	×
109
110	rfm, err := failover.NewFailover(ctx, r.k8sClient, r.eventRecorder, rf, logger)	×
111	if err != nil {	×
112	return err	×
113	}	×
114	if !rfm.IsReady() {	×
115	logger.V(3).Info("instance is not ready", "instance", vkName)	×
116	return fmt.Errorf("instance is not ready")	×
117	}	×
118	for _, node := range rfm.Nodes() {	×
NEW 119	err := node.Setup(ctx, []any{"ACL", "DELUSER", inst.Spec.Username})	×
120	if err != nil {	×
121	logger.Error(err, "acl del user failed", "node", node.GetName())	×
122	return err	×
123	}	×
124	logger.V(3).Info("acl del user success", "node", node.GetName())	×
125	}
126	}
127	return nil	×
128	}
129
UNCOV 130	func (r *UserHandler) Do(ctx context.Context, inst v1alpha1.User, logger logr.Logger) error {	×
UNCOV 131	if inst.Annotations == nil {	×
UNCOV 132	inst.Annotations = map[string]string{}	×
UNCOV 133	}	×
134
UNCOV 135	passwords := []string{}	×
UNCOV 136	userPassword := &user.Password{}	×
UNCOV 137	for _, secretName := range inst.Spec.PasswordSecrets {	×
138	secret, err := r.k8sClient.GetSecret(ctx, inst.Namespace, secretName)	×
139	if err != nil {	×
140	return err	×
141	}	×
142	passwords = append(passwords, string(secret.Data["password"]))	×
143	userPassword = &user.Password{	×
144	SecretName: secretName,	×
145	}	×
146	}
147
UNCOV 148	logger.V(3).Info("reconcile user", "user name", inst.Name, "type", inst.Spec.Arch)	×
UNCOV 149	vkName := inst.Spec.InstanceName	×
UNCOV 150	cmName := aclbuilder.GenerateACLConfigMapName(inst.Spec.Arch, vkName)	×
UNCOV 151		×
UNCOV 152	switch inst.Spec.Arch {	×
UNCOV 153	case core.ValkeyCluster:	×
UNCOV 154	logger.V(3).Info("cluster", "instance", vkName)	×
UNCOV 155	rc, err := r.k8sClient.GetCluster(ctx, inst.Namespace, vkName)	×
UNCOV 156	if err != nil {	×
157	return err	×
158	}	×
159
UNCOV 160	rcm, err := cluster.NewCluster(ctx, r.k8sClient, r.eventRecorder, rc, logger)	×
UNCOV 161	if err != nil {	×
162	return err	×
163	}	×
UNCOV 164	if !rcm.IsReady() {	×
165	logger.V(3).Info("instance is not ready", "instance", vkName)	×
166	return fmt.Errorf("instance is not ready")	×
167	}	×
168
UNCOV 169	aclRules := inst.Spec.AclRules	×
UNCOV 170	rule, err := user.NewRule(inst.Spec.AclRules)	×
UNCOV 171	if err != nil {	×
172	logger.V(3).Info("rule parse failed", "rule", inst.Spec.AclRules)	×
173	return err	×
174	}	×
UNCOV 175	rule = types.PatchClusterClientRequiredRules(rule)	×
UNCOV 176	aclRules = rule.Encode()	×
UNCOV 177		×
UNCOV 178	userObj, err := types.NewUserFromValkeyUser(inst.Spec.Username, aclRules, userPassword)	×
UNCOV 179	if err != nil {	×
180	return err	×
181	}	×
UNCOV 182	info, err := json.Marshal(userObj)	×
UNCOV 183	if err != nil {	×
184	return err	×
185	}	×
186
UNCOV 187	configmap, err := r.k8sClient.GetConfigMap(ctx, inst.Namespace, cmName)	×
UNCOV 188	if err != nil {	×
UNCOV 189	return err	×
UNCOV 190	}	×
191	configmap.Data[inst.Spec.Username] = string(info)	×
192		×
NEW 193	if inst.Spec.AccountType != v1alpha1.SystemAccount {	×
194	for _, node := range rcm.Nodes() {	×
195	_, err := node.SetACLUser(ctx, inst.Spec.Username, passwords, aclRules)	×
196	if err != nil {	×
197	logger.Error(err, "acl set user failed", "node", node.GetName())	×
198	return err	×
199	}	×
200	logger.V(3).Info("acl set user success", "node", node.GetName())	×
201	}
202	} else {	×
203	logger.V(3).Info("skip system account online update", "username", inst.Spec.Username)	×
204	}	×
205
206	if err := r.k8sClient.UpdateConfigMap(ctx, inst.Namespace, configmap); err != nil {	×
207	logger.Error(err, "update configmap failed", "configmap", configmap.Name)	×
208	return err	×
209	}	×
210	case core.ValkeyFailover, core.ValkeyReplica:	×
211	logger.V(3).Info("sentinel", "instance", vkName)	×
212	rf, err := r.k8sClient.GetFailover(ctx, inst.Namespace, vkName)	×
213	if err != nil {	×
214	return err	×
215	}	×
216	rfm, err := failover.NewFailover(ctx, r.k8sClient, r.eventRecorder, rf, logger)	×
217	if err != nil {	×
218	return err	×
219	}	×
220	if !rfm.IsReady() {	×
221	logger.V(3).Info("instance is not ready", "instance", vkName)	×
222	return fmt.Errorf("instance is not ready")	×
223	}	×
224	configmap, err := r.k8sClient.GetConfigMap(ctx, inst.Namespace, cmName)	×
225	if err != nil {	×
226	return err	×
227	}	×
228	userObj, err := types.NewUserFromValkeyUser(inst.Spec.Username, inst.Spec.AclRules, userPassword)	×
229	if err != nil {	×
230	return err	×
231	}	×
232	info, err := json.Marshal(userObj)	×
233	if err != nil {	×
234	return err	×
235	}	×
236	configmap.Data[inst.Spec.Username] = string(info)	×
237		×
NEW 238	if inst.Spec.AccountType != v1alpha1.SystemAccount {	×
239	for _, node := range rfm.Nodes() {	×
240	_, err := node.SetACLUser(ctx, inst.Spec.Username, passwords, inst.Spec.AclRules)	×
241	if err != nil {	×
242	logger.Error(err, "acl set user failed", "node", node.GetName())	×
243	return err	×
244	}	×
245	logger.V(3).Info("acl set user success", "node", node.GetName())	×
246	}
247	} else {	×
248	logger.V(3).Info("skip system account online update", "username", inst.Spec.Username)	×
249	}	×
250
251	if err := r.k8sClient.UpdateConfigMap(ctx, inst.Namespace, configmap); err != nil {	×
252	logger.Error(err, "update configmap failed", "configmap", configmap.Name)	×
253	return err	×
254	}	×
255	}
256	return nil	×
257	}

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