| # -*- 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) |