feat: Allow scaling using redis stream length (#4277)

CHANGELOG.md

@@ -46,7 +46,7 @@ To learn more about active deprecations, we recommend checking [GitHub Discussio
 ### New
 
 - **CPU/Memory scaler**: Add support for scale to zero if there are multiple triggers([#4269](https://github.com/kedacore/keda/issues/4269))
-- TODO ([#XXX](https://github.com/kedacore/keda/issue/XXX))
+- **Redis Scalers**: Allow scaling using redis stream length ([#4277](https://github.com/kedacore/keda/issues/4277))
 - **General:** Introduce new Solr Scaler ([#4234](https://github.com/kedacore/keda/issues/4234))
 
 ### Improvements

pkg/scalers/redis_streams_scaler.go

@@ -14,13 +14,21 @@ import (
 	kedautil "github.com/kedacore/keda/v2/pkg/util"
 )
 
+type scaleFactor int8
+
+const (
+	xPendingFactor scaleFactor = iota + 1
+	xLengthFactor
+)
+
 const (
 	// defaults
-	defaultTargetPendingEntriesCount = 5
-	defaultDBIndex                   = 0
+	defaultDBIndex       = 0
+	defaultTargetEntries = 5
 
 	// metadata names
 	pendingEntriesCountMetadata = "pendingEntriesCount"
+	streamLengthMetadata        = "streamLength"
 	streamNameMetadata          = "stream"
 	consumerGroupNameMetadata   = "consumerGroup"
 	usernameMetadata            = "username"
@@ -30,15 +38,17 @@ const (
 )
 
 type redisStreamsScaler struct {
-	metricType               v2.MetricTargetType
-	metadata                 *redisStreamsMetadata
-	closeFn                  func() error
-	getPendingEntriesCountFn func(ctx context.Context) (int64, error)
-	logger                   logr.Logger
+	metricType        v2.MetricTargetType
+	metadata          *redisStreamsMetadata
+	closeFn           func() error
+	getEntriesCountFn func(ctx context.Context) (int64, error)
+	logger            logr.Logger
 }
 
 type redisStreamsMetadata struct {
+	scaleFactor               scaleFactor
 	targetPendingEntriesCount int64
+	targetStreamLength        int64
 	streamName                string
 	consumerGroupName         string
 	databaseIndex             int
@@ -89,21 +99,15 @@ func createClusteredRedisStreamsScaler(ctx context.Context, meta *redisStreamsMe
 		return nil
 	}
 
-	pendingEntriesCountFn := func(ctx context.Context) (int64, error) {
-		pendingEntries, err := client.XPending(ctx, meta.streamName, meta.consumerGroupName).Result()
-		if err != nil {
-			return -1, err
-		}
-		return pendingEntries.Count, nil
-	}
+	entriesCountFn, err := createEntriesCountFn(client, meta)
 
 	return &redisStreamsScaler{
-		metricType:               metricType,
-		metadata:                 meta,
-		closeFn:                  closeFn,
-		getPendingEntriesCountFn: pendingEntriesCountFn,
-		logger:                   logger,
-	}, nil
+		metricType:        metricType,
+		metadata:          meta,
+		closeFn:           closeFn,
+		getEntriesCountFn: entriesCountFn,
+		logger:            logger,
+	}, err
 }
 
 func createSentinelRedisStreamsScaler(ctx context.Context, meta *redisStreamsMetadata, metricType v2.MetricTargetType, logger logr.Logger) (Scaler, error) {
@@ -133,27 +137,42 @@ func createScaler(client *redis.Client, meta *redisStreamsMetadata, metricType v
 		return nil
 	}
 
-	pendingEntriesCountFn := func(ctx context.Context) (int64, error) {
-		pendingEntries, err := client.XPending(ctx, meta.streamName, meta.consumerGroupName).Result()
-		if err != nil {
-			return -1, err
-		}
-		return pendingEntries.Count, nil
-	}
+	entriesCountFn, err := createEntriesCountFn(client, meta)
 
 	return &redisStreamsScaler{
-		metricType:               metricType,
-		metadata:                 meta,
-		closeFn:                  closeFn,
-		getPendingEntriesCountFn: pendingEntriesCountFn,
-		logger:                   logger,
-	}, nil
+		metricType:        metricType,
+		metadata:          meta,
+		closeFn:           closeFn,
+		getEntriesCountFn: entriesCountFn,
+		logger:            logger,
+	}, err
 }
 
-var (
-	// ErrRedisMissingPendingEntriesCount is returned when "pendingEntriesCount" is missing.
-	ErrRedisMissingPendingEntriesCount = errors.New("missing pending entries count")
+func createEntriesCountFn(client redis.Cmdable, meta *redisStreamsMetadata) (entriesCountFn func(ctx context.Context) (int64, error), err error) {
+	switch meta.scaleFactor {
+	case xPendingFactor:
+		entriesCountFn = func(ctx context.Context) (int64, error) {
+			pendingEntries, err := client.XPending(ctx, meta.streamName, meta.consumerGroupName).Result()
+			if err != nil {
+				return -1, err
+			}
+			return pendingEntries.Count, nil
+		}
+	case xLengthFactor:
+		entriesCountFn = func(ctx context.Context) (int64, error) {
+			entriesLength, err := client.XLen(ctx, meta.streamName).Result()
+			if err != nil {
+				return -1, err
+			}
+			return entriesLength, nil
+		}
+	default:
+		err = fmt.Errorf("unrecognized scale factor %v", meta.scaleFactor)
+	}
+	return
+}
 
+var (
 	// ErrRedisMissingStreamName is returned when "stream" is missing.
 	ErrRedisMissingStreamName = errors.New("missing redis stream name")
 )
@@ -185,18 +204,6 @@ func parseRedisStreamsMetadata(config *ScalerConfig, parseFn redisAddressParser)
 		meta.connectionInfo.unsafeSsl = parsedVal
 	}
 
-	meta.targetPendingEntriesCount = defaultTargetPendingEntriesCount
-
-	if val, ok := config.TriggerMetadata[pendingEntriesCountMetadata]; ok {
-		pendingEntriesCount, err := strconv.ParseInt(val, 10, 64)
-		if err != nil {
-			return nil, fmt.Errorf("error parsing pending entries count: %w", err)
-		}
-		meta.targetPendingEntriesCount = pendingEntriesCount
-	} else {
-		return nil, ErrRedisMissingPendingEntriesCount
-	}
-
 	if val, ok := config.TriggerMetadata[streamNameMetadata]; ok {
 		meta.streamName = val
 	} else {
@@ -205,8 +212,25 @@ func parseRedisStreamsMetadata(config *ScalerConfig, parseFn redisAddressParser)
 
 	if val, ok := config.TriggerMetadata[consumerGroupNameMetadata]; ok {
 		meta.consumerGroupName = val
+		meta.scaleFactor = xPendingFactor
+		meta.targetPendingEntriesCount = defaultTargetEntries
+		if val, ok := config.TriggerMetadata[pendingEntriesCountMetadata]; ok {
+			pendingEntriesCount, err := strconv.ParseInt(val, 10, 64)
+			if err != nil {
+				return nil, fmt.Errorf("error parsing pending entries count: %w", err)
+			}
+			meta.targetPendingEntriesCount = pendingEntriesCount
+		}
 	} else {
-		return nil, fmt.Errorf("missing redis stream consumer group name")
+		meta.scaleFactor = xLengthFactor
+		meta.targetStreamLength = defaultTargetEntries
+		if val, ok := config.TriggerMetadata[streamLengthMetadata]; ok {
+			streamLength, err := strconv.ParseInt(val, 10, 64)
+			if err != nil {
+				return nil, fmt.Errorf("error parsing stream length: %w", err)
+			}
+			meta.targetStreamLength = streamLength
+		}
 	}
 
 	meta.databaseIndex = defaultDBIndex
@@ -228,19 +252,28 @@ func (s *redisStreamsScaler) Close(context.Context) error {
 
 // GetMetricSpecForScaling returns the metric spec for the HPA
 func (s *redisStreamsScaler) GetMetricSpecForScaling(context.Context) []v2.MetricSpec {
+	var metricValue int64
+
+	switch s.metadata.scaleFactor {
+	case xPendingFactor:
+		metricValue = s.metadata.targetPendingEntriesCount
+	case xLengthFactor:
+		metricValue = s.metadata.targetStreamLength
+	}
+
 	externalMetric := &v2.ExternalMetricSource{
 		Metric: v2.MetricIdentifier{
 			Name: GenerateMetricNameWithIndex(s.metadata.scalerIndex, kedautil.NormalizeString(fmt.Sprintf("redis-streams-%s", s.metadata.streamName))),
 		},
-		Target: GetMetricTarget(s.metricType, s.metadata.targetPendingEntriesCount),
+		Target: GetMetricTarget(s.metricType, metricValue),
 	}
 	metricSpec := v2.MetricSpec{External: externalMetric, Type: externalMetricType}
 	return []v2.MetricSpec{metricSpec}
 }
 
 // GetMetricsAndActivity fetches the number of pending entries for a consumer group in a stream
 func (s *redisStreamsScaler) GetMetricsAndActivity(ctx context.Context, metricName string) ([]external_metrics.ExternalMetricValue, bool, error) {
-	pendingEntriesCount, err := s.getPendingEntriesCountFn(ctx)
+	pendingEntriesCount, err := s.getEntriesCountFn(ctx)
 
 	if err != nil {
 		s.logger.Error(err, "error fetching pending entries count")