Implemented a fallback parameter validation for DbClusterParameterGroup (#230)

fixes aws-controllers-k8s/community#1847

`DbClusterParameterGroup` was incorrectly rejecting valid parameters like `slow_query_log`, `long_query_time`, and `log_queries_not_using_indexes` with "unknown parameter" errors. This occurred because the controller only validated parameters against `DescribeEngineDefaultClusterParameters` API, but some valid cluster parameters (particularly MySQL logging parameters) are only listed in the `DescribeEngineDefaultParameters` (instance-level) API response.

Description of changes:

Implemented a fallback parameter validation mechanism in `getParameterMeta()` that first checks cluster-level parameter defaults, and if not found, falls back to instance-level parameter defaults before rejecting a parameter as unknown. This allows the controller to properly validate and apply parameters that AWS RDS Console and CLI support but weren't previously accessible through the ACK controller, while maintaining backward compatibility and proper parameter metadata handling for apply methods and modifiability checks.


```yaml
apiVersion: rds.services.k8s.aws/v1alpha1
kind: DBClusterParameterGroup
metadata:
  name: test-slow-query-log
spec:
  name: test-slow-query-log
  description: Test parameter group to reproduce slow_query_log issue
  family: "aurora-mysql8.0"
  parameterOverrides:
    slow_query_log: "1"
    long_query_time: "10"
    log_queries_not_using_indexes: "1"
```
```
{"level":"debug","ts":"2025-07-21T14:41:49.089-0700","logger":"ackrt","msg":"patched resource status","kind":"DBClusterParameterGroup","namespace":"default","name":"test-slow-query-log","account":"xxx","role":"","region":"us-west-2","is_adopted":false,"generation":1,"json":"{\"metadata\":{\"resourceVersion\":\"103993920\"},\"spec\":{\"tags\":null},\"status\":{\"ackResourceMetadata\":{\"arn\":\"arn:aws:rds:us-west-2:xxxx:cluster-pg:test-slow-query-log\",\"ownerAccountID\":\"xxxx\",\"region\":\"us-west-2\"},\"conditions\":[{\"lastTransitionTime\":\"2025-07-21T21:41:49Z\",\"message\":\"Resource synced successfully\",\"reason\":\"\",\"status\":\"True\",\"type\":\"ACK.ResourceSynced\"}],\"parameterOverrideStatuses\":[{\"applyMethod\":\"immediate\",\"applyType\":\"dynamic\",\"parameterName\":\"log_queries_not_using_indexes\",\"parameterValue\":\"1\"},{\"applyMethod\":\"immediate\",\"applyType\":\"dynamic\",\"parameterName\":\"long_query_time\",\"parameterValue\":\"10\"},{\"applyMethod\":\"immediate\",\"applyType\":\"dynamic\",\"parameterName\":\"slow_query_log\",\"parameterValue\":\"1\"}]}}"}
```

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.

Author: rushmash91

pkg/resource/db_cluster_parameter_group/hooks.go

@@ -37,11 +37,17 @@ const (
 )
 
 var (
-	// cache of parameter defaults
+	// cache of parameter defaults for cluster parameters
 	cachedParamMeta = util.ParamMetaCache{
 		Cache: map[string]map[string]util.ParamMeta{},
 	}
 
+	// cache of parameter defaults for instance parameters
+	// Used as fallback when cluster parameter validation fails
+	instanceParamMeta = util.ParamMetaCache{
+		Cache: map[string]map[string]util.ParamMeta{},
+	}
+
 	errParameterGroupJustCreated = fmt.Errorf("parameter group just got created")
 	requeueWaitWhileCreating     = ackrequeue.NeededAfter(
 		errParameterGroupJustCreated,
@@ -208,6 +214,11 @@ func sdkTagsFromResourceTags(
 // RDS does not have a DeleteParameter or DeleteParameterFromParameterGroup API
 // call. Instead, you need to call ResetDBClusterParameterGroup with a list of
 // DB Cluster Parameters that you want RDS to reset to a default value.
+//
+// Note(rushmash91): This function uses fallback parameter validation to work around an AWS API
+// limitation where DescribeEngineDefaultClusterParameters may not return all valid
+// cluster parameters (e.g., MySQL logging parameters like slow_query_log). See
+// getParameterMeta() for details on the fallback mechanism.
 func (rm *resourceManager) syncParameters(
 	ctx context.Context,
 	desired *resource,
@@ -307,9 +318,7 @@ func (rm *resourceManager) resetParameters(
 		// default to this if something goes wrong looking up parameter
 		// defaults
 		applyMethod := svcsdktypes.ApplyMethodImmediate
-		pMeta, err = cachedParamMeta.Get(
-			ctx, *family, paramName, rm.getFamilyParameters,
-		)
+		pMeta, err = rm.getParameterMeta(ctx, *family, paramName)
 		if err != nil {
 			return err
 		}
@@ -360,9 +369,7 @@ func (rm *resourceManager) modifyParameters(
 	for paramName, paramValue := range toModify {
 		// default to "immediate" if something goes wrong looking up defaults
 		applyMethod := svcsdktypes.ApplyMethodImmediate
-		pMeta, err = cachedParamMeta.Get(
-			ctx, *family, paramName, rm.getFamilyParameters,
-		)
+		pMeta, err = rm.getParameterMeta(ctx, *family, paramName)
 		if err != nil {
 			return err
 		}
@@ -433,3 +440,68 @@ func (rm *resourceManager) getFamilyParameters(
 	}
 	return familyMeta, nil
 }
+
+// getInstanceFamilyParameters calls the RDS DescribeEngineDefaultParameters API to
+// retrieve the set of parameter information for a DB instance parameter group family.
+// This is used as a fallback when cluster parameter validation fails, as some valid
+// cluster parameters may only be listed in the instance parameter defaults.
+func (rm *resourceManager) getInstanceFamilyParameters(
+	ctx context.Context,
+	family string,
+) (map[string]util.ParamMeta, error) {
+	var marker *string
+	familyMeta := map[string]util.ParamMeta{}
+
+	for {
+		resp, err := rm.sdkapi.DescribeEngineDefaultParameters(
+			ctx,
+			&svcsdk.DescribeEngineDefaultParametersInput{
+				DBParameterGroupFamily: aws.String(family),
+				Marker:                 marker,
+			},
+		)
+		rm.metrics.RecordAPICall("GET", "DescribeEngineDefaultParameters", err)
+		if err != nil {
+			return nil, err
+		}
+		for _, param := range resp.EngineDefaults.Parameters {
+			pName := *param.ParameterName
+			familyMeta[pName] = util.ParamMeta{
+				IsModifiable: *param.IsModifiable,
+				IsDynamic:    *param.ApplyType != applyTypeStatic,
+			}
+		}
+		marker = resp.EngineDefaults.Marker
+		if marker == nil {
+			break
+		}
+	}
+	return familyMeta, nil
+}
+
+// getParameterMeta retrieves parameter metadata with fallback validation.
+// First tries cluster-level parameter validation, then falls back to instance-level
+// validation if the parameter is not found. This works around the AWS API limitation
+// where DescribeEngineDefaultClusterParameters may not return all valid cluster
+// parameters (e.g., MySQL logging parameters like slow_query_log).
+func (rm *resourceManager) getParameterMeta(
+	ctx context.Context,
+	family string,
+	paramName string,
+) (*util.ParamMeta, error) {
+	// Try cluster-level parameters first
+	pMeta, err := cachedParamMeta.Get(ctx, family, paramName, rm.getFamilyParameters)
+	if err == nil {
+		return pMeta, nil
+	}
+
+	// If not found in cluster parameters, try instance-level parameters
+	// Some valid cluster parameters may only be listed in instance defaults
+	instanceMeta, instanceErr := instanceParamMeta.Get(ctx, family, paramName, rm.getInstanceFamilyParameters)
+	if instanceErr == nil {
+		return instanceMeta, nil
+	}
+
+	// Return original error if not found in either
+	return nil, err
+}

test/e2e/db_cluster_parameter_group.py

@@ -74,18 +74,50 @@ def get(db_cluster_parameter_group_name):
         return None
 
 def get_parameters(db_cluster_parameter_group_name):
-    """Returns a dict containing the paramters of a given parameter group
+    """Returns a dict containing the parameters of a given parameter group
 
-    If no such DB cluster parameter group exists, returns None.
+    If no such DB cluster parameter group exists, returns empty list.
     """
     c = boto3.client('rds')
     try:
         resp = c.describe_db_cluster_parameters(
             DBClusterParameterGroupName=db_cluster_parameter_group_name,
         )
         return resp['Parameters']
-    except c.exceptions.DBClusterParameterGroupNotFoundFault:
-        return None
+    except c.exceptions.DBParameterGroupNotFoundFault:
+        return []
+
+def get_user_defined_parameters(db_cluster_parameter_group_name):
+    """Returns a dict containing the user-defined parameters of a given cluster parameter group
+
+    If no such DB cluster parameter group exists, returns empty list.
+    Uses Source="user" to get only user-defined parameters (like the controller does).
+    """
+    c = boto3.client('rds')
+    try:
+        all_parameters = []
+        marker = None
+        
+        while True:
+            params = {
+                'DBClusterParameterGroupName': db_cluster_parameter_group_name,
+                'Source': 'user'
+            }
+            if marker:
+                params['Marker'] = marker
+                
+            resp = c.describe_db_cluster_parameters(**params)
+            all_parameters.extend(resp['Parameters'])
+            
+            # Check if there are more results
+            if 'Marker' in resp:
+                marker = resp['Marker']
+            else:
+                break
+                
+        return all_parameters
+    except c.exceptions.DBParameterGroupNotFoundFault:
+        return []
 
 def get_tags(db_cluster_parameter_group_arn):
     """Returns a dict containing the DB cluster parameter group's tag records

test/e2e/db_parameter_group.py

@@ -88,6 +88,42 @@ def get_parameters(db_parameter_group_name):
         return None
 
 
+def get_engine_default_parameters(db_parameter_group_family):
+    """Returns a dict containing the engine default parameters for a given parameter group family
+    
+    This function calls DescribeEngineDefaultParameters to get the default parameter metadata
+    that's used as fallback validation in cluster parameter groups.
+    """
+    c = boto3.client('rds')
+    try:
+        all_parameters = []
+        marker = None
+        
+        while True:
+            if marker:
+                resp = c.describe_engine_default_parameters(
+                    DBParameterGroupFamily=db_parameter_group_family,
+                    Marker=marker
+                )
+            else:
+                resp = c.describe_engine_default_parameters(
+                    DBParameterGroupFamily=db_parameter_group_family,
+                )
+            
+            parameters = resp['EngineDefaults']['Parameters']
+            all_parameters.extend(parameters)
+            
+            # Check if there are more results
+            if 'Marker' in resp['EngineDefaults']:
+                marker = resp['EngineDefaults']['Marker']
+            else:
+                break
+                
+        return all_parameters
+    except Exception as e:
+        return None
+
+
 def get_tags(db_parameter_group_arn):
     """Returns a dict containing the DB parameter group's tag records from the
     RDS API.

test/e2e/resources/db_cluster_parameter_group_aurora_mysql8.0_logging.yaml

@@ -0,0 +1,12 @@
+apiVersion: rds.services.k8s.aws/v1alpha1
+kind: DBClusterParameterGroup
+metadata:
+  name: $DB_CLUSTER_PARAMETER_GROUP_NAME
+spec:
+  name: $DB_CLUSTER_PARAMETER_GROUP_NAME
+  description: $DB_CLUSTER_PARAMETER_GROUP_DESC
+  family: $DB_CLUSTER_PARAMETER_GROUP_FAMILY
+  parameterOverrides:
+    slow_query_log: "$PARAM_SLOW_QUERY_LOG_VALUE"
+    long_query_time: "$PARAM_LONG_QUERY_TIME_VALUE"
+    log_queries_not_using_indexes: "$PARAM_LOG_QUERIES_NOT_USING_INDEXES_VALUE"

test/e2e/tests/test_db_cluster_parameter_group.py

@@ -24,6 +24,7 @@
 from e2e import service_marker, CRD_GROUP, CRD_VERSION, load_rds_resource
 from e2e.replacement_values import REPLACEMENT_VALUES
 from e2e import db_cluster_parameter_group
+from e2e import db_parameter_group
 from e2e import tag
 from e2e import condition
 
@@ -77,6 +78,49 @@ def aurora_mysql57_cluster_param_group():
     db_cluster_parameter_group.wait_until_deleted(resource_name)
 
 
+@pytest.fixture
+def aurora_mysql80_logging_cluster_param_group():
+    resource_name = random_suffix_name("aurora-mysql8-logging", 32)
+
+    replacements = REPLACEMENT_VALUES.copy()
+    replacements["DB_CLUSTER_PARAMETER_GROUP_NAME"] = resource_name
+    replacements["DB_CLUSTER_PARAMETER_GROUP_DESC"] = "Test MySQL logging parameters for Aurora MySQL 8.0"
+    replacements["DB_CLUSTER_PARAMETER_GROUP_FAMILY"] = "aurora-mysql8.0"
+    replacements["PARAM_SLOW_QUERY_LOG_VALUE"] = "1"
+    replacements["PARAM_LONG_QUERY_TIME_VALUE"] = "10"
+    replacements["PARAM_LOG_QUERIES_NOT_USING_INDEXES_VALUE"] = "1"
+
+    resource_data = load_rds_resource(
+        "db_cluster_parameter_group_aurora_mysql8.0_logging",
+        additional_replacements=replacements,
+    )
+    logging.debug(resource_data)
+
+    # Create the k8s resource
+    ref = k8s.CustomResourceReference(
+        CRD_GROUP, CRD_VERSION, RESOURCE_PLURAL,
+        resource_name, namespace="default",
+    )
+    k8s.create_custom_resource(ref, resource_data)
+    cr = k8s.wait_resource_consumed_by_controller(ref)
+    time.sleep(CREATE_WAIT_AFTER_SECONDS)
+
+    assert cr is not None
+    assert k8s.get_resource_exists(ref)
+
+    yield ref, cr, resource_name
+
+    # Try to delete, if doesn't already exist
+    try:
+        _, deleted = k8s.delete_custom_resource(ref, 3, 10)
+        assert deleted
+        time.sleep(DELETE_WAIT_AFTER_SECONDS)
+    except:
+        pass
+
+    db_cluster_parameter_group.wait_until_deleted(resource_name)
+
+
 @service_marker
 @pytest.mark.canary
 class TestDBClusterParameterGroup:
@@ -161,3 +205,70 @@ def test_crud_aurora_mysql5_7(self, aurora_mysql57_cluster_param_group):
                 assert "ParameterValue" in tp, f"No ParameterValue in parameter of name 'aurora_read_replica_read_committed': {tp}"
                 assert tp["ParameterValue"] == "ON", f"Wrong value for parameter of name 'aurora_read_replica_read_committed': {tp}"
         assert found == 2, f"Did not find parameters with names 'aurora_binlog_read_buffer_size' and 'aurora_read_replica_read_committed': {test_params}"
+
+    def test_mysql_logging_parameters(self, aurora_mysql80_logging_cluster_param_group):
+        ref, cr, resource_name = aurora_mysql80_logging_cluster_param_group
+
+        latest = db_cluster_parameter_group.get(resource_name)
+        assert latest is not None
+
+        instance_defaults = db_parameter_group.get_engine_default_parameters("mysql8.0")
+        assert instance_defaults is not None, "Failed to get instance-level engine defaults"
+        
+        fallback_params = list(filter(lambda x: x["ParameterName"] in [
+            "slow_query_log",
+            "long_query_time", 
+            "log_queries_not_using_indexes",
+        ], instance_defaults))
+        
+        # Log debug info about found parameters
+        found_param_names = [p['ParameterName'] for p in fallback_params]
+        logging.debug(f"Found fallback parameters: {found_param_names}")
+        
+        assert len(fallback_params) == 3, f"Expected 3 MySQL logging parameters in instance defaults, found {len(fallback_params)}: {found_param_names}"
+        
+        for param in fallback_params:
+            assert "ParameterName" in param, f"Missing ParameterName in fallback parameter: {param}"
+            assert "IsModifiable" in param, f"Missing IsModifiable in fallback parameter: {param}"
+            assert "ApplyType" in param, f"Missing ApplyType in fallback parameter: {param}"
+
+        assert 'status' in cr
+        assert 'parameterOverrideStatuses' in cr['status']
+        
+        # Verify the parameter statuses show our MySQL logging parameters
+        status_params = cr['status']['parameterOverrideStatuses']
+        param_names = [p['parameterName'] for p in status_params]
+        
+        assert "slow_query_log" in param_names, f"slow_query_log parameter missing from status: {param_names}"
+        assert "long_query_time" in param_names, f"long_query_time parameter missing from status: {param_names}"
+        assert "log_queries_not_using_indexes" in param_names, f"log_queries_not_using_indexes parameter missing from status: {param_names}"
+
+        # Additional wait for AWS RDS parameter propagation
+        # RDS parameter changes can take 5-10 minutes to be fully visible via API
+        logging.debug("Waiting additional time for AWS RDS parameter propagation...")
+        time.sleep(MODIFY_WAIT_AFTER_SECONDS) 
+        
+        latest_params = db_cluster_parameter_group.get_user_defined_parameters(resource_name)
+        assert latest_params is not None, "Failed to get user-defined cluster parameters"
+        
+        test_params = list(filter(lambda x: x["ParameterName"] in [
+            "slow_query_log",
+            "long_query_time", 
+            "log_queries_not_using_indexes",
+        ], latest_params))
+        
+        # Check initial parameter values
+        expected_initial_values = {
+            "slow_query_log": "1",
+            "long_query_time": "10",
+            "log_queries_not_using_indexes": "1"
+        }
+        
+        for tp in test_params:
+            param_name = tp["ParameterName"]
+            assert param_name in expected_initial_values, f"Unexpected parameter: {param_name}"
+            assert tp["ParameterValue"] == expected_initial_values[param_name], \
+                f"Wrong value for {param_name}: expected {expected_initial_values[param_name]}, got {tp['ParameterValue']}"
+        
+        assert len(test_params) == len(expected_initial_values), \
+            f"Expected {len(expected_initial_values)} parameters, found {len(test_params)}: {test_params}"