metrics: added new statsd client and enabled new metrics on app

rhodecode/lib/statsd_client.py

@@ -0,0 +1,49 b''
+from rhodecode.lib._vendor.statsd import client_from_config
+
+
+class StatsdClientNotInitialised(Exception):
+    pass
+
+
+class _Singleton(type):
+    """A metaclass that creates a Singleton base class when called."""
+
+    _instances = {}
+
+    def __call__(cls, *args, **kwargs):
+        if cls not in cls._instances:
+            cls._instances[cls] = super(_Singleton, cls).__call__(*args, **kwargs)
+        return cls._instances[cls]
+
+
+class Singleton(_Singleton("SingletonMeta", (object,), {})):
+    pass
+
+
+class StatsdClientClass(Singleton):
+    setup_run = False
+    statsd_client = None
+    statsd = None
+
+    def __getattribute__(self, name):
+
+        if name.startswith("statsd"):
+            if self.setup_run:
+                return super(StatsdClientClass, self).__getattribute__(name)
+            else:
+                return None
+                #raise StatsdClientNotInitialised("requested key was %s" % name)
+
+        return super(StatsdClientClass, self).__getattribute__(name)
+
+    def setup(self, settings):
+        """
+        Initialize the client
+        """
+        statsd = client_from_config(settings)
+        self.statsd = statsd
+        self.statsd_client = statsd
+        self.setup_run = True
+
+
+StatsdClient = StatsdClientClass()

rhodecode/config/middleware.py

@@ -53,7 +53,7 b' from rhodecode.lib.exc_tracking import s'
 from rhodecode.subscribers import (
     scan_repositories_if_enabled, write_js_routes_if_enabled,
     write_metadata_if_needed, write_usage_data)
-
+from rhodecode.lib.statsd_client import StatsdClient
 
 log = logging.getLogger(__name__)
 
@@ -93,6 +93,9 b' def make_pyramid_app(global_config, **se'
     start_time = time.time()
     log.info('Pyramid app config starting')
 
+    # init and bootstrap StatsdClient
+    StatsdClient.setup(settings)
+
     debug = asbool(global_config.get('debug'))
     if debug:
         enable_debug()
@@ -105,6 +108,8 b' def make_pyramid_app(global_config, **se'
     sanitize_settings_and_apply_defaults(global_config, settings)
 
     config = Configurator(settings=settings)
+    # Init our statsd at very start
+    config.registry.statsd = StatsdClient.statsd
 
     # Apply compatibility patches
     patches.inspect_getargspec()
@@ -124,10 +129,16 b' def make_pyramid_app(global_config, **se'
 
     # creating the app uses a connection - return it after we are done
     meta.Session.remove()
+    statsd = StatsdClient.statsd
+
     total_time = time.time() - start_time
     log.info('Pyramid app `%s` created and configured in %.2fs',
              pyramid_app.func_name, total_time)
-
+    if statsd:
+        elapsed_time_ms = 1000.0 * total_time
+        statsd.timing('rhodecode_app_bootstrap_timing', elapsed_time_ms, tags=[
+            "pyramid_app:{}".format(pyramid_app.func_name)
+        ])
     return pyramid_app
 
 
@@ -169,6 +180,10 b' def error_handler(exception, request):'
         log.exception(
             'error occurred handling this request for path: %s', request.path)
 
+    statsd = request.registry.statsd
+    if statsd and base_response.status_code > 499:
+        statsd.incr('rhodecode_exception')
+
     error_explanation = base_response.explanation or str(base_response)
     if base_response.status_code == 404:
         error_explanation += " Optionally you don't have permission to access this page."
@@ -343,10 +358,6 b' def includeme(config, auth_resources=Non'
         'rhodecode.lib.request_counter.get_request_counter',
         'request_count')
 
-    config.add_request_method(
-        'rhodecode.lib._vendor.statsd.get_statsd_client',
-        'statsd', reify=True)
-
     # Set the authorization policy.
     authz_policy = ACLAuthorizationPolicy()
     config.set_authorization_policy(authz_policy)

rhodecode/lib/_vendor/statsd/base.py

@@ -1,11 +1,27 b''
 from __future__ import absolute_import, division, unicode_literals
 
+import re
 import random
 from collections import deque
 from datetime import timedelta
+from repoze.lru import lru_cache
 
 from .timer import Timer
 
+TAG_INVALID_CHARS_RE = re.compile(r"[^\w\d_\-:/\.]", re.UNICODE)
+TAG_INVALID_CHARS_SUBS = "_"
+
+
+@lru_cache(maxsize=500)
+def _normalize_tags_with_cache(tag_list):
+    return [TAG_INVALID_CHARS_RE.sub(TAG_INVALID_CHARS_SUBS, tag) for tag in tag_list]
+
+
+def normalize_tags(tag_list):
+    # We have to turn our input tag list into a non-mutable tuple for it to
+    # be hashable (and thus usable) by the @lru_cache decorator.
+    return _normalize_tags_with_cache(tuple(tag_list))
+
 
 class StatsClientBase(object):
     """A Base class for various statsd clients."""
@@ -20,10 +36,10 b' class StatsClientBase(object):'
     def pipeline(self):
         raise NotImplementedError()
 
-    def timer(self, stat, rate=1):
-        return Timer(self, stat, rate)
+    def timer(self, stat, rate=1, tags=None):
+        return Timer(self, stat, rate, tags)
 
-    def timing(self, stat, delta, rate=1):
+    def timing(self, stat, delta, rate=1, tags=None):
         """
         Send new timing information.
 
@@ -32,17 +48,17 b' class StatsClientBase(object):'
         if isinstance(delta, timedelta):
             # Convert timedelta to number of milliseconds.
             delta = delta.total_seconds() * 1000.
-        self._send_stat(stat, '%0.6f|ms' % delta, rate)
+        self._send_stat(stat, '%0.6f|ms' % delta, rate, tags)
 
-    def incr(self, stat, count=1, rate=1):
+    def incr(self, stat, count=1, rate=1, tags=None):
         """Increment a stat by `count`."""
-        self._send_stat(stat, '%s|c' % count, rate)
+        self._send_stat(stat, '%s|c' % count, rate, tags)
 
-    def decr(self, stat, count=1, rate=1):
+    def decr(self, stat, count=1, rate=1, tags=None):
         """Decrement a stat by `count`."""
-        self.incr(stat, -count, rate)
+        self.incr(stat, -count, rate, tags)
 
-    def gauge(self, stat, value, rate=1, delta=False):
+    def gauge(self, stat, value, rate=1, delta=False, tags=None):
         """Set a gauge value."""
         if value < 0 and not delta:
             if rate < 1:
@@ -53,16 +69,16 b' class StatsClientBase(object):'
                 pipe._send_stat(stat, '%s|g' % value, 1)
         else:
             prefix = '+' if delta and value >= 0 else ''
-            self._send_stat(stat, '%s%s|g' % (prefix, value), rate)
+            self._send_stat(stat, '%s%s|g' % (prefix, value), rate, tags)
 
     def set(self, stat, value, rate=1):
         """Set a set value."""
         self._send_stat(stat, '%s|s' % value, rate)
 
-    def _send_stat(self, stat, value, rate):
-        self._after(self._prepare(stat, value, rate))
+    def _send_stat(self, stat, value, rate, tags=None):
+        self._after(self._prepare(stat, value, rate, tags))
 
-    def _prepare(self, stat, value, rate):
+    def _prepare(self, stat, value, rate, tags=None):
         if rate < 1:
             if random.random() > rate:
                 return
@@ -71,7 +87,12 b' class StatsClientBase(object):'
         if self._prefix:
             stat = '%s.%s' % (self._prefix, stat)
 
-        return '%s:%s' % (stat, value)
+        res = '%s:%s%s' % (
+            stat,
+            value,
+            ("|#" + ",".join(normalize_tags(tags))) if tags else "",
+        )
+        return res
 
     def _after(self, data):
         if data:

rhodecode/lib/_vendor/statsd/timer.py

@@ -21,10 +21,11 b' def safe_wraps(wrapper, *args, **kwargs)'
 class Timer(object):
     """A context manager/decorator for statsd.timing()."""
 
-    def __init__(self, client, stat, rate=1):
+    def __init__(self, client, stat, rate=1, tags=None):
         self.client = client
         self.stat = stat
         self.rate = rate
+        self.tags = tags
         self.ms = None
         self._sent = False
         self._start_time = None
@@ -38,7 +39,7 b' class Timer(object):'
                 return f(*args, **kwargs)
             finally:
                 elapsed_time_ms = 1000.0 * (time_now() - start_time)
-                self.client.timing(self.stat, elapsed_time_ms, self.rate)
+                self.client.timing(self.stat, elapsed_time_ms, self.rate, self.tags)
         return _wrapped
 
     def __enter__(self):

rhodecode/lib/celerylib/__init__.py

@@ -25,6 +25,7 b' import rhodecode'
 from zope.cachedescriptors.property import Lazy as LazyProperty
 from rhodecode.lib.celerylib.loader import (
     celery_app, RequestContextTask, get_logger)
+from rhodecode.lib.statsd_client import StatsdClient
 
 async_task = celery_app.task
 
@@ -42,16 +43,19 b' class ResultWrapper(object):'
 
 
 def run_task(task, *args, **kwargs):
-    log.debug('Got task `%s` for execution', task)
+    log.debug('Got task `%s` for execution, celery mode enabled:%s', task, rhodecode.CELERY_ENABLED)
     if task is None:
         raise ValueError('Got non-existing task for execution')
 
+    statsd = StatsdClient.statsd
+    exec_mode = 'sync'
+
     if rhodecode.CELERY_ENABLED:
-        celery_is_up = False
+
         try:
             t = task.apply_async(args=args, kwargs=kwargs)
-            celery_is_up = True
             log.debug('executing task %s:%s in async mode', t.task_id, task)
+            exec_mode = 'async'
             return t
 
         except socket.error as e:
@@ -69,4 +73,9 b' def run_task(task, *args, **kwargs):'
     else:
         log.debug('executing task %s:%s in sync mode', 'TASK', task)
 
+    if statsd:
+        statsd.incr('rhodecode_celery_task', tags=[
+            'task:{}'.format(task),
+            'mode:{}'.format(exec_mode)
+        ])
     return ResultWrapper(task(*args, **kwargs))

rhodecode/lib/middleware/request_wrapper.py

@@ -51,19 +51,32 b' class RequestWrapperTween(object):'
         finally:
             count = request.request_count()
             _ver_ = rhodecode.__version__
-            statsd = request.statsd
+            _path = safe_str(get_access_path(request.environ))
+            _auth_user = self._get_user_info(request)
+
             total = time.time() - start
-            if statsd:
-                statsd.timing('rhodecode.req.timing', total)
-                statsd.incr('rhodecode.req.count')
-
             log.info(
                 'Req[%4s] %s %s Request to %s time: %.4fs [%s], RhodeCode %s',
-                count, self._get_user_info(request), request.environ.get('REQUEST_METHOD'),
-                safe_str(get_access_path(request.environ)), total,
-                get_user_agent(request. environ), _ver_
+                count, _auth_user, request.environ.get('REQUEST_METHOD'),
+                _path, total, get_user_agent(request. environ), _ver_
             )
 
+            statsd = request.registry.statsd
+            if statsd:
+                elapsed_time_ms = 1000.0 * total
+                statsd.timing(
+                    'rhodecode_req_timing', elapsed_time_ms,
+                    tags=[
+                        "path:{}".format(_path),
+                        "user:{}".format(_auth_user.user_id)
+                    ]
+                )
+                statsd.incr(
+                    'rhodecode_req_count', tags=[
+                        "path:{}".format(_path),
+                        "user:{}".format(_auth_user.user_id)
+                    ])
+
         return response