Support per-volume resource sizing in spec files

Add get_volume_resources_for(volume_name) to look up per-volume storage
sizes from the spec. Supports both the original global format and a new
per-volume format:

  # Global (unchanged, backwards compatible)
  resources:
    volumes:
      reservations:
        storage: 5Gi

  # Per-volume (new)
  resources:
    volumes:
      my-data:
        reservations:
          storage: 10Gi
      my-cache:
        reservations:
          storage: 1Gi

Fallback chain: per-volume -> global -> default (2Gi).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

stack_orchestrator/deploy/k8s/cluster_info.py

@@ -273,19 +273,25 @@ class ClusterInfo:
         result = []
         spec_volumes = self.spec.get_volumes()
         named_volumes = self._all_named_volumes()
-        resources = self.spec.get_volume_resources()
-        if not resources:
-            resources = DEFAULT_VOLUME_RESOURCES
+        global_resources = self.spec.get_volume_resources()
+        if not global_resources:
+            global_resources = DEFAULT_VOLUME_RESOURCES
         if opts.o.debug:
             print(f"Spec Volumes: {spec_volumes}")
             print(f"Named Volumes: {named_volumes}")
-            print(f"Resources: {resources}")
+            print(f"Resources: {global_resources}")
         for volume_name, volume_path in spec_volumes.items():
             if volume_name not in named_volumes:
                 if opts.o.debug:
                     print(f"{volume_name} not in pod files")
                 continue
 
+            # Per-volume resources override global, which overrides default.
+            vol_resources = (
+                self.spec.get_volume_resources_for(volume_name)
+                or global_resources
+            )
+
             labels = {
                 "app": self.app_name,
                 "volume-label": f"{self.app_name}-{volume_name}",
@@ -301,7 +307,7 @@ class ClusterInfo:
             spec = client.V1PersistentVolumeClaimSpec(
                 access_modes=["ReadWriteOnce"],
                 storage_class_name=storage_class_name,
-                resources=to_k8s_resource_requirements(resources),
+                resources=to_k8s_resource_requirements(vol_resources),
                 volume_name=k8s_volume_name,
             )
             pvc = client.V1PersistentVolumeClaim(
@@ -353,9 +359,9 @@ class ClusterInfo:
         result = []
         spec_volumes = self.spec.get_volumes()
         named_volumes = self._all_named_volumes()
-        resources = self.spec.get_volume_resources()
-        if not resources:
-            resources = DEFAULT_VOLUME_RESOURCES
+        global_resources = self.spec.get_volume_resources()
+        if not global_resources:
+            global_resources = DEFAULT_VOLUME_RESOURCES
         for volume_name, volume_path in spec_volumes.items():
             # We only need to create a volume if it is fully qualified HostPath.
             # Otherwise, we create the PVC and expect the node to allocate the volume
@@ -384,6 +390,10 @@ class ClusterInfo:
                     )
                     continue
 
+            vol_resources = (
+                self.spec.get_volume_resources_for(volume_name)
+                or global_resources
+            )
             if self.spec.is_kind_deployment():
                 host_path = client.V1HostPathVolumeSource(
                     path=get_kind_pv_bind_mount_path(volume_name)
@@ -393,7 +403,7 @@ class ClusterInfo:
             spec = client.V1PersistentVolumeSpec(
                 storage_class_name="manual",
                 access_modes=["ReadWriteOnce"],
-                capacity=to_k8s_resource_requirements(resources).requests,
+                capacity=to_k8s_resource_requirements(vol_resources).requests,
                 host_path=host_path,
             )
             pv = client.V1PersistentVolume(

stack_orchestrator/deploy/spec.py

@@ -149,6 +149,39 @@ class Spec:
             self.obj.get(constants.resources_key, {}).get(constants.volumes_key, {})
         )
 
+    def get_volume_resources_for(self, volume_name: str) -> typing.Optional[Resources]:
+        """Look up per-volume resource overrides from spec.yml.
+
+        Supports two formats under resources.volumes:
+
+        Global (original):
+            resources:
+              volumes:
+                reservations:
+                  storage: 5Gi
+
+        Per-volume (new):
+            resources:
+              volumes:
+                my-volume:
+                  reservations:
+                    storage: 10Gi
+
+        Returns the per-volume Resources if found, otherwise None.
+        The caller should fall back to get_volume_resources() then the default.
+        """
+        vol_section = (
+            self.obj.get(constants.resources_key, {}).get(constants.volumes_key, {})
+        )
+        if volume_name not in vol_section:
+            return None
+        entry = vol_section[volume_name]
+        if isinstance(entry, dict) and (
+            "reservations" in entry or "limits" in entry
+        ):
+            return Resources(entry)
+        return None
+
     def get_http_proxy(self):
         return self.obj.get(constants.network_key, {}).get(constants.http_proxy_key, [])