path: root/crosperf/download_images.py

# -*- coding: utf-8 -*-
# Copyright 2014-2015 The ChromiumOS Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Download images from Cloud Storage."""


import ast
import os

from cros_utils import command_executer
import test_flag


GS_UTIL = "src/chromium/depot_tools/gsutil.py"


class MissingImage(Exception):
    """Raised when the requested image does not exist in gs://"""


class MissingFile(Exception):
    """Raised when the requested file does not exist in gs://"""


class RunCommandExceptionHandler(object):
    """Handle Exceptions from calls to RunCommand"""

    def __init__(self, logger_to_use, log_level, cmd_exec, command):
        self.logger = logger_to_use
        self.log_level = log_level
        self.ce = cmd_exec
        self.cleanup_command = command

    def HandleException(self, _, e):
        # Exception handler, Run specified command
        if self.log_level != "verbose" and self.cleanup_command is not None:
            self.logger.LogOutput("CMD: %s" % self.cleanup_command)
        if self.cleanup_command is not None:
            _ = self.ce.RunCommand(self.cleanup_command)
        # Raise exception again
        raise e


class ImageDownloader(object):
    """Download images from Cloud Storage."""

    def __init__(self, logger_to_use=None, log_level="verbose", cmd_exec=None):
        self._logger = logger_to_use
        self.log_level = log_level
        self._ce = cmd_exec or command_executer.GetCommandExecuter(
            self._logger, log_level=self.log_level
        )

    def GetBuildID(self, chromeos_root, xbuddy_label):
        # Get the translation of the xbuddy_label into the real Google Storage
        # image name.
        command = (
            "cd /mnt/host/source/src/third_party/toolchain-utils/crosperf; "
            "./translate_xbuddy.py '%s'" % xbuddy_label
        )
        _, build_id_tuple_str, _ = self._ce.ChrootRunCommandWOutput(
            chromeos_root, command
        )
        if not build_id_tuple_str:
            raise MissingImage("Unable to find image for '%s'" % xbuddy_label)

        build_id_tuple = ast.literal_eval(build_id_tuple_str)
        build_id = build_id_tuple[0]

        return build_id

    def DownloadImage(self, chromeos_root, build_id, image_name):
        if self.log_level == "average":
            self._logger.LogOutput(
                "Preparing to download %s image to local "
                "directory." % build_id
            )

        # Make sure the directory for downloading the image exists.
        download_path = os.path.join(chromeos_root, "chroot/tmp", build_id)
        image_path = os.path.join(download_path, "chromiumos_test_image.bin")
        if not os.path.exists(download_path):
            os.makedirs(download_path)

        # Check to see if the image has already been downloaded.  If not,
        # download the image.
        if not os.path.exists(image_path):
            gsutil_cmd = os.path.join(chromeos_root, GS_UTIL)
            command = "%s cp %s %s" % (gsutil_cmd, image_name, download_path)

            if self.log_level != "verbose":
                self._logger.LogOutput("CMD: %s" % command)
            status = self._ce.RunCommand(command)
            downloaded_image_name = os.path.join(
                download_path, "chromiumos_test_image.tar.xz"
            )
            if status != 0 or not os.path.exists(downloaded_image_name):
                raise MissingImage(
                    "Cannot download image: %s." % downloaded_image_name
                )

        return image_path

    def UncompressImage(self, chromeos_root, build_id):
        # Check to see if the file has already been uncompresssed, etc.
        if os.path.exists(
            os.path.join(
                chromeos_root,
                "chroot/tmp",
                build_id,
                "chromiumos_test_image.bin",
            )
        ):
            return

        # Uncompress and untar the downloaded image.
        download_path = os.path.join(chromeos_root, "chroot/tmp", build_id)
        command = (
            "cd %s ; tar -Jxf chromiumos_test_image.tar.xz " % download_path
        )
        # Cleanup command for exception handler
        clean_cmd = "cd %s ; rm -f chromiumos_test_image.bin " % download_path
        exception_handler = RunCommandExceptionHandler(
            self._logger, self.log_level, self._ce, clean_cmd
        )
        if self.log_level != "verbose":
            self._logger.LogOutput("CMD: %s" % command)
            print(
                "(Uncompressing and un-tarring may take a couple of minutes..."
                "please be patient.)"
            )
        retval = self._ce.RunCommand(
            command, except_handler=exception_handler.HandleException
        )
        if retval != 0:
            if self.log_level != "verbose":
                self._logger.LogOutput("CMD: %s" % clean_cmd)
                print("(Removing file chromiumos_test_image.bin.)")
            # Remove partially uncompressed file
            _ = self._ce.RunCommand(clean_cmd)
            # Raise exception for failure to uncompress
            raise MissingImage("Cannot uncompress image: %s." % build_id)

        # Remove compressed image
        command = "cd %s ; rm -f chromiumos_test_image.tar.xz; " % download_path
        if self.log_level != "verbose":
            self._logger.LogOutput("CMD: %s" % command)
            print("(Removing file chromiumos_test_image.tar.xz.)")
        # try removing file, its ok to have an error, print if encountered
        retval = self._ce.RunCommand(command)
        if retval != 0:
            print(
                "(Warning: Could not remove file chromiumos_test_image.tar.xz .)"
            )

    def DownloadSingleFile(self, chromeos_root, build_id, package_file_name):
        # Verify if package files exist
        status = 0
        gs_package_name = "gs://chromeos-image-archive/%s/%s" % (
            build_id,
            package_file_name,
        )
        gsutil_cmd = os.path.join(chromeos_root, GS_UTIL)
        if not test_flag.GetTestMode():
            cmd = "%s ls %s" % (gsutil_cmd, gs_package_name)
            status = self._ce.RunCommand(cmd)
        if status != 0:
            raise MissingFile(
                "Cannot find package file: %s." % package_file_name
            )

        if self.log_level == "average":
            self._logger.LogOutput(
                "Preparing to download %s package to local "
                "directory." % package_file_name
            )

        # Make sure the directory for downloading the package exists.
        download_path = os.path.join(chromeos_root, "chroot/tmp", build_id)
        package_path = os.path.join(download_path, package_file_name)
        if not os.path.exists(download_path):
            os.makedirs(download_path)

        # Check to see if the package file has already been downloaded.  If not,
        # download it.
        if not os.path.exists(package_path):
            command = "%s cp %s %s" % (
                gsutil_cmd,
                gs_package_name,
                download_path,
            )

            if self.log_level != "verbose":
                self._logger.LogOutput("CMD: %s" % command)
            status = self._ce.RunCommand(command)
            if status != 0 or not os.path.exists(package_path):
                raise MissingFile(
                    "Cannot download package: %s ." % package_path
                )

    def UncompressSingleFile(
        self, chromeos_root, build_id, package_file_name, uncompress_cmd
    ):
        # Uncompress file
        download_path = os.path.join(chromeos_root, "chroot/tmp", build_id)
        command = "cd %s ; %s %s" % (
            download_path,
            uncompress_cmd,
            package_file_name,
        )

        if self.log_level != "verbose":
            self._logger.LogOutput("CMD: %s" % command)
            print("(Uncompressing file %s .)" % package_file_name)
        retval = self._ce.RunCommand(command)
        if retval != 0:
            raise MissingFile("Cannot uncompress file: %s." % package_file_name)
        # Remove uncompressed downloaded file
        command = "cd %s ; rm -f %s" % (download_path, package_file_name)
        if self.log_level != "verbose":
            self._logger.LogOutput("CMD: %s" % command)
            print("(Removing processed file %s .)" % package_file_name)
        # try removing file, its ok to have an error, print if encountered
        retval = self._ce.RunCommand(command)
        if retval != 0:
            print("(Warning: Could not remove file %s .)" % package_file_name)

    def VerifyFileExists(self, chromeos_root, build_id, package_file):
        # Quickly verify if the files are there
        status = 0
        gs_package_name = "gs://chromeos-image-archive/%s/%s" % (
            build_id,
            package_file,
        )
        gsutil_cmd = os.path.join(chromeos_root, GS_UTIL)
        if not test_flag.GetTestMode():
            cmd = "%s ls %s" % (gsutil_cmd, gs_package_name)
            if self.log_level != "verbose":
                self._logger.LogOutput("CMD: %s" % cmd)
            status = self._ce.RunCommand(cmd)
            if status != 0:
                print("(Warning: Could not find file %s )" % gs_package_name)
                return 1
        # Package exists on server
        return 0

    def DownloadAutotestFiles(self, chromeos_root, build_id):
        # Download autest package files (3 files)
        autotest_packages_name = "autotest_packages.tar"
        autotest_server_package_name = "autotest_server_package.tar.bz2"
        autotest_control_files_name = "control_files.tar"

        download_path = os.path.join(chromeos_root, "chroot/tmp", build_id)
        # Autotest directory relative path wrt chroot
        autotest_rel_path = os.path.join("/tmp", build_id, "autotest_files")
        # Absolute Path to download files
        autotest_path = os.path.join(
            chromeos_root, "chroot/tmp", build_id, "autotest_files"
        )

        if not os.path.exists(autotest_path):
            # Quickly verify if the files are present on server
            # If not, just exit with warning
            status = self.VerifyFileExists(
                chromeos_root, build_id, autotest_packages_name
            )
            if status != 0:
                default_autotest_dir = (
                    "/mnt/host/source/src/third_party/autotest/files"
                )
                print(
                    "(Warning: Could not find autotest packages .)\n"
                    "(Warning: Defaulting autotest path to %s ."
                    % default_autotest_dir
                )
                return default_autotest_dir

            # Files exist on server, download and uncompress them
            self.DownloadSingleFile(
                chromeos_root, build_id, autotest_packages_name
            )
            self.DownloadSingleFile(
                chromeos_root, build_id, autotest_server_package_name
            )
            self.DownloadSingleFile(
                chromeos_root, build_id, autotest_control_files_name
            )

            self.UncompressSingleFile(
                chromeos_root, build_id, autotest_packages_name, "tar -xf "
            )
            self.UncompressSingleFile(
                chromeos_root,
                build_id,
                autotest_server_package_name,
                "tar -jxf ",
            )
            self.UncompressSingleFile(
                chromeos_root, build_id, autotest_control_files_name, "tar -xf "
            )
            # Rename created autotest directory to autotest_files
            command = "cd %s ; mv autotest autotest_files" % download_path
            if self.log_level != "verbose":
                self._logger.LogOutput("CMD: %s" % command)
                print("(Moving downloaded autotest files to autotest_files)")
            retval = self._ce.RunCommand(command)
            if retval != 0:
                raise MissingFile("Could not create directory autotest_files")

        return autotest_rel_path

    def DownloadDebugFile(self, chromeos_root, build_id):
        # Download autest package files (3 files)
        debug_archive_name = "debug.tgz"

        download_path = os.path.join(chromeos_root, "chroot/tmp", build_id)
        # Debug directory relative path wrt chroot
        debug_rel_path = os.path.join("/tmp", build_id, "debug_files")
        # Debug path to download files
        debug_path = os.path.join(
            chromeos_root, "chroot/tmp", build_id, "debug_files"
        )

        if not os.path.exists(debug_path):
            # Quickly verify if the file is present on server
            # If not, just exit with warning
            status = self.VerifyFileExists(
                chromeos_root, build_id, debug_archive_name
            )
            if status != 0:
                self._logger.LogOutput(
                    "WARNING: Could not find debug archive on gs"
                )
                return ""

            # File exists on server, download and uncompress it
            self.DownloadSingleFile(chromeos_root, build_id, debug_archive_name)

            self.UncompressSingleFile(
                chromeos_root, build_id, debug_archive_name, "tar -xf "
            )
            # Extract and move debug files into the proper location.
            debug_dir = "debug_files/usr/lib"
            command = "cd %s ; mkdir -p %s; mv debug %s" % (
                download_path,
                debug_dir,
                debug_dir,
            )
            if self.log_level != "verbose":
                self._logger.LogOutput("CMD: %s" % command)
                print("Moving downloaded debug files to %s" % debug_dir)
            retval = self._ce.RunCommand(command)
            if retval != 0:
                raise MissingFile(
                    "Could not create directory %s"
                    % os.path.join(debug_dir, "debug")
                )

        return debug_rel_path

    def Run(
        self,
        chromeos_root,
        xbuddy_label,
        autotest_path,
        debug_path,
        download_debug,
    ):
        build_id = self.GetBuildID(chromeos_root, xbuddy_label)
        image_name = (
            "gs://chromeos-image-archive/%s/chromiumos_test_image.tar.xz"
            % build_id
        )

        # Verify that image exists for build_id, before attempting to
        # download it.
        status = 0
        if not test_flag.GetTestMode():
            gsutil_cmd = os.path.join(chromeos_root, GS_UTIL)
            cmd = "%s ls %s" % (gsutil_cmd, image_name)
            status = self._ce.RunCommand(cmd)
        if status != 0:
            raise MissingImage("Cannot find official image: %s." % image_name)

        image_path = self.DownloadImage(chromeos_root, build_id, image_name)
        self.UncompressImage(chromeos_root, build_id)

        if self.log_level != "quiet":
            self._logger.LogOutput("Using image from %s." % image_path)

        if autotest_path == "":
            autotest_path = self.DownloadAutotestFiles(chromeos_root, build_id)

        if debug_path == "" and download_debug:
            debug_path = self.DownloadDebugFile(chromeos_root, build_id)

        return image_path, autotest_path, debug_path