gae_ts_mon: instrument Cloud Endpoint methods

appengine_module/gae_ts_mon/config.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 copy
 import datetime
+import functools
 import logging
 import os
 import sys
 import time
 import threading
 
+import endpoints
 import webapp2
 
 from google.appengine.api import modules
 from google.appengine.api.app_identity import app_identity
 from google.appengine.api import runtime
 from google.appengine.ext import ndb
 
 from infra_libs.ts_mon import handlers
 from infra_libs.ts_mon import shared
 from infra_libs.ts_mon.common import http_metrics
@@ skipping 137 matching lines @@
         shared.PUBSUB_TOPIC)
 
   shared.register_global_metrics([shared.appengine_default_version])
   shared.register_global_metrics_callback(
       shared.INTERNAL_CALLBACK_NAME, _internal_callback)
 
   logging.info('Initialized ts_mon with service_name=%s, job_name=%s, '
                'hostname=%s', service_name, job_name, hostname)
 
 
+def update_metrics(name, response_status, elapsed_ms,

[dsansome, 2016/03/15 02:01:20]

This name is quite generic.  Maybe update_http_server_metrics?
Also, maybe move it to http_metrics.py, since it might be useful outside
appengine as well?

[Sergey Berezin, 2016/03/17 22:14:04]

Good idea, done.

+                   request_size=None, response_size=None, user_agent=None):
+  fields = {'status': response_status, 'name': name, 'is_robot': False}
+  if user_agent is not None:
+    # We must not log user agents, but we can store whether or not the
+    # user agent string indicates that the requester was a Google bot.
+    fields['is_robot'] = (
+        'GoogleBot' in user_agent or 'GoogleSecurityScanner' in user_agent)
+
+  http_metrics.server_durations.add(elapsed_ms, fields=fields)
+  http_metrics.server_response_status.increment(fields=fields)
+  if request_size is not None:
+    http_metrics.server_request_bytes.add(request_size, fields=fields)
+  if response_size is not None:  # pragma: no cover
+    http_metrics.server_response_bytes.add(response_size, fields=fields)
+
+
 def _instrumented_dispatcher(dispatcher, request, response, time_fn=time.time):
   start_time = time_fn()
   response_status = 0
   interface.state.store.initialize_context()
   flush_thread = threading.Thread(target=flush_metrics_if_needed)
   flush_thread.start()
   try:
     ret = dispatcher(request, response)
   except webapp2.HTTPException as ex:
     response_status = ex.code
     raise
   except Exception:
     response_status = 500
     raise
   else:
     if isinstance(ret, webapp2.Response):
       response = ret
     response_status = response.status_int
   finally:
     flush_thread.join()
     elapsed_ms = int((time_fn() - start_time) * 1000)
 
-    fields = {'status': response_status, 'name': '', 'is_robot': False}
-    if request.route is not None:
-      # Use the route template regex, not the request path, to prevent an
-      # explosion in possible field values.
-      fields['name'] = request.route.template
-    if request.user_agent is not None:
-      # We must not log user agents, but we can store whether or not the
-      # user agent string indicates that the requester was a Google bot.
-      fields['is_robot'] = (
-          'GoogleBot' in request.user_agent or
-          'GoogleSecurityScanner' in request.user_agent)
+    # Use the route template regex, not the request path, to prevent an
+    # explosion in possible field values.
+    name = request.route.template if request.route is not None else ''
 
-    http_metrics.server_durations.add(elapsed_ms, fields=fields)
-    http_metrics.server_response_status.increment(fields=fields)
-    if request.content_length is not None:
-      http_metrics.server_request_bytes.add(request.content_length,
-                                            fields=fields)
-    if response.content_length is not None:  # pragma: no cover
-      http_metrics.server_response_bytes.add(response.content_length,
-                                             fields=fields)
+    update_metrics(name, response_status, elapsed_ms,
+                   request_size=request.content_length,
+                   response_size=response.content_length,
+                   user_agent=request.user_agent)
+
   return ret
 
 
 def instrument_wsgi_application(app, time_fn=time.time):
   # Don't instrument the same router twice.
   if hasattr(app.router, '__instrumented_by_ts_mon'):
     return
 
   old_dispatcher = app.router.dispatch
 
   def dispatch(router, request, response):
     return _instrumented_dispatcher(old_dispatcher, request, response,
                                     time_fn=time_fn)
 
   app.router.set_dispatcher(dispatch)
   app.router.__instrumented_by_ts_mon = True
 
 
+def instrument_endpoint(time_fn=time.time):
+  """Decorator to instrument Cloud Endpoint methods."""
+  def decorator(fn):
+    method_name = fn.__name__
+    assert method_name
+    @functools.wraps(fn)
+    def decorated(*args, **kwargs):
+      start_time = time_fn()
+      response_status = 0
+      interface.state.store.initialize_context()
+      flush_thread = threading.Thread(target=flush_metrics_if_needed)
+      flush_thread.start()

[nodir, 2016/03/15 18:59:49]

check if it is needed in this thread, do not create a thread in every request

[Sergey Berezin, 2016/03/17 22:14:03]

Done here and in the regular endpoint instrumentation.

+      try:
+        ret = fn(*args, **kwargs)
+        response_status = 200
+        return ret
+      except endpoints.ServiceException as e:
+        response_status = e.http_status
+        raise
+      except Exception:
+        response_status = 500
+        raise
+      finally:
+        flush_thread.join()
+        elapsed_ms = int((time_fn() - start_time) * 1000)
+        update_metrics(method_name, response_status, elapsed_ms)

[nodir, 2016/03/15 18:59:49]

there may be different endpoint services with clashing method names.
I'd prefer it to be actual URL
/_ah/spi/<service>.<method>
e.g. /_ah/spi/BuildBucketApi.search (this is a actual URL for buildbucket
search)

this first parameter of decorated is the service, you can get the class name
there. Also you can retrieve request headers, including user agent, maybe
request size, etc

[Sergey Berezin, 2016/03/17 22:14:03]

Good catch, done.
This requires more assumptions about the method arguments; I'm inclined to do
this in another CL.

+    return decorated
+  return decorator
+
+
 def reset_for_unittest():
   shared.reset_for_unittest()
   interface.reset_for_unittest()