Promethues Exporter

#!/usr/bin/env python # -*- coding: utf-8 -*- import json import sys import prometheus_client import requests from flask import Flask, Response from prometheus_client import CollectorRegistry, Gauge app = Flask(__name__) registry = CollectorRegistry() gauge_QueueMetrics = Gauge("resourcemanager_metrics_queue", "resourcemanager metric queue data", ['instance', 'queuename', 'gauge_queue'], registry=registry) gauge_RpcActivityForPort8032 = Gauge("resourcemanager_metrics_rpcactivityforport8032", "resourcemanager metric rpcactivityforport8032 data", ['instance', 'gauge_rpcactivityforport8032'], registry=registry) gauge_RpcDetailedActivityForPort8032 = Gauge("resourcemanager_metrics_rpcdetailedactivityforport8032", "resourcemanager metric rpcdetailedactivityforport8032 data", ['instance', 'gauge_rpcdetailedactivityforport8032'], registry=registry) gauge_RpcActivityForPort8033 = Gauge("resourcemanager_metrics_rpcactivityforport8033", "resourcemanager metric rpcactivityforport8033 data", ['instance', 'gauge_rpcactivityforport8033'], registry=registry) gauge_RpcDetailedActivityForPort8033 = Gauge("resourcemanager_metrics_rpcdetailedactivityforport8033", "resourcemanager metric rpcdetailedactivityforport8033 data", ['instance', 'gauge_rpcdetailedactivityforport8033'], registry=registry) gauge_RpcActivityForPort8031 = Gauge("resourcemanager_metrics_rpcactivityforport8031", "resourcemanager metric rpcactivityforport8031 data", ['instance', 'gauge_rpcactivityforport8031'], registry=registry) gauge_RpcDetailedActivityForPort8031 = Gauge("resourcemanager_metrics_rpcdetailedactivityforport8031", "resourcemanager metric rpcdetailedactivityforport8031 data", ['instance', 'gauge_rpcdetailedactivityforport8031'], registry=registry) gauge_RpcActivityForPort8030 = Gauge("resourcemanager_metrics_rpcactivityforport8030", "resourcemanager metric rpcactivityforport8030 data", ['instance', 'gauge_rpcactivityforport8030'], registry=registry) gauge_RpcDetailedActivityForPort8030 = Gauge("resourcemanager_metrics_rpcdetailedactivityforport8030", "resourcemanager metric rpcdetailedactivityforport8030 data", ['instance', 'gauge_rpcdetailedactivityforport8030'], registry=registry) gauge_MetricsSystem = Gauge("resourcemanager_metrics_metricssystem", "resourcemanager metric metricssystem data", ['instance', 'gauge_metricssystem'], registry=registry) gauge_JvmMetrics = Gauge("resourcemanager_metrics_jvmmetrics", "resourcemanager metric jvmmetrics data", ['instance', 'gauge_jvmmetrics'], registry=registry) gauge_FSOpDurations = Gauge("resourcemanager_metrics_fsopdurations", "resourcemanager metric fsopdurations data", ['instance', 'gauge_fsopdurations'], registry=registry) gauge_ClusterMetrics = Gauge("resourcemanager_metrics_clustermetrics", "resourcemanager metric clustermetrics data", ['instance', 'gauge_clustermetrics'], registry=registry) # 获取活动的RM def getActiveRM(master1, master2): url1 = 'http://{0}/ws/v1/cluster/info'.format(master1) url2 = 'http://{0}/ws/v1/cluster/info'.format(master2) # Json格式数据解析成Python；json.dumps()方法将Python格式数据解析成Json response = requests.get(url1, timeout=3) data1 = response.json() # print(master1 + " resourcemanager state is :" + data1['clusterInfo']['haState']) # Json格式数据解析成Python；json.dumps()方法将Python格式数据解析成Json response = requests.get(url2, timeout=3) data2 = response.json() # print(master2 + " resourcemanager state is :" + data2['clusterInfo']['haState']) if data1['clusterInfo']['haState'] == 'ACTIVE': activemaster = master1 elif data2['clusterInfo']['haState'] == 'ACTIVE': activemaster = master2 else: raise Exception("on active resourcemanger in %s,%s " % (master1, master2)) return activemaster # Hadoop:service=ResourceManager,name=QueueMetrics,q0=root,q1=* def get_resouecemanager_metrics_queue(url_receive, queuename): try: url_send = 'http://{0}/jmx?qry=Hadoop:service=ResourceManager,name=QueueMetrics,q0=root,q1={1}'.format(url_receive, queuename) response = requests.get(url_send, timeout=3) data = response.json()["beans"][0] # json解析并按key排序 json_str = json.dumps(data, sort_keys=True) # 将 JSON 对象转换为 Python 字典 resourcemanager_metrics_queue = json.loads(json_str) resourcemanager_metrics_queue_items = resourcemanager_metrics_queue.items() for key, value in resourcemanager_metrics_queue_items: if type(value) == int or type(value) == float: gauge_QueueMetrics.labels(instance, queuename, str(key)).set(value) except Exception as e: print("Exit get_resouecemanager_metrics_queue {0} ".format(queuename) + str(e)) # http://10.0.24.27:8088/ws/v1/cluster/scheduler # 获取队列的名称 def getRMQueue(activemaster): url = 'http://{0}/ws/v1/cluster/scheduler'.format(activemaster) response = requests.get(url, timeout=3) data = response.json() listqueue = data["scheduler"]["schedulerInfo"]["rootQueue"]["childQueues"]["queue"] for index in range(len(listqueue)): subqueue = listqueue[index]["queueName"].split(".")[1] get_resouecemanager_metrics_queue(url_receive=activemaster, queuename=subqueue) # Hadoop:service=ResourceManager,name=RpcActivityForPort8032 def get_resourcemanager_metrics_rpcactivityforport8032(url_receive, name): try: url_send = 'http://{0}/jmx?qry=Hadoop:service=ResourceManager,name={1}'.format(url_receive, name) response = requests.get(url_send, timeout=3) data = response.json()["beans"][0] # json解析并按key排序 json_str = json.dumps(data, sort_keys=True) # 将 JSON 对象转换为 Python 字典 resourcemanager_metrics_rpcactivityforport8032 = json.loads(json_str) resourcemanager_metrics_rpcactivityforport8032_items = resourcemanager_metrics_rpcactivityforport8032.items() for key, value in resourcemanager_metrics_rpcactivityforport8032_items: if type(value) == int or type(value) == float: gauge_RpcActivityForPort8032.labels(instance, str(key)).set(value) except Exception as e: print("Exit get_resourcemanager_metrics_rpcactivityforport8032 " + str(e)) # Hadoop:service=ResourceManager,name=RpcDetailedActivityForPort8032 def get_resourcemanager_metrics_rpcdetailedactivityforport8032(url_receive, name): try: url_send = 'http://{0}/jmx?qry=Hadoop:service=ResourceManager,name={1}'.format(url_receive, name) response = requests.get(url_send, timeout=3) data = response.json()["beans"][0] # json解析并按key排序 json_str = json.dumps(data, sort_keys=True) # 将 JSON 对象转换为 Python 字典 resourcemanager_metrics_rpcdetailedactivityforport8032 = json.loads(json_str) resourcemanager_metrics_rpcdetailedactivityforport8032_items = resourcemanager_metrics_rpcdetailedactivityforport8032.items() for key, value in resourcemanager_metrics_rpcdetailedactivityforport8032_items: if type(value) == int or type(value) == float: gauge_RpcDetailedActivityForPort8032.labels(instance, str(key)).set(value) except Exception as e: print("Exit get_resourcemanager_metrics_rpcdetailedactivityforport8032 " + str(e)) # Hadoop:service=ResourceManager,name=RpcActivityForPort8033 def get_resourcemanager_metrics_rpcactivityforport8033(url_receive, name): try: url_send = 'http://{0}/jmx?qry=Hadoop:service=ResourceManager,name={1}'.format(url_receive, name) response = requests.get(url_send, timeout=3) data = response.json()["beans"][0] # json解析并按key排序 json_str = json.dumps(data, sort_keys=True) # 将 JSO