Port extra config changes (#1203)

Author: dragomirp

config.yaml

@@ -25,6 +25,57 @@ options:
       Controls whether Gather and Gather Merge also run subplans.
     type: boolean
     default: true
+  cpu-max-logical-replication-workers:
+    description: |
+      Sets the maximum number of logical replication workers. Should be either "auto"
+      or a positive integer value. Restart is required when changed.
+      auto = max-worker-processes.
+    type: string
+    default: "auto"
+  cpu-max-parallel-apply-workers-per-subscription:
+    description: |
+      Sets the maximum number of parallel apply workers per subscription.
+      Should be either "auto" or a positive integer value.
+      auto = max-worker-processes.
+    type: string
+    default: "auto"
+  cpu-max-parallel-maintenance-workers:
+    description: |
+      Sets the maximum number of workers that can be used for parallel maintenance operations
+      like CREATE INDEX. Should be either "auto" or a positive integer value.
+      Cannot exceed max-parallel-workers.
+      auto = max-worker-processes.
+    type: string
+    default: "auto"
+  cpu-max-parallel-workers:
+    description: |
+      Sets the maximum number of workers that can be used for parallel operations.
+      Should be either "auto" or a positive integer value. Cannot exceed max_worker_processes.
+      auto = max-worker-processes.
+    type: string
+    default: "auto"
+  cpu-max-sync-workers-per-subscription:
+    description: |
+      Sets the maximum number of synchronization workers per subscription.
+      Should be either "auto" or a positive integer value.
+      auto = max-worker-processes.
+    type: string
+    default: "auto"
+  cpu-max-worker-processes:
+    description: |
+      Sets the maximum number of background processes that the system can support.
+      Should be either "auto" or a positive integer value.
+      auto = minimum(8, 2 * vCores). Value will be capped at 10 * vCores if exceeded.
+      Restart is required when changed.
+    type: string
+    default: "auto"
+  cpu-wal-compression:
+    description: |
+      Enable compression of full-page writes written in WAL. Compression reduces
+      the I/O needed for local storage and replication.
+      Default is on (true).
+    type: boolean
+    default: true
   durability-synchronous-commit:
     description: |
       Sets the current transactions synchronization level. This charm allows only the

src/charm.py

@@ -2271,6 +2271,175 @@ def _can_connect_to_postgresql(self) -> bool:
             return False
         return True
 
+    def _calculate_max_worker_processes(self, cpu_cores: int) -> str | None:
+        """Calculate cpu_max_worker_processes configuration value."""
+        if self.config.cpu_max_worker_processes == "auto":
+            # auto = minimum(8, 2 * vCores)
+            return str(min(8, 2 * cpu_cores))
+        elif self.config.cpu_max_worker_processes is not None:
+            value = self.config.cpu_max_worker_processes
+            # Pydantic already enforces minimum of 2 via conint(ge=2)
+            # This is an extra safeguard
+            if value < 2:
+                raise ValueError(f"cpu-max-worker-processes value {value} is below minimum of 2")
+            cap = 10 * cpu_cores
+            if value > cap:
+                raise ValueError(
+                    f"cpu-max-worker-processes value {value} exceeds maximum allowed "
+                    f"of {cap} (10 * vCores). Please set a value <= {cap}."
+                )
+            return str(value)
+        return None
+
+    def _validate_worker_config_value(self, param_name: str, value: int, cpu_cores: int) -> str:
+        """Shared validation logic for worker process parameters.
+
+        Args:
+            param_name: The configuration parameter name (for error messages)
+            value: The integer value to validate
+            cpu_cores: The number of available CPU cores
+
+        Returns:
+            String representation of the validated value
+
+        Raises:
+            ValueError: If value is less than 2 or exceeds 10 * vCores
+        """
+        if value < 2:
+            raise ValueError(f"{param_name} value {value} is below minimum of 2")
+        cap = 10 * cpu_cores
+        if value > cap:
+            raise ValueError(
+                f"{param_name} value {value} exceeds maximum allowed "
+                f"of {cap} (10 * vCores). Please set a value <= {cap}."
+            )
+        return str(value)
+
+    def _calculate_max_parallel_workers(self, base_max_workers: int, cpu_cores: int) -> str | None:
+        """Calculate cpu_max_parallel_workers configuration value."""
+        if self.config.cpu_max_parallel_workers == "auto":
+            return str(base_max_workers)
+        elif self.config.cpu_max_parallel_workers is not None:
+            # Validate the value first
+            validated_value_str = self._validate_worker_config_value(
+                "cpu-max-parallel-workers", self.config.cpu_max_parallel_workers, cpu_cores
+            )
+            # Apply the min constraint with base_max_workers
+            return str(min(int(validated_value_str), base_max_workers))
+        return None
+
+    def _calculate_max_parallel_maintenance_workers(
+        self, base_max_workers: int, cpu_cores: int
+    ) -> str | None:
+        """Calculate cpu_max_parallel_maintenance_workers configuration value."""
+        if self.config.cpu_max_parallel_maintenance_workers == "auto":
+            return str(base_max_workers)
+        elif self.config.cpu_max_parallel_maintenance_workers is not None:
+            return self._validate_worker_config_value(
+                "cpu-max-parallel-maintenance-workers",
+                self.config.cpu_max_parallel_maintenance_workers,
+                cpu_cores,
+            )
+        return None
+
+    def _calculate_max_logical_replication_workers(
+        self, base_max_workers: int, cpu_cores: int
+    ) -> str | None:
+        """Calculate cpu_max_logical_replication_workers configuration value."""
+        if self.config.cpu_max_logical_replication_workers == "auto":
+            return str(base_max_workers)
+        elif self.config.cpu_max_logical_replication_workers is not None:
+            return self._validate_worker_config_value(
+                "cpu-max-logical-replication-workers",
+                self.config.cpu_max_logical_replication_workers,
+                cpu_cores,
+            )
+        return None
+
+    def _calculate_max_sync_workers_per_subscription(
+        self, base_max_workers: int, cpu_cores: int
+    ) -> str | None:
+        """Calculate cpu_max_sync_workers_per_subscription configuration value."""
+        if self.config.cpu_max_sync_workers_per_subscription == "auto":
+            return str(base_max_workers)
+        elif self.config.cpu_max_sync_workers_per_subscription is not None:
+            return self._validate_worker_config_value(
+                "cpu-max-sync-workers-per-subscription",
+                self.config.cpu_max_sync_workers_per_subscription,
+                cpu_cores,
+            )
+        return None
+
+    def _calculate_max_parallel_apply_workers_per_subscription(
+        self, base_max_workers: int, cpu_cores: int
+    ) -> str | None:
+        """Calculate cpu_max_parallel_apply_workers_per_subscription configuration value."""
+        if self.config.cpu_max_parallel_apply_workers_per_subscription == "auto":
+            return str(base_max_workers)
+        elif self.config.cpu_max_parallel_apply_workers_per_subscription is not None:
+            return self._validate_worker_config_value(
+                "cpu-max-parallel-apply-workers-per-subscription",
+                self.config.cpu_max_parallel_apply_workers_per_subscription,
+                cpu_cores,
+            )
+        return None
+
+    def _calculate_worker_process_config(self, cpu_cores: int) -> dict[str, str]:
+        """Calculate worker process configuration values.
+
+        Handles 'auto' values and capping logic for worker process parameters.
+        Returns a dictionary with the calculated values ready for PostgreSQL.
+        """
+        result: dict[str, str] = {}
+
+        # Calculate cpu_max_worker_processes (baseline for other worker configs)
+        cpu_max_worker_processes_value = self._calculate_max_worker_processes(cpu_cores)
+        if cpu_max_worker_processes_value is not None:
+            result["max_worker_processes"] = cpu_max_worker_processes_value
+
+        # Get the effective cpu_max_worker_processes for dependent configs
+        # Use the calculated value, or fall back to PostgreSQL default (8)
+        base_max_workers = int(result.get("max_worker_processes", "8"))
+
+        # Calculate other worker parameters
+        cpu_max_parallel_workers_value = self._calculate_max_parallel_workers(
+            base_max_workers, cpu_cores
+        )
+        if cpu_max_parallel_workers_value is not None:
+            result["max_parallel_workers"] = cpu_max_parallel_workers_value
+
+        cpu_max_parallel_maintenance_workers_value = (
+            self._calculate_max_parallel_maintenance_workers(base_max_workers, cpu_cores)
+        )
+        if cpu_max_parallel_maintenance_workers_value is not None:
+            result["max_parallel_maintenance_workers"] = cpu_max_parallel_maintenance_workers_value
+
+        cpu_max_logical_replication_workers_value = (
+            self._calculate_max_logical_replication_workers(base_max_workers, cpu_cores)
+        )
+        if cpu_max_logical_replication_workers_value is not None:
+            result["max_logical_replication_workers"] = cpu_max_logical_replication_workers_value
+
+        cpu_max_sync_workers_per_subscription_value = (
+            self._calculate_max_sync_workers_per_subscription(base_max_workers, cpu_cores)
+        )
+        if cpu_max_sync_workers_per_subscription_value is not None:
+            result["max_sync_workers_per_subscription"] = (
+                cpu_max_sync_workers_per_subscription_value
+            )
+
+        cpu_max_parallel_apply_workers_per_subscription_value = (
+            self._calculate_max_parallel_apply_workers_per_subscription(
+                base_max_workers, cpu_cores
+            )
+        )
+        if cpu_max_parallel_apply_workers_per_subscription_value is not None:
+            result["max_parallel_apply_workers_per_subscription"] = (
+                cpu_max_parallel_apply_workers_per_subscription_value
+            )
+
+        return result
+
     def _api_update_config(self, available_cpu_cores: int) -> None:
         # Use config value if set, calculate otherwise
         if self.config.experimental_max_connections:
@@ -2286,18 +2455,66 @@ def _api_update_config(self, available_cpu_cores: int) -> None:
             "shared_buffers": self.config.memory_shared_buffers,
             "wal_keep_size": self.config.durability_wal_keep_size,
         }
+
+        # Add restart-required worker process parameters via Patroni API
+        worker_configs = self._calculate_worker_process_config(available_cpu_cores)
+        if "max_worker_processes" in worker_configs:
+            cfg_patch["max_worker_processes"] = worker_configs["max_worker_processes"]
+
+        if "max_logical_replication_workers" in worker_configs:
+            cfg_patch["max_logical_replication_workers"] = worker_configs[
+                "max_logical_replication_workers"
+            ]
+
         base_patch = {}
         if primary_endpoint := self.async_replication.get_primary_cluster_endpoint():
             base_patch["standby_cluster"] = {"host": primary_endpoint}
         self._patroni.bulk_update_parameters_controller_by_patroni(cfg_patch, base_patch)
 
-    def update_config(self, is_creating_backup: bool = False) -> bool:
-        """Updates Patroni config file based on the existence of the TLS files."""
-        # Retrieve PostgreSQL parameters.
+    def _build_postgresql_parameters(
+        self, available_cpu_cores: int, available_memory: int
+    ) -> dict | None:
+        """Build PostgreSQL configuration parameters.
+
+        Args:
+            available_cpu_cores: Number of available CPU cores
+            available_memory: Available memory in bytes
+
+        Returns:
+            Dictionary of PostgreSQL parameters or None if base parameters couldn't be built.
+        """
+        limit_memory = None
         if self.config.profile_limit_memory:
             limit_memory = self.config.profile_limit_memory * 10**6
+
+        # Build PostgreSQL parameters.
+        pg_parameters = self.postgresql.build_postgresql_parameters(
+            self.model.config, available_memory, limit_memory
+        )
+
+        # Calculate and merge worker process configurations
+        worker_configs = self._calculate_worker_process_config(available_cpu_cores)
+
+        # Add cpu_wal_compression configuration (separate from worker processes)
+        if self.config.cpu_wal_compression is not None:
+            cpu_wal_compression = "on" if self.config.cpu_wal_compression else "off"
+        else:
+            # Use config.yaml default when unset (default: true)
+            cpu_wal_compression = "on"
+
+        if pg_parameters is not None:
+            pg_parameters.update(worker_configs)
+            pg_parameters["wal_compression"] = cpu_wal_compression
         else:
-            limit_memory = None
+            pg_parameters = dict(worker_configs)
+            pg_parameters["wal_compression"] = cpu_wal_compression
+            logger.debug(f"pg_parameters set to worker_configs = {pg_parameters}")
+
+        return pg_parameters
+
+    def update_config(self, is_creating_backup: bool = False) -> bool:
+        """Updates Patroni config file based on the existence of the TLS files."""
+        # Retrieve PostgreSQL parameters.
         try:
             available_cpu_cores, available_memory = self.get_available_resources()
         except ApiError as e:
@@ -2306,11 +2523,9 @@ def update_config(self, is_creating_backup: bool = False) -> bool:
                 return False
             raise e
 
-        # TODO Updating the lib should accept ConfigData
-        postgresql_parameters = self.postgresql.build_postgresql_parameters(
-            self.model.config,  # type: ignore
-            available_memory,
-            limit_memory,
+        # Build PostgreSQL parameters
+        postgresql_parameters = self._build_postgresql_parameters(
+            available_cpu_cores, available_memory
         )
 
         # replication_slots = self.logical_replication.replication_slots()

src/config.py

@@ -5,23 +5,42 @@
 """Structured configuration for the PostgreSQL charm."""
 
 import logging
-from typing import Literal
+from typing import Annotated, Literal
 
 from charms.data_platform_libs.v1.data_models import BaseConfigModel
-from pydantic import Field, PositiveInt
+from pydantic import Field, PositiveInt, conint
 
 from locales import ROCK_LOCALES
 
 logger = logging.getLogger(__name__)
 
 
+# Type for worker process parameters that must be >= 2
+WorkerProcessInt = Annotated[int, conint(ge=2)]
+
+
 class CharmConfig(BaseConfigModel):
     """Manager for the structured configuration."""
 
     synchronous_node_count: Literal["all", "majority"] | PositiveInt = Field(default="all")
     connection_authentication_timeout: int | None = Field(ge=1, le=600, default=None)
     connection_statement_timeout: int | None = Field(ge=0, le=2147483647, default=None)
+    cpu_max_logical_replication_workers: Literal["auto"] | WorkerProcessInt | None = Field(
+        default="auto"
+    )
+    cpu_max_parallel_apply_workers_per_subscription: Literal["auto"] | WorkerProcessInt | None = (
+        Field(default="auto")
+    )
+    cpu_max_parallel_maintenance_workers: Literal["auto"] | WorkerProcessInt | None = Field(
+        default="auto"
+    )
+    cpu_max_parallel_workers: Literal["auto"] | WorkerProcessInt | None = Field(default="auto")
+    cpu_max_sync_workers_per_subscription: Literal["auto"] | WorkerProcessInt | None = Field(
+        default="auto"
+    )
+    cpu_max_worker_processes: Literal["auto"] | WorkerProcessInt | None = Field(default="auto")
     cpu_parallel_leader_participation: bool | None = Field(default=None)
+    cpu_wal_compression: bool | None = Field(default=None)
     durability_synchronous_commit: Literal["on", "remote_apply", "remote_write"] | None = Field(
         default=None
     )