py_tasks_melange: comparison thirdparty/google_appengine/google/appengine/tools/appcfg.py

equal deleted inserted replaced

-:27971a13089f
+:2e0b0af889be
 import getpass
 import logging
 import mimetypes
 import optparse
 import os
+import random
 import re
 import sha
 import sys
 import tempfile
 import time
+import urllib
 import urllib2
 import google
 import yaml
 from google.appengine.cron import groctimespecification
 UPDATE_CHECK_TIMEOUT = 3
 NAG_FILE = '.appcfg_nag'
 MAX_LOG_LEVEL = 4
+MAX_BATCH_SIZE = 1000000
+MAX_BATCH_COUNT = 100
+MAX_BATCH_FILE_SIZE = 200000
+BATCH_OVERHEAD = 500
 verbosity = 1
 appinfo.AppInfoExternal.ATTRIBUTES[appinfo.RUNTIME] = 'python'
 delay *= backoff_factor
 max_tries -= 1
 return max_tries > 0
+def _VersionList(release):
+"""Parse a version string into a list of ints.
+Args:
+release: The 'release' version, e.g. '1.2.4'.
+(Due to YAML parsing this may also be an int or float.)
+Returns:
+A list of ints corresponding to the parts of the version string
+between periods.  Example:
+'1.2.4' -> [1, 2, 4]
+'1.2.3.4' -> [1, 2, 3, 4]
+Raises:
+ValueError if not all the parts are valid integers.
+"""
+return [int(part) for part in str(release).split('.')]
 class UpdateCheck(object):
 """Determines if the local SDK is the latest version.
 Nags the user when there are updates to the SDK.  As the SDK becomes
 more out of date, the language in the nagging gets stronger.  We
 except urllib2.URLError, e:
 logging.info('Update check failed: %s', e)
 return
 latest = yaml.safe_load(response)
-if latest['release'] == version['release']:
+if version['release'] == latest['release']:
 logging.info('The SDK is up to date.')
 return
+try:
+this_release = _VersionList(version['release'])
+except ValueError:
+logging.warn('Could not parse this release version (%r)',
+version['release'])
+else:
+try:
+advertised_release = _VersionList(latest['release'])
+except ValueError:
+logging.warn('Could not parse advertised release version (%r)',
+latest['release'])
+else:
+if this_release > advertised_release:
+logging.info('This SDK release is newer than the advertised release.')
+return
 api_versions = latest['api_versions']
 if self.config.api_version not in api_versions:
 self._Nag(
 'The api version you are using (%s) is obsolete!  You should\n'
 return sentinel.rstrip('\n')
 finally:
 fp.close()
+class UploadBatcher(object):
+"""Helper to batch file uploads."""
+def __init__(self, what, app_id, version, server):
+"""Constructor.
+Args:
+what: Either 'file' or 'blob' indicating what kind of objects
+this batcher uploads.  Used in messages and URLs.
+app_id: The application ID.
+version: The application version string.
+server: The RPC server.
+"""
+assert what in ('file', 'blob'), repr(what)
+self.what = what
+self.app_id = app_id
+self.version = version
+self.server = server
+self.single_url = '/api/appversion/add' + what
+self.batch_url = self.single_url + 's'
+self.batching = True
+self.batch = []
+self.batch_size = 0
+def SendBatch(self):
+"""Send the current batch on its way.
+If successful, resets self.batch and self.batch_size.
+Raises:
+HTTPError with code=404 if the server doesn't support batching.
+"""
+boundary = 'boundary'
+parts = []
+for path, payload, mime_type in self.batch:
+while boundary in payload:
+boundary += '%04x' % random.randint(0, 0xffff)
+assert len(boundary) < 80, 'Unexpected error, please try again.'
+part = '\n'.join(['',
+'X-Appcfg-File: %s' % urllib.quote(path),
+'X-Appcfg-Hash: %s' % _Hash(payload),
+'Content-Type: %s' % mime_type,
+'Content-Length: %d' % len(payload),
+'Content-Transfer-Encoding: 8bit',
+'',
+payload,
+])
+parts.append(part)
+parts.insert(0,
+'MIME-Version: 1.0\n'
+'Content-Type: multipart/mixed; boundary="%s"\n'
+'\n'
+'This is a message with multiple parts in MIME format.' %
+boundary)
+parts.append('--\n')
+delimiter = '\n--%s' % boundary
+payload = delimiter.join(parts)
+logging.info('Uploading batch of %d %ss to %s with boundary="%s".',
+len(self.batch), self.what, self.batch_url, boundary)
+self.server.Send(self.batch_url,
+payload=payload,
+content_type='message/rfc822',
+app_id=self.app_id,
+version=self.version)
+self.batch = []
+self.batch_size = 0
+def SendSingleFile(self, path, payload, mime_type):
+"""Send a single file on its way."""
+logging.info('Uploading %s %s (%s bytes, type=%s) to %s.',
+self.what, path, len(payload), mime_type, self.single_url)
+self.server.Send(self.single_url,
+payload=payload,
+content_type=mime_type,
+path=path,
+app_id=self.app_id,
+version=self.version)
+def Flush(self):
+"""Flush the current batch.
+This first attempts to send the batch as a single request; if that
+fails because the server doesn't support batching, the files are
+sent one by one, and self.batching is reset to False.
+At the end, self.batch and self.batch_size are reset.
+"""
+if not self.batch:
+return
+try:
+self.SendBatch()
+except urllib2.HTTPError, err:
+if err.code != 404:
+raise
+logging.info('Old server detected; turning off %s batching.', self.what)
+self.batching = False
+for path, payload, mime_type in self.batch:
+self.SendSingleFile(path, payload, mime_type)
+self.batch = []
+self.batch_size = 0
+def AddToBatch(self, path, payload, mime_type):
+"""Batch a file, possibly flushing first, or perhaps upload it directly.
+Args:
+path: The name of the file.
+payload: The contents of the file.
+mime_type: The MIME Content-type of the file, or None.
+If mime_type is None, application/octet-stream is substituted.
+"""
+if not mime_type:
+mime_type = 'application/octet-stream'
+size = len(payload)
+if size <= MAX_BATCH_FILE_SIZE:
+if (len(self.batch) >= MAX_BATCH_COUNT or
+self.batch_size + size > MAX_BATCH_SIZE):
+self.Flush()
+if self.batching:
+logging.info('Adding %s %s (%s bytes, type=%s) to batch.',
+self.what, path, size, mime_type)
+self.batch.append((path, payload, mime_type))
+self.batch_size += size + BATCH_OVERHEAD
+return
+self.SendSingleFile(path, payload, mime_type)
+def _Hash(content):
+"""Compute the hash of the content.
+Args:
+content: The data to hash as a string.
+Returns:
+The string representation of the hash.
+"""
+h = sha.new(content).hexdigest()
+return '%s_%s_%s_%s_%s' % (h[0:8], h[8:16], h[16:24], h[24:32], h[32:40])
 class AppVersionUpload(object):
 """Provides facilities to upload a new appversion to the hosting service.
 Attributes:
 server: The AbstractRpcServer to use for the upload.
 self.app_id = self.config.application
 self.version = self.config.version
 self.files = {}
 self.in_transaction = False
 self.deployed = False
+self.batching = True
-def _Hash(self, content):
+self.file_batcher = UploadBatcher('file', self.app_id, self.version,
-"""Compute the hash of the content.
+self.server)
+self.blob_batcher = UploadBatcher('blob', self.app_id, self.version,
-Args:
+self.server)
-content: The data to hash as a string.
-Returns:
-The string representation of the hash.
-"""
-h = sha.new(content).hexdigest()
-return '%s_%s_%s_%s_%s' % (h[0:8], h[8:16], h[16:24], h[24:32], h[32:40])
 def AddFile(self, path, file_handle):
 """Adds the provided file to the list to be pushed to the server.
 Args:
 if reason:
 logging.error(reason)
 return
 pos = file_handle.tell()
-content_hash = self._Hash(file_handle.read())
+content_hash = _Hash(file_handle.read())
 file_handle.seek(pos, 0)
 self.files[path] = content_hash
 def Begin(self):
 (f, self.files[f]) for f in result.split(LIST_DELIMITER)))
 CloneFiles('/api/appversion/cloneblobs', blobs_to_clone, 'static')
 CloneFiles('/api/appversion/clonefiles', files_to_clone, 'application')
-logging.info('Files to upload: ' + str(files_to_upload))
+logging.debug('Files to upload: %s', files_to_upload)
 self.files = files_to_upload
 return sorted(files_to_upload.iterkeys())
 def UploadFile(self, path, file_handle):
 raise KeyError('File \'%s\' is not in the list of files to be uploaded.'
 % path)
 del self.files[path]
 mime_type = GetMimeTypeIfStaticFile(self.config, path)
-if mime_type is not None:
+payload = file_handle.read()
-self.server.Send('/api/appversion/addblob', app_id=self.app_id,
+if mime_type is None:
-version=self.version, path=path, content_type=mime_type,
+self.file_batcher.AddToBatch(path, payload, mime_type)
-payload=file_handle.read())
 else:
-self.server.Send('/api/appversion/addfile', app_id=self.app_id,
+self.blob_batcher.AddToBatch(path, payload, mime_type)
-version=self.version, path=path,
-payload=file_handle.read())
 def Commit(self):
 """Commits the transaction, making the new app version available.
 All the files returned by Begin() must have been uploaded with UploadFile()
 raise
 try:
 missing_files = self.Begin()
 if missing_files:
-StatusUpdate('Uploading %d files.' % len(missing_files))
+StatusUpdate('Uploading %d files and blobs.' % len(missing_files))
 num_files = 0
 for missing_file in missing_files:
-logging.info('Uploading file \'%s\'' % missing_file)
 file_handle = openfunc(missing_file)
 try:
 self.UploadFile(missing_file, file_handle)
 finally:
 file_handle.close()
 num_files += 1
 if num_files % 500 == 0:
-StatusUpdate('Uploaded %d files.' % num_files)
+StatusUpdate('Processed %d out of %s.' %
+(num_files, len(missing_files)))
+self.file_batcher.Flush()
+self.blob_batcher.Flush()
+StatusUpdate('Uploaded %d files and blobs' % num_files)
 self.Commit()
 except KeyboardInterrupt:
 logging.info('User interrupted. Aborting.')
+self.Rollback()
+raise
+except urllib2.HTTPError, err:
+logging.info('HTTP Error (%s)', err)
 self.Rollback()
 raise
 except:
 logging.exception('An unexpected error occurred. Aborting.')
 self.Rollback()
 self.parser.error(
 'Severity range is 0 (DEBUG) through %s (CRITICAL).' % MAX_LOG_LEVEL)
 if self.options.num_days is None:
 self.options.num_days = int(not self.options.append)
+try:
+end_date = self._ParseEndDate(self.options.end_date)
+except ValueError:
+self.parser.error('End date must be in the format YYYY-MM-DD.')
 basepath = self.args[0]
 appyaml = self._ParseAppYaml(basepath)
 rpc_server = self._GetRpcServer()
 logs_requester = LogsRequester(rpc_server, appyaml, self.args[1],
 self.options.num_days,
 self.options.append,
 self.options.severity,
-time.time(),
+end_date,
 self.options.vhost,
 self.options.include_vhost)
 logs_requester.DownloadLogs()
+def _ParseEndDate(self, date, time_func=time.time):
+"""Translates a user-readable end date to a POSIX timestamp.
+Args:
+date: A utc date string as YYYY-MM-DD.
+time_func: time.time() function for testing.
+Returns:
+A POSIX timestamp representing the last moment of that day.
+If no date is given, returns a timestamp representing now.
+"""
+if not date:
+return time_func()
+struct_time = time.strptime('%s' % date, '%Y-%m-%d')
+return calendar.timegm(struct_time) + 86400
 def _RequestLogsOptions(self, parser):
 """Adds request_logs-specific options to 'parser'.
 Args:
 help='The virtual host of log messages to get. '
 'If omitted, all log messages are returned.')
 parser.add_option('--include_vhost', dest='include_vhost',
 action='store_true', default=False,
 help='Include virtual host in log messages.')
+parser.add_option('--end_date', dest='end_date',
+action='store', default='',
+help='End date (as YYYY-MM-DD) of period for log data. '
+'Defaults to today.')
 def CronInfo(self, now=None, output=sys.stdout):
 """Displays information about cron definitions.
 Args:
 'log_file',
 'passin',
 'email',
 'debug',
 'exporter_opts',
+'mapper_opts',
 'result_db_filename',
+'mapper_opts',
+'dry_run',
+'dump',
+'restore',
 )])
 def PerformDownload(self, run_fn=None):
 """Performs a datastore download via the bulkloader.
 StatusUpdate('Downloading data records.')
 args = self._MakeLoaderArgs()
 args['download'] = True
 args['has_header'] = False
+args['map'] = False
+args['dump'] = False
+args['restore'] = False
 run_fn(args)
 def PerformUpload(self, run_fn=None):
 """Performs a datastore upload via the bulkloader.
 StatusUpdate('Uploading data records.')
 args = self._MakeLoaderArgs()
 args['download'] = False
+args['map'] = False
+args['dump'] = False
+args['restore'] = False
 run_fn(args)
 def _PerformLoadOptions(self, parser):
 """Adds options common to 'upload_data' and 'download_data'.
 help='The name of the authorization domain to use.')
 parser.add_option('--log_file', type='string', dest='log_file',
 help='File to write bulkloader logs.  If not supplied '
 'then a new log file will be created, named: '
 'bulkloader-log-TIMESTAMP.')
+parser.add_option('--dry_run', action='store_true',
+dest='dry_run', default=False,
+help='Do not execute any remote_api calls')
 def _PerformUploadOptions(self, parser):
 """Adds 'upload_data' specific options to the 'parser' passed in.
 Args:

changeset 2864	2e0b0af889be
parent 2413	d0b7dac5325c
child 3031	7678f72140e6