Validate parameter group changes (#132)

* feat: validate parameter group update with apply_method only changes
* feat: validate parameter group apply_method only changes on creation
* feat: validate immediate can't be used on static parameters
* feat: use paginator when response can be paginated
* chore: bump version to 0.6.7


Signed-off-by: Di Wang <[email protected]>

Author: hemslo

Dockerfile

@@ -1,6 +1,6 @@
 FROM quay.io/redhat-services-prod/app-sre-tenant/er-base-terraform-main/er-base-terraform-main:tf-1.6.6-py-3.12-v0.3.4-1@sha256:1579e58702182a02b55a0841254d188a6b99ff42c774279890567338c863a31b AS base
 # keep in sync with pyproject.toml
-LABEL konflux.additional-tags="0.6.6"
+LABEL konflux.additional-tags="0.6.7"
 
 FROM base AS builder
 COPY --from=ghcr.io/astral-sh/uv:0.6.12@sha256:515b886e8eb99bcf9278776d8ea41eb4553a794195ef5803aa7ca6258653100d /uv /bin/uv

er_aws_rds/errors.py

@@ -11,8 +11,6 @@
     model_validator,
 )
 
-from er_aws_rds.errors import RDSLogicalReplicationError
-
 ENHANCED_MONITORING_ROLE_NAME_MAX_LENGTH = 64
 
 
@@ -264,20 +262,6 @@ def replication(self) -> Self:
         self.backup_retention_period = 0
         return self
 
-    @model_validator(mode="after")
-    def validate_parameter_group_parameters(self) -> Self:
-        """Validate that every parameter complies with our requirements"""
-        if not self.parameter_group:
-            return self
-        for parameter in self.parameter_group.parameters or []:
-            if (
-                parameter.name == "rds.logical_replication"
-                and parameter.apply_method != "pending-reboot"
-            ):
-                msg = "rds.logical_replication must be set to pending-reboot"
-                raise RDSLogicalReplicationError(msg)
-        return self
-
     @model_validator(mode="after")
     def parameter_groups(self) -> Self:
         """

er_aws_rds/input.py

@@ -10,8 +10,12 @@
     ResourceChange,
     TerraformJsonPlanParser,
 )
+from mypy_boto3_rds.type_defs import ParameterOutputTypeDef
 
-from er_aws_rds.input import AppInterfaceInput
+from er_aws_rds.input import (
+    AppInterfaceInput,
+    Parameter,
+)
 from hooks.utils.aws_api import AWSApi
 from hooks.utils.envvars import RuntimeEnvVars
 from hooks.utils.logger import setup_logging
@@ -158,13 +162,122 @@ def _validate_no_changes_when_blue_green_deployment_enabled(self) -> None:
                 f"No changes allowed when Blue/Green Deployment enabled, detected changes: {resource_changes}"
             )
 
+    @staticmethod
+    def _is_apply_method_change_only(
+        before_parameter: Parameter,
+        after_parameter: Parameter,
+    ) -> bool:
+        return (
+            before_parameter.name == after_parameter.name
+            and before_parameter.value == after_parameter.value
+            and before_parameter.apply_method != after_parameter.apply_method
+        )
+
+    def _validate_parameter_group_change(self, change: Change) -> None:
+        if not change.after:
+            return
+
+        after_parameter_by_name = {
+            parameter["name"]: Parameter.model_validate(parameter)
+            for parameter in change.after.get("parameter") or []
+        }
+
+        if not after_parameter_by_name:
+            return
+
+        default_parameter_by_name = self.aws_api.get_engine_default_parameters(
+            change.after["family"],
+            list(after_parameter_by_name.keys()),
+        )
+
+        before_parameter_by_name = (
+            {
+                parameter["name"]: Parameter.model_validate(parameter)
+                for parameter in change.before.get("parameter") or []
+            }
+            if change.before
+            else {
+                parameter["ParameterName"]: Parameter(
+                    name=parameter["ParameterName"],
+                    value=parameter.get("ParameterValue", ""),
+                    apply_method=parameter.get("ApplyMethod", "pending-reboot"),
+                )
+                for parameter in default_parameter_by_name.values()
+            }
+        )
+
+        self._validate_apply_method_change_only(
+            before_parameter_by_name, after_parameter_by_name
+        )
+        self._validate_apply_method_with_apply_type(
+            default_parameter_by_name, after_parameter_by_name
+        )
+
+    def _validate_apply_method_change_only(
+        self,
+        before_parameter_by_name: dict[str, Parameter],
+        after_parameter_by_name: dict[str, Parameter],
+    ) -> None:
+        common_names = before_parameter_by_name.keys() & after_parameter_by_name.keys()
+        apply_method_change_only_parameter_names = [
+            name
+            for name in common_names
+            if self._is_apply_method_change_only(
+                before_parameter_by_name[name],
+                after_parameter_by_name[name],
+            )
+        ]
+        if apply_method_change_only_parameter_names:
+            parameters = ", ".join(apply_method_change_only_parameter_names)
+            self.errors.append(
+                f"Problematic plan changes for parameter group detected for parameters: {parameters}. "
+                "Parameter with only apply_method toggled while value is same as before or default is not allowed, "
+                "remove the parameter OR change value OR align apply_method with AWS default pending-reboot, "
+                "checkout details at https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/db_parameter_group#problematic-plan-changes"
+            )
+
+    def _validate_apply_method_with_apply_type(
+        self,
+        default_parameter_by_name: dict[str, ParameterOutputTypeDef],
+        after_parameter_by_name: dict[str, Parameter],
+    ) -> None:
+        common_names = default_parameter_by_name.keys() & after_parameter_by_name.keys()
+        immediate_on_static_parameter_names = [
+            name
+            for name in common_names
+            if (
+                after_parameter_by_name[name].apply_method == "immediate"
+                and default_parameter_by_name[name].get("ApplyType") == "static"
+            )
+        ]
+        if immediate_on_static_parameter_names:
+            parameters = ", ".join(immediate_on_static_parameter_names)
+            self.errors.append(
+                "cannot use immediate apply method for static parameter, "
+                f"must be set to pending-reboot: {parameters}"
+            )
+
+    def _validate_parameter_group_changes(self) -> None:
+        changed_actions = {
+            Action.ActionCreate,
+            Action.ActionUpdate,
+        }
+        for c in self.plan.resource_changes:
+            if (
+                c.change
+                and changed_actions.intersection(c.change.actions)
+                and c.type == "aws_db_parameter_group"
+            ):
+                self._validate_parameter_group_change(c.change)
+
     def validate(self) -> list[str]:
         """Validate method, return validation errors"""
         self.errors.clear()
         self._validate_version_on_create()
         self._validate_version_upgrade()
         self._validate_deletion_protection_not_enabled_on_destroy()
         self._validate_no_changes_when_blue_green_deployment_enabled()
+        self._validate_parameter_group_changes()
         return self.errors
 
 

hooks/post_plan.py

@@ -13,6 +13,8 @@
     BlueGreenDeploymentTypeDef,
     DBInstanceTypeDef,
     DBParameterGroupTypeDef,
+    DescribeDBParametersMessagePaginateTypeDef,
+    DescribeEngineDefaultParametersMessagePaginateTypeDef,
     ParameterOutputTypeDef,
     UpgradeTargetTypeDef,
 )
@@ -98,19 +100,50 @@ def get_db_parameter_group(self, name: str) -> DBParameterGroupTypeDef | None:
     def get_db_parameters(
         self,
         parameter_group_name: str,
-        parameter_names: list[str],
+        parameter_names: list[str] | None = None,
     ) -> dict[str, ParameterOutputTypeDef]:
         """Get DB parameters"""
-        data = self.rds_client.describe_db_parameters(
-            DBParameterGroupName=parameter_group_name,
-            Filters=[
+        kwargs: DescribeDBParametersMessagePaginateTypeDef = {
+            "DBParameterGroupName": parameter_group_name,
+        }
+        if parameter_names:
+            kwargs["Filters"] = [
                 {
                     "Name": "parameter-name",
                     "Values": parameter_names,
                 }
-            ],
-        )
-        return {item["ParameterName"]: item for item in data["Parameters"] or []}
+            ]
+        paginator = self.rds_client.get_paginator("describe_db_parameters")
+        page_iterator = paginator.paginate(**kwargs)
+        return {
+            item["ParameterName"]: item
+            for data in page_iterator
+            for item in data["Parameters"] or []
+        }
+
+    def get_engine_default_parameters(
+        self,
+        parameter_group_family: str,
+        parameter_names: list[str] | None = None,
+    ) -> dict[str, ParameterOutputTypeDef]:
+        """Get engine default parameters"""
+        kwargs: DescribeEngineDefaultParametersMessagePaginateTypeDef = {
+            "DBParameterGroupFamily": parameter_group_family,
+        }
+        if parameter_names:
+            kwargs["Filters"] = [
+                {
+                    "Name": "parameter-name",
+                    "Values": parameter_names,
+                }
+            ]
+        paginator = self.rds_client.get_paginator("describe_engine_default_parameters")
+        page_iterator = paginator.paginate(**kwargs)
+        return {
+            item["ParameterName"]: item
+            for data in page_iterator
+            for item in data.get("EngineDefaults", {}).get("Parameters") or []
+        }
 
     def get_db_instance(self, identifier: str) -> DBInstanceTypeDef | None:
         """Get DB instance info"""

hooks/utils/aws_api.py

@@ -1,6 +1,6 @@
 [project]
 name = "er-aws-rds"
-version = "0.6.6"
+version = "0.6.7"
 description = "ERv2 module for managing AWS rds instances"
 authors = [{ name = "AppSRE", email = "[email protected]" }]
 license = { text = "Apache 2.0" }

pyproject.toml

@@ -12,7 +12,7 @@
 
 def deep_merge(dict1: dict[str, Any], dict2: dict[str, Any]) -> dict[str, Any]:
     """Merge two dictionaries recursively"""
-    return dict1.copy() | {
+    return dict1 | {
         key: (
             deep_merge(dict1[key], value)
             if key in dict1 and isinstance(dict1[key], dict) and isinstance(value, dict)