stack-orchestrator/stack_orchestrator/deploy/deploy_util.py

91 lines
3.6 KiB
Python
Raw Normal View History

# Copyright © 2022, 2023 Vulcanize
2023-07-30 04:38:46 +00:00
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
# 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
2023-11-20 06:38:06 +00:00
from typing import List, Any
from stack_orchestrator.deploy.deploy_types import DeployCommandContext, VolumeMapping
from stack_orchestrator.util import get_parsed_stack_config, get_yaml, get_compose_file_dir, get_pod_list
2023-11-20 06:38:06 +00:00
from stack_orchestrator.opts import opts
2023-07-30 04:38:46 +00:00
def _container_image_from_service(stack: str, service: str):
2023-07-30 04:38:46 +00:00
# Parse the compose files looking for the image name of the specified service
image_name = None
parsed_stack = get_parsed_stack_config(stack)
pods = get_pod_list(parsed_stack)
2023-07-30 04:38:46 +00:00
yaml = get_yaml()
for pod in pods:
pod_file_path = os.path.join(get_compose_file_dir(), f"docker-compose-{pod}.yml")
parsed_pod_file = yaml.load(open(pod_file_path, "r"))
if "services" in parsed_pod_file:
services = parsed_pod_file["services"]
if service in services:
service_definition = services[service]
if "image" in service_definition:
image_name = service_definition["image"]
return image_name
2023-11-20 06:38:06 +00:00
def parsed_pod_files_map_from_file_names(pod_files):
parsed_pod_yaml_map : Any = {}
for pod_file in pod_files:
with open(pod_file, "r") as pod_file_descriptor:
parsed_pod_file = get_yaml().load(pod_file_descriptor)
parsed_pod_yaml_map[pod_file] = parsed_pod_file
if opts.o.debug:
print(f"parsed_pod_yaml_map: {parsed_pod_yaml_map}")
return parsed_pod_yaml_map
def images_for_deployment(pod_files: List[str]):
image_set = set()
parsed_pod_yaml_map = parsed_pod_files_map_from_file_names(pod_files)
# Find the set of images in the pods
for pod_name in parsed_pod_yaml_map:
pod = parsed_pod_yaml_map[pod_name]
services = pod["services"]
for service_name in services:
service_info = services[service_name]
image = service_info["image"]
image_set.add(image)
if opts.o.debug:
print(f"image_set: {image_set}")
return image_set
2023-07-30 04:38:46 +00:00
def _volumes_to_docker(mounts: List[VolumeMapping]):
2023-08-23 21:20:28 +00:00
# Example from doc: [("/", "/host"), ("/etc/hosts", "/etc/hosts", "rw")]
2023-07-30 04:38:46 +00:00
result = []
for mount in mounts:
docker_volume = (mount.host_path, mount.container_path)
result.append(docker_volume)
return result
def run_container_command(ctx: DeployCommandContext, service: str, command: str, mounts: List[VolumeMapping]):
2023-10-24 20:44:48 +00:00
deployer = ctx.deployer
2023-07-30 04:38:46 +00:00
container_image = _container_image_from_service(ctx.stack, service)
docker_volumes = _volumes_to_docker(mounts)
2023-08-23 21:20:28 +00:00
if ctx.cluster_context.options.debug:
print(f"Running this command in {service} container: {command}")
2023-10-24 20:44:48 +00:00
docker_output = deployer.run(
2023-08-23 21:20:28 +00:00
container_image,
["-c", command], entrypoint="sh",
user=f"{os.getuid()}:{os.getgid()}",
volumes=docker_volumes
)
2023-07-30 04:38:46 +00:00
# There doesn't seem to be a way to get an exit code from docker.run()
return (docker_output, 0)