| #!/usr/bin/env python3 | |
| # Licensed to the Apache Software Foundation (ASF) under one | |
| # or more contributor license agreements. See the NOTICE file | |
| # distributed with this work for additional information | |
| # regarding copyright ownership. The ASF licenses this file | |
| # to you under the Apache License, Version 2.0 (the | |
| # "License"); you may not use this file except in compliance | |
| # with the License. You may obtain a copy of the License at | |
| # | |
| # http://www.apache.org/licenses/LICENSE-2.0 | |
| # | |
| # Unless required by applicable law or agreed to in writing, software | |
| # distributed under the License is distributed on an "AS IS" BASIS, | |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | |
| # See the License for the specific language governing permissions and | |
| # limitations under the License. | |
| """ | |
| docker_to_squash.py is a tool to facilitate the process of converting | |
| Docker images into squashFS layers, manifests, and configs. | |
| Tool dependencies: skopeo, squashfs-tools, tar, setfattr | |
| """ | |
| import argparse | |
| from collections import Iterable | |
| import glob | |
| import hashlib | |
| import json | |
| import logging | |
| import os | |
| import re | |
| import shutil | |
| import subprocess | |
| import time | |
| from threading import Timer | |
| LOG_LEVEL = None | |
| HADOOP_BIN_DIR = "/undefined" | |
| MAX_IMAGE_LAYERS = 0 | |
| MAX_IMAGE_SIZE = 0 | |
| HADOOP_PREFIX = "/undefined" | |
| ARG_MAX = 10 | |
| HDFS_ROOT = "/undefined" | |
| HDFS_MANIFEST_DIR = "/undefined" | |
| HDFS_CONFIG_DIR = "/undefined" | |
| HDFS_LAYERS_DIR = "/undefined" | |
| HDFS_UNREF_DIR = "/undefined" | |
| def shell_command(command, print_stdout, print_stderr, raise_on_error, timeout_sec=600): | |
| global LOG_LEVEL | |
| global ARG_MAX | |
| stdout_val = subprocess.PIPE | |
| stderr_val = subprocess.PIPE | |
| logging.debug("command: %s", command) | |
| for arg in command: | |
| if len(arg) > ARG_MAX: | |
| raise Exception(f"command length ({len(arg)}) greater than ARG_MAX ({ARG_MAX})") | |
| if print_stdout: | |
| stdout_val = None | |
| if print_stderr or LOG_LEVEL == "DEBUG": | |
| stderr_val = None | |
| process = None | |
| timer = None | |
| out = None | |
| err = None | |
| try: | |
| process = subprocess.Popen(command, stdout=stdout_val, stderr=stderr_val) | |
| timer = Timer(timeout_sec, process_timeout, [process]) | |
| timer.start() | |
| out, err = process.communicate() | |
| if raise_on_error and process.returncode: | |
| exception_string = f"Command: {command} failed with returncode: {process.returncode}" | |
| if out: | |
| exception_string = exception_string + "\nstdout: " + str(out) | |
| if err: | |
| exception_string = exception_string + "\nstderr: " + str(err) | |
| raise Exception(exception_string) | |
| except Exception as ex: | |
| if process and process.poll() is None: | |
| process.kill() | |
| raise Exception("Popen failure, " + str(ex)) | |
| finally: | |
| # Note that finally clause executes even when there is no exception, hence the "return" statement | |
| if timer: | |
| timer.cancel() | |
| return out, err, (process.returncode if process else -1) | |
| def process_timeout(process): | |
| process.kill() | |
| logging.error("Process killed due to timeout") | |
| def does_hdfs_entry_exist(entry, raise_on_error=True): | |
| out, err, rc = hdfs_ls(entry, raise_on_error=raise_on_error) | |
| if rc: | |
| return False | |
| return True | |
| def setup_hdfs_dirs(dirs): | |
| if does_hdfs_entry_exist(dirs, raise_on_error=False): | |
| return | |
| hdfs_mkdir(dirs, create_parents=True) | |
| chmod_dirs = [] | |
| for dir_entry in dirs: | |
| directories = dir_entry.split("/")[1:] | |
| dir_path = "" | |
| for directory in directories: | |
| dir_path = dir_path + "/" + directory | |
| logging.info("dir_path: %s", str(dir_path)) | |
| chmod_dirs.append(dir_path) | |
| hdfs_chmod("755", chmod_dirs) | |
| def append_or_extend_to_list(src, src_list): | |
| if isinstance(src, list): | |
| src_list.extend(src) | |
| else: | |
| src_list.append(src) | |
| def hdfs_get(src, dest, print_stdout=False, print_stderr=False, raise_on_error=True): | |
| global HADOOP_BIN_DIR | |
| command = [HADOOP_BIN_DIR + "/hadoop", "fs", "-get"] | |
| append_or_extend_to_list(src, command) | |
| command.append(dest) | |
| out, err, rc = shell_command(command, print_stdout, print_stderr, raise_on_error) | |
| return out, err, rc | |
| def hdfs_ls(file_path, options="", print_stdout=False, print_stderr=False, raise_on_error=True): | |
| global HADOOP_BIN_DIR | |
| command = [HADOOP_BIN_DIR + "/hadoop", "fs", "-ls"] | |
| if options: | |
| append_or_extend_to_list(options, command) | |
| append_or_extend_to_list(file_path, command) | |
| out, err, rc = shell_command(command, print_stdout, print_stderr, raise_on_error) | |
| return out, err, rc | |
| def hdfs_cat(file_path, print_stdout=False, print_stderr=True, raise_on_error=True): | |
| global HADOOP_BIN_DIR | |
| command = [HADOOP_BIN_DIR + "/hadoop", "fs", "-cat"] | |
| append_or_extend_to_list(file_path, command) | |
| out, err, rc = shell_command(command, print_stdout, print_stderr, raise_on_error) | |
| return out, err, rc | |
| def hdfs_mkdir(file_path, print_stdout=False, print_stderr=True, raise_on_error=True, create_parents=False): | |
| global HADOOP_BIN_DIR | |
| command = [HADOOP_BIN_DIR + "/hadoop", "fs", "-mkdir"] | |
| if create_parents: | |
| command.append("-p") | |
| append_or_extend_to_list(file_path, command) | |
| out, err, rc = shell_command(command, print_stdout, print_stderr, raise_on_error) | |
| return out, err, rc | |
| def hdfs_rm(file_path, error_on_file_not_found=False, print_stdout=False, print_stderr=True, raise_on_error=True): | |
| global HADOOP_BIN_DIR | |
| command = [HADOOP_BIN_DIR + "/hadoop", "fs", "-rm"] | |
| if not error_on_file_not_found: | |
| command.append("-f") | |
| append_or_extend_to_list(file_path, command) | |
| out, err, rc = shell_command(command, print_stdout, print_stderr, raise_on_error) | |
| return out, err, rc | |
| def hdfs_put(src, dest, force=False, print_stdout=False, print_stderr=True, raise_on_error=True): | |
| global HADOOP_BIN_DIR | |
| command = [HADOOP_BIN_DIR + "/hadoop", "fs", "-put"] | |
| if force: | |
| command.append("-f") | |
| append_or_extend_to_list(src, command) | |
| command.append(dest) | |
| out, err, rc = shell_command(command, print_stdout, print_stderr, raise_on_error, 60) | |
| return out, err, rc | |
| def hdfs_chmod(mode, file_path, print_stdout=False, print_stderr=True, raise_on_error=True, recursive=False): | |
| global HADOOP_BIN_DIR | |
| command = [HADOOP_BIN_DIR + "/hadoop", "fs", "-chmod"] | |
| if recursive: | |
| command.append("-R") | |
| command.append(mode) | |
| append_or_extend_to_list(file_path, command) | |
| out, err, rc = shell_command(command, print_stdout, print_stderr, raise_on_error) | |
| return out, err, rc | |
| def hdfs_setrep(replication, file_path, print_stdout=False, print_stderr=True, raise_on_error=True): | |
| global HADOOP_BIN_DIR | |
| command = [HADOOP_BIN_DIR + "/hadoop", "fs", "-setrep", str(replication)] | |
| append_or_extend_to_list(file_path, command) | |
| out, err, rc = shell_command(command, print_stdout, print_stderr, raise_on_error) | |
| return out, err, rc | |
| def hdfs_cp(src, dest, force=False, print_stdout=False, print_stderr=True, raise_on_error=True): | |
| global HADOOP_BIN_DIR | |
| command = [HADOOP_BIN_DIR + "/hadoop", "fs", "-cp"] | |
| if force: | |
| command.append("-f") | |
| append_or_extend_to_list(src, command) | |
| command.append(dest) | |
| out, err, rc = shell_command(command, print_stdout, print_stderr, raise_on_error) | |
| return out, err, rc | |
| def hdfs_touchz(file_path, print_stdout=False, print_stderr=True, raise_on_error=True): | |
| global HADOOP_BIN_DIR | |
| command = [HADOOP_BIN_DIR + "/hadoop", "fs", "-touchz"] | |
| append_or_extend_to_list(file_path, command) | |
| out, err, rc = shell_command(command, print_stdout, print_stderr, raise_on_error) | |
| return out, err, rc | |
| def hdfs_stat(file_path, fmt, print_stdout=False, print_stderr=True, raise_on_error=True): | |
| global HADOOP_BIN_DIR | |
| command = [HADOOP_BIN_DIR + "/hadoop", "fs", "-stat"] | |
| append_or_extend_to_list(fmt, command) | |
| command.append(file_path) | |
| out, err, rc = shell_command(command, print_stdout, print_stderr, raise_on_error) | |
| return out, err, rc | |
| def get_working_dir(directory): | |
| working_dir = "(undefined)" | |
| try: | |
| if os.path.isdir(directory): | |
| working_dir = os.path.join(directory, "docker-to-squash") | |
| else: | |
| working_dir = directory | |
| os.makedirs(working_dir) | |
| except Exception as ex: | |
| raise Exception(f"Could not create working_dir: {working_dir}, {ex}") | |
| return working_dir | |
| def is_sha256_hash(string): | |
| if not re.findall(r"^[a-fA-F\d]{64}$", string): | |
| return False | |
| return True | |
| def calculate_file_hash(filename): | |
| sha = hashlib.sha256() | |
| with open(filename, 'rb') as file_pointer: | |
| while True: | |
| data = file_pointer.read(65536) | |
| if not data: | |
| break | |
| sha.update(data) | |
| hex_digest = sha.hexdigest() | |
| if hex_digest == 0: | |
| raise Exception(f"Hex digest for file: {filename} returned 0") | |
| return hex_digest | |
| def calculate_string_hash(string): | |
| sha = hashlib.sha256() | |
| sha.update(string) | |
| return sha.hexdigest() | |
| def get_local_manifest_from_path(manifest_path): | |
| with open(manifest_path, "rb") as file_pointer: | |
| out = file_pointer.read() | |
| manifest_hash = calculate_string_hash(str(out)) | |
| manifest = json.loads(out) | |
| return manifest, manifest_hash | |
| def get_hdfs_manifest_from_path(manifest_path): | |
| out, err, rc = hdfs_cat(manifest_path) | |
| manifest_hash = calculate_string_hash(str(out)) | |
| manifest = json.loads(out) | |
| return manifest, manifest_hash | |
| def get_hdfs_manifests_from_paths(manifest_paths): | |
| out, err, rc = hdfs_cat(manifest_paths) | |
| manifests_list = out.split("}{") | |
| manifests = [] | |
| for manifest_str in manifests_list: | |
| if manifest_str[0] != "{": | |
| manifest_str = "{" + manifest_str | |
| if manifest_str[-1] != "}": | |
| manifest_str = manifest_str + "}" | |
| manifest_hash = calculate_string_hash(manifest_str) | |
| logging.debug("manifest for %s:\n%s", manifest_hash, manifest_str) | |
| manifest = json.loads(manifest_str) | |
| manifests.append((manifest, manifest_hash)) | |
| return manifests | |
| def get_config_hash_from_manifest(manifest): | |
| config_hash = manifest['config']['digest'].split(":", 1)[1] | |
| return config_hash | |
| def check_total_layer_number(layers): | |
| global MAX_IMAGE_LAYERS | |
| if len(layers) > MAX_IMAGE_LAYERS: | |
| logging.error("layers: " + str(layers)) | |
| raise Exception("Image has " + str(len(layers)) + | |
| " layers, which is more than the maximum " + str(MAX_IMAGE_LAYERS) + | |
| " layers. Failing out") | |
| def check_total_layer_size(manifest, size): | |
| global MAX_IMAGE_SIZE | |
| if size > MAX_IMAGE_SIZE: | |
| for layer in manifest['layers']: | |
| logging.error("layer " + layer['digest'] + " has size " + str(layer['size'])) | |
| raise Exception("Image has total size " + str(size) + | |
| " B. which is more than the maximum size " + str(MAX_IMAGE_SIZE) + " B. Failing out") | |
| def get_layer_hashes_from_manifest(manifest, error_on_size_check=True): | |
| layers = [] | |
| size = 0 | |
| for layer in manifest['layers']: | |
| layers.append(layer['digest'].split(":", 1)[1]) | |
| size += layer['size'] | |
| if error_on_size_check: | |
| check_total_layer_number(layers) | |
| check_total_layer_size(manifest, size) | |
| return layers | |
| def get_pull_fmt_string(pull_format): | |
| pull_fmt_string = pull_format + ":" | |
| if pull_format == "docker": | |
| pull_fmt_string = pull_fmt_string + "//" | |
| return pull_fmt_string | |
| def get_manifest_from_docker_image(pull_format, image): | |
| pull_fmt_string = get_pull_fmt_string(pull_format) | |
| out, err, rc = shell_command(["skopeo", "inspect", "--raw", pull_fmt_string + image], False, True, True) | |
| manifest = json.loads(out) | |
| if 'manifests' in manifest: | |
| logging.debug("skopeo inspect --raw returned a list of manifests") | |
| manifests_dict = manifest['manifests'] | |
| sha = None | |
| for mfest in manifests_dict: | |
| if mfest['platform']['architecture'] == "amd64": | |
| sha = mfest['digest'] | |
| break | |
| if not sha: | |
| raise Exception("Could not find amd64 manifest for image " + image) | |
| image_without_tag = image.split("/", 1)[-1].split(":", 1)[0] | |
| image_and_sha = image_without_tag + "@" + sha | |
| logging.debug("amd64 manifest sha is: %s", sha) | |
| manifest, manifest_hash = get_manifest_from_docker_image(pull_format, image_and_sha) | |
| else: | |
| manifest_hash = calculate_string_hash(str(out)) | |
| logging.debug("manifest: %s", str(manifest)) | |
| return manifest, manifest_hash | |
| def split_image_and_tag(image_and_tag): | |
| split = image_and_tag.split(",") | |
| image = split[0] | |
| tags = split[1:] | |
| return image, tags | |
| def read_image_tag_to_hash(image_tag_to_hash): | |
| hash_to_tags = dict() | |
| tag_to_hash = dict() | |
| with open(image_tag_to_hash, 'rb') as file_pointer: | |
| while True: | |
| line = file_pointer.readline() | |
| if not line: | |
| break | |
| line = str(line).rstrip() | |
| if not line: | |
| continue | |
| comment_split_line = line.split("#", 1) | |
| line = comment_split_line[0] | |
| comment = comment_split_line[1:] | |
| split_line = line.rsplit(":", 1) | |
| manifest_hash = split_line[-1] | |
| tags_list = ' '.join(split_line[:-1]).split(",") | |
| if not is_sha256_hash(manifest_hash) or not tags_list: | |
| logging.warning("image-tag-to-hash file malformed. Skipping entry %s", line) | |
| continue | |
| tags_and_comments = hash_to_tags.get(manifest_hash, None) | |
| if tags_and_comments is None: | |
| known_tags = tags_list | |
| known_comment = comment | |
| else: | |
| known_tags = tags_and_comments[0] | |
| for tag in tags_list: | |
| if tag not in known_tags: | |
| known_tags.append(tag) | |
| known_comment = tags_and_comments[1] | |
| known_comment.extend(comment) | |
| hash_to_tags[manifest_hash] = (known_tags, known_comment) | |
| for tag in tags_list: | |
| cur_manifest = tag_to_hash.get(tag, None) | |
| if cur_manifest is not None: | |
| logging.warning(f"tag_to_hash already has manifest {cur_manifest} defined for tag {tag}. " | |
| f"This entry will be overwritten") | |
| tag_to_hash[tag] = manifest_hash | |
| return hash_to_tags, tag_to_hash | |
| def remove_tag_from_dicts(hash_to_tags, tag_to_hash, tag): | |
| if not hash_to_tags: | |
| logging.debug("hash_to_tags is null. Not removing tag %s", tag) | |
| return | |
| prev_hash = tag_to_hash.get(tag, None) | |
| if prev_hash is not None: | |
| del tag_to_hash[tag] | |
| prev_tags, prev_comment = hash_to_tags.get(prev_hash, (None, None)) | |
| prev_tags.remove(tag) | |
| if prev_tags == 0: | |
| del hash_to_tags[prev_hash] | |
| else: | |
| hash_to_tags[prev_hash] = (prev_tags, prev_comment) | |
| else: | |
| logging.debug("Tag not found. Not removing tag: %s", tag) | |
| def remove_image_hash_from_dicts(hash_to_tags, tag_to_hash, image_hash): | |
| if not hash_to_tags: | |
| logging.debug("hash_to_tags is null. Not removing image_hash %s", image_hash) | |
| return | |
| logging.debug("hash_to_tags: %s", str(hash_to_tags)) | |
| logging.debug("Removing image_hash from dicts: %s", image_hash) | |
| prev_tags, prev_comments = hash_to_tags.get(image_hash, None) | |
| if prev_tags is not None: | |
| hash_to_tags.pop(image_hash) | |
| for tag in prev_tags: | |
| del tag_to_hash[tag] | |
| def add_tag_to_dicts(hash_to_tags, tag_to_hash, tag, manifest_hash, comment): | |
| tag_to_hash[tag] = manifest_hash | |
| new_tags_and_comments = hash_to_tags.get(manifest_hash, None) | |
| if new_tags_and_comments is None: | |
| new_tags = [tag] | |
| new_comment = [comment] | |
| else: | |
| new_tags = new_tags_and_comments[0] | |
| new_comment = new_tags_and_comments[1] | |
| if tag not in new_tags: | |
| new_tags.append(tag) | |
| if comment and comment not in new_comment: | |
| new_comment.append(comment) | |
| hash_to_tags[manifest_hash] = (new_tags, new_comment) | |
| def write_local_image_tag_to_hash(image_tag_to_hash, hash_to_tags): | |
| file_contents = [] | |
| for key, value in hash_to_tags.iteritems(): | |
| manifest_hash = key | |
| # Sort tags list to preserve consistent order | |
| value[0].sort() | |
| tags = ','.join(map(str, value[0])) | |
| if tags: | |
| # Sort comments list to preserve consistent order | |
| value[1].sort() | |
| comment = ', '.join(map(str, value[1])) | |
| if comment: | |
| comment = "#" + comment | |
| file_contents.append(tags + ":" + manifest_hash + comment + "\n") | |
| file_contents.sort() | |
| with open(image_tag_to_hash, 'w') as file_pointer: | |
| for val in file_contents: | |
| file_pointer.write(val) | |
| def update_dicts_for_multiple_tags(hash_to_tags, tag_to_hash, tags, manifest_hash, comment): | |
| for tag in tags: | |
| update_dicts(hash_to_tags, tag_to_hash, tag, manifest_hash, comment) | |
| def update_dicts(hash_to_tags, tag_to_hash, tag, manifest_hash, comment): | |
| remove_tag_from_dicts(hash_to_tags, tag_to_hash, tag) | |
| add_tag_to_dicts(hash_to_tags, tag_to_hash, tag, manifest_hash, comment) | |
| def remove_from_dicts(hash_to_tags, tag_to_hash, tags): | |
| for tag in tags: | |
| logging.debug("removing tag: %s", tag) | |
| remove_tag_from_dicts(hash_to_tags, tag_to_hash, tag) | |
| def populate_tag_dicts(image_tag_to_hash, local_image_tag_to_hash): | |
| return populate_tag_dicts_set_root(image_tag_to_hash, local_image_tag_to_hash, None) | |
| def populate_tag_dicts_set_root(image_tag_to_hash, local_image_tag_to_hash, hdfs_root): | |
| # Setting hdfs_root to None will default it to using the global | |
| global HDFS_ROOT | |
| if not hdfs_root: | |
| hdfs_root = HDFS_ROOT | |
| hdfs_get(hdfs_root + "/" + image_tag_to_hash, local_image_tag_to_hash, raise_on_error=True) | |
| image_tag_to_hash_hash = calculate_file_hash(local_image_tag_to_hash) | |
| if image_tag_to_hash_hash: | |
| hash_to_tags, tag_to_hash = read_image_tag_to_hash(local_image_tag_to_hash) | |
| else: | |
| hash_to_tags = {} | |
| tag_to_hash = {} | |
| return hash_to_tags, tag_to_hash, image_tag_to_hash_hash | |
| def setup_squashfs_hdfs_dirs(hdfs_dirs, image_tag_to_hash_path): | |
| logging.debug("Setting up squashfs hdfs_dirs: %s", str(hdfs_dirs)) | |
| setup_hdfs_dirs(hdfs_dirs) | |
| if not does_hdfs_entry_exist(image_tag_to_hash_path, raise_on_error=False): | |
| hdfs_touchz(image_tag_to_hash_path) | |
| hdfs_chmod("755", image_tag_to_hash_path) | |
| def skopeo_copy_image(pull_format, image, skopeo_format, skopeo_dir): | |
| logging.info("Pulling image: %s", image) | |
| if os.path.isdir(skopeo_dir): | |
| raise Exception("Skopeo output directory already exists. " | |
| + "Please delete and try again " | |
| + "Directory: " + skopeo_dir) | |
| pull_fmt_string = get_pull_fmt_string(pull_format) | |
| shell_command(["skopeo", "copy", pull_fmt_string + image, | |
| skopeo_format + ":" + skopeo_dir], False, True, True) | |
| def untar_layer(tmp_dir, layer_path): | |
| shell_command(["tar", "-C", tmp_dir, "--xattrs", "--xattrs-include='*'", "-xf", layer_path], False, True, True) | |
| def tar_file_search(archive, target): | |
| out, err, rc = shell_command(["tar", "-xf", archive, target, "-O"], False, False, False) | |
| return out | |
| def set_fattr(directory): | |
| shell_command(["setfattr", "-n", "trusted.overlay.opaque", "-v", "y", directory], False, True, True) | |
| def make_whiteout_block_device(file_path, whiteout): | |
| shell_command(["mknod", "-m", "000", file_path, "c", "0", "0"], False, True, True) | |
| out, err, rc = shell_command(["stat", "-c", "%U:%G", whiteout], False, True, True) | |
| perms = str(out).strip() | |
| shell_command(["chown", perms, file_path], False, True, True) | |
| def convert_oci_whiteouts(tmp_dir): | |
| out, err, rc = shell_command(["find", tmp_dir, "-name", ".wh.*"], False, False, True) | |
| whiteouts = str(out).splitlines() | |
| for whiteout in whiteouts: | |
| if whiteout == 0: | |
| continue | |
| basename = os.path.basename(whiteout) | |
| directory = os.path.dirname(whiteout) | |
| if basename == ".wh..wh..opq": | |
| set_fattr(directory) | |
| else: | |
| whiteout_string = ".wh." | |
| idx = basename.rfind(whiteout_string) | |
| bname = basename[idx + len(whiteout_string):] | |
| file_path = os.path.join(directory, bname) | |
| make_whiteout_block_device(file_path, whiteout) | |
| shell_command(["rm", whiteout], False, True, True) | |
| def dir_to_squashfs(tmp_dir, squash_path): | |
| shell_command(["mksquashfs", tmp_dir, squash_path, "-write-queue", "4096", "-read-queue", "4096", | |
| "-fragment-queue", "4096"], | |
| False, True, True) | |
| def upload_to_hdfs(src, dest, replication, mode, force=False): | |
| if does_hdfs_entry_exist(dest, raise_on_error=False): | |
| if not force: | |
| logging.warning("Not uploading to HDFS. File already exists: %s", dest) | |
| return | |
| logging.info("File already exists, but overwriting due to force option: %s", dest) | |
| hdfs_put(src, dest, force) | |
| hdfs_setrep(replication, dest) | |
| hdfs_chmod(mode, dest) | |
| logging.info(f"Uploaded file {dest} with replication {replication} and permissions {mode}") | |
| def atomic_upload_mv_to_hdfs(src, dest, replication, image_tag_to_hash_file_hash): | |
| global HADOOP_PREFIX | |
| global HADOOP_BIN_DIR | |
| local_hash = calculate_file_hash(src) | |
| if local_hash == image_tag_to_hash_file_hash: | |
| logging.info("image_tag_to_hash file unchanged. Not uploading") | |
| return | |
| tmp_dest = dest + ".tmp" | |
| try: | |
| if does_hdfs_entry_exist(tmp_dest, raise_on_error=False): | |
| hdfs_rm(tmp_dest) | |
| hdfs_put(src, tmp_dest) | |
| hdfs_setrep(replication, tmp_dest) | |
| hdfs_chmod("444", tmp_dest) | |
| jar_path = HADOOP_PREFIX + "/share/hadoop/tools/lib/hadoop-extras-*.jar" | |
| jar_file = None | |
| for file in glob.glob(jar_path): | |
| jar_file = file | |
| if not jar_file: | |
| raise Exception("SymlinkTool Jar doesn't exist: %s" % jar_path) | |
| logging.debug("jar_file: " + jar_file) | |
| shell_command([HADOOP_BIN_DIR + "/hadoop", "jar", jar_file, "org.apache.hadoop.tools.SymlinkTool", | |
| "mvlink", "-f", tmp_dest, dest], False, False, True) | |
| except Exception as ex: | |
| if does_hdfs_entry_exist(tmp_dest, raise_on_error=False): | |
| hdfs_rm(tmp_dest) | |
| raise Exception("image tag to hash file upload failed, exception=" + str(ex)) | |
| def docker_to_squash(layer_dir, layer, working_dir): | |
| tmp_dir = os.path.join(working_dir, "expand_archive_" + layer) | |
| layer_path = os.path.join(layer_dir, layer) | |
| squash_path = layer_path + ".sqsh" | |
| if os.path.isdir(tmp_dir): | |
| raise Exception("tmp_dir already exists. Please delete and try again " + | |
| "Directory: " + tmp_dir) | |
| os.makedirs(tmp_dir) | |
| try: | |
| untar_layer(tmp_dir, layer_path) | |
| convert_oci_whiteouts(tmp_dir) | |
| dir_to_squashfs(tmp_dir, squash_path) | |
| finally: | |
| os.remove(layer_path) | |
| shell_command(["rm", "-rf", tmp_dir], False, True, True) | |
| def check_image_for_magic_file(magic_file, skopeo_dir, layers): | |
| magic_file_absolute = magic_file.strip("/") | |
| logging.debug("Searching for magic file %s", magic_file_absolute) | |
| for layer in layers: | |
| ret = tar_file_search(os.path.join(skopeo_dir, layer), magic_file_absolute) | |
| if ret: | |
| logging.debug("Found magic file %s in layer %s", magic_file_absolute, layer) | |
| logging.debug("Magic file %s has contents:\n%s", magic_file_absolute, ret) | |
| return ret | |
| raise Exception(f"Magic file {magic_file_absolute} doesn't exist in any layer") | |
| def pull_build_push_update(args): | |
| global HDFS_ROOT | |
| global HDFS_MANIFEST_DIR | |
| global HDFS_CONFIG_DIR | |
| global HDFS_LAYERS_DIR | |
| global HDFS_UNREF_DIR | |
| skopeo_format = args.skopeo_format | |
| pull_format = args.pull_format | |
| image_tag_to_hash = args.image_tag_to_hash | |
| replication = args.replication | |
| force = args.force | |
| images_and_tags = args.images_and_tags | |
| check_magic_file = args.check_magic_file | |
| magic_file = args.magic_file | |
| bootstrap = args.bootstrap | |
| working_dir = None | |
| try: | |
| working_dir = get_working_dir(args.working_dir) | |
| local_image_tag_to_hash = os.path.join(working_dir, os.path.basename(image_tag_to_hash)) | |
| if bootstrap: | |
| hdfs_dirs = [HDFS_ROOT, HDFS_LAYERS_DIR, HDFS_CONFIG_DIR, HDFS_MANIFEST_DIR, HDFS_UNREF_DIR] | |
| image_tag_to_hash_path = HDFS_ROOT + "/" + image_tag_to_hash | |
| setup_squashfs_hdfs_dirs(hdfs_dirs, image_tag_to_hash_path) | |
| hash_to_tags, tag_to_hash, image_tag_to_hash_hash = populate_tag_dicts(image_tag_to_hash, | |
| local_image_tag_to_hash) | |
| for image_and_tag_arg in images_and_tags: | |
| image, tags = split_image_and_tag(image_and_tag_arg) | |
| if not image or not tags: | |
| raise Exception("Positional parameter requires an image and at least 1 tag: " + image_and_tag_arg) | |
| logging.info("Working on image %s with tags %s", image, str(tags)) | |
| manifest, manifest_hash = get_manifest_from_docker_image(pull_format, image) | |
| layers = get_layer_hashes_from_manifest(manifest) | |
| config_hash = get_config_hash_from_manifest(manifest) | |
| logging.debug("Layers: %s", str(layers)) | |
| logging.debug("Config: %s", str(config_hash)) | |
| update_dicts_for_multiple_tags(hash_to_tags, tag_to_hash, tags, manifest_hash, image) | |
| hdfs_files_to_check = [HDFS_MANIFEST_DIR + "/" + manifest_hash, HDFS_CONFIG_DIR + "/" + config_hash] | |
| for layer in layers: | |
| hdfs_files_to_check.append(HDFS_LAYERS_DIR + "/" + layer + ".sqsh") | |
| if does_hdfs_entry_exist(hdfs_files_to_check, raise_on_error=False): | |
| if not force: | |
| logging.info("All image files exist in HDFS, skipping this image") | |
| continue | |
| logging.info("All image files exist in HDFS, but force option set, so overwriting image") | |
| skopeo_dir = os.path.join(working_dir, image.split("/")[-1]) | |
| logging.debug("skopeo_dir: %s", skopeo_dir) | |
| skopeo_copy_image(pull_format, image, skopeo_format, skopeo_dir) | |
| if check_magic_file: | |
| check_image_for_magic_file(magic_file, skopeo_dir, layers) | |
| for layer in layers: | |
| logging.info("Squashifying and uploading layer: %s", layer) | |
| hdfs_squash_path = HDFS_LAYERS_DIR + "/" + layer + ".sqsh" | |
| if does_hdfs_entry_exist(hdfs_squash_path, raise_on_error=False): | |
| if force: | |
| logging.info(f"Layer already exists, but overwriting due to force option: {layer}") | |
| else: | |
| logging.info(f"Layer exists. Skipping and not squashifying or uploading: {layer}") | |
| continue | |
| docker_to_squash(skopeo_dir, layer, working_dir) | |
| squash_path = os.path.join(skopeo_dir, layer + ".sqsh") | |
| squash_name = os.path.basename(squash_path) | |
| upload_to_hdfs(squash_path, HDFS_LAYERS_DIR + "/" + squash_name, replication, "444", force) | |
| config_local_path = os.path.join(skopeo_dir, config_hash) | |
| upload_to_hdfs(config_local_path, | |
| HDFS_CONFIG_DIR + "/" + os.path.basename(config_local_path), | |
| replication, "444", force) | |
| manifest_local_path = os.path.join(skopeo_dir, "manifest.json") | |
| upload_to_hdfs(manifest_local_path, HDFS_MANIFEST_DIR + "/" + manifest_hash, replication, "444", force) | |
| write_local_image_tag_to_hash(local_image_tag_to_hash, hash_to_tags) | |
| atomic_upload_mv_to_hdfs(local_image_tag_to_hash, HDFS_ROOT + "/" + image_tag_to_hash, | |
| replication, image_tag_to_hash_hash) | |
| finally: | |
| if working_dir: | |
| if os.path.isdir(working_dir): | |
| shell_command(["rm", "-rf", working_dir], False, True, True) | |
| def pull_build(args): | |
| skopeo_format = args.skopeo_format | |
| pull_format = args.pull_format | |
| images_and_tags = args.images_and_tags | |
| check_magic_file = args.check_magic_file | |
| magic_file = args.magic_file | |
| for image_and_tag_arg in images_and_tags: | |
| image, tags = split_image_and_tag(image_and_tag_arg) | |
| if not image or not tags: | |
| raise Exception("Positional parameter requires an image and at least 1 tag: " + image_and_tag_arg) | |
| logging.info(f"Working on image {image} with tags {tags}") | |
| manifest, manifest_hash = get_manifest_from_docker_image(pull_format, image) | |
| layers = get_layer_hashes_from_manifest(manifest) | |
| config_hash = get_config_hash_from_manifest(manifest) | |
| logging.debug(f"Layers: {layers}") | |
| logging.debug(f"Config: {config_hash}") | |
| skopeo_dir = None | |
| try: | |
| working_dir = get_working_dir(args.working_dir) | |
| skopeo_dir = os.path.join(working_dir, image.split("/")[-1]) | |
| logging.debug(f"skopeo_dir: {skopeo_dir}") | |
| skopeo_copy_image(pull_format, image, skopeo_format, skopeo_dir) | |
| if check_magic_file: | |
| check_image_for_magic_file(magic_file, skopeo_dir, layers) | |
| for layer in layers: | |
| logging.info(f"Squashifying layer: {layer}") | |
| docker_to_squash(skopeo_dir, layer, working_dir) | |
| except Exception as _: | |
| if skopeo_dir and os.path.isdir(skopeo_dir): | |
| shutil.rmtree(skopeo_dir) | |
| raise | |
| def push_update(args): | |
| global HDFS_ROOT | |
| global HDFS_MANIFEST_DIR | |
| global HDFS_CONFIG_DIR | |
| global HDFS_LAYERS_DIR | |
| global HDFS_UNREF_DIR | |
| image_tag_to_hash = args.image_tag_to_hash | |
| replication = args.replication | |
| force = args.force | |
| images_and_tags = args.images_and_tags | |
| bootstrap = args.bootstrap | |
| local_image_tag_to_hash = None | |
| try: | |
| working_dir = get_working_dir(args.working_dir) | |
| local_image_tag_to_hash = os.path.join(working_dir, os.path.basename(image_tag_to_hash)) | |
| if bootstrap: | |
| hdfs_dirs = [HDFS_ROOT, HDFS_LAYERS_DIR, HDFS_CONFIG_DIR, HDFS_MANIFEST_DIR, HDFS_UNREF_DIR] | |
| image_tag_to_hash_path = HDFS_ROOT + "/" + image_tag_to_hash | |
| setup_squashfs_hdfs_dirs(hdfs_dirs, image_tag_to_hash_path) | |
| hash_to_tags, tag_to_hash, image_tag_to_hash_hash = populate_tag_dicts(image_tag_to_hash, | |
| local_image_tag_to_hash) | |
| for image_and_tag_arg in images_and_tags: | |
| image, tags = split_image_and_tag(image_and_tag_arg) | |
| if not image or not tags: | |
| raise Exception("Positional parameter requires an image and at least 1 tag: " + image_and_tag_arg) | |
| logging.info("Working on image %s with tags %s", image, str(tags)) | |
| skopeo_dir = os.path.join(working_dir, image.split("/")[-1]) | |
| if not os.path.exists(skopeo_dir): | |
| raise Exception(f"skopeo_dir doesn't exists: {skopeo_dir}") | |
| manifest, manifest_hash = get_local_manifest_from_path(f"{skopeo_dir}/manifest.json") | |
| layers = get_layer_hashes_from_manifest(manifest) | |
| config_hash = get_config_hash_from_manifest(manifest) | |
| logging.debug("Layers: %s", str(layers)) | |
| logging.debug("Config: %s", str(config_hash)) | |
| update_dicts_for_multiple_tags(hash_to_tags, tag_to_hash, tags, manifest_hash, image) | |
| hdfs_files_to_check = [f"{HDFS_MANIFEST_DIR}/{manifest_hash}", f"{HDFS_CONFIG_DIR}/{config_hash}"] | |
| for layer in layers: | |
| hdfs_files_to_check.append(f"{HDFS_LAYERS_DIR}/{layer}.sqsh") | |
| if does_hdfs_entry_exist(hdfs_files_to_check, raise_on_error=False): | |
| if not force: | |
| logging.info("All image files exist in HDFS, skipping this image") | |
| continue | |
| logging.info("All image files exist in HDFS, but force option set, so overwriting image") | |
| for layer in layers: | |
| hdfs_squash_path = f"{HDFS_LAYERS_DIR}/{layer}.sqsh" | |
| if does_hdfs_entry_exist(hdfs_squash_path, raise_on_error=False): | |
| if force: | |
| logging.info(f"Layer already exists, but overwriting due to force option: {layer}") | |
| else: | |
| logging.info(f"Layer exists. Skipping and not squashifying or uploading: {layer}") | |
| continue | |
| squash_path = os.path.join(skopeo_dir, layer + ".sqsh") | |
| squash_name = os.path.basename(squash_path) | |
| upload_to_hdfs(squash_path, f"{HDFS_LAYERS_DIR}/{squash_name}", replication, "444", force) | |
| config_local_path = os.path.join(skopeo_dir, config_hash) | |
| upload_to_hdfs(config_local_path, | |
| f"{HDFS_CONFIG_DIR}/" + os.path.basename(config_local_path), | |
| replication, "444", force) | |
| manifest_local_path = os.path.join(skopeo_dir, "manifest.json") | |
| upload_to_hdfs(manifest_local_path, HDFS_MANIFEST_DIR + "/" + manifest_hash, replication, "444", force) | |
| write_local_image_tag_to_hash(local_image_tag_to_hash, hash_to_tags) | |
| atomic_upload_mv_to_hdfs(local_image_tag_to_hash, f"{HDFS_ROOT}/{image_tag_to_hash}", | |
| replication, image_tag_to_hash_hash) | |
| finally: | |
| if local_image_tag_to_hash: | |
| if os.path.isfile(local_image_tag_to_hash): | |
| os.remove(local_image_tag_to_hash) | |
| def remove_image(args): | |
| global HDFS_ROOT | |
| global HDFS_MANIFEST_DIR | |
| global HDFS_CONFIG_DIR | |
| global HDFS_LAYERS_DIR | |
| global HDFS_UNREF_DIR | |
| image_tag_to_hash = args.image_tag_to_hash | |
| replication = args.replication | |
| images_or_tags = args.images_or_tags | |
| working_dir = None | |
| try: | |
| working_dir = get_working_dir(args.working_dir) | |
| local_image_tag_to_hash = os.path.join(working_dir, os.path.basename(image_tag_to_hash)) | |
| images_and_tags_to_remove = [] | |
| for image_or_tag_arg in images_or_tags: | |
| images_and_tags_to_remove.extend(image_or_tag_arg.split(",")) | |
| logging.debug("images_and_tags_to_remove:\n%s", images_and_tags_to_remove) | |
| hash_to_tags, tag_to_hash, image_tag_to_hash_hash = populate_tag_dicts(image_tag_to_hash, | |
| local_image_tag_to_hash) | |
| logging.debug("hash_to_tags: %s", str(hash_to_tags)) | |
| logging.debug("tag_to_hash: %s", str(tag_to_hash)) | |
| known_images = get_all_known_images() | |
| if not known_images: | |
| logging.warn("No known images\n") | |
| return | |
| images_to_remove = get_images_from_args(images_and_tags_to_remove, tag_to_hash, known_images) | |
| logging.debug("images_to_remove:\n%s", images_to_remove) | |
| if not images_to_remove: | |
| logging.warning("No images to remove") | |
| return | |
| delete_list = get_delete_list_from_images_to_remove(images_to_remove, known_images) | |
| for image_to_remove in images_to_remove: | |
| remove_image_hash_from_dicts(hash_to_tags, tag_to_hash, image_to_remove.manifest_stat.name) | |
| write_local_image_tag_to_hash(local_image_tag_to_hash, hash_to_tags) | |
| atomic_upload_mv_to_hdfs(local_image_tag_to_hash, HDFS_ROOT + "/" + image_tag_to_hash, replication, | |
| image_tag_to_hash_hash) | |
| hdfs_rm(delete_list) | |
| finally: | |
| if working_dir: | |
| if os.path.isdir(working_dir): | |
| shutil.rmtree(working_dir) | |
| def get_images_from_args(images_and_tags, tag_to_hash, known_images): | |
| images = [] | |
| if isinstance(images_and_tags, Iterable): | |
| for image_arg in images_and_tags: | |
| image = get_image_hash_from_arg(image_arg, tag_to_hash, known_images) | |
| if image: | |
| images.append(image) | |
| else: | |
| image_arg = images_and_tags[0] | |
| image = get_image_hash_from_arg(image_arg, tag_to_hash, known_images) | |
| if image: | |
| images.append(image) | |
| return images | |
| def get_image_hash_from_arg(image_str, tag_to_hash, known_images): | |
| if is_sha256_hash(image_str): | |
| image_hash = image_str | |
| else: | |
| image_hash = tag_to_hash.get(image_str, None) | |
| if image_hash: | |
| image = get_known_image_by_hash(image_hash, known_images) | |
| else: | |
| logging.warn("image tag unknown: %s", image_str) | |
| return None | |
| return image | |
| def get_delete_list_from_images_to_remove(images_to_remove, known_images): | |
| global HDFS_ROOT | |
| global HDFS_MANIFEST_DIR | |
| global HDFS_CONFIG_DIR | |
| global HDFS_LAYERS_DIR | |
| global HDFS_UNREF_DIR | |
| layers_to_keep = [] | |
| delete_list = [] | |
| for image in known_images: | |
| if image not in images_to_remove: | |
| layers_to_keep.extend(image.layers) | |
| for image_to_remove in images_to_remove: | |
| delete_list.append(HDFS_MANIFEST_DIR + "/" + image_to_remove.manifest_stat.name) | |
| delete_list.append(HDFS_CONFIG_DIR + "/" + image_to_remove.config) | |
| if image_to_remove.unref_file_stat: | |
| delete_list.append(HDFS_UNREF_DIR + "/" + image_to_remove.unref_file_stat.name) | |
| layers = image_to_remove.layers | |
| for layer in layers: | |
| if layer not in layers_to_keep: | |
| layer_path = HDFS_LAYERS_DIR + "/" + layer + ".sqsh" | |
| if layer_path not in delete_list: | |
| delete_list.append(layer_path) | |
| logging.debug("delete_list:\n%s", delete_list) | |
| return delete_list | |
| def add_remove_tag(args): | |
| global HDFS_ROOT | |
| global HDFS_MANIFEST_DIR | |
| pull_format = args.pull_format | |
| image_tag_to_hash = args.image_tag_to_hash | |
| replication = args.replication | |
| sub_command = args.sub_command | |
| images_and_tags = args.images_and_tags | |
| working_dir = None | |
| try: | |
| working_dir = get_working_dir(args.working_dir) | |
| local_image_tag_to_hash = os.path.join(working_dir, os.path.basename(image_tag_to_hash)) | |
| hash_to_tags, tag_to_hash, image_tag_to_hash_hash = populate_tag_dicts(image_tag_to_hash, | |
| local_image_tag_to_hash) | |
| for image_and_tag_arg in images_and_tags: | |
| if sub_command == "add-tag": | |
| image, tags = split_image_and_tag(image_and_tag_arg) | |
| if is_sha256_hash(image): | |
| manifest_hash = image | |
| else: | |
| manifest_hash = tag_to_hash.get(image, None) | |
| if manifest_hash: | |
| manifest_path = HDFS_MANIFEST_DIR + "/" + manifest_hash | |
| out, err, returncode = hdfs_cat(manifest_path) | |
| manifest = json.loads(out) | |
| logging.debug("image tag exists for %s", image) | |
| else: | |
| manifest, manifest_hash = get_manifest_from_docker_image(pull_format, image) | |
| update_dicts_for_multiple_tags(hash_to_tags, tag_to_hash, tags, manifest_hash, image) | |
| elif sub_command == "remove-tag": | |
| tags = image_and_tag_arg.split(",") | |
| image = None | |
| manifest = None | |
| manifest_hash = 0 | |
| remove_from_dicts(hash_to_tags, tag_to_hash, tags) | |
| else: | |
| raise Exception(f"Invalid sub_command: {sub_command}") | |
| write_local_image_tag_to_hash(local_image_tag_to_hash, hash_to_tags) | |
| atomic_upload_mv_to_hdfs(local_image_tag_to_hash, HDFS_ROOT + "/" + image_tag_to_hash, replication, | |
| image_tag_to_hash_hash) | |
| finally: | |
| if working_dir: | |
| if os.path.isdir(working_dir): | |
| shutil.rmtree(working_dir) | |
| def copy_update(args): | |
| image_tag_to_hash = args.image_tag_to_hash | |
| replication = args.replication | |
| force = args.force | |
| src_root = args.src_root | |
| dest_root = args.dest_root | |
| images_and_tags = args.images_and_tags | |
| bootstrap = args.bootstrap | |
| src_layers_dir = src_root + "/layers" | |
| src_config_dir = src_root + "/config" | |
| src_manifest_dir = src_root + "/manifests" | |
| dest_layers_dir = dest_root + "/layers" | |
| dest_config_dir = dest_root + "/config" | |
| dest_manifest_dir = dest_root + "/manifests" | |
| dest_unref_dir = dest_root + "/unreferenced" | |
| if bootstrap: | |
| hdfs_dirs = [dest_root, dest_layers_dir, dest_config_dir, dest_manifest_dir, dest_unref_dir] | |
| image_tag_to_hash_path = dest_root + "/" + image_tag_to_hash | |
| setup_squashfs_hdfs_dirs(hdfs_dirs, image_tag_to_hash_path) | |
| working_dir = None | |
| try: | |
| working_dir = get_working_dir(args.working_dir) | |
| local_src_image_tag_to_hash = os.path.join(working_dir, "src-" | |
| + os.path.basename(image_tag_to_hash)) | |
| local_dest_image_tag_to_hash = os.path.join(working_dir, "dest-" | |
| + os.path.basename(image_tag_to_hash)) | |
| src_hash_to_tags, src_tag_to_hash, src_image_tag_to_hash_hash = populate_tag_dicts_set_root(image_tag_to_hash, | |
| local_src_image_tag_to_hash, | |
| src_root) | |
| dest_hash_to_tags, dest_tag_to_hash, dest_image_tag_to_hash_hash = populate_tag_dicts_set_root( | |
| image_tag_to_hash, local_dest_image_tag_to_hash, dest_root) | |
| for image_and_tag_arg in images_and_tags: | |
| image, tags = split_image_and_tag(image_and_tag_arg) | |
| if not image: | |
| raise Exception("Positional parameter requires an image: " + image_and_tag_arg) | |
| if not tags: | |
| logging.debug("Tag not given. Using image tag instead: %s", image) | |
| tags = [image] | |
| src_manifest_hash = src_tag_to_hash.get(image, None) | |
| if not src_manifest_hash: | |
| logging.info("Manifest not found for image %s. Skipping", image) | |
| continue | |
| src_manifest_path = src_manifest_dir + "/" + src_manifest_hash | |
| dest_manifest_path = dest_manifest_dir + "/" + src_manifest_hash | |
| src_manifest, src_manifest_hash = get_hdfs_manifest_from_path(src_manifest_path) | |
| src_config_hash = get_config_hash_from_manifest(src_manifest) | |
| src_config_path = src_config_dir + "/" + src_config_hash | |
| dest_config_path = dest_config_dir + "/" + src_config_hash | |
| src_layers = get_layer_hashes_from_manifest(src_manifest) | |
| src_layers_paths = [src_layers_dir + "/" + layer + ".sqsh" for layer in src_layers] | |
| dest_layers_paths = [dest_layers_dir + "/" + layer + ".sqsh" for layer in src_layers] | |
| logging.debug("Copying Manifest: %s", str(src_manifest_path)) | |
| logging.debug("Copying Layers: %s", str(src_layers_paths)) | |
| logging.debug("Copying Config: %s", str(src_config_hash)) | |
| if not does_hdfs_entry_exist(dest_layers_paths, raise_on_error=False): | |
| dest_layers_paths = [] | |
| for layer in src_layers: | |
| dest_layer_path = dest_layers_dir + "/" + layer + ".sqsh" | |
| src_layer_path = src_layers_dir + "/" + layer + ".sqsh" | |
| if not does_hdfs_entry_exist(dest_layer_path, raise_on_error=False): | |
| hdfs_cp(src_layer_path, dest_layer_path, force) | |
| dest_layers_paths.append(dest_layer_path) | |
| hdfs_setrep(replication, dest_layers_paths) | |
| hdfs_chmod("444", dest_layers_paths) | |
| if not does_hdfs_entry_exist(dest_config_path, raise_on_error=False): | |
| hdfs_cp(src_config_path, dest_config_dir, force) | |
| hdfs_setrep(replication, dest_config_path) | |
| hdfs_chmod("444", dest_config_path) | |
| if not does_hdfs_entry_exist(dest_manifest_path, raise_on_error=False): | |
| hdfs_cp(src_manifest_path, dest_manifest_dir, force) | |
| hdfs_setrep(replication, dest_manifest_path) | |
| hdfs_chmod("444", dest_manifest_path) | |
| for tag in tags: | |
| comment = None | |
| new_tags_and_comments = src_hash_to_tags.get(src_manifest_hash, None) | |
| if new_tags_and_comments: | |
| comment = ', '.join(map(str, new_tags_and_comments[1])) | |
| if comment is None: | |
| comment = image | |
| update_dicts(dest_hash_to_tags, dest_tag_to_hash, tag, src_manifest_hash, comment) | |
| write_local_image_tag_to_hash(local_dest_image_tag_to_hash, dest_hash_to_tags) | |
| atomic_upload_mv_to_hdfs(local_dest_image_tag_to_hash, dest_root + "/" + image_tag_to_hash, | |
| replication, | |
| dest_image_tag_to_hash_hash) | |
| finally: | |
| if working_dir: | |
| if os.path.isdir(working_dir): | |
| shutil.rmtree(working_dir) | |
| def query_tag(args): | |
| global HDFS_ROOT | |
| global HDFS_MANIFEST_DIR | |
| global HDFS_CONFIG_DIR | |
| global HDFS_LAYERS_DIR | |
| image_tag_to_hash = args.image_tag_to_hash | |
| tags = args.tags | |
| working_dir = None | |
| try: | |
| working_dir = get_working_dir(args.working_dir) | |
| local_image_tag_to_hash = os.path.join(working_dir, os.path.basename(image_tag_to_hash)) | |
| hash_to_tags, tag_to_hash, image_tag_to_hash_hash = populate_tag_dicts(image_tag_to_hash, | |
| local_image_tag_to_hash) | |
| logging.debug("hash_to_tags: %s", str(hash_to_tags)) | |
| logging.debug("tag_to_hash: %s", str(tag_to_hash)) | |
| for tag in tags: | |
| image_hash = tag_to_hash.get(tag, None) | |
| if not image_hash: | |
| logging.info("image hash mapping doesn't exist for tag %s", tag) | |
| continue | |
| manifest_path = HDFS_MANIFEST_DIR + "/" + image_hash | |
| if does_hdfs_entry_exist(manifest_path, raise_on_error=False): | |
| logging.debug("image manifest for %s exists: %s", tag, manifest_path) | |
| else: | |
| logging.info("Image manifest for %s doesn't exist: %s", tag, manifest_path) | |
| continue | |
| manifest, manifest_hash = get_hdfs_manifest_from_path(manifest_path) | |
| layers = get_layer_hashes_from_manifest(manifest, False) | |
| config_hash = get_config_hash_from_manifest(manifest) | |
| config_path = HDFS_CONFIG_DIR + "/" + config_hash | |
| layers_paths = [HDFS_LAYERS_DIR + "/" + layer + ".sqsh" for layer in layers] | |
| logging.info("Image info for '%s'", tag) | |
| logging.info(manifest_path) | |
| logging.info(config_path) | |
| for layer in layers_paths: | |
| logging.info(layer) | |
| finally: | |
| if working_dir: | |
| if os.path.isdir(working_dir): | |
| shutil.rmtree(working_dir) | |
| def list_tags(args): | |
| global HDFS_ROOT | |
| image_tag_to_hash = args.image_tag_to_hash | |
| hdfs_image_tag_to_hash = HDFS_ROOT + "/" + image_tag_to_hash | |
| hdfs_cat(hdfs_image_tag_to_hash, True, True, True) | |
| def bootstrap_setup(args): | |
| global HDFS_ROOT | |
| global HDFS_MANIFEST_DIR | |
| global HDFS_CONFIG_DIR | |
| global HDFS_LAYERS_DIR | |
| global HDFS_UNREF_DIR | |
| image_tag_to_hash = args.image_tag_to_hash | |
| hdfs_dirs = [HDFS_ROOT, HDFS_LAYERS_DIR, HDFS_CONFIG_DIR, HDFS_MANIFEST_DIR, HDFS_UNREF_DIR] | |
| image_tag_to_hash_path = HDFS_ROOT + "/" + image_tag_to_hash | |
| setup_squashfs_hdfs_dirs(hdfs_dirs, image_tag_to_hash_path) | |
| def cleanup_untagged_images(args): | |
| global HDFS_ROOT | |
| global HDFS_MANIFEST_DIR | |
| global HDFS_CONFIG_DIR | |
| global HDFS_LAYERS_DIR | |
| global HDFS_UNREF_DIR | |
| global DEAD_PERMS | |
| image_tag_to_hash = args.image_tag_to_hash | |
| working_dir = get_working_dir(args.working_dir) | |
| local_image_tag_to_hash = os.path.join(working_dir, os.path.basename(image_tag_to_hash)) | |
| try: | |
| hash_to_tags, tag_to_hash, image_tag_to_hash_hash = populate_tag_dicts(image_tag_to_hash, | |
| local_image_tag_to_hash) | |
| logging.debug("hash_to_tags: %s\n", hash_to_tags) | |
| logging.debug("tag_to_hash: %s\n", tag_to_hash) | |
| known_images = get_all_known_images() | |
| tagged_images = [image for image in known_images if image.manifest_stat.name in hash_to_tags.keys()] | |
| untagged_images = get_untagged_images(known_images, tagged_images) | |
| stale_images = get_stale_images(untagged_images) | |
| dead_images = get_dead_images(untagged_images) | |
| cleanup_handle_tagged_images(tagged_images) | |
| cleanup_handle_untagged_images(untagged_images) | |
| cleanup_handle_stale_images(stale_images) | |
| cleanup_handle_dead_images(dead_images, known_images) | |
| finally: | |
| if working_dir: | |
| if os.path.isdir(working_dir): | |
| shutil.rmtree(working_dir) | |
| class Filestat: | |
| def __init__(self, perms, mod_time, name): | |
| self.perms = perms | |
| self.mod_time = mod_time | |
| self.name = name | |
| def __repr__(self): | |
| return "Perms: {}, Mod time: {}, Name: {}\n".format(self.perms, self.mod_time, self.name) | |
| def __str__(self): | |
| return "Perms: {}, Mod time: {}, Name: {}\n".format(self.perms, self.mod_time, self.name) | |
| class Image: | |
| def __init__(self, manifest_stat, layers, config, unref_file_stat): | |
| self.manifest_stat = manifest_stat | |
| self.layers = layers | |
| self.config = config | |
| self.unref_file_stat = unref_file_stat | |
| def __repr__(self): | |
| return "Manifest: {}, Layers: {}, Config: {}, Unreferenced File: {}\n".format(self.manifest_stat, self.layers, | |
| self.config, self.unref_file_stat) | |
| def __str__(self): | |
| return "Manifest: {}, Layers: {}, Config: {}, Unreferenced File: {}\n".format(self.manifest_stat, self.layers, | |
| self.config, self.unref_file_stat) | |
| def get_all_known_images(): | |
| global HDFS_MANIFEST_DIR | |
| global HDFS_UNREF_DIR | |
| known_manifest_paths, err, returncode = hdfs_stat(HDFS_MANIFEST_DIR + "/*", "%a %Y %n", False, False, False) | |
| known_manifests = [image for image in known_manifest_paths.split("\n") if image is not "" or None] | |
| unref_manifest_paths, err, returncode = hdfs_stat(HDFS_UNREF_DIR + "/*", "%a %Y %n", False, False, False) | |
| logging.debug("unref_manifest_paths:\n%s", unref_manifest_paths) | |
| unref_manifests = [] | |
| if unref_manifest_paths: | |
| unref_manifests = [image for image in unref_manifest_paths.split("\n") if image is not "" or None] | |
| logging.debug("unref_manifests:\n%s", unref_manifests) | |
| unref_manifests_stats_dict = {} | |
| if unref_manifests: | |
| for unref_manifest in unref_manifests: | |
| unref_manifest_split = unref_manifest.split() | |
| unref_manifest_perms = unref_manifest_split[0] | |
| unref_manifest_mod_time = long(unref_manifest_split[1]) | |
| unref_manifest_name = unref_manifest_split[2] | |
| unref_manifest_stat = Filestat(unref_manifest_perms, unref_manifest_mod_time, unref_manifest_name) | |
| unref_manifests_stats_dict[unref_manifest_name] = unref_manifest_stat | |
| logging.debug("unref_manifests_stats_dict:\n%s", unref_manifests_stats_dict) | |
| known_manifests_names = [known_manifest.split()[2] for known_manifest in known_manifests] | |
| layers_and_configs = get_all_layers_and_configs(known_manifests_names) | |
| known_images = [] | |
| for manifest in known_manifests: | |
| manifest_split = manifest.split() | |
| manifest_perms = manifest_split[0] | |
| manifest_mod_time = long(manifest_split[1]) | |
| manifest_name = manifest_split[2] | |
| manifest_stat = Filestat(manifest_perms, manifest_mod_time, manifest_name) | |
| unref_image_stat = unref_manifests_stats_dict.get(manifest_name, None) | |
| layers = layers_and_configs[manifest_name][0] | |
| config = layers_and_configs[manifest_name][1] | |
| known_image = Image(manifest_stat, layers, config, unref_image_stat) | |
| known_images.append(known_image) | |
| return known_images | |
| def get_known_image_by_hash(image_hash, known_images): | |
| for image in known_images: | |
| if image_hash == image.manifest_stat.name: | |
| return image | |
| logging.debug("Couldn't find known image by hash:\n%s", image_hash) | |
| return None | |
| def get_all_layers_and_configs(manifest_names): | |
| global HDFS_MANIFEST_DIR | |
| manifests_tuples = get_hdfs_manifests_from_paths( | |
| [HDFS_MANIFEST_DIR + "/" + manifest_name for manifest_name in manifest_names]) | |
| layers_and_configs = {} | |
| for manifest_tuple in manifests_tuples: | |
| manifest = manifest_tuple[0] | |
| manifest_hash = manifest_tuple[1] | |
| layers = [] | |
| layers.extend(get_layer_hashes_from_manifest(manifest, False)) | |
| config = get_config_hash_from_manifest(manifest) | |
| layers_and_configs[manifest_hash] = (layers, config) | |
| logging.debug("layers for %s:\n%s", manifest_hash, layers) | |
| logging.debug("config for %s:\n%s", manifest_hash, config) | |
| return layers_and_configs | |
| def get_untagged_images(known_images, tagged_images): | |
| untagged_images = [] | |
| for image in known_images: | |
| if is_image_untagged(image, tagged_images): | |
| untagged_images.append(image) | |
| logging.debug("known_images:\n%s", known_images) | |
| logging.debug("tagged_images:\n%s", tagged_images) | |
| logging.debug("untagged_images:\n%s", untagged_images) | |
| return untagged_images | |
| def get_stale_images(untagged_images): | |
| stale_images = [image for image in untagged_images if is_image_stale(image)] | |
| logging.debug("stale_images:\n%s", stale_images) | |
| return stale_images | |
| def get_dead_images(untagged_images): | |
| dead_images = [image for image in untagged_images if is_image_dead(image)] | |
| logging.debug("dead_images:\n%s", dead_images) | |
| return dead_images | |
| def is_image_untagged(image, tagged_images): | |
| for tagged_image in tagged_images: | |
| if tagged_image.manifest_stat.name == image.manifest_stat.name: | |
| return False | |
| return True | |
| def is_image_stale(image): | |
| return does_image_have_unref_file(image) and not does_image_have_dead_perms(image) | |
| def is_image_dead(image): | |
| return does_image_have_unref_file(image) and does_image_have_dead_perms(image) | |
| def does_image_have_unref_file(image): | |
| return image.unref_file_stat != None | |
| def does_image_have_dead_perms(image): | |
| global DEAD_PERMS | |
| return image.manifest_stat.perms == DEAD_PERMS | |
| def is_mod_time_old(mod_time): | |
| global UNTAGGED_TRANSITION_SEC | |
| cutoff_time = long(time.time() * 1000) - UNTAGGED_TRANSITION_SEC * 1000 | |
| logging.debug("Mod time: %d, Cutoff time: %d", mod_time, cutoff_time) | |
| return mod_time < cutoff_time | |
| def cleanup_handle_tagged_images(tagged_images): | |
| # Remove unreferenced files if they exist | |
| if not tagged_images: | |
| return | |
| unref_remove_list = [] | |
| for image in tagged_images: | |
| if does_image_have_unref_file(image): | |
| unref_remove_list.append(image) | |
| remove_unref_files(unref_remove_list) | |
| def cleanup_handle_untagged_images(untagged_images): | |
| # Create unreferenced file | |
| if not untagged_images: | |
| return | |
| touch_list = [] | |
| for image in untagged_images: | |
| if not does_image_have_unref_file(image): | |
| touch_list.append(image) | |
| touch_unref_files(touch_list) | |
| def cleanup_handle_stale_images(stale_images): | |
| # Set blob permissions to 400 for old stale images | |
| if not stale_images: | |
| return | |
| make_unreadable_list = [] | |
| for image in stale_images: | |
| if is_mod_time_old(image.unref_file_stat.mod_time): | |
| make_unreadable_list.append(image) | |
| make_manifests_unreadable(make_unreadable_list) | |
| touch_unref_files(make_unreadable_list) | |
| def cleanup_handle_dead_images(dead_images, known_images): | |
| # Remove old dead images | |
| if not dead_images: | |
| return | |
| images_to_remove = [] | |
| for image in dead_images: | |
| if is_mod_time_old(image.unref_file_stat.mod_time): | |
| images_to_remove.append(image) | |
| remove_dead_images(images_to_remove, known_images) | |
| def make_manifests_unreadable(images): | |
| global HDFS_MANIFEST_DIR | |
| global DEAD_PERMS | |
| if not images: | |
| return | |
| manifest_file_paths = [HDFS_MANIFEST_DIR + "/" + image.unref_file_stat.name for image in images] | |
| logging.debug("Chmod %s manifest file:\n%s", DEAD_PERMS, manifest_file_paths) | |
| hdfs_chmod(DEAD_PERMS, manifest_file_paths) | |
| def touch_unref_files(images): | |
| global HDFS_UNREF_DIR | |
| if not images: | |
| return | |
| unref_file_paths = [HDFS_UNREF_DIR + "/" + image.manifest_stat.name for image in images] | |
| logging.debug("Touching unref file:\n%s", unref_file_paths) | |
| hdfs_touchz(unref_file_paths) | |
| def remove_unref_files(images): | |
| global HDFS_UNREF_DIR | |
| if not images: | |
| return | |
| unref_file_paths = [HDFS_UNREF_DIR + "/" + image.manifest_stat.name for image in images] | |
| logging.debug("Removing unref files:\n%s", unref_file_paths) | |
| hdfs_rm(unref_file_paths) | |
| def remove_dead_images(images_to_remove, known_images): | |
| if not images_to_remove: | |
| return | |
| logging.debug("Removing dead images:\n%s", images_to_remove) | |
| delete_list = get_delete_list_from_images_to_remove(images_to_remove, known_images) | |
| if delete_list: | |
| hdfs_rm(delete_list) | |
| def create_parsers(): | |
| parser = argparse.ArgumentParser() | |
| add_parser_default_arguments(parser) | |
| subparsers = parser.add_subparsers(help='sub help', dest='sub_command') | |
| parse_pull_build_push_update = subparsers.add_parser('pull-build-push-update', | |
| help='Pull an image, build its squashfs' | |
| + ' layers, push it to hdfs, and' | |
| + ' atomically update the' | |
| + ' image-tag-to-hash file') | |
| parse_pull_build_push_update.set_defaults(func=pull_build_push_update) | |
| add_parser_default_arguments(parse_pull_build_push_update) | |
| parse_pull_build_push_update.add_argument("images_and_tags", nargs="+", | |
| help="Image and tag argument (can specify multiple)") | |
| parse_pull_build = subparsers.add_parser('pull-build', | |
| help='Pull an image and build its squashfs layers') | |
| parse_pull_build.set_defaults(func=pull_build) | |
| add_parser_default_arguments(parse_pull_build) | |
| parse_pull_build.add_argument("images_and_tags", nargs="+", | |
| help="Image and tag argument (can specify multiple)") | |
| parse_push_update = subparsers.add_parser('push-update', | |
| help='Push the squashfs layers to hdfs and update' | |
| + ' the image-tag-to-hash file') | |
| parse_push_update.set_defaults(func=push_update) | |
| add_parser_default_arguments(parse_push_update) | |
| parse_push_update.add_argument("images_and_tags", nargs="+", | |
| help="Image and tag argument (can specify multiple)") | |
| parse_remove_image = subparsers.add_parser('remove-image', | |
| help='Remove an image (manifest, config, layers)' | |
| + ' from hdfs based on its tag or manifest hash') | |
| parse_remove_image.set_defaults(func=remove_image) | |
| add_parser_default_arguments(parse_remove_image) | |
| parse_remove_image.add_argument("images_or_tags", nargs="+", | |
| help="Image or tag argument (can specify multiple)") | |
| parse_remove_tag = subparsers.add_parser('remove-tag', | |
| help='Remove an image to tag mapping in the' | |
| + ' image-tag-to-hash file') | |
| parse_remove_tag.set_defaults(func=add_remove_tag) | |
| add_parser_default_arguments(parse_remove_tag) | |
| parse_remove_tag.add_argument("images_and_tags", nargs="+", | |
| help="Image and tag argument (can specify multiple)") | |
| parse_add_tag = subparsers.add_parser('add-tag', | |
| help='Add an image to tag mapping in the' | |
| + ' image-tag-to-hash file') | |
| parse_add_tag.set_defaults(func=add_remove_tag) | |
| add_parser_default_arguments(parse_add_tag) | |
| parse_add_tag.add_argument("images_and_tags", nargs="+", | |
| help="Image and tag argument (can specify multiple)") | |
| parse_copy_update = subparsers.add_parser('copy-update', | |
| help='Copy an image from hdfs in one cluster to' | |
| + ' another and then update the' | |
| + ' image-tag-to-hash file') | |
| parse_copy_update.set_defaults(func=copy_update) | |
| add_parser_default_arguments(parse_copy_update) | |
| parse_copy_update.add_argument("src_root", | |
| help="HDFS path for source root directory") | |
| parse_copy_update.add_argument("dest_root", | |
| help="HDFS path for destination root directory") | |
| parse_copy_update.add_argument("images_and_tags", nargs="+", | |
| help="Image and tag argument (can specify multiple)") | |
| parse_query_tag = subparsers.add_parser('query-tag', | |
| help='Get the manifest, config, and layers' | |
| + ' associated with a tag') | |
| parse_query_tag.set_defaults(func=query_tag) | |
| add_parser_default_arguments(parse_query_tag) | |
| parse_query_tag.add_argument("tags", nargs="+", | |
| help="Image or tag argument (can specify multiple)") | |
| parse_list_tags = subparsers.add_parser('list-tags', | |
| help='List all tags in image-tag-to-hash file') | |
| parse_list_tags.set_defaults(func=list_tags) | |
| add_parser_default_arguments(parse_list_tags) | |
| parse_bootstrap_setup = subparsers.add_parser('bootstrap', help='Bootstrap setup of required HDFS directories') | |
| parse_bootstrap_setup.set_defaults(func=bootstrap_setup) | |
| add_parser_default_arguments(parse_bootstrap_setup) | |
| parse_cleanup_untagged_images = subparsers.add_parser('cleanup', | |
| help='Cleanup untagged images in HDFS') | |
| parse_cleanup_untagged_images.set_defaults(func=cleanup_untagged_images) | |
| add_parser_default_arguments(parse_cleanup_untagged_images) | |
| return parser | |
| def add_parser_default_arguments(parser): | |
| parser.add_argument("--working-dir", type=str, dest='working_dir', default="dts-work-dir", | |
| help="Name of working directory") | |
| parser.add_argument("--skopeo-format", type=str, dest='skopeo_format', | |
| default='dir', help="Output format for skopeo copy") | |
| parser.add_argument("--pull-format", type=str, dest='pull_format', | |
| default='docker', help="Pull format for skopeo") | |
| parser.add_argument("-l", "--log", type=str, dest='LOG_LEVEL', | |
| default="INFO", help="Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL)") | |
| parser.add_argument("--hdfs-root", type=str, dest='HDFS_ROOT', | |
| default='/runc-root', help="The root directory in HDFS for all of the" | |
| + "squashfs images") | |
| parser.add_argument("--image-tag-to-hash", type=str, | |
| dest='image_tag_to_hash', default='image-tag-to-hash', | |
| help="image-tag-to-hash filename in hdfs") | |
| parser.add_argument("-r", "--replication", type=int, dest='replication', | |
| default=3, help="Replication factor for all files uploaded to HDFS") | |
| parser.add_argument("--hadoop-prefix", type=str, dest='hadoop_prefix', | |
| default=os.environ.get('HADOOP_PREFIX'), | |
| help="hadoop prefix value for environment") | |
| parser.add_argument("-f", "--force", dest='force', | |
| action="store_true", default=False, help="Force overwrites in HDFS") | |
| parser.add_argument("--check-magic-file", dest='check_magic_file', | |
| action="store_true", default=False, help="Check for a specific magic file" | |
| + "in the image before uploading") | |
| parser.add_argument("--magic-file", type=str, dest='magic_file', | |
| default='etc/dockerfile-version', help="The magic file to check for" | |
| + "in the image") | |
| parser.add_argument("--max-layers", type=int, dest='MAX_IMAGE_LAYERS', | |
| default=37, help="Maximum number of layers an image is allowed to have") | |
| parser.add_argument("--max-size", type=int, dest='MAX_IMAGE_SIZE', | |
| default=10 * 1024 * 1024 * 1024, help="Maximum size an image is allowed to be") | |
| parser.add_argument("--untagged-transition-sec", type=long, dest='UNTAGGED_TRANSITION_SEC', | |
| default=7 * 24 * 60 * 60, help="Time that untagged images will spend in each state" | |
| + "before moving to the next one") | |
| parser.add_argument("--dead-perms", type=str, dest='DEAD_PERMS', | |
| default="400", help="Permissions to set for manifests that are untagged " | |
| + "before they are removed") | |
| parser.add_argument("-b", "--bootstrap", dest='bootstrap', | |
| action="store_true", default=False, help="Bootstrap setup" | |
| + " of required HDFS directories") | |
| return parser | |
| def check_dependencies(): | |
| global HADOOP_BIN_DIR | |
| try: | |
| command = [HADOOP_BIN_DIR + "/hadoop", "-h"] | |
| shell_command(command, False, False, True) | |
| except Exception as ex: | |
| logging.error("Could not find hadoop. Make sure HADOOP_PREFIX " + | |
| "is set correctly either in your environment or on the command line " + | |
| "via --hadoop-prefix" + | |
| ", exception is " + str(ex)) | |
| return 1 | |
| try: | |
| command = ["skopeo", "-v"] | |
| shell_command(command, False, False, True) | |
| except Exception as _: | |
| logging.error("Could not find skopeo. Make sure it is installed and present on the PATH") | |
| return 1 | |
| try: | |
| command = ["/usr/sbin/mksquashfs", "-version"] | |
| shell_command(command, False, False, True) | |
| except Exception as _: | |
| logging.error("Could not find /usr/sbin/mksquashfs. Make sure squashfs-tools is installed " + | |
| "and /usr/sbin/mksquashfs is present on the the PATH") | |
| return 1 | |
| try: | |
| command = ["tar", "--version"] | |
| shell_command(command, False, False, True) | |
| except Exception as _: | |
| logging.error("Could not find tar. Make sure it is installed and present on the PATH") | |
| return 1 | |
| try: | |
| command = ["setfattr", "--version"] | |
| shell_command(command, False, False, True) | |
| except Exception: | |
| logging.error("Could not find setfattr . Make sure it is installed and present on the PATH") | |
| return 1 | |
| return 0 | |
| def main(): | |
| global LOG_LEVEL | |
| global HADOOP_PREFIX | |
| global HADOOP_BIN_DIR | |
| global HDFS_ROOT | |
| global HDFS_MANIFEST_DIR | |
| global HDFS_CONFIG_DIR | |
| global HDFS_LAYERS_DIR | |
| global HDFS_UNREF_DIR | |
| global MAX_IMAGE_LAYERS | |
| global MAX_IMAGE_SIZE | |
| global UNTAGGED_TRANSITION_SEC | |
| global ARG_MAX | |
| global DEAD_PERMS | |
| if os.geteuid() != 0: | |
| logging.error("Script must be run as root") | |
| return | |
| parser = create_parsers() | |
| args, extra = parser.parse_known_args() | |
| if extra: | |
| raise Exception(f"Extra unknown arguments given: {extra}") | |
| ARG_MAX = os.sysconf("SC_ARG_MAX") | |
| HDFS_ROOT = args.HDFS_ROOT | |
| HDFS_MANIFEST_DIR = HDFS_ROOT + "/manifests" | |
| HDFS_CONFIG_DIR = HDFS_ROOT + "/config" | |
| HDFS_LAYERS_DIR = HDFS_ROOT + "/layers" | |
| HDFS_UNREF_DIR = HDFS_ROOT + "/unreferenced" | |
| MAX_IMAGE_LAYERS = args.MAX_IMAGE_LAYERS | |
| MAX_IMAGE_SIZE = args.MAX_IMAGE_SIZE | |
| UNTAGGED_TRANSITION_SEC = args.UNTAGGED_TRANSITION_SEC | |
| DEAD_PERMS = args.DEAD_PERMS | |
| LOG_LEVEL = args.LOG_LEVEL.upper() | |
| image_tag_to_hash = args.image_tag_to_hash | |
| numeric_level = getattr(logging, LOG_LEVEL, None) | |
| if not isinstance(numeric_level, int): | |
| logging.error("Invalid log level: %s", LOG_LEVEL) | |
| return | |
| logging.basicConfig(format="%(levelname)s: %(message)s", level=numeric_level) | |
| if args.hadoop_prefix is None: | |
| logging.error("Hadoop prefix is not set. You may set it either " + | |
| "in your environment or via --hadoop-prefix") | |
| return | |
| HADOOP_PREFIX = args.hadoop_prefix | |
| HADOOP_BIN_DIR = HADOOP_PREFIX + "/bin" | |
| if check_dependencies(): | |
| return | |
| if "/" in image_tag_to_hash: | |
| logging.error("image-tag-to-hash cannot contain a /") | |
| return | |
| logging.debug("args: %s", str(args)) | |
| logging.debug("extra: %s", str(extra)) | |
| logging.debug("image-tag-to-hash: %s", image_tag_to_hash) | |
| logging.debug("LOG_LEVEL: %s", LOG_LEVEL) | |
| logging.debug("HADOOP_BIN_DIR: %s", str(HADOOP_BIN_DIR)) | |
| args.func(args) | |
| if __name__ == "__main__": | |
| main() |