Created
May 1, 2023 21:06
-
-
Save avneesh91/8a7901e30614d96ee2f5552102e71967 to your computer and use it in GitHub Desktop.
Prometheus Metrics Exporter
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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