gae_ts_mon: instrument Cloud Endpoint methods

appengine_module/gae_ts_mon/config.py

Index: appengine_module/gae_ts_mon/config.py
diff --git a/appengine_module/gae_ts_mon/config.py b/appengine_module/gae_ts_mon/config.py
index 2cbe0a344d95538c079dc8fba7c7c115efbdbbcf..73d450cd29a5f0ffcbc056e4f9941fd76ffa2b4a 100644
--- a/appengine_module/gae_ts_mon/config.py
+++ b/appengine_module/gae_ts_mon/config.py
@@ -4,12 +4,14 @@
 
 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
@@ -167,6 +169,23 @@ def initialize(app=None, is_enabled_fn=None, cron_module='default',
                '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
@@ -189,26 +208,15 @@ def _instrumented_dispatcher(dispatcher, request, response, time_fn=time.time):
     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)
-
-    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)
+    # 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 ''
+
+    update_metrics(name, response_status, elapsed_ms,
+                   request_size=request.content_length,
+                   response_size=response.content_length,
+                   user_agent=request.user_agent)
+
   return ret
 
 
@@ -227,6 +235,36 @@ def instrument_wsgi_application(app, time_fn=time.time):
   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()