基于表格存储的高性能监控数据存储计算方案

2021-10-15 02:12:58

概述

随着软件架构的愈发复杂，了解系统现状、调查问题的困难度也增加了很多。此时，一套完善的监控方案能够让开发和运维工程师快速排查问题，更好的维护系统的稳定性。

开源监控方案中，Zabbix、Nagios都是不错的监控软件，可以针对数十万的设备监控数百万的指标，强大的功能让开发和运维都很赞叹。但是，网上经常看到的抱怨是其写入和存储能力的不足，以Zabbix为例，文章[1]提到使用NoSQL方案（HBase、Cassandra、Riak）比利用传统RDBMS方案（MySQL、PostgreSQL、Oracle）其性能提高了1.5-3倍。如果考虑到架构的扩展性以及存储空间的成本，NoSQL方案还会更有优势，因为一般来说，NoSQL的压缩效果都会更好。

下面我们将基于一个实例案例，来讲解如何使用阿里云NoSQL服务“表格存储”来进行监控数据的处理。下面叙述的过程中，我们不会直接拿出最好的方案，而是将逐步优化的思路整理出来分享给大家，期望大家跟我们一起寻求更优的解决方案。心急的同学可以直接跳到最后看结论。

需求定义

一个典型的监控系统包括数据的采集、计算（实时计算和离线计算）、存储和展示，采集和展示是相对独立的模块，这里我们只以存储为重点，在必要的时候也会对计算做相应的说明。

从业务角度看，监控系统需要完成如下功能：

任何时候都要求能写入，不得丢点；
给定某个机器，某个指标，能够查询该机器该指标在一段时间内的连续的值；
给定某个机器，查询该机器所有指标在一段时间内的连续的值；
给定某个指标，查询所有机器在一段时间内的连续的值；
以上三点，时间段可以任意指定，从分钟到月均支持；

从系统设计角度看，监控对存储系统的核心要求如下：

在线变更表模式：监控指标变动频繁，表模式必须做到*改变，同时不对线上业务产生任何影响；
写入性能高：一般批量写数百条数据延时在数百ms内；
高扩展性：随着监控的设备和指标越来越多，写入能力和存储能力需求越来越大，写入要支持每秒千万行监控数据，存储系统要支持数十P数据；扩容过程用户无感知；
低存储成本：监控数据一般比较多，存储成本需要重点关注，存储系统要能够在满足访问需求的前提下尽可能的降低成本；
老数据自动清理：一段时间前的数据一般不再需要，为了节约成本需要系统自动删除，方便用户；

上面列的需求是从宏观层面上理解的，是架构师和CTO关心的事情。而作为一个干活的程序员，我们需要从微观层面，从动手写代码的角度再一次细化需求。扩展、成本咱就不管了，这些老大都决定好了（关于成本最后有个具体的示例），我们就关心如何快速的构建系统。

对某个机器的某个指标，采集间隔为5秒，每行记录约100Byte；平均写入延时低于50ms；
数据保留6个月；
机器个数可以数百万，指标个数可以数千万，数据量可能过P；
典型指标查询的时间范围是最近10分钟、1小时、24小时，亦可自定义起始和终止时间；
对大范围数据做聚合：如果查询的时间范围太大，返回数据点不可太多，否则传输和展示都是负担；
查询要求，已知机器、指标，查询某个时间范围内的指标数据；
查询要求，已知机器，查询某个时间范围内的所有指标数据；
查询要求，已知指标，查询某个时间范围内所有机器的该指标数据；
典型的业务查询操作，延时要求低于200ms；

通用监控架构

下面图1描述了通用的监控系统架构图，以方便继续讲解。监控系统是一个庞大复杂的体系，网上找了个更完善的监控架构图，见[2]。

图1 通用报警系统架构，其中数据聚合任务亦可以使用流计算工具代替，道理类似。某些情况下，数据聚合任务也可以和采集代理合并为一个进程，简化架构。

各个模块的作用大概如下：

监控指标采集代理：部署在应用服务器上收集指标的软件，一些流行的开源软件比如MySQL/Nginx等本身就有很多第三方的监控代理；对于自己开发的系统，可以在程序中埋点，定制化代理；
数据聚合任务：上面需求里面提到需要看长时间周期的数据，如果查看过去一个月的监控图，需要将所有秒级的数据直接拿出来展示，对于传输带宽和前端展示系统都是极大的考验，而且一般业务并无此种精度需求；这个任务的目的就是定时将原始数据拉出来计算更粗粒度的指标，以便观察指标的趋势性；
报警系统：数据聚合任务在计算指标过程中会发现异常指标，此时可以将该指标通知报警系统，报警系统从存储系统里面拿出更细的信息之后，可以通过短信、电话等方式通知相关人员；如果相关事件也能输入报警系统，那么指标展示界面上就可以将异常指标和事件关联，加快问题的解决，如图
可视化展示：丰富的可视化工具可以加速问题的发现，将可视化和后端的数据系统解耦能方便尝试不同的可视化工具；
表格存储系统：无限水平扩展的NoSQL服务，写入能力强大，过期数据自动清理；

图2 报警和系统事件存储在一起，协同处理之后可以得到更有价值的展示

从上面的介绍中能够意识到，最困难的问题是表结构的设计，要求该设计能够避免数据热点，能让聚合任务快速的读取过去一段周期的数据进行聚合，也能让报警任务快速读取异常指标细节，下面我们就开始结构设计之旅，看看如何一步步构建高性能的监控数据存储计算系统。

表结构设计

上述架构中，一个实际的场景如下：机器N指标M通过指标收集代理每5秒将指标数据写入秒表table_5s, 一分钟后N-M向table_5s写入了12条数据，此时聚合任务可以将table_5s最近写入的12条数据读出来，计算均值（也可以按照业务需求做各种计算），然后写入分钟表table_1m，这样1分钟后table_1m写入了一条数据。依次类推，60分钟后小时表table_1h得到了一条数据，24小时后日表table_1d得到了一条数据。如上，有了秒表、分钟表、小时表、日表之后，我们就可以根据不同的精度需求满足业务的查询需要。如果用户要查询最近一个月的趋势，就应该读日表table_1d，一次查询数据在数百条内，如果用户要查一分钟内的细节问题，就应该读表table_5s。

下面表1给出了我们首先想到的表结构设计(秒/分钟/小时/日表结构是类似的)，有3列PK，value是INTEGER类型，

表1 最简单粗暴的表结构设计

*NodeName()**	*MetricName()**	*Timestamp()**	Value
STRING	STRING	INTEGER	INTEGER
Node1	CPU	1000	43
Node1	CPU	1005	50
Node1	Disk	1000	10
Node1	Disk	1005	20
Node2	CPU	1000	60
Node2	CPU	1005	80

从上面的需求来看，代理每5秒针对每个机器每个指标写入一条数据没有问题，

// 伪代码，实际示例见各个SDK中example[3]

tableStore.putRow(table_5s, {Node1, CPU, 1000}, {Value = 43})

而实际上，在某一个时刻会多个机器多个指标都会产生数据，所以更好的做法是使用批量接口减少网络IO次数，也利于存储系统优化实现；

// 伪代码，实际示例见各个SDK中example[3]
// 关于如何高效的将数据批量写入表格存储，[6]提供了非常详尽的说明

tableStore.BatchWrite (table_5s, [{Node1, CPU, 1000}, {Value = 43}],
[ {Node1, Disk, 1000}, {Value = 10}])

聚合任务则定期从高精度表拉数据计算然后写入低精度表，

// 伪代码，实际示例见各个SDK中example[3]
// 最近5s的机器Node1的CPU数据全部读取出来

rows = tableStore.getRange(table_5s, {Node1, CPU, 940}, {Node1, CPU, 1000}, {Value})
For row in rows:
        sum += row.Value

avg = sum/rows.size()
// 下面的PutRow同样可以使用上面提到的BatchWriteRow优化，
// 比如可以同时将多个指标的数据聚合好，然后统一写入

tableStore.PutRow(table_1m, {Node1, CPU, 1000}, {Value = avg})

至此，计算、存储相关的代码已经写完了，系统已经可以正常的运转，如果业务量不大，这个方案是可以的。

随着业务量的增加，上面的设计可能有如下几个问题：

表第一列PK为机器名，可能导致热点问题。想象一下，你打算做一个监控平台，服务其他的用户，那么表第一列不再是机器名，而是用户名，那么一些大用户，其监控指标多达数十万，而这些监控指标因为拥有共同的第一列PK，表格存储无法对其做自动分区（这是表格存储的设计决定的，关于分区的概念，见[4]），从而使得该热点无法被很好的平衡；
扩展性不够灵活：比如有个需求是这样的，业务希望看到机器N的CPU指标超过90%的所有记录，上面的表结构就没法满足了，此时需要的表结构是表2中所示：

表2 为了查找特定机器的CPU高点，需要跟表1不同的表结构

*NodeName()**	*MetricName()**	Value	*Timestamp()**	PlaceHolder
STRING	STRING	INTEGER	INTEGER	INTEGER
Node1	CPU	43	1000	43
Node1	CPU	50	1005	50
Node1	Disk	10	1000	10
Node1	Disk	20	1005	20
Node2	CPU	60	1000	60
Node2	CPU	80	1005	80

/// 伪代码，实际示例见各个SDK中example[3]


Rows = tableStore.GetRowByRange(table_5s, {Node1, CPU, 90, MIN}, {Node1, CPU, MAX, MAX})

这样就可以将该机器CPU超过90%的记录都拿出来。大家看到了这个表结构跟上面的不同，因为PK的列数不同，此时要满足这个业务需求，我们当然可以重建一个表，但是首先表格存储不鼓励建立很多小表，其次类似业务变更可能很多，每次建表影响业务灵活性。

因此，如果我们的表结构变为如表3

表3 新的表结构设计，增加了业务变更的灵活性

*NodeName_MetricName()**	*Timestamp()**	Value
STRING	INTEGER	INTEGER
Node1_CPU	1000	43
Node1_CPU	1005	50
Node1_DISK	1000	10
Node1_DISK	1005	20
Node2_CPU	1000	60
Node2_CPU	1005	80

这种表结构设计比初始设计就要好很多了，第一列是我们自己拼起来的字符串，我们有足够的*按照业务需求来拼，表格存储也能够方便的做负载均衡。按照上述结构设计，上面提到的新的业务得到的表数据如下，如表4

表4 按照新的表结构设计，监控数据展示如下

*NodeName_MetricName()**	*Timestamp()**	PlaceHolder	Value
STRING	INTEGER	INTEGER	INTEGER
Node1_CPU_43	1000	43
Node1_CPU_50	1005	50
Node1_DISK	1000		10
Node1_DISK	1005		20
Node2_CPU_60	1000	60
Node2_CPU_80	1005	80

请注意，上面只是按照业务需求将CPU的PK拼装格式改变了，DISK因为没有这个需求，并不需要改变，表格存储允许不同行的列不同，这样也不会带来额外的存储空间占用。由此我们也能看到，在PK的组织方式上，是有很多花样可以玩的。

问题都解决了吗？还没。我们注意到聚合任务需要定期的读取最近的K条数据做聚合之用，比如从table_5s中读12条生成分钟级别数据，从table_1m中读60条数据生成小时级别数据等，这些读都是通过getRange实现的，也即是一个小的scan读，而表格存储采用LSM[5]模型实现，在写入量巨大的时候scan读会导致大量磁盘IO，从而也容易引起性能下降。而且，随着这些表数据越来越多，这种读取的性能也会越来越难以保证。

如何解决呢？答案是对前面的每类表，建立一个表结构相同的buffer表。比如table_5s表会建立一个对应的table_5s_buffer表。这个表里面只存最近一段时间的数据，比如1天，超过1天的数据自动过期删除，这样数据少了，访问的时候IO次数可控，性能可控。上面保留1天是假设聚合任务可能出问题而多保留了一段时间，实际上对表table_5s_buffer我们几乎只会读最近数秒的数据，而对这种访问刚刚写入的数据的场景，表格存储是有特定的优化的，就是类似文件系统的page cache的概念，数据写入磁盘前首先写入内存，这样访问最新写入的数据命中内存的可能性就变大了。上面的两个特点共同保障了表table_5s_buffer的读性能。

是否还可以继续优化？答案是Yes。我们回头看看，一个机器可能有数千个指标需要监控，包括系统级别的和应用级别的，那么聚合任务对每个指标都要执行scan就有点浪费了，scan的次数跟机器X指标数成正比。实际上，我们可以重新设计各个buffer表，结构如表5所示，第二列是时间，也就是按照时间对指标进行排序，

表5 重新设计buffer表，避免针对每个机器、每个指标都要scan读

NodeName	*Timestamp()**	*MetricName()**	Value
STRING	INTEGER	STRING	INTEGER
Node1_	1000	CPU	43
Node1	1000	DISK	10
Node1	1005	CPU	50
Node1	1005	DISK	20
Node2	1000	CPU	60
Node2	1005	CPU	80

有了上面的设计，聚合任务想拿Node1最近1分钟（12行）的所有秒级监控指标，只要一次getRange查询就可以了，如果指标太多，SDK会自行分阶段多次读取。


// 伪代码，实际示例见各个SDK中example[3]
// 注意，这里数据可能较多，我们使用iterator

iter = tableStore. createRangeIterator (table_5s_buffer, {Node1, 940}, {Node1, 1000}, Value)
Map<metric, list<int> > metrics;
While (iter.MoveNext()):
   metrics[iter.Get(“Metric”)].append(iter->Get(“Value”))

For k,v in metrics:
       avg = sum(v)
       tableStore.PutRow(table_1m, {Node1_k, 1000}, {Value : avg})

    

这样每次scan都可以拿到所有指标的数据，读的IOPS已经降低到很低的水平了，但是还有一个隐藏的问题，如果某个机器的指标特别多，那么会有热点，因为在某个时间点，所有这个机器的指标都是写到一个分区的（还记得我们上面说第一列PK相同的行都在一个分区吗），可以继续优化。新的表结构设计如表6：

表6 为了避免头部热点，对每个机器上的若干指标做分桶，这样也利于聚集任务负载均衡

*BucketId()**	NodeName	*Timestamp()**	*MetricName()**	Value
STRING	STRING	INTEGER	STRING	INTEGER
00001	Node1_	1000	CPU	43
00001	Node1	1000	DISK	10
00003	Node1	1005	CPU	50
00002	Node1	1005	DISK	20
00003	Node2	1000	CPU	60
00003	Node2	1005	CPU	80

其中BucketId是hash(NodeName + MetricName) % K得到的，K可以根据自己的业务规模调整。有了这个设计后，即使某个机器下面需要监控数十万指标，也可以被多个分区均匀处理，避免了只写头部分区的热点问题。这种方案还有一个好处就是，聚合任务也可以启动多个实例，每个实例负责一定数量的桶就可以了，这样聚合任务的负载也是比较均衡的。
总结来说，就是基础表数据采用表1里面给出的结构，而相对应的buffer表采用表6是一个比较好的方案。
上面的各个步骤以一个真实的监控数据处理方案为背景，我们能看到系统如何从一个最简单但是存在性能问题的版本一步步演进到最终的解决性能、访问均匀性等问题的方案。采用最终方案后，该应用至今已经稳定运行4个月，之前的性能问题、访问热点问题均得到了解决。