转自极客时间,仅供非商业用途或交流学习使用,如有侵权请联系删除
你好,我是高楼。
这节课我们来聊聊如何改造分布式压测平台。
在第 6 讲,我们已经详细了解了流量工具的选型,我们一起来回顾下全链路流量平台必须具备的能力:
- 能录制线上真实流量;
- 能实现海量数据的并发请求,并覆盖地域性的 CDN 边缘节点;
- 能支持常见协议的请求;
- 对线上尽量应用透明,也就是说无侵入性;
- 避免写请求的脏数据,压测流量能够被识别,方便压测后清理;
- 工具使用简单,能够满足压测实时监控,服务安全保护(过载熔断)。
按照上面这几条能力需求,我还画了一个比较典型的全链路流量平台架构设计图。
在这张架构图中,我把压测平台分为压测 web 管理端、调度服务、压测引擎、监控服务、对象存储、录制服务六大模块。这样的一个全链路流量平台基本上就可以覆盖大部分企业的需求了。
接下来,我会就这里面部分的技术细节进行拆解。因为内容比较多,我会分成两节课讲解。这节课,我们主要看一下对象存储和性能监控模块如何落地。
对象存储
对象存储简单来说就是一种海量、安全、低成本、高可靠的云存储服务,适合存放任意类型的文件。它还支持快速查询、上传下载等功能。通俗来说就是一个文件仓库。
现在大部分的公有云厂商,都提供了自己的对象存储服务,比如阿里云的OSS、华为云的OBS,腾讯云的COS等,我们只需集成云厂商提供的SDK即可访问。而开源产品方面,比较常见的有 Ceph 和 MinIO。
其中,Ceph是一个比较强大的分布式存储系统,但是它整个系统非常复杂,比较重量级,需要花费大力气去维护,很显然与我们的目标不是很符合,所以暂时不考虑。
而MinIO是一个基于Apache License V2.0开源协议的高性能、分布式的对象存储系统,而且兼容亚马逊S3云存储服务,非常适合存储大容量非结构化的数据,比如图片、视频、日志文件等。而且MinIO系统较为轻量级,可以很简单地和其它的应用结合,很显然,气质和我们的流量平台比较符合。
总而言之,如果想自建对象存储服务,而且有能力、规模又比较大的话,采用 Ceph 更好一点。但是我们只是想要一个对象存储,要求没有那么多,所以我们才选择了 MinIO。它的整体结构图如下:
我们设计的上传文件的大致流程你可以看看下面这张图:
知道了大致的流程,接下来看看我们如何落地。
- 搭建 MinIO Server
这里,为了方便调试,我们使用Docker快速搭建 MinIO Server,并设置端口号、容器名。
docker run -d --name minio-server -p 9000:9000 -p 9001:9001 minio/minio server /data --console-address ":9001"
如果你希望集成到k8s,还可以使用Operator方式搭建MinIO Server,具体方法可以参考官网文档。
启动成功后,访问MinIO的IP地址,因为我的 MinIO Server 安装在本机,所以是 http://localhost:9001,输入默认的账号密码是 minioadmin/minioadmin。
登录后,进入控制台看板页。
好了,搭建完 MinIO Server 之后,我们就需要实现自己的对象存储 HTTP服务了。
- 通过 SpringBoot 实现 HTTP 服务
因为项目使用的是 SpringBoot 应用,所以这里主要演示通过 SpringBoot 实现的 HTTP 服务。
首先, 在 pom.xml中 添加 MinIO 的相关依赖:
<!--MinIO JAVA SDK--><dependency> <groupId>io.minio</groupId> <artifactId>minio</artifactId> <version>3.0.10</version></dependency>
第二步,在 SpringBoot 中开启文件上传功能,在 application.yml 中添加如下配置:
spring: servlet: multipart: enabled: true #开启文件上传 max-file-size: 10MB #限制文件上传大小为10M
然后,添加一个 MinioController 控制器用于实现文件的上传、下载、删除操作:
package com.dunshan.controller;import com.google.api.client.util.IOUtils;import com.dunshan.common.api.CommonResult;import com.dunshan.dto.MinioUploadDto;import io.minio.MinioClient;import io.minio.policy.PolicyType;import io.swagger.annotations.Api;import io.swagger.annotations.ApiOperation;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.beans.factory.annotation.Value;import org.springframework.stereotype.Controller;import org.springframework.web.bind.annotation.RequestMapping;import org.springframework.web.bind.annotation.RequestMethod;import org.springframework.web.bind.annotation.RequestParam;import org.springframework.web.bind.annotation.ResponseBody;import org.springframework.web.multipart.MultipartFile;import javax.servlet.http.HttpServletResponse;import java.io.InputStream;import java.io.OutputStream;import java.net.URLEncoder;import java.text.SimpleDateFormat;import java.util.Date;/** * Created by dunshan on 2019/12/25. /@Api(tags = "MinioController", description = "MinIO对象存储管理")@Controller@RequestMapping("/minio")public class MinioController { private static final Logger LOGGER = LoggerFactory.getLogger(MinioController.class); @Value("$") private String ENDPOINT; @Value("$") private String BUCKET_NAME; @Value("$") private String ACCESS_KEY; @Value("$") private String SECRET_KEY; @ApiOperation("文件上传") @RequestMapping(value = "/upload", method = RequestMethod.POST) @ResponseBody public CommonResult upload(@RequestParam("file") MultipartFile file) { try { //创建一个MinIO的Java客户端 MinioClient minioClient = new MinioClient(ENDPOINT, ACCESS_KEY, SECRET_KEY); boolean isExist = minioClient.bucketExists(BUCKET_NAME); if (isExist) { LOGGER.info("存储桶已经存在!"); } else { //创建存储桶并设置只读权限 minioClient.makeBucket(BUCKET_NAME); minioClient.setBucketPolicy(BUCKET_NAME, ".*", PolicyType.READ_ONLY); } String filename = file.getOriginalFilename(); SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd"); // 设置存储对象名称 String objectName = sdf.format(new Date()) + "/" + filename; // 使用putObject上传一个文件到存储桶中 minioClient.putObject(BUCKET_NAME, objectName, file.getInputStream(), file.getContentType()); LOGGER.info("文件上传成功!"); MinioUploadDto minioUploadDto = new MinioUploadDto(); minioUploadDto.setName(filename); minioUploadDto.setUrl(ENDPOINT + "/" + BUCKET_NAME + "/" + objectName); return CommonResult.success(minioUploadDto); } catch (Exception e) { LOGGER.info("上传发生错误: {}!", e.getMessage()); } return CommonResult.failed(); } @ApiOperation("文件删除") @RequestMapping(value = "/delete", method = RequestMethod.POST) @ResponseBody public CommonResult delete(@RequestParam("objectName") String objectName) { try { MinioClient minioClient = new MinioClient(ENDPOINT, ACCESS_KEY, SECRET_KEY); minioClient.removeObject(BUCKET_NAME, objectName); return CommonResult.success(null); } catch (Exception e) { e.printStackTrace(); } return CommonResult.failed(); } @ApiOperation("文件下载") @RequestMapping(value = "/download", method = RequestMethod.GET) @ResponseBody public CommonResult download(@RequestParam("filename") String filename, HttpServletResponse httpResponse) { try { MinioClient minioClient = new MinioClient(ENDPOINT, ACCESS_KEY, SECRET_KEY); InputStream inputStream= minioClient.getObject(BUCKET_NAME, filename); httpResponse.reset(); httpResponse.setHeader("Content-Disposition", "attachment;filename=" + URLEncoder.encode(filename, "UTF-8")); httpResponse.setContentType("application/octet-stream"); httpResponse.setCharacterEncoding("utf-8"); OutputStream outputStream = httpResponse.getOutputStream(); IOUtils.copy(inputStream, outputStream); outputStream.close(); } catch (Exception e) { LOGGER.info("导出失败:", e.getMessage()); e.printStackTrace(); } return null; }}
最后,在 application.yml 中对 MinIO 客户端进行配置:
# MinIO对象存储相关配置minio: endpoint: http://127.0.0.1:9000 #MinIO服务所在地址 bucketName: goreplay #存储桶名称 accessKey: minioadmin #访问的key secretKey: minioadmin #访问的秘钥
- 接口测试
接下来我们启动 SpringBoot 应用,使用 Postman 来测试验证一下功能。
首先,访问上传接口,进行文件上传。上传接口的地址是:http://localhost:8080/minio/upload。
上传完成后,我们打开 MinIO 的管理界面,可以看到上传后的文件。
我们还可以调用删除接口来删除其中某文件,需要注意的是,objectName 参数值是存储桶(Buckets)中的文件相对路径,删除文件接口地址:http://localhost:8080/minio/delete。
最后,我们还可以调用下载接口来下载文件,下载文件接口地址:http://localhost:8080/minio/download。
导出文件时填好文件名称,选择 send and download按钮就可以在 Postman 中下载文件了。
好了,到这里,我们的对象存储服务就改造完成了,这样,我们流量仓库的功能也就差不多实现了。
性能监控
接下来,我们要对 GoReplay 性能监控进行改造。
- GoReplay 统计请求队列信息
前面我们讲过, GoReplay 通过参数:-stats --out-http-stats 在控制台将统计的发送请求队列信息默认每 5 秒输出到控制台。
下面是参数 --stats --output-http-stats 的说明:
--stats //打开输出队列的状态 Turn on queue stats output-output-http-stats //每5秒钟输出一次输出队列的状态 Report http output queue stats to console every N milliseconds. See output-http-stats-ms-output-http-stats-ms int Report http output queue stats to console every N milliseconds. default: 5000 (default 5000)
统计并发送请求队列信息的核心代码主要会用到下面两个函数。
我们用 output_http.go 实现统计信息收集:
// PluginWrite writes message to this plugin// 统计信息收集func (o *HTTPOutput) PluginWrite(msg *Message) (n int, err error) { if !isRequestPayload(msg.Meta) { return len(msg.Data), nil } select { case <-o.stop: return 0, ErrorStopped case o.queue <- msg: } if o.config.Stats { o.queueStats.Write(len(o.queue)) } if len(o.queue) > 0 { // try to start a new worker to serve if atomic.LoadInt32(&o.activeWorkers) < int32(o.config.WorkersMax) { go o.startWorker() atomic.AddInt32(&o.activeWorkers, 1) } } return len(msg.Data) + len(msg.Meta), nil}
gor_stat.go 类中用 NewGorStat 函数构造统计:
// NewGorStat统计函数func NewGorStat(statName string, rateMs int) (s *GorStat) { s = new(GorStat) s.statName = statName s.rateMs = rateMs s.latest = 0 s.mean = 0 s.max = 0 s.count = 0 if Settings.Stats { go s.reportStats() } return}
控制台输出的发送请求队列信息是这样的:
其中,倒数第二列等同于当前的TPS,但是这就是仅有的统计项了。如果想要更复杂的测试统计结果,就需要我们自己去埋点丰富监控指标了。
- GoReplay 埋点思路
在前面的课程中,我们已经选用了Exporter+Prometheus+Grafana作为我们全局的监控解决方案了,这里我们能不能把 GoReplay 的 Metrics 实时接入进来呢?
事实上是可以做到的。
我们先来看看 node_exporter+Prometheus+Grafana套件的运行效果:
Prometheus 提供了 官方版 Golang 库 ,用于采集并暴露监控数据。我们只需要埋点,让 GoReplay 支持实时 Metrics 统计就可以了。这样,在 GoReplay 执行流量回放时,我们就可以实时采集TPS和响应时间等各项压测指标了,另外,结合 Grafana 看板还可以做到图形可视化展示。
既然我们要去 GoReplay 埋点,那么就得知道去哪里埋,对吧?所以我们得先分析源码,找出 GoReplay 发出请求的代码,理清 GoReplay 埋点的思路。
在用 GoReplay 进行流量回放时,我们主要使用的是HTTP输出的插件output_http.go 。它通过实现 HTTP 协议, 进而实现 io.Writer 接口,最后根据配置注册到 Plugin.outputs 队列里。
在output_http.go 中, NewHTTPOutput 是默认初始化函数:
// NewHTTPOutput constructor for HTTPOutput// Initialize workersfunc NewHTTPOutput(address string, config HTTPOutputConfig) PluginReadWriter { o := new(HTTPOutput) var err error config.url, err = url.Parse(address) if err != nil { log.Fatal(fmt.Sprintf("[OUTPUT-HTTP] parse HTTP output URL error[%q]", err)) } if config.url.Scheme == "" config.rawURL = config.url.String() if config.Timeout < time.Millisecond100 if config.BufferSize <= 0 { config.BufferSize = 100 * 1024 // 100kb } if config.WorkersMin <= 0 if config.WorkersMin > 1000 if config.WorkersMax <= 0 { config.WorkersMax = math.MaxInt32 // idealy so large } if config.WorkersMax < config.WorkersMin if config.QueueLen <= 0 if config.RedirectLimit < 0 if config.WorkerTimeout <= 0 { config.WorkerTimeout = time.Second * 2 } o.config = config o.stop = make(chan bool) //是否收集统计信息,统计输出间隔是多少 if o.config.Stats { o.queueStats = NewGorStat("output_http", o.config.StatsMs) } o.queue = make(chan *Message, o.config.QueueLen) if o.config.TrackResponses { o.responses = make(chan *response, o.config.QueueLen) } // it should not be buffered to avoid races o.stopWorker = make(chan struct{}) if o.config.ElasticSearch != "" { o.elasticSearch = new(ESPlugin) o.elasticSearch.Init(o.config.ElasticSearch) } o.client = NewHTTPClient(o.config) o.activeWorkers += int32(o.config.WorkersMin) for i := 0; i < o.config.WorkersMin; i++ { go o.startWorker() } go o.workerMaster() return o}
初始化配置后,启动 HttpClient 网络库:
紧接着,HttpClient 会启动多个发送请求协程:
HttpClient 执行请求发送:
我们可以看看HttpClient 发送请求的细节,下面这张图中,我圈出的内容是各种配置的生效点:
为了对output_http.go 有一个更清晰的理解,你可以看看下面这张核心代码逻辑调用图:
好了,搞懂了 GoReplay 发出请求的逻辑,接下来,我们就需要具体埋点了。
- GoReplay 实现埋点
我们之所以要埋点,主要是为了实现获取请求的状态码以及对应的 TPS,然后在获取请求返回值的位置插入对应的 Metric。
在此之前,我们先要了解 Prometheus 中常见的四大指标类型:
- Counter(计数器):一个递增的计数器,只增不减,但是它可以被重置为0(例如重启服务)。常用于需要记录请求次数、错误数量的场景;
- Gauge(仪表盘):可以用它来表示一个可以任意变化的浮动值,可增可减。常用于反馈当前情况的场景;
- Histogram(累积直方图):多用于需要统计一些数据分布的情况。它可以计算在一定范围内的分布情况,同时还提供了度量指标值的总和。常用于记录请求延迟、响应大小等统计场景;
- Summary(摘要):在一段时间范围内对数据进行采样,和 Histogram 累积直方图比较类似,主要用于计算在一定时间窗口范围内,度量指标对象的总数以及所有度量指标值的总和。
更多内容你可以参考 Prometheus官网。
了解了基础概念,我们就来看看要怎么在 GoReplay 中进行改造。
第一步,通过 go get 命令来安装相关依赖库,示例如下:
go get github.com/prometheus/client_golang/prometheus/promhttp
有关全面的 API 文档,你可以参考 Prometheus 的各种 Go 库的 GoDoc 文档。
第二步,创建 prometheus_family.go 类,初始化监控 Metric。
package metricsimport "github.com/prometheus/client_golang/prometheus"// 创建 Prometheus 数据Metric, 就相当于SQL 数据库 声明tablevar ( //Counter(计数器) totalRequestsCounter = prometheus.NewCounterVec( prometheus.CounterOpts{ Name: "goreplay_total_requests", Help: "total income requests", }, []string{"location", "code"}, ) //Counter (计数器) subRequestsCounter = prometheus.NewCounterVec( prometheus.CounterOpts{ Name: "test_sub_requests", Help: "sub requests", }, []string{"test"}, ) //Gauge(仪表盘) circuitBreakerRateGauge = prometheus.NewGaugeVec( prometheus.GaugeOpts{ Name: "goreplay_circuit_breaker_rate", Help: "rate of circuit breaker", }, []string{"location", "code"}, ) buckets = []float64{0, 100, 200} // Histogram(累积直方图) totalRequestsTimeHistogram = prometheus.NewHistogramVec( prometheus.HistogramOpts{ Name: "goreplay_total_requests_time", Help: "income requests time", Buckets: buckets, }, []string{"location"}, ))// 注册定义好的Metric 相当于执行SQL create table 语句func init() { prometheus.MustRegister(totalRequestsCounter) prometheus.MustRegister(subRequestsCounter) prometheus.MustRegister(circuitBreakerRateGauge) prometheus.MustRegister(totalRequestsTimeHistogram)}func IncreaseTotalRequests(location,code string) { totalRequestsCounter.With(prometheus.Labels{"location": location, "code": code}).Add(1)}func IncreaseSubRequests() { subRequestsCounter.With(prometheus.Labels{}).Add(1)}func ObserveTotalRequestsTimeHistogram(location string, d float64) { totalRequestsTimeHistogram.With(prometheus.Labels{"location": location}).Observe(d)}
这里,我在项目中创建了 package metrics,初始化了 API 请求相关的 Metric,并且注册到了github.com/prometheus/client_golang/prometheus 。
第三步,改造 output_http.go 类,在业务代码中采集数据。
从这张截图可以看到,我们为了获取请求状态码以及对应的TPS,在获取请求返回值的位置插入了counter metric;为了获取不同请求的响应时间等指标,我们还需要在发出请求前记录开始时间,待请求返回后记录结束时间,同时还要记录时间消耗的 Histogram Metric。
这里大家可以举一反三,扩展其它的 Metric。
第四步,改造 main.go ,进行 Metric 注册,提供/metric 接口给 Prometheus TSDB 时序数据库收集数据。
第五步,重新编译 GoReplay 程序。
在代码所在目录(./src/goreplay)下使用 go build 命令。
go build
第六步,在启动 GoReplay 进行流量回放时,查看监听端口,可以看到我们注册的端口 28081,访问接口:http://localhost:28081/metrics。
。......# HELP goreplay_total_requests total income requests# TYPE goreplay_total_requests countergoreplay_total_requests{code="200 ",location=""} 22# HELP goreplay_total_requests_time income requests time# TYPE goreplay_total_requests_time histogramgoreplay_total_requests_time_bucket{location="",le="0"} 0goreplay_total_requests_time_bucket{location="",le="100"} 22goreplay_total_requests_time_bucket{location="",le="200"} 22goreplay_total_requests_time_bucket{location="",le="+Inf"} 22goreplay_total_requests_time_sum 2.1513923769999996goreplay_total_requests_time_count 22# HELP promhttp_metric_handler_requests_in_flight Current number of scrapes being served.# TYPE promhttp_metric_handler_requests_in_flight gaugepromhttp_metric_handler_requests_in_flight 1# HELP promhttp_metric_handler_requests_total Total number of scrapes by HTTP status code.# TYPE promhttp_metric_handler_requests_total counterpromhttp_metric_handler_requests_total 20promhttp_metric_handler_requests_total 0promhttp_metric_handler_requests_total 0
第七步,在Prometheus 主程序中拉取采集数据。
在Prometheus主程序的配置文件中填写第四步的API接口信息,这样,Prometheus TSDB 时序数据库就可以开始定时收集 Metric 数据了。
###################### GoReolay ######################- job_name: "GoReolays5" static_configs: - targets: ['172.31.184.225:28081'] labels: instance: s5
第八步,在 Grafana 做图形数据展示。
Prometheus 提供了一种功能表达式语言 PromQL,这种语言可以允许用户实时选择和汇聚时间序列数据。另外,表达式的结果可以在结合 Grafana 的控件中显示为图形,也可以显示为表格数据,或者由外部系统通过 HTTP API 调用,因为篇幅有限,网上关于这部分的资料又很多,这里我就不多说了。
到这里,我们对 GoReplay 的埋点改造工作就做完了,通过 Metric 实现了从客户端统计压测过程中的各项指标。
总结
好,这节课就讲到这里。
刚才,我们介绍了流量平台的对象存储和性能监控功能的选型及改造,我还对这两个部分做了详细的演示。通过结合 MinIO Server 和 HTTP 服务,我们可以实现程序二进制包、执行器 jar 包、流量文件等文件管理,并能够使用通用的 HTTP 上传下载功能。另外,通过 Prometheus 在GoReplay 埋点,我们还进一步丰富了性能监控指标。
在当前的全链路压测的市场中,对全链路压测工具的分布式改造总是讳忌莫深的部分,而压力工具对全链路压测来说,目标是能够实现足够的流量即可。从本节课的内容可以看到,相比传统的压力工具,全链路压测工具在改造上的技术成本还算是高的,但值得欣慰的是开源的工具也是完全可以实现的。
下一节课,我们将继续讲解分布式改造的各环节,我会通过案例给你演示怎样进行分布式调度改造工作。
课后题
学完这节课,请你思考两个问题:
- 你有没有做过 Prometheus 埋点,谈谈你对业务埋点的一些心得吧!
- 相比文件系统,你觉得对象存储的优势在什么地方?
欢迎你在留言区与我交流讨论。当然了,你也可以把这节课分享给你身边的朋友,他们的一些想法或许会让你有更大的收获。我们下节课见!