Skip to content

Instantly share code, notes, and snippets.

@avneesh91
Created May 1, 2023 21:06
Show Gist options
  • Save avneesh91/8a7901e30614d96ee2f5552102e71967 to your computer and use it in GitHub Desktop.
Save avneesh91/8a7901e30614d96ee2f5552102e71967 to your computer and use it in GitHub Desktop.
Prometheus Metrics Exporter
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)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment