26 KiB
23 | 压测平台：如何改造对象存储和性能监控？
你好，我是高楼。
这节课我们来聊聊如何改造分布式压测平台。
在第 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("${minio.endpoint}")
private String ENDPOINT;
@Value("${minio.bucketName}")
private String BUCKET_NAME;
@Value("${minio.accessKey}")
private String ACCESS_KEY;
@Value("${minio.secretKey}")
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 workers
func 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.url.Scheme = "http"
}
config.rawURL = config.url.String()
if config.Timeout < time.Millisecond*100 {
config.Timeout = time.Second
}
if config.BufferSize <= 0 {
config.BufferSize = 100 * 1024 // 100kb
}
if config.WorkersMin <= 0 {
config.WorkersMin = 1
}
if config.WorkersMin > 1000 {
config.WorkersMin = 1000
}
if config.WorkersMax <= 0 {
config.WorkersMax = math.MaxInt32 // idealy so large
}
if config.WorkersMax < config.WorkersMin {
config.WorkersMax = config.WorkersMin
}
if config.QueueLen <= 0 {
config.QueueLen = 1000
}
if config.RedirectLimit < 0 {
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 metrics
import "github.com/prometheus/client_golang/prometheus"
// 创建 Prometheus 数据Metric, 就相当于SQL 数据库 声明table
var (
//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 counter
goreplay_total_requests{code="200 ",location=""} 22
# HELP goreplay_total_requests_time income requests time
# TYPE goreplay_total_requests_time histogram
goreplay_total_requests_time_bucket{location="",le="0"} 0
goreplay_total_requests_time_bucket{location="",le="100"} 22
goreplay_total_requests_time_bucket{location="",le="200"} 22
goreplay_total_requests_time_bucket{location="",le="+Inf"} 22
goreplay_total_requests_time_sum{location=""} 2.1513923769999996
goreplay_total_requests_time_count{location=""} 22
# HELP promhttp_metric_handler_requests_in_flight Current number of scrapes being served.
# TYPE promhttp_metric_handler_requests_in_flight gauge
promhttp_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 counter
promhttp_metric_handler_requests_total{code="200"} 20
promhttp_metric_handler_requests_total{code="500"} 0
promhttp_metric_handler_requests_total{code="503"} 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 埋点，谈谈你对业务埋点的一些心得吧！
- 相比文件系统，你觉得对象存储的优势在什么地方？
欢迎你在留言区与我交流讨论。当然了，你也可以把这节课分享给你身边的朋友，他们的一些想法或许会让你有更大的收获。我们下节课见！