5738153381

Committed 02 Aug 2023 11:12AM UTC coverage: 75.091% (+0.08%) from 75.008%

Build # 5738153381

Build Type

Pull #519

github

Committed by

lucastt

Commit Message

Add stack weight verification for HPA reconciling

Signed-off-by: Lucas Thiesen <lucas.thiesen@zalando.de>

Pull Request Pull Request #519: Add stack weight verification for HPA reconciling

Run Details

27 of 27 new or added lines in 1 file covered. (100.0%)

2258 of 3007 relevant lines covered (75.09%)

0.84 hits per line

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

83.79

/controller/stack_resources.go

package controller

import (
        "context"
        "fmt"
        "strconv"

        rgv1 "github.com/szuecs/routegroup-client/apis/zalando.org/v1"
        zv1 "github.com/zalando-incubator/stackset-controller/pkg/apis/zalando.org/v1"
        "github.com/zalando-incubator/stackset-controller/pkg/core"
        apps "k8s.io/api/apps/v1"
        v2 "k8s.io/api/autoscaling/v2"
        apiv1 "k8s.io/api/core/v1"
        networking "k8s.io/api/networking/v1"
        metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

func pint32Equal(p1, p2 *int32) bool {
        if p1 == nil && p2 == nil {
                return true
        }
        if p1 != nil && p2 != nil {
                return *p1 == *p2
        }
        return false
}

func isHPATrafficWeightUpdated(stack *zv1.Stack, hpa *v2.HorizontalPodAutoscaler) (bool, error) {
        // If resource has external RPS metric we need to check if this reconcile was
        // triggered by traffic switch, in this case its necessary to update the weight
        // annotation for the specific metric.
        for _, metric := range hpa.Spec.Metrics {
                if metric.Type == v2.ExternalMetricSourceType {
                        t, ok := metric.External.Metric.Selector.MatchLabels["type"]
                        if ok && t == "requests-per-second" {
                                stackWeight := stack.Status.ActualTrafficWeight
                                key := fmt.Sprintf(
                                        "metric-config.external.%s.requests-per-second/weight",
                                        metric.External.Metric.Name,
                                )
                                hpaWeight, exist := hpa.Annotations[key]
                                w, err := strconv.ParseFloat(hpaWeight, 64)
                                if err != nil {
                                        return false, err
                                }
                                return exist && w == stackWeight, nil
                        }
                }
        }

        return true, nil
}

// syncObjectMeta copies metadata elements such as labels or annotations from source to target
func syncObjectMeta(target, source metav1.Object) {
        target.SetLabels(source.GetLabels())
        target.SetAnnotations(source.GetAnnotations())
}

func (c *StackSetController) ReconcileStackDeployment(ctx context.Context, stack *zv1.Stack, existing *apps.Deployment, generateUpdated func() *apps.Deployment) error {
        deployment := generateUpdated()

        // Create new deployment
        if existing == nil {
                _, err := c.client.AppsV1().Deployments(deployment.Namespace).Create(ctx, deployment, metav1.CreateOptions{})
                if err != nil {
                        return err
                }
                c.recorder.Eventf(
                        stack,
                        apiv1.EventTypeNormal,
                        "CreatedDeployment",
                        "Created Deployment %s",
                        deployment.Name)
                return nil
        }

        // Check if we need to update the deployment
        if core.IsResourceUpToDate(stack, existing.ObjectMeta) && pint32Equal(existing.Spec.Replicas, deployment.Spec.Replicas) {
                return nil
        }

        updated := existing.DeepCopy()
        syncObjectMeta(updated, deployment)
        updated.Spec = deployment.Spec
        updated.Spec.Selector = existing.Spec.Selector

        _, err := c.client.AppsV1().Deployments(updated.Namespace).Update(ctx, updated, metav1.UpdateOptions{})
        if err != nil {
                return err
        }
        c.recorder.Eventf(
                stack,
                apiv1.EventTypeNormal,
                "UpdatedDeployment",
                "Updated Deployment %s",
                deployment.Name)
        return nil
}

func (c *StackSetController) ReconcileStackHPA(ctx context.Context, stack *zv1.Stack, existing *v2.HorizontalPodAutoscaler, generateUpdated func() (*v2.HorizontalPodAutoscaler, error)) error {
        hpa, err := generateUpdated()
        if err != nil {
                return err
        }

        // HPA removed
        if hpa == nil {
                if existing != nil {
                        err := c.client.AutoscalingV2().HorizontalPodAutoscalers(existing.Namespace).Delete(ctx, existing.Name, metav1.DeleteOptions{})
                        if err != nil {
                                return err
                        }
                        c.recorder.Eventf(
                                stack,
                                apiv1.EventTypeNormal,
                                "DeletedHPA",
                                "Deleted HPA %s",
                                existing.Namespace)
                }
                return nil
        }

        // Create new HPA
        if existing == nil {
                _, err := c.client.AutoscalingV2().HorizontalPodAutoscalers(hpa.Namespace).Create(ctx, hpa, metav1.CreateOptions{})
                if err != nil {
                        return err
                }
                c.recorder.Eventf(
                        stack,
                        apiv1.EventTypeNormal,
                        "CreatedHPA",
                        "Created HPA %s",
                        hpa.Name)
                return nil
        }

        // Check if we need to update the HPA
        weightUpdated, err := isHPATrafficWeightUpdated(stack, existing)
        if err != nil {
                return err
        }
        if core.IsResourceUpToDate(stack, existing.ObjectMeta) &&
                pint32Equal(existing.Spec.MinReplicas, hpa.Spec.MinReplicas) &&
                weightUpdated {
                return nil
        }

        updated := existing.DeepCopy()
        syncObjectMeta(updated, hpa)
        updated.Spec = hpa.Spec

        _, err = c.client.AutoscalingV2().HorizontalPodAutoscalers(updated.Namespace).Update(ctx, updated, metav1.UpdateOptions{})
        if err != nil {
                return err
        }
        c.recorder.Eventf(
                stack,
                apiv1.EventTypeNormal,
                "UpdatedHPA",
                "Updated HPA %s",
                hpa.Name)
        return nil
}

func (c *StackSetController) ReconcileStackService(ctx context.Context, stack *zv1.Stack, existing *apiv1.Service, generateUpdated func() (*apiv1.Service, error)) error {
        service, err := generateUpdated()
        if err != nil {
                return err
        }

        // Create new service
        if existing == nil {
                _, err := c.client.CoreV1().Services(service.Namespace).Create(ctx, service, metav1.CreateOptions{})
                if err != nil {
                        return err
                }
                c.recorder.Eventf(
                        stack,
                        apiv1.EventTypeNormal,
                        "CreatedService",
                        "Created Service %s",
                        service.Name)
                return nil
        }

        // Check if we need to update the service
        if core.IsResourceUpToDate(stack, existing.ObjectMeta) {
                return nil
        }

        updated := existing.DeepCopy()
        syncObjectMeta(updated, service)
        updated.Spec = service.Spec
        updated.Spec.ClusterIP = existing.Spec.ClusterIP // ClusterIP is immutable

        _, err = c.client.CoreV1().Services(updated.Namespace).Update(ctx, updated, metav1.UpdateOptions{})
        if err != nil {
                return err
        }
        c.recorder.Eventf(
                stack,
                apiv1.EventTypeNormal,
                "UpdatedService",
                "Updated Service %s",
                service.Name)
        return nil
}

func (c *StackSetController) ReconcileStackIngress(ctx context.Context, stack *zv1.Stack, existing *networking.Ingress, generateUpdated func() (*networking.Ingress, error)) error {
        ingress, err := generateUpdated()
        if err != nil {
                return err
        }

        // Ingress removed
        if ingress == nil {
                if existing != nil {
                        err := c.client.NetworkingV1().Ingresses(existing.Namespace).Delete(ctx, existing.Name, metav1.DeleteOptions{})
                        if err != nil {
                                return err
                        }
                        c.recorder.Eventf(
                                stack,
                                apiv1.EventTypeNormal,
                                "DeletedIngress",
                                "Deleted Ingress %s",
                                existing.Namespace)
                }
                return nil
        }

        // Create new Ingress
        if existing == nil {
                _, err := c.client.NetworkingV1().Ingresses(ingress.Namespace).Create(ctx, ingress, metav1.CreateOptions{})
                if err != nil {
                        return err
                }
                c.recorder.Eventf(
                        stack,
                        apiv1.EventTypeNormal,
                        "CreatedIngress",
                        "Created Ingress %s",
                        ingress.Name)
                return nil
        }

        // Check if we need to update the Ingress
        if core.IsResourceUpToDate(stack, existing.ObjectMeta) {
                return nil
        }

        updated := existing.DeepCopy()
        syncObjectMeta(updated, ingress)
        updated.Spec = ingress.Spec

        _, err = c.client.NetworkingV1().Ingresses(updated.Namespace).Update(ctx, updated, metav1.UpdateOptions{})
        if err != nil {
                return err
        }
        c.recorder.Eventf(
                stack,
                apiv1.EventTypeNormal,
                "UpdatedIngress",
                "Updated Ingress %s",
                ingress.Name)
        return nil
}

func (c *StackSetController) ReconcileStackRouteGroup(ctx context.Context, stack *zv1.Stack, existing *rgv1.RouteGroup, generateUpdated func() (*rgv1.RouteGroup, error)) error {
        routegroup, err := generateUpdated()
        if err != nil {
                return err
        }

        // RouteGroup removed
        if routegroup == nil {
                if existing != nil {
                        err := c.client.RouteGroupV1().RouteGroups(existing.Namespace).Delete(ctx, existing.Name, metav1.DeleteOptions{})
                        if err != nil {
                                return err
                        }
                        c.recorder.Eventf(
                                stack,
                                apiv1.EventTypeNormal,
                                "DeletedRouteGroup",
                                "Deleted RouteGroup %s",
                                existing.Namespace)
                }
                return nil
        }

        // Create new RouteGroup
        if existing == nil {
                _, err := c.client.RouteGroupV1().RouteGroups(routegroup.Namespace).Create(ctx, routegroup, metav1.CreateOptions{})
                if err != nil {
                        return err
                }
                c.recorder.Eventf(
                        stack,
                        apiv1.EventTypeNormal,
                        "CreatedRouteGroup",
                        "Created RouteGroup %s",
                        routegroup.Name)
                return nil
        }

        // Check if we need to update the RouteGroup
        if core.IsResourceUpToDate(stack, existing.ObjectMeta) {
                return nil
        }

        updated := existing.DeepCopy()
        syncObjectMeta(updated, routegroup)
        updated.Spec = routegroup.Spec

        _, err = c.client.RouteGroupV1().RouteGroups(updated.Namespace).Update(ctx, updated, metav1.UpdateOptions{})
        if err != nil {
                return err
        }
        c.recorder.Eventf(
                stack,
                apiv1.EventTypeNormal,
                "UpdatedRouteGroup",
                "Updated RouteGroup %s",
                routegroup.Name)
        return nil
}

1	package controller
2
3	import (
4	"context"
5	"fmt"
6	"strconv"
7
8	rgv1 "github.com/szuecs/routegroup-client/apis/zalando.org/v1"
9	zv1 "github.com/zalando-incubator/stackset-controller/pkg/apis/zalando.org/v1"
10	"github.com/zalando-incubator/stackset-controller/pkg/core"
11	apps "k8s.io/api/apps/v1"
12	v2 "k8s.io/api/autoscaling/v2"
13	apiv1 "k8s.io/api/core/v1"
14	networking "k8s.io/api/networking/v1"
15	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
16	)
17
18	func pint32Equal(p1, p2 *int32) bool {	1✔
19	if p1 == nil && p2 == nil {	1✔
20	return true	×
21	}	×
22	if p1 != nil && p2 != nil {	2✔
23	return p1 == p2	1✔
24	}	1✔
25	return false	×
26	}
27
28	func isHPATrafficWeightUpdated(stack zv1.Stack, hpa v2.HorizontalPodAutoscaler) (bool, error) {	1✔
29	// If resource has external RPS metric we need to check if this reconcile was	1✔
30	// triggered by traffic switch, in this case its necessary to update the weight	1✔
31	// annotation for the specific metric.	1✔
32	for _, metric := range hpa.Spec.Metrics {	2✔
33	if metric.Type == v2.ExternalMetricSourceType {	2✔
34	t, ok := metric.External.Metric.Selector.MatchLabels["type"]	1✔
35	if ok && t == "requests-per-second" {	2✔
36	stackWeight := stack.Status.ActualTrafficWeight	1✔
37	key := fmt.Sprintf(	1✔
38	"metric-config.external.%s.requests-per-second/weight",	1✔
39	metric.External.Metric.Name,	1✔
40	)	1✔
41	hpaWeight, exist := hpa.Annotations[key]	1✔
42	w, err := strconv.ParseFloat(hpaWeight, 64)	1✔
43	if err != nil {	1✔
44	return false, err	×
45	}	×
46	return exist && w == stackWeight, nil	1✔
47	}
48	}
49	}
50
51	return true, nil	1✔
52	}
53
54	// syncObjectMeta copies metadata elements such as labels or annotations from source to target
55	func syncObjectMeta(target, source metav1.Object) {	1✔
56	target.SetLabels(source.GetLabels())	1✔
57	target.SetAnnotations(source.GetAnnotations())	1✔
58	}	1✔
59
60	func (c StackSetController) ReconcileStackDeployment(ctx context.Context, stack zv1.Stack, existing apps.Deployment, generateUpdated func() apps.Deployment) error {	1✔
61	deployment := generateUpdated()	1✔
62		1✔
63	// Create new deployment	1✔
64	if existing == nil {	2✔
65	_, err := c.client.AppsV1().Deployments(deployment.Namespace).Create(ctx, deployment, metav1.CreateOptions{})	1✔
66	if err != nil {	1✔
67	return err	×
68	}	×
69	c.recorder.Eventf(	1✔
70	stack,	1✔
71	apiv1.EventTypeNormal,	1✔
72	"CreatedDeployment",	1✔
73	"Created Deployment %s",	1✔
74	deployment.Name)	1✔
75	return nil	1✔
76	}
77
78	// Check if we need to update the deployment
79	if core.IsResourceUpToDate(stack, existing.ObjectMeta) && pint32Equal(existing.Spec.Replicas, deployment.Spec.Replicas) {	2✔
80	return nil	1✔
81	}	1✔
82
83	updated := existing.DeepCopy()	1✔
84	syncObjectMeta(updated, deployment)	1✔
85	updated.Spec = deployment.Spec	1✔
86	updated.Spec.Selector = existing.Spec.Selector	1✔
87		1✔
88	_, err := c.client.AppsV1().Deployments(updated.Namespace).Update(ctx, updated, metav1.UpdateOptions{})	1✔
89	if err != nil {	1✔
90	return err	×
91	}	×
92	c.recorder.Eventf(	1✔
93	stack,	1✔
94	apiv1.EventTypeNormal,	1✔
95	"UpdatedDeployment",	1✔
96	"Updated Deployment %s",	1✔
97	deployment.Name)	1✔
98	return nil	1✔
99	}
100
101	func (c StackSetController) ReconcileStackHPA(ctx context.Context, stack zv1.Stack, existing v2.HorizontalPodAutoscaler, generateUpdated func() (v2.HorizontalPodAutoscaler, error)) error {	1✔
102	hpa, err := generateUpdated()	1✔
103	if err != nil {	1✔
104	return err	×
105	}	×
106
107	// HPA removed
108	if hpa == nil {	2✔
109	if existing != nil {	2✔
110	err := c.client.AutoscalingV2().HorizontalPodAutoscalers(existing.Namespace).Delete(ctx, existing.Name, metav1.DeleteOptions{})	1✔
111	if err != nil {	1✔
112	return err	×
113	}	×
114	c.recorder.Eventf(	1✔
115	stack,	1✔
116	apiv1.EventTypeNormal,	1✔
117	"DeletedHPA",	1✔
118	"Deleted HPA %s",	1✔
119	existing.Namespace)	1✔
120	}
121	return nil	1✔
122	}
123
124	// Create new HPA
125	if existing == nil {	2✔
126	_, err := c.client.AutoscalingV2().HorizontalPodAutoscalers(hpa.Namespace).Create(ctx, hpa, metav1.CreateOptions{})	1✔
127	if err != nil {	1✔
128	return err	×
129	}	×
130	c.recorder.Eventf(	1✔
131	stack,	1✔
132	apiv1.EventTypeNormal,	1✔
133	"CreatedHPA",	1✔
134	"Created HPA %s",	1✔
135	hpa.Name)	1✔
136	return nil	1✔
137	}
138
139	// Check if we need to update the HPA
140	weightUpdated, err := isHPATrafficWeightUpdated(stack, existing)	1✔
141	if err != nil {	1✔
142	return err	×
143	}	×
144	if core.IsResourceUpToDate(stack, existing.ObjectMeta) &&	1✔
145	pint32Equal(existing.Spec.MinReplicas, hpa.Spec.MinReplicas) &&	1✔
146	weightUpdated {	2✔
147	return nil	1✔
148	}	1✔
149
150	updated := existing.DeepCopy()	1✔
151	syncObjectMeta(updated, hpa)	1✔
152	updated.Spec = hpa.Spec	1✔
153		1✔
154	_, err = c.client.AutoscalingV2().HorizontalPodAutoscalers(updated.Namespace).Update(ctx, updated, metav1.UpdateOptions{})	1✔
155	if err != nil {	1✔
156	return err	×
157	}	×
158	c.recorder.Eventf(	1✔
159	stack,	1✔
160	apiv1.EventTypeNormal,	1✔
161	"UpdatedHPA",	1✔
162	"Updated HPA %s",	1✔
163	hpa.Name)	1✔
164	return nil	1✔
165	}
166
167	func (c StackSetController) ReconcileStackService(ctx context.Context, stack zv1.Stack, existing apiv1.Service, generateUpdated func() (apiv1.Service, error)) error {	1✔
168	service, err := generateUpdated()	1✔
169	if err != nil {	1✔
170	return err	×
171	}	×
172
173	// Create new service
174	if existing == nil {	2✔
175	_, err := c.client.CoreV1().Services(service.Namespace).Create(ctx, service, metav1.CreateOptions{})	1✔
176	if err != nil {	1✔
177	return err	×
178	}	×
179	c.recorder.Eventf(	1✔
180	stack,	1✔
181	apiv1.EventTypeNormal,	1✔
182	"CreatedService",	1✔
183	"Created Service %s",	1✔
184	service.Name)	1✔
185	return nil	1✔
186	}
187
188	// Check if we need to update the service
189	if core.IsResourceUpToDate(stack, existing.ObjectMeta) {	2✔
190	return nil	1✔
191	}	1✔
192
193	updated := existing.DeepCopy()	1✔
194	syncObjectMeta(updated, service)	1✔
195	updated.Spec = service.Spec	1✔
196	updated.Spec.ClusterIP = existing.Spec.ClusterIP // ClusterIP is immutable	1✔
197		1✔
198	_, err = c.client.CoreV1().Services(updated.Namespace).Update(ctx, updated, metav1.UpdateOptions{})	1✔
199	if err != nil {	1✔
200	return err	×
201	}	×
202	c.recorder.Eventf(	1✔
203	stack,	1✔
204	apiv1.EventTypeNormal,	1✔
205	"UpdatedService",	1✔
206	"Updated Service %s",	1✔
207	service.Name)	1✔
208	return nil	1✔
209	}
210
211	func (c StackSetController) ReconcileStackIngress(ctx context.Context, stack zv1.Stack, existing networking.Ingress, generateUpdated func() (networking.Ingress, error)) error {	1✔
212	ingress, err := generateUpdated()	1✔
213	if err != nil {	1✔
214	return err	×
215	}	×
216
217	// Ingress removed
218	if ingress == nil {	2✔
219	if existing != nil {	2✔
220	err := c.client.NetworkingV1().Ingresses(existing.Namespace).Delete(ctx, existing.Name, metav1.DeleteOptions{})	1✔
221	if err != nil {	1✔
222	return err	×
223	}	×
224	c.recorder.Eventf(	1✔
225	stack,	1✔
226	apiv1.EventTypeNormal,	1✔
227	"DeletedIngress",	1✔
228	"Deleted Ingress %s",	1✔
229	existing.Namespace)	1✔
230	}
231	return nil	1✔
232	}
233
234	// Create new Ingress
235	if existing == nil {	2✔
236	_, err := c.client.NetworkingV1().Ingresses(ingress.Namespace).Create(ctx, ingress, metav1.CreateOptions{})	1✔
237	if err != nil {	1✔
238	return err	×
239	}	×
240	c.recorder.Eventf(	1✔
241	stack,	1✔
242	apiv1.EventTypeNormal,	1✔
243	"CreatedIngress",	1✔
244	"Created Ingress %s",	1✔
245	ingress.Name)	1✔
246	return nil	1✔
247	}
248
249	// Check if we need to update the Ingress
250	if core.IsResourceUpToDate(stack, existing.ObjectMeta) {	2✔
251	return nil	1✔
252	}	1✔
253
254	updated := existing.DeepCopy()	1✔
255	syncObjectMeta(updated, ingress)	1✔
256	updated.Spec = ingress.Spec	1✔
257		1✔
258	_, err = c.client.NetworkingV1().Ingresses(updated.Namespace).Update(ctx, updated, metav1.UpdateOptions{})	1✔
259	if err != nil {	1✔
260	return err	×
261	}	×
262	c.recorder.Eventf(	1✔
263	stack,	1✔
264	apiv1.EventTypeNormal,	1✔
265	"UpdatedIngress",	1✔
266	"Updated Ingress %s",	1✔
267	ingress.Name)	1✔
268	return nil	1✔
269	}
270
271	func (c StackSetController) ReconcileStackRouteGroup(ctx context.Context, stack zv1.Stack, existing rgv1.RouteGroup, generateUpdated func() (rgv1.RouteGroup, error)) error {	1✔
272	routegroup, err := generateUpdated()	1✔
273	if err != nil {	1✔
274	return err	×
275	}	×
276
277	// RouteGroup removed
278	if routegroup == nil {	2✔
279	if existing != nil {	2✔
280	err := c.client.RouteGroupV1().RouteGroups(existing.Namespace).Delete(ctx, existing.Name, metav1.DeleteOptions{})	1✔
281	if err != nil {	1✔
282	return err	×
283	}	×
284	c.recorder.Eventf(	1✔
285	stack,	1✔
286	apiv1.EventTypeNormal,	1✔
287	"DeletedRouteGroup",	1✔
288	"Deleted RouteGroup %s",	1✔
289	existing.Namespace)	1✔
290	}
291	return nil	1✔
292	}
293
294	// Create new RouteGroup
295	if existing == nil {	2✔
296	_, err := c.client.RouteGroupV1().RouteGroups(routegroup.Namespace).Create(ctx, routegroup, metav1.CreateOptions{})	1✔
297	if err != nil {	1✔
298	return err	×
299	}	×
300	c.recorder.Eventf(	1✔
301	stack,	1✔
302	apiv1.EventTypeNormal,	1✔
303	"CreatedRouteGroup",	1✔
304	"Created RouteGroup %s",	1✔
305	routegroup.Name)	1✔
306	return nil	1✔
307	}
308
309	// Check if we need to update the RouteGroup
310	if core.IsResourceUpToDate(stack, existing.ObjectMeta) {	2✔
311	return nil	1✔
312	}	1✔
313
314	updated := existing.DeepCopy()	1✔
315	syncObjectMeta(updated, routegroup)	1✔
316	updated.Spec = routegroup.Spec	1✔
317		1✔
318	_, err = c.client.RouteGroupV1().RouteGroups(updated.Namespace).Update(ctx, updated, metav1.UpdateOptions{})	1✔
319	if err != nil {	1✔
320	return err	×
321	}	×
322	c.recorder.Eventf(	1✔
323	stack,	1✔
324	apiv1.EventTypeNormal,	1✔
325	"UpdatedRouteGroup",	1✔
326	"Updated RouteGroup %s",	1✔
327	routegroup.Name)	1✔
328	return nil	1✔
329	}

zalando-incubator / stackset-controller / 5738153381

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