Index: tools/telemetry/third_party/gsutil/gslib/commands/rsync.py |
diff --git a/tools/telemetry/third_party/gsutil/gslib/commands/rsync.py b/tools/telemetry/third_party/gsutil/gslib/commands/rsync.py |
deleted file mode 100644 |
index 4eb9b92e92ad9eb1155393634fc999e8f3cbb7e0..0000000000000000000000000000000000000000 |
--- a/tools/telemetry/third_party/gsutil/gslib/commands/rsync.py |
+++ /dev/null |
@@ -1,1032 +0,0 @@ |
-# -*- coding: utf-8 -*- |
-# Copyright 2014 Google Inc. All Rights Reserved. |
-# |
-# 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. |
-"""Implementation of Unix-like rsync command.""" |
- |
-from __future__ import absolute_import |
- |
-import errno |
-import heapq |
-import io |
-from itertools import islice |
-import os |
-import re |
-import tempfile |
-import textwrap |
-import traceback |
-import urllib |
- |
-from boto import config |
-import crcmod |
- |
-from gslib import copy_helper |
-from gslib.cloud_api import NotFoundException |
-from gslib.command import Command |
-from gslib.command import DummyArgChecker |
-from gslib.command_argument import CommandArgument |
-from gslib.copy_helper import CreateCopyHelperOpts |
-from gslib.copy_helper import SkipUnsupportedObjectError |
-from gslib.cs_api_map import ApiSelector |
-from gslib.exception import CommandException |
-from gslib.hashing_helper import CalculateB64EncodedCrc32cFromContents |
-from gslib.hashing_helper import CalculateB64EncodedMd5FromContents |
-from gslib.hashing_helper import SLOW_CRCMOD_WARNING |
-from gslib.plurality_checkable_iterator import PluralityCheckableIterator |
-from gslib.sig_handling import GetCaughtSignals |
-from gslib.sig_handling import RegisterSignalHandler |
-from gslib.storage_url import StorageUrlFromString |
-from gslib.util import GetCloudApiInstance |
-from gslib.util import IsCloudSubdirPlaceholder |
-from gslib.util import TEN_MIB |
-from gslib.util import UsingCrcmodExtension |
-from gslib.util import UTF8 |
-from gslib.wildcard_iterator import CreateWildcardIterator |
- |
- |
-_SYNOPSIS = """ |
- gsutil rsync [-c] [-C] [-d] [-e] [-n] [-p] [-r] [-U] [-x] src_url dst_url |
-""" |
- |
-_DETAILED_HELP_TEXT = (""" |
-<B>SYNOPSIS</B> |
-""" + _SYNOPSIS + """ |
- |
- |
-<B>DESCRIPTION</B> |
- The gsutil rsync command makes the contents under dst_url the same as the |
- contents under src_url, by copying any missing files/objects, and (if the |
- -d option is specified) deleting any extra files/objects. For example, to |
- make gs://mybucket/data match the contents of the local directory "data" |
- you could do: |
- |
- gsutil rsync -d data gs://mybucket/data |
- |
- To recurse into directories use the -r option: |
- |
- gsutil rsync -d -r data gs://mybucket/data |
- |
- To copy only new/changed files without deleting extra files from |
- gs://mybucket/data leave off the -d option: |
- |
- gsutil rsync -r data gs://mybucket/data |
- |
- If you have a large number of objects to synchronize you might want to use the |
- gsutil -m option, to perform parallel (multi-threaded/multi-processing) |
- synchronization: |
- |
- gsutil -m rsync -d -r data gs://mybucket/data |
- |
- The -m option typically will provide a large performance boost if either the |
- source or destination (or both) is a cloud URL. If both source and |
- destination are file URLs the -m option will typically thrash the disk and |
- slow synchronization down. |
- |
- To make the local directory "data" the same as the contents of |
- gs://mybucket/data: |
- |
- gsutil rsync -d -r gs://mybucket/data data |
- |
- To make the contents of gs://mybucket2 the same as gs://mybucket1: |
- |
- gsutil rsync -d -r gs://mybucket1 gs://mybucket2 |
- |
- You can also mirror data across local directories: |
- |
- gsutil rsync -d -r dir1 dir2 |
- |
- To mirror your content across clouds: |
- |
- gsutil rsync -d -r gs://my-gs-bucket s3://my-s3-bucket |
- |
- Note: If you are synchronizing a large amount of data between clouds you might |
- consider setting up a |
- `Google Compute Engine <https://cloud.google.com/products/compute-engine>`_ |
- account and running gsutil there. Since cross-provider gsutil data transfers |
- flow through the machine where gsutil is running, doing this can make your |
- transfer run significantly faster than running gsutil on your local |
- workstation. |
- |
- |
-<B>BE CAREFUL WHEN USING -d OPTION!</B> |
- The rsync -d option is very useful and commonly used, because it provides a |
- means of making the contents of a destination bucket or directory match those |
- of a source bucket or directory. However, please exercise caution when you |
- use this option: It's possible to delete large amounts of data accidentally |
- if, for example, you erroneously reverse source and destination. For example, |
- if you meant to synchronize a local directory from a bucket in the cloud but |
- instead run the command: |
- |
- gsutil -m rsync -r -d ./your-dir gs://your-bucket |
- |
- and your-dir is currently empty, you will quickly delete all of the objects in |
- gs://your-bucket. |
- |
- You can also cause large amounts of data to be lost quickly by specifying a |
- subdirectory of the destination as the source of an rsync. For example, the |
- command: |
- |
- gsutil -m rsync -r -d gs://your-bucket/data gs://your-bucket |
- |
- would cause most or all of the objects in gs://your-bucket to be deleted |
- (some objects may survive if there are any with names that sort lower than |
- "data" under gs://your-bucket/data). |
- |
- In addition to paying careful attention to the source and destination you |
- specify with the rsync command, there are two more safety measures your can |
- take when using gsutil rsync -d: |
- |
- 1. Try running the command with the rsync -n option first, to see what it |
- would do without actually performing the operations. For example, if |
- you run the command: |
- |
- gsutil -m rsync -r -d -n gs://your-bucket/data gs://your-bucket |
- |
- it will be immediately evident that running that command without the -n |
- option would cause many objects to be deleted. |
- |
- 2. Enable object versioning in your bucket, which will allow you to restore |
- objects if you accidentally delete them. For more details see |
- "gsutil help versions". |
- |
- |
-<B>IMPACT OF BUCKET LISTING EVENTUAL CONSISTENCY</B> |
- The rsync command operates by listing the source and destination URLs, and |
- then performing copy and remove operations according to the differences |
- between these listings. Because bucket listing is eventually (not strongly) |
- consistent, if you upload new objects or delete objects from a bucket and then |
- immediately run gsutil rsync with that bucket as the source or destination, |
- it's possible the rsync command will not see the recent updates and thus |
- synchronize incorrectly. You can rerun the rsync operation again later to |
- correct the incorrect synchronization. |
- |
- |
-<B>CHECKSUM VALIDATION AND FAILURE HANDLING</B> |
- At the end of every upload or download, the gsutil rsync command validates |
- that the checksum of the source file/object matches the checksum of the |
- destination file/object. If the checksums do not match, gsutil will delete |
- the invalid copy and print a warning message. This very rarely happens, but |
- if it does, please contact gs-team@google.com. |
- |
- The rsync command will retry when failures occur, but if enough failures |
- happen during a particular copy or delete operation the command will skip that |
- object and move on. At the end of the synchronization run if any failures were |
- not successfully retried, the rsync command will report the count of failures, |
- and exit with non-zero status. At this point you can run the rsync command |
- again, and it will attempt any remaining needed copy and/or delete operations. |
- |
- Note that there are cases where retrying will never succeed, such as if you |
- don't have write permission to the destination bucket or if the destination |
- path for some objects is longer than the maximum allowed length. |
- |
- For more details about gsutil's retry handling, please see |
- "gsutil help retries". |
- |
- |
-<B>CHANGE DETECTION ALGORITHM</B> |
- To determine if a file or object has changed gsutil rsync first checks whether |
- the source and destination sizes match. If they match, it next checks if their |
- checksums match, using checksums if available (see below). Unlike the Unix |
- rsync command, gsutil rsync does not use timestamps to determine if the |
- file/object changed, because the GCS API does not permit the caller to set an |
- object's timestamp (hence, timestamps of identical files/objects cannot be |
- made to match). |
- |
- Checksums will not be available in two cases: |
- |
- 1. When synchronizing to or from a file system. By default, gsutil does not |
- checksum files, because of the slowdown caused when working with large |
- files. You can cause gsutil to checksum files by using the gsutil rsync -c |
- option, at the cost of increased local disk I/O and run time when working |
- with large files. You should consider using the -c option if your files can |
- change without changing sizes (e.g., if you have files that contain fixed |
- width data, such as timestamps). |
- |
- 2. When comparing composite GCS objects with objects at a cloud provider that |
- does not support CRC32C (which is the only checksum available for composite |
- objects). See 'gsutil help compose' for details about composite objects. |
- |
- |
-<B>COPYING IN THE CLOUD AND METADATA PRESERVATION</B> |
- If both the source and destination URL are cloud URLs from the same provider, |
- gsutil copies data "in the cloud" (i.e., without downloading to and uploading |
- from the machine where you run gsutil). In addition to the performance and |
- cost advantages of doing this, copying in the cloud preserves metadata (like |
- Content-Type and Cache-Control). In contrast, when you download data from the |
- cloud it ends up in a file, which has no associated metadata. Thus, unless you |
- have some way to hold on to or re-create that metadata, synchronizing a bucket |
- to a directory in the local file system will not retain the metadata. |
- |
- Note that by default, the gsutil rsync command does not copy the ACLs of |
- objects being synchronized and instead will use the default bucket ACL (see |
- "gsutil help defacl"). You can override this behavior with the -p option (see |
- OPTIONS below). |
- |
- |
-<B>SLOW CHECKSUMS</B> |
- If you find that CRC32C checksum computation runs slowly, this is likely |
- because you don't have a compiled CRC32c on your system. Try running: |
- |
- gsutil ver -l |
- |
- If the output contains: |
- |
- compiled crcmod: False |
- |
- you are running a Python library for computing CRC32C, which is much slower |
- than using the compiled code. For information on getting a compiled CRC32C |
- implementation, see 'gsutil help crc32c'. |
- |
- |
-<B>LIMITATIONS</B> |
- 1. The gsutil rsync command doesn't make the destination object's timestamps |
- match those of the source object (it can't; timestamp setting is not |
- allowed by the GCS API). |
- |
- 2. The gsutil rsync command ignores versioning, synchronizing only the live |
- object versions in versioned buckets. |
- |
- |
-<B>OPTIONS</B> |
- -c Causes the rsync command to compute checksums for files if the |
- size of source and destination match, and then compare |
- checksums. This option increases local disk I/O and run time |
- if either src_url or dst_url are on the local file system. |
- |
- -C If an error occurs, continue to attempt to copy the remaining |
- files. If errors occurred, gsutil's exit status will be non-zero |
- even if this flag is set. This option is implicitly set when |
- running "gsutil -m rsync...". Note: -C only applies to the |
- actual copying operation. If an error occurs while iterating |
- over the files in the local directory (e.g., invalid Unicode |
- file name) gsutil will print an error message and abort. |
- |
- -d Delete extra files under dst_url not found under src_url. By |
- default extra files are not deleted. Note: this option can |
- delete data quickly if you specify the wrong source/destination |
- combination. See the help section above, |
- "BE CAREFUL WHEN USING -d OPTION!". |
- |
- -e Exclude symlinks. When specified, symbolic links will be |
- ignored. |
- |
- -n Causes rsync to run in "dry run" mode, i.e., just outputting |
- what would be copied or deleted without actually doing any |
- copying/deleting. |
- |
- -p Causes ACLs to be preserved when synchronizing in the cloud. |
- Note that this option has performance and cost implications when |
- using the XML API, as it requires separate HTTP calls for |
- interacting with ACLs. The performance issue can be mitigated to |
- some degree by using gsutil -m rsync to cause parallel |
- synchronization. Also, this option only works if you have OWNER |
- access to all of the objects that are copied. |
- |
- You can avoid the additional performance and cost of using |
- rsync -p if you want all objects in the destination bucket to |
- end up with the same ACL by setting a default object ACL on that |
- bucket instead of using rsync -p. See 'help gsutil defacl'. |
- |
- -R, -r Causes directories, buckets, and bucket subdirectories to be |
- synchronized recursively. If you neglect to use this option |
- gsutil will make only the top-level directory in the source |
- and destination URLs match, skipping any sub-directories. |
- |
- -U Skip objects with unsupported object types instead of failing. |
- Unsupported object types are s3 glacier objects. |
- |
- -x pattern Causes files/objects matching pattern to be excluded, i.e., any |
- matching files/objects will not be copied or deleted. Note that |
- the pattern is a Python regular expression, not a wildcard (so, |
- matching any string ending in 'abc' would be specified using |
- '.*abc' rather than '*abc'). Note also that the exclude path is |
- always relative (similar to Unix rsync or tar exclude options). |
- For example, if you run the command: |
- |
- gsutil rsync -x 'data./.*\\.txt' dir gs://my-bucket |
- |
- it will skip the file dir/data1/a.txt. |
- |
- You can use regex alternation to specify multiple exclusions, |
- for example: |
- |
- gsutil rsync -x '.*\\.txt|.*\\.jpg' dir gs://my-bucket |
-""") |
- |
- |
-class _DiffAction(object): |
- COPY = 'copy' |
- REMOVE = 'remove' |
- |
- |
-_NA = '-' |
-_OUTPUT_BUFFER_SIZE = 64 * 1024 |
-_PROGRESS_REPORT_LISTING_COUNT = 10000 |
- |
- |
-# Tracks files we need to clean up at end or if interrupted. |
-_tmp_files = [] |
- |
- |
-# pylint: disable=unused-argument |
-def _HandleSignals(signal_num, cur_stack_frame): |
- """Called when rsync command is killed with SIGINT, SIGQUIT or SIGTERM.""" |
- CleanUpTempFiles() |
- |
- |
-def CleanUpTempFiles(): |
- """Cleans up temp files. |
- |
- This function allows the main (RunCommand) function to clean up at end of |
- operation, or if gsutil rsync is interrupted (e.g., via ^C). This is necessary |
- because tempfile.NamedTemporaryFile doesn't allow the created file to be |
- re-opened in read mode on Windows, so we have to use tempfile.mkstemp, which |
- doesn't automatically delete temp files. |
- """ |
- try: |
- for fname in _tmp_files: |
- os.unlink(fname) |
- except: # pylint: disable=bare-except |
- pass |
- |
- |
-class _DiffToApply(object): |
- """Class that encapsulates info needed to apply diff for one object.""" |
- |
- def __init__(self, src_url_str, dst_url_str, diff_action): |
- """Constructor. |
- |
- Args: |
- src_url_str: The source URL string, or None if diff_action is REMOVE. |
- dst_url_str: The destination URL string. |
- diff_action: _DiffAction to be applied. |
- """ |
- self.src_url_str = src_url_str |
- self.dst_url_str = dst_url_str |
- self.diff_action = diff_action |
- |
- |
-def _DiffToApplyArgChecker(command_instance, diff_to_apply): |
- """Arg checker that skips symlinks if -e flag specified.""" |
- if (diff_to_apply.diff_action == _DiffAction.REMOVE |
- or not command_instance.exclude_symlinks): |
- # No src URL is populated for REMOVE actions. |
- return True |
- exp_src_url = StorageUrlFromString(diff_to_apply.src_url_str) |
- if exp_src_url.IsFileUrl() and os.path.islink(exp_src_url.object_name): |
- command_instance.logger.info('Skipping symbolic link %s...', exp_src_url) |
- return False |
- return True |
- |
- |
-def _ComputeNeededFileChecksums(logger, src_url_str, src_size, src_crc32c, |
- src_md5, dst_url_str, dst_size, dst_crc32c, |
- dst_md5): |
- """Computes any file checksums needed by _ObjectsMatch. |
- |
- Args: |
- logger: logging.logger for outputting log messages. |
- src_url_str: Source URL string. |
- src_size: Source size |
- src_crc32c: Source CRC32c. |
- src_md5: Source MD5. |
- dst_url_str: Destination URL string. |
- dst_size: Destination size |
- dst_crc32c: Destination CRC32c. |
- dst_md5: Destination MD5. |
- |
- Returns: |
- (src_crc32c, src_md5, dst_crc32c, dst_md5) |
- """ |
- src_url = StorageUrlFromString(src_url_str) |
- dst_url = StorageUrlFromString(dst_url_str) |
- if src_url.IsFileUrl(): |
- if dst_crc32c != _NA or dst_url.IsFileUrl(): |
- if src_size > TEN_MIB: |
- logger.info('Computing MD5 for %s...', src_url_str) |
- with open(src_url.object_name, 'rb') as fp: |
- src_crc32c = CalculateB64EncodedCrc32cFromContents(fp) |
- elif dst_md5 != _NA or dst_url.IsFileUrl(): |
- if dst_size > TEN_MIB: |
- logger.info('Computing MD5 for %s...', dst_url_str) |
- with open(src_url.object_name, 'rb') as fp: |
- src_md5 = CalculateB64EncodedMd5FromContents(fp) |
- if dst_url.IsFileUrl(): |
- if src_crc32c != _NA: |
- if src_size > TEN_MIB: |
- logger.info('Computing CRC32C for %s...', src_url_str) |
- with open(dst_url.object_name, 'rb') as fp: |
- dst_crc32c = CalculateB64EncodedCrc32cFromContents(fp) |
- elif src_md5 != _NA: |
- if dst_size > TEN_MIB: |
- logger.info('Computing CRC32C for %s...', dst_url_str) |
- with open(dst_url.object_name, 'rb') as fp: |
- dst_md5 = CalculateB64EncodedMd5FromContents(fp) |
- return (src_crc32c, src_md5, dst_crc32c, dst_md5) |
- |
- |
-def _ListUrlRootFunc(cls, args_tuple, thread_state=None): |
- """Worker function for listing files/objects under to be sync'd. |
- |
- Outputs sorted list to out_file_name, formatted per _BuildTmpOutputLine. We |
- sort the listed URLs because we don't want to depend on consistent sort |
- order across file systems and cloud providers. |
- |
- Args: |
- cls: Command instance. |
- args_tuple: (base_url_str, out_file_name, desc), where base_url_str is |
- top-level URL string to list; out_filename is name of file to |
- which sorted output should be written; desc is 'source' or |
- 'destination'. |
- thread_state: gsutil Cloud API instance to use. |
- """ |
- gsutil_api = GetCloudApiInstance(cls, thread_state=thread_state) |
- (base_url_str, out_filename, desc) = args_tuple |
- # We sort while iterating over base_url_str, allowing parallelism of batched |
- # sorting with collecting the listing. |
- out_file = io.open(out_filename, mode='w', encoding=UTF8) |
- try: |
- _BatchSort(_FieldedListingIterator(cls, gsutil_api, base_url_str, desc), |
- out_file) |
- except Exception as e: # pylint: disable=broad-except |
- # Abandon rsync if an exception percolates up to this layer - retryable |
- # exceptions are handled in the lower layers, so we got a non-retryable |
- # exception (like 404 bucket not found) and proceeding would either be |
- # futile or could result in data loss - for example: |
- # gsutil rsync -d gs://non-existent-bucket ./localdir |
- # would delete files from localdir. |
- cls.logger.error( |
- 'Caught non-retryable exception while listing %s: %s' % |
- (base_url_str, e)) |
- cls.non_retryable_listing_failures = 1 |
- out_file.close() |
- |
- |
-def _FieldedListingIterator(cls, gsutil_api, base_url_str, desc): |
- """Iterator over base_url_str formatting output per _BuildTmpOutputLine. |
- |
- Args: |
- cls: Command instance. |
- gsutil_api: gsutil Cloud API instance to use for bucket listing. |
- base_url_str: The top-level URL string over which to iterate. |
- desc: 'source' or 'destination'. |
- |
- Yields: |
- Output line formatted per _BuildTmpOutputLine. |
- """ |
- if cls.recursion_requested: |
- wildcard = '%s/**' % base_url_str.rstrip('/\\') |
- else: |
- wildcard = '%s/*' % base_url_str.rstrip('/\\') |
- i = 0 |
- for blr in CreateWildcardIterator( |
- wildcard, gsutil_api, debug=cls.debug, |
- project_id=cls.project_id).IterObjects( |
- # Request just the needed fields, to reduce bandwidth usage. |
- bucket_listing_fields=['crc32c', 'md5Hash', 'name', 'size']): |
- # Various GUI tools (like the GCS web console) create placeholder objects |
- # ending with '/' when the user creates an empty directory. Normally these |
- # tools should delete those placeholders once objects have been written |
- # "under" the directory, but sometimes the placeholders are left around. |
- # We need to filter them out here, otherwise if the user tries to rsync |
- # from GCS to a local directory it will result in a directory/file |
- # conflict (e.g., trying to download an object called "mydata/" where the |
- # local directory "mydata" exists). |
- url = blr.storage_url |
- if IsCloudSubdirPlaceholder(url, blr=blr): |
- cls.logger.info('Skipping cloud sub-directory placeholder object (%s) ' |
- 'because such objects aren\'t needed in (and would ' |
- 'interfere with) directories in the local file system', |
- url) |
- continue |
- if (cls.exclude_symlinks and url.IsFileUrl() |
- and os.path.islink(url.object_name)): |
- continue |
- if cls.exclude_pattern: |
- str_to_check = url.url_string[len(base_url_str):] |
- if str_to_check.startswith(url.delim): |
- str_to_check = str_to_check[1:] |
- if cls.exclude_pattern.match(str_to_check): |
- continue |
- i += 1 |
- if i % _PROGRESS_REPORT_LISTING_COUNT == 0: |
- cls.logger.info('At %s listing %d...', desc, i) |
- yield _BuildTmpOutputLine(blr) |
- |
- |
-def _BuildTmpOutputLine(blr): |
- """Builds line to output to temp file for given BucketListingRef. |
- |
- Args: |
- blr: The BucketListingRef. |
- |
- Returns: |
- The output line, formatted as _EncodeUrl(URL)<sp>size<sp>crc32c<sp>md5 |
- where crc32c will only be present for GCS URLs, and md5 will only be |
- present for cloud URLs that aren't composite objects. A missing field is |
- populated with '-'. |
- """ |
- crc32c = _NA |
- md5 = _NA |
- url = blr.storage_url |
- if url.IsFileUrl(): |
- size = os.path.getsize(url.object_name) |
- elif url.IsCloudUrl(): |
- size = blr.root_object.size |
- crc32c = blr.root_object.crc32c or _NA |
- md5 = blr.root_object.md5Hash or _NA |
- else: |
- raise CommandException('Got unexpected URL type (%s)' % url.scheme) |
- return '%s %d %s %s\n' % (_EncodeUrl(url.url_string), size, crc32c, md5) |
- |
- |
-def _EncodeUrl(url_string): |
- """Encodes url_str with quote plus encoding and UTF8 character encoding. |
- |
- We use this for all URL encodings. |
- |
- Args: |
- url_string: String URL to encode. |
- |
- Returns: |
- encoded URL. |
- """ |
- return urllib.quote_plus(url_string.encode(UTF8)) |
- |
- |
-def _DecodeUrl(enc_url_string): |
- """Inverts encoding from EncodeUrl. |
- |
- Args: |
- enc_url_string: String URL to decode. |
- |
- Returns: |
- decoded URL. |
- """ |
- return urllib.unquote_plus(enc_url_string).decode(UTF8) |
- |
- |
-# pylint: disable=bare-except |
-def _BatchSort(in_iter, out_file): |
- """Sorts input lines from in_iter and outputs to out_file. |
- |
- Sorts in batches as input arrives, so input file does not need to be loaded |
- into memory all at once. Derived from Python Recipe 466302: Sorting big |
- files the Python 2.4 way by Nicolas Lehuen. |
- |
- Sorted format is per _BuildTmpOutputLine. We're sorting on the entire line |
- when we could just sort on the first record (URL); but the sort order is |
- identical either way. |
- |
- Args: |
- in_iter: Input iterator. |
- out_file: Output file. |
- """ |
- # Note: If chunk_files gets very large we can run out of open FDs. See .boto |
- # file comments about rsync_buffer_lines. If increasing rsync_buffer_lines |
- # doesn't suffice (e.g., for someone synchronizing with a really large |
- # bucket), an option would be to make gsutil merge in passes, never |
- # opening all chunk files simultaneously. |
- buffer_size = config.getint('GSUtil', 'rsync_buffer_lines', 32000) |
- chunk_files = [] |
- try: |
- while True: |
- current_chunk = sorted(islice(in_iter, buffer_size)) |
- if not current_chunk: |
- break |
- output_chunk = io.open('%s-%06i' % (out_file.name, len(chunk_files)), |
- mode='w+', encoding=UTF8) |
- chunk_files.append(output_chunk) |
- output_chunk.writelines(unicode(''.join(current_chunk))) |
- output_chunk.flush() |
- output_chunk.seek(0) |
- out_file.writelines(heapq.merge(*chunk_files)) |
- except IOError as e: |
- if e.errno == errno.EMFILE: |
- raise CommandException('\n'.join(textwrap.wrap( |
- 'Synchronization failed because too many open file handles were ' |
- 'needed while building synchronization state. Please see the ' |
- 'comments about rsync_buffer_lines in your .boto config file for a ' |
- 'possible way to address this problem.'))) |
- raise |
- finally: |
- for chunk_file in chunk_files: |
- try: |
- chunk_file.close() |
- os.remove(chunk_file.name) |
- except: |
- pass |
- |
- |
-class _DiffIterator(object): |
- """Iterator yielding sequence of _DiffToApply objects.""" |
- |
- def __init__(self, command_obj, base_src_url, base_dst_url): |
- self.command_obj = command_obj |
- self.compute_file_checksums = command_obj.compute_file_checksums |
- self.delete_extras = command_obj.delete_extras |
- self.recursion_requested = command_obj.recursion_requested |
- self.logger = self.command_obj.logger |
- self.base_src_url = base_src_url |
- self.base_dst_url = base_dst_url |
- self.logger.info('Building synchronization state...') |
- |
- (src_fh, self.sorted_list_src_file_name) = tempfile.mkstemp( |
- prefix='gsutil-rsync-src-') |
- _tmp_files.append(self.sorted_list_src_file_name) |
- (dst_fh, self.sorted_list_dst_file_name) = tempfile.mkstemp( |
- prefix='gsutil-rsync-dst-') |
- _tmp_files.append(self.sorted_list_dst_file_name) |
- # Close the file handles; the file will be opened in write mode by |
- # _ListUrlRootFunc. |
- os.close(src_fh) |
- os.close(dst_fh) |
- |
- # Build sorted lists of src and dst URLs in parallel. To do this, pass args |
- # to _ListUrlRootFunc as tuple (base_url_str, out_filename, desc) |
- # where base_url_str is the starting URL string for listing. |
- args_iter = iter([ |
- (self.base_src_url.url_string, self.sorted_list_src_file_name, |
- 'source'), |
- (self.base_dst_url.url_string, self.sorted_list_dst_file_name, |
- 'destination') |
- ]) |
- |
- # Contains error message from non-retryable listing failure. |
- command_obj.non_retryable_listing_failures = 0 |
- shared_attrs = ['non_retryable_listing_failures'] |
- command_obj.Apply(_ListUrlRootFunc, args_iter, _RootListingExceptionHandler, |
- shared_attrs, arg_checker=DummyArgChecker, |
- parallel_operations_override=True, |
- fail_on_error=True) |
- |
- if command_obj.non_retryable_listing_failures: |
- raise CommandException('Caught non-retryable exception - aborting rsync') |
- |
- self.sorted_list_src_file = open(self.sorted_list_src_file_name, 'r') |
- self.sorted_list_dst_file = open(self.sorted_list_dst_file_name, 'r') |
- |
- # Wrap iterators in PluralityCheckableIterator so we can check emptiness. |
- self.sorted_src_urls_it = PluralityCheckableIterator( |
- iter(self.sorted_list_src_file)) |
- self.sorted_dst_urls_it = PluralityCheckableIterator( |
- iter(self.sorted_list_dst_file)) |
- |
- def _ParseTmpFileLine(self, line): |
- """Parses output from _BuildTmpOutputLine. |
- |
- Parses into tuple: |
- (URL, size, crc32c, md5) |
- where crc32c and/or md5 can be _NA. |
- |
- Args: |
- line: The line to parse. |
- |
- Returns: |
- Parsed tuple: (url, size, crc32c, md5) |
- """ |
- (encoded_url, size, crc32c, md5) = line.split() |
- return (_DecodeUrl(encoded_url), int(size), crc32c, md5.strip()) |
- |
- def _WarnIfMissingCloudHash(self, url_str, crc32c, md5): |
- """Warns if given url_str is a cloud URL and is missing both crc32c and md5. |
- |
- Args: |
- url_str: Destination URL string. |
- crc32c: Destination CRC32c. |
- md5: Destination MD5. |
- |
- Returns: |
- True if issued warning. |
- """ |
- # One known way this can currently happen is when rsync'ing objects larger |
- # than 5 GB from S3 (for which the etag is not an MD5). |
- if (StorageUrlFromString(url_str).IsCloudUrl() |
- and crc32c == _NA and md5 == _NA): |
- self.logger.warn( |
- 'Found no hashes to validate %s. Integrity cannot be assured without ' |
- 'hashes.', url_str) |
- return True |
- return False |
- |
- def _ObjectsMatch(self, src_url_str, src_size, src_crc32c, src_md5, |
- dst_url_str, dst_size, dst_crc32c, dst_md5): |
- """Returns True if src and dst objects are the same. |
- |
- Uses size plus whatever checksums are available. |
- |
- Args: |
- src_url_str: Source URL string. |
- src_size: Source size |
- src_crc32c: Source CRC32c. |
- src_md5: Source MD5. |
- dst_url_str: Destination URL string. |
- dst_size: Destination size |
- dst_crc32c: Destination CRC32c. |
- dst_md5: Destination MD5. |
- |
- Returns: |
- True/False. |
- """ |
- # Note: This function is called from __iter__, which is called from the |
- # Command.Apply driver. Thus, all checksum computation will be run in a |
- # single thread, which is good (having multiple threads concurrently |
- # computing checksums would thrash the disk). |
- if src_size != dst_size: |
- return False |
- if self.compute_file_checksums: |
- (src_crc32c, src_md5, dst_crc32c, dst_md5) = _ComputeNeededFileChecksums( |
- self.logger, src_url_str, src_size, src_crc32c, src_md5, dst_url_str, |
- dst_size, dst_crc32c, dst_md5) |
- if src_md5 != _NA and dst_md5 != _NA: |
- self.logger.debug('Comparing md5 for %s and %s', src_url_str, dst_url_str) |
- return src_md5 == dst_md5 |
- if src_crc32c != _NA and dst_crc32c != _NA: |
- self.logger.debug( |
- 'Comparing crc32c for %s and %s', src_url_str, dst_url_str) |
- return src_crc32c == dst_crc32c |
- if not self._WarnIfMissingCloudHash(src_url_str, src_crc32c, src_md5): |
- self._WarnIfMissingCloudHash(dst_url_str, dst_crc32c, dst_md5) |
- # Without checksums to compare we depend only on basic size comparison. |
- return True |
- |
- def __iter__(self): |
- """Iterates over src/dst URLs and produces a _DiffToApply sequence. |
- |
- Yields: |
- The _DiffToApply. |
- """ |
- # Strip trailing slashes, if any, so we compute tail length against |
- # consistent position regardless of whether trailing slashes were included |
- # or not in URL. |
- base_src_url_len = len(self.base_src_url.url_string.rstrip('/\\')) |
- base_dst_url_len = len(self.base_dst_url.url_string.rstrip('/\\')) |
- src_url_str = dst_url_str = None |
- # Invariant: After each yield, the URLs in src_url_str, dst_url_str, |
- # self.sorted_src_urls_it, and self.sorted_dst_urls_it are not yet |
- # processed. Each time we encounter None in src_url_str or dst_url_str we |
- # populate from the respective iterator, and we reset one or the other value |
- # to None after yielding an action that disposes of that URL. |
- while not self.sorted_src_urls_it.IsEmpty() or src_url_str is not None: |
- if src_url_str is None: |
- (src_url_str, src_size, src_crc32c, src_md5) = self._ParseTmpFileLine( |
- self.sorted_src_urls_it.next()) |
- # Skip past base URL and normalize slashes so we can compare across |
- # clouds/file systems (including Windows). |
- src_url_str_to_check = _EncodeUrl( |
- src_url_str[base_src_url_len:].replace('\\', '/')) |
- dst_url_str_would_copy_to = copy_helper.ConstructDstUrl( |
- self.base_src_url, StorageUrlFromString(src_url_str), True, True, |
- self.base_dst_url, False, self.recursion_requested).url_string |
- if self.sorted_dst_urls_it.IsEmpty(): |
- # We've reached end of dst URLs, so copy src to dst. |
- yield _DiffToApply( |
- src_url_str, dst_url_str_would_copy_to, _DiffAction.COPY) |
- src_url_str = None |
- continue |
- if not dst_url_str: |
- (dst_url_str, dst_size, dst_crc32c, dst_md5) = ( |
- self._ParseTmpFileLine(self.sorted_dst_urls_it.next())) |
- # Skip past base URL and normalize slashes so we can compare acros |
- # clouds/file systems (including Windows). |
- dst_url_str_to_check = _EncodeUrl( |
- dst_url_str[base_dst_url_len:].replace('\\', '/')) |
- |
- if src_url_str_to_check < dst_url_str_to_check: |
- # There's no dst object corresponding to src object, so copy src to dst. |
- yield _DiffToApply( |
- src_url_str, dst_url_str_would_copy_to, _DiffAction.COPY) |
- src_url_str = None |
- elif src_url_str_to_check > dst_url_str_to_check: |
- # dst object without a corresponding src object, so remove dst if -d |
- # option was specified. |
- if self.delete_extras: |
- yield _DiffToApply(None, dst_url_str, _DiffAction.REMOVE) |
- dst_url_str = None |
- else: |
- # There is a dst object corresponding to src object, so check if objects |
- # match. |
- if self._ObjectsMatch( |
- src_url_str, src_size, src_crc32c, src_md5, |
- dst_url_str, dst_size, dst_crc32c, dst_md5): |
- # Continue iterating without yielding a _DiffToApply. |
- pass |
- else: |
- yield _DiffToApply(src_url_str, dst_url_str, _DiffAction.COPY) |
- src_url_str = None |
- dst_url_str = None |
- |
- # If -d option specified any files/objects left in dst iteration should be |
- # removed. |
- if not self.delete_extras: |
- return |
- if dst_url_str: |
- yield _DiffToApply(None, dst_url_str, _DiffAction.REMOVE) |
- dst_url_str = None |
- for line in self.sorted_dst_urls_it: |
- (dst_url_str, _, _, _) = self._ParseTmpFileLine(line) |
- yield _DiffToApply(None, dst_url_str, _DiffAction.REMOVE) |
- |
- |
-def _RsyncFunc(cls, diff_to_apply, thread_state=None): |
- """Worker function for performing the actual copy and remove operations.""" |
- gsutil_api = GetCloudApiInstance(cls, thread_state=thread_state) |
- dst_url_str = diff_to_apply.dst_url_str |
- dst_url = StorageUrlFromString(dst_url_str) |
- if diff_to_apply.diff_action == _DiffAction.REMOVE: |
- if cls.dryrun: |
- cls.logger.info('Would remove %s', dst_url) |
- else: |
- cls.logger.info('Removing %s', dst_url) |
- if dst_url.IsFileUrl(): |
- os.unlink(dst_url.object_name) |
- else: |
- try: |
- gsutil_api.DeleteObject( |
- dst_url.bucket_name, dst_url.object_name, |
- generation=dst_url.generation, provider=dst_url.scheme) |
- except NotFoundException: |
- # If the object happened to be deleted by an external process, this |
- # is fine because it moves us closer to the desired state. |
- pass |
- elif diff_to_apply.diff_action == _DiffAction.COPY: |
- src_url_str = diff_to_apply.src_url_str |
- src_url = StorageUrlFromString(src_url_str) |
- if cls.dryrun: |
- cls.logger.info('Would copy %s to %s', src_url, dst_url) |
- else: |
- try: |
- copy_helper.PerformCopy(cls.logger, src_url, dst_url, gsutil_api, cls, |
- _RsyncExceptionHandler, |
- headers=cls.headers) |
- except SkipUnsupportedObjectError, e: |
- cls.logger.info('Skipping item %s with unsupported object type %s', |
- src_url, e.unsupported_type) |
- |
- else: |
- raise CommandException('Got unexpected DiffAction (%d)' |
- % diff_to_apply.diff_action) |
- |
- |
-def _RootListingExceptionHandler(cls, e): |
- """Simple exception handler for exceptions during listing URLs to sync.""" |
- cls.logger.error(str(e)) |
- |
- |
-def _RsyncExceptionHandler(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 syncing:\n%s\n', |
- traceback.format_exc()) |
- |
- |
-class RsyncCommand(Command): |
- """Implementation of gsutil rsync command.""" |
- |
- # Command specification. See base class for documentation. |
- command_spec = Command.CreateCommandSpec( |
- 'rsync', |
- command_name_aliases=[], |
- usage_synopsis=_SYNOPSIS, |
- min_args=2, |
- max_args=2, |
- supported_sub_args='cCdenprRUx:', |
- file_url_ok=True, |
- provider_url_ok=False, |
- urls_start_arg=0, |
- gs_api_support=[ApiSelector.XML, ApiSelector.JSON], |
- gs_default_api=ApiSelector.JSON, |
- argparse_arguments=[ |
- CommandArgument.MakeNCloudOrFileURLsArgument(2) |
- ] |
- ) |
- # Help specification. See help_provider.py for documentation. |
- help_spec = Command.HelpSpec( |
- help_name='rsync', |
- help_name_aliases=['sync', 'synchronize'], |
- help_type='command_help', |
- help_one_line_summary='Synchronize content of two buckets/directories', |
- help_text=_DETAILED_HELP_TEXT, |
- subcommand_help_text={}, |
- ) |
- total_bytes_transferred = 0 |
- |
- def _InsistContainer(self, url_str, treat_nonexistent_object_as_subdir): |
- """Sanity checks that URL names an existing container. |
- |
- Args: |
- url_str: URL string to check. |
- treat_nonexistent_object_as_subdir: indicates if should treat a |
- non-existent object as a subdir. |
- |
- Returns: |
- URL for checked string. |
- |
- Raises: |
- CommandException if url_str doesn't name an existing container. |
- """ |
- (url, have_existing_container) = ( |
- copy_helper.ExpandUrlToSingleBlr(url_str, self.gsutil_api, self.debug, |
- self.project_id, |
- treat_nonexistent_object_as_subdir)) |
- if not have_existing_container: |
- raise CommandException( |
- 'arg (%s) does not name a directory, bucket, or bucket subdir.' |
- % url_str) |
- return url |
- |
- def RunCommand(self): |
- """Command entry point for the rsync command.""" |
- self._ParseOpts() |
- if self.compute_file_checksums and not UsingCrcmodExtension(crcmod): |
- self.logger.warn(SLOW_CRCMOD_WARNING) |
- |
- src_url = self._InsistContainer(self.args[0], False) |
- dst_url = self._InsistContainer(self.args[1], True) |
- |
- # Tracks if any copy or rm operations failed. |
- self.op_failure_count = 0 |
- |
- # List of attributes to share/manage across multiple processes in |
- # parallel (-m) mode. |
- shared_attrs = ['op_failure_count'] |
- |
- for signal_num in GetCaughtSignals(): |
- RegisterSignalHandler(signal_num, _HandleSignals) |
- |
- # Perform sync requests in parallel (-m) mode, if requested, using |
- # configured number of parallel processes and threads. Otherwise, |
- # perform requests with sequential function calls in current process. |
- diff_iterator = _DiffIterator(self, src_url, dst_url) |
- self.logger.info('Starting synchronization') |
- try: |
- self.Apply(_RsyncFunc, diff_iterator, _RsyncExceptionHandler, |
- shared_attrs, arg_checker=_DiffToApplyArgChecker, |
- fail_on_error=True) |
- finally: |
- CleanUpTempFiles() |
- |
- if self.op_failure_count: |
- plural_str = 's' if self.op_failure_count else '' |
- raise CommandException( |
- '%d file%s/object%s could not be copied/removed.' % |
- (self.op_failure_count, plural_str, plural_str)) |
- |
- def _ParseOpts(self): |
- # exclude_symlinks is handled by Command parent class, so save in Command |
- # state rather than CopyHelperOpts. |
- self.exclude_symlinks = False |
- # continue_on_error is handled by Command parent class, so save in Command |
- # state rather than CopyHelperOpts. |
- self.continue_on_error = False |
- self.delete_extras = False |
- preserve_acl = False |
- self.compute_file_checksums = False |
- self.dryrun = False |
- self.exclude_pattern = None |
- self.skip_unsupported_objects = False |
- # self.recursion_requested is initialized in command.py (so it can be |
- # checked in parent class for all commands). |
- |
- if self.sub_opts: |
- for o, a in self.sub_opts: |
- if o == '-c': |
- self.compute_file_checksums = True |
- # Note: In gsutil cp command this is specified using -c but here we use |
- # -C so we can use -c for checksum arg (to be consistent with Unix rsync |
- # command options). |
- elif o == '-C': |
- self.continue_on_error = True |
- elif o == '-d': |
- self.delete_extras = True |
- elif o == '-e': |
- self.exclude_symlinks = True |
- elif o == '-n': |
- self.dryrun = True |
- elif o == '-p': |
- preserve_acl = True |
- elif o == '-r' or o == '-R': |
- self.recursion_requested = True |
- elif o == '-U': |
- self.skip_unsupported_objects = True |
- elif o == '-x': |
- if not a: |
- raise CommandException('Invalid blank exclude filter') |
- try: |
- self.exclude_pattern = re.compile(a) |
- except re.error: |
- raise CommandException('Invalid exclude filter (%s)' % a) |
- return CreateCopyHelperOpts( |
- preserve_acl=preserve_acl, |
- skip_unsupported_objects=self.skip_unsupported_objects) |