Prometheus Metrics Exporter

metrics_exporter.py

import os
from prometheus_client import start_http_server
from prometheus_client import Histogram, Summary, Counter
from app_io.constants import PROMETHEUS_PORT

class MetricType:
    SUMMARY='SUMMARY'
    HISTOGRAM='HISTOGRAM'
    COUNTER='COUNTER'

class MetricInfoObject:

    def __init__(self, **kwargs):

        self.metric_instance = kwargs.get('metric_instance')
        self.metric_type = kwargs.get('metric_type')
        self.metric_name = kwargs.get('metric_name')
        self.metric_description = kwargs.get('metric_description')
        self.requestor_name = kwargs.get('requestor_name')

    def check_equality(self, metric_name, requestor_name, metric_type):
        return self.metric_name == metric_name and self.requestor_name == requestor_name and self.metric_type == metric_type

    @property
    def complete_metric_name(self):
        return f'search_service_{self.metric_name}_{self.metric_type}_{self.requestor_name}'

    def _create_metric(self):
        if self.metric_type == MetricType.SUMMARY:
            self.metric_instance = Summary(self.complete_metric_name, self.description)
        elif self.metric_type == MetricType.HISTOGRAM:
            self.metric_instance = Histogram(self.complete_metric_name, self.description)
        elif self.metric_type == MetricType.COUNTER:
            self.metric_instance = Counter(self.complete_metric_name, self.description)
        else:
            raise Exception(f'Metric Type Not defined {self.metric_type}')

    @staticmethod
    def get_instance(metric_type, name, description, entity_name):
        curr_obj = MetricInfoObject()
        curr_obj.metric_type = metric_type
        curr_obj.metric_name = name
        curr_obj.description = description
        curr_obj.requestor_name = entity_name
        curr_obj._create_metric()
        return curr_obj

class MetricsProvider(object):

    def __new__(cls):
         if not hasattr(cls, '_instance'):
             orig = super(MetricsProvider, cls)
             cls._instance = orig.__new__(cls)
             cls.initiated = True
         return cls._instance

    def __init__(self):
        if self.initiated == True:
            self.metrics_info_dict = {}
            self.initiated = False
            try:
                self.start_metrics_server()
            except:
                pass

    def get_metric_obj(self, metric_type, name, description, entity_name):
        info_obj = self.check_for_prev_instance(metric_type, name, entity_name)

        if info_obj:
            return info_obj.metric_instance

        metric_obj = MetricInfoObject.get_instance(metric_type, name, description, entity_name)

        if not self.metrics_info_dict.get(entity_name):
            self.metrics_info_dict[entity_name] = []

        self.metrics_info_dict[entity_name].append(metric_obj)

        return metric_obj.metric_instance


    def check_for_prev_instance(self, metric_type, name, entity_name):
        metric_list = self.metrics_info_dict.get(entity_name, [])

        for info_obj in metric_list:
            if info_obj.check_equality(name, entity_name, metric_type):
                return info_obj
    def start_metrics_server(self):
        start_http_server(PROMETHEUS_PORT)