Python?prometheus_client怎么用

這篇文章給大家分享的是有關Python prometheus_client怎么用的內容。小編覺得挺實用的，因此分享給大家做個參考，一起跟隨小編過來看看吧。

Python prometheus-client 安裝

pip install prometheus-client

Python封裝

# encoding: utf-8
from prometheus_client import Counter, Gauge, Summary
from prometheus_client.core import CollectorRegistry
from prometheus_client.exposition import choose_encoder


class Monitor:
    def __init__(self):
    # 注冊收集器&最大耗時map
    self.collector_registry = CollectorRegistry(auto_describe=False)
    self.request_time_max_map = {}

    # 接口調用summary統計
    self.http_request_summary = Summary(name="http_server_requests_seconds",
                                   documentation="Num of request time summary",
                                   labelnames=("method", "code", "uri"),
                                   registry=self.collector_registry)
    # 接口最大耗時統計
    self.http_request_max_cost = Gauge(name="http_server_requests_seconds_max",
                                  documentation="Number of request max cost",
                                  labelnames=("method", "code", "uri"),
                                  registry=self.collector_registry)

    # 請求失敗次數統計
    self.http_request_fail_count = Counter(name="http_server_requests_error",
                                      documentation="Times of request fail in total",
                                      labelnames=("method", "code", "uri"),
                                      registry=self.collector_registry)

    # 模型預測耗時統計
    self.http_request_predict_cost = Counter(name="http_server_requests_seconds_predict",
                                        documentation="Seconds of prediction cost in total",
                                        labelnames=("method", "code", "uri"),
                                        registry=self.collector_registry)
    # 圖片下載耗時統計
    self.http_request_download_cost = Counter(name="http_server_requests_seconds_download",
                                         documentation="Seconds of download cost in total",
                                         labelnames=("method", "code", "uri"),
                                         registry=self.collector_registry)

    # 獲取/metrics結果
    def get_prometheus_metrics_info(self, handler):
        encoder, content_type = choose_encoder(handler.request.headers.get('accept'))
        handler.set_header("Content-Type", content_type)
        handler.write(encoder(self.collector_registry))
        self.reset_request_time_max_map()

    # summary統計
    def set_prometheus_request_summary(self, handler):
        self.http_request_summary.labels(handler.request.method, handler.get_status(), handler.request.path).observe(handler.request.request_time())
        self.set_prometheus_request_max_cost(handler)

    # 自定義summary統計
    def set_prometheus_request_summary_customize(self, method, status, path, cost_time):
        self.http_request_summary.labels(method, status, path).observe(cost_time)
        self.set_prometheus_request_max_cost_customize(method, status, path, cost_time)

    # 失敗統計
    def set_prometheus_request_fail_count(self, handler, amount=1.0):
        self.http_request_fail_count.labels(handler.request.method, handler.get_status(), handler.request.path).inc(amount)

    # 自定義失敗統計
    def set_prometheus_request_fail_count_customize(self, method, status, path, amount=1.0):
        self.http_request_fail_count.labels(method, status, path).inc(amount)

    # 最大耗時統計
    def set_prometheus_request_max_cost(self, handler):
        requset_cost = handler.request.request_time()
        if self.check_request_time_max_map(handler.request.path, requset_cost):
            self.http_request_max_cost.labels(handler.request.method, handler.get_status(), handler.request.path).set(requset_cost)
            self.request_time_max_map[handler.request.path] = requset_cost

    # 自定義最大耗時統計
    def set_prometheus_request_max_cost_customize(self, method, status, path, cost_time):
        if self.check_request_time_max_map(path, cost_time):
            self.http_request_max_cost.labels(method, status, path).set(cost_time)
            self.request_time_max_map[path] = cost_time

    # 預測耗時統計
    def set_prometheus_request_predict_cost(self, handler, amount=1.0):
        self.http_request_predict_cost.labels(handler.request.method, handler.get_status(), handler.request.path).inc(amount)

    # 自定義預測耗時統計
    def set_prometheus_request_predict_cost_customize(self, method, status, path, cost_time):
        self.http_request_predict_cost.labels(method, status, path).inc(cost_time)

    # 下載耗時統計
    def set_prometheus_request_download_cost(self, handler, amount=1.0):
        self.http_request_download_cost.labels(handler.request.method, handler.get_status(), handler.request.path).inc(amount)

    # 自定義下載耗時統計
    def set_prometheus_request_download_cost_customize(self, method, status, path, cost_time):
        self.http_request_download_cost.labels(method, status, path).inc(cost_time)

    # 校驗是否賦值最大耗時map
    def check_request_time_max_map(self, uri, cost):
        if uri not in self.request_time_max_map:
            return True
        if self.request_time_max_map[uri] < cost:
            return True
        return False

    # 重置最大耗時map
    def reset_request_time_max_map(self):
        for key in self.request_time_max_map:
            self.request_time_max_map[key] = 0.0

調用

import tornado
import tornado.ioloop
import tornado.web
import tornado.gen
from datetime import datetime
from tools.monitor import Monitor

global g_monitor

class ClassifierHandler(tornado.web.RequestHandler):
    def post(self):
        # TODO Something you need
        # work....
        # 統計Summary，包括請求次數和每次耗時
        g_monitor.set_prometheus_request_summary(self)
        self.write("OK")


class PingHandler(tornado.web.RequestHandler):
    def head(self):
        print('INFO', datetime.now(), "/ping Head.")
        g_monitor.set_prometheus_request_summary(self)
        self.write("OK")

    def get(self):
        print('INFO', datetime.now(), "/ping Get.")
        g_monitor.set_prometheus_request_summary(self)
        self.write("OK")


class MetricsHandler(tornado.web.RequestHandler):
    def get(self):
        print('INFO', datetime.now(), "/metrics Get.")
		g_monitor.set_prometheus_request_summary(self)
		# 通過Metrics接口返回統計結果
    	g_monitor.get_prometheus_metrics_info(self)
    

def make_app():
    return tornado.web.Application([
        (r"/ping?", PingHandler),
        (r"/metrics?", MetricsHandler),
        (r"/work?", ClassifierHandler)
    ])

if __name__ == "__main__":
    g_monitor = Monitor()
	
	app = make_app()
    app.listen(port)
    tornado.ioloop.IOLoop.current().start()

Metrics返回結果實例

感謝各位的閱讀！關于“Python prometheus_client怎么用”這篇文章就分享到這里了，希望以上內容可以對大家有一定的幫助，讓大家可以學到更多知識，如果覺得文章不錯，可以把它分享出去讓更多的人看到吧！