Use Keda context in some cases

Signed-off-by: Jeroen Bobbeldijk <jeroen@klippa.com>

pkg/scalers/redis_scaler.go

@@ -24,14 +24,12 @@ const (
 	defaultEnableTLS        = false
 )
 
-var ctx = context.Background()
-
 type redisAddressParser func(metadata, resolvedEnv, authParams map[string]string) (redisConnectionInfo, error)
 
 type redisScaler struct {
 	metadata        *redisMetadata
 	closeFn         func() error
-	getListLengthFn func() (int64, error)
+	getListLengthFn func(ctx context.Context) (int64, error)
 }
 
 type redisConnectionInfo struct {
@@ -105,7 +103,7 @@ func createClusteredRedisScaler(meta *redisMetadata, script string) (Scaler, err
 		return nil
 	}
 
-	listLengthFn := func() (int64, error) {
+	listLengthFn := func(ctx context.Context) (int64, error) {
 		cmd := client.Eval(ctx, script, []string{meta.listName})
 		if cmd.Err() != nil {
 			return -1, cmd.Err()
@@ -135,7 +133,7 @@ func createSentinelRedisScaler(meta *redisMetadata, script string) (Scaler, erro
 		return nil
 	}
 
-	listLengthFn := func() (int64, error) {
+	listLengthFn := func(ctx context.Context) (int64, error) {
 		cmd := client.Eval(ctx, script, []string{meta.listName})
 		if cmd.Err() != nil {
 			return -1, cmd.Err()
@@ -165,7 +163,7 @@ func createRedisScaler(meta *redisMetadata, script string) (Scaler, error) {
 		return nil
 	}
 
-	listLengthFn := func() (int64, error) {
+	listLengthFn := func(ctx context.Context) (int64, error) {
 		cmd := client.Eval(ctx, script, []string{meta.listName})
 		if cmd.Err() != nil {
 			return -1, cmd.Err()
@@ -219,7 +217,7 @@ func parseRedisMetadata(config *ScalerConfig, parserFn redisAddressParser) (*red
 
 // IsActive checks if there is any element in the Redis list
 func (s *redisScaler) IsActive(ctx context.Context) (bool, error) {
-	length, err := s.getListLengthFn()
+	length, err := s.getListLengthFn(ctx)
 
 	if err != nil {
 		redisLog.Error(err, "error")
@@ -253,7 +251,7 @@ func (s *redisScaler) GetMetricSpecForScaling() []v2beta2.MetricSpec {
 
 // GetMetrics connects to Redis and finds the length of the list
 func (s *redisScaler) GetMetrics(ctx context.Context, metricName string, metricSelector labels.Selector) ([]external_metrics.ExternalMetricValue, error) {
-	listLen, err := s.getListLengthFn()
+	listLen, err := s.getListLengthFn(ctx)
 
 	if err != nil {
 		redisLog.Error(err, "error getting list length")
@@ -482,7 +480,7 @@ func getRedisClusterClient(info redisConnectionInfo) (*redis.ClusterClient, erro
 
 	// confirm if connected
 	c := redis.NewClusterClient(options)
-	err := c.Ping(ctx).Err()
+	err := c.Ping(context.Background()).Err()
 	if err != nil {
 		return nil, err
 	}
@@ -507,7 +505,7 @@ func getRedisSentinelClient(info redisConnectionInfo, dbIndex int) (*redis.Clien
 
 	// confirm if connected
 	c := redis.NewFailoverClient(options)
-	err := c.Ping(ctx).Err()
+	err := c.Ping(context.Background()).Err()
 	if err != nil {
 		return nil, err
 	}
@@ -529,7 +527,7 @@ func getRedisClient(info redisConnectionInfo, dbIndex int) (*redis.Client, error
 
 	// confirm if connected
 	c := redis.NewClient(options)
-	err := c.Ping(ctx).Err()
+	err := c.Ping(context.Background()).Err()
 	if err != nil {
 		return nil, err
 	}

pkg/scalers/redis_scaler_test.go

@@ -1,6 +1,7 @@
 package scalers
 
 import (
+	"context"
 	"errors"
 	"testing"
 
@@ -79,7 +80,7 @@ func TestRedisGetMetricSpecForScaling(t *testing.T) {
 			t.Fatal("Could not parse metadata:", err)
 		}
 		closeFn := func() error { return nil }
-		lengthFn := func() (int64, error) { return -1, nil }
+		lengthFn := func(ctx context.Context) (int64, error) { return -1, nil }
 		mockRedisScaler := redisScaler{
 			meta,
 			closeFn,

pkg/scalers/redis_streams_scaler.go

@@ -33,7 +33,7 @@ const (
 type redisStreamsScaler struct {
 	metadata                 *redisStreamsMetadata
 	closeFn                  func() error
-	getPendingEntriesCountFn func() (int64, error)
+	getPendingEntriesCountFn func(ctx context.Context) (int64, error)
 }
 
 type redisStreamsMetadata struct {
@@ -82,7 +82,7 @@ func createClusteredRedisStreamsScaler(meta *redisStreamsMetadata) (Scaler, erro
 		return nil
 	}
 
-	pendingEntriesCountFn := func() (int64, error) {
+	pendingEntriesCountFn := func(ctx context.Context) (int64, error) {
 		pendingEntries, err := client.XPending(ctx, meta.streamName, meta.consumerGroupName).Result()
 		if err != nil {
 			return -1, err
@@ -111,7 +111,7 @@ func createSentinelRedisStreamsScaler(meta *redisStreamsMetadata) (Scaler, error
 		return nil
 	}
 
-	pendingEntriesCountFn := func() (int64, error) {
+	pendingEntriesCountFn := func(ctx context.Context) (int64, error) {
 		pendingEntries, err := client.XPending(ctx, meta.streamName, meta.consumerGroupName).Result()
 		if err != nil {
 			return -1, err
@@ -140,7 +140,7 @@ func createRedisStreamsScaler(meta *redisStreamsMetadata) (Scaler, error) {
 		return nil
 	}
 
-	pendingEntriesCountFn := func() (int64, error) {
+	pendingEntriesCountFn := func(ctx context.Context) (int64, error) {
 		pendingEntries, err := client.XPending(ctx, meta.streamName, meta.consumerGroupName).Result()
 		if err != nil {
 			return -1, err
@@ -201,7 +201,7 @@ func parseRedisStreamsMetadata(config *ScalerConfig, parseFn redisAddressParser)
 
 // IsActive checks if there are pending entries in the 'Pending Entries List' for consumer group of a stream
 func (s *redisStreamsScaler) IsActive(ctx context.Context) (bool, error) {
-	count, err := s.getPendingEntriesCountFn()
+	count, err := s.getPendingEntriesCountFn(ctx)
 
 	if err != nil {
 		redisStreamsLog.Error(err, "error")
@@ -233,7 +233,7 @@ func (s *redisStreamsScaler) GetMetricSpecForScaling() []v2beta2.MetricSpec {
 
 // GetMetrics fetches the number of pending entries for a consumer group in a stream
 func (s *redisStreamsScaler) GetMetrics(ctx context.Context, metricName string, metricSelector labels.Selector) ([]external_metrics.ExternalMetricValue, error) {
-	pendingEntriesCount, err := s.getPendingEntriesCountFn()
+	pendingEntriesCount, err := s.getPendingEntriesCountFn(ctx)
 
 	if err != nil {
 		redisStreamsLog.Error(err, "error fetching pending entries count")

pkg/scalers/redis_streams_scaler_test.go

@@ -1,6 +1,7 @@
 package scalers
 
 import (
+	"context"
 	"errors"
 	"strconv"
 	"testing"
@@ -141,7 +142,7 @@ func TestRedisStreamsGetMetricSpecForScaling(t *testing.T) {
 			t.Fatal("Could not parse metadata:", err)
 		}
 		closeFn := func() error { return nil }
-		getPendingEntriesCountFn := func() (int64, error) { return -1, nil }
+		getPendingEntriesCountFn := func(ctx context.Context) (int64, error) { return -1, nil }
 		mockRedisStreamsScaler := redisStreamsScaler{meta, closeFn, getPendingEntriesCountFn}
 
 		metricSpec := mockRedisStreamsScaler.GetMetricSpecForScaling()