Propagate contexts down scaler call stacks (#2202)

Signed-off-by: Aaron <aaron@ecomaz.net>
kedacore · Oct 26, 2021 · d1bed54 · d1bed54
1 parent 04aac84
commit d1bed54
Show file tree

Hide file tree

Showing 92 changed files with 406 additions and 365 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -22,7 +22,6 @@
 
 ### New
 
-- TODO ([#XXX](https://github.com/kedacore/keda/pull/XXX))
 - ScaledJob: introduce MultipleScalersCalculation ([#2016](https://github.com/kedacore/keda/pull/2016))
 - Add Graphite Scaler ([#1628](https://github.com/kedacore/keda/pull/2092))
 
@@ -54,6 +53,7 @@
 
 ### Other
 
+- Ensure that `context.Context` values are passed down the stack from all scaler gRPC handler implementation to scaler implementation code ([#2202](https://github.com/kedacore/keda/pull/2202))
 - Migrate to Kubebuilder v3 ([#2082](https://github.com/kedacore/keda/pull/2082))
     - API path has been changed: `github.com/kedacore/keda/v2/api/v1alpha1` -> `github.com/kedacore/keda/v2/apis/keda/v1alpha1`
 - Use Patch to set FallbackCondition on ScaledObject.Status ([#2037](https://github.com/kedacore/keda/pull/2037))

diff --git a/apis/keda/v1alpha1/zz_generated.deepcopy.go b/apis/keda/v1alpha1/zz_generated.deepcopy.go
diff --git a/controllers/keda/hpa.go b/controllers/keda/hpa.go
@@ -40,10 +40,10 @@ const (
 )
 
 // createAndDeployNewHPA creates and deploy HPA in the cluster for specified ScaledObject
-func (r *ScaledObjectReconciler) createAndDeployNewHPA(logger logr.Logger, scaledObject *kedav1alpha1.ScaledObject, gvkr *kedav1alpha1.GroupVersionKindResource) error {
+func (r *ScaledObjectReconciler) createAndDeployNewHPA(ctx context.Context, logger logr.Logger, scaledObject *kedav1alpha1.ScaledObject, gvkr *kedav1alpha1.GroupVersionKindResource) error {
 	hpaName := getHPAName(scaledObject)
 	logger.Info("Creating a new HPA", "HPA.Namespace", scaledObject.Namespace, "HPA.Name", hpaName)
-	hpa, err := r.newHPAForScaledObject(logger, scaledObject, gvkr)
+	hpa, err := r.newHPAForScaledObject(ctx, logger, scaledObject, gvkr)
 	if err != nil {
 		logger.Error(err, "Failed to create new HPA resource", "HPA.Namespace", scaledObject.Namespace, "HPA.Name", hpaName)
 		return err
@@ -59,8 +59,8 @@ func (r *ScaledObjectReconciler) createAndDeployNewHPA(logger logr.Logger, scale
 }
 
 // newHPAForScaledObject returns HPA as it is specified in ScaledObject
-func (r *ScaledObjectReconciler) newHPAForScaledObject(logger logr.Logger, scaledObject *kedav1alpha1.ScaledObject, gvkr *kedav1alpha1.GroupVersionKindResource) (*autoscalingv2beta2.HorizontalPodAutoscaler, error) {
-	scaledObjectMetricSpecs, err := r.getScaledObjectMetricSpecs(logger, scaledObject)
+func (r *ScaledObjectReconciler) newHPAForScaledObject(ctx context.Context, logger logr.Logger, scaledObject *kedav1alpha1.ScaledObject, gvkr *kedav1alpha1.GroupVersionKindResource) (*autoscalingv2beta2.HorizontalPodAutoscaler, error) {
+	scaledObjectMetricSpecs, err := r.getScaledObjectMetricSpecs(ctx, logger, scaledObject)
 	if err != nil {
 		return nil, err
 	}
@@ -120,8 +120,8 @@ func (r *ScaledObjectReconciler) newHPAForScaledObject(logger logr.Logger, scale
 }
 
 // updateHPAIfNeeded checks whether update of HPA is needed
-func (r *ScaledObjectReconciler) updateHPAIfNeeded(logger logr.Logger, scaledObject *kedav1alpha1.ScaledObject, foundHpa *autoscalingv2beta2.HorizontalPodAutoscaler, gvkr *kedav1alpha1.GroupVersionKindResource) error {
-	hpa, err := r.newHPAForScaledObject(logger, scaledObject, gvkr)
+func (r *ScaledObjectReconciler) updateHPAIfNeeded(ctx context.Context, logger logr.Logger, scaledObject *kedav1alpha1.ScaledObject, foundHpa *autoscalingv2beta2.HorizontalPodAutoscaler, gvkr *kedav1alpha1.GroupVersionKindResource) error {
+	hpa, err := r.newHPAForScaledObject(ctx, logger, scaledObject, gvkr)
 	if err != nil {
 		logger.Error(err, "Failed to create new HPA resource", "HPA.Namespace", scaledObject.Namespace, "HPA.Name", getHPAName(scaledObject))
 		return err
@@ -155,19 +155,19 @@ func (r *ScaledObjectReconciler) updateHPAIfNeeded(logger logr.Logger, scaledObj
 }
 
 // getScaledObjectMetricSpecs returns MetricSpec for HPA, generater from Triggers defitinion in ScaledObject
-func (r *ScaledObjectReconciler) getScaledObjectMetricSpecs(logger logr.Logger, scaledObject *kedav1alpha1.ScaledObject) ([]autoscalingv2beta2.MetricSpec, error) {
+func (r *ScaledObjectReconciler) getScaledObjectMetricSpecs(ctx context.Context, logger logr.Logger, scaledObject *kedav1alpha1.ScaledObject) ([]autoscalingv2beta2.MetricSpec, error) {
 	var scaledObjectMetricSpecs []autoscalingv2beta2.MetricSpec
 	var externalMetricNames []string
 	var resourceMetricNames []string
 
-	scalers, err := r.scaleHandler.GetScalers(scaledObject)
+	scalers, err := r.scaleHandler.GetScalers(ctx, scaledObject)
 	if err != nil {
 		logger.Error(err, "Error getting scalers")
 		return nil, err
 	}
 
 	for _, scaler := range scalers {
-		metricSpecs := scaler.GetMetricSpecForScaling()
+		metricSpecs := scaler.GetMetricSpecForScaling(ctx)
 
 		for _, metricSpec := range metricSpecs {
 			if metricSpec.Resource != nil {
@@ -187,7 +187,7 @@ func (r *ScaledObjectReconciler) getScaledObjectMetricSpecs(logger logr.Logger,
 			}
 		}
 		scaledObjectMetricSpecs = append(scaledObjectMetricSpecs, metricSpecs...)
-		scaler.Close()
+		scaler.Close(ctx)
 	}
 
 	// sort metrics in ScaledObject, this way we always check the same resource in Reconcile loop and we can prevent unnecessary HPA updates,

diff --git a/controllers/keda/hpa_test.go b/controllers/keda/hpa_test.go
@@ -17,6 +17,8 @@ limitations under the License.
 package keda
 
 import (
+	"context"
+
 	"github.com/go-logr/logr"
 	"github.com/golang/mock/gomock"
 	. "github.com/onsi/ginkgo"
@@ -75,7 +77,7 @@ var _ = Describe("hpa", func() {
 			capturedScaledObject = *scaledObject
 		})
 
-		_, err := reconciler.getScaledObjectMetricSpecs(logger, scaledObject)
+		_, err := reconciler.getScaledObjectMetricSpecs(context.Background(), logger, scaledObject)
 
 		Expect(err).ToNot(HaveOccurred())
 		Expect(capturedScaledObject.Status.Health).To(BeEmpty())
@@ -102,7 +104,7 @@ var _ = Describe("hpa", func() {
 			capturedScaledObject = *scaledObject
 		})
 
-		_, err := reconciler.getScaledObjectMetricSpecs(logger, scaledObject)
+		_, err := reconciler.getScaledObjectMetricSpecs(context.Background(), logger, scaledObject)
 
 		expectedHealth := make(map[string]v1alpha1.HealthStatus)
 		expectedHealth["some metric name"] = v1alpha1.HealthStatus{
@@ -136,9 +138,10 @@ func setupTest(health map[string]v1alpha1.HealthStatus, scaler *mock_scalers.Moc
 		},
 	}
 	metricSpecs := []v2beta2.MetricSpec{metricSpec}
-	scaler.EXPECT().GetMetricSpecForScaling().Return(metricSpecs)
-	scaler.EXPECT().Close()
-	scaleHandler.EXPECT().GetScalers(gomock.Eq(scaledObject)).Return(scalers, nil)
+	ctx := context.Background()
+	scaler.EXPECT().GetMetricSpecForScaling(ctx).Return(metricSpecs)
+	scaler.EXPECT().Close(ctx)
+	scaleHandler.EXPECT().GetScalers(context.Background(), gomock.Eq(scaledObject)).Return(scalers, nil)
 
 	return scaledObject
 }
diff --git a/controllers/keda/scaledjob_controller.go b/controllers/keda/scaledjob_controller.go
@@ -110,7 +110,7 @@ func (r *ScaledJobReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
 		reqLogger.Error(err, "scaledJob.spec.jobTargetRef not found")
 		return ctrl.Result{}, err
 	}
-	msg, err := r.reconcileScaledJob(reqLogger, scaledJob)
+	msg, err := r.reconcileScaledJob(ctx, reqLogger, scaledJob)
 	conditions := scaledJob.Status.Conditions.DeepCopy()
 	if err != nil {
 		reqLogger.Error(err, msg)
@@ -133,14 +133,14 @@ func (r *ScaledJobReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
 }
 
 // reconcileScaledJob implements reconciler logic for K8s Jobs based ScaledJob
-func (r *ScaledJobReconciler) reconcileScaledJob(logger logr.Logger, scaledJob *kedav1alpha1.ScaledJob) (string, error) {
+func (r *ScaledJobReconciler) reconcileScaledJob(ctx context.Context, logger logr.Logger, scaledJob *kedav1alpha1.ScaledJob) (string, error) {
 	msg, err := r.deletePreviousVersionScaleJobs(logger, scaledJob)
 	if err != nil {
 		return msg, err
 	}
 
 	// Check ScaledJob is Ready or not
-	_, err = r.scaleHandler.GetScalers(scaledJob)
+	_, err = r.scaleHandler.GetScalers(ctx, scaledJob)
 	if err != nil {
 		logger.Error(err, "Error getting scalers")
 		return "Failed to ensure ScaledJob is correctly created", err

diff --git a/controllers/keda/scaledobject_controller.go b/controllers/keda/scaledobject_controller.go
@@ -175,7 +175,7 @@ func (r *ScaledObjectReconciler) Reconcile(ctx context.Context, req ctrl.Request
 	}
 
 	// reconcile ScaledObject and set status appropriately
-	msg, err := r.reconcileScaledObject(reqLogger, scaledObject)
+	msg, err := r.reconcileScaledObject(ctx, reqLogger, scaledObject)
 	conditions := scaledObject.Status.Conditions.DeepCopy()
 	if err != nil {
 		reqLogger.Error(err, msg)
@@ -199,7 +199,7 @@ func (r *ScaledObjectReconciler) Reconcile(ctx context.Context, req ctrl.Request
 }
 
 // reconcileScaledObject implements reconciler logic for ScaleObject
-func (r *ScaledObjectReconciler) reconcileScaledObject(logger logr.Logger, scaledObject *kedav1alpha1.ScaledObject) (string, error) {
+func (r *ScaledObjectReconciler) reconcileScaledObject(ctx context.Context, logger logr.Logger, scaledObject *kedav1alpha1.ScaledObject) (string, error) {
 	// Check scale target Name is specified
 	if scaledObject.Spec.ScaleTargetRef.Name == "" {
 		err := fmt.Errorf("ScaledObject.spec.scaleTargetRef.name is missing")
@@ -224,7 +224,7 @@ func (r *ScaledObjectReconciler) reconcileScaledObject(logger logr.Logger, scale
 	}
 
 	// Create a new HPA or update existing one according to ScaledObject
-	newHPACreated, err := r.ensureHPAForScaledObjectExists(logger, scaledObject, &gvkr)
+	newHPACreated, err := r.ensureHPAForScaledObjectExists(ctx, logger, scaledObject, &gvkr)
 	if err != nil {
 		return "Failed to ensure HPA is correctly created for ScaledObject", err
 	}
@@ -349,14 +349,14 @@ func (r *ScaledObjectReconciler) checkReplicaCountBoundsAreValid(scaledObject *k
 }
 
 // ensureHPAForScaledObjectExists ensures that in cluster exist up-to-date HPA for specified ScaledObject, returns true if a new HPA was created
-func (r *ScaledObjectReconciler) ensureHPAForScaledObjectExists(logger logr.Logger, scaledObject *kedav1alpha1.ScaledObject, gvkr *kedav1alpha1.GroupVersionKindResource) (bool, error) {
+func (r *ScaledObjectReconciler) ensureHPAForScaledObjectExists(ctx context.Context, logger logr.Logger, scaledObject *kedav1alpha1.ScaledObject, gvkr *kedav1alpha1.GroupVersionKindResource) (bool, error) {
 	hpaName := getHPAName(scaledObject)
 	foundHpa := &autoscalingv2beta2.HorizontalPodAutoscaler{}
 	// Check if HPA for this ScaledObject already exists
 	err := r.Client.Get(context.TODO(), types.NamespacedName{Name: hpaName, Namespace: scaledObject.Namespace}, foundHpa)
 	if err != nil && errors.IsNotFound(err) {
 		// HPA wasn't found -> let's create a new one
-		err = r.createAndDeployNewHPA(logger, scaledObject, gvkr)
+		err = r.createAndDeployNewHPA(ctx, logger, scaledObject, gvkr)
 		if err != nil {
 			return false, err
 		}
@@ -372,7 +372,7 @@ func (r *ScaledObjectReconciler) ensureHPAForScaledObjectExists(logger logr.Logg
 	}
 
 	// HPA was found -> let's check if we need to update it
-	err = r.updateHPAIfNeeded(logger, scaledObject, foundHpa, gvkr)
+	err = r.updateHPAIfNeeded(ctx, logger, scaledObject, foundHpa, gvkr)
 	if err != nil {
 		logger.Error(err, "Failed to check HPA for possible update")
 		return false, err

diff --git a/controllers/keda/scaledobject_controller_test.go b/controllers/keda/scaledobject_controller_test.go
@@ -100,20 +100,20 @@ var _ = Describe("ScaledObjectController", func() {
 					}
 
 					testScalers = append(testScalers, s)
-					for _, metricSpec := range s.GetMetricSpecForScaling() {
+					for _, metricSpec := range s.GetMetricSpecForScaling(context.Background()) {
 						if metricSpec.External != nil {
 							expectedExternalMetricNames = append(expectedExternalMetricNames, metricSpec.External.Metric.Name)
 						}
 					}
 				}
 
 				// Set up expectations
-				mockScaleHandler.EXPECT().GetScalers(uniquelyNamedScaledObject).Return(testScalers, nil)
+				mockScaleHandler.EXPECT().GetScalers(context.Background(), uniquelyNamedScaledObject).Return(testScalers, nil)
 				mockClient.EXPECT().Status().Return(mockStatusWriter)
 				mockStatusWriter.EXPECT().Patch(gomock.Any(), gomock.Any(), gomock.Any())
 
 				// Call function to be tested
-				metricSpecs, err := metricNameTestReconciler.getScaledObjectMetricSpecs(testLogger, uniquelyNamedScaledObject)
+				metricSpecs, err := metricNameTestReconciler.getScaledObjectMetricSpecs(context.Background(), testLogger, uniquelyNamedScaledObject)
 
 				// Test that the status was updated with metric names
 				Ω(uniquelyNamedScaledObject.Status.ExternalMetricNames).Should(Equal(expectedExternalMetricNames))
@@ -139,19 +139,19 @@ var _ = Describe("ScaledObjectController", func() {
 				if err != nil {
 					Fail(err.Error())
 				}
-				for _, metricSpec := range s.GetMetricSpecForScaling() {
+				for _, metricSpec := range s.GetMetricSpecForScaling(context.Background()) {
 					if metricSpec.External != nil {
 						expectedExternalMetricNames = append(expectedExternalMetricNames, metricSpec.External.Metric.Name)
 					}
 				}
 
 				// Set up expectations
-				mockScaleHandler.EXPECT().GetScalers(uniquelyNamedScaledObject).Return([]scalers.Scaler{s}, nil)
+				mockScaleHandler.EXPECT().GetScalers(context.Background(), uniquelyNamedScaledObject).Return([]scalers.Scaler{s}, nil)
 				mockClient.EXPECT().Status().Return(mockStatusWriter)
 				mockStatusWriter.EXPECT().Patch(gomock.Any(), gomock.Any(), gomock.Any())
 
 				// Call function to be tested
-				metricSpecs, err := metricNameTestReconciler.getScaledObjectMetricSpecs(testLogger, uniquelyNamedScaledObject)
+				metricSpecs, err := metricNameTestReconciler.getScaledObjectMetricSpecs(context.Background(), testLogger, uniquelyNamedScaledObject)
 
 				// Test that the status was updated
 				Ω(uniquelyNamedScaledObject.Status.ExternalMetricNames).Should(Equal(expectedExternalMetricNames))
@@ -186,10 +186,10 @@ var _ = Describe("ScaledObjectController", func() {
 				}
 
 				// Set up expectations
-				mockScaleHandler.EXPECT().GetScalers(duplicateNamedScaledObject).Return(testScalers, nil)
+				mockScaleHandler.EXPECT().GetScalers(context.Background(), duplicateNamedScaledObject).Return(testScalers, nil)
 
 				// Call function tobe tested
-				metricSpecs, err := metricNameTestReconciler.getScaledObjectMetricSpecs(testLogger, duplicateNamedScaledObject)
+				metricSpecs, err := metricNameTestReconciler.getScaledObjectMetricSpecs(context.Background(), testLogger, duplicateNamedScaledObject)
 
 				// Test that the status was not updated
 				Ω(duplicateNamedScaledObject.Status.ExternalMetricNames).Should(BeNil())

diff --git a/pkg/mock/mock_client/mock_interfaces.go b/pkg/mock/mock_client/mock_interfaces.go
diff --git a/pkg/mock/mock_scale/mock_interfaces.go b/pkg/mock/mock_scale/mock_interfaces.go
diff --git a/pkg/mock/mock_scaler/mock_scaler.go b/pkg/mock/mock_scaler/mock_scaler.go
diff --git a/pkg/mock/mock_scaling/mock_interface.go b/pkg/mock/mock_scaling/mock_interface.go