Python調(diào)用Prometheus監(jiān)控?cái)?shù)據(jù)并計(jì)算

第Python調(diào)用Prometheus監(jiān)控?cái)?shù)據(jù)并計(jì)算目錄Prometheus是什么Prometheus基礎(chǔ)概念什么是時(shí)間序列數(shù)據(jù)什么是targets(目標(biāo))什么是metrics(指標(biāo))什么是PromQL(函數(shù)式查詢語(yǔ)言)如何監(jiān)控遠(yuǎn)程Linux主機(jī)PrometheusHTTPAPI支持的API認(rèn)證方法數(shù)據(jù)返回格式數(shù)據(jù)寫入監(jiān)控?cái)?shù)據(jù)查詢什么是Grafana工作使用場(chǎng)景CPU峰值計(jì)算CPU均值計(jì)算內(nèi)存峰值計(jì)算內(nèi)存均值計(jì)算導(dǎo)出excel參考鏈接：

Prometheus是什么

Prometheus是一套開源監(jiān)控系統(tǒng)和告警為一體，由go語(yǔ)言(golang)開發(fā)，是監(jiān)控+報(bào)警+時(shí)間序列數(shù)

據(jù)庫(kù)的組合。適合監(jiān)控docker容器。因?yàn)閗ubernetes(k8s)的流行帶動(dòng)其發(fā)展。

Prometheus的主要特點(diǎn)

多維度數(shù)據(jù)模型，由指標(biāo)名稱和鍵/值對(duì)標(biāo)識(shí)的時(shí)間序列數(shù)據(jù)。

作為一個(gè)時(shí)間序列數(shù)據(jù)庫(kù)，其采集的數(shù)據(jù)會(huì)以文件的形式存儲(chǔ)在本地中。

靈活的查詢語(yǔ)言，PromQL(PrometheusQueryLanguage)函數(shù)式查詢語(yǔ)言。

不依賴分布式存儲(chǔ)，單個(gè)服務(wù)器節(jié)點(diǎn)是自治的。

以HTTP方式，通過(guò)pull模型拉取時(shí)間序列數(shù)據(jù)。

也可以通過(guò)中間網(wǎng)關(guān)支持push模型。

通過(guò)服務(wù)發(fā)現(xiàn)或者靜態(tài)配置，來(lái)發(fā)現(xiàn)目標(biāo)服務(wù)對(duì)象。

支持多種多樣的圖表和界面展示。

Prometheus原理架構(gòu)圖

Prometheus基礎(chǔ)概念

什么是時(shí)間序列數(shù)據(jù)

時(shí)間序列數(shù)據(jù)(TimeSeriesData):按照時(shí)間順序記錄系統(tǒng)、設(shè)備狀態(tài)變化的數(shù)據(jù)被稱為時(shí)序數(shù)據(jù)。

應(yīng)用的場(chǎng)景很多，如：

無(wú)人駕駛運(yùn)行中記錄的經(jīng)度，緯度，速度，方向，旁邊物體距離等。

某一個(gè)地區(qū)的各車輛的行駛軌跡數(shù)據(jù)。

傳統(tǒng)證券行業(yè)實(shí)時(shí)交易數(shù)據(jù)。

實(shí)時(shí)運(yùn)維監(jiān)控?cái)?shù)據(jù)等。

時(shí)間序列數(shù)據(jù)特點(diǎn)：

性能好、存儲(chǔ)成本低

什么是targets(目標(biāo))

Prometheus是一個(gè)監(jiān)控平臺(tái)，它通過(guò)抓取監(jiān)控目標(biāo)(targets)上的指標(biāo)HTTP端點(diǎn)來(lái)從這些目標(biāo)收集指標(biāo)。

安裝完P(guān)rometheusServer端之后，第一個(gè)targets就是它本身。

具體可以參考官方文檔

什么是metrics(指標(biāo))

Prometheus存在多種不同的監(jiān)控指標(biāo)(Metrics)，在不同的場(chǎng)景下應(yīng)該要選擇不同的Metrics。

Prometheus的merics類型有四種，分別為Counter、Gauge、Summary、Histogram。

Counter：只增不減的計(jì)數(shù)器

Gauge：可增可減的儀表盤

Histogram：分析數(shù)據(jù)分布情況

Summary：使用較少

簡(jiǎn)單了解即可，暫不需要深入理解。

通過(guò)瀏覽器訪問(wèn)http://被監(jiān)控端IP:9100(被監(jiān)控端口)/metrics

就可以查到node_exporter在被監(jiān)控端收集的監(jiān)控信息

什么是PromQL(函數(shù)式查詢語(yǔ)言)

Prometheus內(nèi)置了一個(gè)強(qiáng)大的數(shù)據(jù)查詢語(yǔ)言PromQL。通過(guò)PromQL可以實(shí)現(xiàn)對(duì)監(jiān)控?cái)?shù)據(jù)的查詢、聚合。

同時(shí)PromQL也被應(yīng)用于數(shù)據(jù)可視化(如Grafana)以及告警當(dāng)中。

通過(guò)PromQL可以輕松回答以下問(wèn)題：

在過(guò)去一段時(shí)間中95%應(yīng)用延遲時(shí)間的分布范圍？

預(yù)測(cè)在4小時(shí)后，磁盤空間占用大致會(huì)是什么情況？

CPU占用率前5位的服務(wù)有哪些？(過(guò)濾)

具體查詢細(xì)節(jié)可以參考官方。

如何監(jiān)控遠(yuǎn)程Linux主機(jī)

安裝Prometheus組件其實(shí)很簡(jiǎn)單，下載包--解壓--后臺(tái)啟動(dòng)運(yùn)行即可，不做具體演示。

在遠(yuǎn)程linux主機(jī)(被監(jiān)控端)上安裝node_exporter組件，可看下載地址

下載解壓后，里面就一個(gè)啟動(dòng)命令node_exporter，直接啟動(dòng)即可。

nohup/usr/local/node_exporter/node_exporter/dev/null21

lsof-i:9100

nohup：如果直接啟動(dòng)node_exporter的話，終端關(guān)閉進(jìn)程也會(huì)隨之關(guān)閉，這個(gè)命令幫你解決問(wèn)題。

PrometheusHTTPAPI

Prometheus所有穩(wěn)定的HTTPAPI都在/api/v1路徑下。當(dāng)我們有數(shù)據(jù)查詢需求時(shí)，可以通過(guò)查詢API請(qǐng)求監(jiān)控?cái)?shù)據(jù)，提交數(shù)據(jù)可以使用remotewrite協(xié)議或者Pushgateway的方式。

支持的API

認(rèn)證方法

默認(rèn)開啟認(rèn)證，因此所有的接口都需要認(rèn)證，且所有的認(rèn)證方式都支持BearerToken和BasicAuth。

調(diào)用接口的時(shí)候，我們需要攜帶BasicAuth請(qǐng)求頭的認(rèn)證，否則會(huì)出現(xiàn)401。

BearerToken

BearerToken隨著實(shí)例產(chǎn)生而生成，可以通過(guò)控制臺(tái)進(jìn)行查詢。了解BearerToken更多信息，請(qǐng)參見BearerAuthentication。

BasicAuth

BasicAuth兼容原生PrometheusQuery的認(rèn)證方式，用戶名為用戶的APPID，密碼為bearertoken（實(shí)例產(chǎn)生時(shí)生成），可以通過(guò)控制臺(tái)進(jìn)行查詢。了解BasicAuth更多信息，請(qǐng)參見BasicAuthentication。

數(shù)據(jù)返回格式

所有API的響應(yīng)數(shù)據(jù)格式都為JSON。每一次成功的請(qǐng)求會(huì)返回2xx狀態(tài)碼。

無(wú)效的請(qǐng)求會(huì)返回一個(gè)包含錯(cuò)誤對(duì)象的JSON格式數(shù)據(jù)，同時(shí)也將包含一個(gè)如下表格的狀態(tài)碼：

無(wú)效請(qǐng)求響應(yīng)返回模板如下：

"status":"success"|"error",

"data":data,

//當(dāng)status狀態(tài)為error時(shí)，下面的數(shù)據(jù)將被返回

"errorType":"string",

"error":"string",

//當(dāng)執(zhí)行請(qǐng)求時(shí)有警告信息時(shí)，該字段將被填充返回

"warnings":["string"]

數(shù)據(jù)寫入

運(yùn)維過(guò)程不需要對(duì)數(shù)據(jù)進(jìn)行寫入，所以暫時(shí)不深入理解。

有興趣的同學(xué)可以看看官方文檔

監(jiān)控?cái)?shù)據(jù)查詢

當(dāng)我們有數(shù)據(jù)查詢需求時(shí)，可以通過(guò)查詢API請(qǐng)求監(jiān)控?cái)?shù)據(jù)。

查詢API接口

GET/api/v1/query

POST/api/v1/query

查詢參數(shù)：

query=:Prometheus：查詢表達(dá)式。

time=rfc3339|unix_timestamp：時(shí)間戳，可選。

timeout=：檢測(cè)超時(shí)時(shí)間，可選。默認(rèn)由-query.timeout參數(shù)指定。

簡(jiǎn)單的查詢

查詢當(dāng)前狀態(tài)為up的監(jiān)控主機(jī)：

curl-u"appid:token"'http://IP:PORT/api/v1/queryquery=up'

范圍查詢

GET/api/v1/query_range

POST/api/v1/query_range

根據(jù)時(shí)間范圍查詢需要的數(shù)據(jù)，這也是我們用得最多的場(chǎng)景，

這時(shí)我們需要用到/api/v1/query_range接口，示例如下：

$curl'http://localhost:9090/api/v1/query_rangequery=upstart=2015-07-01T20:10:30.781Zend=2015-07-01T20:11:00.781Zstep=15s'

"status":"success",

"data":{

"resultType":"matrix",

"result":[

"metric":{

"__name__":"up",

"job":"prometheus",

"instance":"localhost:9090"

"values":[

[1435781430.781,"1"],

[1435781445.781,"1"],

[1435781460.781,"1"]

"metric":{

"__name__":"up",

"job":"node",

"instance":"localhost:9091"

"values":[

[1435781430.781,"0"],

[1435781445.781,"0"],

[1435781460.781,"1"]

什么是Grafana

Grafana是一個(gè)開源的度量分析和可視化工具，可以通過(guò)將采集的數(shù)據(jù)分析、查詢，

然后進(jìn)行可視化的展示，并能實(shí)現(xiàn)報(bào)警。

網(wǎng)址:/

使用Grafana連接Prometheus

連接不再做具體演示，操作思路如下：

1.在Grafana服務(wù)器上安裝，下載地址：/grafana/download

2.瀏覽器http://grafana服務(wù)器IP:3000登錄，默認(rèn)賬號(hào)密碼都是admin，就可以登陸了。

3.把Prometheus服務(wù)器收集的數(shù)據(jù)做為一個(gè)數(shù)據(jù)源添加到Grafana，得到Prometheus數(shù)據(jù)。

4.然后為添加好的數(shù)據(jù)源做圖形顯示，最后在dashboard就可以查看到。

操作流程不難，就不講解重點(diǎn)，后面正式開始上查詢腳本。

工作使用場(chǎng)景

工作中需要通過(guò)CPU、內(nèi)存生成資源利用率報(bào)表，可以通過(guò)Prometheus的API寫一個(gè)Python腳本。

可通過(guò)API獲取數(shù)據(jù)，然后再進(jìn)行數(shù)據(jù)排序、過(guò)濾、運(yùn)算、聚合，最后寫入Mysql數(shù)據(jù)庫(kù)。

CPU峰值計(jì)算

取最近一周CPU數(shù)值，再排序取最高的值。

defget_cpu_peak(self):

CPU取最近一周所有數(shù)值，再排序取最高的值,TOP1

:return:{'IP':value}

#拼接URL

pre_url=self.server_ip+'/api/v1/query_rangequery='

expr='100-(avgby(instance)(irate(node_cpu_seconds_total{mode="idle"}[5m]))*100)'\

'start=%send=%sstep=300'%(self.time_list[0],self.time_list[-1]-1)

url=pre_url+expr

#print(url)

result={}

#請(qǐng)求URL后將Json數(shù)據(jù)轉(zhuǎn)為字典對(duì)象

res=json.loads(requests.post(url=url,headers=self.headers).content.decode('utf8','ignore'))

#print(data)

#循環(huán)取出字典里每個(gè)IP的values，排序取最高值，最后存入result字典

fordainres.get('data').get('result'):

values=da.get('values')

cpu_values=[float(v[1])forvinvalues]#取出數(shù)值并存入列表

#取出IP并消除端口號(hào)

ip=da.get('metric').get('instance')

ip=ip[:ip.index(':')]if':'inipelseip

#ifip=='81':

#print(ip)

#cpu_peak=round(sorted(cpu_values,reverse=True)[0],2)

cpu_peak=sorted(cpu_values,reverse=True)[0]

#取出IP和最高值之后，寫入字典

result[ip]=cpu_peak

#print(result)

returnresult

CPU均值計(jì)算

取最近一周CPU每一天的TOP20除以20得到當(dāng)時(shí)忙時(shí)平均值，

再將7天平均值的和除以n，得到時(shí)間范圍內(nèi)忙時(shí)平均值。

defget_cpu_average(self):

CPU忙時(shí)平均值：取最近一周CPU數(shù)據(jù)，每一天的TOP20除以20得到忙時(shí)平均值；

再將一周得到的忙時(shí)平均值相加，再除以7，得到時(shí)間范圍內(nèi)一周的忙時(shí)平均值。

:return:

cpu_average={}

fortinrange(len(self.time_list)):

ift+1len(self.time_list):

start_time=self.time_list[t]

end_time=self.time_list[t+1]

#print(start_time,end_time)

#拼接URL

pre_url=server_ip+'/api/v1/query_rangequery='

expr='100-(avgby(instance)(irate(node_cpu_seconds_total{mode="idle"}[5m]))*100)'\

'start=%send=%sstep=300'%(start_time,end_time-1)

url=pre_url+expr

#print(url)

#請(qǐng)求接口數(shù)據(jù)

data=json.loads(requests.post(url=url,headers=self.headers).content.decode('utf8','ignore'))

fordaindata.get('data').get('result'):#循環(huán)拿到result數(shù)據(jù)

values=da.get('values')

cpu_load=[float(v[1])forvinvalues]#循環(huán)拿到values里面的所有值

ip=da.get('metric').get('instance')#拿到instance里面的ip

ip=ip[:ip.index(':')]if':'inipelseip#去除個(gè)別后面帶的端口號(hào)

#avg_cup_load=sum(sorted(cpu_load,reverse=True)[:20])/20

#取top20%再除以20%，得出top20%的平均值

#avg_cup_load=round(sum(sorted(cpu_load,reverse=True)[:round(len(cpu_load)*0.2)])/round(len(cpu_load)*0.2),2)

#倒序后取前面20%除以個(gè)數(shù)，得到前20%的平均值

avg_cup_load=sum(sorted(cpu_load,reverse=True)[:round(len(cpu_load)*0.2)])/round(len(cpu_load)*0.2)

#print(avg_cup_load)

#將計(jì)算后的數(shù)據(jù)以ip為key寫入字典

ifcpu_average.get(ip):

cpu_average[ip].append(avg_cup_load)

else:

cpu_average[ip]=[avg_cup_load]

#每日top20的平均值累加，共7天的再除以7

fork,vincpu_average.items():

#cpu_average[k]=round(sum(v)/7,2)

cpu_average[k]=sum(v)

#print(cpu_average)

returncpu_average

內(nèi)存峰值計(jì)算

取7天內(nèi)存數(shù)值，排序后取最高峰值TOP1

defget_mem_peak(self):

內(nèi)存單臺(tái)峰值：取7天內(nèi)存最高峰值TOP1

:return:7天內(nèi)存使用率最高峰值

pre_url=self.server_ip+'/api/v1/query_rangequery='

#expr='(node_memory_MemTotal_bytes-(node_memory_MemFree_bytes+node_memory_Buffers_bytes+node_memory_Cached_bytes))/node_memory_MemTotal_bytes*100start=%send=%sstep=300'%(start_time,end_time)

#字符太長(zhǎng)會(huì)導(dǎo)致報(bào)錯(cuò)，所以這里進(jìn)行拆分字段計(jì)算

expr_MenTotal='node_memory_MemTotal_bytesstart=%send=%sstep=300'%(self.time_list[0],self.time_list[-1]-1)

expr_MemFree='node_memory_MemFree_bytesstart=%send=%sstep=300'%(self.time_list[0],self.time_list[-1]-1)

expr_Buffers='node_memory_Buffers_bytesstart=%send=%sstep=300'%(self.time_list[0],self.time_list[-1]-1)

expr_Cached='node_memory_Cached_bytesstart=%send=%sstep=300'%(self.time_list[0],self.time_list[-1]-1)

result={}

#循環(huán)分別取出總內(nèi)存、可用內(nèi)存、Buffer塊、緩存塊四個(gè)字段

forurinexpr_MenTotal,expr_MemFree,expr_Buffers,expr_Cached:

url=pre_url+ur

data=json.loads(requests.post(url=url,headers=self.headers).content.decode('utf8','ignore'))

ip_dict={}

#循環(huán)單個(gè)字段所有值

fordaindata.get('data').get('result'):

ip=da.get('metric').get('instance')

ip=ip[:ip.index(':')]if':'inipelseip

#ifip!='2':

#continue

ifip_dict.get(ip):#過(guò)濾重復(fù)的ip，重復(fù)ip會(huì)導(dǎo)致計(jì)算多次

#print("重復(fù)ip：%s"%(ip))

continue

values=da.get('values')

#將列表里的值轉(zhuǎn)為字典方便計(jì)算

values_dict={}

forvinvalues:

values_dict[str(v[0])]=v[1]

#標(biāo)記ip存在

ip_dict[ip]=True

#建立列表追加字典

ifresult.get(ip):

result[ip].append(values_dict)

else:

result[ip]=[values_dict]

#print(result)

#對(duì)取出的四個(gè)值進(jìn)行計(jì)算，得出峰值

forip,valuesinresult.items():

values_list=[]

fork,vinvalues[0].items():

try:

values_MenTotal=float(v)

values_MemFree=float(values[1].get(k,0))

values_Buffers=float(values[2].get(k,0))ifvalues[2]else0

values_Cached=float(values[3].get(k,0))ifvalues[3]else0

#如果是0，不參與計(jì)算

ifvalues_MemFree==0.0orvalues_Buffers==0.0orvalues_Cached==0.0:

continue

#values_list.append(round((values_MenTotal-(values_MemFree+values_Buffers+values_Cached))/values_MenTotal*100,2))

#合并后計(jì)算，得出列表

values_list.append((values_MenTotal-(values_MemFree+values_Buffers+values_Cached))/values_MenTotal*100)

#對(duì)得出結(jié)果進(jìn)行排序

result[ip]=sorted(values_list,reverse=True)[0]

exceptExceptionase:

#print(values[0])

logging.exception(e)

#print(result)

returnresult

內(nèi)存均值計(jì)算

先取出7天的日期，根據(jù)多條鏈接循環(huán)取出每天數(shù)據(jù)，排序value取top20除以20，最終7天數(shù)據(jù)再除以7

defget_mem_average(self):

內(nèi)存忙時(shí)平均值：先取出7天的日期，根據(jù)多條鏈接循環(huán)取出每天數(shù)據(jù)，排序value取top20除以20，最終7天數(shù)據(jù)再除以7

:return:

avg_mem_util={}

fortinrange(len(self.time_list)):

ift+1len(self.time_list):

start_time=self.time_list[t]

end_time=self.time_list[t+1]

#根據(jù)多條鏈接循環(huán)取出每天數(shù)據(jù)

pre_url=self.server_ip+'/api/v1/query_rangequery='

#expr='(node_memory_MemTotal_bytes-(node_memory_MemFree_bytes+node_memory_Buffers_bytes+node_memory_Cached_bytes))/node_memory_MemTotal_bytes*100start=%send=%sstep=300'%(start_time,end_time)

expr_MenTotal='node_memory_MemTotal_bytesstart=%send=%sstep=600'%(start_time,end_time-1)

expr_MemFree='node_memory_MemFree_bytesstart=%send=%sstep=600'%(start_time,end_time-1)

expr_Buffers='node_memory_Buffers_bytesstart=%send=%sstep=600'%(start_time,end_time-1)

expr_Cached='node_memory_Cached_bytesstart=%send=%sstep=600'%(start_time,end_time-1)

result={}

#循環(huán)取出四個(gè)字段

forurinexpr_MenTotal,expr_MemFree,expr_Buffers,expr_Cached:

url=pre_url+ur

data=json.loads(requests.post(url=url,headers=self.headers).content.decode('utf8','ignore'))

ip_dict={}

#循環(huán)單個(gè)字段所有值

fordaindata.get('data').get('result'):

ip=da.get('metric').get('instance')

ip=ip[:ip.index(':')]if':'inipelseip

ifip_dict.get(ip):

#print("重復(fù)ip：%s"%(ip))

continue

values=da.get('values')

#將列表里的值轉(zhuǎn)為字典方便計(jì)算

values_dict={}

forvinvalues:

values_dict[str(v[0])]=v[1]

#標(biāo)記ip存在

ip_dict[ip]=True

#建立列表追加字典

ifresult.get(ip):

result[ip].append(values_dict)

else:

result[ip]=[values_dict]

#print(result)

forip,valuesinresult.items():

values_list=[]

fork,vinvalues[0].items():

try:

values_MenTotal=float(v)

values_MemFree=float(values[1].get(k,0))ifvalues[1]else0

values_Buffers=float(values[2].get(k,0))ifvalues[2]else0

values_Cached=float(values[3].get(k,0))ifvalues[3]else0

ifvalues_MemFree==0.0orvalues_Buffers==0.0orvalues_Cached==0.0:

continue

value_calc=(values_MenTotal-(values_MemFree+values_Buffers+values_Cached))/values_MenTotal*100

ifvalue_calc!=float(0):

values_list.append(value_calc)

exceptExceptionase:

print(values[0])

#logging.exception(e)

continue

#排序value取top20除以20

#avg_mem=round(sum(sorted(values_list,reverse=True)[:round(len(values_list)*0.2)])/round(len(values_list)*0.2),2)

try:

avg_mem=sum(sorted(values_list,reverse=True)[:round(len(values_list)*0.2)])/round(len(values_list)*0.2)

exceptExceptiona