sdk/python/arvados/commands/keepdocker.py

   1 # Copyright (C) The Arvados Authors. All rights reserved.
   2 #
   3 # SPDX-License-Identifier: Apache-2.0
   4
   5 from builtins import next
   6 import argparse
   7 import collections
   8 import datetime
   9 import errno
  10 import json
  11 import os
  12 import re
  13 import sys
  14 import tarfile
  15 import tempfile
  16 import shutil
  17 import _strptime
  18 import fcntl
  19 from operator import itemgetter
  20 from stat import *
  21
  22 if os.name == "posix" and sys.version_info[0] < 3:
  23     import subprocess32 as subprocess
  24 else:
  25     import subprocess
  26
  27 import arvados
  28 import arvados.util
  29 import arvados.commands._util as arv_cmd
  30 import arvados.commands.put as arv_put
  31 from arvados.collection import CollectionReader
  32 import ciso8601
  33 import logging
  34 import arvados.config
  35
  36 from arvados._version import __version__
  37
  38 logger = logging.getLogger('arvados.keepdocker')
  39 logger.setLevel(logging.DEBUG if arvados.config.get('ARVADOS_DEBUG')
  40                 else logging.INFO)
  41
  42 EARLIEST_DATETIME = datetime.datetime(datetime.MINYEAR, 1, 1, 0, 0, 0)
  43 STAT_CACHE_ERRORS = (IOError, OSError, ValueError)
  44
  45 DockerImage = collections.namedtuple(
  46     'DockerImage', ['repo', 'tag', 'hash', 'created', 'vsize'])
  47
  48 keepdocker_parser = argparse.ArgumentParser(add_help=False)
  49 keepdocker_parser.add_argument(
  50     '--version', action='version', version="%s %s" % (sys.argv[0], __version__),
  51     help='Print version and exit.')
  52 keepdocker_parser.add_argument(
  53     '-f', '--force', action='store_true', default=False,
  54     help="Re-upload the image even if it already exists on the server")
  55 keepdocker_parser.add_argument(
  56     '--force-image-format', action='store_true', default=False,
  57     help="Proceed even if the image format is not supported by the server")
  58
  59 _group = keepdocker_parser.add_mutually_exclusive_group()
  60 _group.add_argument(
  61     '--pull', action='store_true', default=False,
  62     help="Try to pull the latest image from Docker registry")
  63 _group.add_argument(
  64     '--no-pull', action='store_false', dest='pull',
  65     help="Use locally installed image only, don't pull image from Docker registry (default)")
  66
  67 # Combine keepdocker options listed above with run_opts options of arv-put.
  68 # The options inherited from arv-put include --name, --project-uuid,
  69 # --progress/--no-progress/--batch-progress and --resume/--no-resume.
  70 arg_parser = argparse.ArgumentParser(
  71         description="Upload or list Docker images in Arvados",
  72         parents=[keepdocker_parser, arv_put.run_opts, arv_cmd.retry_opt])
  73
  74 arg_parser.add_argument(
  75     'image', nargs='?',
  76     help="Docker image to upload: repo, repo:tag, or hash")
  77 arg_parser.add_argument(
  78     'tag', nargs='?',
  79     help="Tag of the Docker image to upload (default 'latest'), if image is given as an untagged repo name")
  80
  81 class DockerError(Exception):
  82     pass
  83
  84
  85 def popen_docker(cmd, *args, **kwargs):
  86     manage_stdin = ('stdin' not in kwargs)
  87     kwargs.setdefault('stdin', subprocess.PIPE)
  88     kwargs.setdefault('stdout', subprocess.PIPE)
  89     kwargs.setdefault('stderr', subprocess.PIPE)
  90     try:
  91         docker_proc = subprocess.Popen(['docker'] + cmd, *args, **kwargs)
  92     except OSError:  # No docker in $PATH, try docker.io
  93         docker_proc = subprocess.Popen(['docker.io'] + cmd, *args, **kwargs)
  94     if manage_stdin:
  95         docker_proc.stdin.close()
  96     return docker_proc
  97
  98 def check_docker(proc, description):
  99     proc.wait()
 100     if proc.returncode != 0:
 101         raise DockerError("docker {} returned status code {}".
 102                           format(description, proc.returncode))
 103
 104 def docker_image_format(image_hash):
 105     """Return the registry format ('v1' or 'v2') of the given image."""
 106     cmd = popen_docker(['inspect', '--format={{.Id}}', image_hash],
 107                         stdout=subprocess.PIPE)
 108     try:
 109         image_id = next(cmd.stdout).decode('utf-8').strip()
 110         if image_id.startswith('sha256:'):
 111             return 'v2'
 112         elif ':' not in image_id:
 113             return 'v1'
 114         else:
 115             return 'unknown'
 116     finally:
 117         check_docker(cmd, "inspect")
 118
 119 def docker_image_compatible(api, image_hash):
 120     supported = api._rootDesc.get('dockerImageFormats', [])
 121     if not supported:
 122         logger.warning("server does not specify supported image formats (see docker_image_formats in server config).")
 123         return False
 124
 125     fmt = docker_image_format(image_hash)
 126     if fmt in supported:
 127         return True
 128     else:
 129         logger.error("image format is {!r} " \
 130             "but server supports only {!r}".format(fmt, supported))
 131         return False
 132
 133 def docker_images():
 134     # Yield a DockerImage tuple for each installed image.
 135     list_proc = popen_docker(['images', '--no-trunc'], stdout=subprocess.PIPE)
 136     list_output = iter(list_proc.stdout)
 137     next(list_output)  # Ignore the header line
 138     for line in list_output:
 139         words = line.split()
 140         words = [word.decode('utf-8') for word in words]
 141         size_index = len(words) - 2
 142         repo, tag, imageid = words[:3]
 143         ctime = ' '.join(words[3:size_index])
 144         vsize = ' '.join(words[size_index:])
 145         yield DockerImage(repo, tag, imageid, ctime, vsize)
 146     list_proc.stdout.close()
 147     check_docker(list_proc, "images")
 148
 149 def find_image_hashes(image_search, image_tag=None):
 150     # Query for a Docker images with the repository and tag and return
 151     # the image ids in a list.  Returns empty list if no match is
 152     # found.
 153
 154     list_proc = popen_docker(['inspect', "%s%s" % (image_search, ":"+image_tag if image_tag else "")], stdout=subprocess.PIPE)
 155
 156     inspect = list_proc.stdout.read()
 157     list_proc.stdout.close()
 158
 159     imageinfo = json.loads(inspect)
 160
 161     return [i["Id"] for i in imageinfo]
 162
 163 def find_one_image_hash(image_search, image_tag=None):
 164     hashes = find_image_hashes(image_search, image_tag)
 165     hash_count = len(hashes)
 166     if hash_count == 1:
 167         return hashes.pop()
 168     elif hash_count == 0:
 169         raise DockerError("no matching image found")
 170     else:
 171         raise DockerError("{} images match {}".format(hash_count, image_search))
 172
 173 def stat_cache_name(image_file):
 174     return getattr(image_file, 'name', image_file) + '.stat'
 175
 176 def pull_image(image_name, image_tag):
 177     check_docker(popen_docker(['pull', '{}:{}'.format(image_name, image_tag)]),
 178                  "pull")
 179
 180 def save_image(image_hash, image_file):
 181     # Save the specified Docker image to image_file, then try to save its
 182     # stats so we can try to resume after interruption.
 183     check_docker(popen_docker(['save', image_hash], stdout=image_file),
 184                  "save")
 185     image_file.flush()
 186     try:
 187         with open(stat_cache_name(image_file), 'w') as statfile:
 188             json.dump(tuple(os.fstat(image_file.fileno())), statfile)
 189     except STAT_CACHE_ERRORS:
 190         pass  # We won't resume from this cache.  No big deal.
 191
 192 def get_cache_dir():
 193     return arv_cmd.make_home_conf_dir(
 194         os.path.join('.cache', 'arvados', 'docker'), 0o700)
 195
 196 def prep_image_file(filename):
 197     # Return a file object ready to save a Docker image,
 198     # and a boolean indicating whether or not we need to actually save the
 199     # image (False if a cached save is available).
 200     cache_dir = get_cache_dir()
 201     if cache_dir is None:
 202         image_file = tempfile.NamedTemporaryFile(suffix='.tar')
 203         need_save = True
 204     else:
 205         file_path = os.path.join(cache_dir, filename)
 206         try:
 207             with open(stat_cache_name(file_path)) as statfile:
 208                 prev_stat = json.load(statfile)
 209             now_stat = os.stat(file_path)
 210             need_save = any(prev_stat[field] != now_stat[field]
 211                             for field in [ST_MTIME, ST_SIZE])
 212         except STAT_CACHE_ERRORS + (AttributeError, IndexError):
 213             need_save = True  # We couldn't compare against old stats
 214         image_file = open(file_path, 'w+b' if need_save else 'rb')
 215     return image_file, need_save
 216
 217 def make_link(api_client, num_retries, link_class, link_name, **link_attrs):
 218     link_attrs.update({'link_class': link_class, 'name': link_name})
 219     return api_client.links().create(body=link_attrs).execute(
 220         num_retries=num_retries)
 221
 222 def docker_link_sort_key(link):
 223     """Build a sort key to find the latest available Docker image.
 224
 225     To find one source collection for a Docker image referenced by
 226     name or image id, the API server looks for a link with the most
 227     recent `image_timestamp` property; then the most recent
 228     `created_at` timestamp.  This method generates a sort key for
 229     Docker metadata links to sort them from least to most preferred.
 230     """
 231     try:
 232         image_timestamp = ciso8601.parse_datetime_as_naive(
 233             link['properties']['image_timestamp'])
 234     except (KeyError, ValueError):
 235         image_timestamp = EARLIEST_DATETIME
 236     try:
 237         created_timestamp = ciso8601.parse_datetime_as_naive(link['created_at'])
 238     except ValueError:
 239         created_timestamp = None
 240     return (image_timestamp, created_timestamp)
 241
 242 def _get_docker_links(api_client, num_retries, **kwargs):
 243     links = arvados.util.list_all(api_client.links().list,
 244                                   num_retries, **kwargs)
 245     for link in links:
 246         link['_sort_key'] = docker_link_sort_key(link)
 247     links.sort(key=itemgetter('_sort_key'), reverse=True)
 248     return links
 249
 250 def _new_image_listing(link, dockerhash, repo='<none>', tag='<none>'):
 251     timestamp_index = 1 if (link['_sort_key'][0] is EARLIEST_DATETIME) else 0
 252     return {
 253         '_sort_key': link['_sort_key'],
 254         'timestamp': link['_sort_key'][timestamp_index],
 255         'collection': link['head_uuid'],
 256         'dockerhash': dockerhash,
 257         'repo': repo,
 258         'tag': tag,
 259         }
 260
 261 def list_images_in_arv(api_client, num_retries, image_name=None, image_tag=None, project_uuid=None):
 262     """List all Docker images known to the api_client with image_name and
 263     image_tag.  If no image_name is given, defaults to listing all
 264     Docker images.
 265
 266     Returns a list of tuples representing matching Docker images,
 267     sorted in preference order (i.e. the first collection in the list
 268     is the one that the API server would use). Each tuple is a
 269     (collection_uuid, collection_info) pair, where collection_info is
 270     a dict with fields "dockerhash", "repo", "tag", and "timestamp".
 271
 272     """
 273     search_filters = []
 274     repo_links = None
 275     hash_links = None
 276
 277     project_filter = []
 278     if project_uuid is not None:
 279         project_filter = [["owner_uuid", "=", project_uuid]]
 280
 281     if image_name:
 282         # Find images with the name the user specified.
 283         search_links = _get_docker_links(
 284             api_client, num_retries,
 285             filters=[['link_class', '=', 'docker_image_repo+tag'],
 286                      ['name', '=',
 287                       '{}:{}'.format(image_name, image_tag or 'latest')]]+project_filter)
 288         if search_links:
 289             repo_links = search_links
 290         else:
 291             # Fall back to finding images with the specified image hash.
 292             search_links = _get_docker_links(
 293                 api_client, num_retries,
 294                 filters=[['link_class', '=', 'docker_image_hash'],
 295                          ['name', 'ilike', image_name + '%']]+project_filter)
 296             hash_links = search_links
 297         # Only list information about images that were found in the search.
 298         search_filters.append(['head_uuid', 'in',
 299                                [link['head_uuid'] for link in search_links]])
 300
 301     # It should be reasonable to expect that each collection only has one
 302     # image hash (though there may be many links specifying this).  Find
 303     # the API server's most preferred image hash link for each collection.
 304     if hash_links is None:
 305         hash_links = _get_docker_links(
 306             api_client, num_retries,
 307             filters=search_filters + [['link_class', '=', 'docker_image_hash']]+project_filter)
 308     hash_link_map = {link['head_uuid']: link for link in reversed(hash_links)}
 309
 310     # Each collection may have more than one name (though again, one name
 311     # may be specified more than once).  Build an image listing from name
 312     # tags, sorted by API server preference.
 313     if repo_links is None:
 314         repo_links = _get_docker_links(
 315             api_client, num_retries,
 316             filters=search_filters + [['link_class', '=',
 317                                        'docker_image_repo+tag']]+project_filter)
 318     seen_image_names = collections.defaultdict(set)
 319     images = []
 320     for link in repo_links:
 321         collection_uuid = link['head_uuid']
 322         if link['name'] in seen_image_names[collection_uuid]:
 323             continue
 324         seen_image_names[collection_uuid].add(link['name'])
 325         try:
 326             dockerhash = hash_link_map[collection_uuid]['name']
 327         except KeyError:
 328             dockerhash = '<unknown>'
 329         name_parts = link['name'].rsplit(':', 1)
 330         images.append(_new_image_listing(link, dockerhash, *name_parts))
 331
 332     # Find any image hash links that did not have a corresponding name link,
 333     # and add image listings for them, retaining the API server preference
 334     # sorting.
 335     images_start_size = len(images)
 336     for collection_uuid, link in hash_link_map.items():
 337         if not seen_image_names[collection_uuid]:
 338             images.append(_new_image_listing(link, link['name']))
 339     if len(images) > images_start_size:
 340         images.sort(key=itemgetter('_sort_key'), reverse=True)
 341
 342     # Remove any image listings that refer to unknown collections.
 343     existing_coll_uuids = {coll['uuid'] for coll in arvados.util.list_all(
 344             api_client.collections().list, num_retries,
 345             filters=[['uuid', 'in', [im['collection'] for im in images]]]+project_filter,
 346             select=['uuid'])}
 347     return [(image['collection'], image) for image in images
 348             if image['collection'] in existing_coll_uuids]
 349
 350 def items_owned_by(owner_uuid, arv_items):
 351     return (item for item in arv_items if item['owner_uuid'] == owner_uuid)
 352
 353 def _uuid2pdh(api, uuid):
 354     return api.collections().list(
 355         filters=[['uuid', '=', uuid]],
 356         select=['portable_data_hash'],
 357     ).execute()['items'][0]['portable_data_hash']
 358
 359 def main(arguments=None, stdout=sys.stdout, install_sig_handlers=True, api=None):
 360     args = arg_parser.parse_args(arguments)
 361     if api is None:
 362         api = arvados.api('v1')
 363
 364     if args.image is None or args.image == 'images':
 365         fmt = "{:30}  {:10}  {:12}  {:29}  {:20}\n"
 366         stdout.write(fmt.format("REPOSITORY", "TAG", "IMAGE ID", "COLLECTION", "CREATED"))
 367         try:
 368             for i, j in list_images_in_arv(api, args.retries):
 369                 stdout.write(fmt.format(j["repo"], j["tag"], j["dockerhash"][0:12], i, j["timestamp"].strftime("%c")))
 370         except IOError as e:
 371             if e.errno == errno.EPIPE:
 372                 pass
 373             else:
 374                 raise
 375         sys.exit(0)
 376
 377     if re.search(r':\w[-.\w]{0,127}$', args.image):
 378         # image ends with :valid-tag
 379         if args.tag is not None:
 380             logger.error(
 381                 "image %r already includes a tag, cannot add tag argument %r",
 382                 args.image, args.tag)
 383             sys.exit(1)
 384         # rsplit() accommodates "myrepo.example:8888/repo/image:tag"
 385         args.image, args.tag = args.image.rsplit(':', 1)
 386     elif args.tag is None:
 387         args.tag = 'latest'
 388
 389     if '/' in args.image:
 390         hostport, path = args.image.split('/', 1)
 391         if hostport.endswith(':443'):
 392             # "docker pull host:443/asdf" transparently removes the
 393             # :443 (which is redundant because https is implied) and
 394             # after it succeeds "docker images" will list "host/asdf",
 395             # not "host:443/asdf".  If we strip the :443 then the name
 396             # doesn't change underneath us.
 397             args.image = '/'.join([hostport[:-4], path])
 398
 399     # Pull the image if requested, unless the image is specified as a hash
 400     # that we already have.
 401     if args.pull and not find_image_hashes(args.image):
 402         pull_image(args.image, args.tag)
 403
 404     images_in_arv = list_images_in_arv(api, args.retries, args.image, args.tag)
 405
 406     image_hash = None
 407     try:
 408         image_hash = find_one_image_hash(args.image, args.tag)
 409         if not docker_image_compatible(api, image_hash):
 410             if args.force_image_format:
 411                 logger.warning("forcing incompatible image")
 412             else:
 413                 logger.error("refusing to store " \
 414                     "incompatible format (use --force-image-format to override)")
 415                 sys.exit(1)
 416     except DockerError as error:
 417         if images_in_arv:
 418             # We don't have Docker / we don't have the image locally,
 419             # use image that's already uploaded to Arvados
 420             image_hash = images_in_arv[0][1]['dockerhash']
 421         else:
 422             logger.error(str(error))
 423             sys.exit(1)
 424
 425     image_repo_tag = '{}:{}'.format(args.image, args.tag) if not image_hash.startswith(args.image.lower()) else None
 426
 427     if args.name is None:
 428         if image_repo_tag:
 429             collection_name = 'Docker image {} {}'.format(image_repo_tag.replace("/", " "), image_hash[0:12])
 430         else:
 431             collection_name = 'Docker image {}'.format(image_hash[0:12])
 432     else:
 433         collection_name = args.name
 434
 435     # Acquire a lock so that only one arv-keepdocker process will
 436     # dump/upload a particular docker image at a time.  Do this before
 437     # checking if the image already exists in Arvados so that if there
 438     # is an upload already underway, when that upload completes and
 439     # this process gets a turn, it will discover the Docker image is
 440     # already available and exit quickly.
 441     outfile_name = '{}.tar'.format(image_hash)
 442     lockfile_name = '{}.lock'.format(outfile_name)
 443     lockfile = None
 444     cache_dir = get_cache_dir()
 445     if cache_dir:
 446         lockfile = open(os.path.join(cache_dir, lockfile_name), 'w+')
 447         fcntl.flock(lockfile, fcntl.LOCK_EX)
 448
 449     try:
 450         if not args.force:
 451             # Check if this image is already in Arvados.
 452
 453             # Project where everything should be owned
 454             parent_project_uuid = args.project_uuid or api.users().current().execute(
 455                 num_retries=args.retries)['uuid']
 456
 457             # Find image hash tags
 458             existing_links = _get_docker_links(
 459                 api, args.retries,
 460                 filters=[['link_class', '=', 'docker_image_hash'],
 461                          ['name', '=', image_hash]])
 462             if existing_links:
 463                 # get readable collections
 464                 collections = api.collections().list(
 465                     filters=[['uuid', 'in', [link['head_uuid'] for link in existing_links]]],
 466                     select=["uuid", "owner_uuid", "name", "manifest_text"]
 467                     ).execute(num_retries=args.retries)['items']
 468
 469                 if collections:
 470                     # check for repo+tag links on these collections
 471                     if image_repo_tag:
 472                         existing_repo_tag = _get_docker_links(
 473                             api, args.retries,
 474                             filters=[['link_class', '=', 'docker_image_repo+tag'],
 475                                      ['name', '=', image_repo_tag],
 476                                      ['head_uuid', 'in', [c["uuid"] for c in collections]]])
 477                     else:
 478                         existing_repo_tag = []
 479
 480                     try:
 481                         coll_uuid = next(items_owned_by(parent_project_uuid, collections))['uuid']
 482                     except StopIteration:
 483                         # create new collection owned by the project
 484                         coll_uuid = api.collections().create(
 485                             body={"manifest_text": collections[0]['manifest_text'],
 486                                   "name": collection_name,
 487                                   "owner_uuid": parent_project_uuid,
 488                                   "properties": {"docker-image-repo-tag": image_repo_tag}},
 489                             ensure_unique_name=True
 490                             ).execute(num_retries=args.retries)['uuid']
 491
 492                     link_base = {'owner_uuid': parent_project_uuid,
 493                                  'head_uuid':  coll_uuid,
 494                                  'properties': existing_links[0]['properties']}
 495
 496                     if not any(items_owned_by(parent_project_uuid, existing_links)):
 497                         # create image link owned by the project
 498                         make_link(api, args.retries,
 499                                   'docker_image_hash', image_hash, **link_base)
 500
 501                     if image_repo_tag and not any(items_owned_by(parent_project_uuid, existing_repo_tag)):
 502                         # create repo+tag link owned by the project
 503                         make_link(api, args.retries, 'docker_image_repo+tag',
 504                                   image_repo_tag, **link_base)
 505
 506                     stdout.write(coll_uuid + "\n")
 507
 508                     sys.exit(0)
 509
 510         # Open a file for the saved image, and write it if needed.
 511         image_file, need_save = prep_image_file(outfile_name)
 512         if need_save:
 513             save_image(image_hash, image_file)
 514
 515         # Call arv-put with switches we inherited from it
 516         # (a.k.a., switches that aren't our own).
 517         if arguments is None:
 518             arguments = sys.argv[1:]
 519         arguments = [i for i in arguments if i not in (args.image, args.tag, image_repo_tag)]
 520         put_args = keepdocker_parser.parse_known_args(arguments)[1]
 521
 522         # Don't fail when cached manifest is invalid, just ignore the cache.
 523         put_args += ['--batch']
 524
 525         if args.name is None:
 526             put_args += ['--name', collection_name]
 527
 528         coll_uuid = arv_put.main(
 529             put_args + ['--filename', outfile_name, image_file.name], stdout=stdout,
 530             install_sig_handlers=install_sig_handlers).strip()
 531
 532         # Managed properties could be already set
 533         coll_properties = api.collections().get(uuid=coll_uuid).execute(num_retries=args.retries).get('properties', {})
 534         coll_properties.update({"docker-image-repo-tag": image_repo_tag})
 535
 536         api.collections().update(uuid=coll_uuid, body={"properties": coll_properties}).execute(num_retries=args.retries)
 537
 538         # Read the image metadata and make Arvados links from it.
 539         image_file.seek(0)
 540         image_tar = tarfile.open(fileobj=image_file)
 541         image_hash_type, _, raw_image_hash = image_hash.rpartition(':')
 542         if image_hash_type:
 543             json_filename = raw_image_hash + '.json'
 544         else:
 545             json_filename = raw_image_hash + '/json'
 546         json_file = image_tar.extractfile(image_tar.getmember(json_filename))
 547         image_metadata = json.loads(json_file.read().decode('utf-8'))
 548         json_file.close()
 549         image_tar.close()
 550         link_base = {'head_uuid': coll_uuid, 'properties': {}}
 551         if 'created' in image_metadata:
 552             link_base['properties']['image_timestamp'] = image_metadata['created']
 553         if args.project_uuid is not None:
 554             link_base['owner_uuid'] = args.project_uuid
 555
 556         make_link(api, args.retries, 'docker_image_hash', image_hash, **link_base)
 557         if image_repo_tag:
 558             make_link(api, args.retries,
 559                       'docker_image_repo+tag', image_repo_tag, **link_base)
 560
 561         # Clean up.
 562         image_file.close()
 563         for filename in [stat_cache_name(image_file), image_file.name]:
 564             try:
 565                 os.unlink(filename)
 566             except OSError as error:
 567                 if error.errno != errno.ENOENT:
 568                     raise
 569     finally:
 570         if lockfile is not None:
 571             # Closing the lockfile unlocks it.
 572             lockfile.close()
 573
 574 if __name__ == '__main__':
 575     main()