python/google/appengine/ext/datastore_admin/remote_api_put_stub.py

   1 #!/usr/bin/env python
   2 #
   3 # Copyright 2007 Google Inc.
   4 #
   5 # Licensed under the Apache License, Version 2.0 (the "License");
   6 # you may not use this file except in compliance with the License.
   7 # You may obtain a copy of the License at
   8 #
   9 #     http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 # Unless required by applicable law or agreed to in writing, software
  12 # distributed under the License is distributed on an "AS IS" BASIS,
  13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 # See the License for the specific language governing permissions and
  15 # limitations under the License.
  16 #
  17
  18
  19
  20
  21 """An apiproxy stub that calls a remote handler via HTTP.
  22
  23 This is a special version of the remote_api_stub which sends all traffic
  24 to the local backends *except* for datastore_v3 Put and datastore_v4
  25 AllocateIds calls where the key contains a remote app_id.
  26
  27 Calls to datastore_v3 Put and datastore_v4 AllocateIds for which the entity
  28 keys contain a remote app_id are sent to the remote app.
  29
  30 It re-implements parts of the remote_api_stub so as to replace dependencies on
  31 the (SDK only) appengine_rpc with urlfetch.
  32 """
  33
  34
  35
  36
  37
  38
  39
  40
  41 import logging
  42 import pickle
  43 import random
  44 import yaml
  45
  46 from google.appengine.api import apiproxy_rpc
  47 from google.appengine.api import apiproxy_stub_map
  48 from google.appengine.api import urlfetch
  49 from google.appengine.ext.remote_api import remote_api_pb
  50 from google.appengine.runtime import apiproxy_errors
  51
  52
  53 class Error(Exception):
  54   """Base class for exceptions in this module."""
  55
  56
  57 class ConfigurationError(Error):
  58   """Exception for configuration errors."""
  59
  60
  61 class FetchFailed(Exception):
  62   """Remote fetch failed."""
  63
  64
  65 class UnknownJavaServerError(Error):
  66   """Exception for exceptions returned from a Java remote_api handler."""
  67
  68
  69 class RemoteTransactionsUnimplemented(Error):
  70   """Remote Put requests do not support transactions."""
  71
  72
  73 class RemoteApiDatastoreStub(object):
  74   """A specialised stub for writing to a remote App Engine datastore.
  75
  76   This class supports checking the app_id of a datastore op and either passing
  77   the request through to the local app or sending it to a remote app.
  78   Subclassed to implement supported services datastore_v3 and datastore_v4.
  79   """
  80
  81
  82   _SERVICE_NAME = None
  83
  84   def __init__(self, remote_url, target_app_id, extra_headers, normal_stub):
  85     """Constructor.
  86
  87     Args:
  88       remote_url: The URL of the remote_api handler.
  89       target_app_id: The app_id to intercept calls for.
  90       extra_headers: Headers to send (for authentication).
  91       normal_stub: The standard stub to delegate most calls to.
  92     """
  93     self.remote_url = remote_url
  94     self.target_app_id = target_app_id
  95     self.extra_headers = InsertDefaultExtraHeaders(extra_headers)
  96     self.normal_stub = normal_stub
  97
  98   def CreateRPC(self):
  99     """Creates RPC object instance.
 100
 101     Returns:
 102       a instance of RPC.
 103     """
 104     return apiproxy_rpc.RPC(stub=self)
 105
 106   @classmethod
 107   def ServiceName(cls):
 108     """Return the name of the datastore service supported by this stub."""
 109     return cls._SERVICE_NAME
 110
 111   def MakeSyncCall(self, service, call, request, response):
 112     """Handle all calls to this stub; delegate as appropriate."""
 113     assert service == self.ServiceName(), '%s does not support service %s' % (
 114         type(self), service)
 115
 116     explanation = []
 117     assert request.IsInitialized(explanation), explanation
 118
 119     handler = getattr(self, '_Dynamic_' + call, None)
 120     if handler:
 121       handler(request, response)
 122     else:
 123       self.normal_stub.MakeSyncCall(service, call, request, response)
 124
 125     assert response.IsInitialized(explanation), explanation
 126
 127   def _MakeRemoteSyncCall(self, service, call, request, response):
 128     """Send an RPC to a remote_api endpoint."""
 129     request_pb = remote_api_pb.Request()
 130     request_pb.set_service_name(service)
 131     request_pb.set_method(call)
 132     request_pb.set_request(request.Encode())
 133
 134     response_pb = remote_api_pb.Response()
 135     encoded_request = request_pb.Encode()
 136     try:
 137       urlfetch_response = urlfetch.fetch(self.remote_url, encoded_request,
 138                                          urlfetch.POST, self.extra_headers,
 139                                          follow_redirects=False,
 140                                          deadline=10)
 141     except Exception, e:
 142
 143
 144       logging.exception('Fetch failed to %s', self.remote_url)
 145       raise FetchFailed(e)
 146     if urlfetch_response.status_code != 200:
 147       logging.error('Fetch failed to %s; Status %s; body %s',
 148                     self.remote_url,
 149                     urlfetch_response.status_code,
 150                     urlfetch_response.content)
 151       raise FetchFailed(urlfetch_response.status_code)
 152     response_pb.ParseFromString(urlfetch_response.content)
 153
 154     if response_pb.has_application_error():
 155       error_pb = response_pb.application_error()
 156       raise apiproxy_errors.ApplicationError(error_pb.code(),
 157                                              error_pb.detail())
 158     elif response_pb.has_exception():
 159       raise pickle.loads(response_pb.exception())
 160     elif response_pb.has_java_exception():
 161       raise UnknownJavaServerError('An unknown error has occured in the '
 162                                    'Java remote_api handler for this call.')
 163     else:
 164       response.ParseFromString(response_pb.response())
 165
 166
 167 class RemoteApiDatastoreV3Stub(RemoteApiDatastoreStub):
 168   """A specialised stub for calling datastore_v3 Put on a foreign datastore.
 169
 170   This stub passes through all requests to the normal stub except for
 171   datastore v3 Put. It will check those to see if the put is for the local app
 172   or a remote app, and if remote will send traffic remotely.
 173   """
 174
 175   _SERVICE_NAME = 'datastore_v3'
 176
 177   def _Dynamic_Put(self, request, response):
 178     """Handle a Put request and route remotely if it matches the target app.
 179
 180     Args:
 181       request: A datastore_pb.PutRequest
 182       response: A datastore_pb.PutResponse
 183
 184     Raises:
 185       RemoteTransactionsUnimplemented: Remote transactions are unimplemented.
 186     """
 187
 188     if request.entity_list():
 189       entity = request.entity(0)
 190       if entity.has_key() and entity.key().app() == self.target_app_id:
 191         if request.has_transaction():
 192
 193
 194           raise RemoteTransactionsUnimplemented()
 195         self._MakeRemoteSyncCall(self.ServiceName(), 'Put', request, response)
 196         return
 197
 198
 199     self.normal_stub.MakeSyncCall(self.ServiceName(), 'Put', request, response)
 200
 201
 202 class RemoteApiDatastoreV4Stub(RemoteApiDatastoreStub):
 203   """A remote api stub to call datastore_v4 AllocateIds on a foreign datastore.
 204
 205   This stub passes through all requests to the normal datastore_v4 stub except
 206   for datastore v4 AllocateIds. It will check those to see if the keys are for
 207   the local app or a remote app, and if remote will send traffic remotely.
 208   """
 209
 210   _SERVICE_NAME = 'datastore_v4'
 211
 212   def _Dynamic_AllocateIds(self, request, response):
 213     """Handle v4 AllocateIds and route remotely if it matches the target app.
 214
 215     Args:
 216       request: A datastore_v4_pb.AllocateIdsRequest
 217       response: A datastore_v4_pb.AllocateIdsResponse
 218     """
 219
 220     if request.reserve_size() > 0:
 221       app_id = request.reserve(0).partition_id().dataset_id()
 222     elif request.allocate_size() > 0:
 223       app_id = request.allocate(0).partition_id().dataset_id()
 224     else:
 225       app_id = None
 226
 227     if app_id == self.target_app_id:
 228       self._MakeRemoteSyncCall(self.ServiceName(), 'AllocateIds', request,
 229                                response)
 230     else:
 231       self.normal_stub.MakeSyncCall(self.ServiceName(), 'AllocateIds', request,
 232                                     response)
 233
 234
 235
 236 def get_remote_app_id(remote_url, extra_headers=None):
 237   """Get the app_id from the remote_api endpoint.
 238
 239   This also has the side effect of verifying that it is a remote_api endpoint.
 240
 241   Args:
 242     remote_url: The url to the remote_api handler.
 243     extra_headers: Headers to send (for authentication).
 244
 245   Returns:
 246     app_id: The app_id of the target app.
 247
 248   Raises:
 249     FetchFailed: Urlfetch call failed.
 250     ConfigurationError: URLfetch succeeded but results were invalid.
 251   """
 252   rtok = str(random.random())[2:]
 253   url = remote_url + '?rtok=' + rtok
 254   if not extra_headers:
 255     extra_headers = {}
 256   if 'X-appcfg-api-version' not in extra_headers:
 257     extra_headers['X-appcfg-api-version'] = '1'
 258   try:
 259     urlfetch_response = urlfetch.fetch(url, None, urlfetch.GET,
 260                                        extra_headers, follow_redirects=False,
 261                                        deadline=10)
 262   except Exception, e:
 263
 264
 265     logging.exception('Fetch failed to %s', remote_url)
 266     raise FetchFailed('Fetch to %s failed: %r' % (remote_url, e))
 267   if urlfetch_response.status_code != 200:
 268     logging.error('Fetch failed to %s; Status %s; body %s',
 269                   remote_url,
 270                   urlfetch_response.status_code,
 271                   urlfetch_response.content)
 272     raise FetchFailed('Fetch to %s failed with status %s' %
 273                       (remote_url, urlfetch_response.status_code))
 274   response = urlfetch_response.content
 275   if not response.startswith('{'):
 276     logging.info('Response unparasable: %s', response)
 277     raise ConfigurationError(
 278         'Invalid response received from server: %s' % response)
 279   app_info = yaml.load(response)
 280   if not app_info or 'rtok' not in app_info or 'app_id' not in app_info:
 281     logging.info('Response unparsable: %s', response)
 282     raise ConfigurationError('Error parsing app_id lookup response')
 283   if str(app_info['rtok']) != rtok:
 284     logging.info('Response invalid token (expected %s): %s', rtok, response)
 285     raise ConfigurationError('Token validation failed during app_id lookup. '
 286                              '(sent %s, got %s)' % (repr(rtok),
 287                                                     repr(app_info['rtok'])))
 288   return app_info['app_id']
 289
 290
 291 def InsertDefaultExtraHeaders(extra_headers):
 292   """Add defaults to extra_headers arg for stub configuration.
 293
 294   This permits comparison of a proposed RemoteApiDatastoreStub config with
 295   an existing config.
 296
 297   Args:
 298     extra_headers: The dict of headers to transform.
 299
 300   Returns:
 301     A new copy of the input dict with defaults set.
 302   """
 303   extra_headers = extra_headers.copy() if extra_headers else {}
 304   if 'X-appcfg-api-version' not in extra_headers:
 305     extra_headers['X-appcfg-api-version'] = '1'
 306   return extra_headers
 307
 308
 309 def StubConfigEqualsRequestedConfig(stub, remote_url, target_app_id,
 310                                     extra_headers):
 311   """Return true if the stub and requseted stub config match.
 312
 313   Args:
 314     stub: a RemoteApiDatastore stub.
 315     remote_url: requested remote_api url of target app.
 316     target_app_id: requested app_id of target (remote) app.
 317     extra_headers: requested headers for auth, possibly not yet including
 318       defaults applied at stub instantiation time.
 319
 320   Returns:
 321     True if the requested config matches the stub, else False.
 322   """
 323   return (stub.remote_url == remote_url and
 324           stub.target_app_id == target_app_id and
 325           stub.extra_headers == InsertDefaultExtraHeaders(extra_headers))
 326
 327
 328 def configure_remote_put(remote_url, target_app_id, extra_headers=None):
 329   """Does necessary setup to intercept v3 Put and v4 AllocateIds.
 330
 331   Args:
 332     remote_url: The url to the remote_api handler.
 333     target_app_id: The app_id of the target app.
 334     extra_headers: Headers to send (for authentication).
 335
 336   Raises:
 337     ConfigurationError: if there is a error configuring the stubs.
 338   """
 339   if not target_app_id or not remote_url:
 340     raise ConfigurationError('app_id and remote_url required')
 341
 342   for stub_class in (RemoteApiDatastoreV3Stub, RemoteApiDatastoreV4Stub):
 343     service_name = stub_class.ServiceName()
 344     original_datastore_stub = apiproxy_stub_map.apiproxy.GetStub(service_name)
 345     if isinstance(original_datastore_stub, stub_class):
 346       logging.info('Datastore Admin %s RemoteApi stub is already configured.',
 347                    service_name)
 348       if not StubConfigEqualsRequestedConfig(
 349           original_datastore_stub, remote_url, target_app_id, extra_headers):
 350         logging.warning('Requested Datastore Admin %s RemoteApi stub '
 351                         'configuration differs from existing configuration, '
 352                         'attempting reconfiguration.', service_name)
 353         datastore_stub = stub_class(remote_url, target_app_id, extra_headers,
 354                                     original_datastore_stub.normal_stub)
 355         apiproxy_stub_map.apiproxy.ReplaceStub(service_name, datastore_stub)
 356     else:
 357       datastore_stub = stub_class(remote_url, target_app_id, extra_headers,
 358                                   original_datastore_stub)
 359       apiproxy_stub_map.apiproxy.RegisterStub(service_name, datastore_stub)