quay/endpoints/verbs.py

import logging
import json

from flask import redirect, Blueprint, abort, send_file

from app import storage as store, app
from auth.auth import process_auth
from auth.permissions import ReadRepositoryPermission
from data import model
from data import database

from util.queuefile import QueueFile
from util.queueprocess import QueueProcess
from util.gzipwrap import GzipWrap
from util.streamlayerformat import StreamLayerMerger


verbs = Blueprint('verbs', __name__)
logger = logging.getLogger(__name__)


def _open_stream(namespace, repository, image_list):
  def get_next_layer():
    for current_image_id in image_list:
      current_image_entry = model.get_repo_image(namespace, repository, current_image_id)
      current_image_path = store.image_layer_path(current_image_entry.storage.uuid)
      current_image_stream = store.stream_read_file(current_image_entry.storage.locations,
                                                    current_image_path)
        
      logger.debug('Returning image layer %s: %s' % (current_image_id, current_image_path))
      yield current_image_stream

  database.configure(app.config)
  stream = GzipWrap(StreamLayerMerger(get_next_layer).get_generator())
  return stream.read


def _write_synthetic_image_to_storage(linked_storage_uuid, linked_locations, queue_file):
  image_path = store.image_layer_path(linked_storage_uuid)
  store.stream_write(linked_locations, image_path, queue_file)
  queue_file.close()

  database.configure(app.config)
  done_uploading = model.get_storage_by_uuid(linked_storage_uuid)
  done_uploading.uploading = False
  done_uploading.save()


@verbs.route('/<namespace>/<repository>/<tag>/squash', methods=['GET'])
@process_auth
def get_squashed_tag(namespace, repository, tag):
  permission = ReadRepositoryPermission(namespace, repository)
  if permission.can() or model.repository_is_public(namespace, repository):
    # Lookup the requested tag.
    tag_image = model.get_tag_image(namespace, repository, tag)
    if not tag_image:
      abort(404)

    # Lookup the tag's image and storage.
    repo_image = model.get_repo_image(namespace, repository, tag_image.docker_image_id)
    if not repo_image:
      abort(404)

    derived = model.find_or_create_derived_storage(repo_image.storage, 'squash',
                                                   store.preferred_locations[0])
    if not derived.uploading:
      logger.debug('Derived image %s exists in storage', derived.uuid)
      derived_layer_path = store.image_layer_path(derived.uuid)
      download_url = store.get_direct_download_url(derived.locations, derived_layer_path)
      if download_url:
        logger.debug('Redirecting to download URL for derived image %s', derived.uuid)
        return redirect(download_url)

      logger.debug('Sending cached derived image %s', derived.uuid)
      return send_file(store.stream_read_file(derived.locations, derived_layer_path))

    # Load the ancestry for the image.
    logger.debug('Building and returning derived image %s', derived.uuid)
    uuid = repo_image.storage.uuid
    ancestry_data = store.get_content(repo_image.storage.locations, store.image_ancestry_path(uuid))
    full_image_list = json.loads(ancestry_data)

    # Create a queue process to generate the data. The queue files will read from the process
    # and send the results to the client and storage.
    args = (namespace, repository, full_image_list)
    queue_process = QueueProcess(_open_stream, 8 * 1024, 10 * 1024 * 1024, args) # 8K/10M chunk/max

    client_queue_file = QueueFile(queue_process.create_queue(), 'client')
    storage_queue_file = QueueFile(queue_process.create_queue(), 'storage')

    # Start building.
    queue_process.run()

    # Start the storage saving.
    storage_args = (derived.uuid, derived.locations, storage_queue_file)
    QueueProcess.run_process(_write_synthetic_image_to_storage, storage_args)

    # Return the client's data.
    return send_file(client_queue_file)

  abort(403)