gae_ts_mon: flush metrics asynchronously

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 logging
 import os
 import sys
 import time
@@ skipping 153 matching lines @@
       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 _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)
 
     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)
-    flush_metrics_if_needed()
-
   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 reset_for_unittest():
   shared.reset_for_unittest()
   interface.reset_for_unittest()