Explorar o código

HUE-7569 [core] Finer grained Thrift pool instrumentation

Romain Rigaux %!s(int64=8) %!d(string=hai) anos
pai
achega
62e358f
Modificáronse 1 ficheiros con 12 adicións e 14 borrados
  1. 12 14
      desktop/core/src/desktop/lib/thrift_util.py

+ 12 - 14
desktop/core/src/desktop/lib/thrift_util.py

@@ -219,16 +219,14 @@ class ConnectionPooler(object):
         this_round_timeout = None
 
       try:
-        connection = self.pooldict[_get_pool_key(conf)].get(
-          block=True, timeout=this_round_timeout)
+        connection = self.pooldict[_get_pool_key(conf)].get(block=True, timeout=this_round_timeout)
+        logging.debug("Thrift client %s got connection %s after %.2f seconds" % (self, connection, time.time() - start_pool_get_time))
       except Queue.Empty:
         has_waited_for = time.time() - start_pool_get_time
         if get_client_timeout is not None and has_waited_for > get_client_timeout:
           raise socket.timeout(
-            ("Timed out after %.2f seconds waiting to retrieve a " +
-             "%s client from the pool.") % (has_waited_for, conf.service_name))
-        logging.warn("Waited %d seconds for a thrift client to %s:%d" %
-          (has_waited_for, conf.host, conf.port))
+            ("Timed out after %.2f seconds waiting to retrieve a %s client from the pool.") % (has_waited_for, conf.service_name))
+        logging.warn("Waited %d seconds for a thrift client to %s:%d" % (has_waited_for, conf.host, conf.port))
 
     return connection
 
@@ -322,6 +320,7 @@ def _grab_transport_from_wrapper(outer_transport):
   else:
     raise Exception("Unknown transport type: " + outer_transport.__class__)
 
+
 class PooledClient(object):
   """
   A wrapper for a SuperClient
@@ -334,8 +333,7 @@ class PooledClient(object):
       return self.__dict__[attr_name]
 
     # Fetch the thrift client from the pool
-    superclient = _connection_pool.get_client(self.conf,
-        get_client_timeout=self.conf.timeout_seconds)
+    superclient = _connection_pool.get_client(self.conf, get_client_timeout=self.conf.timeout_seconds)
 
     # Fetch the attribute. If it's callable, wrap it in a wrapper that re-gets
     # the client.
@@ -374,6 +372,7 @@ class PooledClient(object):
 
           superclient.set_timeout(self.conf.timeout_seconds)
 
+          logging.debug("Thrift client %s call" % superclient)
           return attr(*args, **kwargs)
         except TApplicationException, e:
           # Unknown thrift exception... typically IO errors
@@ -421,6 +420,7 @@ class SuperClient(object):
     res = getattr(self.wrapped, attr)
     if not hasattr(res, '__call__'):
       return res
+
     def wrapper(*args, **kwargs):
       tries_left = 3
       while tries_left:
@@ -433,8 +433,7 @@ class SuperClient(object):
           st = time.time()
 
           str_args = _unpack_guid_secret_in_handle(repr(args))
-          logging.debug("Thrift call: %s.%s(args=%s, kwargs=%s)"
-            % (str(self.wrapped.__class__), attr, str_args, repr(kwargs)))
+          logging.debug("Thrift call: %s.%s(args=%s, kwargs=%s)" % (str(self.wrapped.__class__), attr, str_args, repr(kwargs)))
 
           ret = res(*args, **kwargs)
           log_msg = _unpack_guid_secret_in_handle(repr(ret))
@@ -445,10 +444,8 @@ class SuperClient(object):
 
           duration = time.time() - st
 
-          # Log the duration at different levels, depending on how long
-          # it took.
-          logmsg = "Thrift call %s.%s returned in %dms: %s" % (
-            str(self.wrapped.__class__), attr, duration * 1000, log_msg)
+          # Log the duration at different levels, depending on how long it took.
+          logmsg = "Thrift call %s.%s returned in %dms: %s" % (str(self.wrapped.__class__), attr, duration * 1000, log_msg)
           if duration >= WARN_LEVEL_CALL_DURATION_MS:
             logging.warn(logmsg)
           elif duration >= INFO_LEVEL_CALL_DURATION_MS:
@@ -464,6 +461,7 @@ class SuperClient(object):
         except Exception, e:
           logging.exception("Thrift saw exception (this may be expected).")
           raise
+
         self.transport.close()
 
         if isinstance(e, socket.timeout) or 'read operation timed out' in str(e): # Can come from ssl.SSLError