downloader.py

#!/usr/bin/python
#
# Copyright (c) 2012 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

import Queue
import os
import shutil
import tempfile
import threading

import common_util
import log_util


class Downloader(log_util.Loggable):
  """Download images to the devsever.

  Given a URL to a build on the archive server:

    - Determine if the build already exists.
    - Download and extract the build to a staging directory.
    - Package autotest tests.
    - Install components to static dir.
  """

  # This filename must be kept in sync with clean_staged_images.py
  _TIMESTAMP_FILENAME = 'staged.timestamp'

  def __init__(self, static_dir):
    self._static_dir = static_dir
    self._build_dir = None
    self._staging_dir = None
    self._status_queue = Queue.Queue(maxsize=1)
    self._lock_tag = None

  @staticmethod
  def ParseUrl(archive_url):
    """Parse archive_url into rel_path and short_build
    e.g. gs://chromeos-image-archive/{rel_path}/{short_build}

    @param archive_url: a URL at which build artifacts are archived.
    @return a tuple of (build relative path, short build name)
    """
    # The archive_url is of the form gs://server/[some_path/target]/...]/build
    # This function discards 'gs://server/' and extracts the [some_path/target]
    # as rel_path and the build as short_build.
    sub_url = archive_url.partition('://')[2]
    split_sub_url = sub_url.split('/')
    rel_path = '/'.join(split_sub_url[1:-1])
    short_build = split_sub_url[-1]
    return rel_path, short_build

  @staticmethod
  def GenerateLockTag(rel_path, short_build):
    """Generate a name for a lock scoped to this rel_path/build pair.

    @param rel_path: the relative path for the build.
    @param short_build: short build name
    @return a name to use with AcquireLock that will scope the lock.
    """
    return '/'.join([rel_path, short_build])

  @staticmethod
  def _TouchTimestampForStaged(directory_path):
    file_name = os.path.join(directory_path, Downloader._TIMESTAMP_FILENAME)
    # Easiest python version of |touch file_name|
    with file(file_name, 'a'):
      os.utime(file_name, None)

  @staticmethod
  def BuildStaged(archive_url, static_dir):
    """Returns True if the build is already staged."""
    rel_path, short_build = Downloader.ParseUrl(archive_url)
    sub_directory = Downloader.GenerateLockTag(rel_path, short_build)
    directory_path = os.path.join(static_dir, sub_directory)
    exists = os.path.isdir(directory_path)
    # If the build exists, then touch the timestamp to tell
    # clean_stages_images.py that we're using this build.
    if exists:
      Downloader._TouchTimestampForStaged(directory_path)
    return exists

  def Download(self, archive_url, background=False):
    """Downloads the given build artifacts defined by the |archive_url|.

    If background is set to True, will return back early before all artifacts
    have been downloaded. The artifacts that can be backgrounded are all those
    that are not set as synchronous.

    TODO: refactor this into a common Download method, once unit tests are
    fixed up to make iterating on the code easier.
    """
    # Parse archive_url into rel_path (contains the build target) and
    # short_build.
    # e.g. gs://chromeos-image-archive/{rel_path}/{short_build}
    rel_path, short_build = self.ParseUrl(archive_url)
    # This should never happen. The Devserver should only try to call this
    # method if no previous downloads have been staged for this archive_url.
    assert not Downloader.BuildStaged(archive_url, self._static_dir)
    # Bind build_dir and staging_dir here so we can tell if we need to do any
    # cleanup after an exception occurs before build_dir is set.
    self._lock_tag = self.GenerateLockTag(rel_path, short_build)
    try:
      # Create Dev Server directory for this build and tell other Downloader
      # instances we have processed this build. Note that during normal
      # execution, this lock is only released in the actual downloading
      # procedure called below.
      self._build_dir = common_util.AcquireLock(
          static_dir=self._static_dir, tag=self._lock_tag)

      # Replace '/' with '_' in rel_path because it may contain multiple levels
      # which would not be qualified as part of the suffix.
      self._staging_dir = tempfile.mkdtemp(suffix='_'.join(
          [rel_path.replace('/', '_'), short_build]))
      Downloader._TouchTimestampForStaged(self._build_dir)
      self._Log('Gathering download requirements %s' % archive_url)
      artifacts = self.GatherArtifactDownloads(
          self._staging_dir, archive_url, self._build_dir, short_build)
      common_util.PrepareBuildDirectory(self._build_dir)

      self._Log('Downloading foreground artifacts from %s' % archive_url)
      background_artifacts = []
      for artifact in artifacts:
        if artifact.Synchronous():
          artifact.Download()
          artifact.Stage()
        else:
          background_artifacts.append(artifact)

      if background:
        self._DownloadArtifactsInBackground(background_artifacts)
      else:
        self._DownloadArtifactsSerially(background_artifacts)

    except Exception, e:
      # Release processing lock, which will remove build components directory
      # so future runs can retry.
      if self._build_dir:
        common_util.ReleaseLock(static_dir=self._static_dir, tag=self._lock_tag,
                                destroy=True)

      self._status_queue.put(e)
      self._Cleanup()
      raise
    return 'Success'

  def _Cleanup(self):
    """Cleans up the staging dir for this downloader instanfce."""
    if self._staging_dir:
      self._Log('Cleaning up staging directory %s' % self._staging_dir)
      shutil.rmtree(self._staging_dir)

    self._staging_dir = None

  def _DownloadArtifactsSerially(self, artifacts):
    """Simple function to download all the given artifacts serially."""
    self._Log('Downloading artifacts serially.')
    try:
      for artifact in artifacts:
        artifact.Download()
        artifact.Stage()
    except Exception, e:
      self._status_queue.put(e)

      # Release processing lock, which will remove build components directory
      # so future runs can retry.
      if self._build_dir:
        common_util.ReleaseLock(static_dir=self._static_dir, tag=self._lock_tag,
                                destroy=True)
    else:
      # Release processing lock, keeping directory intact.
      if self._build_dir:
        common_util.ReleaseLock(static_dir=self._static_dir, tag=self._lock_tag)
      self._status_queue.put('Success')
    finally:
      self._Cleanup()

  def _DownloadArtifactsInBackground(self, artifacts):
    """Downloads |artifacts| in the background and signals when complete."""
    self._Log('Invoking background download of artifacts')
    thread = threading.Thread(target=self._DownloadArtifactsSerially,
                              args=(artifacts,))
    thread.start()

  def GatherArtifactDownloads(self, main_staging_dir, archive_url, build_dir,
                              short_build):
    """Wrapper around common_util.GatherArtifactDownloads().

    The wrapper allows mocking and overriding in derived classes.
    """
    return common_util.GatherArtifactDownloads(
        main_staging_dir, archive_url, build_dir, short_build)

  def GetStatusOfBackgroundDownloads(self):
    """Returns the status of the background downloads.

    This commands returns the status of the background downloads and blocks
    until a status is returned.
    """
    status = self._status_queue.get()
    # In case anyone else is calling.
    self._status_queue.put(status)
    # If someone is curious about the status of a build, then we should
    # probably keep it around for a bit longer.
    if self._build_dir and os.path.exists(self._build_dir):
      Downloader._TouchTimestampForStaged(self._build_dir)
    # It's possible we received an exception, if so, re-raise it here.
    if isinstance(status, Exception):
      raise status

    return status


class SymbolDownloader(Downloader):
  """Download and stage debug symbols for a build on the devsever.

  Given a URL to a build on the archive server:

    - Determine if the build already exists.
    - Download and extract the debug symbols to a staging directory.
    - Install symbols to static dir.
  """

  _DONE_FLAG = 'done'

  @staticmethod
  def GenerateLockTag(rel_path, short_build):
    return '/'.join([rel_path, short_build, 'symbols'])

  def Download(self, archive_url, _background=False):
    """Downloads debug symbols for the build defined by the |archive_url|.

    The symbols will be downloaded synchronously
    """
    # Parse archive_url into rel_path (contains the build target) and
    # short_build.
    # e.g. gs://chromeos-image-archive/{rel_path}/{short_build}
    rel_path, short_build = self.ParseUrl(archive_url)

    # Bind build_dir and staging_dir here so we can tell if we need to do any
    # cleanup after an exception occurs before build_dir is set.
    self._lock_tag = self.GenerateLockTag(rel_path, short_build)
    if self.SymbolsStaged(archive_url, self._static_dir):
      self._Log('Symbols for build %s have already been staged.' %
                self._lock_tag)
      return 'Success'

    try:
      # Create Dev Server directory for this build and tell other Downloader
      # instances we have processed this build.
      self._build_dir = common_util.AcquireLock(
          static_dir=self._static_dir, tag=self._lock_tag)

      # Replace '/' with '_' in rel_path because it may contain multiple levels
      # which would not be qualified as part of the suffix.
      self._staging_dir = tempfile.mkdtemp(suffix='_'.join(
          [rel_path.replace('/', '_'), short_build]))
      self._Log('Downloading debug symbols from %s' % archive_url)

      [symbol_artifact] = self.GatherArtifactDownloads(
          self._staging_dir, archive_url, self._static_dir)
      symbol_artifact.Download()
      symbol_artifact.Stage()
      self.MarkSymbolsStaged()

    except Exception:
      # Release processing "lock", which will indicate to future runs that we
      # did not succeed, and so they should try again.
      if self._build_dir:
        common_util.ReleaseLock(static_dir=self._static_dir, tag=self._lock_tag,
                                destroy=True)

      raise
    else:
      # Release processing "lock", keeping directory intact.
      if self._build_dir:
        common_util.ReleaseLock(static_dir=self._static_dir, tag=self._lock_tag)
    finally:
      self._Cleanup()

    return 'Success'

  def GatherArtifactDownloads(self, temp_download_dir, archive_url, static_dir,
                              short_build=None):
    """Call SymbolDownloader-appropriate artifact gathering method.

    @param temp_download_dir: the tempdir into which we're downloading artifacts
                              prior to staging them.
    @param archive_url: the google storage url of the bucket where the debug
                        symbols for the desired build are stored.
    @param staging_dir: the dir into which to stage the symbols
    @param short_build: (ignored)

    @return an iterable of one DebugTarballBuildArtifact pointing to the right
            debug symbols.  This is an iterable so that it's similar to
            GatherArtifactDownloads.  Also, it's possible that someday we might
            have more than one.
    """
    return common_util.GatherSymbolArtifactDownloads(
        temp_download_dir, archive_url, static_dir)

  def MarkSymbolsStaged(self):
    """Puts a flag file on disk to signal that symbols are staged."""
    with open(os.path.join(self._build_dir, self._DONE_FLAG), 'w') as flag:
      flag.write(self._DONE_FLAG)

  def SymbolsStaged(self, archive_url, static_dir):
    """Returns True if the build is already staged."""
    rel_path, short_build = self.ParseUrl(archive_url)
    sub_directory = self.GenerateLockTag(rel_path, short_build)
    return os.path.isfile(os.path.join(static_dir,
                                       sub_directory,
                                       self._DONE_FLAG))


class ImagesDownloader(Downloader):
  """Download and stage prebuilt images for a given build.

  Given a URL to a build on the archive server and a list of images:
   - Determine which images have not been staged yet.
   - Download the image archive.
   - Extract missing images to the staging directory.

  """
  _DONE_FLAG = 'staged'

  # List of images to be staged; empty (default) means all.
  _image_list = []

  # A mapping from test image types to their archived file names.
  _IMAGE_TO_FNAME = {
    'test': 'chromiumos_test_image.bin',
    'base': 'chromiumos_base_image.bin',
    'recovery': 'recovery_image.bin',
  }

  @staticmethod
  def GenerateLockTag(rel_path, short_build):
    return os.path.join('images', rel_path, short_build)

  def Download(self, archive_url, image_list, _background=False):
    """Downloads images in |image_list| from the build defined by |archive_url|.

    Download happens synchronously. |images| may include any of those in
    self._IMAGE_TO_FNAME.keys().

    """
    # Check correctness of image list, remove duplicates.
    if not image_list:
      raise DevServerError('empty list of image types')
    invalid_images = list(set(image_list) - set(self._IMAGE_TO_FNAME.keys()))
    if invalid_images:
      raise DevServerError('invalid images requested: %s' % invalid_images)
    image_list = list(set(image_list))

    # Parse archive_url into rel_path (contains the build target) and
    # short_build.
    # e.g. gs://chromeos-image-archive/{rel_path}/{short_build}
    rel_path, short_build = self.ParseUrl(archive_url)

    # Bind build_dir and staging_dir here so we can tell if we need to do any
    # cleanup after an exception occurs before build_dir is set.
    self._lock_tag = self.GenerateLockTag(rel_path, short_build)
    staged_image_list = self._CheckStagedImages(archive_url, self._static_dir)
    unstaged_image_list = [image for image in image_list
                                 if image not in staged_image_list]
    if not unstaged_image_list:
      self._Log(
          'All requested images (%s) for build %s have already been staged.' %
          (common_util.CommaSeparatedList(image_list, is_quoted=True)
           if image_list else 'none',
           self._lock_tag))
      return 'Success'

    self._Log(
        'Image(s) %s for build %s will be staged' %
        (common_util.CommaSeparatedList(unstaged_image_list, is_quoted=True),
         self._lock_tag))
    self._image_list = unstaged_image_list

    try:
      # Create a static target directory and lock it for processing. We permit
      # the directory to preexist, as different images might be downloaded and
      # extracted at different times.
      self._build_dir = common_util.AcquireLock(
          static_dir=self._static_dir, tag=self._lock_tag,
          create_once=False)

      # Replace '/' with '_' in rel_path because it may contain multiple levels
      # which would not be qualified as part of the suffix.
      self._staging_dir = tempfile.mkdtemp(suffix='_'.join(
          [rel_path.replace('/', '_'), short_build]))
      self._Log('Downloading image archive from %s' % archive_url)
      dest_static_dir = os.path.join(self._static_dir, self._lock_tag)
      [image_archive_artifact] = self.GatherArtifactDownloads(
          self._staging_dir, archive_url, dest_static_dir)
      image_archive_artifact.Download()
      self._Log('Staging images to %s' % dest_static_dir)
      image_archive_artifact.Stage()
      self._MarkStagedImages(unstaged_image_list)

    except Exception:
      # Release processing "lock", which will indicate to future runs that we
      # did not succeed, and so they should try again.
      if self._build_dir:
        common_util.ReleaseLock(static_dir=self._static_dir, tag=self._lock_tag,
                                destroy=True)
      raise
    else:
      # Release processing "lock", keeping directory intact.
      if self._build_dir:
        common_util.ReleaseLock(static_dir=self._static_dir, tag=self._lock_tag)
    finally:
      self._Cleanup()

    return 'Success'

  def GatherArtifactDownloads(self, temp_download_dir, archive_url, static_dir,
                              short_build=None):
    """Call appropriate artifact gathering method.

    Args:
      temp_download_dir: temporary directory for downloading artifacts to
      archive_url:       URI to the bucket where image archive is stored
      staging_dir:       directory into which to stage extracted images
      short_build:       (ignored)
    Returns:
      list of downloadable artifacts (of type ZipfileBuildArtifact), currently
      containing a single object, configured for extracting a predetermined
      list of images
    """
    return common_util.GatherImageArchiveArtifactDownloads(
        temp_download_dir, archive_url, static_dir,
        [self._IMAGE_TO_FNAME[image] for image in self._image_list])

  def _MarkStagedImages(self, image_list):
    """Update the on-disk flag file with the list of newly staged images.

    This does not check for duplicates against already listed images, and will
    add any listed images regardless.

    """
    flag_fname = os.path.join(self._build_dir, self._DONE_FLAG)
    with open(flag_fname, 'a') as flag_file:
      flag_file.writelines([image + '\n' for image in image_list])

  def _CheckStagedImages(self, archive_url, static_dir):
    """Returns a list of images that were already staged.

    Reads the list of images from a flag file, if one is present, and returns
    after removing duplicates.

    """
    rel_path, short_build = self.ParseUrl(archive_url)
    sub_directory = self.GenerateLockTag(rel_path, short_build)
    flag_fname = os.path.join(static_dir, sub_directory, self._DONE_FLAG)
    staged_image_list = []
    # TODO(garnold) make this code immune to race conditions, probably by
    # acquiring a lock around the file access code.
    if os.path.isfile(flag_fname):
      with open(flag_fname) as flag_file:
        staged_image_list = [image.strip() for image in flag_file.readlines()]
    return list(set(staged_image_list))