Instrument HTTP client requests to buildbucket

scripts/master/deferred_resource.py

 # Copyright 2014 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.
 
 """deferred_resource converts blocking apiclient resource to deferred."""
 
 import collections
 import datetime
 import functools
 import httplib
 import ssl
 import threading
 import traceback
 
 from twisted.internet import defer, reactor, threads
 from twisted.python import log as twistedLog
 from twisted.python.threadpool import ThreadPool
 import apiclient
 import apiclient.discovery
-import httplib2
 import oauth2client
 
+from infra_libs import InstrumentedHttp
+
 
 DEFAULT_RETRY_ATTEMPT_COUNT = 5
 DEFAULT_RETRY_WAIT_SECONDS = 1
 
 
 if httplib.FORBIDDEN not in oauth2client.client.REFRESH_STATUS_CODES:
   oauth2client.client.REFRESH_STATUS_CODES.append(httplib.FORBIDDEN)
 
 
 class NotStartedError(Exception):
@@ skipping 81 matching lines @@
       new_prefix = self._path + (name,)
       sub_api = self._api_cache.get(new_prefix)
       if not sub_api:
         sub_api = self._owner.Api(self._owner, path=self._path + (name,))
         self._api_cache[new_prefix] = sub_api
       return sub_api
 
   def __init__(
       self, resource, credentials=None, max_concurrent_requests=1,
       retry_wait_seconds=None, retry_attempt_count=None, verbose=False,
-      log_prefix='', timeout=None, _pool=None):
+      log_prefix='', timeout=None, _pool=None, http_client_name=None):
     """Creates a DeferredResource.
 
     Args:
       resource (apiclient.Resource): a resource, such as one generated by
         apiclient.discovery.build.
       credentials (oauth2client.client.Credentials or CredentialFactory):
         credentials to use to make API requests.
       max_concurrent_requests (int): maximum number of concurrent requests.
         Defaults to 1.
       retry_wait_seconds (int, float): initial wait interval for request
         retrial. In seconds, defaults to 1.
       retry_attempt_count (int): number of attempts before giving up.
         Defaults to 5.
       verbose (bool): if True, log each request/response.
       log_prefix (str): prefix for log messages.
       timeout (int): request timeout in seconds. If None is passed
         then Python's default timeout for sockets will be used. See
         for example the docs of socket.setdefaulttimeout():
         http://docs.python.org/library/socket.html#socket.setdefaulttimeout
+      http_client_name (str): an identifier for the HTTP requests made by this
+        resource.  Included with monitoring metrics.
     """
     max_concurrent_requests = max_concurrent_requests or 1
     assert resource, 'resource not specified'
     if retry_wait_seconds is None:
       retry_wait_seconds = DEFAULT_RETRY_WAIT_SECONDS
     assert isinstance(retry_wait_seconds, (int, float))
     if retry_attempt_count is None:
       retry_attempt_count = DEFAULT_RETRY_ATTEMPT_COUNT
     assert isinstance(retry_attempt_count, int)
+    if http_client_name is None:
+      http_client_name = 'deferred_resource'
 
     self._pool = _pool or self._create_thread_pool(max_concurrent_requests)
     self._resource = resource
     self.credentials = credentials
     self.retry_wait_seconds = retry_wait_seconds
     self.retry_attempt_count = retry_attempt_count
     self.verbose = verbose
     self.log_prefix = log_prefix
     self.api = self.Api(self)
     self._th_local = threading.local()
     self.started = False
     self.timeout = timeout
+    self.http_client_name = http_client_name
 
   @classmethod
   def _create_thread_pool(cls, max_concurrent_requests):
     return DaemonThreadPool(minthreads=1, maxthreads=max_concurrent_requests)
 
   @classmethod
   def _create_async(
       cls, resource_factory, max_concurrent_requests=1, _pool=None, **kwargs):
     _pool = _pool or cls._create_thread_pool(max_concurrent_requests)
     result = defer.Deferred()
@@ skipping 153 matching lines @@
         create_creds = False
         if getattr(self._th_local, 'http', None) is None:
           create_creds = True
         elif (self._th_local.credentials_expiry is not None and
               self._th_local.credentials_expiry <= now):
           create_creds = True
 
         if create_creds:
           self._th_local.credentials = None
           self._th_local.credentials_expiry = None
-          self._th_local.http = httplib2.Http(timeout=self.timeout)
+          self._th_local.http = InstrumentedHttp(
+              self.http_client_name, timeout=self.timeout)
           if self.credentials is not None:
             creds = self.credentials
             ttl = None
             if isinstance(creds, CredentialFactory):
               ttl = creds.ttl
               creds = creds()
             self._th_local.credentials = creds.from_json(creds.to_json())
             if ttl is not None:
               self._th_local.credentials_expiry = now + ttl
             self._th_local.http = self._th_local.credentials.authorize(
@@ skipping 23 matching lines @@
   return d
 
 
 def is_transient(ex):
   if isinstance(ex, apiclient.errors.HttpError) and ex.resp:
     return ex.resp.status >= 500;
   if isinstance(ex, ssl.SSLError):
     # No reason, no errcode.
     return "timed out" in str(ex)
   return False