Add external-services and ca-certificates spec keys

New spec.yml features for routing external service dependencies:

external-services:
  s3:
    host: example.com  # ExternalName Service (production)
    port: 443
  s3:
    selector: {app: mock}  # headless Service + Endpoints (testing)
    namespace: mock-ns
    port: 443

ca-certificates:
  - ~/.local/share/mkcert/rootCA.pem  # testing only

laconic-so creates the appropriate k8s Service type per mode:
- host mode: ExternalName (DNS CNAME to external provider)
- selector mode: headless Service + Endpoints with pod IPs
  discovered from the target namespace at deploy time

ca-certificates mounts CA files into all containers at
/etc/ssl/certs/ and sets NODE_EXTRA_CA_CERTS for Node/Bun.

Also includes the previously committed PV Released state fix.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
afd-dumpster-local-testing
Snake Game Developer 2026-03-21 15:25:47 +00:00
parent 98ff221a21
commit 713a81c245
4 changed files with 285 additions and 0 deletions

View File

@ -47,3 +47,5 @@ high_memlock_runtime = "high-memlock"
high_memlock_spec_filename = "high-memlock-spec.json" high_memlock_spec_filename = "high-memlock-spec.json"
acme_email_key = "acme-email" acme_email_key = "acme-email"
kind_mount_root_key = "kind-mount-root" kind_mount_root_key = "kind-mount-root"
external_services_key = "external-services"
ca_certificates_key = "ca-certificates"

View File

@ -845,6 +845,20 @@ class ClusterInfo:
if multi_pod: if multi_pod:
selector_labels["app.kubernetes.io/component"] = pod_name selector_labels["app.kubernetes.io/component"] = pod_name
# Add CA certificate volume and env vars if configured
_ca_secret, ca_volume, ca_mount, ca_envs = (
self.get_ca_certificate_resources()
)
if ca_volume:
volumes.append(ca_volume)
for container in containers:
if container.volume_mounts is None:
container.volume_mounts = []
container.volume_mounts.append(ca_mount)
if container.env is None:
container.env = []
container.env.extend(ca_envs)
template = client.V1PodTemplateSpec( template = client.V1PodTemplateSpec(
metadata=client.V1ObjectMeta(annotations=annotations, labels=labels), metadata=client.V1ObjectMeta(annotations=annotations, labels=labels),
spec=client.V1PodSpec( spec=client.V1PodSpec(
@ -1055,3 +1069,121 @@ class ClusterInfo:
jobs.append(job) jobs.append(job)
return jobs return jobs
def get_external_service_resources(self) -> List:
"""Build k8s Services (and Endpoints) for external-services in spec.
Two modes:
- host mode: ExternalName Service (DNS CNAME to external host)
- selector mode: headless Service + Endpoints (cross-namespace
routing to a mock pod, IP discovered at deploy time)
Returns a flat list of k8s resource objects (Services + Endpoints).
"""
ext_services = self.spec.get_external_services()
if not ext_services:
return []
resources = []
for name, config in ext_services.items():
port = config.get("port", 443)
if "host" in config:
# ExternalName: DNS CNAME to external host
svc = client.V1Service(
metadata=client.V1ObjectMeta(
name=name,
labels={"app": self.app_name},
),
spec=client.V1ServiceSpec(
type="ExternalName",
external_name=config["host"],
ports=[
client.V1ServicePort(port=port, name=f"port-{port}")
],
),
)
resources.append(svc)
elif "selector" in config and "namespace" in config:
# Cross-namespace headless Service + Endpoints.
# The Endpoints IP is populated in deploy_k8s.py at deploy
# time by querying the target namespace for matching pods.
svc = client.V1Service(
metadata=client.V1ObjectMeta(
name=name,
labels={"app": self.app_name},
),
spec=client.V1ServiceSpec(
cluster_ip="None",
ports=[
client.V1ServicePort(port=port, name=f"port-{port}")
],
),
)
resources.append(svc)
# Endpoints object is created in deploy_k8s.py after pod
# IP discovery — we just return the Service here.
return resources
def get_ca_certificate_resources(self) -> tuple:
"""Build k8s Secret and volume mount config for CA certificates.
Returns (secret, volume, volume_mount, env_vars) or (None, ...) if
no CA certificates are configured. The caller must add the volume
and mount to all containers, and the env vars to all containers.
"""
ca_files = self.spec.get_ca_certificates()
if not ca_files:
return None, None, None, []
# Concatenate all CA files into one Secret
secret_data = {}
for i, ca_path in enumerate(ca_files):
expanded = os.path.expanduser(ca_path)
if not os.path.exists(expanded):
print(f"Warning: CA certificate file not found: {expanded}")
continue
with open(expanded, "rb") as f:
ca_bytes = f.read()
key = f"laconic-extra-ca-{i}.pem"
secret_data[key] = base64.b64encode(ca_bytes).decode()
if not secret_data:
return None, None, None, []
secret_name = f"{self.app_name}-ca-certificates"
secret = client.V1Secret(
metadata=client.V1ObjectMeta(
name=secret_name,
labels={"app": self.app_name},
),
data=secret_data,
)
volume = client.V1Volume(
name="laconic-ca-certs",
secret=client.V1SecretVolumeSource(
secret_name=secret_name,
),
)
# Mount each CA file into /etc/ssl/certs/ (Go reads this dir)
volume_mount = client.V1VolumeMount(
name="laconic-ca-certs",
mount_path="/etc/ssl/certs/laconic-extra-ca",
read_only=True,
)
# Set NODE_EXTRA_CA_CERTS for Node/Bun containers.
# Point at the first CA file (most common: single mkcert root CA).
first_key = list(secret_data.keys())[0]
env_vars = [
client.V1EnvVar(
name="NODE_EXTRA_CA_CERTS",
value=f"/etc/ssl/certs/laconic-extra-ca/{first_key}",
),
]
return secret, volume, volume_mount, env_vars

View File

@ -420,6 +420,134 @@ class K8sDeployer(Deployer):
else: else:
raise raise
def _create_external_services(self):
"""Create k8s Services for external-services declared in the spec.
For host mode: ExternalName Service (DNS CNAME).
For selector mode: headless Service + Endpoints with pod IPs
discovered from the target namespace.
"""
resources = self.cluster_info.get_external_service_resources()
ext_services = self.cluster_info.spec.get_external_services()
for resource in resources:
if opts.o.dry_run:
print(f"Dry run: would create external service: {resource.metadata.name}")
continue
svc_name = resource.metadata.name
try:
self.core_api.create_namespaced_service(
body=resource, namespace=self.k8s_namespace
)
print(f"Created external service '{svc_name}'")
except ApiException as e:
if e.status == 409:
self.core_api.replace_namespaced_service(
name=svc_name,
namespace=self.k8s_namespace,
body=resource,
)
print(f"Updated external service '{svc_name}'")
else:
raise
# Create Endpoints for selector-mode services
for name, config in ext_services.items():
if "selector" not in config or "namespace" not in config:
continue
if opts.o.dry_run:
continue
target_ns = config["namespace"]
selector = config["selector"]
port = config.get("port", 443)
# Build label selector string from dict
label_selector = ",".join(f"{k}={v}" for k, v in selector.items())
# Discover pod IPs in target namespace
pods = self.core_api.list_namespaced_pod(
namespace=target_ns, label_selector=label_selector
)
pod_ips = [
p.status.pod_ip
for p in pods.items
if p.status and p.status.pod_ip
]
if not pod_ips:
print(
f"Warning: no pods found in {target_ns} matching "
f"{label_selector} for external service '{name}'"
)
continue
endpoints = client.V1Endpoints(
metadata=client.V1ObjectMeta(
name=name,
labels={"app": self.cluster_info.app_name},
),
subsets=[
client.V1EndpointSubset(
addresses=[
client.V1EndpointAddress(ip=ip) for ip in pod_ips
],
ports=[
client.CoreV1EndpointPort(
port=port, name=f"port-{port}"
)
],
)
],
)
try:
self.core_api.create_namespaced_endpoints(
body=endpoints, namespace=self.k8s_namespace
)
print(f"Created endpoints for '{name}'{pod_ips}")
except ApiException as e:
if e.status == 409:
self.core_api.replace_namespaced_endpoints(
name=name,
namespace=self.k8s_namespace,
body=endpoints,
)
print(f"Updated endpoints for '{name}'{pod_ips}")
else:
raise
def _create_ca_certificates(self):
"""Create k8s Secret for CA certificates declared in the spec.
The Secret is mounted into containers by get_deployments() in
cluster_info.py. This method just ensures the Secret exists.
"""
ca_secret, _, _, _ = self.cluster_info.get_ca_certificate_resources()
if not ca_secret:
return
if opts.o.dry_run:
print(f"Dry run: would create CA certificate secret")
return
secret_name = ca_secret.metadata.name
try:
self.core_api.create_namespaced_secret(
body=ca_secret, namespace=self.k8s_namespace
)
print(f"Created CA certificate secret '{secret_name}'")
except ApiException as e:
if e.status == 409:
self.core_api.replace_namespaced_secret(
name=secret_name,
namespace=self.k8s_namespace,
body=ca_secret,
)
print(f"Updated CA certificate secret '{secret_name}'")
else:
raise
def _create_deployment(self): def _create_deployment(self):
# Skip if there are no pods to deploy (e.g. jobs-only stacks) # Skip if there are no pods to deploy (e.g. jobs-only stacks)
if not self.cluster_info.parsed_pod_yaml_map: if not self.cluster_info.parsed_pod_yaml_map:
@ -701,6 +829,8 @@ class K8sDeployer(Deployer):
) )
self._create_volume_data() self._create_volume_data()
self._create_external_services()
self._create_ca_certificates()
self._create_deployment() self._create_deployment()
self._create_jobs() self._create_jobs()
self._create_ingress() self._create_ingress()

View File

@ -284,5 +284,26 @@ class Spec:
""" """
return self.obj.get("maintenance-service") return self.obj.get("maintenance-service")
def get_external_services(self) -> typing.Dict[str, typing.Dict]:
"""Return external-services config from spec.
Each entry maps a service name to its routing config:
- host mode: {host: "example.com", port: 443}
ExternalName k8s Service (DNS CNAME)
- selector mode: {selector: {app: "foo"}, namespace: "ns", port: 443}
Headless Service + Endpoints (cross-namespace routing to mock pod)
"""
return self.obj.get(constants.external_services_key, {})
def get_ca_certificates(self) -> typing.List[str]:
"""Return list of CA certificate file paths to trust.
Used in testing specs to inject mkcert root CAs so containers
trust TLS certs on mock services. Files are mounted into all
containers at /etc/ssl/certs/ and NODE_EXTRA_CA_CERTS is set.
Production specs omit this key entirely.
"""
return self.obj.get(constants.ca_certificates_key, [])
def is_docker_deployment(self): def is_docker_deployment(self):
return self.get_deployment_type() in [constants.compose_deploy_type] return self.get_deployment_type() in [constants.compose_deploy_type]