封装Grafana，获取监控数据

1-31 3 views

基类

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# @Author         : Eric Winn
# @Email          : eng.eric.winn@gmail.com
# @Time           : 2020/1/30 1:34 PM
# @Version        : 1.0
# @File           : grafana_api
# @Software       : PyCharm

import json
import logging

import requests

logger = logging.getLogger(__file__)

clients = {}


class GrafanaAPI(object):
    """
    Grafana对接基类
    使用用户名和密码登录Grafana，使用Grafana的Query Range获取监控采集数据
    """

    def __init__(self, host=None, username=None, passport=None, email=None):
        """

        :param host:
        :param username:
        :param passport:
        :param email:
        """
        try:
            self.session = requests.session()
            self.username = username
            self.password = passport
            self.email = email or ''
            self.host = host or 'https://grafana.itnotebooks.com'
            self.login_url = f'{self.host}/login'
            self.request = None

        except Exception as e:
            logger.error(e, exc_info=True)

    def login(self):
        """
        登录Grafana
        :return:
        """

        # 调用Grafana Login接口登录
        ret = self.session.post(self.login_url, headers={'Content-Type': 'application/json'},
                                data=json.dumps({
                                    "user": self.username,
                                    "password": self.password,
                                    "email": self.email
                                }))
        # 判断否登录成功
        if ret.status_code != 200:
            logger.error(ret.text)

            return False

    def alive(self):
        """
        检测当前是否处于登录状态

        :return:
        """
        ret = self.session.get(f'{self.host}/api/user')
        if ret.status_code == 200:
            logger.debug(ret.json())
            return True
        else:
            return False

    def get(self, url, params):
        """
        调用API获取监控数据
        :param url:
        :param params:
        :return:
        """
        # 拼接整个请求的query
        params_encode = f'query={params.get("query")}&start={params.get("start")}&end={params.get("end")}&step={params.get("step")}'
        # 将域名、接口和params合并成一条request url
        ret = self.session.get(f'{self.host}{url}?{params_encode}')
        if ret.status_code == 200:

            # 如果返回的数据不是json格式，就返回空
            try:
                return ret.json()
            except Exception as e:
                logger.error(e, exc_info=True)
                return []

        elif ret.status_code == 401:
            # 如果认证过期，重新认证
            self.login()
            # 再次调用
            return self.get(url, params)
        else:
            logger.error(ret.text)
            return []

100

101

102

103

104

105

#!/usr/bin/env python3

# -*- coding: utf-8 -*-

# @Author : Eric Winn

# @Email : eng.eric.winn@gmail.com

# @Time : 2020/1/30 1:34 PM

# @Version : 1.0

# @File : grafana_api

# @Software : PyCharm

import json

import logging

import requests

logger = logging.getLogger(__file__)

clients = {}

class GrafanaAPI(object):

"""

Grafana对接基类

使用用户名和密码登录Grafana，使用Grafana的Query Range获取监控采集数据

"""

def __init__(self, host=None, username=None, passport=None, email=None):

"""

:param host:

:param username:

:param passport:

:param email:

"""

try:

self.session = requests.session()

self.username = username

self.password = passport

self.email = email or ''

self.host = host or 'https://grafana.itnotebooks.com'

self.login_url = f'{self.host}/login'

self.request = None

except Exception as e:

logger.error(e, exc_info=True)

def login(self):

"""

登录Grafana

:return:

"""

# 调用Grafana Login接口登录

ret = self.session.post(self.login_url, headers={'Content-Type': 'application/json'},

data=json.dumps({

"user": self.username,

"password": self.password,

"email": self.email

}))

# 判断否登录成功

if ret.status_code != 200:

logger.error(ret.text)

return False

def alive(self):

"""

检测当前是否处于登录状态

:return:

"""

ret = self.session.get(f'{self.host}/api/user')

if ret.status_code == 200:

logger.debug(ret.json())

return True

else:

return False

def get(self, url, params):

"""

调用API获取监控数据

:param url:

:param params:

:return:

"""

# 拼接整个请求的query

params_encode = f'query={params.get("query")}&start={params.get("start")}&end={params.get("end")}&step={params.get("step")}'

# 将域名、接口和params合并成一条request url

ret = self.session.get(f'{self.host}{url}?{params_encode}')

if ret.status_code == 200:

# 如果返回的数据不是json格式，就返回空

try:

return ret.json()

except Exception as e:

logger.error(e, exc_info=True)

return []

elif ret.status_code == 401:

# 如果认证过期，重新认证

self.login()

# 再次调用

return self.get(url, params)

else:

logger.error(ret.text)

return []

接口

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# @Author         : Eric Winn
# @Email          : eng.eric.winn@gmail.com
# @Time           : 2020/1/30 1:34 PM
# @Version        : 1.0
# @File           : api
# @Software       : PyCharm

from rest_framework.views import APIView
from .hands import IsValidUser
from .grafana_api import GrafanaAPI, clients


class GrafanaQueryRangeAPI(APIView):
    """
    Grafana数据查询接口，默认查询近15分钟的数据，步长为1分钟

    :param request:
        {
            "api": "/api/datasources/proxy/98/api/v1/query_range",
            "query": "sum(rate(http_call_request_count%7Bjob%3D~%22app-index%7Capp-index%22%2Cendpoint%3D~%22%5C%5C%5BPOST%5C%5C%5D%2Fice%2Fflow%2Fproduct%2Fdetail%22%2C%20instance%3D~%22.*%22%2C%20succeeded%3D%22false%22%7D%5B1m%5D))%20by%20(endpoint)",
            "start": "1590083760",  # Unix时间戳
            "end": "1590085560",    # Unix时间戳
            "step": "15",
        }
    :return
    """
    http_method_names = ['get']
    permission_classes = (IsValidUser,)

    def get(self, request):
        data = request.GET
        query = 'sum(rate(http_call_request_count%7Bjob%3D~%22app-index%7Capp-index%22%2Cendpoint%3D~%22%5C%5C%5BPOST%5C%5C%5D%2Fice%2Fflow%2Fproduct%2Fdetail%22%2C%20instance%3D~%22.*%22%2C%20succeeded%3D%22false%22%7D%5B1m%5D))%20by%20(endpoint)'

        current_time = datetime.datetime.now()
        start = data.get('start', int((current_time + datetime.timedelta(minutes=-15)).timestamp()))
        end = data.get('end', int(current_time.timestamp()))
        step = data.get('step', '15')
        query = data.get('query', '') or query
        api = data.get('api', '') or '/api/datasources/proxy/98/api/v1/query_range'

        # 判断当前用户请求是否已登录过grafana
        # 通过容器来管理登录请求的问题，用用户session key做为唯一标识符
        if clients.get(request.session.session_key):
            worker = clients.get(request.session.session_key)
            grafana = worker['grafana']
        else:
            # 新的登录，用户未请求过grafana
            worker = clients.setdefault(request.session.session_key, {})
            grafana = GrafanaAPI()
            grafana.login()
            worker['grafana'] = grafana

        params = {
            'query': query,
            'start': start,
            'end': end,
            'step': step
        }
        ret = grafana.get(api, params)
        return Response(ret)

#!/usr/bin/env python3

# -*- coding: utf-8 -*-

# @Author : Eric Winn

# @Email : eng.eric.winn@gmail.com

# @Time : 2020/1/30 1:34 PM

# @Version : 1.0

# @File : api

# @Software : PyCharm

from rest_framework.views import APIView

from .hands import IsValidUser

from .grafana_api import GrafanaAPI, clients

class GrafanaQueryRangeAPI(APIView):

"""

Grafana数据查询接口，默认查询近15分钟的数据，步长为1分钟

:param request:

{

"api": "/api/datasources/proxy/98/api/v1/query_range",

"query": "sum(rate(http_call_request_count%7Bjob%3D~%22app-index%7Capp-index%22%2Cendpoint%3D~%22%5C%5C%5BPOST%5C%5C%5D%2Fice%2Fflow%2Fproduct%2Fdetail%22%2C%20instance%3D~%22.*%22%2C%20succeeded%3D%22false%22%7D%5B1m%5D))%20by%20(endpoint)",

"start": "1590083760", # Unix时间戳

"end": "1590085560", # Unix时间戳

"step": "15",

}

:return

"""

http_method_names = ['get']

permission_classes = (IsValidUser,)

def get(self, request):

data = request.GET

query = 'sum(rate(http_call_request_count%7Bjob%3D~%22app-index%7Capp-index%22%2Cendpoint%3D~%22%5C%5C%5BPOST%5C%5C%5D%2Fice%2Fflow%2Fproduct%2Fdetail%22%2C%20instance%3D~%22.*%22%2C%20succeeded%3D%22false%22%7D%5B1m%5D))%20by%20(endpoint)'

current_time = datetime.datetime.now()

start = data.get('start', int((current_time + datetime.timedelta(minutes=-15)).timestamp()))

end = data.get('end', int(current_time.timestamp()))

step = data.get('step', '15')

query = data.get('query', '') or query

api = data.get('api', '') or '/api/datasources/proxy/98/api/v1/query_range'

# 判断当前用户请求是否已登录过grafana

# 通过容器来管理登录请求的问题，用用户session key做为唯一标识符

if clients.get(request.session.session_key):

worker = clients.get(request.session.session_key)

grafana = worker['grafana']

else:

# 新的登录，用户未请求过grafana

worker = clients.setdefault(request.session.session_key, {})

grafana = GrafanaAPI()

grafana.login()

worker['grafana'] = grafana

params = {

'query': query,

'start': start,

'end': end,

'step': step

}

ret = grafana.get(api, params)

return Response(ret)

前端ECharts

如果想赏钱，可以用微信扫描下面的二维码，一来能刺激我写博客的欲望，二来好维护云主机的费用；另外再次标注博客原地址 itnotebooks.com 感谢！

版权属于: Eric

原文地址: https://www.itnotebooks.com/?p=1475

转载时必须以链接形式注明原始出处及本声明。

封装Grafana，获取监控数据

封装Grafana，获取监控数据

基类

接口

前端ECharts

1 条评论

回复 airmir 取消回复