arvados.commands.put

   1# Copyright (C) The Arvados Authors. All rights reserved.
   2#
   3# SPDX-License-Identifier: Apache-2.0
   4
   5import argparse
   6import arvados
   7import arvados.collection
   8import base64
   9import ciso8601
  10import copy
  11import datetime
  12import errno
  13import fcntl
  14import fnmatch
  15import hashlib
  16import json
  17import logging
  18import os
  19import pwd
  20import re
  21import signal
  22import socket
  23import sys
  24import tempfile
  25import threading
  26import time
  27import traceback
  28
  29from pathlib import Path
  30
  31import arvados.util
  32import arvados.commands._util as arv_cmd
  33
  34from apiclient import errors as apiclient_errors
  35from arvados._internal import basedirs
  36from arvados._version import __version__
  37
  38api_client = None
  39
  40
  41def get_run_opts():
  42    """Returns an argparse.ArgumentParser instances that includes
  43    a subset of parameters used both by this script ('arv put') and also reused
  44    by 'arv keep docker'.
  45    """
  46    run_opts = argparse.ArgumentParser(add_help=False)
  47
  48    run_opts.add_argument('--project-uuid', metavar='UUID', help="""
  49    Store the collection in the specified project, instead of your Home
  50    project.
  51    """)
  52
  53    run_opts.add_argument('--name', help="""
  54    Save the collection with the specified name.
  55    """)
  56
  57    _group = run_opts.add_mutually_exclusive_group()
  58    _group.add_argument('--progress', action='store_true',
  59                        help="""
  60    Display human-readable progress on stderr (bytes and, if possible,
  61    percentage of total data size). This is the default behavior when
  62    stderr is a tty.
  63    """)
  64
  65    _group.add_argument('--no-progress', action='store_true',
  66                        help="""
  67    Do not display human-readable progress on stderr, even if stderr is a
  68    tty.
  69    """)
  70
  71    _group.add_argument('--batch-progress', action='store_true',
  72                        help="""
  73    Display machine-readable progress on stderr (bytes and, if known,
  74    total data size).
  75    """)
  76
  77    run_opts.add_argument('--silent', action='store_true',
  78                          help="""
  79    Do not print any debug messages to console. (Any error messages will
  80    still be displayed.)
  81    """)
  82
  83    run_opts.add_argument('--batch', action='store_true', default=False,
  84                          help="""
  85    Retries with '--no-resume --no-cache' if cached state contains invalid/expired
  86    block signatures.
  87    """)
  88
  89    _group = run_opts.add_mutually_exclusive_group()
  90    _group.add_argument('--resume', action='store_true', default=True,
  91                        help="""
  92    Continue interrupted uploads from cached state (default).
  93    """)
  94    _group.add_argument('--no-resume', action='store_false', dest='resume',
  95                        help="""
  96    Do not continue interrupted uploads from cached state.
  97    """)
  98
  99    _group = run_opts.add_mutually_exclusive_group()
 100    _group.add_argument('--cache', action='store_true', dest='use_cache', default=True,
 101                        help="""
 102    Save upload state in a cache file for resuming (default).
 103    """)
 104    _group.add_argument('--no-cache', action='store_false', dest='use_cache',
 105                        help="""
 106    Do not save upload state in a cache file for resuming.
 107    """)
 108
 109    return run_opts
 110
 111
 112def get_argument_parser():
 113    """Returns an argparse.ArgumentParser instance, the main argument parser
 114    used by 'arv put', given the shared argument parser instance.
 115    """
 116    upload_opts = argparse.ArgumentParser(add_help=False)
 117
 118    upload_opts.add_argument('--version', action='version',
 119                             version=f"%(prog)s {__version__}",
 120                             help='Print version and exit.')
 121    upload_opts.add_argument('paths', metavar='path', type=str, nargs='*',
 122                             help="""
 123    Local file or directory. If path is a directory reference with a trailing
 124    slash, then just upload the directory's contents; otherwise upload the
 125    directory itself. Default: read from standard input.
 126    """)
 127
 128    _group = upload_opts.add_mutually_exclusive_group()
 129
 130    _group.add_argument('--max-manifest-depth', type=int, metavar='N',
 131                        default=-1, help=argparse.SUPPRESS)
 132
 133    _group.add_argument('--normalize', action='store_true',
 134                        help="""
 135    Normalize the manifest by re-ordering files and streams after writing
 136    data.
 137    """)
 138
 139    _group.add_argument('--dry-run', action='store_true', default=False,
 140                        help="""
 141    Don't actually upload files, but only check if any file should be
 142    uploaded. Exit with code=2 when files are pending for upload.
 143    """)
 144
 145    _group = upload_opts.add_mutually_exclusive_group()
 146
 147    _group.add_argument('--as-stream', action='store_true', dest='stream',
 148                        help="""
 149    Synonym for --stream.
 150    """)
 151
 152    _group.add_argument('--stream', action='store_true',
 153                        help="""
 154    Store the file content and display the resulting manifest on
 155    stdout. Do not save a Collection object in Arvados.
 156    """)
 157
 158    _group.add_argument('--as-manifest', action='store_true', dest='manifest',
 159                        help="""
 160    Synonym for --manifest.
 161    """)
 162
 163    _group.add_argument('--in-manifest', action='store_true', dest='manifest',
 164                        help="""
 165    Synonym for --manifest.
 166    """)
 167
 168    _group.add_argument('--manifest', action='store_true',
 169                        help="""
 170    Store the file data and resulting manifest in Keep, save a Collection
 171    object in Arvados, and display the manifest locator (Collection uuid)
 172    on stdout. This is the default behavior.
 173    """)
 174
 175    _group.add_argument('--as-raw', action='store_true', dest='raw',
 176                        help="""
 177    Synonym for --raw.
 178    """)
 179
 180    _group.add_argument('--raw', action='store_true',
 181                        help="""
 182    Store the file content and display the data block locators on stdout,
 183    separated by commas, with a trailing newline. Do not store a
 184    manifest.
 185    """)
 186
 187    upload_opts.add_argument('--update-collection', type=str, default=None,
 188                             dest='update_collection', metavar="UUID", help="""
 189    Update an existing collection identified by the given Arvados collection
 190    UUID. All new local files will be uploaded.
 191    """)
 192
 193    upload_opts.add_argument('--use-filename', type=str, default=None,
 194                             dest='filename', help="""
 195    Synonym for --filename.
 196    """)
 197
 198    upload_opts.add_argument('--filename', type=str, default=None,
 199                             help="""
 200    Use the given filename in the manifest, instead of the name of the
 201    local file. This is useful when "-" or "/dev/stdin" is given as an
 202    input file. It can be used only if there is exactly one path given and
 203    it is not a directory. Implies --manifest.
 204    """)
 205
 206    upload_opts.add_argument('--portable-data-hash', action='store_true',
 207                             help="""
 208    Print the portable data hash instead of the Arvados UUID for the collection
 209    created by the upload.
 210    """)
 211
 212    upload_opts.add_argument('--replication', type=int, metavar='N',
 213                             default=None,
 214                             help="""
 215    Set the replication level for the new collection: how many different
 216    physical storage devices (e.g., disks) should have a copy of each data
 217    block. Default is to use the server-provided default (if any) or 2.
 218    """)
 219
 220    upload_opts.add_argument(
 221        '--storage-classes',
 222        type=arv_cmd.UniqueSplit(),
 223        help="""
 224    Specify comma separated list of storage classes to be used when saving data to Keep.
 225    """)
 226
 227    upload_opts.add_argument('--threads', type=int, metavar='N', default=None,
 228                             help="""
 229    Set the number of upload threads to be used. Take into account that
 230    using lots of threads will increase the RAM requirements. Default is
 231    to use 2 threads.
 232    On high latency installations, using a greater number will improve
 233    overall throughput.
 234    """)
 235
 236    upload_opts.add_argument('--exclude', metavar='PATTERN', default=[],
 237                          action='append', help="""
 238    Exclude files and directories whose names match the given glob pattern. When
 239    using a path-like pattern like 'subdir/*.txt', all text files inside 'subdir'
 240    directory, relative to the provided input dirs will be excluded.
 241    When using a filename pattern like '*.txt', any text file will be excluded
 242    no matter where it is placed.
 243    For the special case of needing to exclude only files or dirs directly below
 244    the given input directory, you can use a pattern like './exclude_this.gif'.
 245    You can specify multiple patterns by using this argument more than once.
 246    """)
 247
 248    _group = upload_opts.add_mutually_exclusive_group()
 249    _group.add_argument('--follow-links', action='store_true', default=True,
 250                        dest='follow_links', help="""
 251    Follow file and directory symlinks (default).
 252    """)
 253    _group.add_argument('--no-follow-links', action='store_false', dest='follow_links',
 254                        help="""
 255    Ignore file and directory symlinks. Even paths given explicitly on the
 256    command line will be skipped if they are symlinks.
 257    """)
 258
 259    _group = upload_opts.add_mutually_exclusive_group()
 260    _group.add_argument('--trash-at', metavar='YYYY-MM-DDTHH:MM', default=None,
 261                        help="""
 262    Set the trash date of the resulting collection to an absolute date in the future.
 263    The accepted format is defined by the ISO 8601 standard. Examples: 20090103, 2009-01-03, 20090103T181505, 2009-01-03T18:15:05.\n
 264    Timezone information can be added. If not, the provided date/time is assumed as being in the local system's timezone.
 265    """)
 266    _group.add_argument('--trash-after', type=int, metavar='DAYS', default=None,
 267                        help="""
 268    Set the trash date of the resulting collection to an amount of days from the
 269    date/time that the upload process finishes.
 270    """)
 271
 272    return argparse.ArgumentParser(
 273        description='Copy data from the local filesystem to Keep.',
 274        parents=[upload_opts, get_run_opts(), arv_cmd.retry_opt])
 275
 276
 277def parse_arguments(arguments):
 278    arg_parser = get_argument_parser()
 279    args = arg_parser.parse_args(arguments)
 280
 281    if len(args.paths) == 0:
 282        args.paths = ['-']
 283
 284    args.paths = ["-" if x == "/dev/stdin" else x for x in args.paths]
 285
 286    if args.filename and (len(args.paths) != 1 or os.path.isdir(args.paths[0])):
 287        arg_parser.error("""
 288    --filename argument cannot be used when storing a directory or
 289    multiple files.
 290    """)
 291
 292    # Turn on --progress by default if stderr is a tty.
 293    if (not (args.batch_progress or args.no_progress or args.silent)
 294        and sys.stderr.isatty()):
 295        args.progress = True
 296
 297    # Turn off --resume (default) if --no-cache is used.
 298    if not args.use_cache:
 299        args.resume = False
 300
 301    if args.paths == ['-']:
 302        if args.update_collection:
 303            arg_parser.error("""
 304    --update-collection cannot be used when reading from stdin.
 305    """)
 306        args.resume = False
 307        args.use_cache = False
 308        if not args.filename:
 309            args.filename = 'stdin'
 310
 311    # Remove possible duplicated patterns
 312    if len(args.exclude) > 0:
 313        args.exclude = list(set(args.exclude))
 314
 315    return args
 316
 317
 318class PathDoesNotExistError(Exception):
 319    pass
 320
 321
 322class CollectionUpdateError(Exception):
 323    pass
 324
 325
 326class ResumeCacheConflict(Exception):
 327    pass
 328
 329
 330class ResumeCacheInvalidError(Exception):
 331    pass
 332
 333class ArvPutArgumentConflict(Exception):
 334    pass
 335
 336
 337class ArvPutUploadIsPending(Exception):
 338    pass
 339
 340
 341class ArvPutUploadNotPending(Exception):
 342    pass
 343
 344
 345class FileUploadList(list):
 346    def __init__(self, dry_run=False):
 347        list.__init__(self)
 348        self.dry_run = dry_run
 349
 350    def append(self, other):
 351        if self.dry_run:
 352            raise ArvPutUploadIsPending()
 353        super(FileUploadList, self).append(other)
 354
 355
 356# Appends the X-Request-Id to the log message when log level is ERROR or DEBUG
 357class ArvPutLogFormatter(logging.Formatter):
 358    std_fmtr = logging.Formatter(arvados.log_format, arvados.log_date_format)
 359    err_fmtr = None
 360    request_id_informed = False
 361
 362    def __init__(self, request_id):
 363        self.err_fmtr = logging.Formatter(
 364            arvados.log_format+' (X-Request-Id: {})'.format(request_id),
 365            arvados.log_date_format)
 366
 367    def format(self, record):
 368        if (not self.request_id_informed) and (record.levelno in (logging.DEBUG, logging.ERROR)):
 369            self.request_id_informed = True
 370            return self.err_fmtr.format(record)
 371        return self.std_fmtr.format(record)
 372
 373
 374class ResumeCache(object):
 375    CACHE_DIR = 'arv-put'
 376
 377    def __init__(self, file_spec):
 378        self.cache_file = open(file_spec, 'a+')
 379        self._lock_file(self.cache_file)
 380        self.filename = self.cache_file.name
 381
 382    @classmethod
 383    def make_path(cls, args):
 384        md5 = hashlib.md5()
 385        md5.update(arvados.config.get('ARVADOS_API_HOST', '!nohost').encode())
 386        realpaths = sorted(os.path.realpath(path) for path in args.paths)
 387        md5.update(b'\0'.join([p.encode() for p in realpaths]))
 388        if any(os.path.isdir(path) for path in realpaths):
 389            md5.update(b'-1')
 390        elif args.filename:
 391            md5.update(args.filename.encode())
 392        cache_path = Path(cls.CACHE_DIR)
 393        if len(cache_path.parts) == 1:
 394            cache_path = basedirs.BaseDirectories('CACHE').storage_path(cache_path)
 395        else:
 396            # Note this is a noop if cache_path is absolute, which is what we want.
 397            cache_path = Path.home() / cache_path
 398            cache_path.mkdir(parents=True, exist_ok=True, mode=0o700)
 399        return str(cache_path / md5.hexdigest())
 400
 401    def _lock_file(self, fileobj):
 402        try:
 403            fcntl.flock(fileobj, fcntl.LOCK_EX | fcntl.LOCK_NB)
 404        except IOError:
 405            raise ResumeCacheConflict(u"{} locked".format(fileobj.name))
 406
 407    def load(self):
 408        self.cache_file.seek(0)
 409        return json.load(self.cache_file)
 410
 411    def check_cache(self, api_client=None, num_retries=0):
 412        try:
 413            state = self.load()
 414            locator = None
 415            try:
 416                if "_finished_streams" in state and len(state["_finished_streams"]) > 0:
 417                    locator = state["_finished_streams"][0][1][0]
 418                elif "_current_stream_locators" in state and len(state["_current_stream_locators"]) > 0:
 419                    locator = state["_current_stream_locators"][0]
 420                if locator is not None:
 421                    kc = arvados.keep.KeepClient(api_client=api_client)
 422                    kc.head(locator, num_retries=num_retries)
 423            except Exception as e:
 424                self.restart()
 425        except (ValueError):
 426            pass
 427
 428    def save(self, data):
 429        try:
 430            new_cache_fd, new_cache_name = tempfile.mkstemp(
 431                dir=os.path.dirname(self.filename))
 432            self._lock_file(new_cache_fd)
 433            new_cache = os.fdopen(new_cache_fd, 'r+')
 434            json.dump(data, new_cache)
 435            os.rename(new_cache_name, self.filename)
 436        except (IOError, OSError, ResumeCacheConflict):
 437            try:
 438                os.unlink(new_cache_name)
 439            except NameError:  # mkstemp failed.
 440                pass
 441        else:
 442            self.cache_file.close()
 443            self.cache_file = new_cache
 444
 445    def close(self):
 446        self.cache_file.close()
 447
 448    def destroy(self):
 449        try:
 450            os.unlink(self.filename)
 451        except OSError as error:
 452            if error.errno != errno.ENOENT:  # That's what we wanted anyway.
 453                raise
 454        self.close()
 455
 456    def restart(self):
 457        self.destroy()
 458        self.__init__(self.filename)
 459
 460
 461class ArvPutUploadJob(object):
 462    CACHE_DIR = 'arv-put'
 463    EMPTY_STATE = {
 464        'manifest' : None, # Last saved manifest checkpoint
 465        'files' : {} # Previous run file list: {path : {size, mtime}}
 466    }
 467
 468    def __init__(self, paths, resume=True, use_cache=True, reporter=None,
 469                 name=None, owner_uuid=None, api_client=None, batch_mode=False,
 470                 ensure_unique_name=False, num_retries=None,
 471                 put_threads=None, replication_desired=None, filename=None,
 472                 update_time=60.0, update_collection=None, storage_classes=None,
 473                 logger=logging.getLogger('arvados.arv_put'), dry_run=False,
 474                 follow_links=True, exclude_paths=[], exclude_names=None,
 475                 trash_at=None):
 476        self.paths = paths
 477        self.resume = resume
 478        self.use_cache = use_cache
 479        self.batch_mode = batch_mode
 480        self.update = False
 481        self.reporter = reporter
 482        # This will set to 0 before start counting, if no special files are going
 483        # to be read.
 484        self.bytes_expected = None
 485        self.bytes_written = 0
 486        self.bytes_skipped = 0
 487        self.name = name
 488        self.owner_uuid = owner_uuid
 489        self.ensure_unique_name = ensure_unique_name
 490        self.num_retries = num_retries
 491        self.replication_desired = replication_desired
 492        self.put_threads = put_threads
 493        self.filename = filename
 494        self.storage_classes = storage_classes
 495        self._api_client = api_client
 496        self._state_lock = threading.Lock()
 497        self._state = None # Previous run state (file list & manifest)
 498        self._current_files = [] # Current run file list
 499        self._cache_file = None
 500        self._collection_lock = threading.Lock()
 501        self._remote_collection = None # Collection being updated (if asked)
 502        self._local_collection = None # Collection from previous run manifest
 503        self._file_paths = set() # Files to be updated in remote collection
 504        self._stop_checkpointer = threading.Event()
 505        self._checkpointer = threading.Thread(target=self._update_task)
 506        self._checkpointer.daemon = True
 507        self._update_task_time = update_time  # How many seconds wait between update runs
 508        self._files_to_upload = FileUploadList(dry_run=dry_run)
 509        self._upload_started = False
 510        self.logger = logger
 511        self.dry_run = dry_run
 512        self._checkpoint_before_quit = True
 513        self.follow_links = follow_links
 514        self.exclude_paths = exclude_paths
 515        self.exclude_names = exclude_names
 516        self._trash_at = trash_at
 517
 518        if self._trash_at is not None:
 519            if type(self._trash_at) not in [datetime.datetime, datetime.timedelta]:
 520                raise TypeError('trash_at should be None, timezone-naive datetime or timedelta')
 521            if type(self._trash_at) == datetime.datetime and self._trash_at.tzinfo is not None:
 522                raise TypeError('provided trash_at datetime should be timezone-naive')
 523
 524        if not self.use_cache and self.resume:
 525            raise ArvPutArgumentConflict('resume cannot be True when use_cache is False')
 526
 527        # Check for obvious dry-run responses
 528        if self.dry_run and (not self.use_cache or not self.resume):
 529            raise ArvPutUploadIsPending()
 530
 531        # Load cached data if any and if needed
 532        self._setup_state(update_collection)
 533
 534        # Build the upload file list, excluding requested files and counting the
 535        # bytes expected to be uploaded.
 536        self._build_upload_list()
 537
 538    def _build_upload_list(self):
 539        """
 540        Scan the requested paths to count file sizes, excluding requested files
 541        and dirs and building the upload file list.
 542        """
 543        # If there aren't special files to be read, reset total bytes count to zero
 544        # to start counting.
 545        if not any([p for p in self.paths
 546                    if not (os.path.isfile(p) or os.path.isdir(p))]):
 547            self.bytes_expected = 0
 548
 549        for path in self.paths:
 550            # Test for stdin first, in case some file named '-' exist
 551            if path == '-':
 552                if self.dry_run:
 553                    raise ArvPutUploadIsPending()
 554                self._write_stdin(self.filename or 'stdin')
 555            elif not os.path.exists(path):
 556                 raise PathDoesNotExistError(u"file or directory '{}' does not exist.".format(path))
 557            elif (not self.follow_links) and os.path.islink(path):
 558                self.logger.warning("Skipping symlink '{}'".format(path))
 559                continue
 560            elif os.path.isdir(path):
 561                # Use absolute paths on cache index so CWD doesn't interfere
 562                # with the caching logic.
 563                orig_path = path
 564                path = os.path.abspath(path)
 565                if orig_path[-1:] == os.sep:
 566                    # When passing a directory reference with a trailing slash,
 567                    # its contents should be uploaded directly to the
 568                    # collection's root.
 569                    prefixdir = path
 570                else:
 571                    # When passing a directory reference with no trailing slash,
 572                    # upload the directory to the collection's root.
 573                    prefixdir = os.path.dirname(path)
 574                prefixdir += os.sep
 575                for root, dirs, files in os.walk(path,
 576                                                 followlinks=self.follow_links):
 577                    root_relpath = os.path.relpath(root, path)
 578                    if root_relpath == '.':
 579                        root_relpath = ''
 580                    # Exclude files/dirs by full path matching pattern
 581                    if self.exclude_paths:
 582                        dirs[:] = [d for d in dirs
 583                                   if not any(pathname_match(
 584                                           os.path.join(root_relpath, d), pat)
 585                                              for pat in self.exclude_paths)]
 586                        files = [f for f in files
 587                                 if not any(pathname_match(
 588                                         os.path.join(root_relpath, f), pat)
 589                                            for pat in self.exclude_paths)]
 590                    # Exclude files/dirs by name matching pattern
 591                    if self.exclude_names is not None:
 592                        dirs[:] = [d for d in dirs
 593                                   if not self.exclude_names.match(d)]
 594                        files = [f for f in files
 595                                 if not self.exclude_names.match(f)]
 596                    # Make os.walk()'s dir traversing order deterministic
 597                    dirs.sort()
 598                    files.sort()
 599                    for f in files:
 600                        filepath = os.path.join(root, f)
 601                        if not os.path.isfile(filepath):
 602                            self.logger.warning("Skipping non-regular file '{}'".format(filepath))
 603                            continue
 604                        # Add its size to the total bytes count (if applicable)
 605                        if self.follow_links or (not os.path.islink(filepath)):
 606                            if self.bytes_expected is not None:
 607                                self.bytes_expected += os.path.getsize(filepath)
 608                        self._check_file(filepath,
 609                                         os.path.join(root[len(prefixdir):], f))
 610            else:
 611                filepath = os.path.abspath(path)
 612                # Add its size to the total bytes count (if applicable)
 613                if self.follow_links or (not os.path.islink(filepath)):
 614                    if self.bytes_expected is not None:
 615                        self.bytes_expected += os.path.getsize(filepath)
 616                self._check_file(filepath,
 617                                 self.filename or os.path.basename(path))
 618        # If dry-mode is on, and got up to this point, then we should notify that
 619        # there aren't any file to upload.
 620        if self.dry_run:
 621            raise ArvPutUploadNotPending()
 622        # Remove local_collection's files that don't exist locally anymore, so the
 623        # bytes_written count is correct.
 624        for f in self.collection_file_paths(self._local_collection,
 625                                            path_prefix=""):
 626            if f != 'stdin' and f != self.filename and not f in self._file_paths:
 627                self._local_collection.remove(f)
 628
 629    def start(self, save_collection):
 630        """
 631        Start supporting thread & file uploading
 632        """
 633        self._checkpointer.start()
 634        try:
 635            # Update bytes_written from current local collection and
 636            # report initial progress.
 637            self._update()
 638            # Actual file upload
 639            self._upload_started = True # Used by the update thread to start checkpointing
 640            self._upload_files()
 641        except (SystemExit, Exception) as e:
 642            self._checkpoint_before_quit = False
 643            # Log stack trace only when Ctrl-C isn't pressed (SIGINT)
 644            # Note: We're expecting SystemExit instead of
 645            # KeyboardInterrupt because we have a custom signal
 646            # handler in place that raises SystemExit with the catched
 647            # signal's code.
 648            if isinstance(e, PathDoesNotExistError):
 649                # We aren't interested in the traceback for this case
 650                pass
 651            elif not isinstance(e, SystemExit) or e.code != -2:
 652                self.logger.warning("Abnormal termination:\n{}".format(
 653                    traceback.format_exc()))
 654            raise
 655        finally:
 656            if not self.dry_run:
 657                # Stop the thread before doing anything else
 658                self._stop_checkpointer.set()
 659                self._checkpointer.join()
 660                if self._checkpoint_before_quit:
 661                    # Commit all pending blocks & one last _update()
 662                    self._local_collection.manifest_text()
 663                    self._update(final=True)
 664                    if save_collection:
 665                        self.save_collection()
 666            if self.use_cache:
 667                self._cache_file.close()
 668
 669    def _collection_trash_at(self):
 670        """
 671        Returns the trash date that the collection should use at save time.
 672        Takes into account absolute/relative trash_at values requested
 673        by the user.
 674        """
 675        if type(self._trash_at) == datetime.timedelta:
 676            # Get an absolute datetime for trash_at
 677            return datetime.datetime.utcnow() + self._trash_at
 678        return self._trash_at
 679
 680    def save_collection(self):
 681        if self.update:
 682            # Check if files should be updated on the remote collection.
 683            for fp in self._file_paths:
 684                remote_file = self._remote_collection.find(fp)
 685                if not remote_file:
 686                    # File don't exist on remote collection, copy it.
 687                    self._remote_collection.copy(fp, fp, self._local_collection)
 688                elif remote_file != self._local_collection.find(fp):
 689                    # A different file exist on remote collection, overwrite it.
 690                    self._remote_collection.copy(fp, fp, self._local_collection, overwrite=True)
 691                else:
 692                    # The file already exist on remote collection, skip it.
 693                    pass
 694            self._remote_collection.save(num_retries=self.num_retries,
 695                                         trash_at=self._collection_trash_at())
 696        else:
 697            if len(self._local_collection) == 0:
 698                self.logger.warning("No files were uploaded, skipping collection creation.")
 699                return
 700            self._local_collection.save_new(
 701                name=self.name, owner_uuid=self.owner_uuid,
 702                ensure_unique_name=self.ensure_unique_name,
 703                num_retries=self.num_retries,
 704                trash_at=self._collection_trash_at())
 705
 706    def destroy_cache(self):
 707        if self.use_cache:
 708            try:
 709                os.unlink(self._cache_filename)
 710            except OSError as error:
 711                # That's what we wanted anyway.
 712                if error.errno != errno.ENOENT:
 713                    raise
 714            self._cache_file.close()
 715
 716    def _collection_size(self, collection):
 717        """
 718        Recursively get the total size of the collection
 719        """
 720        size = 0
 721        for item in collection.values():
 722            if isinstance(item, arvados.collection.Collection) or isinstance(item, arvados.collection.Subcollection):
 723                size += self._collection_size(item)
 724            else:
 725                size += item.size()
 726        return size
 727
 728    def _update_task(self):
 729        """
 730        Periodically called support task. File uploading is
 731        asynchronous so we poll status from the collection.
 732        """
 733        while not self._stop_checkpointer.wait(1 if not self._upload_started else self._update_task_time):
 734            self._update()
 735
 736    def _update(self, final=False):
 737        """
 738        Update cached manifest text and report progress.
 739        """
 740        if self._upload_started:
 741            with self._collection_lock:
 742                self.bytes_written = self._collection_size(self._local_collection)
 743                if self.use_cache:
 744                    if final:
 745                        manifest = self._local_collection.manifest_text()
 746                    else:
 747                        # Get the manifest text without comitting pending blocks
 748                        manifest = self._local_collection.manifest_text(strip=False,
 749                                                                        normalize=False,
 750                                                                        only_committed=True)
 751                    # Update cache
 752                    with self._state_lock:
 753                        self._state['manifest'] = manifest
 754            if self.use_cache:
 755                try:
 756                    self._save_state()
 757                except Exception as e:
 758                    self.logger.error("Unexpected error trying to save cache file: {}".format(e))
 759            # Keep remote collection's trash_at attribute synced when using relative expire dates
 760            if self._remote_collection is not None and type(self._trash_at) == datetime.timedelta:
 761                try:
 762                    self._api_client.collections().update(
 763                        uuid=self._remote_collection.manifest_locator(),
 764                        body={'trash_at': self._collection_trash_at().strftime("%Y-%m-%dT%H:%M:%S.%fZ")}
 765                    ).execute(num_retries=self.num_retries)
 766                except Exception as e:
 767                    self.logger.error("Unexpected error trying to update remote collection's expire date: {}".format(e))
 768        else:
 769            self.bytes_written = self.bytes_skipped
 770        # Call the reporter, if any
 771        self.report_progress()
 772
 773    def report_progress(self):
 774        if self.reporter is not None:
 775            self.reporter(self.bytes_written, self.bytes_expected)
 776
 777    def _write_stdin(self, filename):
 778        output = self._local_collection.open(filename, 'wb')
 779        self._write(sys.stdin.buffer, output)
 780        output.close()
 781
 782    def _check_file(self, source, filename):
 783        """
 784        Check if this file needs to be uploaded
 785        """
 786        # Ignore symlinks when requested
 787        if (not self.follow_links) and os.path.islink(source):
 788            return
 789        resume_offset = 0
 790        should_upload = False
 791        new_file_in_cache = False
 792        # Record file path for updating the remote collection before exiting
 793        self._file_paths.add(filename)
 794
 795        with self._state_lock:
 796            # If no previous cached data on this file, store it for an eventual
 797            # repeated run.
 798            if source not in self._state['files']:
 799                self._state['files'][source] = {
 800                    'mtime': os.path.getmtime(source),
 801                    'size' : os.path.getsize(source)
 802                }
 803                new_file_in_cache = True
 804            cached_file_data = self._state['files'][source]
 805
 806        # Check if file was already uploaded (at least partially)
 807        file_in_local_collection = self._local_collection.find(filename)
 808
 809        # If not resuming, upload the full file.
 810        if not self.resume:
 811            should_upload = True
 812        # New file detected from last run, upload it.
 813        elif new_file_in_cache:
 814            should_upload = True
 815        # Local file didn't change from last run.
 816        elif cached_file_data['mtime'] == os.path.getmtime(source) and cached_file_data['size'] == os.path.getsize(source):
 817            if not file_in_local_collection:
 818                # File not uploaded yet, upload it completely
 819                should_upload = True
 820            elif file_in_local_collection.permission_expired():
 821                # Permission token expired, re-upload file. This will change whenever
 822                # we have a API for refreshing tokens.
 823                self.logger.warning(u"Uploaded file '{}' access token expired, will re-upload it from scratch".format(filename))
 824                should_upload = True
 825                self._local_collection.remove(filename)
 826            elif cached_file_data['size'] == file_in_local_collection.size():
 827                # File already there, skip it.
 828                self.bytes_skipped += cached_file_data['size']
 829            elif cached_file_data['size'] > file_in_local_collection.size():
 830                # File partially uploaded, resume!
 831                resume_offset = file_in_local_collection.size()
 832                self.bytes_skipped += resume_offset
 833                should_upload = True
 834            else:
 835                # Inconsistent cache, re-upload the file
 836                should_upload = True
 837                self._local_collection.remove(filename)
 838                self.logger.warning(u"Uploaded version of file '{}' is bigger than local version, will re-upload it from scratch.".format(source))
 839        # Local file differs from cached data, re-upload it.
 840        else:
 841            if file_in_local_collection:
 842                self._local_collection.remove(filename)
 843            should_upload = True
 844
 845        if should_upload:
 846            try:
 847                self._files_to_upload.append((source, resume_offset, filename))
 848            except ArvPutUploadIsPending:
 849                # This could happen when running on dry-mode, close cache file to
 850                # avoid locking issues.
 851                self._cache_file.close()
 852                raise
 853
 854    def _upload_files(self):
 855        for source, resume_offset, filename in self._files_to_upload:
 856            with open(source, 'rb') as source_fd:
 857                with self._state_lock:
 858                    self._state['files'][source]['mtime'] = os.path.getmtime(source)
 859                    self._state['files'][source]['size'] = os.path.getsize(source)
 860                if resume_offset > 0:
 861                    # Start upload where we left off
 862                    output = self._local_collection.open(filename, 'ab')
 863                    source_fd.seek(resume_offset)
 864                else:
 865                    # Start from scratch
 866                    output = self._local_collection.open(filename, 'wb')
 867                self._write(source_fd, output)
 868                output.close(flush=False)
 869
 870    def _write(self, source_fd, output):
 871        while True:
 872            data = source_fd.read(arvados.config.KEEP_BLOCK_SIZE)
 873            if not data:
 874                break
 875            output.write(data)
 876
 877    def _my_collection(self):
 878        return self._remote_collection if self.update else self._local_collection
 879
 880    def _get_cache_filepath(self):
 881        # Set up cache file name from input paths.
 882        md5 = hashlib.md5()
 883        md5.update(arvados.config.get('ARVADOS_API_HOST', '!nohost').encode())
 884        realpaths = sorted(os.path.realpath(path) for path in self.paths)
 885        md5.update(b'\0'.join([p.encode() for p in realpaths]))
 886        if self.filename:
 887            md5.update(self.filename.encode())
 888        cache_path = Path(self.CACHE_DIR)
 889        if len(cache_path.parts) == 1:
 890            cache_path = basedirs.BaseDirectories('CACHE').storage_path(cache_path)
 891        else:
 892            # Note this is a noop if cache_path is absolute, which is what we want.
 893            cache_path = Path.home() / cache_path
 894            cache_path.mkdir(parents=True, exist_ok=True, mode=0o700)
 895        return str(cache_path / md5.hexdigest())
 896
 897    def _setup_state(self, update_collection):
 898        """
 899        Create a new cache file or load a previously existing one.
 900        """
 901        # Load an already existing collection for update
 902        if update_collection and re.match(arvados.util.collection_uuid_pattern,
 903                                          update_collection):
 904            try:
 905                self._remote_collection = arvados.collection.Collection(
 906                    update_collection,
 907                    api_client=self._api_client,
 908                    storage_classes_desired=self.storage_classes,
 909                    num_retries=self.num_retries)
 910            except arvados.errors.ApiError as error:
 911                raise CollectionUpdateError("Cannot read collection {} ({})".format(update_collection, error))
 912            else:
 913                self.update = True
 914        elif update_collection:
 915            # Collection locator provided, but unknown format
 916            raise CollectionUpdateError("Collection locator unknown: '{}'".format(update_collection))
 917
 918        if self.use_cache:
 919            cache_filepath = self._get_cache_filepath()
 920            if self.resume and os.path.exists(cache_filepath):
 921                self.logger.info(u"Resuming upload from cache file {}".format(cache_filepath))
 922                self._cache_file = open(cache_filepath, 'a+')
 923            else:
 924                # --no-resume means start with a empty cache file.
 925                self.logger.info(u"Creating new cache file at {}".format(cache_filepath))
 926                self._cache_file = open(cache_filepath, 'w+')
 927            self._cache_filename = self._cache_file.name
 928            self._lock_file(self._cache_file)
 929            self._cache_file.seek(0)
 930
 931        with self._state_lock:
 932            if self.use_cache:
 933                try:
 934                    self._state = json.load(self._cache_file)
 935                    if not set(['manifest', 'files']).issubset(set(self._state.keys())):
 936                        # Cache at least partially incomplete, set up new cache
 937                        self._state = copy.deepcopy(self.EMPTY_STATE)
 938                except ValueError:
 939                    # Cache file empty, set up new cache
 940                    self._state = copy.deepcopy(self.EMPTY_STATE)
 941            else:
 942                self.logger.info("No cache usage requested for this run.")
 943                # No cache file, set empty state
 944                self._state = copy.deepcopy(self.EMPTY_STATE)
 945            if not self._cached_manifest_valid():
 946                if not self.batch_mode:
 947                    raise ResumeCacheInvalidError()
 948                else:
 949                    self.logger.info("Invalid signatures on cache file '{}' while being run in 'batch mode' -- continuing anyways.".format(self._cache_file.name))
 950                    self.use_cache = False # Don't overwrite preexisting cache file.
 951                    self._state = copy.deepcopy(self.EMPTY_STATE)
 952            # Load the previous manifest so we can check if files were modified remotely.
 953            self._local_collection = arvados.collection.Collection(
 954                self._state['manifest'],
 955                replication_desired=self.replication_desired,
 956                storage_classes_desired=self.storage_classes,
 957                put_threads=self.put_threads,
 958                api_client=self._api_client,
 959                num_retries=self.num_retries)
 960
 961    def _cached_manifest_valid(self):
 962        """
 963        Validate the oldest non-expired block signature to check if cached manifest
 964        is usable: checking if the cached manifest was not created with a different
 965        arvados account.
 966        """
 967        if self._state.get('manifest', None) is None:
 968            # No cached manifest yet, all good.
 969            return True
 970        now = datetime.datetime.utcnow()
 971        oldest_exp = None
 972        oldest_loc = None
 973        block_found = False
 974        for m in arvados.util.keep_locator_pattern.finditer(self._state['manifest']):
 975            loc = m.group(0)
 976            try:
 977                exp = datetime.datetime.utcfromtimestamp(int(loc.split('@')[1], 16))
 978            except IndexError:
 979                # Locator without signature
 980                continue
 981            block_found = True
 982            if exp > now and (oldest_exp is None or exp < oldest_exp):
 983                oldest_exp = exp
 984                oldest_loc = loc
 985        if not block_found:
 986            # No block signatures found => no invalid block signatures.
 987            return True
 988        if oldest_loc is None:
 989            # Locator signatures found, but all have expired.
 990            # Reset the cache and move on.
 991            self.logger.info('Cache expired, starting from scratch.')
 992            self._state['manifest'] = ''
 993            return True
 994        kc = arvados.KeepClient(api_client=self._api_client,
 995                                num_retries=self.num_retries)
 996        try:
 997            kc.head(oldest_loc)
 998        except arvados.errors.KeepRequestError:
 999            # Something is wrong, cached manifest is not valid.
1000            return False
1001        return True
1002
1003    def collection_file_paths(self, col, path_prefix='.'):
1004        """Return a list of file paths by recursively go through the entire collection `col`"""
1005        file_paths = []
1006        for name, item in col.items():
1007            if isinstance(item, arvados.arvfile.ArvadosFile):
1008                file_paths.append(os.path.join(path_prefix, name))
1009            elif isinstance(item, arvados.collection.Subcollection):
1010                new_prefix = os.path.join(path_prefix, name)
1011                file_paths += self.collection_file_paths(item, path_prefix=new_prefix)
1012        return file_paths
1013
1014    def _lock_file(self, fileobj):
1015        try:
1016            fcntl.flock(fileobj, fcntl.LOCK_EX | fcntl.LOCK_NB)
1017        except IOError:
1018            raise ResumeCacheConflict(u"{} locked".format(fileobj.name))
1019
1020    def _save_state(self):
1021        """
1022        Atomically save current state into cache.
1023        """
1024        with self._state_lock:
1025            # We're not using copy.deepcopy() here because it's a lot slower
1026            # than json.dumps(), and we're already needing JSON format to be
1027            # saved on disk.
1028            state = json.dumps(self._state)
1029        try:
1030            new_cache = tempfile.NamedTemporaryFile(
1031                mode='w+',
1032                dir=os.path.dirname(self._cache_filename), delete=False)
1033            self._lock_file(new_cache)
1034            new_cache.write(state)
1035            new_cache.flush()
1036            os.fsync(new_cache)
1037            os.rename(new_cache.name, self._cache_filename)
1038        except (IOError, OSError, ResumeCacheConflict) as error:
1039            self.logger.error("There was a problem while saving the cache file: {}".format(error))
1040            try:
1041                os.unlink(new_cache_name)
1042            except NameError:  # mkstemp failed.
1043                pass
1044        else:
1045            self._cache_file.close()
1046            self._cache_file = new_cache
1047
1048    def collection_name(self):
1049        return self._my_collection().api_response()['name'] if self._my_collection().api_response() else None
1050
1051    def collection_trash_at(self):
1052        return self._my_collection().get_trash_at()
1053
1054    def manifest_locator(self):
1055        return self._my_collection().manifest_locator()
1056
1057    def portable_data_hash(self):
1058        pdh = self._my_collection().portable_data_hash()
1059        m = self._my_collection().stripped_manifest().encode()
1060        local_pdh = '{}+{}'.format(hashlib.md5(m).hexdigest(), len(m))
1061        if pdh != local_pdh:
1062            self.logger.warning("\n".join([
1063                "arv-put: API server provided PDH differs from local manifest.",
1064                "         This should not happen; showing API server version."]))
1065        return pdh
1066
1067    def manifest_text(self, stream_name=".", strip=False, normalize=False):
1068        return self._my_collection().manifest_text(stream_name, strip, normalize)
1069
1070    def _datablocks_on_item(self, item):
1071        """
1072        Return a list of datablock locators, recursively navigating
1073        through subcollections
1074        """
1075        if isinstance(item, arvados.arvfile.ArvadosFile):
1076            if item.size() == 0:
1077                # Empty file locator
1078                return ["d41d8cd98f00b204e9800998ecf8427e+0"]
1079            else:
1080                locators = []
1081                for segment in item.segments():
1082                    loc = segment.locator
1083                    locators.append(loc)
1084                return locators
1085        elif isinstance(item, arvados.collection.Collection):
1086            l = [self._datablocks_on_item(x) for x in item.values()]
1087            # Fast list flattener method taken from:
1088            # http://stackoverflow.com/questions/952914/making-a-flat-list-out-of-list-of-lists-in-python
1089            return [loc for sublist in l for loc in sublist]
1090        else:
1091            return None
1092
1093    def data_locators(self):
1094        with self._collection_lock:
1095            # Make sure all datablocks are flushed before getting the locators
1096            self._my_collection().manifest_text()
1097            datablocks = self._datablocks_on_item(self._my_collection())
1098        return datablocks
1099
1100_machine_format = "{} {}: {{}} written {{}} total\n".format(sys.argv[0],
1101                                                            os.getpid())
1102
1103# Simulate glob.glob() matching behavior without the need to scan the filesystem
1104# Note: fnmatch() doesn't work correctly when used with pathnames. For example the
1105# pattern 'tests/*.py' will match 'tests/run_test.py' and also 'tests/subdir/run_test.py',
1106# so instead we're using it on every path component.
1107def pathname_match(pathname, pattern):
1108    name = pathname.split(os.sep)
1109    # Fix patterns like 'some/subdir/' or 'some//subdir'
1110    pat = [x for x in pattern.split(os.sep) if x != '' and x != '.']
1111    if len(name) != len(pat):
1112        return False
1113    for i in range(len(name)):
1114        if not fnmatch.fnmatch(name[i], pat[i]):
1115            return False
1116    return True
1117
1118def machine_progress(bytes_written, bytes_expected):
1119    return _machine_format.format(
1120        bytes_written, -1 if (bytes_expected is None) else bytes_expected)
1121
1122def human_progress(bytes_written, bytes_expected):
1123    if bytes_expected:
1124        return "\r{}M / {}M {:.1%} ".format(
1125            bytes_written >> 20, bytes_expected >> 20,
1126            float(bytes_written) / bytes_expected)
1127    else:
1128        return "\r{} ".format(bytes_written)
1129
1130def progress_writer(progress_func, outfile=sys.stderr):
1131    def write_progress(bytes_written, bytes_expected):
1132        outfile.write(progress_func(bytes_written, bytes_expected))
1133    return write_progress
1134
1135def desired_project_uuid(api_client, project_uuid, num_retries):
1136    if not project_uuid:
1137        query = api_client.users().current()
1138    elif arvados.util.user_uuid_pattern.match(project_uuid):
1139        query = api_client.users().get(uuid=project_uuid)
1140    elif arvados.util.group_uuid_pattern.match(project_uuid):
1141        query = api_client.groups().get(uuid=project_uuid)
1142    else:
1143        raise ValueError("Not a valid project UUID: {}".format(project_uuid))
1144    return query.execute(num_retries=num_retries)['uuid']
1145
1146def main(arguments=None, stdout=sys.stdout, stderr=sys.stderr,
1147         install_sig_handlers=True):
1148    global api_client
1149
1150    args = parse_arguments(arguments)
1151    logger = logging.getLogger('arvados.arv_put')
1152    if args.silent:
1153        logger.setLevel(logging.WARNING)
1154    else:
1155        logger.setLevel(logging.INFO)
1156    status = 0
1157
1158    request_id = arvados.util.new_request_id()
1159
1160    formatter = ArvPutLogFormatter(request_id)
1161    logging.getLogger('arvados').handlers[0].setFormatter(formatter)
1162
1163    if api_client is None:
1164        api_client = arvados.api('v1', request_id=request_id, num_retries=args.retries)
1165
1166    if install_sig_handlers:
1167        arv_cmd.install_signal_handlers()
1168
1169    # Trash arguments validation
1170    trash_at = None
1171    if args.trash_at is not None:
1172        # ciso8601 considers YYYYMM as invalid but YYYY-MM as valid, so here we
1173        # make sure the user provides a complete YYYY-MM-DD date.
1174        if not re.match(r'^\d{4}(?P<dash>-?)\d{2}?(?P=dash)\d{2}', args.trash_at):
1175            logger.error("--trash-at argument format invalid, use --help to see examples.")
1176            sys.exit(1)
1177        # Check if no time information was provided. In that case, assume end-of-day.
1178        if re.match(r'^\d{4}(?P<dash>-?)\d{2}?(?P=dash)\d{2}$', args.trash_at):
1179            args.trash_at += 'T23:59:59'
1180        try:
1181            trash_at = ciso8601.parse_datetime(args.trash_at)
1182        except:
1183            logger.error("--trash-at argument format invalid, use --help to see examples.")
1184            sys.exit(1)
1185        else:
1186            if trash_at.tzinfo is not None:
1187                # Timezone aware datetime provided.
1188                utcoffset = -trash_at.utcoffset()
1189            else:
1190                # Timezone naive datetime provided. Assume is local.
1191                if time.daylight:
1192                    utcoffset = datetime.timedelta(seconds=time.altzone)
1193                else:
1194                    utcoffset = datetime.timedelta(seconds=time.timezone)
1195            # Convert to UTC timezone naive datetime.
1196            trash_at = trash_at.replace(tzinfo=None) + utcoffset
1197
1198        if trash_at <= datetime.datetime.utcnow():
1199            logger.error("--trash-at argument must be set in the future")
1200            sys.exit(1)
1201    if args.trash_after is not None:
1202        if args.trash_after < 1:
1203            logger.error("--trash-after argument must be >= 1")
1204            sys.exit(1)
1205        trash_at = datetime.timedelta(seconds=(args.trash_after * 24 * 60 * 60))
1206
1207    # Determine the name to use
1208    if args.name:
1209        if args.stream or args.raw:
1210            logger.error("Cannot use --name with --stream or --raw")
1211            sys.exit(1)
1212        elif args.update_collection:
1213            logger.error("Cannot use --name with --update-collection")
1214            sys.exit(1)
1215        collection_name = args.name
1216    else:
1217        collection_name = "Saved at {} by {}@{}".format(
1218            datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S UTC"),
1219            pwd.getpwuid(os.getuid()).pw_name,
1220            socket.gethostname())
1221
1222    if args.project_uuid and (args.stream or args.raw):
1223        logger.error("Cannot use --project-uuid with --stream or --raw")
1224        sys.exit(1)
1225
1226    # Determine the parent project
1227    try:
1228        project_uuid = desired_project_uuid(api_client, args.project_uuid,
1229                                            args.retries)
1230    except (apiclient_errors.Error, ValueError) as error:
1231        logger.error(error)
1232        sys.exit(1)
1233
1234    if args.progress:
1235        reporter = progress_writer(human_progress)
1236    elif args.batch_progress:
1237        reporter = progress_writer(machine_progress)
1238    else:
1239        reporter = None
1240
1241    # Setup exclude regex from all the --exclude arguments provided
1242    name_patterns = []
1243    exclude_paths = []
1244    exclude_names = None
1245    if len(args.exclude) > 0:
1246        # We're supporting 2 kinds of exclusion patterns:
1247        # 1)   --exclude '*.jpg'    (file/dir name patterns, will only match
1248        #                            the name, wherever the file is on the tree)
1249        # 2.1) --exclude 'foo/bar'  (file/dir path patterns, will match the
1250        #                            entire path, and should be relative to
1251        #                            any input dir argument)
1252        # 2.2) --exclude './*.jpg'  (Special case for excluding files/dirs
1253        #                            placed directly underneath the input dir)
1254        for p in args.exclude:
1255            # Only relative paths patterns allowed
1256            if p.startswith(os.sep):
1257                logger.error("Cannot use absolute paths with --exclude")
1258                sys.exit(1)
1259            if os.path.dirname(p):
1260                # We don't support of path patterns with '..'
1261                p_parts = p.split(os.sep)
1262                if '..' in p_parts:
1263                    logger.error(
1264                        "Cannot use path patterns that include or '..'")
1265                    sys.exit(1)
1266                # Path search pattern
1267                exclude_paths.append(p)
1268            else:
1269                # Name-only search pattern
1270                name_patterns.append(p)
1271        # For name only matching, we can combine all patterns into a single
1272        # regexp, for better performance.
1273        exclude_names = re.compile('|'.join(
1274            [fnmatch.translate(p) for p in name_patterns]
1275        )) if len(name_patterns) > 0 else None
1276        # Show the user the patterns to be used, just in case they weren't
1277        # specified inside quotes and got changed by the shell expansion.
1278        logger.info("Exclude patterns: {}".format(args.exclude))
1279
1280    # If this is used by a human, and there's at least one directory to be
1281    # uploaded, the expected bytes calculation can take a moment.
1282    if args.progress and any([os.path.isdir(f) for f in args.paths]):
1283        logger.info("Calculating upload size, this could take some time...")
1284    try:
1285        writer = ArvPutUploadJob(paths = args.paths,
1286                                 resume = args.resume,
1287                                 use_cache = args.use_cache,
1288                                 batch_mode= args.batch,
1289                                 filename = args.filename,
1290                                 reporter = reporter,
1291                                 api_client = api_client,
1292                                 num_retries = args.retries,
1293                                 replication_desired = args.replication,
1294                                 put_threads = args.threads,
1295                                 name = collection_name,
1296                                 owner_uuid = project_uuid,
1297                                 ensure_unique_name = True,
1298                                 update_collection = args.update_collection,
1299                                 storage_classes=args.storage_classes,
1300                                 logger=logger,
1301                                 dry_run=args.dry_run,
1302                                 follow_links=args.follow_links,
1303                                 exclude_paths=exclude_paths,
1304                                 exclude_names=exclude_names,
1305                                 trash_at=trash_at)
1306    except ResumeCacheConflict:
1307        logger.error("\n".join([
1308            "arv-put: Another process is already uploading this data.",
1309            "         Use --no-cache if this is really what you want."]))
1310        sys.exit(1)
1311    except ResumeCacheInvalidError:
1312        logger.error("\n".join([
1313            "arv-put: Resume cache contains invalid signature: it may have expired",
1314            "         or been created with another Arvados user's credentials.",
1315            "         Switch user or use one of the following options to restart upload:",
1316            "         --no-resume to start a new resume cache.",
1317            "         --no-cache to disable resume cache.",
1318            "         --batch to ignore the resume cache if invalid."]))
1319        sys.exit(1)
1320    except (CollectionUpdateError, PathDoesNotExistError) as error:
1321        logger.error("\n".join([
1322            "arv-put: %s" % str(error)]))
1323        sys.exit(1)
1324    except ArvPutUploadIsPending:
1325        # Dry run check successful, return proper exit code.
1326        sys.exit(2)
1327    except ArvPutUploadNotPending:
1328        # No files pending for upload
1329        sys.exit(0)
1330
1331    if not args.dry_run and not args.update_collection and args.resume and writer.bytes_written > 0:
1332        logger.warning("\n".join([
1333            "arv-put: Resuming previous upload from last checkpoint.",
1334            "         Use the --no-resume option to start over."]))
1335
1336    if not args.dry_run:
1337        writer.report_progress()
1338    output = None
1339    try:
1340        writer.start(save_collection=not(args.stream or args.raw))
1341    except (arvados.errors.ApiError, arvados.errors.KeepWriteError) as error:
1342        logger.error("\n".join([
1343            "arv-put: %s" % str(error)]))
1344        sys.exit(1)
1345
1346    if args.progress:  # Print newline to split stderr from stdout for humans.
1347        logger.info("\n")
1348
1349    if args.stream:
1350        if args.normalize:
1351            output = writer.manifest_text(normalize=True)
1352        else:
1353            output = writer.manifest_text()
1354    elif args.raw:
1355        output = ','.join(writer.data_locators())
1356    elif writer.manifest_locator() is not None:
1357        try:
1358            expiration_notice = ""
1359            if writer.collection_trash_at() is not None:
1360                # Get the local timezone-naive version, and log it with timezone information.
1361                if time.daylight:
1362                    local_trash_at = writer.collection_trash_at().replace(tzinfo=None) - datetime.timedelta(seconds=time.altzone)
1363                else:
1364                    local_trash_at = writer.collection_trash_at().replace(tzinfo=None) - datetime.timedelta(seconds=time.timezone)
1365                expiration_notice = ". It will expire on {} {}.".format(
1366                    local_trash_at.strftime("%Y-%m-%d %H:%M:%S"), time.strftime("%z"))
1367            if args.update_collection:
1368                logger.info(u"Collection updated: '{}'{}".format(
1369                    writer.collection_name(), expiration_notice))
1370            else:
1371                logger.info(u"Collection saved as '{}'{}".format(
1372                    writer.collection_name(), expiration_notice))
1373            if args.portable_data_hash:
1374                output = writer.portable_data_hash()
1375            else:
1376                output = writer.manifest_locator()
1377        except apiclient_errors.Error as error:
1378            logger.error(
1379                "arv-put: Error creating Collection on project: {}.".format(
1380                    error))
1381            status = 1
1382    else:
1383        status = 1
1384
1385    # Print the locator (uuid) of the new collection.
1386    if output is None:
1387        status = status or 1
1388    elif not args.silent:
1389        stdout.write(output)
1390        if not output.endswith('\n'):
1391            stdout.write('\n')
1392
1393    if install_sig_handlers:
1394        arv_cmd.restore_signal_handlers()
1395
1396    if status != 0:
1397        sys.exit(status)
1398
1399    # Success!
1400    return output
1401
1402
1403if __name__ == '__main__':
1404    main()