Monitoring proxy for time series data

appengine/chrome_infra_mon_proxy/vm_module.py

Index: appengine/chrome_infra_mon_proxy/vm_module.py
diff --git a/appengine/chrome_infra_mon_proxy/vm_module.py b/appengine/chrome_infra_mon_proxy/vm_module.py
new file mode 100644
index 0000000000000000000000000000000000000000..5f7f73b1bc67d97f6693ab47563cd10a28690fb1
--- /dev/null
+++ b/appengine/chrome_infra_mon_proxy/vm_module.py
@@ -0,0 +1,97 @@
+# Copyright 2015 The Chromium Authors. All rights reserved.
+# Use of this source code is governed by a BSD-style license that can be
+# found in the LICENSE file.
+
+import googleapiclient.http
+import httplib2
+import json
+import logging
+import os
+import traceback
+import webapp2
+
+from google.appengine.api import app_identity
+from google.appengine.ext import ndb
+
+import common
+
+
+def _get_config_data():
+  data_entity = common.MonAcqData.get_by_id(common.CONFIG_DATA_KEY)
+  logging.info('get_config_data(): entity = %r', data_entity)
+  if not data_entity:
+    return None
+  return data_entity.to_dict()
+
+
+def _get_credentials(credentials_dict, scopes):
+  """Obtain Aquisition API credentials as Credentials object."""
+  from oauth2client.client import SignedJwtAssertionCredentials
+
+  # Ideally, we should have loaded credentials with GoogleCredentials.
+  # However, it insists to load only from a file. So, here's a hack.
+  return SignedJwtAssertionCredentials(
+      service_account_name=credentials_dict['client_email'],
+      private_key=credentials_dict['private_key'],
+      scope=scopes,
+      # Extra **kwargs, just in case.
+      service_account_id=credentials_dict['client_id'],
+      private_key_id=credentials_dict['private_key_id'],
+  )
+
+
+class VMHandler(webapp2.RequestHandler):
+
+  def requester_is_me(self):
+    requester_id = self.request.headers.get('X-Appengine-Inbound-Appid')
+    return requester_id == app_identity.get_application_id()
+
+  def requester_is_task(self):
+    return bool(self.request.headers.get('X-AppEngine-TaskName'))
+
+  def post(self):
+    authorized = (
+        common.is_development_server() or
+        self.requester_is_task() or self.requester_is_me())
+    if not authorized:
+      self.abort(403)
+    data = _get_config_data()
+    if not data:
+      self.abort_admin_error('Endpoint data is not set')
+    if not all(f in data for f in ['credentials', 'url']):
+      self.abort_admin_error('Missing required fields: credentials, url')
+
+    url = data['url']
+    http_auth = httplib2.Http()
+    try:
+      credentials = _get_credentials(data['credentials'], data['scopes'])
+    except KeyError as e:
+      self.abort_admin_error('Bad credentials format: missing field %s' % e)
+    http_auth = credentials.authorize(http_auth)

[Vadim Sh., 2015/04/28 18:10:43]

this thing does RSA signing and calls google servers (via urlfetch) to grab
access_token _in each request_, no wonder it is slow

SignedJwtAssertionCredentials object caches access_token internally. It is
mutable stateful object. You need to cache it across requests, so that its
access token is reused. Then you'll pay the penalty of token refresh only once
an hour (when access token expires). Be careful with multithreading though, it
is not thread safe (but request handlers are).

(For high QPS it should be made even more carefully, to avoid multiple requests
all start refreshing the token at once when it expiration comes).

Also oauth2client library is not the best kind of library for writing this sort
of logic in my opinion. It is very flexible and supports all kinds of flows and
storage mechanisms, but when you need to implement something that works fast in
multithreaded environment in very specific conditions (e.g. your case), the
flexibility gets in the way.

[Sergey Berezin, 2015/04/28 19:35:27]

Thanks, this is great info. I'll definitely think about this, and will fix it in
future CLs. It is not mission-critical at the moment though, so I'd say it can
wait for a bit.

Can you suggest alternative libraries? Otherwise, I'll look for a way to store
the token in the Datastore, and reconstruct SignedJwtAssertionCredentials with a
cached token. Refreshing a token at 500 qps without a race storm will indeed be
an interesting challenge. I'll think about it.

[Vadim Sh., 2015/04/28 20:28:52]

Swarming components have simple "get_access_token(...)" function that works with
service accounts and that uses memcache to store access token. It can be
slightly adjusted to use local memory as yet another caching layer (for existing
use cases memcache was more than enough, for long lived managed VM local cache
is justifiable).

We also have wrapper around URLfetch that knows how to use get_access_token and
that implements retries.

Roughly, the usage would be:

from components import auth
from components import net

service_account_key = auth.ServiceAccountKey(
  client_email=...,
  private_key=...,
  private_key_id=...)

response = net.request(
  url=url, 
  method='POST', 
  payload=self.request.body,
  headers={'Content-Type': 'application/x-protobuf'},
  scopes=data['scopes'],
  service_account_key=service_account_key)

(succeeds on  200 <= code <= 299)
(raises net.Error on 300 <= code < 500)
(retries >= 500 a bunch of times)

net.py:
https://github.com/luci/luci-py/blob/master/appengine/components/components/n...
service_account.py:
https://github.com/luci/luci-py/blob/master/appengine/components/components/a...

+    def callback(response, _content):
+      success = 200 <= response.status and response.status <= 299
+      if not success:
+        logging.error('Failed to send data to %s: %d %s',
+                      url, response.status, response.reason)
+    # Important: set content-type to binary, otherwise httplib2 mangles it.
+    data.setdefault('headers', {}).update({
+        'content-length': str(len(self.request.body)),
+        'content-type': 'application/x-protobuf',
+    })
+    request = googleapiclient.http.HttpRequest(
+        http_auth, callback, url, method='POST', body=self.request.body,
+        headers=data['headers'])
+    logging.debug('Sending the payload (synchronously).')
+    request.execute()
+    logging.debug('Done sending the payload.')
+
+  def abort_admin_error(self, message):
+      logging.error('%s; please visit https://%s/admin/',
+                    message, app_identity.get_default_version_hostname())
+      self.abort(500)
+
+
+logging.basicConfig(level=logging.DEBUG)
+app = webapp2.WSGIApplication([
+    ('/vm.*', VMHandler),
+    ], debug=True)