fix(aws-dynamodb): paginate Query to avoid undercounting large result sets

CHANGELOG.md

@@ -88,6 +88,7 @@ To learn more about active deprecations, we recommend checking [GitHub Discussio
 - **General**: Fix `ScaledJob` CRD validation to include "default" as a valid value for `scalingStrategy.strategy` ([#7855](https://github.com/kedacore/keda/issues/7855))
 - **General**: Restore gRPC reconnect backoff in the metrics service client; an unset `Backoff` in `WithConnectParams` disabled backoff and caused a zero-delay reconnect loop that flooded logs when keda-operator was unreachable ([#7856](https://github.com/kedacore/keda/issues/7856))
 - **General**: Treat negative external metric values as zero to prevent incorrect HPA scaling ([#7880](https://github.com/kedacore/keda/issues/7880))
+- **AWS DynamoDB Scaler**: Paginate the `Query` request so items beyond the first 1 MB page are counted, instead of undercounting large result sets and under-scaling ([#7912](https://github.com/kedacore/keda/pull/7912))
 - **Azure Blob Storage Scaler**: Fix `globPattern` never matching when written in path-style with a leading `/`, since blob names never have one ([#6492](https://github.com/kedacore/keda/issues/6492))
 - **MongoDB Scaler**: Disconnect the client when the initial `Ping` fails so the background topology-monitoring connections opened by `mongo.Connect` are not leaked ([#5612](https://github.com/kedacore/keda/issues/5612))
 

pkg/scalers/aws_dynamodb_scaler.go

@@ -199,14 +199,22 @@ func (s *awsDynamoDBScaler) GetQueryMetrics(ctx context.Context) (float64, error
 		dimensions.IndexName = aws.String(s.metadata.IndexName)
 	}
 
-	res, err := s.dbClient.Query(ctx, &dimensions)
-
-	if err != nil {
-		s.logger.Error(err, "Failed to get output")
-		return 0, err
+	// A DynamoDB Query returns at most 1 MB of data per call, so the result set
+	// can span multiple pages. Follow LastEvaluatedKey and sum the per-page
+	// counts, otherwise the scaler only sees the first page and undercounts the
+	// matching items, leading to under-scaling.
+	var itemCount int32
+	paginator := dynamodb.NewQueryPaginator(s.dbClient, &dimensions)
+	for paginator.HasMorePages() {
+		res, err := paginator.NextPage(ctx)
+		if err != nil {
+			s.logger.Error(err, "Failed to get output")
+			return 0, err
+		}
+		itemCount += res.Count
 	}
 
-	return float64(res.Count), nil
+	return float64(itemCount), nil
 }
 
 // json2Map convert Json to map[string]string

pkg/scalers/aws_dynamodb_scaler_test.go

@@ -21,6 +21,7 @@ const (
 	testAWSDynamoErrorTable      = "Error"
 	testAWSDynamoNoValueTable    = "NoValue"
 	testAWSDynamoIndexTable      = "Index"
+	testAWSDynamoMultiPageTable  = "MultiPage"
 )
 
 var testAWSDynamoAuthentication = map[string]string{
@@ -405,6 +406,8 @@ type mockDynamoDB struct {
 var result int32 = 4
 var indexResult int32 = 2
 var empty int32
+var multiPageFirst int32 = 4
+var multiPageSecond int32 = 3
 
 func (c *mockDynamoDB) Query(_ context.Context, input *dynamodb.QueryInput, _ ...func(*dynamodb.Options)) (*dynamodb.QueryOutput, error) {
 	switch *input.TableName {
@@ -414,6 +417,19 @@ func (c *mockDynamoDB) Query(_ context.Context, input *dynamodb.QueryInput, _ ..
 		return &dynamodb.QueryOutput{
 			Count: empty,
 		}, nil
+	case testAWSDynamoMultiPageTable:
+		// Simulate a paginated Query response: the first page carries a
+		// LastEvaluatedKey so the scaler must request the next page and sum
+		// the per-page counts.
+		if input.ExclusiveStartKey == nil {
+			return &dynamodb.QueryOutput{
+				Count:            multiPageFirst,
+				LastEvaluatedKey: map[string]types.AttributeValue{"pk": &types.AttributeValueMemberS{Value: "next"}},
+			}, nil
+		}
+		return &dynamodb.QueryOutput{
+			Count: multiPageSecond,
+		}, nil
 	}
 
 	if input.IndexName != nil {
@@ -504,6 +520,24 @@ func TestDynamoGetQueryMetrics(t *testing.T) {
 	}
 }
 
+func TestDynamoGetQueryMetricsPaginated(t *testing.T) {
+	meta := awsDynamoDBMetadata{
+		TableName:                 testAWSDynamoMultiPageTable,
+		AwsRegion:                 "eu-west-1",
+		KeyConditionExpression:    "#yr = :yyyy",
+		expressionAttributeNames:  map[string]string{"#yr": year},
+		expressionAttributeValues: map[string]types.AttributeValue{":yyyy": yearAttr},
+		TargetValue:               3,
+	}
+	scaler := awsDynamoDBScaler{"", &meta, &mockDynamoDB{}, nil, logr.Discard()}
+
+	value, err := scaler.GetQueryMetrics(context.Background())
+	assert.NoError(t, err)
+	// The mock returns two pages (4 and 3 items). A DynamoDB Query response is
+	// paginated, so the scaler must follow LastEvaluatedKey and sum every page.
+	assert.EqualValues(t, int64(multiPageFirst+multiPageSecond), value)
+}
+
 func TestDynamoIsActive(t *testing.T) {
 	for _, meta := range awsDynamoDBGetMetricTestData {
 		t.Run(meta.TableName, func(t *testing.T) {