| # Copyright 2010 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. |
| |
| """Base class for gsutil commands. |
| |
| In addition to base class code, this file contains helpers that depend on base |
| class state (such as GetAclCommandHelper, which depends on self.gsutil_bin_dir, |
| self.bucket_storage_uri_class, etc.) In general, functions that depend on class |
| state and that are used by multiple commands belong in this file. Functions that |
| don't depend on class state belong in util.py, and non-shared helpers belong in |
| individual subclasses. |
| """ |
| |
| import boto |
| import getopt |
| import gslib |
| import logging |
| import multiprocessing |
| import os |
| import platform |
| import re |
| import sys |
| import wildcard_iterator |
| import xml.dom.minidom |
| |
| from boto import handler |
| from boto.storage_uri import StorageUri |
| from getopt import GetoptError |
| from gslib import util |
| from gslib.exception import CommandException |
| from gslib.help_provider import HelpProvider |
| from gslib.name_expansion import NameExpansionIterator |
| from gslib.name_expansion import NameExpansionIteratorQueue |
| from gslib.project_id import ProjectIdHandler |
| from gslib.storage_uri_builder import StorageUriBuilder |
| from gslib.thread_pool import ThreadPool |
| from gslib.util import HAVE_OAUTH2 |
| from gslib.util import NO_MAX |
| |
| from gslib.wildcard_iterator import ContainsWildcard |
| |
| |
| def _ThreadedLogger(): |
| """Creates a logger that resembles 'print' output, but is thread safe. |
| |
| The logger will display all messages logged with level INFO or above. Log |
| propagation is disabled. |
| |
| Returns: |
| A logger object. |
| """ |
| log = logging.getLogger('threaded-logging') |
| log.propagate = False |
| log.setLevel(logging.INFO) |
| log_handler = logging.StreamHandler() |
| log_handler.setFormatter(logging.Formatter('%(message)s')) |
| log.addHandler(log_handler) |
| return log |
| |
| # command_spec key constants. |
| COMMAND_NAME = 'command_name' |
| COMMAND_NAME_ALIASES = 'command_name_aliases' |
| MIN_ARGS = 'min_args' |
| MAX_ARGS = 'max_args' |
| SUPPORTED_SUB_ARGS = 'supported_sub_args' |
| FILE_URIS_OK = 'file_uri_ok' |
| PROVIDER_URIS_OK = 'provider_uri_ok' |
| URIS_START_ARG = 'uris_start_arg' |
| CONFIG_REQUIRED = 'config_required' |
| |
| _EOF_NAME_EXPANSION_RESULT = ("EOF") |
| |
| |
| class Command(object): |
| # Global instance of a threaded logger object. |
| THREADED_LOGGER = _ThreadedLogger() |
| |
| REQUIRED_SPEC_KEYS = [COMMAND_NAME] |
| |
| # Each subclass must define the following map, minimally including the |
| # keys in REQUIRED_SPEC_KEYS; other values below will be used as defaults, |
| # although for readbility subclasses should specify the complete map. |
| command_spec = { |
| # Name of command. |
| COMMAND_NAME : None, |
| # List of command name aliases. |
| COMMAND_NAME_ALIASES : [], |
| # Min number of args required by this command. |
| MIN_ARGS : 0, |
| # Max number of args required by this command, or NO_MAX. |
| MAX_ARGS : NO_MAX, |
| # Getopt-style string specifying acceptable sub args. |
| SUPPORTED_SUB_ARGS : '', |
| # True if file URIs are acceptable for this command. |
| FILE_URIS_OK : False, |
| # True if provider-only URIs are acceptable for this command. |
| PROVIDER_URIS_OK : False, |
| # Index in args of first URI arg. |
| URIS_START_ARG : 0, |
| # True if must configure gsutil before running command. |
| CONFIG_REQUIRED : True, |
| } |
| _default_command_spec = command_spec |
| help_spec = HelpProvider.help_spec |
| |
| """Define an empty test specification, which derived classes must populate. |
| |
| This is a list of tuples containing the following values: |
| |
| step_name - mnemonic name for test, displayed when test is run |
| cmd_line - shell command line to run test |
| expect_ret or None - expected return code from test (None means ignore) |
| (result_file, expect_file) or None - tuple of result file and expected |
| file to diff for additional test |
| verification beyond the return code |
| (None means no diff requested) |
| Notes: |
| |
| - Setting expected_ret to None means there is no expectation and, |
| hence, any returned value will pass. |
| |
| - Any occurrences of the string 'gsutil' in the cmd_line parameter |
| are expanded to the full path to the gsutil command under test. |
| |
| - The cmd_line, result_file and expect_file parameters may |
| contain the following special substrings: |
| |
| $Bn - converted to one of 10 unique-for-testing bucket names (n=0..9) |
| $On - converted to one of 10 unique-for-testing object names (n=0..9) |
| $Fn - converted to one of 10 unique-for-testing file names (n=0..9) |
| $G - converted to the directory where gsutil is installed. Useful for |
| referencing test data. |
| |
| - The generated file names are full pathnames, whereas the generated |
| bucket and object names are simple relative names. |
| |
| - Tests with a non-None result_file and expect_file automatically |
| trigger an implicit diff of the two files. |
| |
| - These test specifications, in combination with the conversion strings |
| allow tests to be constructed parametrically. For example, here's an |
| annotated subset of a test_steps for the cp command: |
| |
| # Copy local file to object, verify 0 return code. |
| ('simple cp', 'gsutil cp $F1 gs://$B1/$O1', 0, None, None), |
| # Copy uploaded object back to local file and diff vs. orig file. |
| ('verify cp', 'gsutil cp gs://$B1/$O1 $F2', 0, '$F2', '$F1'), |
| |
| - After pattern substitution, the specs are run sequentially, in the |
| order in which they appear in the test_steps list. |
| """ |
| test_steps = [] |
| |
| # Define a convenience property for command name, since it's used many places. |
| def _GetDefaultCommandName(self): |
| return self.command_spec[COMMAND_NAME] |
| command_name = property(_GetDefaultCommandName) |
| |
| def __init__(self, command_runner, args, headers, debug, parallel_operations, |
| gsutil_bin_dir, boto_lib_dir, config_file_list, gsutil_ver, |
| bucket_storage_uri_class, test_method=None, |
| bypass_prodaccess=True): |
| """ |
| Args: |
| command_runner: CommandRunner (for commands built atop other commands). |
| args: Command-line args (arg0 = actual arg, not command name ala bash). |
| headers: Dictionary containing optional HTTP headers to pass to boto. |
| debug: Debug level to pass in to boto connection (range 0..3). |
| parallel_operations: Should command operations be executed in parallel? |
| gsutil_bin_dir: Bin dir from which gsutil is running. |
| boto_lib_dir: Lib dir where boto runs. |
| config_file_list: Config file list returned by _GetBotoConfigFileList(). |
| gsutil_ver: Version string of currently running gsutil command. |
| bucket_storage_uri_class: Class to instantiate for cloud StorageUris. |
| Settable for testing/mocking. |
| test_method: Optional general purpose method for testing purposes. |
| Application and semantics of this method will vary by |
| command and test type. |
| bypass_prodaccess: Boolean to ignore the existance of prodaccess. |
| |
| Implementation note: subclasses shouldn't need to define an __init__ |
| method, and instead depend on the shared initialization that happens |
| here. If you do define an __init__ method in a subclass you'll need to |
| explicitly call super().__init__(). But you're encouraged not to do this, |
| because it will make changing the __init__ interface more painful. |
| """ |
| # Save class values from constructor params. |
| self.command_runner = command_runner |
| self.args = args |
| self.unparsed_args = args |
| self.headers = headers |
| self.debug = debug |
| self.parallel_operations = parallel_operations |
| self.gsutil_bin_dir = gsutil_bin_dir |
| self.boto_lib_dir = boto_lib_dir |
| self.config_file_list = config_file_list |
| self.gsutil_ver = gsutil_ver |
| self.bucket_storage_uri_class = bucket_storage_uri_class |
| self.test_method = test_method |
| self.exclude_symlinks = False |
| self.recursion_requested = False |
| self.all_versions = False |
| self.bypass_prodaccess = bypass_prodaccess |
| |
| # Process sub-command instance specifications. |
| # First, ensure subclass implementation sets all required keys. |
| for k in self.REQUIRED_SPEC_KEYS: |
| if k not in self.command_spec or self.command_spec[k] is None: |
| raise CommandException('"%s" command implementation is missing %s ' |
| 'specification' % (self.command_name, k)) |
| # Now override default command_spec with subclass-specified values. |
| tmp = self._default_command_spec |
| tmp.update(self.command_spec) |
| self.command_spec = tmp |
| del tmp |
| |
| # Make sure command provides a test specification. |
| if not self.test_steps: |
| # TODO: Uncomment following lines when test feature is ready. |
| #raise CommandException('"%s" command implementation is missing test ' |
| #'specification' % self.command_name) |
| pass |
| |
| # Parse and validate args. |
| try: |
| (self.sub_opts, self.args) = getopt.getopt( |
| args, self.command_spec[SUPPORTED_SUB_ARGS]) |
| except GetoptError, e: |
| raise CommandException('%s for "%s" command.' % (e.msg, |
| self.command_name)) |
| if (len(self.args) < self.command_spec[MIN_ARGS] |
| or len(self.args) > self.command_spec[MAX_ARGS]): |
| raise CommandException('Wrong number of arguments for "%s" command.' % |
| self.command_name) |
| if (not self.command_spec[FILE_URIS_OK] |
| and self.HaveFileUris(self.args[self.command_spec[URIS_START_ARG]:])): |
| raise CommandException('"%s" command does not support "file://" URIs. ' |
| 'Did you mean to use a gs:// URI?' % |
| self.command_name) |
| if (not self.command_spec[PROVIDER_URIS_OK] |
| and self._HaveProviderUris( |
| self.args[self.command_spec[URIS_START_ARG]:])): |
| raise CommandException('"%s" command does not support provider-only ' |
| 'URIs.' % self.command_name) |
| if self.command_spec[CONFIG_REQUIRED]: |
| self._ConfigureNoOpAuthIfNeeded() |
| |
| self.proj_id_handler = ProjectIdHandler() |
| self.suri_builder = StorageUriBuilder(debug, bucket_storage_uri_class) |
| |
| # Cross-platform path to run gsutil binary. |
| self.gsutil_cmd = '' |
| # Cross-platform list containing gsutil path for use with subprocess. |
| self.gsutil_exec_list = [] |
| # If running on Windows, invoke python interpreter explicitly. |
| if platform.system() == "Windows": |
| self.gsutil_cmd += 'python ' |
| self.gsutil_exec_list += ['python'] |
| # Add full path to gsutil to make sure we test the correct version. |
| self.gsutil_path = os.path.join(self.gsutil_bin_dir, 'gsutil') |
| self.gsutil_cmd += self.gsutil_path |
| self.gsutil_exec_list += [self.gsutil_path] |
| |
| # We're treating recursion_requested like it's used by all commands, but |
| # only some of the commands accept the -R option. |
| if self.sub_opts: |
| for o, unused_a in self.sub_opts: |
| if o == '-r' or o == '-R': |
| self.recursion_requested = True |
| break |
| |
| def WildcardIterator(self, uri_or_str, all_versions=False): |
| """ |
| Helper to instantiate gslib.WildcardIterator. Args are same as |
| gslib.WildcardIterator interface, but this method fills in most of the |
| values from instance state. |
| |
| Args: |
| uri_or_str: StorageUri or URI string naming wildcard objects to iterate. |
| """ |
| return wildcard_iterator.wildcard_iterator( |
| uri_or_str, self.proj_id_handler, |
| bucket_storage_uri_class=self.bucket_storage_uri_class, |
| all_versions=all_versions, |
| headers=self.headers, debug=self.debug) |
| |
| def RunCommand(self): |
| """Abstract function in base class. Subclasses must implement this. The |
| return value of this function will be used as the exit status of the |
| process, so subclass commands should return an integer exit code (0 for |
| success, a value in [1,255] for failure). |
| """ |
| raise CommandException('Command %s is missing its RunCommand() ' |
| 'implementation' % self.command_name) |
| |
| ############################################################ |
| # Shared helper functions that depend on base class state. # |
| ############################################################ |
| |
| def UrisAreForSingleProvider(self, uri_args): |
| """Tests whether the uris are all for a single provider. |
| |
| Returns: a StorageUri for one of the uris on success, None on failure. |
| """ |
| provider = None |
| uri = None |
| for uri_str in uri_args: |
| # validate=False because we allow wildcard uris. |
| uri = boto.storage_uri( |
| uri_str, debug=self.debug, validate=False, |
| bucket_storage_uri_class=self.bucket_storage_uri_class) |
| if not provider: |
| provider = uri.scheme |
| elif uri.scheme != provider: |
| return None |
| return uri |
| |
| def SetAclCommandHelper(self): |
| """ |
| Common logic for setting ACLs. Sets the standard ACL or the default |
| object ACL depending on self.command_name. |
| """ |
| |
| acl_arg = self.args[0] |
| uri_args = self.args[1:] |
| # Disallow multi-provider setacl requests, because there are differences in |
| # the ACL models. |
| storage_uri = self.UrisAreForSingleProvider(uri_args) |
| if not storage_uri: |
| raise CommandException('"%s" command spanning providers not allowed.' % |
| self.command_name) |
| |
| # Determine whether acl_arg names a file containing XML ACL text vs. the |
| # string name of a canned ACL. |
| if os.path.isfile(acl_arg): |
| acl_file = open(acl_arg, 'r') |
| acl_arg = acl_file.read() |
| |
| # TODO: Remove this workaround when GCS allows |
| # whitespace in the Permission element on the server-side |
| acl_arg = re.sub(r'<Permission>\s*(\S+)\s*</Permission>', |
| r'<Permission>\1</Permission>', acl_arg) |
| |
| acl_file.close() |
| self.canned = False |
| else: |
| # No file exists, so expect a canned ACL string. |
| canned_acls = storage_uri.canned_acls() |
| if acl_arg not in canned_acls: |
| raise CommandException('Invalid canned ACL "%s".' % acl_arg) |
| self.canned = True |
| |
| # Used to track if any ACLs failed to be set. |
| self.everything_set_okay = True |
| |
| def _SetAclExceptionHandler(e): |
| """Simple exception handler to allow post-completion status.""" |
| self.THREADED_LOGGER.error(str(e)) |
| self.everything_set_okay = False |
| |
| def _SetAclFunc(name_expansion_result): |
| exp_src_uri = self.suri_builder.StorageUri( |
| name_expansion_result.GetExpandedUriStr()) |
| # We don't do bucket operations multi-threaded (see comment below). |
| assert self.command_name != 'setdefacl' |
| self.THREADED_LOGGER.info('Setting ACL on %s...' % |
| name_expansion_result.expanded_uri_str) |
| if self.canned: |
| exp_src_uri.set_acl(acl_arg, exp_src_uri.object_name, False, |
| self.headers) |
| else: |
| exp_src_uri.set_xml_acl(acl_arg, exp_src_uri.object_name, False, |
| self.headers) |
| |
| # If user specified -R option, convert any bucket args to bucket wildcards |
| # (e.g., gs://bucket/*), to prevent the operation from being applied to |
| # the buckets themselves. |
| if self.recursion_requested: |
| for i in range(len(uri_args)): |
| uri = self.suri_builder.StorageUri(uri_args[i]) |
| if uri.names_bucket(): |
| uri_args[i] = uri.clone_replace_name('*').uri |
| else: |
| # Handle bucket ACL setting operations single-threaded, because |
| # our threading machinery currently assumes it's working with objects |
| # (name_expansion_iterator), and normally we wouldn't expect users to need |
| # to set ACLs on huge numbers of buckets at once anyway. |
| for i in range(len(uri_args)): |
| uri_str = uri_args[i] |
| if self.suri_builder.StorageUri(uri_str).names_bucket(): |
| self._RunSingleThreadedSetAcl(acl_arg, uri_args) |
| return |
| |
| name_expansion_iterator = NameExpansionIterator( |
| self.command_name, self.proj_id_handler, self.headers, self.debug, |
| self.bucket_storage_uri_class, uri_args, self.recursion_requested, |
| self.recursion_requested, all_versions=self.all_versions) |
| |
| # Perform 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. |
| self.Apply(_SetAclFunc, name_expansion_iterator, _SetAclExceptionHandler) |
| |
| if not self.everything_set_okay: |
| raise CommandException('ACLs for some objects could not be set.') |
| |
| def _RunSingleThreadedSetAcl(self, acl_arg, uri_args): |
| some_matched = False |
| for uri_str in uri_args: |
| for blr in self.WildcardIterator(uri_str): |
| if blr.HasPrefix(): |
| continue |
| some_matched = True |
| uri = blr.GetUri() |
| if self.command_name == 'setdefacl': |
| print 'Setting default object ACL on %s...' % uri |
| if self.canned: |
| uri.set_def_acl(acl_arg, uri.object_name, False, self.headers) |
| else: |
| uri.set_def_xml_acl(acl_arg, False, self.headers) |
| else: |
| print 'Setting ACL on %s...' % uri |
| if self.canned: |
| uri.set_acl(acl_arg, uri.object_name, False, self.headers) |
| else: |
| uri.set_xml_acl(acl_arg, uri.object_name, False, self.headers) |
| if not some_matched: |
| raise CommandException('No URIs matched') |
| |
| def GetAclCommandHelper(self): |
| """Common logic for getting ACLs. Gets the standard ACL or the default |
| object ACL depending on self.command_name.""" |
| |
| # Resolve to just one object. |
| # Handle wildcard-less URI specially in case this is a version-specific |
| # URI, because WildcardIterator().IterUris() would lose the versioning info. |
| if not ContainsWildcard(self.args[0]): |
| uri = self.suri_builder.StorageUri(self.args[0]) |
| else: |
| uris = list(self.WildcardIterator(self.args[0]).IterUris()) |
| if len(uris) == 0: |
| raise CommandException('No URIs matched') |
| if len(uris) != 1: |
| raise CommandException('%s matched more than one URI, which is not ' |
| 'allowed by the %s command' % (self.args[0], self.command_name)) |
| uri = uris[0] |
| if not uri.names_bucket() and not uri.names_object(): |
| raise CommandException('"%s" command must specify a bucket or ' |
| 'object.' % self.command_name) |
| if self.command_name == 'getdefacl': |
| acl = uri.get_def_acl(False, self.headers) |
| else: |
| acl = uri.get_acl(False, self.headers) |
| # Pretty-print the XML to make it more easily human editable. |
| parsed_xml = xml.dom.minidom.parseString(acl.to_xml().encode('utf-8')) |
| print parsed_xml.toprettyxml(indent=' ') |
| |
| def GetXmlSubresource(self, subresource, uri_arg): |
| """Print an xml subresource, e.g. logging, for a bucket/object. |
| |
| Args: |
| subresource: The subresource name. |
| uri_arg: URI for the bucket/object. Wildcards will be expanded. |
| |
| Raises: |
| CommandException: if errors encountered. |
| """ |
| # Wildcarding is allowed but must resolve to just one bucket. |
| uris = list(self.WildcardIterator(uri_arg).IterUris()) |
| if len(uris) != 1: |
| raise CommandException('Wildcards must resolve to exactly one item for ' |
| 'get %s' % subresource) |
| uri = uris[0] |
| xml_str = uri.get_subresource(subresource, False, self.headers) |
| # Pretty-print the XML to make it more easily human editable. |
| parsed_xml = xml.dom.minidom.parseString(xml_str.encode('utf-8')) |
| print parsed_xml.toprettyxml(indent=' ') |
| |
| def Apply(self, func, name_expansion_iterator, thr_exc_handler, |
| shared_attrs=None): |
| """Dispatch input URI assignments across a pool of parallel OS |
| processes and/or Python threads, based on options (-m or not) |
| and settings in the user's config file. If non-parallel mode |
| or only one OS process requested, execute requests sequentially |
| in the current OS process. |
| |
| Args: |
| func: Function to call to process each URI. |
| name_expansion_iterator: Iterator of NameExpansionResult. |
| thr_exc_handler: Exception handler for ThreadPool class. |
| shared_attrs: List of attributes to manage across sub-processes. |
| |
| Raises: |
| CommandException if invalid config encountered. |
| """ |
| |
| # Set OS process and python thread count as a function of options |
| # and config. |
| if self.parallel_operations: |
| process_count = boto.config.getint( |
| 'GSUtil', 'parallel_process_count', |
| gslib.commands.config.DEFAULT_PARALLEL_PROCESS_COUNT) |
| if process_count < 1: |
| raise CommandException('Invalid parallel_process_count "%d".' % |
| process_count) |
| thread_count = boto.config.getint( |
| 'GSUtil', 'parallel_thread_count', |
| gslib.commands.config.DEFAULT_PARALLEL_THREAD_COUNT) |
| if thread_count < 1: |
| raise CommandException('Invalid parallel_thread_count "%d".' % |
| thread_count) |
| else: |
| # If -m not specified, then assume 1 OS process and 1 Python thread. |
| process_count = 1 |
| thread_count = 1 |
| |
| if self.debug: |
| self.THREADED_LOGGER.info('process count: %d', process_count) |
| self.THREADED_LOGGER.info('thread count: %d', thread_count) |
| |
| if self.parallel_operations and process_count > 1: |
| procs = [] |
| # If any shared attributes passed by caller, create a dictionary of |
| # shared memory variables for every element in the list of shared |
| # attributes. |
| shared_vars = None |
| if shared_attrs: |
| for name in shared_attrs: |
| if not shared_vars: |
| shared_vars = {} |
| shared_vars[name] = multiprocessing.Value('i', 0) |
| # Construct work queue for parceling out work to multiprocessing workers, |
| # setting the max queue length of 50k so we will block if workers don't |
| # empty the queue as fast as we can continue iterating over the bucket |
| # listing. This number may need tuning; it should be large enough to |
| # keep workers busy (overlapping bucket list next-page retrieval with |
| # operations being fed from the queue) but small enough that we don't |
| # overfill memory when runing across a slow network link. |
| work_queue = multiprocessing.Queue(50000) |
| for shard in range(process_count): |
| # Spawn a separate OS process for each shard. |
| if self.debug: |
| self.THREADED_LOGGER.info('spawning process for shard %d', shard) |
| p = multiprocessing.Process(target=self._ApplyThreads, |
| args=(func, work_queue, shard, |
| thread_count, thr_exc_handler, |
| shared_vars)) |
| procs.append(p) |
| p.start() |
| |
| last_name_expansion_result = None |
| try: |
| # Feed all work into the queue being emptied by the workers. |
| for name_expansion_result in name_expansion_iterator: |
| last_name_expansion_result = name_expansion_result |
| work_queue.put(name_expansion_result) |
| except: |
| sys.stderr.write('Failed URI iteration. Last result (prior to ' |
| 'exception) was: %s\n' |
| % repr(last_name_expansion_result)) |
| finally: |
| # We do all of the process cleanup in a finally cause in case the name |
| # expansion iterator throws an exception. This will send EOF to all the |
| # child processes and join them back into the parent process. |
| |
| # Send an EOF per worker. |
| for shard in range(process_count): |
| work_queue.put(_EOF_NAME_EXPANSION_RESULT) |
| |
| # Wait for all spawned OS processes to finish. |
| failed_process_count = 0 |
| for p in procs: |
| p.join() |
| # Count number of procs that returned non-zero exit code. |
| if p.exitcode != 0: |
| failed_process_count += 1 |
| |
| # Propagate shared variables back to caller's attributes. |
| if shared_vars: |
| for (name, var) in shared_vars.items(): |
| setattr(self, name, var.value) |
| |
| # Abort main process if one or more sub-processes failed. Note that this |
| # is outside the finally clause, because we only want to raise a new |
| # exception if an exception wasn't already raised in the try clause above. |
| if failed_process_count: |
| plural_str = '' |
| if failed_process_count > 1: |
| plural_str = 'es' |
| raise Exception('unexpected failure in %d sub-process%s, ' |
| 'aborting...' % (failed_process_count, plural_str)) |
| |
| else: |
| # Using just 1 process, so funnel results to _ApplyThreads using facade |
| # that makes NameExpansionIterator look like a Multiprocessing.Queue |
| # that sends one EOF once the iterator empties. |
| work_queue = NameExpansionIteratorQueue(name_expansion_iterator, |
| _EOF_NAME_EXPANSION_RESULT) |
| self._ApplyThreads(func, work_queue, 0, thread_count, thr_exc_handler, |
| None) |
| |
| def HaveFileUris(self, args_to_check): |
| """Checks whether args_to_check contain any file URIs. |
| |
| Args: |
| args_to_check: Command-line argument subset to check. |
| |
| Returns: |
| True if args_to_check contains any file URIs. |
| """ |
| for uri_str in args_to_check: |
| if uri_str.lower().startswith('file://') or uri_str.find(':') == -1: |
| return True |
| return False |
| |
| ###################### |
| # Private functions. # |
| ###################### |
| |
| def _HaveProviderUris(self, args_to_check): |
| """Checks whether args_to_check contains any provider URIs (like 'gs://'). |
| |
| Args: |
| args_to_check: Command-line argument subset to check. |
| |
| Returns: |
| True if args_to_check contains any provider URIs. |
| """ |
| for uri_str in args_to_check: |
| if re.match('^[a-z]+://$', uri_str): |
| return True |
| return False |
| |
| def _ConfigureNoOpAuthIfNeeded(self): |
| """Sets up no-op auth handler if no boto credentials are configured.""" |
| config = boto.config |
| if not util.HasConfiguredCredentials(self.bypass_prodaccess): |
| if self.config_file_list: |
| if (config.has_option('Credentials', 'gs_oauth2_refresh_token') |
| and not HAVE_OAUTH2): |
| raise CommandException( |
| 'Your gsutil is configured with OAuth2 authentication ' |
| 'credentials.\nHowever, OAuth2 is only supported when running ' |
| 'under Python 2.6 or later\n(unless additional dependencies are ' |
| 'installed, see README for details); you are running Python %s.' % |
| sys.version) |
| raise CommandException('You have no storage service credentials in any ' |
| 'of the following boto config\nfiles. Please ' |
| 'add your credentials as described in the ' |
| 'gsutil README file, or else\nre-run ' |
| '"gsutil config" to re-create a config ' |
| 'file:\n%s' % self.config_file_list) |
| else: |
| # With no boto config file the user can still access publicly readable |
| # buckets and objects. |
| from gslib import no_op_auth_plugin |
| |
| def _ApplyThreads(self, func, work_queue, shard, num_threads, |
| thr_exc_handler=None, shared_vars=None): |
| """ |
| Perform subset of required requests across a caller specified |
| number of parallel Python threads, which may be one, in which |
| case the requests are processed in the current thread. |
| |
| Args: |
| func: Function to call for each request. |
| work_queue: shared queue of NameExpansionResult to process. |
| shard: Assigned subset (shard number) for this function. |
| num_threads: Number of Python threads to spawn to process this shard. |
| thr_exc_handler: Exception handler for ThreadPool class. |
| shared_vars: Dict of shared memory variables to be managed. |
| (only relevant, and non-None, if this function is |
| run in a separate OS process). |
| """ |
| # Each OS process needs to establish its own set of connections to |
| # the server to avoid writes from different OS processes interleaving |
| # onto the same socket (and garbling the underlying SSL session). |
| # We ensure each process gets its own set of connections here by |
| # closing all connections in the storage provider connection pool. |
| connection_pool = StorageUri.provider_pool |
| if connection_pool: |
| for i in connection_pool: |
| connection_pool[i].connection.close() |
| |
| if num_threads > 1: |
| thread_pool = ThreadPool(num_threads, thr_exc_handler) |
| try: |
| while True: # Loop until we hit EOF marker. |
| name_expansion_result = work_queue.get() |
| if name_expansion_result == _EOF_NAME_EXPANSION_RESULT: |
| break |
| exp_src_uri = self.suri_builder.StorageUri( |
| name_expansion_result.GetExpandedUriStr()) |
| if self.debug: |
| self.THREADED_LOGGER.info('process %d shard %d is handling uri %s', |
| os.getpid(), shard, exp_src_uri) |
| if (self.exclude_symlinks and exp_src_uri.is_file_uri() |
| and os.path.islink(exp_src_uri.object_name)): |
| self.THREADED_LOGGER.info('Skipping symbolic link %s...', exp_src_uri) |
| elif num_threads > 1: |
| thread_pool.AddTask(func, name_expansion_result) |
| else: |
| func(name_expansion_result) |
| # If any Python threads created, wait here for them to finish. |
| if num_threads > 1: |
| thread_pool.WaitCompletion() |
| finally: |
| if num_threads > 1: |
| thread_pool.Shutdown() |
| # If any shared variables (which means we are running in a separate OS |
| # process), increment value for each shared variable. |
| if shared_vars: |
| for (name, var) in shared_vars.items(): |
| var.value += getattr(self, name) |
