deep-stack/stack-orchestrator
3
0 Fork 0
forked from cerc-io/stack-orchestrator
Code Pull Requests Activity

fix: add pre-commit hooks and fix all lint/type/format errors

Browse Source 
Process bug fix: no pre-commit existed for this repo's Python code.
Added pyproject.toml with unified dependencies (ruff, mypy, ansible-lint),
.pre-commit-config.yaml with repo-based hooks (ruff) and local uv-run
hooks (mypy, ansible-lint).

Fixed 249 ruff errors (B023, B904, B006, B007, UP008, UP031, C408),
~13 mypy type errors, 11 ansible-lint violations, and ruff-format
across all Python files including stack-orchestrator subtree.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
A. F. DudleyandClaude Opus 4.6 committed 2026-03-10 14:56:22 +00:00
1 parent 7f12270939
commit fdde3be5c8
52 files changed
+692 -1221
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

stack_orchestrator/deploy/k8s/cluster_info.py
+34 -62
View File
@@ -13,33 +13,31 @@
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http:#www.gnu.org/licenses/>.
import os
import base64
import os
from typing import Any
from kubernetes import client
from typing import Any, List, Optional, Set
from stack_orchestrator.opts import opts
from stack_orchestrator.util import env_var_map_from_file
from stack_orchestrator.deploy.deploy_types import DeployEnvVars
from stack_orchestrator.deploy.deploy_util import (
images_for_deployment,
parsed_pod_files_map_from_file_names,
)
from stack_orchestrator.deploy.images import remote_tag_for_image_unique
from stack_orchestrator.deploy.k8s.helpers import (
envs_from_compose_file,
envs_from_environment_variables_map,
get_kind_pv_bind_mount_path,
merge_envs,
named_volumes_from_pod_files,
translate_sidecar_service_names,
volume_mounts_for_service,
volumes_for_pod_files,
)
from stack_orchestrator.deploy.k8s.helpers import get_kind_pv_bind_mount_path
from stack_orchestrator.deploy.k8s.helpers import (
envs_from_environment_variables_map,
envs_from_compose_file,
merge_envs,
translate_sidecar_service_names,
)
from stack_orchestrator.deploy.deploy_util import (
parsed_pod_files_map_from_file_names,
images_for_deployment,
)
from stack_orchestrator.deploy.deploy_types import DeployEnvVars
from stack_orchestrator.deploy.spec import Spec, Resources, ResourceLimits
from stack_orchestrator.deploy.images import remote_tag_for_image_unique
from stack_orchestrator.deploy.spec import ResourceLimits, Resources, Spec
from stack_orchestrator.opts import opts
from stack_orchestrator.util import env_var_map_from_file
DEFAULT_VOLUME_RESOURCES = Resources({"reservations": {"storage": "2Gi"}})
@@ -52,7 +50,7 @@ DEFAULT_CONTAINER_RESOURCES = Resources(
def to_k8s_resource_requirements(resources: Resources) -> client.V1ResourceRequirements:
def to_dict(limits: Optional[ResourceLimits]):
def to_dict(limits: ResourceLimits | None):
if not limits:
return None
@@ -72,7 +70,7 @@ def to_k8s_resource_requirements(resources: Resources) -> client.V1ResourceRequi
class ClusterInfo:
parsed_pod_yaml_map: Any
image_set: Set[str] = set()
image_set: set[str] = set()
app_name: str
environment_variables: DeployEnvVars
spec: Spec
@@ -80,14 +78,12 @@ class ClusterInfo:
def __init__(self) -> None:
pass
def int(self, pod_files: List[str], compose_env_file, deployment_name, spec: Spec):
def int(self, pod_files: list[str], compose_env_file, deployment_name, spec: Spec):
self.parsed_pod_yaml_map = parsed_pod_files_map_from_file_names(pod_files)
# Find the set of images in the pods
self.image_set = images_for_deployment(pod_files)
# Filter out None values from env file
env_vars = {
k: v for k, v in env_var_map_from_file(compose_env_file).items() if v
}
env_vars = {k: v for k, v in env_var_map_from_file(compose_env_file).items() if v}
self.environment_variables = DeployEnvVars(env_vars)
self.app_name = deployment_name
self.spec = spec
@@ -124,8 +120,7 @@ class ClusterInfo:
service = client.V1Service(
metadata=client.V1ObjectMeta(
name=(
f"{self.app_name}-nodeport-"
f"{pod_port}-{protocol.lower()}"
f"{self.app_name}-nodeport-" f"{pod_port}-{protocol.lower()}"
),
labels={"app": self.app_name},
),
@@ -145,9 +140,7 @@ class ClusterInfo:
nodeports.append(service)
return nodeports
def get_ingress(
self, use_tls=False, certificate=None, cluster_issuer="letsencrypt-prod"
):
def get_ingress(self, use_tls=False, certificate=None, cluster_issuer="letsencrypt-prod"):
# No ingress for a deployment that has no http-proxy defined, for now
http_proxy_info_list = self.spec.get_http_proxy()
ingress = None
@@ -162,9 +155,7 @@ class ClusterInfo:
tls = (
[
client.V1IngressTLS(
hosts=certificate["spec"]["dnsNames"]
if certificate
else [host_name],
hosts=certificate["spec"]["dnsNames"] if certificate else [host_name],
secret_name=certificate["spec"]["secretName"]
if certificate
else f"{self.app_name}-tls",
@@ -237,8 +228,7 @@ class ClusterInfo:
return None
service_ports = [
client.V1ServicePort(port=p, target_port=p, name=f"port-{p}")
for p in sorted(ports_set)
client.V1ServicePort(port=p, target_port=p, name=f"port-{p}") for p in sorted(ports_set)
]
service = client.V1Service(
@@ -290,9 +280,7 @@ class ClusterInfo:
volume_name=k8s_volume_name,
)
pvc = client.V1PersistentVolumeClaim(
metadata=client.V1ObjectMeta(
name=f"{self.app_name}-{volume_name}", labels=labels
),
metadata=client.V1ObjectMeta(name=f"{self.app_name}-{volume_name}", labels=labels),
spec=spec,
)
result.append(pvc)
@@ -309,9 +297,7 @@ class ClusterInfo:
continue
if not cfg_map_path.startswith("/") and self.spec.file_path is not None:
cfg_map_path = os.path.join(
os.path.dirname(str(self.spec.file_path)), cfg_map_path
)
cfg_map_path = os.path.join(os.path.dirname(str(self.spec.file_path)), cfg_map_path)
# Read in all the files at a single-level of the directory.
# This mimics the behavior of
@@ -320,9 +306,7 @@ class ClusterInfo:
for f in os.listdir(cfg_map_path):
full_path = os.path.join(cfg_map_path, f)
if os.path.isfile(full_path):
data[f] = base64.b64encode(open(full_path, "rb").read()).decode(
"ASCII"
)
data[f] = base64.b64encode(open(full_path, "rb").read()).decode("ASCII")
spec = client.V1ConfigMap(
metadata=client.V1ObjectMeta(
@@ -425,7 +409,7 @@ class ClusterInfo:
return global_resources
# TODO: put things like image pull policy into an object-scope struct
def get_deployment(self, image_pull_policy: Optional[str] = None):
def get_deployment(self, image_pull_policy: str | None = None):
containers = []
services = {}
global_resources = self.spec.get_container_resources()
@@ -453,9 +437,7 @@ class ClusterInfo:
port_str = port_str.split(":")[-1]
port = int(port_str)
container_ports.append(
client.V1ContainerPort(
container_port=port, protocol=protocol
)
client.V1ContainerPort(container_port=port, protocol=protocol)
)
if opts.o.debug:
print(f"image: {image}")
@@ -473,9 +455,7 @@ class ClusterInfo:
# Translate docker-compose service names to localhost for sidecars
# All services in the same pod share the network namespace
sibling_services = [s for s in services.keys() if s != service_name]
merged_envs = translate_sidecar_service_names(
merged_envs, sibling_services
)
merged_envs = translate_sidecar_service_names(merged_envs, sibling_services)
envs = envs_from_environment_variables_map(merged_envs)
if opts.o.debug:
print(f"Merged envs: {envs}")
@@ -488,18 +468,14 @@ class ClusterInfo:
if self.spec.get_image_registry() is not None
else image
)
volume_mounts = volume_mounts_for_service(
self.parsed_pod_yaml_map, service_name
)
volume_mounts = volume_mounts_for_service(self.parsed_pod_yaml_map, service_name)
# Handle command/entrypoint from compose file
# In docker-compose: entrypoint -> k8s command, command -> k8s args
container_command = None
container_args = None
if "entrypoint" in service_info:
entrypoint = service_info["entrypoint"]
container_command = (
entrypoint if isinstance(entrypoint, list) else [entrypoint]
)
container_command = entrypoint if isinstance(entrypoint, list) else [entrypoint]
if "command" in service_info:
cmd = service_info["command"]
container_args = cmd if isinstance(cmd, list) else cmd.split()
@@ -528,18 +504,14 @@ class ClusterInfo:
volume_mounts=volume_mounts,
security_context=client.V1SecurityContext(
privileged=self.spec.get_privileged(),
capabilities=client.V1Capabilities(
add=self.spec.get_capabilities()
)
capabilities=client.V1Capabilities(add=self.spec.get_capabilities())
if self.spec.get_capabilities()
else None,
),
resources=to_k8s_resource_requirements(container_resources),
)
containers.append(container)
volumes = volumes_for_pod_files(
self.parsed_pod_yaml_map, self.spec, self.app_name
)
volumes = volumes_for_pod_files(self.parsed_pod_yaml_map, self.spec, self.app_name)
registry_config = self.spec.get_image_registry_config()
if registry_config:
secret_name = f"{self.app_name}-registry"
stack_orchestrator/deploy/k8s/deploy_k8s.py
+33 -62
View File
@@ -14,42 +14,36 @@
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, cast
from kubernetes import client, config
from kubernetes.client.exceptions import ApiException
from typing import Any, Dict, List, Optional, cast
from stack_orchestrator import constants
from stack_orchestrator.deploy.deployer import Deployer, DeployerConfigGenerator
from stack_orchestrator.deploy.deployment_context import DeploymentContext
from stack_orchestrator.deploy.k8s.cluster_info import ClusterInfo
from stack_orchestrator.deploy.k8s.helpers import (
containers_in_pod,
create_cluster,
destroy_cluster,
load_images_into_kind,
)
from stack_orchestrator.deploy.k8s.helpers import (
install_ingress_for_kind,
wait_for_ingress_in_kind,
is_ingress_running,
)
from stack_orchestrator.deploy.k8s.helpers import (
pods_in_deployment,
containers_in_pod,
log_stream_from_string,
)
from stack_orchestrator.deploy.k8s.helpers import (
generate_kind_config,
generate_high_memlock_spec_json,
generate_kind_config,
install_ingress_for_kind,
is_ingress_running,
load_images_into_kind,
log_stream_from_string,
pods_in_deployment,
wait_for_ingress_in_kind,
)
from stack_orchestrator.deploy.k8s.cluster_info import ClusterInfo
from stack_orchestrator.opts import opts
from stack_orchestrator.deploy.deployment_context import DeploymentContext
from stack_orchestrator.util import error_exit
class AttrDict(dict):
def __init__(self, *args, **kwargs):
super(AttrDict, self).__init__(*args, **kwargs)
super().__init__(*args, **kwargs)
self.__dict__ = self
@@ -144,9 +138,7 @@ class K8sDeployer(Deployer):
else:
# Get the config file and pass to load_kube_config()
config.load_kube_config(
config_file=self.deployment_dir.joinpath(
constants.kube_config_filename
).as_posix()
config_file=self.deployment_dir.joinpath(constants.kube_config_filename).as_posix()
)
self.core_api = client.CoreV1Api()
self.networking_api = client.NetworkingV1Api()
@@ -213,10 +205,7 @@ class K8sDeployer(Deployer):
)
if opts.o.debug:
print(
f"Namespace {self.k8s_namespace} is terminating, "
f"waiting for deletion..."
)
print(f"Namespace {self.k8s_namespace} is terminating, " f"waiting for deletion...")
time.sleep(2)
def _delete_namespace(self):
@@ -276,9 +265,7 @@ class K8sDeployer(Deployer):
name=deployment.metadata.name,
namespace=self.k8s_namespace,
)
deployment.metadata.resource_version = (
existing.metadata.resource_version
)
deployment.metadata.resource_version = existing.metadata.resource_version
resp = cast(
client.V1Deployment,
self.apps_api.replace_namespaced_deployment(
@@ -391,9 +378,7 @@ class K8sDeployer(Deployer):
print(f"Sending this pv: {pv}")
if not opts.o.dry_run:
try:
pv_resp = self.core_api.read_persistent_volume(
name=pv.metadata.name
)
pv_resp = self.core_api.read_persistent_volume(name=pv.metadata.name)
if pv_resp:
if opts.o.debug:
print("PVs already present:")
@@ -500,9 +485,9 @@ class K8sDeployer(Deployer):
if before < now < after:
# Check the status is Ready
for condition in status.get("conditions", []):
if "True" == condition.get(
"status"
) and "Ready" == condition.get("type"):
if "True" == condition.get("status") and "Ready" == condition.get(
"type"
):
return cert
return None
@@ -519,15 +504,11 @@ class K8sDeployer(Deployer):
self.skip_cluster_management = skip_cluster_management
if not opts.o.dry_run:
if self.is_kind() and not self.skip_cluster_management:
kind_config = str(
self.deployment_dir.joinpath(constants.kind_config_filename)
)
kind_config = str(self.deployment_dir.joinpath(constants.kind_config_filename))
actual_cluster = create_cluster(self.kind_cluster_name, kind_config)
if actual_cluster != self.kind_cluster_name:
self.kind_cluster_name = actual_cluster
local_containers = self.deployment_context.stack.obj.get(
"containers", []
)
local_containers = self.deployment_context.stack.obj.get("containers", [])
if local_containers:
local_images = {
img
@@ -579,9 +560,7 @@ class K8sDeployer(Deployer):
if opts.o.debug and certificate:
print(f"Using existing certificate: {certificate}")
ingress = self.cluster_info.get_ingress(
use_tls=use_tls, certificate=certificate
)
ingress = self.cluster_info.get_ingress(use_tls=use_tls, certificate=certificate)
if ingress:
if opts.o.debug:
print(f"Sending this ingress: {ingress}")
@@ -590,7 +569,7 @@ class K8sDeployer(Deployer):
elif opts.o.debug:
print("No ingress configured")
nodeports: List[client.V1Service] = self.cluster_info.get_nodeports()
nodeports: list[client.V1Service] = self.cluster_info.get_nodeports()
for nodeport in nodeports:
if opts.o.debug:
print(f"Sending this nodeport: {nodeport}")
@@ -670,7 +649,7 @@ class K8sDeployer(Deployer):
return
cert = cast(
Dict[str, Any],
dict[str, Any],
self.custom_obj_api.get_namespaced_custom_object(
group="cert-manager.io",
version="v1",
@@ -686,7 +665,7 @@ class K8sDeployer(Deployer):
if lb_ingress:
ip = lb_ingress[0].ip or "?"
cert_status = cert.get("status", {})
tls = "notBefore: %s; notAfter: %s; names: %s" % (
tls = "notBefore: {}; notAfter: {}; names: {}".format(
cert_status.get("notBefore", "?"),
cert_status.get("notAfter", "?"),
ingress.spec.tls[0].hosts,
@@ -727,9 +706,7 @@ class K8sDeployer(Deployer):
if c.ports:
for prt in c.ports:
ports[str(prt.container_port)] = [
AttrDict(
{"HostIp": pod_ip, "HostPort": prt.container_port}
)
AttrDict({"HostIp": pod_ip, "HostPort": prt.container_port})
]
ret.append(
@@ -791,9 +768,7 @@ class K8sDeployer(Deployer):
deployment = cast(
client.V1Deployment,
self.apps_api.read_namespaced_deployment(
name=ref_name, namespace=self.k8s_namespace
),
self.apps_api.read_namespaced_deployment(name=ref_name, namespace=self.k8s_namespace),
)
if not deployment.spec or not deployment.spec.template:
return
@@ -832,14 +807,14 @@ class K8sDeployer(Deployer):
user=None,
volumes=None,
entrypoint=None,
env={},
ports=[],
env=None,
ports=None,
detach=False,
):
# We need to figure out how to do this -- check why we're being called first
pass
def run_job(self, job_name: str, helm_release: Optional[str] = None):
def run_job(self, job_name: str, helm_release: str | None = None):
if not opts.o.dry_run:
from stack_orchestrator.deploy.k8s.helm.job_runner import run_helm_job
@@ -881,13 +856,9 @@ class K8sDeployerConfigGenerator(DeployerConfigGenerator):
# Must be done before generate_kind_config() which references it.
if self.deployment_context.spec.get_unlimited_memlock():
spec_content = generate_high_memlock_spec_json()
spec_file = deployment_dir.joinpath(
constants.high_memlock_spec_filename
)
spec_file = deployment_dir.joinpath(constants.high_memlock_spec_filename)
if opts.o.debug:
print(
f"Creating high-memlock spec for unlimited memlock: {spec_file}"
)
print(f"Creating high-memlock spec for unlimited memlock: {spec_file}")
with open(spec_file, "w") as output_file:
output_file.write(spec_content)
stack_orchestrator/deploy/k8s/helm/chart_generator.py
+16 -23
View File
@@ -16,21 +16,21 @@
from pathlib import Path
from stack_orchestrator import constants
from stack_orchestrator.opts import opts
from stack_orchestrator.util import (
get_parsed_stack_config,
get_pod_list,
get_pod_file_path,
get_job_list,
get_job_file_path,
error_exit,
)
from stack_orchestrator.deploy.k8s.helm.kompose_wrapper import (
check_kompose_available,
get_kompose_version,
convert_to_helm_chart,
get_kompose_version,
)
from stack_orchestrator.opts import opts
from stack_orchestrator.util import (
error_exit,
get_job_file_path,
get_job_list,
get_parsed_stack_config,
get_pod_file_path,
get_pod_list,
get_yaml,
)
from stack_orchestrator.util import get_yaml
def _wrap_job_templates_with_conditionals(chart_dir: Path, jobs: list) -> None:
@@ -88,7 +88,7 @@ def _post_process_chart(chart_dir: Path, chart_name: str, jobs: list) -> None:
# Fix Chart.yaml
chart_yaml_path = chart_dir / "Chart.yaml"
if chart_yaml_path.exists():
chart_yaml = yaml.load(open(chart_yaml_path, "r"))
chart_yaml = yaml.load(open(chart_yaml_path))
# Fix name
chart_yaml["name"] = chart_name
@@ -108,9 +108,7 @@ def _post_process_chart(chart_dir: Path, chart_name: str, jobs: list) -> None:
_wrap_job_templates_with_conditionals(chart_dir, jobs)
def generate_helm_chart(
stack_path: str, spec_file: str, deployment_dir_path: Path
) -> None:
def generate_helm_chart(stack_path: str, spec_file: str, deployment_dir_path: Path) -> None:
"""
Generate a self-sufficient Helm chart from stack compose files using Kompose.
@@ -152,7 +150,7 @@ def generate_helm_chart(
error_exit(f"Deployment file not found: {deployment_file}")
yaml = get_yaml()
deployment_config = yaml.load(open(deployment_file, "r"))
deployment_config = yaml.load(open(deployment_file))
cluster_id = deployment_config.get(constants.cluster_id_key)
if not cluster_id:
error_exit(f"cluster-id not found in {deployment_file}")
@@ -219,10 +217,7 @@ def generate_helm_chart(
# 5. Create chart directory and invoke Kompose
chart_dir = deployment_dir_path / "chart"
print(
f"Converting {len(compose_files)} compose file(s) to Helm chart "
"using Kompose..."
)
print(f"Converting {len(compose_files)} compose file(s) to Helm chart " "using Kompose...")
try:
output = convert_to_helm_chart(
@@ -304,9 +299,7 @@ Edit the generated template files in `templates/` to customize:
# Count generated files
template_files = (
list((chart_dir / "templates").glob("*.yaml"))
if (chart_dir / "templates").exists()
else []
list((chart_dir / "templates").glob("*.yaml")) if (chart_dir / "templates").exists() else []
)
print(f" Files: {len(template_files)} template(s) generated")
stack_orchestrator/deploy/k8s/helm/job_runner.py
+11 -16
View File
@@ -13,12 +13,12 @@
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http:#www.gnu.org/licenses/>.
import json
import os
import subprocess
import tempfile
import os
import json
from pathlib import Path
from typing import Optional
from stack_orchestrator.util import get_yaml
@@ -40,18 +40,19 @@ def get_release_name_from_chart(chart_dir: Path) -> str:
raise Exception(f"Chart.yaml not found: {chart_yaml_path}")
yaml = get_yaml()
chart_yaml = yaml.load(open(chart_yaml_path, "r"))
chart_yaml = yaml.load(open(chart_yaml_path))
if "name" not in chart_yaml:
raise Exception(f"Chart name not found in {chart_yaml_path}")
return chart_yaml["name"]
name: str = chart_yaml["name"]
return name
def run_helm_job(
chart_dir: Path,
job_name: str,
release: Optional[str] = None,
release: str | None = None,
namespace: str = "default",
timeout: int = 600,
verbose: bool = False,
@@ -94,9 +95,7 @@ def run_helm_job(
print(f"Running job '{job_name}' from helm chart: {chart_dir}")
# Use helm template to render the job manifest
with tempfile.NamedTemporaryFile(
mode="w", suffix=".yaml", delete=False
) as tmp_file:
with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as tmp_file:
try:
# Render job template with job enabled
# Use --set-json to properly handle job names with dashes
@@ -116,9 +115,7 @@ def run_helm_job(
if verbose:
print(f"Running: {' '.join(helm_cmd)}")
result = subprocess.run(
helm_cmd, check=True, capture_output=True, text=True
)
result = subprocess.run(helm_cmd, check=True, capture_output=True, text=True)
tmp_file.write(result.stdout)
tmp_file.flush()
@@ -139,9 +136,7 @@ def run_helm_job(
"-n",
namespace,
]
subprocess.run(
kubectl_apply_cmd, check=True, capture_output=True, text=True
)
subprocess.run(kubectl_apply_cmd, check=True, capture_output=True, text=True)
if verbose:
print(f"Job {actual_job_name} created, waiting for completion...")
@@ -164,7 +159,7 @@ def run_helm_job(
except subprocess.CalledProcessError as e:
error_msg = e.stderr if e.stderr else str(e)
raise Exception(f"Job failed: {error_msg}")
raise Exception(f"Job failed: {error_msg}") from e
finally:
# Clean up temp file
if os.path.exists(tmp_file.name):
stack_orchestrator/deploy/k8s/helm/kompose_wrapper.py
+5 -11
View File
@@ -13,10 +13,9 @@
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http:#www.gnu.org/licenses/>.
import subprocess
import shutil
import subprocess
from pathlib import Path
from typing import List, Optional
def check_kompose_available() -> bool:
@@ -37,9 +36,7 @@ def get_kompose_version() -> str:
if not check_kompose_available():
raise Exception("kompose not found in PATH")
result = subprocess.run(
["kompose", "version"], capture_output=True, text=True, timeout=10
)
result = subprocess.run(["kompose", "version"], capture_output=True, text=True, timeout=10)
if result.returncode != 0:
raise Exception(f"Failed to get kompose version: {result.stderr}")
@@ -53,7 +50,7 @@ def get_kompose_version() -> str:
def convert_to_helm_chart(
compose_files: List[Path], output_dir: Path, chart_name: Optional[str] = None
compose_files: list[Path], output_dir: Path, chart_name: str | None = None
) -> str:
"""
Invoke kompose to convert Docker Compose files to a Helm chart.
@@ -71,8 +68,7 @@ def convert_to_helm_chart(
"""
if not check_kompose_available():
raise Exception(
"kompose not found in PATH. "
"Install from: https://kompose.io/installation/"
"kompose not found in PATH. " "Install from: https://kompose.io/installation/"
)
# Ensure output directory exists
@@ -95,9 +91,7 @@ def convert_to_helm_chart(
if result.returncode != 0:
raise Exception(
f"Kompose conversion failed:\n"
f"Command: {' '.join(cmd)}\n"
f"Error: {result.stderr}"
f"Kompose conversion failed:\n" f"Command: {' '.join(cmd)}\n" f"Error: {result.stderr}"
)
return result.stdout
stack_orchestrator/deploy/k8s/helpers.py
+53 -90
View File
@@ -13,20 +13,22 @@
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http:#www.gnu.org/licenses/>.
import os
import re
import subprocess
from collections.abc import Mapping
from pathlib import Path
from typing import cast
import yaml
from kubernetes import client, utils, watch
from kubernetes.client.exceptions import ApiException
import os
from pathlib import Path
import subprocess
import re
from typing import Set, Mapping, List, Optional, cast
import yaml
from stack_orchestrator.util import get_k8s_dir, error_exit
from stack_orchestrator.opts import opts
from stack_orchestrator import constants
from stack_orchestrator.deploy.deploy_util import parsed_pod_files_map_from_file_names
from stack_orchestrator.deploy.deployer import DeployerException
from stack_orchestrator import constants
from stack_orchestrator.opts import opts
from stack_orchestrator.util import error_exit, get_k8s_dir
def is_host_path_mount(volume_name: str) -> bool:
@@ -77,9 +79,7 @@ def get_kind_cluster():
Uses `kind get clusters` to find existing clusters.
Returns the cluster name or None if no cluster exists.
"""
result = subprocess.run(
"kind get clusters", shell=True, capture_output=True, text=True
)
result = subprocess.run("kind get clusters", shell=True, capture_output=True, text=True)
if result.returncode != 0:
return None
@@ -98,12 +98,12 @@ def _run_command(command: str):
return result
def _get_etcd_host_path_from_kind_config(config_file: str) -> Optional[str]:
def _get_etcd_host_path_from_kind_config(config_file: str) -> str | None:
"""Extract etcd host path from kind config extraMounts."""
import yaml
try:
with open(config_file, "r") as f:
with open(config_file) as f:
config = yaml.safe_load(f)
except Exception:
return None
@@ -113,7 +113,8 @@ def _get_etcd_host_path_from_kind_config(config_file: str) -> Optional[str]:
extra_mounts = node.get("extraMounts", [])
for mount in extra_mounts:
if mount.get("containerPath") == "/var/lib/etcd":
return mount.get("hostPath")
host_path: str | None = mount.get("hostPath")
return host_path
return None
@@ -133,8 +134,7 @@ def _clean_etcd_keeping_certs(etcd_path: str) -> bool:
db_path = Path(etcd_path) / "member" / "snap" / "db"
# Check existence using docker since etcd dir is root-owned
check_cmd = (
f"docker run --rm -v {etcd_path}:/etcd:ro alpine:3.19 "
"test -f /etcd/member/snap/db"
f"docker run --rm -v {etcd_path}:/etcd:ro alpine:3.19 " "test -f /etcd/member/snap/db"
)
check_result = subprocess.run(check_cmd, shell=True, capture_output=True)
if check_result.returncode != 0:
@@ -337,7 +337,7 @@ def is_ingress_running() -> bool:
def wait_for_ingress_in_kind():
core_v1 = client.CoreV1Api()
for i in range(20):
for _i in range(20):
warned_waiting = False
w = watch.Watch()
for event in w.stream(
@@ -364,9 +364,7 @@ def wait_for_ingress_in_kind():
def install_ingress_for_kind(acme_email: str = ""):
api_client = client.ApiClient()
ingress_install = os.path.abspath(
get_k8s_dir().joinpath(
"components", "ingress", "ingress-caddy-kind-deploy.yaml"
)
get_k8s_dir().joinpath("components", "ingress", "ingress-caddy-kind-deploy.yaml")
)
if opts.o.debug:
print("Installing Caddy ingress controller in kind cluster")
@@ -400,11 +398,9 @@ def install_ingress_for_kind(acme_email: str = ""):
)
def load_images_into_kind(kind_cluster_name: str, image_set: Set[str]):
def load_images_into_kind(kind_cluster_name: str, image_set: set[str]):
for image in image_set:
result = _run_command(
f"kind load docker-image {image} --name {kind_cluster_name}"
)
result = _run_command(f"kind load docker-image {image} --name {kind_cluster_name}")
if result.returncode != 0:
raise DeployerException(f"kind load docker-image failed: {result}")
@@ -422,11 +418,9 @@ def pods_in_deployment(core_api: client.CoreV1Api, deployment_name: str):
return pods
def containers_in_pod(core_api: client.CoreV1Api, pod_name: str) -> List[str]:
containers: List[str] = []
pod_response = cast(
client.V1Pod, core_api.read_namespaced_pod(pod_name, namespace="default")
)
def containers_in_pod(core_api: client.CoreV1Api, pod_name: str) -> list[str]:
containers: list[str] = []
pod_response = cast(client.V1Pod, core_api.read_namespaced_pod(pod_name, namespace="default"))
if opts.o.debug:
print(f"pod_response: {pod_response}")
if not pod_response.spec or not pod_response.spec.containers:
@@ -449,7 +443,7 @@ def named_volumes_from_pod_files(parsed_pod_files):
parsed_pod_file = parsed_pod_files[pod]
if "volumes" in parsed_pod_file:
volumes = parsed_pod_file["volumes"]
for volume, value in volumes.items():
for volume, _value in volumes.items():
# Volume definition looks like:
# 'laconicd-data': None
named_volumes.append(volume)
@@ -481,14 +475,10 @@ def volume_mounts_for_service(parsed_pod_files, service):
mount_split = mount_string.split(":")
volume_name = mount_split[0]
mount_path = mount_split[1]
mount_options = (
mount_split[2] if len(mount_split) == 3 else None
)
mount_options = mount_split[2] if len(mount_split) == 3 else None
# For host path mounts, use sanitized name
if is_host_path_mount(volume_name):
k8s_volume_name = sanitize_host_path_to_volume_name(
volume_name
)
k8s_volume_name = sanitize_host_path_to_volume_name(volume_name)
else:
k8s_volume_name = volume_name
if opts.o.debug:
@@ -527,9 +517,7 @@ def volumes_for_pod_files(parsed_pod_files, spec, app_name):
claim = client.V1PersistentVolumeClaimVolumeSource(
claim_name=f"{app_name}-{volume_name}"
)
volume = client.V1Volume(
name=volume_name, persistent_volume_claim=claim
)
volume = client.V1Volume(name=volume_name, persistent_volume_claim=claim)
result.append(volume)
# Handle host path mounts from service volumes
@@ -542,15 +530,11 @@ def volumes_for_pod_files(parsed_pod_files, spec, app_name):
mount_split = mount_string.split(":")
volume_source = mount_split[0]
if is_host_path_mount(volume_source):
sanitized_name = sanitize_host_path_to_volume_name(
volume_source
)
sanitized_name = sanitize_host_path_to_volume_name(volume_source)
if sanitized_name not in seen_host_path_volumes:
seen_host_path_volumes.add(sanitized_name)
# Create hostPath volume for mount inside kind node
kind_mount_path = get_kind_host_path_mount_path(
sanitized_name
)
kind_mount_path = get_kind_host_path_mount_path(sanitized_name)
host_path_source = client.V1HostPathVolumeSource(
path=kind_mount_path, type="FileOrCreate"
)
@@ -585,18 +569,14 @@ def _generate_kind_mounts(parsed_pod_files, deployment_dir, deployment_context):
deployment_id = deployment_context.id
backup_subdir = f"cluster-backups/{deployment_id}"
etcd_host_path = _make_absolute_host_path(
Path(f"./data/{backup_subdir}/etcd"), deployment_dir
)
etcd_host_path = _make_absolute_host_path(Path(f"./data/{backup_subdir}/etcd"), deployment_dir)
volume_definitions.append(
f" - hostPath: {etcd_host_path}\n"
f" containerPath: /var/lib/etcd\n"
f" propagation: HostToContainer\n"
)
pki_host_path = _make_absolute_host_path(
Path(f"./data/{backup_subdir}/pki"), deployment_dir
)
pki_host_path = _make_absolute_host_path(Path(f"./data/{backup_subdir}/pki"), deployment_dir)
volume_definitions.append(
f" - hostPath: {pki_host_path}\n"
f" containerPath: /etc/kubernetes/pki\n"
@@ -626,18 +606,12 @@ def _generate_kind_mounts(parsed_pod_files, deployment_dir, deployment_context):
if is_host_path_mount(volume_name):
# Host path mount - add extraMount for kind
sanitized_name = sanitize_host_path_to_volume_name(
volume_name
)
sanitized_name = sanitize_host_path_to_volume_name(volume_name)
if sanitized_name not in seen_host_path_mounts:
seen_host_path_mounts.add(sanitized_name)
# Resolve path relative to compose directory
host_path = resolve_host_path_for_kind(
volume_name, deployment_dir
)
container_path = get_kind_host_path_mount_path(
sanitized_name
)
host_path = resolve_host_path_for_kind(volume_name, deployment_dir)
container_path = get_kind_host_path_mount_path(sanitized_name)
volume_definitions.append(
f" - hostPath: {host_path}\n"
f" containerPath: {container_path}\n"
@@ -651,10 +625,7 @@ def _generate_kind_mounts(parsed_pod_files, deployment_dir, deployment_context):
print(f"volume_name: {volume_name}")
print(f"map: {volume_host_path_map}")
print(f"mount path: {mount_path}")
if (
volume_name
not in deployment_context.spec.get_configmaps()
):
if volume_name not in deployment_context.spec.get_configmaps():
if (
volume_name in volume_host_path_map
and volume_host_path_map[volume_name]
@@ -663,9 +634,7 @@ def _generate_kind_mounts(parsed_pod_files, deployment_dir, deployment_context):
volume_host_path_map[volume_name],
deployment_dir,
)
container_path = get_kind_pv_bind_mount_path(
volume_name
)
container_path = get_kind_pv_bind_mount_path(volume_name)
volume_definitions.append(
f" - hostPath: {host_path}\n"
f" containerPath: {container_path}\n"
@@ -693,8 +662,7 @@ def _generate_kind_port_mappings_from_services(parsed_pod_files):
# TODO handle the complex cases
# Looks like: 80 or something more complicated
port_definitions.append(
f" - containerPort: {port_string}\n"
f" hostPort: {port_string}\n"
f" - containerPort: {port_string}\n" f" hostPort: {port_string}\n"
)
return (
""
@@ -707,9 +675,7 @@ def _generate_kind_port_mappings(parsed_pod_files):
port_definitions = []
# Map port 80 and 443 for the Caddy ingress controller (HTTPS support)
for port_string in ["80", "443"]:
port_definitions.append(
f" - containerPort: {port_string}\n hostPort: {port_string}\n"
)
port_definitions.append(f" - containerPort: {port_string}\n hostPort: {port_string}\n")
return (
""
if len(port_definitions) == 0
@@ -903,9 +869,7 @@ def generate_cri_base_json():
return generate_high_memlock_spec_json()
def _generate_containerd_config_patches(
deployment_dir: Path, has_high_memlock: bool
) -> str:
def _generate_containerd_config_patches(deployment_dir: Path, has_high_memlock: bool) -> str:
"""Generate containerdConfigPatches YAML for custom runtime handlers.
This configures containerd to have a runtime handler named 'high-memlock'
@@ -932,9 +896,7 @@ def merge_envs(a: Mapping[str, str], b: Mapping[str, str]) -> Mapping[str, str]:
return result
def _expand_shell_vars(
raw_val: str, env_map: Optional[Mapping[str, str]] = None
) -> str:
def _expand_shell_vars(raw_val: str, env_map: Mapping[str, str] | None = None) -> str:
# Expand docker-compose style variable substitution:
# ${VAR} - use VAR value or empty string
# ${VAR:-default} - use VAR value or default if unset/empty
@@ -959,7 +921,7 @@ def _expand_shell_vars(
def envs_from_compose_file(
compose_file_envs: Mapping[str, str], env_map: Optional[Mapping[str, str]] = None
compose_file_envs: Mapping[str, str], env_map: Mapping[str, str] | None = None
) -> Mapping[str, str]:
result = {}
for env_var, env_val in compose_file_envs.items():
@@ -969,7 +931,7 @@ def envs_from_compose_file(
def translate_sidecar_service_names(
envs: Mapping[str, str], sibling_service_names: List[str]
envs: Mapping[str, str], sibling_service_names: list[str]
) -> Mapping[str, str]:
"""Translate docker-compose service names to localhost for sidecar containers.
@@ -996,7 +958,12 @@ def translate_sidecar_service_names(
# Handle URLs like: postgres://user:pass@db:5432/dbname
# and simple refs like: db:5432 or just db
pattern = rf"\b{re.escape(service_name)}(:\d+)?\b"
new_val = re.sub(pattern, lambda m: f'localhost{m.group(1) or ""}', new_val)
def _replace_with_localhost(m: re.Match[str]) -> str:
port: str = m.group(1) or ""
return "localhost" + port
new_val = re.sub(pattern, _replace_with_localhost, new_val)
result[env_var] = new_val
@@ -1004,8 +971,8 @@ def translate_sidecar_service_names(
def envs_from_environment_variables_map(
map: Mapping[str, str]
) -> List[client.V1EnvVar]:
map: Mapping[str, str],
) -> list[client.V1EnvVar]:
result = []
for env_var, env_val in map.items():
result.append(client.V1EnvVar(env_var, env_val))
@@ -1036,17 +1003,13 @@ def generate_kind_config(deployment_dir: Path, deployment_context):
pod_files = [p for p in compose_file_dir.iterdir() if p.is_file()]
parsed_pod_files_map = parsed_pod_files_map_from_file_names(pod_files)
port_mappings_yml = _generate_kind_port_mappings(parsed_pod_files_map)
mounts_yml = _generate_kind_mounts(
parsed_pod_files_map, deployment_dir, deployment_context
)
mounts_yml = _generate_kind_mounts(parsed_pod_files_map, deployment_dir, deployment_context)
# Check if unlimited_memlock is enabled
unlimited_memlock = deployment_context.spec.get_unlimited_memlock()
# Generate containerdConfigPatches for RuntimeClass support
containerd_patches_yml = _generate_containerd_config_patches(
deployment_dir, unlimited_memlock
)
containerd_patches_yml = _generate_containerd_config_patches(deployment_dir, unlimited_memlock)
# Add high-memlock spec file mount if needed
if unlimited_memlock: