feat(config): port PostgreSQL parameter building + resource introspection

pyproject.toml

@@ -40,6 +40,10 @@ vm = [
     "charmlibs-snap>=1.0.1; python_version >= '3.12'",
     "charmlibs-systemd>=1.0.0.post0; python_version >= '3.12'"
 ]
+k8s = [
+    "lightkube>=0.21.0; python_version >= '3.12'",
+    "lightkube-models>=1.28.1.4; python_version >= '3.12'",
+]
 db-driver = [
     "psycopg2>=2.9.10",
 ]

single_kernel_postgresql/charms/k8s_charm.py

@@ -52,7 +52,10 @@ def workload(self) -> BaseWorkload:
             BaseWorkload: The K8sWorkload instance for this charm
         """
         return K8sWorkload(
-            charm_dir=self.charm_dir, container=self.unit.get_container(CONTAINER_NAME)
+            charm_dir=self.charm_dir,
+            container=self.unit.get_container(CONTAINER_NAME),
+            unit_name=self.unit.name,
+            namespace=self.model.name,
         )
 
     @property

single_kernel_postgresql/config/exceptions.py

@@ -73,3 +73,7 @@ class SwitchoverNotSyncError(SwitchoverFailedError):
 
 class UpdateSyncNodeCountError(Exception):
     """Raised when updating synchronous_node_count failed for some reason."""
+
+
+class DeployedWithoutTrustError(Exception):
+    """Raised when the K8s API denies access because the app wasn't deployed with --trust."""

single_kernel_postgresql/managers/config.py

@@ -65,6 +65,168 @@ def configure_patroni_on_unit(self):
             self.workload.paths.data, mode=POSTGRESQL_STORAGE_PERMISSIONS, exist_ok=True
         )
 
+    def _calculate_max_worker_processes(self, cpu_cores: int) -> str | None:
+        """Calculate cpu_max_worker_processes configuration value."""
+        if self.state.config.cpu_max_worker_processes == "auto":
+            # auto = minimum(8, 2 * vCores)
+            return str(min(8, 2 * cpu_cores))
+        elif self.state.config.cpu_max_worker_processes is not None:
+            value = self.state.config.cpu_max_worker_processes
+            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 exceeds 10 * vCores.
+        """
+        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.state.config.cpu_max_parallel_workers == "auto":
+            return str(base_max_workers)
+        elif self.state.config.cpu_max_parallel_workers is not None:
+            validated_value_str = self._validate_worker_config_value(
+                "cpu-max-parallel-workers", self.state.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.state.config.cpu_max_parallel_maintenance_workers == "auto":
+            return str(base_max_workers)
+        elif self.state.config.cpu_max_parallel_maintenance_workers is not None:
+            return self._validate_worker_config_value(
+                "cpu-max-parallel-maintenance-workers",
+                self.state.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.state.config.cpu_max_logical_replication_workers == "auto":
+            return str(base_max_workers)
+        elif self.state.config.cpu_max_logical_replication_workers is not None:
+            return self._validate_worker_config_value(
+                "cpu-max-logical-replication-workers",
+                self.state.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.state.config.cpu_max_sync_workers_per_subscription == "auto":
+            return str(base_max_workers)
+        elif self.state.config.cpu_max_sync_workers_per_subscription is not None:
+            return self._validate_worker_config_value(
+                "cpu-max-sync-workers-per-subscription",
+                self.state.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.state.config.cpu_max_parallel_apply_workers_per_subscription == "auto":
+            return str(base_max_workers)
+        elif self.state.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.state.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 _build_postgresql_parameters(
         self, postgresql_client: PostgreSQLClient
     ) -> dict[str, str] | None:
@@ -73,17 +235,33 @@ def _build_postgresql_parameters(
         Returns:
             Dictionary of PostgreSQL parameters or None if base parameters couldn't be built.
         """
+        cpu_cores, available_memory = self.workload.get_available_resources()
+
         limit_memory = None
         if self.state.config.profile_limit_memory:
             limit_memory = self.state.config.profile_limit_memory * 10**6
 
         # Build PostgreSQL parameters.
         pg_parameters = postgresql_client.build_postgresql_parameters(
-            self.state.model_config, self.workload.get_available_memory(), limit_memory
+            self.state.model_config, available_memory, limit_memory
         )
 
         # Calculate and merge worker process configurations
-        # TODO: Add additional parameters
+        worker_configs = self._calculate_worker_process_config(cpu_cores)
+
+        # Add cpu_wal_compression configuration (separate from worker processes)
+        if self.state.config.cpu_wal_compression is not None:
+            cpu_wal_compression = "on" if self.state.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:
+            pg_parameters = dict(worker_configs)
+            pg_parameters["wal_compression"] = cpu_wal_compression
 
         return pg_parameters
 

single_kernel_postgresql/workload/base.py

@@ -273,3 +273,8 @@ def get_postgresql_version(self) -> str:
     def get_available_memory(self) -> int:
         """Returns the system available memory in bytes."""
         pass
+
+    @abstractmethod
+    def get_available_resources(self) -> tuple[int, int]:
+        """Returns the available (cpu_cores, memory_bytes) for the workload."""
+        pass

single_kernel_postgresql/workload/k8s.py

@@ -12,14 +12,25 @@
 
 from charmlibs import pathops
 from charmlibs.pathops import PathProtocol
+from lightkube import Client
+from lightkube.core.exceptions import ApiError
+from lightkube.resources.core_v1 import Node, Pod
 from ops import Container, ModelError
 from ops.pebble import Plan, ServiceStatus
 
-from single_kernel_postgresql.config.exceptions import PostgreSQLFileOperationError
+from single_kernel_postgresql.config.exceptions import (
+    DeployedWithoutTrustError,
+    PostgreSQLFileOperationError,
+)
 from single_kernel_postgresql.config.literals import (
     DIR_PERMISSIONS_READONLY,
     K8S_POSTGRESQL_SERVICE_NAME,
 )
+from single_kernel_postgresql.utils import (
+    any_cpu_to_cores,
+    any_memory_to_bytes,
+    unit_name_to_pod_name,
+)
 from single_kernel_postgresql.workload.base import BaseWorkload
 from single_kernel_postgresql.workload.paths.base import Paths as BasePaths
 from single_kernel_postgresql.workload.paths.k8s import K8sPaths
@@ -30,15 +41,20 @@
 class K8sWorkload(BaseWorkload):
     """Kubernetes PostgreSQL Workload."""
 
-    def __init__(self, charm_dir: Path, container: Container):
+    def __init__(self, charm_dir: Path, container: Container, *, unit_name: str, namespace: str):
         """Initialize workload.
 
         Args:
             charm_dir: the path to charm code.
             container: the Container instance.
+            unit_name: the Juju unit name (e.g. "postgresql-k8s/0"), used to resolve
+                this unit's own pod for K8s resource-limit lookups.
+            namespace: the Juju model name, i.e. the K8s namespace the pod lives in.
         """
         super().__init__(charm_dir=charm_dir)
         self.container = container
+        self.unit_name = unit_name
+        self.namespace = namespace
         self._paths: BasePaths | None = None
 
     def install(self) -> None:
@@ -225,3 +241,70 @@ def workload_present(self) -> bool:
     def get_available_memory(self) -> int:
         """Returns the system available memory in bytes."""
         raise NotImplementedError
+
+    def _get_node_name_for_pod(self) -> str:
+        """Return the node name for this unit's own pod."""
+        client = Client()
+        pod = client.get(Pod, name=unit_name_to_pod_name(self.unit_name), namespace=self.namespace)
+        if pod.spec and pod.spec.nodeName:
+            return pod.spec.nodeName
+        raise RuntimeError("Pod doesn't exist")
+
+    def get_resources_limits(self, container_name: str) -> dict:
+        """Return resources limits for a given container.
+
+        Args:
+            container_name: name of the container to get resources limits for.
+        """
+        client = Client()
+        pod = client.get(Pod, name=unit_name_to_pod_name(self.unit_name), namespace=self.namespace)
+        if pod.spec:
+            for container in pod.spec.containers:
+                if container.name == container_name and container.resources:
+                    return container.resources.limits or {}
+        return {}
+
+    def get_node_allocable_memory(self) -> int:
+        """Return the allocable memory in bytes for the current K8s node."""
+        client = Client()
+        node = client.get(Node, name=self._get_node_name_for_pod(), namespace=self.namespace)  # type: ignore
+        return any_memory_to_bytes(node.status.allocatable["memory"])
+
+    def get_node_cpu_cores(self) -> int:
+        """Return the number of CPU cores for the current K8s node."""
+        client = Client()
+        node = client.get(Node, name=self._get_node_name_for_pod(), namespace=self.namespace)  # type: ignore
+        return any_cpu_to_cores(node.status.allocatable["cpu"])
+
+    def get_available_resources(self) -> tuple[int, int]:
+        """Returns the available (cpu_cores, memory_bytes) for the workload.
+
+        Raises:
+            DeployedWithoutTrustError: if the K8s API denies access (403), meaning the
+                app wasn't deployed with ``--trust``.
+        """
+        try:
+            cpu_cores = self.get_node_cpu_cores()
+            allocable_memory = self.get_node_allocable_memory()
+            container_limits = self.get_resources_limits(
+                container_name=K8S_POSTGRESQL_SERVICE_NAME
+            )
+        except ApiError as e:
+            if e.status.code == 403:
+                raise DeployedWithoutTrustError from e
+            raise
+
+        if "cpu" in container_limits:
+            cpu_str = container_limits["cpu"]
+            constrained_cpu = any_cpu_to_cores(cpu_str)
+            if constrained_cpu < cpu_cores:
+                logger.debug(f"CPU constrained to {cpu_str} cores from resource limit")
+                cpu_cores = constrained_cpu
+        if "memory" in container_limits:
+            memory_str = container_limits["memory"]
+            constrained_memory = any_memory_to_bytes(memory_str)
+            if constrained_memory < allocable_memory:
+                logger.debug(f"Memory constrained to {memory_str} from resource limit")
+                allocable_memory = constrained_memory
+
+        return cpu_cores, allocable_memory

single_kernel_postgresql/workload/vm.py

@@ -4,6 +4,7 @@
 """Machine Workload."""
 
 import logging
+import os
 import pathlib
 import platform
 import subprocess
@@ -233,3 +234,7 @@ def get_available_memory(self) -> int:
                     return int(line.split()[1]) * 1024
 
         return 0
+
+    def get_available_resources(self) -> tuple[int, int]:
+        """Returns the available (cpu_cores, memory_bytes) for the workload."""
+        return os.cpu_count() or 1, self.get_available_memory()