clusterd: set TCP keepalive sysctls on pods (#36351)

## Summary
- Container network namespaces get default kernel values, not the host's
custom sysctl settings — host-level TCP keepalive tuning was being
silently overwritten by container defaults
- Adds `net.ipv4.tcp_keepalive_time=300`,
`net.ipv4.tcp_keepalive_intvl=30`, `net.ipv4.tcp_keepalive_probes=3` to
the clusterd `PodSecurityContext`
- Adds `validate_clusterd_pod()` to the orchestratord integration tests,
called from all three existing clusterd pod validation sites
(`SwapEnabledGlobal`, `StorageClass`, `ClusterdCpu`)

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: jubrad

src/orchestrator-kubernetes/src/lib.rs

@@ -30,7 +30,7 @@ use k8s_openapi::api::core::v1::{
     PersistentVolumeClaimTemplate, Pod, PodAffinity, PodAffinityTerm, PodAntiAffinity,
     PodSecurityContext, PodSpec, PodTemplateSpec, PreferredSchedulingTerm, ResourceRequirements,
     SeccompProfile, Secret, SecurityContext, Service as K8sService, ServicePort, ServiceSpec,
-    Toleration, TopologySpreadConstraint, Volume, VolumeMount, VolumeResourceRequirements,
+    Sysctl, Toleration, TopologySpreadConstraint, Volume, VolumeMount, VolumeResourceRequirements,
     WeightedPodAffinityTerm,
 };
 use k8s_openapi::apimachinery::pkg::api::resource::Quantity;
@@ -1089,15 +1089,34 @@ impl NamespacedOrchestrator for NamespacedKubernetesOrchestrator {
             None
         };
 
+        let tcp_keepalive_sysctls = vec![
+            Sysctl {
+                name: "net.ipv4.tcp_keepalive_time".to_string(),
+                value: "300".to_string(),
+            },
+            Sysctl {
+                name: "net.ipv4.tcp_keepalive_intvl".to_string(),
+                value: "30".to_string(),
+            },
+            Sysctl {
+                name: "net.ipv4.tcp_keepalive_probes".to_string(),
+                value: "3".to_string(),
+            },
+        ];
+
         let security_context = if let Some(fs_group) = self.config.service_fs_group {
             Some(PodSecurityContext {
                 fs_group: Some(fs_group),
                 run_as_user: Some(fs_group),
                 run_as_group: Some(fs_group),
+                sysctls: Some(tcp_keepalive_sysctls),
                 ..Default::default()
             })
         } else {
-            None
+            Some(PodSecurityContext {
+                sysctls: Some(tcp_keepalive_sysctls),
+                ..Default::default()
+            })
         };
 
         let mut tolerations = vec![

test/orchestratord/mzcompose.py

@@ -1148,6 +1148,26 @@ def validate_node_selector(
         assert "materialize.cloud/scratch-fs" not in selector
 
 
+EXPECTED_CLUSTERD_SYSCTLS = {
+    "net.ipv4.tcp_keepalive_time": "300",
+    "net.ipv4.tcp_keepalive_intvl": "30",
+    "net.ipv4.tcp_keepalive_probes": "3",
+}
+
+
+def validate_clusterd_pod(pod: dict[str, Any]) -> None:
+    """Validate always-on properties of a clusterd pod."""
+    sysctls = {
+        s["name"]: s["value"]
+        for s in pod["spec"].get("securityContext", {}).get("sysctls", [])
+    }
+    for name, expected in EXPECTED_CLUSTERD_SYSCTLS.items():
+        actual = sysctls.get(name)
+        assert (
+            actual == expected
+        ), f"Expected sysctl {name}={expected}, but got {actual}"
+
+
 def validate_container_resources(
     resources: dict[str, dict[str, str]],
     swap_enabled: bool,
@@ -1196,6 +1216,7 @@ def validate(self, mods: dict[type[Modification], Any]) -> None:
 
         def check_pods() -> None:
             clusterd = get_pod_data(labels)["items"][0]
+            validate_clusterd_pod(clusterd)
 
             resources = clusterd["spec"]["containers"][0]["resources"]
             validate_container_resources(resources, self.value)
@@ -1251,6 +1272,7 @@ def validate(self, mods: dict[type[Modification], Any]) -> None:
 
         def check_pods() -> None:
             clusterd = get_pod_data(labels)["items"][0]
+            validate_clusterd_pod(clusterd)
 
             resources = clusterd["spec"]["containers"][0]["resources"]
             validate_container_resources(resources, mods[SwapEnabledGlobal])
@@ -1366,6 +1388,7 @@ def cpu_to_k8s_quantity(cpu: float | None) -> str | None:
 
         def check_pods() -> None:
             clusterd = get_pod_data(labels)["items"][0]
+            validate_clusterd_pod(clusterd)
             resources = clusterd["spec"]["containers"][0]["resources"]
             actual_cpu_request = resources.get("requests", {}).get("cpu")
             actual_cpu_limit = resources.get("limits", {}).get("cpu")