Monitoring proxy for time series data

appengine/chrome_infra_mon_proxy/main.py

+# 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 logging
+import random
+import os
+import time
+import sys
+import webapp2
+
+from protorpc import messages
+from protorpc import message_types
+from protorpc import remote
+
+from google.appengine.api import app_identity, taskqueue
+
+import admin_handler
+import common
+
+BASE_DIR = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, os.path.join(BASE_DIR, 'components', 'third_party'))
+
+from components import auth
+
+VM_MODULES = ['vm1', 'vm2', 'vm3']

[ghost stip (do not use), 2015/04/14 00:36:42]

oppan dijkstra style:

VM_MODULE_COUNT = 3
VM_MODULES = ['vm%d' % i for i in range(VM_MODULE_COUNT)]

[Sergey Berezin (google), 2015/04/16 04:39:07]

I feel the temptation :-) But this suggests the number of modules can be
parameterized, and honestly, I want to fix it to exactly 3, and make it hard to
add more. It's already too many as it is for maintenance. I need N+1 redundancy,
and N=2 is the minimum viable in this case (avoiding SPOF and just handling
load). So I'm pulling an exception to the generalization rule. And it's 2 lines
instead of 1 :-)

Context: the number of modules is dictated by the number of NAT boxes I need to
have. I couldn't make those load balance separately, so I have to load balance
at the level of modules, each with its own NAT box. 3 us-central1 zones, 3 NAT
boxes. 

+
+
+def is_group_member(group_name):
+  """Skip authorization for dev appserver."""
+  if common.is_development_server():
+    auth_decorator = auth.public
+  else:
+    auth_decorator = auth.require( # pragma: no branch
+        lambda: auth.is_group_member(group_name))
+
+  def decorator(fn):
+    return auth_decorator(fn)
+  return decorator
+
+
+class NoBackendException(Exception):
+  pass
+
+
+class LoadBalancer(object):
+  """Balance the load among VM modules.
+
+  TODO(sergeyberezin): take into account health checks on the
+  corresponding NAT boxes. Specifically, fetch the health status
+  from the datastore in __init__(), and update it periodically as needed.
+  """
+  def __init__(self):
+    # dict: module name -> bool: whether module is healthy.
+    self._health = {module: True for module in VM_MODULES}
+    # Seconds since UNIX epoch when last health check was made.

[ghost stip (do not use), 2015/04/14 00:36:42]

probably want to remove now until we figure out how to check NAT health?

[Sergey Berezin (google), 2015/04/16 04:39:07]

Done.

+    self._last_health_check = 0
+
+  def _healthy_modules(self):
+    """Return the list of names of all healthy modules."""
+    # TODO(sergeyberezin): perform health checks and update the
+    # datastore once in a while.
+    return [module for module in VM_MODULES if self._health[module]]
+
+  def choose_module(self):
+    """Select a module to send the data to."""
+    modules = self._healthy_modules()
+    if not modules:
+      raise NoBackendException('Error: no healthy backends are available')
+    return random.randint(0, len(modules)-1)

[ghost stip (do not use), 2015/04/14 00:36:42]

return random.choice(modules)

[Sergey Berezin (google), 2015/04/16 04:39:06]

Done. Nice, thanks!

+
+
+def forward_data(data):
+  """Forwards the raw data to the backend."""
+  # logging.debug('Forwarding data: %s', common.payload_stats(data))

[ghost stip (do not use), 2015/04/14 00:36:42]

why commented out? seems legit

[Sergey Berezin (google), 2015/04/16 04:39:06]

It computes md5, and now that I debugged the headers, I don't really need it
anymore. Saves CPU.

+  # Task queue should work correctly both in dev and prod server.
+  lb = LoadBalancer()
+  queue_name = VM_MODULES[lb.choose_module()]

[ghost stip (do not use), 2015/04/14 00:36:42]

queue_name = lb.choose_module()

[Sergey Berezin (google), 2015/04/16 04:39:07]

Done.

+  logging.info('Selected queue: %s', queue_name)
+  taskqueue.add(url='/vm', payload=data, queue_name=queue_name)
+
+
+class MonacqHandler(auth.AuthenticatingHandler):
+  # Disable XSRF in local dev appserver; otherwise requests will fail.
+  if common.is_development_server():
+    xsrf_token_enforce_on = [] # pragma: no cover

[ghost stip (do not use), 2015/04/14 00:36:43]

nit: two spaces before comment

[Sergey Berezin (google), 2015/04/16 04:39:07]

Done.

+
+  @is_group_member('service-account-monitoring-proxy')
+  def get(self):
+    self.response.headers.add_header('Content-Type', 'text/plain')
+    self.response.write('GET requests are not allowed.')

[ghost stip (do not use), 2015/04/14 00:36:42]

I think if you don't provide a get function to begin with the person gets a 405
automatically

[Sergey Berezin (google), 2015/04/16 04:39:06]

Indeed - removed.

+    self.response.set_status(403)

[ghost stip (do not use), 2015/04/14 00:36:42]

at least make it 405

[Sergey Berezin (google), 2015/04/16 04:39:07]

N/A - removed def get().

+    return
+
+  @is_group_member('service-account-monitoring-proxy')
+  def post(self):
+    # logging.info('Received POST /monacq: %s',
+    #              common.payload_stats(self.request.body))
+    forward_data(self.request.body)
+
+
+class MainHandler(webapp2.RequestHandler):
+  def get(self):
+    # TODO(sergeyberezin): create a meaningful welcome page with some
+    # documentation.
+    msg = 'There is nothing here yet.\n'

[ghost stip (do not use), 2015/04/14 00:36:42]

make this a little more friendly!

msg = 'Welcome to the monitoring proxy! Nothing to see here (yet).'

[Sergey Berezin (google), 2015/04/16 04:39:06]

Done - created an actual page with some useful content.

+    self.response.headers['Content-Type'] = 'text/plain'

[ghost stip (do not use), 2015/04/14 00:36:42]

is this needed?

[Sergey Berezin (google), 2015/04/16 04:39:07]

Nope - it's all gone now.

+    self.response.out.write(msg)
+
+
+
+logging.basicConfig(level=logging.DEBUG)
+
+main_handlers = [
+    (r'/', MainHandler),
+    (r'/monacq', MonacqHandler),
+    (r'/admin/(.*)', admin_handler.AdminDispatch),
+]
+
+app = webapp2.WSGIApplication(main_handlers, debug=True)