接口测试接口监控可视化系统

xinxi · 2018年12月27日 · 最后由 LiuZeng520 回复于 2020年03月31日 · 3913 次阅读

前言

在之前公司做过一个项目叫监控 app 首页接口及其下的二级接口的状态码和接口响应时间.

当时的做法是写一个爬虫脚本,请求首页接口成功后返回结果递归解析并且再请求二级接口,大概能请求 200 多个接口,会记录错误状态码和失败原因,最终存在数据库中.

执行策略是运行在 linux 下做定时运行脚本,当错误数量满足一定条件时发邮件报警.

问题

以上就是我离职之前做的事,当时遇到几个问题:

监控是否会误报

通过修改策略减少了误报,已经解决了大部分

是否能监控其他接口

通过把接口写入配置文件,就可以监控其他接口

- 是否能分布式

使用多线程技术,但是也没感觉多块

- 数据怎么展示出来

这个问题在我走之前并没有解决,在领导看来监控就是一个脚本在跑,什么有效数据他都看不到,总不能让领导去看数据库吧.

本文就围绕最后一个问题来探索下解决方案,打造一款接口监控可视化系统.

监控脚本

爬虫

这个监控脚本并没有用爬虫框架,就是使用 python 的 requests 库实现的发送请求.

爬虫的核心是递归,如何从一个接口一直一直的爬.

下面这段代码是一段递归函数,通过指定规则递归解析,
把获取的 url 加入到数组中.

def parse_url(obj):
    """ 遍历解析json的每个url元素， 将url的加入到list中"""
    json_string = json.dumps(obj)
    python_obj = json.loads(json_string)
    try:
        if isinstance(python_obj, dict):
            for key, value in python_obj.items():
                # print key,value
                if key != r'' and re.match(r'^(http|https)://.*', str(value)):
                    # print "PARSED URL:         %s，%s" % (key, value)
                    url = url_local_hander(value, city)
                    sublist.append(url)  # 未过滤重复的二级url
                    # print "FIND URL : " + value
                elif key == r'action' and re.match(r'^(wbmain)://.*', str(value)):
                    # print"ACTION_INFO :     %s, %s" % (value, type(value))
                    content = urllib.unquote(value)
                    get_protocol = content[content.find('{'): content.rfind('}') + len('}')]
                    json_str = str_to_json(get_protocol)
                    finderurl = parse_json(json_str)
                    if finderurl != r'':
                        url = url_local_hander(finderurl, city)
                        sublist.append(url)  # 未过滤重复的二级url
                # 再次迭代查找
                parse_url(value)

        elif isinstance(python_obj, (list, tuple)):
            for el in obj:
                # print el
                parse_url(el)
    except Exception, e:
        # raise e
        print str(e)

断言

断言一个接口请求成功,有几个要素.

状态码
请求耗时
请求结果

请求结果和业务有很大关系,在监控中需要验证每个接口的返回值需要大量的接口文档.

所以暂时先考虑状态码和请求耗时,接口耗时通过设置最大超时时间决定.

elk

之前简单使用了 elk 系统,简直是太香了.做数据可视化再合适不过了.

接入准备

监控脚本接入 elk 有几步操作:
监控脚本写请求 log 日志
docker 启动 elk
配置 logstash 采集请求 log 日志
kibana 设计报表

请求 log 日志

格式如下:

{
    "request_time": "2018-12-26 21:56:10",
    "url": "https://zprecommend.58.com/api/abtest?ptype=appnearbyhomepagedispcatelist",
    "request_code": 404,
    "usdtime": 50.628,
    "interface_name": "首页",
    "issue": "请求超时"
}

需要 url、耗时、状态码、请求时间、请求状态等字段组装成 json 格式写入本地 log 日志.

启动 elk

使用 sebp/elk 这个镜像,会把 elk 三个组件组件打在一起使用.

docker run \
--name elk \
-d \
-e LOGSTASH_START=0 \
-p 5601:5601 \
-p 9200:9200 \
-p 5044:5044 -d \
sebp/elk

启动完成以后过一会儿,浏览器访问: http://127.0.0.1:5601

配置 logstash

elk 集成 logstash 组件,为了更好的隔离环境,单独配置 logstash 镜像.

docker run -it --rm \
-v $PWD/data/:/data/  \
-v $PWD:/conf/  \
registry.docker-cn.com/library/logstash -f /conf/request.conf --verbose

-f 指定使用 logstash 的配置文件.

-v 指定本地文件挂载.

logstash 的配置文件如下:

input {
    file {
        path => "/data/send.log"
        start_position => beginning
        codec => "json"
    }
}
filter {
  date {
        match => ["log_time", "ISO8601"]
        timezone => "Asia/Shanghai"
    }
}
output {
    elasticsearch {
        hosts => ["192.168.1.104:9200"]
        index => "requests-json-%{+YYYY.MM.dd}"
    }
    stdout{  
    codec => rubydebug  
    }
}

start_position 必须配置,当时就是没配置造成一条数据都没展示.

codec 是会把 log 日志自动解析字段,比如上面说响应时间、状态码字段、url 字段.

hosts 是 elasticsearch 的 ip 地址,不能用 localhost.

stdout 中的 codec => rubydebug,会把日志打印出来,这个很有用.

启动以后就会实时加载本地 log 日志并且打印在控制台上.

kibana 设计报表

下图是创建索引以后,默认的柱状数据统计,可以设置展示刷新时间.

下图是创建饼图面板,统计请求状态和接口

下图是创建饼图,统计状态码和接口

下图是创建柱状图,统计接口耗时和接口

使用面板汇总图表,elk 提供很多图表和查询规则,总之能生成一份很炫的汇总数据.

grafana

随着不断发送请求,脚本的宿主机性能也会下降,所以使用 grafana 来监控发送请求数量和宿主机的 cpu、内存等.

启动 grafana

docker run \
  -d \
  -p 3000:3000 \
  --name=grafana \
  -e "GF_SERVER_ROOT_URL=http://grafana.server.name" \
  -e "GF_SECURITY_ADMIN_PASSWORD=secret" \
  grafana/grafana

启动成功后,浏览器访问:http://127.0.0.1:3000

账号:admin 密码:secret