MatrixSynapse/synapse/util/metrics.py

# -*- coding: utf-8 -*-
# Copyright 2016 OpenMarket Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from twisted.internet import defer

from prometheus_client import Counter
from synapse.util.logcontext import LoggingContext

from functools import wraps
import logging


logger = logging.getLogger(__name__)

block_counter = Counter("synapse_util_metrics_block_count", "", ["block_name"])

block_timer = Counter("synapse_util_metrics_block_time_seconds", "", ["block_name"])

block_ru_utime = Counter("synapse_util_metrics_block_ru_utime_seconds", "", ["block_name"])

block_ru_stime = Counter("synapse_util_metrics_block_ru_stime_seconds", "", ["block_name"])

block_db_txn_count = Counter("synapse_util_metrics_block_db_txn_count", "", ["block_name"])

# seconds spent waiting for db txns, excluding scheduling time, in this block
block_db_txn_duration = Counter("synapse_util_metrics_block_db_txn_duration_seconds", "", ["block_name"])

# seconds spent waiting for a db connection, in this block
block_db_sched_duration = Counter("synapse_util_metrics_block_db_sched_duration_seconds", "", ["block_name"])


def measure_func(name):
    def wrapper(func):
        @wraps(func)
        @defer.inlineCallbacks
        def measured_func(self, *args, **kwargs):
            with Measure(self.clock, name):
                r = yield func(self, *args, **kwargs)
            defer.returnValue(r)
        return measured_func
    return wrapper


class Measure(object):
    __slots__ = [
        "clock", "name", "start_context", "start", "new_context", "ru_utime",
        "ru_stime",
        "db_txn_count", "db_txn_duration_ms", "db_sched_duration_ms",
        "created_context",
    ]

    def __init__(self, clock, name):
        self.clock = clock
        self.name = name
        self.start_context = None
        self.start = None
        self.created_context = False

    def __enter__(self):
        self.start = self.clock.time_msec()
        self.start_context = LoggingContext.current_context()
        if not self.start_context:
            self.start_context = LoggingContext("Measure")
            self.start_context.__enter__()
            self.created_context = True

        self.ru_utime, self.ru_stime = self.start_context.get_resource_usage()
        self.db_txn_count = self.start_context.db_txn_count
        self.db_txn_duration_ms = self.start_context.db_txn_duration_ms
        self.db_sched_duration_ms = self.start_context.db_sched_duration_ms

    def __exit__(self, exc_type, exc_val, exc_tb):
        if isinstance(exc_type, Exception) or not self.start_context:
            return

        duration = self.clock.time_msec() - self.start

        block_counter.labels(self.name).inc()
        block_timer.labels(self.name).inc(duration)

        context = LoggingContext.current_context()

        if context != self.start_context:
            logger.warn(
                "Context has unexpectedly changed from '%s' to '%s'. (%r)",
                self.start_context, context, self.name
            )
            return

        if not context:
            logger.warn("Expected context. (%r)", self.name)
            return

        ru_utime, ru_stime = context.get_resource_usage()

        block_ru_utime.labels(self.name).inc(ru_utime - self.ru_utime)
        block_ru_stime.labels(self.name).inc(ru_stime - self.ru_stime)
        block_db_txn_count.labels(self.name).inc(context.db_txn_count - self.db_txn_count)
        block_db_txn_duration.labels(self.name).inc(
            (context.db_txn_duration_ms - self.db_txn_duration_ms) / 1000.)
        block_db_sched_duration.labels(self.name).inc(
            (context.db_sched_duration_ms - self.db_sched_duration_ms) / 1000.)

        if self.created_context:
            self.start_context.__exit__(exc_type, exc_val, exc_tb)