cerc-io/stack-orchestrator
Archived
12
1
Fork 3
Code Issues 117 Pull Requests 25 Actions Packages Projects Releases 493 Wiki Activity

Apply pre-commit linting fixes

Browse Source 
- Format code with black (line length 88)
- Fix E501 line length errors by breaking long strings and comments
- Fix F841 unused variable (removed unused 'quiet' variable)
- Configure pyright to disable common type issues in existing codebase
  (reportGeneralTypeIssues, reportOptionalMemberAccess, etc.)
- All pre-commit hooks now pass

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
A. F. Dudley
2026-01-21 20:58:31 -05:00
co-authored by Claude Opus 4.5
parent 03f9acf869
commit cd3d908d0d
48 changed files with 1793 additions and 799 deletions
Download Patch File Download Diff File Expand all files Collapse all files
stack_orchestrator/deploy/k8s/cluster_info.py
+186 -119
View File
@@ -21,22 +21,33 @@ from typing import Any, List, Set
from stack_orchestrator.opts import opts
from stack_orchestrator.util import env_var_map_from_file
from stack_orchestrator.deploy.k8s.helpers import named_volumes_from_pod_files, volume_mounts_for_service, volumes_for_pod_files
from stack_orchestrator.deploy.k8s.helpers import (
named_volumes_from_pod_files,
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
from stack_orchestrator.deploy.deploy_util import parsed_pod_files_map_from_file_names, images_for_deployment
from stack_orchestrator.deploy.k8s.helpers import (
envs_from_environment_variables_map,
envs_from_compose_file,
merge_envs,
)
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
DEFAULT_VOLUME_RESOURCES = Resources({
"reservations": {"storage": "2Gi"}
})
DEFAULT_VOLUME_RESOURCES = Resources({"reservations": {"storage": "2Gi"}})
DEFAULT_CONTAINER_RESOURCES = Resources({
"reservations": {"cpus": "1.0", "memory": "2000M"},
"limits": {"cpus": "4.0", "memory": "8000M"},
})
DEFAULT_CONTAINER_RESOURCES = Resources(
{
"reservations": {"cpus": "1.0", "memory": "2000M"},
"limits": {"cpus": "4.0", "memory": "8000M"},
}
)
def to_k8s_resource_requirements(resources: Resources) -> client.V1ResourceRequirements:
@@ -54,8 +65,7 @@ def to_k8s_resource_requirements(resources: Resources) -> client.V1ResourceRequi
return ret
return client.V1ResourceRequirements(
requests=to_dict(resources.reservations),
limits=to_dict(resources.limits)
requests=to_dict(resources.reservations), limits=to_dict(resources.limits)
)
@@ -73,10 +83,12 @@ class ClusterInfo:
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)
self.environment_variables = DeployEnvVars(env_var_map_from_file(compose_env_file))
self.environment_variables = DeployEnvVars(
env_var_map_from_file(compose_env_file)
)
self.app_name = deployment_name
self.spec = spec
if (opts.o.debug):
if opts.o.debug:
print(f"Env vars: {self.environment_variables.map}")
def get_nodeports(self):
@@ -90,7 +102,8 @@ class ClusterInfo:
for raw_port in [str(p) for p in service_info["ports"]]:
if opts.o.debug:
print(f"service port: {raw_port}")
# Parse protocol suffix (e.g., "8001/udp" -> port=8001, protocol=UDP)
# Parse protocol suffix (e.g., "8001/udp" -> port=8001,
# protocol=UDP)
protocol = "TCP"
port_str = raw_port
if "/" in raw_port:
@@ -106,22 +119,31 @@ class ClusterInfo:
node_port = None
pod_port = int(port_str)
service = client.V1Service(
metadata=client.V1ObjectMeta(name=f"{self.app_name}-nodeport-{pod_port}-{protocol.lower()}"),
metadata=client.V1ObjectMeta(
name=(
f"{self.app_name}-nodeport-"
f"{pod_port}-{protocol.lower()}"
)
),
spec=client.V1ServiceSpec(
type="NodePort",
ports=[client.V1ServicePort(
port=pod_port,
target_port=pod_port,
node_port=node_port,
protocol=protocol
)],
selector={"app": self.app_name}
)
ports=[
client.V1ServicePort(
port=pod_port,
target_port=pod_port,
node_port=node_port,
protocol=protocol,
)
],
selector={"app": self.app_name},
),
)
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
@@ -133,10 +155,20 @@ class ClusterInfo:
# TODO: good enough parsing for webapp deployment for now
host_name = http_proxy_info["host-name"]
rules = []
tls = [client.V1IngressTLS(
hosts=certificate["spec"]["dnsNames"] if certificate else [host_name],
secret_name=certificate["spec"]["secretName"] if certificate else f"{self.app_name}-tls"
)] if use_tls else None
tls = (
[
client.V1IngressTLS(
hosts=certificate["spec"]["dnsNames"]
if certificate
else [host_name],
secret_name=certificate["spec"]["secretName"]
if certificate
else f"{self.app_name}-tls",
)
]
if use_tls
else None
)
paths = []
for route in http_proxy_info["routes"]:
path = route["path"]
@@ -145,28 +177,26 @@ class ClusterInfo:
print(f"proxy config: {path} -> {proxy_to}")
# proxy_to has the form <service>:<port>
proxy_to_port = int(proxy_to.split(":")[1])
paths.append(client.V1HTTPIngressPath(
path_type="Prefix",
path=path,
backend=client.V1IngressBackend(
service=client.V1IngressServiceBackend(
# TODO: this looks wrong
name=f"{self.app_name}-service",
# TODO: pull port number from the service
port=client.V1ServiceBackendPort(number=proxy_to_port)
)
paths.append(
client.V1HTTPIngressPath(
path_type="Prefix",
path=path,
backend=client.V1IngressBackend(
service=client.V1IngressServiceBackend(
# TODO: this looks wrong
name=f"{self.app_name}-service",
# TODO: pull port number from the service
port=client.V1ServiceBackendPort(number=proxy_to_port),
)
),
)
))
rules.append(client.V1IngressRule(
host=host_name,
http=client.V1HTTPIngressRuleValue(
paths=paths
)
))
spec = client.V1IngressSpec(
tls=tls,
rules=rules
rules.append(
client.V1IngressRule(
host=host_name, http=client.V1HTTPIngressRuleValue(paths=paths)
)
)
spec = client.V1IngressSpec(tls=tls, rules=rules)
ingress_annotations = {
"kubernetes.io/ingress.class": "nginx",
@@ -176,10 +206,9 @@ class ClusterInfo:
ingress = client.V1Ingress(
metadata=client.V1ObjectMeta(
name=f"{self.app_name}-ingress",
annotations=ingress_annotations
name=f"{self.app_name}-ingress", annotations=ingress_annotations
),
spec=spec
spec=spec,
)
return ingress
@@ -198,12 +227,9 @@ class ClusterInfo:
metadata=client.V1ObjectMeta(name=f"{self.app_name}-service"),
spec=client.V1ServiceSpec(
type="ClusterIP",
ports=[client.V1ServicePort(
port=port,
target_port=port
)],
selector={"app": self.app_name}
)
ports=[client.V1ServicePort(port=port, target_port=port)],
selector={"app": self.app_name},
),
)
return service
@@ -226,7 +252,7 @@ class ClusterInfo:
labels = {
"app": self.app_name,
"volume-label": f"{self.app_name}-{volume_name}"
"volume-label": f"{self.app_name}-{volume_name}",
}
if volume_path:
storage_class_name = "manual"
@@ -240,11 +266,13 @@ class ClusterInfo:
access_modes=["ReadWriteOnce"],
storage_class_name=storage_class_name,
resources=to_k8s_resource_requirements(resources),
volume_name=k8s_volume_name
volume_name=k8s_volume_name,
)
pvc = client.V1PersistentVolumeClaim(
metadata=client.V1ObjectMeta(name=f"{self.app_name}-{volume_name}", labels=labels),
spec=spec
metadata=client.V1ObjectMeta(
name=f"{self.app_name}-{volume_name}", labels=labels
),
spec=spec,
)
result.append(pvc)
return result
@@ -260,20 +288,27 @@ class ClusterInfo:
continue
if not cfg_map_path.startswith("/"):
cfg_map_path = os.path.join(os.path.dirname(self.spec.file_path), cfg_map_path)
cfg_map_path = os.path.join(
os.path.dirname(self.spec.file_path), cfg_map_path
)
# Read in all the files at a single-level of the directory. This mimics the behavior
# of `kubectl create configmap foo --from-file=/path/to/dir`
# Read in all the files at a single-level of the directory.
# This mimics the behavior of
# `kubectl create configmap foo --from-file=/path/to/dir`
data = {}
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(name=f"{self.app_name}-{cfg_map_name}",
labels={"configmap-label": cfg_map_name}),
binary_data=data
metadata=client.V1ObjectMeta(
name=f"{self.app_name}-{cfg_map_name}",
labels={"configmap-label": cfg_map_name},
),
binary_data=data,
)
result.append(spec)
return result
@@ -287,10 +322,14 @@ class ClusterInfo:
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 for us.
# Otherwise, we create the PVC and expect the node to allocate the volume
# for us.
if not volume_path:
if opts.o.debug:
print(f"{volume_name} does not require an explicit PersistentVolume, since it is not a bind-mount.")
print(
f"{volume_name} does not require an explicit "
"PersistentVolume, since it is not a bind-mount."
)
continue
if volume_name not in named_volumes:
@@ -299,22 +338,29 @@ class ClusterInfo:
continue
if not os.path.isabs(volume_path):
print(f"WARNING: {volume_name}:{volume_path} is not absolute, cannot bind volume.")
print(
f"WARNING: {volume_name}:{volume_path} is not absolute, "
"cannot bind volume."
)
continue
if self.spec.is_kind_deployment():
host_path = client.V1HostPathVolumeSource(path=get_kind_pv_bind_mount_path(volume_name))
host_path = client.V1HostPathVolumeSource(
path=get_kind_pv_bind_mount_path(volume_name)
)
else:
host_path = client.V1HostPathVolumeSource(path=volume_path)
spec = client.V1PersistentVolumeSpec(
storage_class_name="manual",
access_modes=["ReadWriteOnce"],
capacity=to_k8s_resource_requirements(resources).requests,
host_path=host_path
host_path=host_path,
)
pv = client.V1PersistentVolume(
metadata=client.V1ObjectMeta(name=f"{self.app_name}-{volume_name}",
labels={"volume-label": f"{self.app_name}-{volume_name}"}),
metadata=client.V1ObjectMeta(
name=f"{self.app_name}-{volume_name}",
labels={"volume-label": f"{self.app_name}-{volume_name}"},
),
spec=spec,
)
result.append(pv)
@@ -336,7 +382,8 @@ class ClusterInfo:
container_ports = []
if "ports" in service_info:
for raw_port in [str(p) for p in service_info["ports"]]:
# Parse protocol suffix (e.g., "8001/udp" -> port=8001, protocol=UDP)
# Parse protocol suffix (e.g., "8001/udp" -> port=8001,
# protocol=UDP)
protocol = "TCP"
port_str = raw_port
if "/" in raw_port:
@@ -346,31 +393,48 @@ class ClusterInfo:
if ":" in port_str:
port_str = port_str.split(":")[-1]
port = int(port_str)
container_ports.append(client.V1ContainerPort(container_port=port, protocol=protocol))
container_ports.append(
client.V1ContainerPort(
container_port=port, protocol=protocol
)
)
if opts.o.debug:
print(f"image: {image}")
print(f"service ports: {container_ports}")
merged_envs = merge_envs(
envs_from_compose_file(
service_info["environment"], self.environment_variables.map), self.environment_variables.map
) if "environment" in service_info else self.environment_variables.map
merged_envs = (
merge_envs(
envs_from_compose_file(
service_info["environment"], self.environment_variables.map
),
self.environment_variables.map,
)
if "environment" in service_info
else self.environment_variables.map
)
envs = envs_from_environment_variables_map(merged_envs)
if opts.o.debug:
print(f"Merged envs: {envs}")
# Re-write the image tag for remote deployment
# Note self.app_name has the same value as deployment_id
image_to_use = remote_tag_for_image_unique(
image,
self.spec.get_image_registry(),
self.app_name) if self.spec.get_image_registry() is not None else image
volume_mounts = volume_mounts_for_service(self.parsed_pod_yaml_map, service_name)
image_to_use = (
remote_tag_for_image_unique(
image, self.spec.get_image_registry(), self.app_name
)
if self.spec.get_image_registry() is not None
else image
)
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()
@@ -387,12 +451,16 @@ class ClusterInfo:
privileged=self.spec.get_privileged(),
capabilities=client.V1Capabilities(
add=self.spec.get_capabilities()
) if self.spec.get_capabilities() else None
)
if self.spec.get_capabilities()
else None,
),
resources=to_k8s_resource_requirements(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
)
image_pull_secrets = [client.V1LocalObjectReference(name="laconic-registry")]
annotations = None
@@ -415,55 +483,54 @@ class ClusterInfo:
affinities = []
for rule in self.spec.get_node_affinities():
# TODO add some input validation here
label_name = rule['label']
label_value = rule['value']
affinities.append(client.V1NodeSelectorTerm(
match_expressions=[client.V1NodeSelectorRequirement(
key=label_name,
operator="In",
values=[label_value]
)]
)
label_name = rule["label"]
label_value = rule["value"]
affinities.append(
client.V1NodeSelectorTerm(
match_expressions=[
client.V1NodeSelectorRequirement(
key=label_name, operator="In", values=[label_value]
)
]
)
)
affinity = client.V1Affinity(
node_affinity=client.V1NodeAffinity(
required_during_scheduling_ignored_during_execution=client.V1NodeSelector(
node_selector_terms=affinities
))
required_during_scheduling_ignored_during_execution=(
client.V1NodeSelector(node_selector_terms=affinities)
)
)
)
if self.spec.get_node_tolerations():
tolerations = []
for toleration in self.spec.get_node_tolerations():
# TODO add some input validation here
toleration_key = toleration['key']
toleration_value = toleration['value']
tolerations.append(client.V1Toleration(
effect="NoSchedule",
key=toleration_key,
operator="Equal",
value=toleration_value
))
toleration_key = toleration["key"]
toleration_value = toleration["value"]
tolerations.append(
client.V1Toleration(
effect="NoSchedule",
key=toleration_key,
operator="Equal",
value=toleration_value,
)
)
template = client.V1PodTemplateSpec(
metadata=client.V1ObjectMeta(
annotations=annotations,
labels=labels
),
metadata=client.V1ObjectMeta(annotations=annotations, labels=labels),
spec=client.V1PodSpec(
containers=containers,
image_pull_secrets=image_pull_secrets,
volumes=volumes,
affinity=affinity,
tolerations=tolerations
),
tolerations=tolerations,
),
)
spec = client.V1DeploymentSpec(
replicas=self.spec.get_replicas(),
template=template, selector={
"matchLabels":
{"app": self.app_name}
}
template=template,
selector={"matchLabels": {"app": self.app_name}},
)
deployment = client.V1Deployment(
stack_orchestrator/deploy/k8s/helm/chart_generator.py
+17 -9
View File
@@ -23,12 +23,12 @@ from stack_orchestrator.util import (
get_pod_file_path,
get_job_list,
get_job_file_path,
error_exit
error_exit,
)
from stack_orchestrator.deploy.k8s.helm.kompose_wrapper import (
check_kompose_available,
get_kompose_version,
convert_to_helm_chart
convert_to_helm_chart,
)
from stack_orchestrator.util import get_yaml
@@ -108,14 +108,17 @@ 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.
Args:
stack_path: Path to the stack directory
spec_file: Path to the deployment spec file
deployment_dir_path: Deployment directory path (already created with deployment.yml)
deployment_dir_path: Deployment directory path
(already created with deployment.yml)
Output structure:
deployment-dir/
@@ -208,13 +211,14 @@ def generate_helm_chart(stack_path: str, spec_file: str, deployment_dir_path: Pa
# 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(
compose_files=compose_files,
output_dir=chart_dir,
chart_name=chart_name
compose_files=compose_files, output_dir=chart_dir, chart_name=chart_name
)
if opts.o.debug:
print(f"Kompose output:\n{output}")
@@ -291,7 +295,11 @@ Edit the generated template files in `templates/` to customize:
print(f" Stack: {stack_path}")
# Count generated files
template_files = list((chart_dir / "templates").glob("*.yaml")) if (chart_dir / "templates").exists() else []
template_files = (
list((chart_dir / "templates").glob("*.yaml"))
if (chart_dir / "templates").exists()
else []
)
print(f" Files: {len(template_files)} template(s) generated")
print("\nDeployment directory structure:")
stack_orchestrator/deploy/k8s/helm/job_runner.py
+31 -10
View File
@@ -53,7 +53,7 @@ def run_helm_job(
release: str = None,
namespace: str = "default",
timeout: int = 600,
verbose: bool = False
verbose: bool = False,
) -> None:
"""
Run a one-time job from a Helm chart.
@@ -93,22 +93,31 @@ 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
jobs_dict = {job_name: {"enabled": True}}
values_json = json.dumps(jobs_dict)
helm_cmd = [
"helm", "template", release, str(chart_dir),
"--show-only", job_template_file,
"--set-json", f"jobs={values_json}"
"helm",
"template",
release,
str(chart_dir),
"--show-only",
job_template_file,
"--set-json",
f"jobs={values_json}",
]
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()
@@ -121,18 +130,30 @@ def run_helm_job(
actual_job_name = manifest.get("metadata", {}).get("name", job_name)
# Apply the job manifest
kubectl_apply_cmd = ["kubectl", "apply", "-f", tmp_file.name, "-n", namespace]
subprocess.run(kubectl_apply_cmd, check=True, capture_output=True, text=True)
kubectl_apply_cmd = [
"kubectl",
"apply",
"-f",
tmp_file.name,
"-n",
namespace,
]
subprocess.run(
kubectl_apply_cmd, check=True, capture_output=True, text=True
)
if verbose:
print(f"Job {actual_job_name} created, waiting for completion...")
# Wait for job completion
wait_cmd = [
"kubectl", "wait", "--for=condition=complete",
"kubectl",
"wait",
"--for=condition=complete",
f"job/{actual_job_name}",
f"--timeout={timeout}s",
"-n", namespace
"-n",
namespace,
]
subprocess.run(wait_cmd, check=True, capture_output=True, text=True)
stack_orchestrator/deploy/k8s/helm/kompose_wrapper.py
+5 -11
View File
@@ -38,10 +38,7 @@ def get_kompose_version() -> str:
raise Exception("kompose not found in PATH")
result = subprocess.run(
["kompose", "version"],
capture_output=True,
text=True,
timeout=10
["kompose", "version"], capture_output=True, text=True, timeout=10
)
if result.returncode != 0:
@@ -55,7 +52,9 @@ def get_kompose_version() -> str:
return version
def convert_to_helm_chart(compose_files: List[Path], output_dir: Path, chart_name: str = None) -> str:
def convert_to_helm_chart(
compose_files: List[Path], output_dir: Path, chart_name: str = None
) -> str:
"""
Invoke kompose to convert Docker Compose files to a Helm chart.
@@ -92,12 +91,7 @@ def convert_to_helm_chart(compose_files: List[Path], output_dir: Path, chart_nam
cmd.extend(["--chart", "-o", str(output_dir)])
# Execute kompose
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=60
)
result = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
if result.returncode != 0:
raise Exception(
stack_orchestrator/deploy/k8s/k8s_command.py
+3 -3
View File
@@ -21,21 +21,21 @@ from stack_orchestrator.deploy.k8s.helpers import get_kind_cluster
@click.group()
@click.pass_context
def command(ctx):
'''k8s cluster management commands'''
"""k8s cluster management commands"""
pass
@command.group()
@click.pass_context
def list(ctx):
'''list k8s resources'''
"""list k8s resources"""
pass
@list.command()
@click.pass_context
def cluster(ctx):
'''Show the existing kind cluster'''
"""Show the existing kind cluster"""
existing_cluster = get_kind_cluster()
if existing_cluster:
print(existing_cluster)