Merge pull request #760 from whoarethebritons/dispatch-routing

Dispatch routing from yaml

appscale/tools/admin_api/client.py

@@ -321,3 +321,35 @@ def update_queues(self, project_id, queues):
       message = 'AdminServer returned: {}'.format(response.status_code)
 
     raise AdminError(message)
+
+  @retry(**RETRY_POLICY)
+  def update_dispatch(self, project_id, dispatch_rules):
+    """ Updates the the project's dispatch configuration.
+
+    Args:
+      project_id: A string specifying the project ID.
+      dispatch_rules: A dictionary containing dispatch configuration details.
+    Raises:
+      AdminError if unable to update dispatch configuration.
+    """
+    versions_url = ('{prefix}/{project}'
+                    .format(prefix=self.prefix, project=project_id))
+    headers = {
+      'AppScale-Secret': self.secret,
+      'Content-Type': 'application/json'
+    }
+    params = {
+      'updateMask': 'dispatchRules'
+    }
+    body = {'dispatchRules': dispatch_rules}
+
+    response = requests.patch(versions_url, headers=headers, params=params,
+                              json=body, verify=False)
+
+    operation = self.extract_response(response)
+    try:
+      operation_id = operation['name'].split('/')[-1]
+    except (KeyError, IndexError):
+      raise AdminError('Invalid operation: {}'.format(operation))
+
+    return operation_id

appscale/tools/appscale.py

@@ -8,7 +8,6 @@
 import shutil
 import subprocess
 import sys
-
 import yaml
 
 from appscale.tools.appengine_helper import AppEngineHelper
@@ -23,7 +22,7 @@
 from appscale.tools.parse_args import ParseArgs
 from appscale.tools.registration_helper import RegistrationHelper
 from appscale.tools.remote_helper import RemoteHelper
-
+from appscale.tools.admin_api.client import AdminError
 
 class AppScale():
   """ AppScale provides a configuration-file-based alternative to the
@@ -578,6 +577,13 @@ def deploy(self, app, project_id=None):
     AppScaleTools.update_indexes(options.file, options.keyname, options.project)
     AppScaleTools.update_cron(options.file, options.keyname, options.project)
     AppScaleTools.update_queues(options.file, options.keyname, options.project)
+    try:
+      AppScaleTools.update_dispatch(options.file, options.keyname,
+                                    options.project, options.verbose)
+    except (AdminError, AppScaleException) as e:
+      AppScaleLogger.warn('Request to update dispatch failed, if your '
+                          'dispatch references undeployed services, ignore '
+                          'this exception: {}'.format(e))
     return login_host, http_port
 
 

appscale/tools/appscale_tools.py

@@ -1075,6 +1075,70 @@ def upload_app(cls, options):
     http_port = int(match.group(2))
     return login_host, http_port
 
+
+  @classmethod
+  def update_dispatch(cls, source_location, keyname, project_id, is_verbose):
+    """ Updates an application's dispatch routing rules from the configuration
+      file.
+
+    Args:
+      options: A Namespace that has fields for each parameter that can be
+        passed in via the command-line interface.
+    """
+    if cls.TAR_GZ_REGEX.search(source_location):
+      fetch_function = utils.config_from_tar_gz
+      version = Version.from_tar_gz(source_location)
+    elif cls.ZIP_REGEX.search(source_location):
+      fetch_function = utils.config_from_zip
+      version = Version.from_zip(source_location)
+    elif os.path.isdir(source_location):
+      fetch_function = utils.config_from_dir
+      version = Version.from_directory(source_location)
+    elif source_location.endswith('.yaml'):
+      fetch_function = utils.config_from_dir
+      version = Version.from_yaml_file(source_location)
+      source_location = os.path.dirname(source_location)
+    else:
+      raise BadConfigurationException(
+        '{} must be a directory, tar.gz, or zip'.format(source_location))
+
+    if project_id:
+      version.project_id = project_id
+
+    dispatch_rules = utils.dispatch_from_yaml(source_location, fetch_function)
+    if dispatch_rules is None:
+        return
+    AppScaleLogger.log('Updating dispatch for {}'.format(version.project_id))
+
+    load_balancer_ip = LocalState.get_host_with_role(keyname, 'load_balancer')
+    secret_key = LocalState.get_secret_key(keyname)
+    admin_client = AdminClient(load_balancer_ip, secret_key)
+    operation_id = admin_client.update_dispatch(version.project_id, dispatch_rules)
+
+    # Check on the operation.
+    AppScaleLogger.log("Please wait for your dispatch to be updated.")
+
+    deadline = time.time() + cls.MAX_OPERATION_TIME
+    while True:
+      if time.time() > deadline:
+        raise AppScaleException('The operation took too long.')
+      operation = admin_client.get_operation(version.project_id, operation_id)
+      if not operation['done']:
+        time.sleep(1)
+        continue
+
+      if 'error' in operation:
+        raise AppScaleException(operation['error']['message'])
+      dispatch_rules = operation['response']['dispatchRules']
+      break
+
+    AppScaleLogger.verbose(
+        "The following dispatchRules have been applied to your application's "
+        "configuration : {}".format(dispatch_rules), is_verbose)
+    AppScaleLogger.success('Dispatch has been updated for {}'.format(
+        version.project_id))
+
+
   @classmethod
   def update_cron(cls, source_location, keyname, project_id):
     """ Updates a project's cron jobs from the configuration file.

appscale/tools/utils.py

@@ -2,13 +2,18 @@
 
 import errno
 import os
+import re
 import tarfile
 import yaml
 import zipfile
 from xml.etree import ElementTree
 
 from .custom_exceptions import BadConfigurationException
 
+# The regex used to group the dispatch url into 'domain' and 'path'.
+# taken from GAE's 1.9.69 SDK (google/appengine/api/dispatchinfo.py)
+_URL_SPLITTER_RE = re.compile(r'^([^/]+)(/.*)$')
+
 
 def shortest_path_from_list(file_name, name_list):
   """ Determines the shortest path to a file in a list of candidates.
@@ -272,6 +277,34 @@ def queues_from_xml(contents):
 
   return queues
 
+def dispatch_from_yaml(source_location, fetch_function):
+  dispatch_config = fetch_function('dispatch.yaml', source_location)
+  if dispatch_config is None:
+    return
+
+  dispatch_rules = yaml.safe_load(dispatch_config)
+
+  if not dispatch_rules or not dispatch_rules.get('dispatch'):
+    raise BadConfigurationException('Could not retrieve anything from '
+                                    'specified dispatch.yaml')
+  modified_rules = []
+  for dispatch_rule in dispatch_rules['dispatch']:
+    rule = {}
+    module = dispatch_rule.get('module')
+    service = dispatch_rule.get('service')
+    if module and service:
+      raise BadConfigurationException('Both module: and service: in dispatch '
+                                      'entry. Please use only one.')
+    if not (module or service):
+      raise BadConfigurationException("Missing required value 'service'.")
+
+    rule['service'] = module or service
+    rule['domain'], rule['path'] = _URL_SPLITTER_RE.match(
+        dispatch_rule['url']).groups()
+    modified_rules.append(rule)
+
+  return modified_rules
+
 
 def mkdir(dir_path):
   """ Creates a directory.

test/test_appscale.py

@@ -429,7 +429,6 @@ def testDeployWithCloudAppScalefile(self):
     }
     yaml_dumped_contents = yaml.dump(contents)
     self.addMockForAppScalefile(appscale, yaml_dumped_contents)
-
     # finally, mock out the actual appscale-run-instances call
     fake_port = 8080
     fake_host = 'fake_host'
@@ -439,6 +438,7 @@ def testDeployWithCloudAppScalefile(self):
     AppScaleTools.should_receive('update_indexes')
     AppScaleTools.should_receive('update_cron')
     AppScaleTools.should_receive('update_queues')
+    AppScaleTools.should_receive('update_dispatch')
     app = '/bar/app'
     (host, port) = appscale.deploy(app)
     self.assertEquals(fake_host, host)
@@ -499,7 +499,6 @@ def testDeployWithCloudAppScalefileAndTestFlag(self):
     }
     yaml_dumped_contents = yaml.dump(contents)
     self.addMockForAppScalefile(appscale, yaml_dumped_contents)
-
     # finally, mock out the actual appscale-run-instances call
     fake_port = 8080
     fake_host = 'fake_host'
@@ -509,6 +508,7 @@ def testDeployWithCloudAppScalefileAndTestFlag(self):
     AppScaleTools.should_receive('update_indexes')
     AppScaleTools.should_receive('update_cron')
     AppScaleTools.should_receive('update_queues')
+    AppScaleTools.should_receive('update_dispatch')
     app = '/bar/app'
     (host, port) = appscale.deploy(app)
     self.assertEquals(fake_host, host)