From 9dcc95584d06f62fa74a4c2c3d0e99499de95a38 Mon Sep 17 00:00:00 2001 From: Alex Emelyanov Date: Mon, 16 Mar 2020 18:05:43 +0300 Subject: [PATCH] Fix scalers leaking --- pkg/handler/scale_handler.go | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/pkg/handler/scale_handler.go b/pkg/handler/scale_handler.go index fa6e3a109d7..ca5669e5e17 100644 --- a/pkg/handler/scale_handler.go +++ b/pkg/handler/scale_handler.go @@ -186,6 +186,12 @@ func (h *ScaleHandler) resolveConfigValue(configKeyRef *corev1.ConfigMapKeySelec return string(configMap.Data[keyName]), nil } +func closeScalers(scalers []scalers.Scaler) { + for _, scaler := range scalers { + defer scaler.Close() + } +} + // GetDeploymentScalers returns list of Scalers and Deployment for the specified ScaledObject func (h *ScaleHandler) GetDeploymentScalers(scaledObject *kedav1alpha1.ScaledObject) ([]scalers.Scaler, *appsv1.Deployment, error) { scalers := []scalers.Scaler{} @@ -214,7 +220,8 @@ func (h *ScaleHandler) GetDeploymentScalers(scaledObject *kedav1alpha1.ScaledObj serviceAccount := &v1.ServiceAccount{} err = h.client.Get(context.TODO(), types.NamespacedName{Name: serviceAccountName, Namespace: scaledObject.GetNamespace()}, serviceAccount) if err != nil { - return scalers, nil, fmt.Errorf("error getting deployment: %s", err) + closeScalers(scalers) + return []scalers.Scaler{}, nil, fmt.Errorf("error getting deployment: %s", err) } authParams["awsRoleArn"] = serviceAccount.Annotations[kedav1alpha1.PodIdentityAnnotationEKS] } else if podIdentity == kedav1alpha1.PodIdentityProviderAwsKiam { @@ -223,7 +230,8 @@ func (h *ScaleHandler) GetDeploymentScalers(scaledObject *kedav1alpha1.ScaledObj scaler, err := h.getScaler(scaledObject.Name, scaledObject.Namespace, trigger.Type, resolvedEnv, trigger.Metadata, authParams, podIdentity) if err != nil { - return scalers, nil, fmt.Errorf("error getting scaler for trigger #%d: %s", i, err) + closeScalers(scalers) + return []scalers.Scaler{}, nil, fmt.Errorf("error getting scaler for trigger #%d: %s", i, err) } scalers = append(scalers, scaler) @@ -244,7 +252,8 @@ func (h *ScaleHandler) getJobScalers(scaledObject *kedav1alpha1.ScaledObject) ([ authParams, podIdentity := h.parseJobAuthRef(trigger.AuthenticationRef, scaledObject) scaler, err := h.getScaler(scaledObject.Name, scaledObject.Namespace, trigger.Type, resolvedEnv, trigger.Metadata, authParams, podIdentity) if err != nil { - return scalers, fmt.Errorf("error getting scaler for trigger #%d: %s", i, err) + closeScalers(scalers) + return []scalers.Scaler{}, fmt.Errorf("error getting scaler for trigger #%d: %s", i, err) } scalers = append(scalers, scaler)