Common - Perf Eval Decorator

}

class MetricsPool:

    def __init__(self) -> None:

    def __init__(self, labels : Dict[str, str] = {}) -> None:

        self._labels = labels

        self._metrics : Dict[str, MetricWrapperBase] = dict()

        self._lock = threading.Lock()

    def get_labels(self) -> Dict[str, str]: return self._labels

    def get_or_create(self, function_name : str, metric_type : MetricTypeEnum, **metric_params) -> MetricWrapperBase:

        metric_name = str(metric_type.value).format(function_name).upper()

        with self._lock:

                metric_tuple : Tuple[MetricWrapperBase, Dict] = METRIC_TO_CLASS_PARAMS.get(metric_type)

                metric_class, default_metric_params = metric_tuple

                if len(metric_params) == 0: metric_params = default_metric_params

                self._metrics[metric_name] = metric_class(metric_name, '', **metric_params)

                labels = sorted(self._labels.keys())

                self._metrics[metric_name] = metric_class(metric_name, '', labels, **metric_params)

            return self._metrics[metric_name]

def meter_method(metrics_pool : MetricsPool):

    def outer_wrapper(func):

        func_name = func.__name__

        labels = metrics_pool.get_labels()

        histogram_duration : Histogram = metrics_pool.get_or_create(func_name, MetricTypeEnum.HISTOGRAM_DURATION)

        counter_started    : Counter   = metrics_pool.get_or_create(func_name, MetricTypeEnum.COUNTER_STARTED)

        counter_completed  : Counter   = metrics_pool.get_or_create(func_name, MetricTypeEnum.COUNTER_COMPLETED)

        counter_failed     : Counter   = metrics_pool.get_or_create(func_name, MetricTypeEnum.COUNTER_FAILED)

        @histogram_duration.time()

        @histogram_duration.labels(**labels).time()

        def inner_wrapper(self, *args, **kwargs):

            counter_started.inc()

            counter_started.labels(**labels).inc()

            try:

                reply = func(self, *args, **kwargs)

                counter_completed.inc()

                counter_completed.labels(**labels).inc()

                return reply

            except KeyboardInterrupt:   # pylint: disable=try-except-raise

                raise

            except Exception:           # pylint: disable=broad-except

                counter_failed.inc()

                counter_failed.labels(**labels).inc()

        return inner_wrapper

    return outer_wrapper

EXCEPTION_RATIO = 0.05

METRICS_POOL = MetricsPool()

METRICS_POOL = MetricsPool(labels={'driver': 'dummy'})

class DummyDeviceDriver:

    def __init__(self) -> None:

kubectl port-forward -n monitoring service/alertmanager-main --address 0.0.0.0 9093:9093

```

terminal 4 (tun test_set):

```

export PYTHONPATH=/home/tfs/tfs-ctrl/src

python -m common.perf_eval_method_wrapper.tests.test_set

```

- log into grafana:

  - 127.0.0.1:3000

  - admin/admin

# See the License for the specific language governing permissions and

# limitations under the License.

import grpc, logging, time

from common.rpc_method_wrapper.Decorator import create_metrics, safe_and_metered_rpc_method

import logging, random

from prometheus_client import start_http_server

from .DummyDeviceDriver import DummyDeviceDriver

logging.basicConfig(level=logging.DEBUG)

LOGGER = logging.getLogger(__name__)

def test_database_instantiation():

    SERVICE_NAME = 'Context'

    METHOD_NAMES = [

        'ListContextIds',  'ListContexts',   'GetContext',  'SetContext',  'RemoveContext',  'GetContextEvents',

        'ListTopologyIds', 'ListTopologies', 'GetTopology', 'SetTopology', 'RemoveTopology', 'GetTopologyEvents',

        'ListDeviceIds',   'ListDevices',    'GetDevice',   'SetDevice',   'RemoveDevice',   'GetDeviceEvents',

        'ListLinkIds',     'ListLinks',      'GetLink',     'SetLink',     'RemoveLink',     'GetLinkEvents',

        'ListServiceIds',  'ListServices',   'GetService',  'SetService',  'RemoveService',  'GetServiceEvents',

    ]

    METRICS = create_metrics(SERVICE_NAME, METHOD_NAMES)

    class TestServiceServicerImpl:

        @safe_and_metered_rpc_method(METRICS, LOGGER)

        def GetTopology(self, request, grpc_context : grpc.ServicerContext):

            print('doing funny things')

            time.sleep(0.1)

            return 'done'

    tssi = TestServiceServicerImpl()

    tssi.GetTopology(1, 2)

    for metric_name,metric in METRICS.items():

        if 'GETTOPOLOGY_' not in metric_name: continue

        print(metric_name, metric._child_samples()) # pylint: disable=protected-access

import random

from prometheus_client import start_http_server

from DummyDeviceDriver import DummyDeviceDriver

def main():

    # Start up the server to expose the metrics

    start_http_server(8000)

# If not already set, disable skip-build flag.

# If TFS_SKIP_BUILD is "YES", the containers are not rebuilt-retagged-repushed and existing ones are used.

export TFS_SKIP_BUILD=${TFS_SKIP_BUILD:-"YES"}

export TFS_SKIP_BUILD="NO" #${TFS_SKIP_BUILD:-"YES"}