实现目的:不需要打开Grafana查看监控图,自动根据告警将图片保存并发送
1)获取Grafana Service Token

2)获取面板ID,并构建图形URL

GRAFANA_API_URL = os.environ['GRAFANA_API_URL']
GRAFANA_SERVICE_TOKEN = GetDecryptToken(os.environ['GRAFANA_SERVICE_TOKEN'])
GRAFANA_SERVICE_USER = GetDecryptToken(os.environ['GRAFANA_SERVICE_USER'])
DNS_DASHBOARD_LIST = {
"DNS1": "/render/d-solo/3hfXaZV7k/dns?orgId=1&panelId=120&tz=Asia%2FShanghai",
"DNS2": "/render/d-solo/3hfXaZV7k/dns?orgId=1&panelId=108&tz=Asia%2FShanghai"
}
def Combine_Images_Vertically(images_list=[], output_path_filename=None):
images = [Image.open(image) for image in images_list]
widths, heights = zip(*(i.size for i in images))
new_width = max(widths)
new_height = sum(heights)
new_image = Image.new('RGB', (new_width, new_height))
y_offset = 0
for img in images:
new_image.paste(img, (0, y_offset))
y_offset += img.height
new_image.save(output_path_filename)
if os.path.exists(output_path_filename):
return True
else:
return False
def Show_Grafana_Dashboard(dashboard_url_list=None, dashboard_name=None, from_time=None, to_time=None, last_minutes=5, group_id=None, erp=None, image_combine=False):
IMG_HEIGHT = 500
IMG_WIDTH = 1500
headers = {
"Authorization" : f"Bearer {GRAFANA_SERVICE_TOKEN}"
}
if from_time is not None and to_time is not None:
current_year = datetime.now().year
from_time = f"{current_year}/{from_time}"
to_time = f"{current_year}/{to_time}"
from_time_obj = datetime.strptime(from_time, '%Y/%m/%d %H:%M')
to_time_obj = datetime.strptime(to_time, '%Y/%m/%d %H:%M')
from_time_unix = int(from_time_obj.timestamp() * 1000)
to_time_unix = int(to_time_obj.timestamp() * 1000)
elif last_minutes is not None and 0 < last_minutes < 60:
to_time = datetime.now()
from_time = to_time - timedelta(minutes=last_minutes)
from_time_unix = int(from_time.timestamp() * 1000)
to_time_unix = int(to_time.timestamp() * 1000)
rendered_images_fullpath_list = []
rendered_images_filename_list = []
for dashboard_url in dashboard_url_list:
URL = GRAFANA_API_URL + dashboard_url + f"&from={from_time_unix}&to={to_time_unix}&width={IMG_WIDTH}&height={IMG_HEIGHT}"
timestamp = int(time.time())
image_file_name = f"grafana_images_{timestamp}.png"
image_file_path = os.path.join(IMG_WEB_DIR, image_file_name)
file_counter = 1
while os.path.exists(image_file_path):
image_file_name = f"grafana_images_{timestamp}_{file_counter}.png"
image_file_path = os.path.join(IMG_WEB_DIR, image_file_name)
file_counter += 1
try:
response = requests.get(URL, headers=headers, timeout=10)
if response.status_code == 200:
with open(image_file_path, 'wb') as f:
f.write(response.content)
rendered_images_fullpath_list.append(image_file_path)
rendered_images_filename_list.append(image_file_name)
else:
running_logger.error(f'Server Error, {response.status_code}, Grafana获取图片失败')
except Exception as e:
running_logger.error(f'Server Error, {e}, Grafana获取图片失败')
BotSendMessage(erp=erp,group_id=group,msg=f"【ERROR】Grafana API获取失败!")
pass
if rendered_images_fullpath_list is not None:
if image_combine is True:
timestamp_combine = int(time.time())
image_file_name_combine = f"grafana_combine_images_{timestamp}.png"
image_file_path_combine = os.path.join(IMG_WEB_DIR, image_file_name_combine)
file_counter = 1
while os.path.exists(image_file_path_combine):
image_file_name_combine = f"grafana_combine_images_{timestamp}_{file_counter}.png"
image_file_path_combine = os.path.join(IMG_WEB_DIR, image_file_name_combine)
file_counter += 1
if Combine_Images_Vertically(rendered_images_fullpath_list, output_path_filename=image_file_path_combine):
BotSendInteractiveCardsMessage(erp=erp, group_id=group_id, img=image_file_name_combine, title=dashboard_name, deeplink_url=None)
else:
for image_file in rendered_images_filename_list:
BotSendInteractiveCardsMessage(erp=erp, group_id=group_id, img=image_file, title=dashboard_name, deeplink_url=None, deeplink_url_name=None)
dashboard_url_list = []
dashboard_url_list.append(f"{DNS_DASHBOARD_LIST ['DNS1']}&var-IP={ip}")
dashboard_url_list.append(f"{DNS_DASHBOARD_LIST ['DNS2']}&var-IP={ip}")
dashboard_title = f"最近5分钟 {IP} DNS解析峰值 {_peak}{_unit}"
Show_Grafana_Dashboard(dashboard_url_list=dashboard_url_list, dashboard_name=f"{dashboard_title}", last_minutes=5, group_id=group_id, erp=erp, image_combine=True)



