| Index: third_party/gsutil/gslib/copy_helper.py
|
| diff --git a/third_party/gsutil/gslib/copy_helper.py b/third_party/gsutil/gslib/copy_helper.py
|
| new file mode 100644
|
| index 0000000000000000000000000000000000000000..456b15cf47a011e3c5b71a426c46dffeb520a833
|
| --- /dev/null
|
| +++ b/third_party/gsutil/gslib/copy_helper.py
|
| @@ -0,0 +1,2787 @@
|
| +# -*- coding: utf-8 -*-
|
| +# Copyright 2011 Google Inc. All Rights Reserved.
|
| +# Copyright 2011, Nexenta Systems Inc.
|
| +#
|
| +# Licensed under the Apache License, Version 2.0 (the "License");
|
| +# you may not use this file except in compliance with the License.
|
| +# You may obtain a copy of the License at
|
| +#
|
| +# http://www.apache.org/licenses/LICENSE-2.0
|
| +#
|
| +# Unless required by applicable law or agreed to in writing, software
|
| +# distributed under the License is distributed on an "AS IS" BASIS,
|
| +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
| +# See the License for the specific language governing permissions and
|
| +# limitations under the License.
|
| +"""Helper functions for copy functionality."""
|
| +
|
| +from __future__ import absolute_import
|
| +
|
| +import base64
|
| +from collections import namedtuple
|
| +import csv
|
| +import datetime
|
| +import errno
|
| +import gzip
|
| +from hashlib import md5
|
| +import json
|
| +import logging
|
| +import mimetypes
|
| +import multiprocessing
|
| +import os
|
| +import pickle
|
| +import random
|
| +import re
|
| +import shutil
|
| +import stat
|
| +import subprocess
|
| +import tempfile
|
| +import textwrap
|
| +import time
|
| +import traceback
|
| +
|
| +from boto import config
|
| +import crcmod
|
| +
|
| +import gslib
|
| +from gslib.cloud_api import ArgumentException
|
| +from gslib.cloud_api import CloudApi
|
| +from gslib.cloud_api import NotFoundException
|
| +from gslib.cloud_api import PreconditionException
|
| +from gslib.cloud_api import Preconditions
|
| +from gslib.cloud_api import ResumableDownloadException
|
| +from gslib.cloud_api import ResumableUploadAbortException
|
| +from gslib.cloud_api import ResumableUploadException
|
| +from gslib.cloud_api import ResumableUploadStartOverException
|
| +from gslib.cloud_api_helper import GetDownloadSerializationDict
|
| +from gslib.commands.compose import MAX_COMPOSE_ARITY
|
| +from gslib.commands.config import DEFAULT_PARALLEL_COMPOSITE_UPLOAD_COMPONENT_SIZE
|
| +from gslib.commands.config import DEFAULT_PARALLEL_COMPOSITE_UPLOAD_THRESHOLD
|
| +from gslib.cs_api_map import ApiSelector
|
| +from gslib.daisy_chain_wrapper import DaisyChainWrapper
|
| +from gslib.exception import CommandException
|
| +from gslib.exception import HashMismatchException
|
| +from gslib.file_part import FilePart
|
| +from gslib.hashing_helper import Base64EncodeHash
|
| +from gslib.hashing_helper import CalculateB64EncodedMd5FromContents
|
| +from gslib.hashing_helper import CalculateHashesFromContents
|
| +from gslib.hashing_helper import GetDownloadHashAlgs
|
| +from gslib.hashing_helper import GetUploadHashAlgs
|
| +from gslib.hashing_helper import HashingFileUploadWrapper
|
| +from gslib.parallelism_framework_util import ThreadAndProcessSafeDict
|
| +from gslib.parallelism_framework_util import ThreadSafeDict
|
| +from gslib.progress_callback import ConstructAnnounceText
|
| +from gslib.progress_callback import FileProgressCallbackHandler
|
| +from gslib.progress_callback import ProgressCallbackWithBackoff
|
| +from gslib.resumable_streaming_upload import ResumableStreamingJsonUploadWrapper
|
| +from gslib.storage_url import ContainsWildcard
|
| +from gslib.storage_url import StorageUrlFromString
|
| +from gslib.third_party.storage_apitools import storage_v1_messages as apitools_messages
|
| +from gslib.tracker_file import DeleteTrackerFile
|
| +from gslib.tracker_file import GetTrackerFilePath
|
| +from gslib.tracker_file import RaiseUnwritableTrackerFileException
|
| +from gslib.tracker_file import ReadOrCreateDownloadTrackerFile
|
| +from gslib.tracker_file import TrackerFileType
|
| +from gslib.translation_helper import AddS3MarkerAclToObjectMetadata
|
| +from gslib.translation_helper import CopyObjectMetadata
|
| +from gslib.translation_helper import DEFAULT_CONTENT_TYPE
|
| +from gslib.translation_helper import GenerationFromUrlAndString
|
| +from gslib.translation_helper import ObjectMetadataFromHeaders
|
| +from gslib.translation_helper import PreconditionsFromHeaders
|
| +from gslib.translation_helper import S3MarkerAclFromObjectMetadata
|
| +from gslib.util import CreateLock
|
| +from gslib.util import DEFAULT_FILE_BUFFER_SIZE
|
| +from gslib.util import GetCloudApiInstance
|
| +from gslib.util import GetFileSize
|
| +from gslib.util import GetJsonResumableChunkSize
|
| +from gslib.util import GetMaxRetryDelay
|
| +from gslib.util import GetNumRetries
|
| +from gslib.util import GetStreamFromFileUrl
|
| +from gslib.util import HumanReadableToBytes
|
| +from gslib.util import IS_WINDOWS
|
| +from gslib.util import IsCloudSubdirPlaceholder
|
| +from gslib.util import MakeHumanReadable
|
| +from gslib.util import MIN_SIZE_COMPUTE_LOGGING
|
| +from gslib.util import MultiprocessingIsAvailable
|
| +from gslib.util import ResumableThreshold
|
| +from gslib.util import TEN_MIB
|
| +from gslib.util import UTF8
|
| +from gslib.wildcard_iterator import CreateWildcardIterator
|
| +
|
| +# pylint: disable=g-import-not-at-top
|
| +if IS_WINDOWS:
|
| + import msvcrt
|
| + from ctypes import c_int
|
| + from ctypes import c_uint64
|
| + from ctypes import c_char_p
|
| + from ctypes import c_wchar_p
|
| + from ctypes import windll
|
| + from ctypes import POINTER
|
| + from ctypes import WINFUNCTYPE
|
| + from ctypes import WinError
|
| +
|
| +# Declare copy_helper_opts as a global because namedtuple isn't aware of
|
| +# assigning to a class member (which breaks pickling done by multiprocessing).
|
| +# For details see
|
| +# http://stackoverflow.com/questions/16377215/how-to-pickle-a-namedtuple-instance-correctly
|
| +# Similarly can't pickle logger.
|
| +# pylint: disable=global-at-module-level
|
| +global global_copy_helper_opts, global_logger
|
| +
|
| +# In-memory map of local files that are currently opened for write. Used to
|
| +# ensure that if we write to the same file twice (say, for example, because the
|
| +# user specified two identical source URLs), the writes occur serially.
|
| +global open_files_map
|
| +open_files_map = (
|
| + ThreadSafeDict() if (IS_WINDOWS or not MultiprocessingIsAvailable()[0])
|
| + else ThreadAndProcessSafeDict(multiprocessing.Manager()))
|
| +
|
| +# For debugging purposes; if True, files and objects that fail hash validation
|
| +# will be saved with the below suffix appended.
|
| +_RENAME_ON_HASH_MISMATCH = False
|
| +_RENAME_ON_HASH_MISMATCH_SUFFIX = '_corrupt'
|
| +
|
| +PARALLEL_UPLOAD_TEMP_NAMESPACE = (
|
| + u'/gsutil/tmp/parallel_composite_uploads/for_details_see/gsutil_help_cp/')
|
| +
|
| +PARALLEL_UPLOAD_STATIC_SALT = u"""
|
| +PARALLEL_UPLOAD_SALT_TO_PREVENT_COLLISIONS.
|
| +The theory is that no user will have prepended this to the front of
|
| +one of their object names and then done an MD5 hash of the name, and
|
| +then prepended PARALLEL_UPLOAD_TEMP_NAMESPACE to the front of their object
|
| +name. Note that there will be no problems with object name length since we
|
| +hash the original name.
|
| +"""
|
| +
|
| +# When uploading a file, get the following fields in the response for
|
| +# filling in command output and manifests.
|
| +UPLOAD_RETURN_FIELDS = ['crc32c', 'etag', 'generation', 'md5Hash', 'size']
|
| +
|
| +# This tuple is used only to encapsulate the arguments needed for
|
| +# command.Apply() in the parallel composite upload case.
|
| +# Note that content_type is used instead of a full apitools Object() because
|
| +# apitools objects are not picklable.
|
| +# filename: String name of file.
|
| +# file_start: start byte of file (may be in the middle of a file for partitioned
|
| +# files).
|
| +# file_length: length of upload (may not be the entire length of a file for
|
| +# partitioned files).
|
| +# src_url: FileUrl describing the source file.
|
| +# dst_url: CloudUrl describing the destination component file.
|
| +# canned_acl: canned_acl to apply to the uploaded file/component.
|
| +# content_type: content-type for final object, used for setting content-type
|
| +# of components and final object.
|
| +# tracker_file: tracker file for this component.
|
| +# tracker_file_lock: tracker file lock for tracker file(s).
|
| +PerformParallelUploadFileToObjectArgs = namedtuple(
|
| + 'PerformParallelUploadFileToObjectArgs',
|
| + 'filename file_start file_length src_url dst_url canned_acl '
|
| + 'content_type tracker_file tracker_file_lock')
|
| +
|
| +ObjectFromTracker = namedtuple('ObjectFromTracker',
|
| + 'object_name generation')
|
| +
|
| +# TODO: Refactor this file to be less cumbersome. In particular, some of the
|
| +# different paths (e.g., uploading a file to an object vs. downloading an
|
| +# object to a file) could be split into separate files.
|
| +
|
| +# Chunk size to use while zipping/unzipping gzip files.
|
| +GZIP_CHUNK_SIZE = 8192
|
| +
|
| +PARALLEL_COMPOSITE_SUGGESTION_THRESHOLD = 150 * 1024 * 1024
|
| +
|
| +# S3 requires special Multipart upload logic (that we currently don't implement)
|
| +# for files > 5GiB in size.
|
| +S3_MAX_UPLOAD_SIZE = 5 * 1024 * 1024 * 1024
|
| +
|
| +suggested_parallel_composites = False
|
| +
|
| +
|
| +class FileConcurrencySkipError(Exception):
|
| + """Raised when skipping a file due to a concurrent, duplicate copy."""
|
| +
|
| +
|
| +def _RmExceptionHandler(cls, e):
|
| + """Simple exception handler to allow post-completion status."""
|
| + cls.logger.error(str(e))
|
| +
|
| +
|
| +def _ParallelUploadCopyExceptionHandler(cls, e):
|
| + """Simple exception handler to allow post-completion status."""
|
| + cls.logger.error(str(e))
|
| + cls.op_failure_count += 1
|
| + cls.logger.debug('\n\nEncountered exception while copying:\n%s\n',
|
| + traceback.format_exc())
|
| +
|
| +
|
| +def _PerformParallelUploadFileToObject(cls, args, thread_state=None):
|
| + """Function argument to Apply for performing parallel composite uploads.
|
| +
|
| + Args:
|
| + cls: Calling Command class.
|
| + args: PerformParallelUploadFileToObjectArgs tuple describing the target.
|
| + thread_state: gsutil Cloud API instance to use for the operation.
|
| +
|
| + Returns:
|
| + StorageUrl representing a successfully uploaded component.
|
| + """
|
| + fp = FilePart(args.filename, args.file_start, args.file_length)
|
| + gsutil_api = GetCloudApiInstance(cls, thread_state=thread_state)
|
| + with fp:
|
| + # We take many precautions with the component names that make collisions
|
| + # effectively impossible. Specifying preconditions will just allow us to
|
| + # reach a state in which uploads will always fail on retries.
|
| + preconditions = None
|
| +
|
| + # Fill in content type if one was provided.
|
| + dst_object_metadata = apitools_messages.Object(
|
| + name=args.dst_url.object_name,
|
| + bucket=args.dst_url.bucket_name,
|
| + contentType=args.content_type)
|
| +
|
| + try:
|
| + if global_copy_helper_opts.canned_acl:
|
| + # No canned ACL support in JSON, force XML API to be used for
|
| + # upload/copy operations.
|
| + orig_prefer_api = gsutil_api.prefer_api
|
| + gsutil_api.prefer_api = ApiSelector.XML
|
| + ret = _UploadFileToObject(args.src_url, fp, args.file_length,
|
| + args.dst_url, dst_object_metadata,
|
| + preconditions, gsutil_api, cls.logger, cls,
|
| + _ParallelUploadCopyExceptionHandler,
|
| + gzip_exts=None, allow_splitting=False)
|
| + finally:
|
| + if global_copy_helper_opts.canned_acl:
|
| + gsutil_api.prefer_api = orig_prefer_api
|
| +
|
| + component = ret[2]
|
| + _AppendComponentTrackerToParallelUploadTrackerFile(
|
| + args.tracker_file, component, args.tracker_file_lock)
|
| + return ret
|
| +
|
| +
|
| +CopyHelperOpts = namedtuple('CopyHelperOpts', [
|
| + 'perform_mv',
|
| + 'no_clobber',
|
| + 'daisy_chain',
|
| + 'read_args_from_stdin',
|
| + 'print_ver',
|
| + 'use_manifest',
|
| + 'preserve_acl',
|
| + 'canned_acl',
|
| + 'skip_unsupported_objects',
|
| + 'test_callback_file'])
|
| +
|
| +
|
| +# pylint: disable=global-variable-undefined
|
| +def CreateCopyHelperOpts(perform_mv=False, no_clobber=False, daisy_chain=False,
|
| + read_args_from_stdin=False, print_ver=False,
|
| + use_manifest=False, preserve_acl=False,
|
| + canned_acl=None, skip_unsupported_objects=False,
|
| + test_callback_file=None):
|
| + """Creates CopyHelperOpts for passing options to CopyHelper."""
|
| + # We create a tuple with union of options needed by CopyHelper and any
|
| + # copy-related functionality in CpCommand, RsyncCommand, or Command class.
|
| + global global_copy_helper_opts
|
| + global_copy_helper_opts = CopyHelperOpts(
|
| + perform_mv=perform_mv,
|
| + no_clobber=no_clobber,
|
| + daisy_chain=daisy_chain,
|
| + read_args_from_stdin=read_args_from_stdin,
|
| + print_ver=print_ver,
|
| + use_manifest=use_manifest,
|
| + preserve_acl=preserve_acl,
|
| + canned_acl=canned_acl,
|
| + skip_unsupported_objects=skip_unsupported_objects,
|
| + test_callback_file=test_callback_file)
|
| + return global_copy_helper_opts
|
| +
|
| +
|
| +# pylint: disable=global-variable-undefined
|
| +# pylint: disable=global-variable-not-assigned
|
| +def GetCopyHelperOpts():
|
| + """Returns namedtuple holding CopyHelper options."""
|
| + global global_copy_helper_opts
|
| + return global_copy_helper_opts
|
| +
|
| +
|
| +def _SelectDownloadStrategy(dst_url):
|
| + """Get download strategy based on the destination object.
|
| +
|
| + Args:
|
| + dst_url: Destination StorageUrl.
|
| +
|
| + Returns:
|
| + gsutil Cloud API DownloadStrategy.
|
| + """
|
| + dst_is_special = False
|
| + if dst_url.IsFileUrl():
|
| + # Check explicitly first because os.stat doesn't work on 'nul' in Windows.
|
| + if dst_url.object_name == os.devnull:
|
| + dst_is_special = True
|
| + try:
|
| + mode = os.stat(dst_url.object_name).st_mode
|
| + if stat.S_ISCHR(mode):
|
| + dst_is_special = True
|
| + except OSError:
|
| + pass
|
| +
|
| + if dst_is_special:
|
| + return CloudApi.DownloadStrategy.ONE_SHOT
|
| + else:
|
| + return CloudApi.DownloadStrategy.RESUMABLE
|
| +
|
| +
|
| +def _GetUploadTrackerData(tracker_file_name, logger):
|
| + """Reads tracker data from an upload tracker file if it exists.
|
| +
|
| + Args:
|
| + tracker_file_name: Tracker file name for this upload.
|
| + logger: for outputting log messages.
|
| +
|
| + Returns:
|
| + Serialization data if the tracker file already exists (resume existing
|
| + upload), None otherwise.
|
| + """
|
| + tracker_file = None
|
| +
|
| + # If we already have a matching tracker file, get the serialization data
|
| + # so that we can resume the upload.
|
| + try:
|
| + tracker_file = open(tracker_file_name, 'r')
|
| + tracker_data = tracker_file.read()
|
| + return tracker_data
|
| + except IOError as e:
|
| + # Ignore non-existent file (happens first time a upload is attempted on an
|
| + # object, or when re-starting an upload after a
|
| + # ResumableUploadStartOverException), but warn user for other errors.
|
| + if e.errno != errno.ENOENT:
|
| + logger.warn('Couldn\'t read upload tracker file (%s): %s. Restarting '
|
| + 'upload from scratch.', tracker_file_name, e.strerror)
|
| + finally:
|
| + if tracker_file:
|
| + tracker_file.close()
|
| +
|
| +
|
| +def InsistDstUrlNamesContainer(exp_dst_url, have_existing_dst_container,
|
| + command_name):
|
| + """Ensures the destination URL names a container.
|
| +
|
| + Acceptable containers include directory, bucket, bucket
|
| + subdir, and non-existent bucket subdir.
|
| +
|
| + Args:
|
| + exp_dst_url: Wildcard-expanded destination StorageUrl.
|
| + have_existing_dst_container: bool indicator of whether exp_dst_url
|
| + names a container (directory, bucket, or existing bucket subdir).
|
| + command_name: Name of command making call. May not be the same as the
|
| + calling class's self.command_name in the case of commands implemented
|
| + atop other commands (like mv command).
|
| +
|
| + Raises:
|
| + CommandException: if the URL being checked does not name a container.
|
| + """
|
| + if ((exp_dst_url.IsFileUrl() and not exp_dst_url.IsDirectory()) or
|
| + (exp_dst_url.IsCloudUrl() and exp_dst_url.IsBucket()
|
| + and not have_existing_dst_container)):
|
| + raise CommandException('Destination URL must name a directory, bucket, '
|
| + 'or bucket\nsubdirectory for the multiple '
|
| + 'source form of the %s command.' % command_name)
|
| +
|
| +
|
| +def _ShouldTreatDstUrlAsBucketSubDir(have_multiple_srcs, dst_url,
|
| + have_existing_dest_subdir,
|
| + src_url_names_container,
|
| + recursion_requested):
|
| + """Checks whether dst_url should be treated as a bucket "sub-directory".
|
| +
|
| + The decision about whether something constitutes a bucket "sub-directory"
|
| + depends on whether there are multiple sources in this request and whether
|
| + there is an existing bucket subdirectory. For example, when running the
|
| + command:
|
| + gsutil cp file gs://bucket/abc
|
| + if there's no existing gs://bucket/abc bucket subdirectory we should copy
|
| + file to the object gs://bucket/abc. In contrast, if
|
| + there's an existing gs://bucket/abc bucket subdirectory we should copy
|
| + file to gs://bucket/abc/file. And regardless of whether gs://bucket/abc
|
| + exists, when running the command:
|
| + gsutil cp file1 file2 gs://bucket/abc
|
| + we should copy file1 to gs://bucket/abc/file1 (and similarly for file2).
|
| + Finally, for recursive copies, if the source is a container then we should
|
| + copy to a container as the target. For example, when running the command:
|
| + gsutil cp -r dir1 gs://bucket/dir2
|
| + we should copy the subtree of dir1 to gs://bucket/dir2.
|
| +
|
| + Note that we don't disallow naming a bucket "sub-directory" where there's
|
| + already an object at that URL. For example it's legitimate (albeit
|
| + confusing) to have an object called gs://bucket/dir and
|
| + then run the command
|
| + gsutil cp file1 file2 gs://bucket/dir
|
| + Doing so will end up with objects gs://bucket/dir, gs://bucket/dir/file1,
|
| + and gs://bucket/dir/file2.
|
| +
|
| + Args:
|
| + have_multiple_srcs: Bool indicator of whether this is a multi-source
|
| + operation.
|
| + dst_url: StorageUrl to check.
|
| + have_existing_dest_subdir: bool indicator whether dest is an existing
|
| + subdirectory.
|
| + src_url_names_container: bool indicator of whether the source URL
|
| + is a container.
|
| + recursion_requested: True if a recursive operation has been requested.
|
| +
|
| + Returns:
|
| + bool indicator.
|
| + """
|
| + if have_existing_dest_subdir:
|
| + return True
|
| + if dst_url.IsCloudUrl():
|
| + return (have_multiple_srcs or
|
| + (src_url_names_container and recursion_requested))
|
| +
|
| +
|
| +def _ShouldTreatDstUrlAsSingleton(have_multiple_srcs,
|
| + have_existing_dest_subdir, dst_url,
|
| + recursion_requested):
|
| + """Checks that dst_url names a single file/object after wildcard expansion.
|
| +
|
| + It is possible that an object path might name a bucket sub-directory.
|
| +
|
| + Args:
|
| + have_multiple_srcs: Bool indicator of whether this is a multi-source
|
| + operation.
|
| + have_existing_dest_subdir: bool indicator whether dest is an existing
|
| + subdirectory.
|
| + dst_url: StorageUrl to check.
|
| + recursion_requested: True if a recursive operation has been requested.
|
| +
|
| + Returns:
|
| + bool indicator.
|
| + """
|
| + if recursion_requested:
|
| + return False
|
| + if dst_url.IsFileUrl():
|
| + return not dst_url.IsDirectory()
|
| + else: # dst_url.IsCloudUrl()
|
| + return (not have_multiple_srcs and
|
| + not have_existing_dest_subdir and
|
| + dst_url.IsObject())
|
| +
|
| +
|
| +def ConstructDstUrl(src_url, exp_src_url, src_url_names_container,
|
| + have_multiple_srcs, exp_dst_url, have_existing_dest_subdir,
|
| + recursion_requested):
|
| + """Constructs the destination URL for a given exp_src_url/exp_dst_url pair.
|
| +
|
| + Uses context-dependent naming rules that mimic Linux cp and mv behavior.
|
| +
|
| + Args:
|
| + src_url: Source StorageUrl to be copied.
|
| + exp_src_url: Single StorageUrl from wildcard expansion of src_url.
|
| + src_url_names_container: True if src_url names a container (including the
|
| + case of a wildcard-named bucket subdir (like gs://bucket/abc,
|
| + where gs://bucket/abc/* matched some objects).
|
| + have_multiple_srcs: True if this is a multi-source request. This can be
|
| + true if src_url wildcard-expanded to multiple URLs or if there were
|
| + multiple source URLs in the request.
|
| + exp_dst_url: the expanded StorageUrl requested for the cp destination.
|
| + Final written path is constructed from this plus a context-dependent
|
| + variant of src_url.
|
| + have_existing_dest_subdir: bool indicator whether dest is an existing
|
| + subdirectory.
|
| + recursion_requested: True if a recursive operation has been requested.
|
| +
|
| + Returns:
|
| + StorageUrl to use for copy.
|
| +
|
| + Raises:
|
| + CommandException if destination object name not specified for
|
| + source and source is a stream.
|
| + """
|
| + if _ShouldTreatDstUrlAsSingleton(
|
| + have_multiple_srcs, have_existing_dest_subdir, exp_dst_url,
|
| + recursion_requested):
|
| + # We're copying one file or object to one file or object.
|
| + return exp_dst_url
|
| +
|
| + if exp_src_url.IsFileUrl() and exp_src_url.IsStream():
|
| + if have_existing_dest_subdir:
|
| + raise CommandException('Destination object name needed when '
|
| + 'source is a stream')
|
| + return exp_dst_url
|
| +
|
| + if not recursion_requested and not have_multiple_srcs:
|
| + # We're copying one file or object to a subdirectory. Append final comp
|
| + # of exp_src_url to exp_dst_url.
|
| + src_final_comp = exp_src_url.object_name.rpartition(src_url.delim)[-1]
|
| + return StorageUrlFromString('%s%s%s' % (
|
| + exp_dst_url.url_string.rstrip(exp_dst_url.delim),
|
| + exp_dst_url.delim, src_final_comp))
|
| +
|
| + # Else we're copying multiple sources to a directory, bucket, or a bucket
|
| + # "sub-directory".
|
| +
|
| + # Ensure exp_dst_url ends in delim char if we're doing a multi-src copy or
|
| + # a copy to a directory. (The check for copying to a directory needs
|
| + # special-case handling so that the command:
|
| + # gsutil cp gs://bucket/obj dir
|
| + # will turn into file://dir/ instead of file://dir -- the latter would cause
|
| + # the file "dirobj" to be created.)
|
| + # Note: need to check have_multiple_srcs or src_url.names_container()
|
| + # because src_url could be a bucket containing a single object, named
|
| + # as gs://bucket.
|
| + if ((have_multiple_srcs or src_url_names_container or
|
| + (exp_dst_url.IsFileUrl() and exp_dst_url.IsDirectory()))
|
| + and not exp_dst_url.url_string.endswith(exp_dst_url.delim)):
|
| + exp_dst_url = StorageUrlFromString('%s%s' % (exp_dst_url.url_string,
|
| + exp_dst_url.delim))
|
| +
|
| + # Making naming behavior match how things work with local Linux cp and mv
|
| + # operations depends on many factors, including whether the destination is a
|
| + # container, the plurality of the source(s), and whether the mv command is
|
| + # being used:
|
| + # 1. For the "mv" command that specifies a non-existent destination subdir,
|
| + # renaming should occur at the level of the src subdir, vs appending that
|
| + # subdir beneath the dst subdir like is done for copying. For example:
|
| + # gsutil rm -r gs://bucket
|
| + # gsutil cp -r dir1 gs://bucket
|
| + # gsutil cp -r dir2 gs://bucket/subdir1
|
| + # gsutil mv gs://bucket/subdir1 gs://bucket/subdir2
|
| + # would (if using cp naming behavior) end up with paths like:
|
| + # gs://bucket/subdir2/subdir1/dir2/.svn/all-wcprops
|
| + # whereas mv naming behavior should result in:
|
| + # gs://bucket/subdir2/dir2/.svn/all-wcprops
|
| + # 2. Copying from directories, buckets, or bucket subdirs should result in
|
| + # objects/files mirroring the source directory hierarchy. For example:
|
| + # gsutil cp dir1/dir2 gs://bucket
|
| + # should create the object gs://bucket/dir2/file2, assuming dir1/dir2
|
| + # contains file2).
|
| + # To be consistent with Linux cp behavior, there's one more wrinkle when
|
| + # working with subdirs: The resulting object names depend on whether the
|
| + # destination subdirectory exists. For example, if gs://bucket/subdir
|
| + # exists, the command:
|
| + # gsutil cp -r dir1/dir2 gs://bucket/subdir
|
| + # should create objects named like gs://bucket/subdir/dir2/a/b/c. In
|
| + # contrast, if gs://bucket/subdir does not exist, this same command
|
| + # should create objects named like gs://bucket/subdir/a/b/c.
|
| + # 3. Copying individual files or objects to dirs, buckets or bucket subdirs
|
| + # should result in objects/files named by the final source file name
|
| + # component. Example:
|
| + # gsutil cp dir1/*.txt gs://bucket
|
| + # should create the objects gs://bucket/f1.txt and gs://bucket/f2.txt,
|
| + # assuming dir1 contains f1.txt and f2.txt.
|
| +
|
| + recursive_move_to_new_subdir = False
|
| + if (global_copy_helper_opts.perform_mv and recursion_requested
|
| + and src_url_names_container and not have_existing_dest_subdir):
|
| + # Case 1. Handle naming rules for bucket subdir mv. Here we want to
|
| + # line up the src_url against its expansion, to find the base to build
|
| + # the new name. For example, running the command:
|
| + # gsutil mv gs://bucket/abcd gs://bucket/xyz
|
| + # when processing exp_src_url=gs://bucket/abcd/123
|
| + # exp_src_url_tail should become /123
|
| + # Note: mv.py code disallows wildcard specification of source URL.
|
| + recursive_move_to_new_subdir = True
|
| + exp_src_url_tail = (
|
| + exp_src_url.url_string[len(src_url.url_string):])
|
| + dst_key_name = '%s/%s' % (exp_dst_url.object_name.rstrip('/'),
|
| + exp_src_url_tail.strip('/'))
|
| +
|
| + elif src_url_names_container and (exp_dst_url.IsCloudUrl() or
|
| + exp_dst_url.IsDirectory()):
|
| + # Case 2. Container copy to a destination other than a file.
|
| + # Build dst_key_name from subpath of exp_src_url past
|
| + # where src_url ends. For example, for src_url=gs://bucket/ and
|
| + # exp_src_url=gs://bucket/src_subdir/obj, dst_key_name should be
|
| + # src_subdir/obj.
|
| + src_url_path_sans_final_dir = GetPathBeforeFinalDir(src_url)
|
| + dst_key_name = exp_src_url.versionless_url_string[
|
| + len(src_url_path_sans_final_dir):].lstrip(src_url.delim)
|
| + # Handle case where dst_url is a non-existent subdir.
|
| + if not have_existing_dest_subdir:
|
| + dst_key_name = dst_key_name.partition(src_url.delim)[-1]
|
| + # Handle special case where src_url was a directory named with '.' or
|
| + # './', so that running a command like:
|
| + # gsutil cp -r . gs://dest
|
| + # will produce obj names of the form gs://dest/abc instead of
|
| + # gs://dest/./abc.
|
| + if dst_key_name.startswith('.%s' % os.sep):
|
| + dst_key_name = dst_key_name[2:]
|
| +
|
| + else:
|
| + # Case 3.
|
| + dst_key_name = exp_src_url.object_name.rpartition(src_url.delim)[-1]
|
| +
|
| + if (not recursive_move_to_new_subdir and (
|
| + exp_dst_url.IsFileUrl() or _ShouldTreatDstUrlAsBucketSubDir(
|
| + have_multiple_srcs, exp_dst_url, have_existing_dest_subdir,
|
| + src_url_names_container, recursion_requested))):
|
| + if exp_dst_url.object_name and exp_dst_url.object_name.endswith(
|
| + exp_dst_url.delim):
|
| + dst_key_name = '%s%s%s' % (
|
| + exp_dst_url.object_name.rstrip(exp_dst_url.delim),
|
| + exp_dst_url.delim, dst_key_name)
|
| + else:
|
| + delim = exp_dst_url.delim if exp_dst_url.object_name else ''
|
| + dst_key_name = '%s%s%s' % (exp_dst_url.object_name or '',
|
| + delim, dst_key_name)
|
| +
|
| + new_exp_dst_url = exp_dst_url.Clone()
|
| + new_exp_dst_url.object_name = dst_key_name.replace(src_url.delim,
|
| + exp_dst_url.delim)
|
| + return new_exp_dst_url
|
| +
|
| +
|
| +def _CreateDigestsFromDigesters(digesters):
|
| + digests = {}
|
| + if digesters:
|
| + for alg in digesters:
|
| + digests[alg] = base64.encodestring(
|
| + digesters[alg].digest()).rstrip('\n')
|
| + return digests
|
| +
|
| +
|
| +def _CreateDigestsFromLocalFile(logger, algs, file_name, src_obj_metadata):
|
| + """Creates a base64 CRC32C and/or MD5 digest from file_name.
|
| +
|
| + Args:
|
| + logger: for outputting log messages.
|
| + algs: list of algorithms to compute.
|
| + file_name: file to digest.
|
| + src_obj_metadata: metadta of source object.
|
| +
|
| + Returns:
|
| + Dict of algorithm name : base 64 encoded digest
|
| + """
|
| + hash_dict = {}
|
| + if 'md5' in algs:
|
| + if src_obj_metadata.size and src_obj_metadata.size > TEN_MIB:
|
| + logger.info(
|
| + 'Computing MD5 for %s...', file_name)
|
| + hash_dict['md5'] = md5()
|
| + if 'crc32c' in algs:
|
| + hash_dict['crc32c'] = crcmod.predefined.Crc('crc-32c')
|
| + with open(file_name, 'rb') as fp:
|
| + CalculateHashesFromContents(
|
| + fp, hash_dict, ProgressCallbackWithBackoff(
|
| + src_obj_metadata.size,
|
| + FileProgressCallbackHandler(
|
| + ConstructAnnounceText('Hashing', file_name), logger).call))
|
| + digests = {}
|
| + for alg_name, digest in hash_dict.iteritems():
|
| + digests[alg_name] = Base64EncodeHash(digest.hexdigest())
|
| + return digests
|
| +
|
| +
|
| +def _CheckCloudHashes(logger, src_url, dst_url, src_obj_metadata,
|
| + dst_obj_metadata):
|
| + """Validates integrity of two cloud objects copied via daisy-chain.
|
| +
|
| + Args:
|
| + logger: for outputting log messages.
|
| + src_url: CloudUrl for source cloud object.
|
| + dst_url: CloudUrl for destination cloud object.
|
| + src_obj_metadata: Cloud Object metadata for object being downloaded from.
|
| + dst_obj_metadata: Cloud Object metadata for object being uploaded to.
|
| +
|
| + Raises:
|
| + CommandException: if cloud digests don't match local digests.
|
| + """
|
| + checked_one = False
|
| + download_hashes = {}
|
| + upload_hashes = {}
|
| + if src_obj_metadata.md5Hash:
|
| + download_hashes['md5'] = src_obj_metadata.md5Hash
|
| + if src_obj_metadata.crc32c:
|
| + download_hashes['crc32c'] = src_obj_metadata.crc32c
|
| + if dst_obj_metadata.md5Hash:
|
| + upload_hashes['md5'] = dst_obj_metadata.md5Hash
|
| + if dst_obj_metadata.crc32c:
|
| + upload_hashes['crc32c'] = dst_obj_metadata.crc32c
|
| +
|
| + for alg, upload_b64_digest in upload_hashes.iteritems():
|
| + if alg not in download_hashes:
|
| + continue
|
| +
|
| + download_b64_digest = download_hashes[alg]
|
| + logger.debug(
|
| + 'Comparing source vs destination %s-checksum for %s. (%s/%s)', alg,
|
| + dst_url, download_b64_digest, upload_b64_digest)
|
| + if download_b64_digest != upload_b64_digest:
|
| + raise HashMismatchException(
|
| + '%s signature for source object (%s) doesn\'t match '
|
| + 'destination object digest (%s). Object (%s) will be deleted.' % (
|
| + alg, download_b64_digest, upload_b64_digest, dst_url))
|
| + checked_one = True
|
| + if not checked_one:
|
| + # One known way this can currently happen is when downloading objects larger
|
| + # than 5 GiB from S3 (for which the etag is not an MD5).
|
| + logger.warn(
|
| + 'WARNING: Found no hashes to validate object downloaded from %s and '
|
| + 'uploaded to %s. Integrity cannot be assured without hashes.',
|
| + src_url, dst_url)
|
| +
|
| +
|
| +def _CheckHashes(logger, obj_url, obj_metadata, file_name, digests,
|
| + is_upload=False):
|
| + """Validates integrity by comparing cloud digest to local digest.
|
| +
|
| + Args:
|
| + logger: for outputting log messages.
|
| + obj_url: CloudUrl for cloud object.
|
| + obj_metadata: Cloud Object being downloaded from or uploaded to.
|
| + file_name: Local file name on disk being downloaded to or uploaded from.
|
| + digests: Computed Digests for the object.
|
| + is_upload: If true, comparing for an uploaded object (controls logging).
|
| +
|
| + Raises:
|
| + CommandException: if cloud digests don't match local digests.
|
| + """
|
| + local_hashes = digests
|
| + cloud_hashes = {}
|
| + if obj_metadata.md5Hash:
|
| + cloud_hashes['md5'] = obj_metadata.md5Hash.rstrip('\n')
|
| + if obj_metadata.crc32c:
|
| + cloud_hashes['crc32c'] = obj_metadata.crc32c.rstrip('\n')
|
| +
|
| + checked_one = False
|
| + for alg in local_hashes:
|
| + if alg not in cloud_hashes:
|
| + continue
|
| +
|
| + local_b64_digest = local_hashes[alg]
|
| + cloud_b64_digest = cloud_hashes[alg]
|
| + logger.debug(
|
| + 'Comparing local vs cloud %s-checksum for %s. (%s/%s)', alg, file_name,
|
| + local_b64_digest, cloud_b64_digest)
|
| + if local_b64_digest != cloud_b64_digest:
|
| +
|
| + raise HashMismatchException(
|
| + '%s signature computed for local file (%s) doesn\'t match '
|
| + 'cloud-supplied digest (%s). %s (%s) will be deleted.' % (
|
| + alg, local_b64_digest, cloud_b64_digest,
|
| + 'Cloud object' if is_upload else 'Local file',
|
| + obj_url if is_upload else file_name))
|
| + checked_one = True
|
| + if not checked_one:
|
| + if is_upload:
|
| + logger.warn(
|
| + 'WARNING: Found no hashes to validate object uploaded to %s. '
|
| + 'Integrity cannot be assured without hashes.', obj_url)
|
| + else:
|
| + # One known way this can currently happen is when downloading objects larger
|
| + # than 5 GB from S3 (for which the etag is not an MD5).
|
| + logger.warn(
|
| + 'WARNING: Found no hashes to validate object downloaded to %s. '
|
| + 'Integrity cannot be assured without hashes.', file_name)
|
| +
|
| +
|
| +def IsNoClobberServerException(e):
|
| + """Checks to see if the server attempted to clobber a file.
|
| +
|
| + In this case we specified via a precondition that we didn't want the file
|
| + clobbered.
|
| +
|
| + Args:
|
| + e: The Exception that was generated by a failed copy operation
|
| +
|
| + Returns:
|
| + bool indicator - True indicates that the server did attempt to clobber
|
| + an existing file.
|
| + """
|
| + return ((isinstance(e, PreconditionException)) or
|
| + (isinstance(e, ResumableUploadException) and '412' in e.message))
|
| +
|
| +
|
| +def CheckForDirFileConflict(exp_src_url, dst_url):
|
| + """Checks whether copying exp_src_url into dst_url is not possible.
|
| +
|
| + This happens if a directory exists in local file system where a file
|
| + needs to go or vice versa. In that case we print an error message and
|
| + exits. Example: if the file "./x" exists and you try to do:
|
| + gsutil cp gs://mybucket/x/y .
|
| + the request can't succeed because it requires a directory where
|
| + the file x exists.
|
| +
|
| + Note that we don't enforce any corresponding restrictions for buckets,
|
| + because the flat namespace semantics for buckets doesn't prohibit such
|
| + cases the way hierarchical file systems do. For example, if a bucket
|
| + contains an object called gs://bucket/dir and then you run the command:
|
| + gsutil cp file1 file2 gs://bucket/dir
|
| + you'll end up with objects gs://bucket/dir, gs://bucket/dir/file1, and
|
| + gs://bucket/dir/file2.
|
| +
|
| + Args:
|
| + exp_src_url: Expanded source StorageUrl.
|
| + dst_url: Destination StorageUrl.
|
| +
|
| + Raises:
|
| + CommandException: if errors encountered.
|
| + """
|
| + if dst_url.IsCloudUrl():
|
| + # The problem can only happen for file destination URLs.
|
| + return
|
| + dst_path = dst_url.object_name
|
| + final_dir = os.path.dirname(dst_path)
|
| + if os.path.isfile(final_dir):
|
| + raise CommandException('Cannot retrieve %s because a file exists '
|
| + 'where a directory needs to be created (%s).' %
|
| + (exp_src_url.url_string, final_dir))
|
| + if os.path.isdir(dst_path):
|
| + raise CommandException('Cannot retrieve %s because a directory exists '
|
| + '(%s) where the file needs to be created.' %
|
| + (exp_src_url.url_string, dst_path))
|
| +
|
| +
|
| +def _PartitionFile(fp, file_size, src_url, content_type, canned_acl,
|
| + dst_bucket_url, random_prefix, tracker_file,
|
| + tracker_file_lock):
|
| + """Partitions a file into FilePart objects to be uploaded and later composed.
|
| +
|
| + These objects, when composed, will match the original file. This entails
|
| + splitting the file into parts, naming and forming a destination URL for each
|
| + part, and also providing the PerformParallelUploadFileToObjectArgs
|
| + corresponding to each part.
|
| +
|
| + Args:
|
| + fp: The file object to be partitioned.
|
| + file_size: The size of fp, in bytes.
|
| + src_url: Source FileUrl from the original command.
|
| + content_type: content type for the component and final objects.
|
| + canned_acl: The user-provided canned_acl, if applicable.
|
| + dst_bucket_url: CloudUrl for the destination bucket
|
| + random_prefix: The randomly-generated prefix used to prevent collisions
|
| + among the temporary component names.
|
| + tracker_file: The path to the parallel composite upload tracker file.
|
| + tracker_file_lock: The lock protecting access to the tracker file.
|
| +
|
| + Returns:
|
| + dst_args: The destination URIs for the temporary component objects.
|
| + """
|
| + parallel_composite_upload_component_size = HumanReadableToBytes(
|
| + config.get('GSUtil', 'parallel_composite_upload_component_size',
|
| + DEFAULT_PARALLEL_COMPOSITE_UPLOAD_COMPONENT_SIZE))
|
| + (num_components, component_size) = _GetPartitionInfo(
|
| + file_size, MAX_COMPOSE_ARITY, parallel_composite_upload_component_size)
|
| +
|
| + dst_args = {} # Arguments to create commands and pass to subprocesses.
|
| + file_names = [] # Used for the 2-step process of forming dst_args.
|
| + for i in range(num_components):
|
| + # "Salt" the object name with something a user is very unlikely to have
|
| + # used in an object name, then hash the extended name to make sure
|
| + # we don't run into problems with name length. Using a deterministic
|
| + # naming scheme for the temporary components allows users to take
|
| + # advantage of resumable uploads for each component.
|
| + encoded_name = (PARALLEL_UPLOAD_STATIC_SALT + fp.name).encode(UTF8)
|
| + content_md5 = md5()
|
| + content_md5.update(encoded_name)
|
| + digest = content_md5.hexdigest()
|
| + temp_file_name = (random_prefix + PARALLEL_UPLOAD_TEMP_NAMESPACE +
|
| + digest + '_' + str(i))
|
| + tmp_dst_url = dst_bucket_url.Clone()
|
| + tmp_dst_url.object_name = temp_file_name
|
| +
|
| + if i < (num_components - 1):
|
| + # Every component except possibly the last is the same size.
|
| + file_part_length = component_size
|
| + else:
|
| + # The last component just gets all of the remaining bytes.
|
| + file_part_length = (file_size - ((num_components -1) * component_size))
|
| + offset = i * component_size
|
| + func_args = PerformParallelUploadFileToObjectArgs(
|
| + fp.name, offset, file_part_length, src_url, tmp_dst_url, canned_acl,
|
| + content_type, tracker_file, tracker_file_lock)
|
| + file_names.append(temp_file_name)
|
| + dst_args[temp_file_name] = func_args
|
| +
|
| + return dst_args
|
| +
|
| +
|
| +def _DoParallelCompositeUpload(fp, src_url, dst_url, dst_obj_metadata,
|
| + canned_acl, file_size, preconditions, gsutil_api,
|
| + command_obj, copy_exception_handler):
|
| + """Uploads a local file to a cloud object using parallel composite upload.
|
| +
|
| + The file is partitioned into parts, and then the parts are uploaded in
|
| + parallel, composed to form the original destination object, and deleted.
|
| +
|
| + Args:
|
| + fp: The file object to be uploaded.
|
| + src_url: FileUrl representing the local file.
|
| + dst_url: CloudUrl representing the destination file.
|
| + dst_obj_metadata: apitools Object describing the destination object.
|
| + canned_acl: The canned acl to apply to the object, if any.
|
| + file_size: The size of the source file in bytes.
|
| + preconditions: Cloud API Preconditions for the final object.
|
| + gsutil_api: gsutil Cloud API instance to use.
|
| + command_obj: Command object (for calling Apply).
|
| + copy_exception_handler: Copy exception handler (for use in Apply).
|
| +
|
| + Returns:
|
| + Elapsed upload time, uploaded Object with generation, crc32c, and size
|
| + fields populated.
|
| + """
|
| + start_time = time.time()
|
| + dst_bucket_url = StorageUrlFromString(dst_url.bucket_url_string)
|
| + api_selector = gsutil_api.GetApiSelector(provider=dst_url.scheme)
|
| + # Determine which components, if any, have already been successfully
|
| + # uploaded.
|
| + tracker_file = GetTrackerFilePath(dst_url, TrackerFileType.PARALLEL_UPLOAD,
|
| + api_selector, src_url)
|
| + tracker_file_lock = CreateLock()
|
| + (random_prefix, existing_components) = (
|
| + _ParseParallelUploadTrackerFile(tracker_file, tracker_file_lock))
|
| +
|
| + # Create the initial tracker file for the upload.
|
| + _CreateParallelUploadTrackerFile(tracker_file, random_prefix,
|
| + existing_components, tracker_file_lock)
|
| +
|
| + # Get the set of all components that should be uploaded.
|
| + dst_args = _PartitionFile(
|
| + fp, file_size, src_url, dst_obj_metadata.contentType, canned_acl,
|
| + dst_bucket_url, random_prefix, tracker_file, tracker_file_lock)
|
| +
|
| + (components_to_upload, existing_components, existing_objects_to_delete) = (
|
| + FilterExistingComponents(dst_args, existing_components, dst_bucket_url,
|
| + gsutil_api))
|
| +
|
| + # In parallel, copy all of the file parts that haven't already been
|
| + # uploaded to temporary objects.
|
| + cp_results = command_obj.Apply(
|
| + _PerformParallelUploadFileToObject, components_to_upload,
|
| + copy_exception_handler, ('op_failure_count', 'total_bytes_transferred'),
|
| + arg_checker=gslib.command.DummyArgChecker,
|
| + parallel_operations_override=True, should_return_results=True)
|
| + uploaded_components = []
|
| + for cp_result in cp_results:
|
| + uploaded_components.append(cp_result[2])
|
| + components = uploaded_components + existing_components
|
| +
|
| + if len(components) == len(dst_args):
|
| + # Only try to compose if all of the components were uploaded successfully.
|
| +
|
| + def _GetComponentNumber(component):
|
| + return int(component.object_name[component.object_name.rfind('_')+1:])
|
| + # Sort the components so that they will be composed in the correct order.
|
| + components = sorted(components, key=_GetComponentNumber)
|
| +
|
| + request_components = []
|
| + for component_url in components:
|
| + src_obj_metadata = (
|
| + apitools_messages.ComposeRequest.SourceObjectsValueListEntry(
|
| + name=component_url.object_name))
|
| + if component_url.HasGeneration():
|
| + src_obj_metadata.generation = long(component_url.generation)
|
| + request_components.append(src_obj_metadata)
|
| +
|
| + composed_object = gsutil_api.ComposeObject(
|
| + request_components, dst_obj_metadata, preconditions=preconditions,
|
| + provider=dst_url.scheme, fields=['generation', 'crc32c', 'size'])
|
| +
|
| + try:
|
| + # Make sure only to delete things that we know were successfully
|
| + # uploaded (as opposed to all of the objects that we attempted to
|
| + # create) so that we don't delete any preexisting objects, except for
|
| + # those that were uploaded by a previous, failed run and have since
|
| + # changed (but still have an old generation lying around).
|
| + objects_to_delete = components + existing_objects_to_delete
|
| + command_obj.Apply(_DeleteObjectFn, objects_to_delete, _RmExceptionHandler,
|
| + arg_checker=gslib.command.DummyArgChecker,
|
| + parallel_operations_override=True)
|
| + except Exception: # pylint: disable=broad-except
|
| + # If some of the delete calls fail, don't cause the whole command to
|
| + # fail. The copy was successful iff the compose call succeeded, so
|
| + # reduce this to a warning.
|
| + logging.warning(
|
| + 'Failed to delete some of the following temporary objects:\n' +
|
| + '\n'.join(dst_args.keys()))
|
| + finally:
|
| + with tracker_file_lock:
|
| + if os.path.exists(tracker_file):
|
| + os.unlink(tracker_file)
|
| + else:
|
| + # Some of the components failed to upload. In this case, we want to exit
|
| + # without deleting the objects.
|
| + raise CommandException(
|
| + 'Some temporary components were not uploaded successfully. '
|
| + 'Please retry this upload.')
|
| +
|
| + elapsed_time = time.time() - start_time
|
| + return elapsed_time, composed_object
|
| +
|
| +
|
| +def _ShouldDoParallelCompositeUpload(logger, allow_splitting, src_url, dst_url,
|
| + file_size, canned_acl=None):
|
| + """Determines whether parallel composite upload strategy should be used.
|
| +
|
| + Args:
|
| + logger: for outputting log messages.
|
| + allow_splitting: If false, then this function returns false.
|
| + src_url: FileUrl corresponding to a local file.
|
| + dst_url: CloudUrl corresponding to destination cloud object.
|
| + file_size: The size of the source file, in bytes.
|
| + canned_acl: Canned ACL to apply to destination object, if any.
|
| +
|
| + Returns:
|
| + True iff a parallel upload should be performed on the source file.
|
| + """
|
| + global suggested_parallel_composites
|
| + parallel_composite_upload_threshold = HumanReadableToBytes(config.get(
|
| + 'GSUtil', 'parallel_composite_upload_threshold',
|
| + DEFAULT_PARALLEL_COMPOSITE_UPLOAD_THRESHOLD))
|
| +
|
| + all_factors_but_size = (
|
| + allow_splitting # Don't split the pieces multiple times.
|
| + and not src_url.IsStream() # We can't partition streams.
|
| + and dst_url.scheme == 'gs' # Compose is only for gs.
|
| + and not canned_acl) # TODO: Implement canned ACL support for compose.
|
| +
|
| + # Since parallel composite uploads are disabled by default, make user aware of
|
| + # them.
|
| + # TODO: Once compiled crcmod is being distributed by major Linux distributions
|
| + # remove this check.
|
| + if (all_factors_but_size and parallel_composite_upload_threshold == 0
|
| + and file_size >= PARALLEL_COMPOSITE_SUGGESTION_THRESHOLD
|
| + and not suggested_parallel_composites):
|
| + logger.info('\n'.join(textwrap.wrap(
|
| + '==> NOTE: You are uploading one or more large file(s), which would '
|
| + 'run significantly faster if you enable parallel composite uploads. '
|
| + 'This feature can be enabled by editing the '
|
| + '"parallel_composite_upload_threshold" value in your .boto '
|
| + 'configuration file. However, note that if you do this you and any '
|
| + 'users that download such composite files will need to have a compiled '
|
| + 'crcmod installed (see "gsutil help crcmod").')) + '\n')
|
| + suggested_parallel_composites = True
|
| +
|
| + return (all_factors_but_size
|
| + and parallel_composite_upload_threshold > 0
|
| + and file_size >= parallel_composite_upload_threshold)
|
| +
|
| +
|
| +def ExpandUrlToSingleBlr(url_str, gsutil_api, debug, project_id,
|
| + treat_nonexistent_object_as_subdir=False):
|
| + """Expands wildcard if present in url_str.
|
| +
|
| + Args:
|
| + url_str: String representation of requested url.
|
| + gsutil_api: gsutil Cloud API instance to use.
|
| + debug: debug level to use (for iterators).
|
| + project_id: project ID to use (for iterators).
|
| + treat_nonexistent_object_as_subdir: indicates if should treat a non-existent
|
| + object as a subdir.
|
| +
|
| + Returns:
|
| + (exp_url, have_existing_dst_container)
|
| + where exp_url is a StorageUrl
|
| + and have_existing_dst_container is a bool indicating whether
|
| + exp_url names an existing directory, bucket, or bucket subdirectory.
|
| + In the case where we match a subdirectory AND an object, the
|
| + object is returned.
|
| +
|
| + Raises:
|
| + CommandException: if url_str matched more than 1 URL.
|
| + """
|
| + # Handle wildcarded url case.
|
| + if ContainsWildcard(url_str):
|
| + blr_expansion = list(CreateWildcardIterator(url_str, gsutil_api,
|
| + debug=debug,
|
| + project_id=project_id))
|
| + if len(blr_expansion) != 1:
|
| + raise CommandException('Destination (%s) must match exactly 1 URL' %
|
| + url_str)
|
| + blr = blr_expansion[0]
|
| + # BLR is either an OBJECT, PREFIX, or BUCKET; the latter two represent
|
| + # directories.
|
| + return (StorageUrlFromString(blr.url_string), not blr.IsObject())
|
| +
|
| + storage_url = StorageUrlFromString(url_str)
|
| +
|
| + # Handle non-wildcarded URL.
|
| + if storage_url.IsFileUrl():
|
| + return (storage_url, storage_url.IsDirectory())
|
| +
|
| + # At this point we have a cloud URL.
|
| + if storage_url.IsBucket():
|
| + return (storage_url, True)
|
| +
|
| + # For object/prefix URLs check 3 cases: (a) if the name ends with '/' treat
|
| + # as a subdir; otherwise, use the wildcard iterator with url to
|
| + # find if (b) there's a Prefix matching url, or (c) name is of form
|
| + # dir_$folder$ (and in both these cases also treat dir as a subdir).
|
| + # Cloud subdirs are always considered to be an existing container.
|
| + if IsCloudSubdirPlaceholder(storage_url):
|
| + return (storage_url, True)
|
| +
|
| + # Check for the special case where we have a folder marker object.
|
| + folder_expansion = CreateWildcardIterator(
|
| + storage_url.versionless_url_string + '_$folder$', gsutil_api,
|
| + debug=debug, project_id=project_id).IterAll(
|
| + bucket_listing_fields=['name'])
|
| + for blr in folder_expansion:
|
| + return (storage_url, True)
|
| +
|
| + blr_expansion = CreateWildcardIterator(url_str, gsutil_api,
|
| + debug=debug,
|
| + project_id=project_id).IterAll(
|
| + bucket_listing_fields=['name'])
|
| + expansion_empty = True
|
| + for blr in blr_expansion:
|
| + expansion_empty = False
|
| + if blr.IsPrefix():
|
| + return (storage_url, True)
|
| +
|
| + return (storage_url,
|
| + expansion_empty and treat_nonexistent_object_as_subdir)
|
| +
|
| +
|
| +def FixWindowsNaming(src_url, dst_url):
|
| + """Translates Windows pathnames to cloud pathnames.
|
| +
|
| + Rewrites the destination URL built by ConstructDstUrl().
|
| +
|
| + Args:
|
| + src_url: Source StorageUrl to be copied.
|
| + dst_url: The destination StorageUrl built by ConstructDstUrl().
|
| +
|
| + Returns:
|
| + StorageUrl to use for copy.
|
| + """
|
| + if (src_url.IsFileUrl() and src_url.delim == '\\'
|
| + and dst_url.IsCloudUrl()):
|
| + trans_url_str = re.sub(r'\\', '/', dst_url.url_string)
|
| + dst_url = StorageUrlFromString(trans_url_str)
|
| + return dst_url
|
| +
|
| +
|
| +def SrcDstSame(src_url, dst_url):
|
| + """Checks if src_url and dst_url represent the same object or file.
|
| +
|
| + We don't handle anything about hard or symbolic links.
|
| +
|
| + Args:
|
| + src_url: Source StorageUrl.
|
| + dst_url: Destination StorageUrl.
|
| +
|
| + Returns:
|
| + Bool indicator.
|
| + """
|
| + if src_url.IsFileUrl() and dst_url.IsFileUrl():
|
| + # Translate a/b/./c to a/b/c, so src=dst comparison below works.
|
| + new_src_path = os.path.normpath(src_url.object_name)
|
| + new_dst_path = os.path.normpath(dst_url.object_name)
|
| + return new_src_path == new_dst_path
|
| + else:
|
| + return (src_url.url_string == dst_url.url_string and
|
| + src_url.generation == dst_url.generation)
|
| +
|
| +
|
| +def _LogCopyOperation(logger, src_url, dst_url, dst_obj_metadata):
|
| + """Logs copy operation, including Content-Type if appropriate.
|
| +
|
| + Args:
|
| + logger: logger instance to use for output.
|
| + src_url: Source StorageUrl.
|
| + dst_url: Destination StorageUrl.
|
| + dst_obj_metadata: Object-specific metadata that should be overidden during
|
| + the copy.
|
| + """
|
| + if (dst_url.IsCloudUrl() and dst_obj_metadata and
|
| + dst_obj_metadata.contentType):
|
| + content_type_msg = ' [Content-Type=%s]' % dst_obj_metadata.contentType
|
| + else:
|
| + content_type_msg = ''
|
| + if src_url.IsFileUrl() and src_url.IsStream():
|
| + logger.info('Copying from <STDIN>%s...', content_type_msg)
|
| + else:
|
| + logger.info('Copying %s%s...', src_url.url_string, content_type_msg)
|
| +
|
| +
|
| +# pylint: disable=undefined-variable
|
| +def _CopyObjToObjInTheCloud(src_url, src_obj_metadata, dst_url,
|
| + dst_obj_metadata, preconditions, gsutil_api,
|
| + logger):
|
| + """Performs copy-in-the cloud from specified src to dest object.
|
| +
|
| + Args:
|
| + src_url: Source CloudUrl.
|
| + src_obj_metadata: Metadata for source object; must include etag and size.
|
| + dst_url: Destination CloudUrl.
|
| + dst_obj_metadata: Object-specific metadata that should be overidden during
|
| + the copy.
|
| + preconditions: Preconditions to use for the copy.
|
| + gsutil_api: gsutil Cloud API instance to use for the copy.
|
| + logger: logging.Logger for log message output.
|
| +
|
| + Returns:
|
| + (elapsed_time, bytes_transferred, dst_url with generation,
|
| + md5 hash of destination) excluding overhead like initial GET.
|
| +
|
| + Raises:
|
| + CommandException: if errors encountered.
|
| + """
|
| + start_time = time.time()
|
| +
|
| + progress_callback = FileProgressCallbackHandler(
|
| + ConstructAnnounceText('Copying', dst_url.url_string), logger).call
|
| + if global_copy_helper_opts.test_callback_file:
|
| + with open(global_copy_helper_opts.test_callback_file, 'rb') as test_fp:
|
| + progress_callback = pickle.loads(test_fp.read()).call
|
| + dst_obj = gsutil_api.CopyObject(
|
| + src_obj_metadata, dst_obj_metadata, src_generation=src_url.generation,
|
| + canned_acl=global_copy_helper_opts.canned_acl,
|
| + preconditions=preconditions, progress_callback=progress_callback,
|
| + provider=dst_url.scheme, fields=UPLOAD_RETURN_FIELDS)
|
| +
|
| + end_time = time.time()
|
| +
|
| + result_url = dst_url.Clone()
|
| + result_url.generation = GenerationFromUrlAndString(result_url,
|
| + dst_obj.generation)
|
| +
|
| + return (end_time - start_time, src_obj_metadata.size, result_url,
|
| + dst_obj.md5Hash)
|
| +
|
| +
|
| +def _CheckFreeSpace(path):
|
| + """Return path/drive free space (in bytes)."""
|
| + if IS_WINDOWS:
|
| + # pylint: disable=g-import-not-at-top
|
| + try:
|
| + # pylint: disable=invalid-name
|
| + get_disk_free_space_ex = WINFUNCTYPE(c_int, c_wchar_p,
|
| + POINTER(c_uint64),
|
| + POINTER(c_uint64),
|
| + POINTER(c_uint64))
|
| + get_disk_free_space_ex = get_disk_free_space_ex(
|
| + ('GetDiskFreeSpaceExW', windll.kernel32), (
|
| + (1, 'lpszPathName'),
|
| + (2, 'lpFreeUserSpace'),
|
| + (2, 'lpTotalSpace'),
|
| + (2, 'lpFreeSpace'),))
|
| + except AttributeError:
|
| + get_disk_free_space_ex = WINFUNCTYPE(c_int, c_char_p,
|
| + POINTER(c_uint64),
|
| + POINTER(c_uint64),
|
| + POINTER(c_uint64))
|
| + get_disk_free_space_ex = get_disk_free_space_ex(
|
| + ('GetDiskFreeSpaceExA', windll.kernel32), (
|
| + (1, 'lpszPathName'),
|
| + (2, 'lpFreeUserSpace'),
|
| + (2, 'lpTotalSpace'),
|
| + (2, 'lpFreeSpace'),))
|
| +
|
| + def GetDiskFreeSpaceExErrCheck(result, unused_func, args):
|
| + if not result:
|
| + raise WinError()
|
| + return args[1].value
|
| + get_disk_free_space_ex.errcheck = GetDiskFreeSpaceExErrCheck
|
| +
|
| + return get_disk_free_space_ex(os.getenv('SystemDrive'))
|
| + else:
|
| + (_, f_frsize, _, _, f_bavail, _, _, _, _, _) = os.statvfs(path)
|
| + return f_frsize * f_bavail
|
| +
|
| +
|
| +def _SetContentTypeFromFile(src_url, dst_obj_metadata):
|
| + """Detects and sets Content-Type if src_url names a local file.
|
| +
|
| + Args:
|
| + src_url: Source StorageUrl.
|
| + dst_obj_metadata: Object-specific metadata that should be overidden during
|
| + the copy.
|
| + """
|
| + # contentType == '' if user requested default type.
|
| + if (dst_obj_metadata.contentType is None and src_url.IsFileUrl()
|
| + and not src_url.IsStream()):
|
| + # Only do content type recognition if src_url is a file. Object-to-object
|
| + # copies with no -h Content-Type specified re-use the content type of the
|
| + # source object.
|
| + object_name = src_url.object_name
|
| + content_type = None
|
| + # Streams (denoted by '-') are expected to be 'application/octet-stream'
|
| + # and 'file' would partially consume them.
|
| + if object_name != '-':
|
| + if config.getbool('GSUtil', 'use_magicfile', False):
|
| + p = subprocess.Popen(['file', '--mime-type', object_name],
|
| + stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
| + output, error = p.communicate()
|
| + p.stdout.close()
|
| + p.stderr.close()
|
| + if p.returncode != 0 or error:
|
| + raise CommandException(
|
| + 'Encountered error running "file --mime-type %s" '
|
| + '(returncode=%d).\n%s' % (object_name, p.returncode, error))
|
| + # Parse output by removing line delimiter and splitting on last ":
|
| + content_type = output.rstrip().rpartition(': ')[2]
|
| + else:
|
| + content_type = mimetypes.guess_type(object_name)[0]
|
| + if not content_type:
|
| + content_type = DEFAULT_CONTENT_TYPE
|
| + dst_obj_metadata.contentType = content_type
|
| +
|
| +
|
| +# pylint: disable=undefined-variable
|
| +def _UploadFileToObjectNonResumable(src_url, src_obj_filestream,
|
| + src_obj_size, dst_url, dst_obj_metadata,
|
| + preconditions, gsutil_api, logger):
|
| + """Uploads the file using a non-resumable strategy.
|
| +
|
| + Args:
|
| + src_url: Source StorageUrl to upload.
|
| + src_obj_filestream: File pointer to uploadable bytes.
|
| + src_obj_size: Size of the source object.
|
| + dst_url: Destination StorageUrl for the upload.
|
| + dst_obj_metadata: Metadata for the target object.
|
| + preconditions: Preconditions for the upload, if any.
|
| + gsutil_api: gsutil Cloud API instance to use for the upload.
|
| + logger: For outputting log messages.
|
| +
|
| + Returns:
|
| + Elapsed upload time, uploaded Object with generation, md5, and size fields
|
| + populated.
|
| + """
|
| + progress_callback = FileProgressCallbackHandler(
|
| + ConstructAnnounceText('Uploading', dst_url.url_string), logger).call
|
| + if global_copy_helper_opts.test_callback_file:
|
| + with open(global_copy_helper_opts.test_callback_file, 'rb') as test_fp:
|
| + progress_callback = pickle.loads(test_fp.read()).call
|
| + start_time = time.time()
|
| +
|
| + if src_url.IsStream():
|
| + # TODO: gsutil-beta: Provide progress callbacks for streaming uploads.
|
| + uploaded_object = gsutil_api.UploadObjectStreaming(
|
| + src_obj_filestream, object_metadata=dst_obj_metadata,
|
| + canned_acl=global_copy_helper_opts.canned_acl,
|
| + preconditions=preconditions, progress_callback=progress_callback,
|
| + provider=dst_url.scheme, fields=UPLOAD_RETURN_FIELDS)
|
| + else:
|
| + uploaded_object = gsutil_api.UploadObject(
|
| + src_obj_filestream, object_metadata=dst_obj_metadata,
|
| + canned_acl=global_copy_helper_opts.canned_acl, size=src_obj_size,
|
| + preconditions=preconditions, progress_callback=progress_callback,
|
| + provider=dst_url.scheme, fields=UPLOAD_RETURN_FIELDS)
|
| + end_time = time.time()
|
| + elapsed_time = end_time - start_time
|
| +
|
| + return elapsed_time, uploaded_object
|
| +
|
| +
|
| +# pylint: disable=undefined-variable
|
| +def _UploadFileToObjectResumable(src_url, src_obj_filestream,
|
| + src_obj_size, dst_url, dst_obj_metadata,
|
| + preconditions, gsutil_api, logger):
|
| + """Uploads the file using a resumable strategy.
|
| +
|
| + Args:
|
| + src_url: Source FileUrl to upload. Must not be a stream.
|
| + src_obj_filestream: File pointer to uploadable bytes.
|
| + src_obj_size: Size of the source object.
|
| + dst_url: Destination StorageUrl for the upload.
|
| + dst_obj_metadata: Metadata for the target object.
|
| + preconditions: Preconditions for the upload, if any.
|
| + gsutil_api: gsutil Cloud API instance to use for the upload.
|
| + logger: for outputting log messages.
|
| +
|
| + Returns:
|
| + Elapsed upload time, uploaded Object with generation, md5, and size fields
|
| + populated.
|
| + """
|
| + tracker_file_name = GetTrackerFilePath(
|
| + dst_url, TrackerFileType.UPLOAD,
|
| + gsutil_api.GetApiSelector(provider=dst_url.scheme))
|
| +
|
| + def _UploadTrackerCallback(serialization_data):
|
| + """Creates a new tracker file for starting an upload from scratch.
|
| +
|
| + This function is called by the gsutil Cloud API implementation and the
|
| + the serialization data is implementation-specific.
|
| +
|
| + Args:
|
| + serialization_data: Serialization data used in resuming the upload.
|
| + """
|
| + tracker_file = None
|
| + try:
|
| + tracker_file = open(tracker_file_name, 'w')
|
| + tracker_file.write(str(serialization_data))
|
| + except IOError as e:
|
| + RaiseUnwritableTrackerFileException(tracker_file_name, e.strerror)
|
| + finally:
|
| + if tracker_file:
|
| + tracker_file.close()
|
| +
|
| + # This contains the upload URL, which will uniquely identify the
|
| + # destination object.
|
| + tracker_data = _GetUploadTrackerData(tracker_file_name, logger)
|
| + if tracker_data:
|
| + logger.info(
|
| + 'Resuming upload for %s', src_url.url_string)
|
| +
|
| + retryable = True
|
| +
|
| + progress_callback = FileProgressCallbackHandler(
|
| + ConstructAnnounceText('Uploading', dst_url.url_string), logger).call
|
| + if global_copy_helper_opts.test_callback_file:
|
| + with open(global_copy_helper_opts.test_callback_file, 'rb') as test_fp:
|
| + progress_callback = pickle.loads(test_fp.read()).call
|
| +
|
| + start_time = time.time()
|
| + num_startover_attempts = 0
|
| + # This loop causes us to retry when the resumable upload failed in a way that
|
| + # requires starting over with a new upload ID. Retries within a single upload
|
| + # ID within the current process are handled in
|
| + # gsutil_api.UploadObjectResumable, and retries within a single upload ID
|
| + # spanning processes happens if an exception occurs not caught below (which
|
| + # will leave the tracker file in place, and cause the upload ID to be reused
|
| + # the next time the user runs gsutil and attempts the same upload).
|
| + while retryable:
|
| + try:
|
| + uploaded_object = gsutil_api.UploadObjectResumable(
|
| + src_obj_filestream, object_metadata=dst_obj_metadata,
|
| + canned_acl=global_copy_helper_opts.canned_acl,
|
| + preconditions=preconditions, provider=dst_url.scheme,
|
| + size=src_obj_size, serialization_data=tracker_data,
|
| + fields=UPLOAD_RETURN_FIELDS,
|
| + tracker_callback=_UploadTrackerCallback,
|
| + progress_callback=progress_callback)
|
| + retryable = False
|
| + except ResumableUploadStartOverException, e:
|
| + # This can happen, for example, if the server sends a 410 response code.
|
| + # In that case the current resumable upload ID can't be reused, so delete
|
| + # the tracker file and try again up to max retries.
|
| + num_startover_attempts += 1
|
| + retryable = (num_startover_attempts < GetNumRetries())
|
| + if not retryable:
|
| + raise
|
| +
|
| + # If the server sends a 404 response code, then the upload should only
|
| + # be restarted if it was the object (and not the bucket) that was missing.
|
| + try:
|
| + gsutil_api.GetBucket(dst_obj_metadata.bucket, provider=dst_url.scheme)
|
| + except NotFoundException:
|
| + raise
|
| +
|
| + logger.info('Restarting upload from scratch after exception %s', e)
|
| + DeleteTrackerFile(tracker_file_name)
|
| + tracker_data = None
|
| + src_obj_filestream.seek(0)
|
| + # Reset the progress callback handler.
|
| + progress_callback = FileProgressCallbackHandler(
|
| + ConstructAnnounceText('Uploading', dst_url.url_string), logger).call
|
| + logger.info('\n'.join(textwrap.wrap(
|
| + 'Resumable upload of %s failed with a response code indicating we '
|
| + 'need to start over with a new resumable upload ID. Backing off '
|
| + 'and retrying.' % src_url.url_string)))
|
| + time.sleep(min(random.random() * (2 ** num_startover_attempts),
|
| + GetMaxRetryDelay()))
|
| + except ResumableUploadAbortException:
|
| + retryable = False
|
| + raise
|
| + finally:
|
| + if not retryable:
|
| + DeleteTrackerFile(tracker_file_name)
|
| +
|
| + end_time = time.time()
|
| + elapsed_time = end_time - start_time
|
| +
|
| + return (elapsed_time, uploaded_object)
|
| +
|
| +
|
| +def _CompressFileForUpload(src_url, src_obj_filestream, src_obj_size, logger):
|
| + """Compresses a to-be-uploaded local file to save bandwidth.
|
| +
|
| + Args:
|
| + src_url: Source FileUrl.
|
| + src_obj_filestream: Read stream of the source file - will be consumed
|
| + and closed.
|
| + src_obj_size: Size of the source file.
|
| + logger: for outputting log messages.
|
| +
|
| + Returns:
|
| + StorageUrl path to compressed file, compressed file size.
|
| + """
|
| + # TODO: Compress using a streaming model as opposed to all at once here.
|
| + if src_obj_size >= MIN_SIZE_COMPUTE_LOGGING:
|
| + logger.info(
|
| + 'Compressing %s (to tmp)...', src_url)
|
| + (gzip_fh, gzip_path) = tempfile.mkstemp()
|
| + gzip_fp = None
|
| + try:
|
| + # Check for temp space. Assume the compressed object is at most 2x
|
| + # the size of the object (normally should compress to smaller than
|
| + # the object)
|
| + if _CheckFreeSpace(gzip_path) < 2*int(src_obj_size):
|
| + raise CommandException('Inadequate temp space available to compress '
|
| + '%s. See the CHANGING TEMP DIRECTORIES section '
|
| + 'of "gsutil help cp" for more info.' % src_url)
|
| + gzip_fp = gzip.open(gzip_path, 'wb')
|
| + data = src_obj_filestream.read(GZIP_CHUNK_SIZE)
|
| + while data:
|
| + gzip_fp.write(data)
|
| + data = src_obj_filestream.read(GZIP_CHUNK_SIZE)
|
| + finally:
|
| + if gzip_fp:
|
| + gzip_fp.close()
|
| + os.close(gzip_fh)
|
| + src_obj_filestream.close()
|
| + gzip_size = os.path.getsize(gzip_path)
|
| + return StorageUrlFromString(gzip_path), gzip_size
|
| +
|
| +
|
| +def _UploadFileToObject(src_url, src_obj_filestream, src_obj_size,
|
| + dst_url, dst_obj_metadata, preconditions, gsutil_api,
|
| + logger, command_obj, copy_exception_handler,
|
| + gzip_exts=None, allow_splitting=True):
|
| + """Uploads a local file to an object.
|
| +
|
| + Args:
|
| + src_url: Source FileUrl.
|
| + src_obj_filestream: Read stream of the source file to be read and closed.
|
| + src_obj_size: Size of the source file.
|
| + dst_url: Destination CloudUrl.
|
| + dst_obj_metadata: Metadata to be applied to the destination object.
|
| + preconditions: Preconditions to use for the copy.
|
| + gsutil_api: gsutil Cloud API to use for the copy.
|
| + logger: for outputting log messages.
|
| + command_obj: command object for use in Apply in parallel composite uploads.
|
| + copy_exception_handler: For handling copy exceptions during Apply.
|
| + gzip_exts: List of file extensions to gzip prior to upload, if any.
|
| + allow_splitting: Whether to allow the file to be split into component
|
| + pieces for an parallel composite upload.
|
| +
|
| + Returns:
|
| + (elapsed_time, bytes_transferred, dst_url with generation,
|
| + md5 hash of destination) excluding overhead like initial GET.
|
| +
|
| + Raises:
|
| + CommandException: if errors encountered.
|
| + """
|
| + if not dst_obj_metadata or not dst_obj_metadata.contentLanguage:
|
| + content_language = config.get_value('GSUtil', 'content_language')
|
| + if content_language:
|
| + dst_obj_metadata.contentLanguage = content_language
|
| +
|
| + fname_parts = src_url.object_name.split('.')
|
| + upload_url = src_url
|
| + upload_stream = src_obj_filestream
|
| + upload_size = src_obj_size
|
| + zipped_file = False
|
| + if gzip_exts and len(fname_parts) > 1 and fname_parts[-1] in gzip_exts:
|
| + upload_url, upload_size = _CompressFileForUpload(
|
| + src_url, src_obj_filestream, src_obj_size, logger)
|
| + upload_stream = open(upload_url.object_name, 'rb')
|
| + dst_obj_metadata.contentEncoding = 'gzip'
|
| + zipped_file = True
|
| +
|
| + elapsed_time = None
|
| + uploaded_object = None
|
| + hash_algs = GetUploadHashAlgs()
|
| + digesters = dict((alg, hash_algs[alg]()) for alg in hash_algs or {})
|
| +
|
| + parallel_composite_upload = _ShouldDoParallelCompositeUpload(
|
| + logger, allow_splitting, upload_url, dst_url, src_obj_size,
|
| + canned_acl=global_copy_helper_opts.canned_acl)
|
| +
|
| + if (src_url.IsStream() and
|
| + gsutil_api.GetApiSelector(provider=dst_url.scheme) == ApiSelector.JSON):
|
| + orig_stream = upload_stream
|
| + # Add limited seekable properties to the stream via buffering.
|
| + upload_stream = ResumableStreamingJsonUploadWrapper(
|
| + orig_stream, GetJsonResumableChunkSize())
|
| +
|
| + if not parallel_composite_upload and len(hash_algs):
|
| + # Parallel composite uploads calculate hashes per-component in subsequent
|
| + # calls to this function, but the composition of the final object is a
|
| + # cloud-only operation.
|
| + wrapped_filestream = HashingFileUploadWrapper(upload_stream, digesters,
|
| + hash_algs, upload_url, logger)
|
| + else:
|
| + wrapped_filestream = upload_stream
|
| +
|
| + try:
|
| + if parallel_composite_upload:
|
| + elapsed_time, uploaded_object = _DoParallelCompositeUpload(
|
| + upload_stream, upload_url, dst_url, dst_obj_metadata,
|
| + global_copy_helper_opts.canned_acl, upload_size, preconditions,
|
| + gsutil_api, command_obj, copy_exception_handler)
|
| + elif upload_size < ResumableThreshold() or src_url.IsStream():
|
| + elapsed_time, uploaded_object = _UploadFileToObjectNonResumable(
|
| + upload_url, wrapped_filestream, upload_size, dst_url,
|
| + dst_obj_metadata, preconditions, gsutil_api, logger)
|
| + else:
|
| + elapsed_time, uploaded_object = _UploadFileToObjectResumable(
|
| + upload_url, wrapped_filestream, upload_size, dst_url,
|
| + dst_obj_metadata, preconditions, gsutil_api, logger)
|
| +
|
| + finally:
|
| + if zipped_file:
|
| + try:
|
| + os.unlink(upload_url.object_name)
|
| + # Windows sometimes complains the temp file is locked when you try to
|
| + # delete it.
|
| + except Exception: # pylint: disable=broad-except
|
| + logger.warning(
|
| + 'Could not delete %s. This can occur in Windows because the '
|
| + 'temporary file is still locked.', upload_url.object_name)
|
| + # In the gzip case, this is the gzip stream. _CompressFileForUpload will
|
| + # have already closed the original source stream.
|
| + upload_stream.close()
|
| +
|
| + if not parallel_composite_upload:
|
| + try:
|
| + digests = _CreateDigestsFromDigesters(digesters)
|
| + _CheckHashes(logger, dst_url, uploaded_object, src_url.object_name,
|
| + digests, is_upload=True)
|
| + except HashMismatchException:
|
| + if _RENAME_ON_HASH_MISMATCH:
|
| + corrupted_obj_metadata = apitools_messages.Object(
|
| + name=dst_obj_metadata.name,
|
| + bucket=dst_obj_metadata.bucket,
|
| + etag=uploaded_object.etag)
|
| + dst_obj_metadata.name = (dst_url.object_name +
|
| + _RENAME_ON_HASH_MISMATCH_SUFFIX)
|
| + gsutil_api.CopyObject(corrupted_obj_metadata,
|
| + dst_obj_metadata, provider=dst_url.scheme)
|
| + # If the digest doesn't match, delete the object.
|
| + gsutil_api.DeleteObject(dst_url.bucket_name, dst_url.object_name,
|
| + generation=uploaded_object.generation,
|
| + provider=dst_url.scheme)
|
| + raise
|
| +
|
| + result_url = dst_url.Clone()
|
| +
|
| + result_url.generation = uploaded_object.generation
|
| + result_url.generation = GenerationFromUrlAndString(
|
| + result_url, uploaded_object.generation)
|
| +
|
| + return (elapsed_time, uploaded_object.size, result_url,
|
| + uploaded_object.md5Hash)
|
| +
|
| +
|
| +# TODO: Refactor this long function into smaller pieces.
|
| +# pylint: disable=too-many-statements
|
| +def _DownloadObjectToFile(src_url, src_obj_metadata, dst_url,
|
| + gsutil_api, logger, test_method=None):
|
| + """Downloads an object to a local file.
|
| +
|
| + Args:
|
| + src_url: Source CloudUrl.
|
| + src_obj_metadata: Metadata from the source object.
|
| + dst_url: Destination FileUrl.
|
| + gsutil_api: gsutil Cloud API instance to use for the download.
|
| + logger: for outputting log messages.
|
| + test_method: Optional test method for modifying the file before validation
|
| + during unit tests.
|
| + Returns:
|
| + (elapsed_time, bytes_transferred, dst_url, md5), excluding overhead like
|
| + initial GET.
|
| +
|
| + Raises:
|
| + CommandException: if errors encountered.
|
| + """
|
| + global open_files_map
|
| + file_name = dst_url.object_name
|
| + dir_name = os.path.dirname(file_name)
|
| + if dir_name and not os.path.exists(dir_name):
|
| + # Do dir creation in try block so can ignore case where dir already
|
| + # exists. This is needed to avoid a race condition when running gsutil
|
| + # -m cp.
|
| + try:
|
| + os.makedirs(dir_name)
|
| + except OSError, e:
|
| + if e.errno != errno.EEXIST:
|
| + raise
|
| + api_selector = gsutil_api.GetApiSelector(provider=src_url.scheme)
|
| + # For gzipped objects download to a temp file and unzip. For the XML API,
|
| + # the represents the result of a HEAD request. For the JSON API, this is
|
| + # the stored encoding which the service may not respect. However, if the
|
| + # server sends decompressed bytes for a file that is stored compressed
|
| + # (double compressed case), there is no way we can validate the hash and
|
| + # we will fail our hash check for the object.
|
| + if (src_obj_metadata.contentEncoding and
|
| + src_obj_metadata.contentEncoding.lower().endswith('gzip')):
|
| + # We can't use tempfile.mkstemp() here because we need a predictable
|
| + # filename for resumable downloads.
|
| + download_file_name = _GetDownloadZipFileName(file_name)
|
| + logger.info(
|
| + 'Downloading to temp gzip filename %s', download_file_name)
|
| + need_to_unzip = True
|
| + else:
|
| + download_file_name = file_name
|
| + need_to_unzip = False
|
| +
|
| + if download_file_name.endswith(dst_url.delim):
|
| + logger.warn('\n'.join(textwrap.wrap(
|
| + 'Skipping attempt to download to filename ending with slash (%s). This '
|
| + 'typically happens when using gsutil to download from a subdirectory '
|
| + 'created by the Cloud Console (https://cloud.google.com/console)'
|
| + % download_file_name)))
|
| + return (0, 0, dst_url, '')
|
| +
|
| + # Set up hash digesters.
|
| + hash_algs = GetDownloadHashAlgs(
|
| + logger, src_has_md5=src_obj_metadata.md5Hash,
|
| + src_has_crc32c=src_obj_metadata.crc32c)
|
| + digesters = dict((alg, hash_algs[alg]()) for alg in hash_algs or {})
|
| +
|
| + fp = None
|
| + # Tracks whether the server used a gzip encoding.
|
| + server_encoding = None
|
| + download_complete = False
|
| + download_strategy = _SelectDownloadStrategy(dst_url)
|
| + download_start_point = 0
|
| + # This is used for resuming downloads, but also for passing the mediaLink
|
| + # and size into the download for new downloads so that we can avoid
|
| + # making an extra HTTP call.
|
| + serialization_data = None
|
| + serialization_dict = GetDownloadSerializationDict(src_obj_metadata)
|
| + open_files = []
|
| + try:
|
| + if download_strategy is CloudApi.DownloadStrategy.ONE_SHOT:
|
| + fp = open(download_file_name, 'wb')
|
| + elif download_strategy is CloudApi.DownloadStrategy.RESUMABLE:
|
| + # If this is a resumable download, we need to open the file for append and
|
| + # manage a tracker file.
|
| + if open_files_map.get(download_file_name, False):
|
| + # Ensure another process/thread is not already writing to this file.
|
| + raise FileConcurrencySkipError
|
| + open_files.append(download_file_name)
|
| + open_files_map[download_file_name] = True
|
| + fp = open(download_file_name, 'ab')
|
| +
|
| + resuming = ReadOrCreateDownloadTrackerFile(
|
| + src_obj_metadata, dst_url, api_selector)
|
| + if resuming:
|
| + # Find out how far along we are so we can request the appropriate
|
| + # remaining range of the object.
|
| + existing_file_size = GetFileSize(fp, position_to_eof=True)
|
| + if existing_file_size > src_obj_metadata.size:
|
| + DeleteTrackerFile(GetTrackerFilePath(
|
| + dst_url, TrackerFileType.DOWNLOAD, api_selector))
|
| + raise CommandException(
|
| + '%s is larger (%d) than %s (%d).\nDeleting tracker file, so '
|
| + 'if you re-try this download it will start from scratch' %
|
| + (download_file_name, existing_file_size, src_url.object_name,
|
| + src_obj_metadata.size))
|
| + else:
|
| + if existing_file_size == src_obj_metadata.size:
|
| + logger.info(
|
| + 'Download already complete for file %s, skipping download but '
|
| + 'will run integrity checks.', download_file_name)
|
| + download_complete = True
|
| + else:
|
| + download_start_point = existing_file_size
|
| + serialization_dict['progress'] = download_start_point
|
| + logger.info('Resuming download for %s', src_url.url_string)
|
| + # Catch up our digester with the hash data.
|
| + if existing_file_size > TEN_MIB:
|
| + for alg_name in digesters:
|
| + logger.info(
|
| + 'Catching up %s for %s', alg_name, download_file_name)
|
| + with open(download_file_name, 'rb') as hash_fp:
|
| + while True:
|
| + data = hash_fp.read(DEFAULT_FILE_BUFFER_SIZE)
|
| + if not data:
|
| + break
|
| + for alg_name in digesters:
|
| + digesters[alg_name].update(data)
|
| + else:
|
| + # Starting a new download, blow away whatever is already there.
|
| + fp.truncate(0)
|
| + fp.seek(0)
|
| +
|
| + else:
|
| + raise CommandException('Invalid download strategy %s chosen for'
|
| + 'file %s' % (download_strategy, fp.name))
|
| +
|
| + if not dst_url.IsStream():
|
| + serialization_data = json.dumps(serialization_dict)
|
| +
|
| + progress_callback = FileProgressCallbackHandler(
|
| + ConstructAnnounceText('Downloading', dst_url.url_string),
|
| + logger).call
|
| + if global_copy_helper_opts.test_callback_file:
|
| + with open(global_copy_helper_opts.test_callback_file, 'rb') as test_fp:
|
| + progress_callback = pickle.loads(test_fp.read()).call
|
| +
|
| + start_time = time.time()
|
| + # TODO: With gzip encoding (which may occur on-the-fly and not be part of
|
| + # the object's metadata), when we request a range to resume, it's possible
|
| + # that the server will just resend the entire object, which means our
|
| + # caught-up hash will be incorrect. We recalculate the hash on
|
| + # the local file in the case of a failed gzip hash anyway, but it would
|
| + # be better if we actively detected this case.
|
| + if not download_complete:
|
| + server_encoding = gsutil_api.GetObjectMedia(
|
| + src_url.bucket_name, src_url.object_name, fp,
|
| + start_byte=download_start_point, generation=src_url.generation,
|
| + object_size=src_obj_metadata.size,
|
| + download_strategy=download_strategy, provider=src_url.scheme,
|
| + serialization_data=serialization_data, digesters=digesters,
|
| + progress_callback=progress_callback)
|
| +
|
| + end_time = time.time()
|
| +
|
| + # If a custom test method is defined, call it here. For the copy command,
|
| + # test methods are expected to take one argument: an open file pointer,
|
| + # and are used to perturb the open file during download to exercise
|
| + # download error detection.
|
| + if test_method:
|
| + test_method(fp)
|
| +
|
| + except ResumableDownloadException as e:
|
| + logger.warning('Caught ResumableDownloadException (%s) for file %s.',
|
| + e.reason, file_name)
|
| + raise
|
| +
|
| + finally:
|
| + if fp:
|
| + fp.close()
|
| + for file_name in open_files:
|
| + open_files_map.delete(file_name)
|
| +
|
| + # If we decompressed a content-encoding gzip file on the fly, this may not
|
| + # be accurate, but it is the best we can do without going deep into the
|
| + # underlying HTTP libraries. Note that this value is only used for
|
| + # reporting in log messages; inaccuracy doesn't impact the integrity of the
|
| + # download.
|
| + bytes_transferred = src_obj_metadata.size - download_start_point
|
| + server_gzip = server_encoding and server_encoding.lower().endswith('gzip')
|
| + local_md5 = _ValidateDownloadHashes(logger, src_url, src_obj_metadata,
|
| + dst_url, need_to_unzip, server_gzip,
|
| + digesters, hash_algs, api_selector,
|
| + bytes_transferred)
|
| +
|
| + return (end_time - start_time, bytes_transferred, dst_url, local_md5)
|
| +
|
| +
|
| +def _GetDownloadZipFileName(file_name):
|
| + """Returns the file name for a temporarily compressed downloaded file."""
|
| + return '%s_.gztmp' % file_name
|
| +
|
| +
|
| +def _ValidateDownloadHashes(logger, src_url, src_obj_metadata, dst_url,
|
| + need_to_unzip, server_gzip, digesters, hash_algs,
|
| + api_selector, bytes_transferred):
|
| + """Validates a downloaded file's integrity.
|
| +
|
| + Args:
|
| + logger: For outputting log messages.
|
| + src_url: StorageUrl for the source object.
|
| + src_obj_metadata: Metadata for the source object, potentially containing
|
| + hash values.
|
| + dst_url: StorageUrl describing the destination file.
|
| + need_to_unzip: If true, a temporary zip file was used and must be
|
| + uncompressed as part of validation.
|
| + server_gzip: If true, the server gzipped the bytes (regardless of whether
|
| + the object metadata claimed it was gzipped).
|
| + digesters: dict of {string, hash digester} that contains up-to-date digests
|
| + computed during the download. If a digester for a particular
|
| + algorithm is None, an up-to-date digest is not available and the
|
| + hash must be recomputed from the local file.
|
| + hash_algs: dict of {string, hash algorithm} that can be used if digesters
|
| + don't have up-to-date digests.
|
| + api_selector: The Cloud API implementation used (used tracker file naming).
|
| + bytes_transferred: Number of bytes downloaded (used for logging).
|
| +
|
| + Returns:
|
| + An MD5 of the local file, if one was calculated as part of the integrity
|
| + check.
|
| + """
|
| + file_name = dst_url.object_name
|
| + download_file_name = (_GetDownloadZipFileName(file_name) if need_to_unzip else
|
| + file_name)
|
| + digesters_succeeded = True
|
| + for alg in digesters:
|
| + # If we get a digester with a None algorithm, the underlying
|
| + # implementation failed to calculate a digest, so we will need to
|
| + # calculate one from scratch.
|
| + if not digesters[alg]:
|
| + digesters_succeeded = False
|
| + break
|
| +
|
| + if digesters_succeeded:
|
| + local_hashes = _CreateDigestsFromDigesters(digesters)
|
| + else:
|
| + local_hashes = _CreateDigestsFromLocalFile(
|
| + logger, hash_algs, download_file_name, src_obj_metadata)
|
| +
|
| + digest_verified = True
|
| + hash_invalid_exception = None
|
| + try:
|
| + _CheckHashes(logger, src_url, src_obj_metadata, download_file_name,
|
| + local_hashes)
|
| + DeleteTrackerFile(GetTrackerFilePath(
|
| + dst_url, TrackerFileType.DOWNLOAD, api_selector))
|
| + except HashMismatchException, e:
|
| + # If an non-gzipped object gets sent with gzip content encoding, the hash
|
| + # we calculate will match the gzipped bytes, not the original object. Thus,
|
| + # we'll need to calculate and check it after unzipping.
|
| + if server_gzip:
|
| + logger.debug(
|
| + 'Hash did not match but server gzipped the content, will '
|
| + 'recalculate.')
|
| + digest_verified = False
|
| + elif api_selector == ApiSelector.XML:
|
| + logger.debug(
|
| + 'Hash did not match but server may have gzipped the content, will '
|
| + 'recalculate.')
|
| + # Save off the exception in case this isn't a gzipped file.
|
| + hash_invalid_exception = e
|
| + digest_verified = False
|
| + else:
|
| + DeleteTrackerFile(GetTrackerFilePath(
|
| + dst_url, TrackerFileType.DOWNLOAD, api_selector))
|
| + if _RENAME_ON_HASH_MISMATCH:
|
| + os.rename(download_file_name,
|
| + download_file_name + _RENAME_ON_HASH_MISMATCH_SUFFIX)
|
| + else:
|
| + os.unlink(download_file_name)
|
| + raise
|
| +
|
| + if server_gzip and not need_to_unzip:
|
| + # Server compressed bytes on-the-fly, thus we need to rename and decompress.
|
| + # We can't decompress on-the-fly because prior to Python 3.2 the gzip
|
| + # module makes a bunch of seek calls on the stream.
|
| + download_file_name = _GetDownloadZipFileName(file_name)
|
| + os.rename(file_name, download_file_name)
|
| +
|
| + if need_to_unzip or server_gzip:
|
| + # Log that we're uncompressing if the file is big enough that
|
| + # decompressing would make it look like the transfer "stalled" at the end.
|
| + if bytes_transferred > TEN_MIB:
|
| + logger.info(
|
| + 'Uncompressing downloaded tmp file to %s...', file_name)
|
| +
|
| + # Downloaded gzipped file to a filename w/o .gz extension, so unzip.
|
| + gzip_fp = None
|
| + try:
|
| + gzip_fp = gzip.open(download_file_name, 'rb')
|
| + with open(file_name, 'wb') as f_out:
|
| + data = gzip_fp.read(GZIP_CHUNK_SIZE)
|
| + while data:
|
| + f_out.write(data)
|
| + data = gzip_fp.read(GZIP_CHUNK_SIZE)
|
| + except IOError, e:
|
| + # In the XML case where we don't know if the file was gzipped, raise
|
| + # the original hash exception if we find that it wasn't.
|
| + if 'Not a gzipped file' in str(e) and hash_invalid_exception:
|
| + # Linter improperly thinks we're raising None despite the above check.
|
| + # pylint: disable=raising-bad-type
|
| + raise hash_invalid_exception
|
| + finally:
|
| + if gzip_fp:
|
| + gzip_fp.close()
|
| +
|
| + os.unlink(download_file_name)
|
| +
|
| + if not digest_verified:
|
| + try:
|
| + # Recalculate hashes on the unzipped local file.
|
| + local_hashes = _CreateDigestsFromLocalFile(logger, hash_algs, file_name,
|
| + src_obj_metadata)
|
| + _CheckHashes(logger, src_url, src_obj_metadata, file_name, local_hashes)
|
| + DeleteTrackerFile(GetTrackerFilePath(
|
| + dst_url, TrackerFileType.DOWNLOAD, api_selector))
|
| + except HashMismatchException:
|
| + DeleteTrackerFile(GetTrackerFilePath(
|
| + dst_url, TrackerFileType.DOWNLOAD, api_selector))
|
| + if _RENAME_ON_HASH_MISMATCH:
|
| + os.rename(file_name,
|
| + file_name + _RENAME_ON_HASH_MISMATCH_SUFFIX)
|
| + else:
|
| + os.unlink(file_name)
|
| + raise
|
| +
|
| + if 'md5' in local_hashes:
|
| + return local_hashes['md5']
|
| +
|
| +
|
| +def _CopyFileToFile(src_url, dst_url):
|
| + """Copies a local file to a local file.
|
| +
|
| + Args:
|
| + src_url: Source FileUrl.
|
| + dst_url: Destination FileUrl.
|
| + Returns:
|
| + (elapsed_time, bytes_transferred, dst_url, md5=None).
|
| +
|
| + Raises:
|
| + CommandException: if errors encountered.
|
| + """
|
| + src_fp = GetStreamFromFileUrl(src_url)
|
| + dir_name = os.path.dirname(dst_url.object_name)
|
| + if dir_name and not os.path.exists(dir_name):
|
| + os.makedirs(dir_name)
|
| + dst_fp = open(dst_url.object_name, 'wb')
|
| + start_time = time.time()
|
| + shutil.copyfileobj(src_fp, dst_fp)
|
| + end_time = time.time()
|
| + return (end_time - start_time, os.path.getsize(dst_url.object_name),
|
| + dst_url, None)
|
| +
|
| +
|
| +def _DummyTrackerCallback(_):
|
| + pass
|
| +
|
| +
|
| +# pylint: disable=undefined-variable
|
| +def _CopyObjToObjDaisyChainMode(src_url, src_obj_metadata, dst_url,
|
| + dst_obj_metadata, preconditions, gsutil_api,
|
| + logger):
|
| + """Copies from src_url to dst_url in "daisy chain" mode.
|
| +
|
| + See -D OPTION documentation about what daisy chain mode is.
|
| +
|
| + Args:
|
| + src_url: Source CloudUrl
|
| + src_obj_metadata: Metadata from source object
|
| + dst_url: Destination CloudUrl
|
| + dst_obj_metadata: Object-specific metadata that should be overidden during
|
| + the copy.
|
| + preconditions: Preconditions to use for the copy.
|
| + gsutil_api: gsutil Cloud API to use for the copy.
|
| + logger: For outputting log messages.
|
| +
|
| + Returns:
|
| + (elapsed_time, bytes_transferred, dst_url with generation,
|
| + md5 hash of destination) excluding overhead like initial GET.
|
| +
|
| + Raises:
|
| + CommandException: if errors encountered.
|
| + """
|
| + # We don't attempt to preserve ACLs across providers because
|
| + # GCS and S3 support different ACLs and disjoint principals.
|
| + if (global_copy_helper_opts.preserve_acl
|
| + and src_url.scheme != dst_url.scheme):
|
| + raise NotImplementedError(
|
| + 'Cross-provider cp -p not supported')
|
| + if not global_copy_helper_opts.preserve_acl:
|
| + dst_obj_metadata.acl = []
|
| +
|
| + # Don't use callbacks for downloads on the daisy chain wrapper because
|
| + # upload callbacks will output progress, but respect test hooks if present.
|
| + progress_callback = None
|
| + if global_copy_helper_opts.test_callback_file:
|
| + with open(global_copy_helper_opts.test_callback_file, 'rb') as test_fp:
|
| + progress_callback = pickle.loads(test_fp.read()).call
|
| +
|
| + start_time = time.time()
|
| + upload_fp = DaisyChainWrapper(src_url, src_obj_metadata.size, gsutil_api,
|
| + progress_callback=progress_callback)
|
| + uploaded_object = None
|
| + if src_obj_metadata.size == 0:
|
| + # Resumable uploads of size 0 are not supported.
|
| + uploaded_object = gsutil_api.UploadObject(
|
| + upload_fp, object_metadata=dst_obj_metadata,
|
| + canned_acl=global_copy_helper_opts.canned_acl,
|
| + preconditions=preconditions, provider=dst_url.scheme,
|
| + fields=UPLOAD_RETURN_FIELDS, size=src_obj_metadata.size)
|
| + else:
|
| + # TODO: Support process-break resumes. This will resume across connection
|
| + # breaks and server errors, but the tracker callback is a no-op so this
|
| + # won't resume across gsutil runs.
|
| + # TODO: Test retries via test_callback_file.
|
| + uploaded_object = gsutil_api.UploadObjectResumable(
|
| + upload_fp, object_metadata=dst_obj_metadata,
|
| + canned_acl=global_copy_helper_opts.canned_acl,
|
| + preconditions=preconditions, provider=dst_url.scheme,
|
| + fields=UPLOAD_RETURN_FIELDS, size=src_obj_metadata.size,
|
| + progress_callback=FileProgressCallbackHandler(
|
| + ConstructAnnounceText('Uploading', dst_url.url_string),
|
| + logger).call,
|
| + tracker_callback=_DummyTrackerCallback)
|
| + end_time = time.time()
|
| +
|
| + try:
|
| + _CheckCloudHashes(logger, src_url, dst_url, src_obj_metadata,
|
| + uploaded_object)
|
| + except HashMismatchException:
|
| + if _RENAME_ON_HASH_MISMATCH:
|
| + corrupted_obj_metadata = apitools_messages.Object(
|
| + name=dst_obj_metadata.name,
|
| + bucket=dst_obj_metadata.bucket,
|
| + etag=uploaded_object.etag)
|
| + dst_obj_metadata.name = (dst_url.object_name +
|
| + _RENAME_ON_HASH_MISMATCH_SUFFIX)
|
| + gsutil_api.CopyObject(corrupted_obj_metadata,
|
| + dst_obj_metadata, provider=dst_url.scheme)
|
| + # If the digest doesn't match, delete the object.
|
| + gsutil_api.DeleteObject(dst_url.bucket_name, dst_url.object_name,
|
| + generation=uploaded_object.generation,
|
| + provider=dst_url.scheme)
|
| + raise
|
| +
|
| + result_url = dst_url.Clone()
|
| + result_url.generation = GenerationFromUrlAndString(
|
| + result_url, uploaded_object.generation)
|
| +
|
| + return (end_time - start_time, src_obj_metadata.size, result_url,
|
| + uploaded_object.md5Hash)
|
| +
|
| +
|
| +# pylint: disable=undefined-variable
|
| +# pylint: disable=too-many-statements
|
| +def PerformCopy(logger, src_url, dst_url, gsutil_api, command_obj,
|
| + copy_exception_handler, allow_splitting=True,
|
| + headers=None, manifest=None, gzip_exts=None, test_method=None):
|
| + """Performs copy from src_url to dst_url, handling various special cases.
|
| +
|
| + Args:
|
| + logger: for outputting log messages.
|
| + src_url: Source StorageUrl.
|
| + dst_url: Destination StorageUrl.
|
| + gsutil_api: gsutil Cloud API instance to use for the copy.
|
| + command_obj: command object for use in Apply in parallel composite uploads.
|
| + copy_exception_handler: for handling copy exceptions during Apply.
|
| + allow_splitting: Whether to allow the file to be split into component
|
| + pieces for an parallel composite upload.
|
| + headers: optional headers to use for the copy operation.
|
| + manifest: optional manifest for tracking copy operations.
|
| + gzip_exts: List of file extensions to gzip for uploads, if any.
|
| + test_method: optional test method for modifying files during unit tests.
|
| +
|
| + Returns:
|
| + (elapsed_time, bytes_transferred, version-specific dst_url) excluding
|
| + overhead like initial GET.
|
| +
|
| + Raises:
|
| + ItemExistsError: if no clobber flag is specified and the destination
|
| + object already exists.
|
| + SkipUnsupportedObjectError: if skip_unsupported_objects flag is specified
|
| + and the source is an unsupported type.
|
| + CommandException: if other errors encountered.
|
| + """
|
| + if headers:
|
| + dst_obj_headers = headers.copy()
|
| + else:
|
| + dst_obj_headers = {}
|
| +
|
| + # Create a metadata instance for each destination object so metadata
|
| + # such as content-type can be applied per-object.
|
| + # Initialize metadata from any headers passed in via -h.
|
| + dst_obj_metadata = ObjectMetadataFromHeaders(dst_obj_headers)
|
| +
|
| + if dst_url.IsCloudUrl() and dst_url.scheme == 'gs':
|
| + preconditions = PreconditionsFromHeaders(dst_obj_headers)
|
| + else:
|
| + preconditions = Preconditions()
|
| +
|
| + src_obj_metadata = None
|
| + src_obj_filestream = None
|
| + if src_url.IsCloudUrl():
|
| + src_obj_fields = None
|
| + if dst_url.IsCloudUrl():
|
| + # For cloud or daisy chain copy, we need every copyable field.
|
| + # If we're not modifying or overriding any of the fields, we can get
|
| + # away without retrieving the object metadata because the copy
|
| + # operation can succeed with just the destination bucket and object
|
| + # name. But if we are sending any metadata, the JSON API will expect a
|
| + # complete object resource. Since we want metadata like the object size
|
| + # for our own tracking, we just get all of the metadata here.
|
| + src_obj_fields = ['cacheControl', 'componentCount',
|
| + 'contentDisposition', 'contentEncoding',
|
| + 'contentLanguage', 'contentType', 'crc32c',
|
| + 'etag', 'generation', 'md5Hash', 'mediaLink',
|
| + 'metadata', 'metageneration', 'size']
|
| + # We only need the ACL if we're going to preserve it.
|
| + if global_copy_helper_opts.preserve_acl:
|
| + src_obj_fields.append('acl')
|
| + if (src_url.scheme == dst_url.scheme
|
| + and not global_copy_helper_opts.daisy_chain):
|
| + copy_in_the_cloud = True
|
| + else:
|
| + copy_in_the_cloud = False
|
| + else:
|
| + # Just get the fields needed to validate the download.
|
| + src_obj_fields = ['crc32c', 'contentEncoding', 'contentType', 'etag',
|
| + 'mediaLink', 'md5Hash', 'size']
|
| +
|
| + if (src_url.scheme == 's3' and
|
| + global_copy_helper_opts.skip_unsupported_objects):
|
| + src_obj_fields.append('storageClass')
|
| +
|
| + try:
|
| + src_generation = GenerationFromUrlAndString(src_url, src_url.generation)
|
| + src_obj_metadata = gsutil_api.GetObjectMetadata(
|
| + src_url.bucket_name, src_url.object_name,
|
| + generation=src_generation, provider=src_url.scheme,
|
| + fields=src_obj_fields)
|
| + except NotFoundException:
|
| + raise CommandException(
|
| + 'NotFoundException: Could not retrieve source object %s.' %
|
| + src_url.url_string)
|
| + if (src_url.scheme == 's3' and
|
| + global_copy_helper_opts.skip_unsupported_objects and
|
| + src_obj_metadata.storageClass == 'GLACIER'):
|
| + raise SkipGlacierError()
|
| +
|
| + src_obj_size = src_obj_metadata.size
|
| + dst_obj_metadata.contentType = src_obj_metadata.contentType
|
| + if global_copy_helper_opts.preserve_acl:
|
| + dst_obj_metadata.acl = src_obj_metadata.acl
|
| + # Special case for S3-to-S3 copy URLs using
|
| + # global_copy_helper_opts.preserve_acl.
|
| + # dst_url will be verified in _CopyObjToObjDaisyChainMode if it
|
| + # is not s3 (and thus differs from src_url).
|
| + if src_url.scheme == 's3':
|
| + acl_text = S3MarkerAclFromObjectMetadata(src_obj_metadata)
|
| + if acl_text:
|
| + AddS3MarkerAclToObjectMetadata(dst_obj_metadata, acl_text)
|
| + else:
|
| + try:
|
| + src_obj_filestream = GetStreamFromFileUrl(src_url)
|
| + except Exception, e: # pylint: disable=broad-except
|
| + raise CommandException('Error opening file "%s": %s.' % (src_url,
|
| + e.message))
|
| + if src_url.IsStream():
|
| + src_obj_size = None
|
| + else:
|
| + src_obj_size = os.path.getsize(src_url.object_name)
|
| +
|
| + if global_copy_helper_opts.use_manifest:
|
| + # Set the source size in the manifest.
|
| + manifest.Set(src_url.url_string, 'size', src_obj_size)
|
| +
|
| + if (dst_url.scheme == 's3' and src_obj_size > S3_MAX_UPLOAD_SIZE
|
| + and src_url != 's3'):
|
| + raise CommandException(
|
| + '"%s" exceeds the maximum gsutil-supported size for an S3 upload. S3 '
|
| + 'objects greater than %s in size require multipart uploads, which '
|
| + 'gsutil does not support.' % (src_url,
|
| + MakeHumanReadable(S3_MAX_UPLOAD_SIZE)))
|
| +
|
| + # On Windows, stdin is opened as text mode instead of binary which causes
|
| + # problems when piping a binary file, so this switches it to binary mode.
|
| + if IS_WINDOWS and src_url.IsFileUrl() and src_url.IsStream():
|
| + msvcrt.setmode(GetStreamFromFileUrl(src_url).fileno(), os.O_BINARY)
|
| +
|
| + if global_copy_helper_opts.no_clobber:
|
| + # There are two checks to prevent clobbering:
|
| + # 1) The first check is to see if the URL
|
| + # already exists at the destination and prevent the upload/download
|
| + # from happening. This is done by the exists() call.
|
| + # 2) The second check is only relevant if we are writing to gs. We can
|
| + # enforce that the server only writes the object if it doesn't exist
|
| + # by specifying the header below. This check only happens at the
|
| + # server after the complete file has been uploaded. We specify this
|
| + # header to prevent a race condition where a destination file may
|
| + # be created after the first check and before the file is fully
|
| + # uploaded.
|
| + # In order to save on unnecessary uploads/downloads we perform both
|
| + # checks. However, this may come at the cost of additional HTTP calls.
|
| + if preconditions.gen_match:
|
| + raise ArgumentException('Specifying x-goog-if-generation-match is '
|
| + 'not supported with cp -n')
|
| + else:
|
| + preconditions.gen_match = 0
|
| + if dst_url.IsFileUrl() and os.path.exists(dst_url.object_name):
|
| + # The local file may be a partial. Check the file sizes.
|
| + if src_obj_size == os.path.getsize(dst_url.object_name):
|
| + raise ItemExistsError()
|
| + elif dst_url.IsCloudUrl():
|
| + try:
|
| + dst_object = gsutil_api.GetObjectMetadata(
|
| + dst_url.bucket_name, dst_url.object_name, provider=dst_url.scheme)
|
| + except NotFoundException:
|
| + dst_object = None
|
| + if dst_object:
|
| + raise ItemExistsError()
|
| +
|
| + if dst_url.IsCloudUrl():
|
| + # Cloud storage API gets object and bucket name from metadata.
|
| + dst_obj_metadata.name = dst_url.object_name
|
| + dst_obj_metadata.bucket = dst_url.bucket_name
|
| + if src_url.IsCloudUrl():
|
| + # Preserve relevant metadata from the source object if it's not already
|
| + # provided from the headers.
|
| + CopyObjectMetadata(src_obj_metadata, dst_obj_metadata, override=False)
|
| + src_obj_metadata.name = src_url.object_name
|
| + src_obj_metadata.bucket = src_url.bucket_name
|
| + else:
|
| + _SetContentTypeFromFile(src_url, dst_obj_metadata)
|
| + else:
|
| + # Files don't have Cloud API metadata.
|
| + dst_obj_metadata = None
|
| +
|
| + _LogCopyOperation(logger, src_url, dst_url, dst_obj_metadata)
|
| +
|
| + if src_url.IsCloudUrl():
|
| + if dst_url.IsFileUrl():
|
| + return _DownloadObjectToFile(src_url, src_obj_metadata, dst_url,
|
| + gsutil_api, logger, test_method=test_method)
|
| + elif copy_in_the_cloud:
|
| + return _CopyObjToObjInTheCloud(src_url, src_obj_metadata, dst_url,
|
| + dst_obj_metadata, preconditions,
|
| + gsutil_api, logger)
|
| + else:
|
| + return _CopyObjToObjDaisyChainMode(src_url, src_obj_metadata,
|
| + dst_url, dst_obj_metadata,
|
| + preconditions, gsutil_api, logger)
|
| + else: # src_url.IsFileUrl()
|
| + if dst_url.IsCloudUrl():
|
| + return _UploadFileToObject(
|
| + src_url, src_obj_filestream, src_obj_size, dst_url,
|
| + dst_obj_metadata, preconditions, gsutil_api, logger, command_obj,
|
| + copy_exception_handler, gzip_exts=gzip_exts,
|
| + allow_splitting=allow_splitting)
|
| + else: # dst_url.IsFileUrl()
|
| + return _CopyFileToFile(src_url, dst_url)
|
| +
|
| +
|
| +class Manifest(object):
|
| + """Stores the manifest items for the CpCommand class."""
|
| +
|
| + def __init__(self, path):
|
| + # self.items contains a dictionary of rows
|
| + self.items = {}
|
| + self.manifest_filter = {}
|
| + self.lock = CreateLock()
|
| +
|
| + self.manifest_path = os.path.expanduser(path)
|
| + self._ParseManifest()
|
| + self._CreateManifestFile()
|
| +
|
| + def _ParseManifest(self):
|
| + """Load and parse a manifest file.
|
| +
|
| + This information will be used to skip any files that have a skip or OK
|
| + status.
|
| + """
|
| + try:
|
| + if os.path.exists(self.manifest_path):
|
| + with open(self.manifest_path, 'rb') as f:
|
| + first_row = True
|
| + reader = csv.reader(f)
|
| + for row in reader:
|
| + if first_row:
|
| + try:
|
| + source_index = row.index('Source')
|
| + result_index = row.index('Result')
|
| + except ValueError:
|
| + # No header and thus not a valid manifest file.
|
| + raise CommandException(
|
| + 'Missing headers in manifest file: %s' % self.manifest_path)
|
| + first_row = False
|
| + source = row[source_index]
|
| + result = row[result_index]
|
| + if result in ['OK', 'skip']:
|
| + # We're always guaranteed to take the last result of a specific
|
| + # source url.
|
| + self.manifest_filter[source] = result
|
| + except IOError:
|
| + raise CommandException('Could not parse %s' % self.manifest_path)
|
| +
|
| + def WasSuccessful(self, src):
|
| + """Returns whether the specified src url was marked as successful."""
|
| + return src in self.manifest_filter
|
| +
|
| + def _CreateManifestFile(self):
|
| + """Opens the manifest file and assigns it to the file pointer."""
|
| + try:
|
| + if ((not os.path.exists(self.manifest_path))
|
| + or (os.stat(self.manifest_path).st_size == 0)):
|
| + # Add headers to the new file.
|
| + with open(self.manifest_path, 'wb', 1) as f:
|
| + writer = csv.writer(f)
|
| + writer.writerow(['Source',
|
| + 'Destination',
|
| + 'Start',
|
| + 'End',
|
| + 'Md5',
|
| + 'UploadId',
|
| + 'Source Size',
|
| + 'Bytes Transferred',
|
| + 'Result',
|
| + 'Description'])
|
| + except IOError:
|
| + raise CommandException('Could not create manifest file.')
|
| +
|
| + def Set(self, url, key, value):
|
| + if value is None:
|
| + # In case we don't have any information to set we bail out here.
|
| + # This is so that we don't clobber existing information.
|
| + # To zero information pass '' instead of None.
|
| + return
|
| + if url in self.items:
|
| + self.items[url][key] = value
|
| + else:
|
| + self.items[url] = {key: value}
|
| +
|
| + def Initialize(self, source_url, destination_url):
|
| + # Always use the source_url as the key for the item. This is unique.
|
| + self.Set(source_url, 'source_uri', source_url)
|
| + self.Set(source_url, 'destination_uri', destination_url)
|
| + self.Set(source_url, 'start_time', datetime.datetime.utcnow())
|
| +
|
| + def SetResult(self, source_url, bytes_transferred, result,
|
| + description=''):
|
| + self.Set(source_url, 'bytes', bytes_transferred)
|
| + self.Set(source_url, 'result', result)
|
| + self.Set(source_url, 'description', description)
|
| + self.Set(source_url, 'end_time', datetime.datetime.utcnow())
|
| + self._WriteRowToManifestFile(source_url)
|
| + self._RemoveItemFromManifest(source_url)
|
| +
|
| + def _WriteRowToManifestFile(self, url):
|
| + """Writes a manifest entry to the manifest file for the url argument."""
|
| + row_item = self.items[url]
|
| + data = [
|
| + str(row_item['source_uri'].encode(UTF8)),
|
| + str(row_item['destination_uri'].encode(UTF8)),
|
| + '%sZ' % row_item['start_time'].isoformat(),
|
| + '%sZ' % row_item['end_time'].isoformat(),
|
| + row_item['md5'] if 'md5' in row_item else '',
|
| + row_item['upload_id'] if 'upload_id' in row_item else '',
|
| + str(row_item['size']) if 'size' in row_item else '',
|
| + str(row_item['bytes']) if 'bytes' in row_item else '',
|
| + row_item['result'],
|
| + row_item['description'].encode(UTF8)]
|
| +
|
| + # Aquire a lock to prevent multiple threads writing to the same file at
|
| + # the same time. This would cause a garbled mess in the manifest file.
|
| + with self.lock:
|
| + with open(self.manifest_path, 'a', 1) as f: # 1 == line buffered
|
| + writer = csv.writer(f)
|
| + writer.writerow(data)
|
| +
|
| + def _RemoveItemFromManifest(self, url):
|
| + # Remove the item from the dictionary since we're done with it and
|
| + # we don't want the dictionary to grow too large in memory for no good
|
| + # reason.
|
| + del self.items[url]
|
| +
|
| +
|
| +class ItemExistsError(Exception):
|
| + """Exception class for objects that are skipped because they already exist."""
|
| + pass
|
| +
|
| +
|
| +class SkipUnsupportedObjectError(Exception):
|
| + """Exception for objects skipped because they are an unsupported type."""
|
| +
|
| + def __init__(self):
|
| + super(SkipUnsupportedObjectError, self).__init__()
|
| + self.unsupported_type = 'Unknown'
|
| +
|
| +
|
| +class SkipGlacierError(SkipUnsupportedObjectError):
|
| + """Exception for objects skipped because they are an unsupported type."""
|
| +
|
| + def __init__(self):
|
| + super(SkipGlacierError, self).__init__()
|
| + self.unsupported_type = 'GLACIER'
|
| +
|
| +
|
| +def GetPathBeforeFinalDir(url):
|
| + """Returns the path section before the final directory component of the URL.
|
| +
|
| + This handles cases for file system directories, bucket, and bucket
|
| + subdirectories. Example: for gs://bucket/dir/ we'll return 'gs://bucket',
|
| + and for file://dir we'll return file://
|
| +
|
| + Args:
|
| + url: StorageUrl representing a filesystem directory, cloud bucket or
|
| + bucket subdir.
|
| +
|
| + Returns:
|
| + String name of above-described path, sans final path separator.
|
| + """
|
| + sep = url.delim
|
| + if url.IsFileUrl():
|
| + past_scheme = url.url_string[len('file://'):]
|
| + if past_scheme.find(sep) == -1:
|
| + return 'file://'
|
| + else:
|
| + return 'file://%s' % past_scheme.rstrip(sep).rpartition(sep)[0]
|
| + if url.IsBucket():
|
| + return '%s://' % url.scheme
|
| + # Else it names a bucket subdir.
|
| + return url.url_string.rstrip(sep).rpartition(sep)[0]
|
| +
|
| +
|
| +def _DivideAndCeil(dividend, divisor):
|
| + """Returns ceil(dividend / divisor).
|
| +
|
| + Takes care to avoid the pitfalls of floating point arithmetic that could
|
| + otherwise yield the wrong result for large numbers.
|
| +
|
| + Args:
|
| + dividend: Dividend for the operation.
|
| + divisor: Divisor for the operation.
|
| +
|
| + Returns:
|
| + Quotient.
|
| + """
|
| + quotient = dividend // divisor
|
| + if (dividend % divisor) != 0:
|
| + quotient += 1
|
| + return quotient
|
| +
|
| +
|
| +def _GetPartitionInfo(file_size, max_components, default_component_size):
|
| + """Gets info about a file partition for parallel composite uploads.
|
| +
|
| + Args:
|
| + file_size: The number of bytes in the file to be partitioned.
|
| + max_components: The maximum number of components that can be composed.
|
| + default_component_size: The size of a component, assuming that
|
| + max_components is infinite.
|
| + Returns:
|
| + The number of components in the partitioned file, and the size of each
|
| + component (except the last, which will have a different size iff
|
| + file_size != 0 (mod num_components)).
|
| + """
|
| + # num_components = ceil(file_size / default_component_size)
|
| + num_components = _DivideAndCeil(file_size, default_component_size)
|
| +
|
| + # num_components must be in the range [2, max_components]
|
| + num_components = max(min(num_components, max_components), 2)
|
| +
|
| + # component_size = ceil(file_size / num_components)
|
| + component_size = _DivideAndCeil(file_size, num_components)
|
| + return (num_components, component_size)
|
| +
|
| +
|
| +def _DeleteObjectFn(cls, url_to_delete, thread_state=None):
|
| + """Wrapper function to be used with command.Apply()."""
|
| + gsutil_api = GetCloudApiInstance(cls, thread_state)
|
| + gsutil_api.DeleteObject(
|
| + url_to_delete.bucket_name, url_to_delete.object_name,
|
| + generation=url_to_delete.generation, provider=url_to_delete.scheme)
|
| +
|
| +
|
| +def _ParseParallelUploadTrackerFile(tracker_file, tracker_file_lock):
|
| + """Parse the tracker file from the last parallel composite upload attempt.
|
| +
|
| + If it exists, the tracker file is of the format described in
|
| + _CreateParallelUploadTrackerFile. If the file doesn't exist or cannot be
|
| + read, then the upload will start from the beginning.
|
| +
|
| + Args:
|
| + tracker_file: The name of the file to parse.
|
| + tracker_file_lock: Lock protecting access to the tracker file.
|
| +
|
| + Returns:
|
| + random_prefix: A randomly-generated prefix to the name of the
|
| + temporary components.
|
| + existing_objects: A list of ObjectFromTracker objects representing
|
| + the set of files that have already been uploaded.
|
| + """
|
| +
|
| + def GenerateRandomPrefix():
|
| + return str(random.randint(1, (10 ** 10) - 1))
|
| +
|
| + existing_objects = []
|
| + try:
|
| + with tracker_file_lock:
|
| + with open(tracker_file, 'r') as fp:
|
| + lines = fp.readlines()
|
| + lines = [line.strip() for line in lines]
|
| + if not lines:
|
| + print('Parallel upload tracker file (%s) was invalid. '
|
| + 'Restarting upload from scratch.' % tracker_file)
|
| + lines = [GenerateRandomPrefix()]
|
| +
|
| + except IOError as e:
|
| + # We can't read the tracker file, so generate a new random prefix.
|
| + lines = [GenerateRandomPrefix()]
|
| +
|
| + # Ignore non-existent file (happens first time an upload
|
| + # is attempted on a file), but warn user for other errors.
|
| + if e.errno != errno.ENOENT:
|
| + # Will restart because we failed to read in the file.
|
| + print('Couldn\'t read parallel upload tracker file (%s): %s. '
|
| + 'Restarting upload from scratch.' % (tracker_file, e.strerror))
|
| +
|
| + # The first line contains the randomly-generated prefix.
|
| + random_prefix = lines[0]
|
| +
|
| + # The remaining lines were written in pairs to describe a single component
|
| + # in the form:
|
| + # object_name (without random prefix)
|
| + # generation
|
| + # Newlines are used as the delimiter because only newlines and carriage
|
| + # returns are invalid characters in object names, and users can specify
|
| + # a custom prefix in the config file.
|
| + i = 1
|
| + while i < len(lines):
|
| + (name, generation) = (lines[i], lines[i+1])
|
| + if not generation:
|
| + # Cover the '' case.
|
| + generation = None
|
| + existing_objects.append(ObjectFromTracker(name, generation))
|
| + i += 2
|
| + return (random_prefix, existing_objects)
|
| +
|
| +
|
| +def _AppendComponentTrackerToParallelUploadTrackerFile(tracker_file, component,
|
| + tracker_file_lock):
|
| + """Appends info about the uploaded component to an existing tracker file.
|
| +
|
| + Follows the format described in _CreateParallelUploadTrackerFile.
|
| +
|
| + Args:
|
| + tracker_file: Tracker file to append to.
|
| + component: Component that was uploaded.
|
| + tracker_file_lock: Thread and process-safe Lock for the tracker file.
|
| + """
|
| + lines = _GetParallelUploadTrackerFileLinesForComponents([component])
|
| + lines = [line + '\n' for line in lines]
|
| + with tracker_file_lock:
|
| + with open(tracker_file, 'a') as f:
|
| + f.writelines(lines)
|
| +
|
| +
|
| +def _CreateParallelUploadTrackerFile(tracker_file, random_prefix, components,
|
| + tracker_file_lock):
|
| + """Writes information about components that were successfully uploaded.
|
| +
|
| + This way the upload can be resumed at a later date. The tracker file has
|
| + the format:
|
| + random_prefix
|
| + temp_object_1_name
|
| + temp_object_1_generation
|
| + .
|
| + .
|
| + .
|
| + temp_object_N_name
|
| + temp_object_N_generation
|
| + where N is the number of components that have been successfully uploaded.
|
| +
|
| + Args:
|
| + tracker_file: The name of the parallel upload tracker file.
|
| + random_prefix: The randomly-generated prefix that was used for
|
| + for uploading any existing components.
|
| + components: A list of ObjectFromTracker objects that were uploaded.
|
| + tracker_file_lock: The lock protecting access to the tracker file.
|
| + """
|
| + lines = [random_prefix]
|
| + lines += _GetParallelUploadTrackerFileLinesForComponents(components)
|
| + lines = [line + '\n' for line in lines]
|
| + try:
|
| + with tracker_file_lock:
|
| + open(tracker_file, 'w').close() # Clear the file.
|
| + with open(tracker_file, 'w') as f:
|
| + f.writelines(lines)
|
| + except IOError as e:
|
| + RaiseUnwritableTrackerFileException(tracker_file, e.strerror)
|
| +
|
| +
|
| +def _GetParallelUploadTrackerFileLinesForComponents(components):
|
| + """Return a list of the lines for use in a parallel upload tracker file.
|
| +
|
| + The lines represent the given components, using the format as described in
|
| + _CreateParallelUploadTrackerFile.
|
| +
|
| + Args:
|
| + components: A list of ObjectFromTracker objects that were uploaded.
|
| +
|
| + Returns:
|
| + Lines describing components with their generation for outputting to the
|
| + tracker file.
|
| + """
|
| + lines = []
|
| + for component in components:
|
| + generation = None
|
| + generation = component.generation
|
| + if not generation:
|
| + generation = ''
|
| + lines += [component.object_name, str(generation)]
|
| + return lines
|
| +
|
| +
|
| +def FilterExistingComponents(dst_args, existing_components, bucket_url,
|
| + gsutil_api):
|
| + """Determines course of action for component objects.
|
| +
|
| + Given the list of all target objects based on partitioning the file and
|
| + the list of objects that have already been uploaded successfully,
|
| + this function determines which objects should be uploaded, which
|
| + existing components are still valid, and which existing components should
|
| + be deleted.
|
| +
|
| + Args:
|
| + dst_args: The map of file_name -> PerformParallelUploadFileToObjectArgs
|
| + calculated by partitioning the file.
|
| + existing_components: A list of ObjectFromTracker objects that have been
|
| + uploaded in the past.
|
| + bucket_url: CloudUrl of the bucket in which the components exist.
|
| + gsutil_api: gsutil Cloud API instance to use for retrieving object metadata.
|
| +
|
| + Returns:
|
| + components_to_upload: List of components that need to be uploaded.
|
| + uploaded_components: List of components that have already been
|
| + uploaded and are still valid.
|
| + existing_objects_to_delete: List of components that have already
|
| + been uploaded, but are no longer valid
|
| + and are in a versioned bucket, and
|
| + therefore should be deleted.
|
| + """
|
| + components_to_upload = []
|
| + existing_component_names = [component.object_name
|
| + for component in existing_components]
|
| + for component_name in dst_args:
|
| + if component_name not in existing_component_names:
|
| + components_to_upload.append(dst_args[component_name])
|
| +
|
| + objects_already_chosen = []
|
| +
|
| + # Don't reuse any temporary components whose MD5 doesn't match the current
|
| + # MD5 of the corresponding part of the file. If the bucket is versioned,
|
| + # also make sure that we delete the existing temporary version.
|
| + existing_objects_to_delete = []
|
| + uploaded_components = []
|
| + for tracker_object in existing_components:
|
| + if (tracker_object.object_name not in dst_args.keys()
|
| + or tracker_object.object_name in objects_already_chosen):
|
| + # This could happen if the component size has changed. This also serves
|
| + # to handle object names that get duplicated in the tracker file due
|
| + # to people doing things they shouldn't (e.g., overwriting an existing
|
| + # temporary component in a versioned bucket).
|
| +
|
| + url = bucket_url.Clone()
|
| + url.object_name = tracker_object.object_name
|
| + url.generation = tracker_object.generation
|
| + existing_objects_to_delete.append(url)
|
| + continue
|
| +
|
| + dst_arg = dst_args[tracker_object.object_name]
|
| + file_part = FilePart(dst_arg.filename, dst_arg.file_start,
|
| + dst_arg.file_length)
|
| + # TODO: calculate MD5's in parallel when possible.
|
| + content_md5 = CalculateB64EncodedMd5FromContents(file_part)
|
| +
|
| + try:
|
| + # Get the MD5 of the currently-existing component.
|
| + dst_url = dst_arg.dst_url
|
| + dst_metadata = gsutil_api.GetObjectMetadata(
|
| + dst_url.bucket_name, dst_url.object_name,
|
| + generation=dst_url.generation, provider=dst_url.scheme,
|
| + fields=['md5Hash', 'etag'])
|
| + cloud_md5 = dst_metadata.md5Hash
|
| + except Exception: # pylint: disable=broad-except
|
| + # We don't actually care what went wrong - we couldn't retrieve the
|
| + # object to check the MD5, so just upload it again.
|
| + cloud_md5 = None
|
| +
|
| + if cloud_md5 != content_md5:
|
| + components_to_upload.append(dst_arg)
|
| + objects_already_chosen.append(tracker_object.object_name)
|
| + if tracker_object.generation:
|
| + # If the old object doesn't have a generation (i.e., it isn't in a
|
| + # versioned bucket), then we will just overwrite it anyway.
|
| + invalid_component_with_generation = dst_arg.dst_url.Clone()
|
| + invalid_component_with_generation.generation = tracker_object.generation
|
| + existing_objects_to_delete.append(invalid_component_with_generation)
|
| + else:
|
| + url = dst_arg.dst_url.Clone()
|
| + url.generation = tracker_object.generation
|
| + uploaded_components.append(url)
|
| + objects_already_chosen.append(tracker_object.object_name)
|
| +
|
| + if uploaded_components:
|
| + logging.info('Found %d existing temporary components to reuse.',
|
| + len(uploaded_components))
|
| +
|
| + return (components_to_upload, uploaded_components,
|
| + existing_objects_to_delete)
|
|
|
Chromium Code Reviews
Issue 1377933002:
[catapult] - Copy Telemetry's gsutilz over to third_party. (Closed)
Base URL: https://github.com/catapult-project/catapult.git@master