1-31 3 views
基类
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 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 [] |
接口
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 |
#!/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 感谢!

大佬好 能加个联系方式吗