较新版本的 kibana 前端都有一个 Share -> Short URL 的功能，用起来非常方便。因此想着能否将该功能用代码自动生成，将其集成到告警通知中，让用户收到告警通知的时候可以通过短链接快速定位到错误原因。
查询文档发现官方提供 Shorten URL API 实现锻炼功能，该功能处于实验阶段，没有详细的参数介绍，因此需要自己结合kibana 的url 和查询的字段分析其参数生成模式。笔者多次测试后， 结合shorten_url 和 query 语句实现该功能，分享在这里供有需要的小伙伴们学习。
kibana 官方文档 Shorten URL API 的介绍相当简洁，提供了一个很简单的案例，然而并没有介绍 url 中的参数，因此还需要自己通过开发工具一步步测试验证。
笔者在完成这个demo的时候，也是结合kibana URL、share-> PERMALINK-> Short URL-> Chrome 开发工具, 多次对比才找到处一种比较可靠的方式。
主要过程：
以下为源码实现和测试过程
源码结构如下：
alarm_shorten_url$ tree -L 2
.
├── generate_url.py
├── readme.md
└── rison
├── constants.py
├── decoder.py
├── encoder.py
├── __init__.py
└── utils.py
1）从 [github pifantastic/python-rison](https://github.com/pifantastic/python-rison) 下载rison库， 注意 py3 没有basestring类型，因此需要修改 basestring 为 str，否则会报错;
按需拷贝python-rison 的 rison 目录即可，此处就不再贴一长串代码了；
2）按需获取 index pattern id， 然后组装kibana url 中的参数
vim generate_url.py
#!/usr/bin/pytnon3
# -*- coding:utf-8 -*-
import sys
import requests
from base64 import b64encode
sys.path.append('./rison')
import rison
def get_base64_str(str_text):
str_user_pwd = b64encode(str_text.encode('ascii'))
return str_user_pwd.decode('ascii')
def generate_rison(bool_query):
ret = rison.dumps(bool_query)
return ret
class GenerateShortUrl:
def __init__(self, base_url, user, pwd, time_from, time_to, index_pattern, bool_query):
"""
:param base_url:
:param user:
:param pwd:
:param time_from: 'now-15d'
:param time_to: 'now'
:param index_pattern: 'kibana_sample_data_logs'
:param bool_query:
注： 本函数只验证了一个完整类型的bool查询，
因此bool_query_dict必须是 {"query": {"bool":{ ...must,must_not,should...}}}
"""
self.base_url = base_url
self.user = user
self.pwd = pwd
self.time_from = time_from
self.time_to = time_to
self.index_pattern = index_pattern
self.bool_query = bool_query
def get_index_pattern_id(self):
url = f'{self.base_url}/api/saved_objects/_find?fields=title&fields=type&per_page=10000&type=index-pattern'
headers = {'Authorization': 'Basic ' + get_base64_str(self.user + ':' + self.pwd)}
ret = requests.get(url=url, headers=headers)
for item in (ret.json())['saved_objects']:
if item['attributes']['title'] == self.index_pattern:
return item['id']
return None
def generate_shorten_url(self):
url = f'{self.base_url}/api/shorten_url'
index_pattern_id = self.get_index_pattern_id()
bool_query_rison = generate_rison(self.bool_query)
url_dict = {"url": f"/app/discover#/?_g=(filters:!(),refreshInterval:(pause:!t,value:0),"
f"time:(from:{self.time_from},to:{self.time_to}))&_a=(columns:!(_source),"
f"filters:!({bool_query_rison}),index:'{index_pattern_id}',interval:auto,"
f"query:(language:lucene,query:''),sort:!())"}
headers = {'Authorization': 'Basic ' + get_base64_str(self.user + ':' + self.pwd), 'kbn-xsrf': 'reporting'}
ret = requests.post(url=url, json=url_dict, headers=headers)
ret_dict = ret.json()
return f'{self.base_url}/goto/' + ret_dict['urlId']
def test_case01():
base_url = 'http://127.0.0.1:5601'
user = "elastic"
pwd = "elastic"
index_pattern_name = 'kibana_sample_data_logs'
bool_query = {
"query": {
"bool": {
"must": [
{
"term": {
"tags.keyword": {
"value": "error"
}
}
},
{
"term": {
"response.keyword": {
"value": "200"
}
}
}
]
}
}
}
time_from = 'now-15d'
time_to = 'now'
gen_url_obj = GenerateShortUrl(base_url, user, pwd, time_from, time_to, index_pattern_name, bool_query)
index_pattern_id = gen_url_obj.get_index_pattern_id()
short_url = gen_url_obj.generate_shorten_url()
print(f'{index_pattern_name} id= {index_pattern_id}\n{short_url}')
if __name__ == '__main__':
test_case01()
执行输出
kibana 前端
点击链接后会自动跳转到查询界面，可以在filter 处看到代码中的query语句，如下图所示：
{"statusCode":400,"error":"Bad Request","message":"Request must contain a kbn-xsrf header."}
headers = {'Authorization': 'Basic ' + get_base64_str(USER + ':' + PWD), 'kbn-xsrf': 'reporting'}