气象行业数据种类繁多, 包括各类结构化、半结构化和非结构化数据, 随着互联网技术、信息化技术以及气象观测技术的迅速发展, 气象数据采集频次从逐天逐时转变为逐分钟, 使气象行业累积了海量数据, 数据量已达到PB以上, 具有典型的大数据特征. 传统的计算方式无法对海量数据进行深度挖掘, 在应对海量数据的高并发访问同样遇到了处理瓶颈, 导致在气象防灾减灾领域, 气象数据的价值无法有效的实现. 因此, 气象部门对于应用大数据相关技术以提升气象信息化水平有着迫切的需求, 在此背景下, 气象大数据云平台“天擎”^{[1, 2]}作为气象部门信息化的核心系统应运而生.

“天擎”是一个庞大的、复杂的大型综合性气象业务系统, 它是云计算、大数据、分布式计算、人工智能等各种技术的综合体, “云+端”业务应用云化是“天擎”设计的一个亮点, “天擎”通过提供“数算一体”的业务新形态, 整合气象部门的各类数据资源和处理流程, 消除数据烟囱和信息孤岛, 实现气象业务数据资源的汇聚互联、共建共享, 从而推进气象业务技术体制改革和转型.

2021年底“天擎”正式投入业务运行, “天擎”包括CTS传输、DPC解码、SOD存储、MUSIC接口、PORTAL等多个集群子系统共100多台服务器, 业务环节冗繁且运行流程复杂, 如何保障整个平台全链条全天候的正常运行给运维人员带来了巨大的挑战, 因此, 建立“天擎”的全链条自动化监控告警系统是十分有必要的, 高效的监控告警系统可以及时定位故障并提升故障处理的响应时效保证系统持续稳定运行, 减少运维成本. 目前, 在系统及应用监控方面出现了一些优秀的开源分布式监控软件, 如Zabbix、Prometheus、Nagios等^[3-5], 可以实现系统资源和应用状态的图形化动态监控, 在资源监控领域使用比较广泛. 但是这些开源的框架部署困难、开发技术难度大、并且都是面向一些通用性的监控任务, 对于特殊的气象业务场景并不适用.

针对上述问题, 本文对基于“天擎”整体业务处理流程的关键环节和技术进行了研究, 实时监视和采集各处理环节的运行状态并转换为告警信息, 在保证网络安全的前提下, 通过企业微信将告警信息实时的发送给系统运维人员, 实现对“天擎”各业务系统运行状态的全方位的监控, 助力运维人员及时定位并处理系统故障.

1 系统总体架构 1.1 系统总体架构

本文系统架构分为3层, 包括数据源支持层、数据加工处理层及应用层. 其中数据源支持层以“天擎”的数据环境为基础, “天擎”数据环境包含地面、高空、雷达、数值预报、行业共享数据等14大类的数据集和数据产品, 集成了“虚谷分布式关系型数据库、Gbase分布式列式数据库^[6]、Cassandra分布式表格系统^[7]、Ceph分布式文件系统^{[8, 9]}、分布式对象存储^[10]”等于一体的适合不同数据格式、满足不同应用场景的大数据存储体系. “天擎”以传统分布式NAS结合数据湖技术^[11], 对于运行在此基础上的CTS (传输)、DPC (解码)、SOD (存储)、MUSIC (API接口)、微服务众创接口等各个子系统运行产生的状态消息、数据入库信息、运行处理日志、API接口访问日志、管理信息、NAS存储状态^[12]等信息, 从原始数据接收、分发、质控解码开始, 到最终数据的存储与发布, 都详细地记录每一条数据的产生和动态流向的过程, 完成了对不同类型的数据处理全流程的分析和溯源, 并具备完善的数据生命周期管控能力, 为管理员提供了一体化数据全流程的溯源监控. 此外, 数据湖有着多样化的分析处理能力, 例如基于STORM的雷达、地面Bufr分钟数据流的解码入库架构, 基于SPARK并行分布式微服务众创接口、数据图谱构建与日志分析等, 满足用户对多种应用场景的数据服务需求并能为用户主动推送热点数据. 本系统的告警数据源就是源于上述“天擎”各模块子系统运行过程中所产生的综合状态信息.

数据加工处理层是整个系统的核心模块, 它按照一定的频次, 通过周期性的任务请求, 以API接口、数据库直连等方式从数据源支撑层对数据进行检索采集、然后经过数据清洗、格式转换、数据分析计算并完成最终数据推送的全过程. 数据加工处理层根据业务的实际需求, 设计了多个独立的采集处理模块, 每个采集处理模块都是独立开发, 相互之间没有依赖关系, 并且都是并行运行, 这样极大地提高了数据采集效率. 监控数据在采集过程中需要进行数据清洗以尽可能地消除脏数据, 比如网络传输异常或者故障产生的重复数据、不完整数据等, 其数据无法被利用和纠正, 此类数据只能进行清洗、修复和剔除才能使用. 数据计算主要对是数据表的操作, 比如聚合计算, 通过计算得到具有业务意义的相关指标, 随后将数据格式化, 形成基础告警数据.

应用层主要是对数据的正确性进行验证并完成告警数据的迭代更新, 是通过被动的方式获取加工处理层的基础数据, 即数据加工处理层是将数据“推”到应用层, 而不是应用层从数据加工层“拉”取数据, 应用层被动的接收数据加工层上报的数据. 采用这种数据传输方式符合网络安全的要求, 最后将告警数据发送给微信客户端. 系统总体架构如图1所示.

1.2 系统硬件结构

系统硬件包含8个部分, 分别为万兆交换机、负载均衡设备、数据处理服务器集群、SAN交换机、磁盘阵列、接口、防火墙和安装了企业微信的终端. 本系统采用分布式负载均衡的架构, 以满足系统长时间不间断密集型的稳定运行、以及大流量、高并发和高可用的应用需求. 硬件都是以集群的方式部署, 这是实现分布式架构和高可用的物理基础^[13], 在数据加工处理层与数据源支持层的数据交换是通过代理程序实现的, 代理程序负责数据采集, 随后将采集的监控任务发送给数据加工处理层前端的负载均衡设备, 由负载均衡统一调度分配任务给后台加工服务器, 将加工处理后的信息通过单向传输和感知的方式推送给应用层, 同时为了保证数据的一致性和数据同步带来的延迟, 本系统采用共享SAN存储方式, 将每个节点访问的数据存放在同一个共享存储空间^[14], 即多个节点共享同一份数据有效的解决不同节点数据同步带来的延迟和数据不一致的现象. 集群中的每个部分相互协作共同完成数据交互, 共同实现本系统的各项功能. 其物理架构如图2所示.

2 系统实现关键技术

本系统开发的硬件环境为: (1) Windows环境Intel(R) Core(TM) i7-7700HQ CPU @ 2.80 GHz 8核CPU, 8 GB内存, 500 GB SSD; (2) Linux 16核CPU, 8 GB内存, 500 GB SSD. 软件支撑环境: 操作系统采用CentOS 7.4 和 Windows 10. 开发工具采用 IntelliJ IDEA和Vim. IntelliJ IDEA 用于Java代码开发, Vim用于Python代码开发. 开发和测试环境中, IntelliJ IDEA集成开发环境和所在的Windows主机需要能够访问互联网, 同时Vim开发环境所在的主机也需要能够访问互联网和“天擎”服务接口, 便于系统开发和测试.

2.1 RSA加密算法

根据系统开发的规范标准, 配置文件和程序中对于出现的敏感字符, 比如数据库的用户名、密码等不能出现明文. 本系统采用RSA加密算法^[15]主要是对程序和配置文件中出现的用户名、密码及URL进行加密, 避免主机被攻击之后导致安全信息被泄漏. 其加密示例代码如下.

from Crypto.PublicKey import RSA

from Crypto.Cipher import PKCS1_v1_5 as Cipher_pkcs1_v1_5

import base64

message = " http://localhost:15672/api/queues"

rsakey = RSA.importKey(open("public.pem").read())

cipher = Cipher_pkcs1_v1_5.new(rsakey) cipher_text = base64.b64encode(cipher.encrypt(message.encode('utf-8')))

print(cipher_text.decode('utf-8'))

2.2 Python多进程

为了能够加快站点数据的处理速度, 本系统以多核CPU资源为基础, 利用Python提供的多进程Multi-processing包, 完成了从单进程到并行执行的切换. Multiprocessing能够支持子进程、通信和共享数据、执行不同形式的同步, 同时提供了Process、Queue、Pipe、Lock等组件. 本系统在“天擎” Web服务和FTP服务监控中采用了多进程技术, 其示例代码如下所示.

if __name__ == "__main__":

　　file_exist_res = is_file_exist('url_address.txt')

　　if file_exist_res != 0:

　　　　　file_num, is_ack = split_file('url_address.txt')

　　　　　print("---cpu的核数为:{}".format(cpu_count()) + '---')

　　　　　print("---父进程开始执行---")

　　　　　print("父进程PID: %s" % os.getpid())

　　　　　p = Pool(file_num)

　　　　　for i in range(0, file_num):

　　　　　　　print('生成的文件名为: '+ 'url_address.00%d'%i + " 记录的日志为: failed%d"%i +'.log')

　　　　　　　p.apply_async(get_result, args=('url_address.00%d'%i, 'failed%d'%i +'.log', ))

　　　　　print("---等待所有子进程结束---")

　　　　　p.close()

　　　　　p.join()

　　　　　print("---父进程执行结束----")

2.3 Python调度器

APScheduler (advanced Python scheduler)是Python的一个轻量级定时任务框架, 包含的任务类型有Linux Crontab风格的调度、基于时间间隔的周期性执行调度等, 能够以守护进程的方式启动应用, 并且具有任务持久化的功能. APScheduler包含4个组件, 分别是触发器、作业存储、执行器和调度器. 本系统采用APScheduler中的BlockingScheduler调度器实现间隔执行任务, 主要用于FTP服务采集模块, 时间间隔设置为每10分钟采集一次.

2.4 Python网络爬虫

网络爬虫^[16]就是向网络站点不断的发送HTTP请求, 通过程序模拟浏览器访问站点的行为, 把站点返回的JSON数据、图片、视频等保存到本地, 提取需要的数据进行存储. 流程为发送请求、获取响应内容、解析内容、保存数据. 本系统是通过服务器管理口爬取硬件日志进行分析, 爬虫请求库为Requests, 解析库为正则和beautifulsoup. 本系统网络爬虫技术主要用于“天擎”服务器硬件监控. 其示例代码如下所示.

sess = requests.session()

sess.cookies = cookielib.LWPCookieJar(filename = "cookies.txt")

userAgent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:93.0) Gecko/20100101 Firefox/93.0"

header = {

　　"Referer": " https://ip/api/session",

　　'User-Agent': userAgent,

}

sess.cookies.load()

url = ' https://ip/api/alarm_info'

result = sess.get(routeUrl, headers = header, allow_redirects = False, verify = False)

print(result.text)

if result.status_code != 200:

　　print("Failed")

else:

　　print("Success")

2.5 Java线程池

在线程池中^[17], 线程是可以复用的, 不需要每次都创建一个新的线程, 有效地减少创建和销毁线程的时间开销. 线程的数量可以限制和手动设置, 这样可以避免由于创建线程过多而耗尽系统资源导致服务器宕机, 同时也把实际的任务与线程本身进行了解耦, 本系统使用Executors类提供工厂方法创建不同类型的线程池, ExecutorService executorService = Executors.newFixedThreadPool(6), 通过调用返回值ExecutorService接口的基础方法execute或submit, 将具体执行的线程任务加入线程池中并执行. 线程池技术主要用于主机网络连通性监控.

2.6 Python正则表达式

所有的监控告警信息最终都需要在互联网上进行传输, 由于网络安全的要求, 本系统采用正则表达式^[18]对一些告警信息中的敏感字符进行屏蔽和转换, 保证传输的数据不包含敏感信息, 正则表达式可以处理任何关于字符串的操作, 是一个强大的字符串处理工具, 具有自己独特的语法和数据处理引擎. 其示例代码如下所示.

ip_port_re = r'(6[0-5]{2}[0-3][0-5]|[1-5]\d{4}|[1-9]\d{1, 3}|[0-9])'

re_result = re.findall(ip_port_re, first_part)

　　if len(re_result) == 5:

　　　　ip_address = re_result[0] + '.' + re_result[1] + '.' + re_result[2] + '.' + re_result[3]

　　　　ip_rep = first_part.replace(ip_address, 'ip')

　　　　port_rep = ip_rep.replace(re_result[4], 'port')

　　elif len(re_result) == 4:

　　　　ip_address = re_result[0] + '.' + re_result[1] + '.' + re_result[2] + '.' + re_result[3]

　　　　ip_rep = first_part.replace(ip_address, 'ip')

3 系统核心功能设计与实现

系统的数据采集处理流程位于数据加工处理层, 数据加工处理层是数据平台建设的核心, 目前包括11个独立处理的模块, 可以进行扩展, 这些模块实现了对“天擎”核心服务器运行状况信息的采集和处理存储, 包含CTS、DPC等分系统的消息采集解码, SOD存储、数据同步、清除策略运行任务监控信息的提取以及各类上行数据传输状况信息、后台日志采集等. 采集后的数据使用MySQL和文件索引的方式进行存储. 数据加工处理层监控信息的采集与处理流程如图3所示, MySQL数据库表结构如表1所示.

3.1 “天擎”消息队列数据采集功能设计

“天擎”消息队列主要包括Kafka消息队列^[19]、Bufr数据消息队列、CTS文件消息队列和DPC文件通知消息队列. 消息队列及消费者类型如表2所示.

(1) Kafka数据同步消息队列. 作为“天擎”虚谷和Gbase数据库的数据同步消息中间件, “天擎”数据流从CTS输入、经过DPC进行解码存入虚谷缓冲库, 同时将数据写入Kafka消息队列中, 虚谷服务库和Gbase历史分析库以消费者身份获取该消息数据, 完成数据从缓冲库到服务库和分析库的同步. 当前需要监控的Kafka消息队列(主题)有149个. 通过Kafka集群目录/bin下的kafka-consumer-groups.sh -bootstrap-server HostName:9092 -list 获取每个队列名称, kafka-consumer-groups.sh --bootstrap-server HostName:9092 --describe --group queue name计算每个队列所有分区的消息数目. 本系统采集Kafka消息队列信息的示例代码如下所示.

cat /space/cmadaas/sod/shell_ly/new_code/consumer_group.tmp| while read consumer_group

do

　　if [ "consumergroup"="music−group−2"]||["consumergroup"="music−group−2"]||["{consumer_group}" = "mygroup" ] || [ "${consumer_group}" = "getendpos" ];then

　　　　continue

　　fi

sh /space/cmadaas/sod/kafka_2.11-2.1.1/bin/kafka-consumer-groups.sh --bootstrap-server ${node01}:${port} --describe --group ${consumer_group}|grep '^bclz_'|awk '{print $1, $5, $7}' > /space/cmadaas/sod/shell_ly/new_code/consumer.tmp

　　partition_name='cat /space/cmadaas/sod/shell_ly/new_code/consumer.tmp |head -1|awk -F '''{print $1}''

　　partition_num='cat /space/cmadaas/sod/shell_ly/new_code/consumer.tmp |wc -l '

　　jiya_sum='cat /space/cmadaas/sod/shell_ly/new_code/consumer.tmp |awk '/bclz_/ {sum += $2};END {print sum}''

　　echo ${jiya_sum}

　　consumer_node='cat /space/cmadaas/sod/shell_ly/new_code/consumer.tmp |head -1|awk -F '''{print $3}''

　　fi

done

(2) RabbitMQ 消息队列. 此消息队列包含了Bufr数据消息队列、CTS文件消息队列和DPC文件通知消息队列. Bufr数据消息队列用于接收地面自动站上传的Bufr二进制数据, 此类数据主要包括小时和分钟数据, 自动站将数据以Bufr的形式写入到RabbitMQ的消息队列中, CTS的质控程序进行数据质量控制, 将质控后的数据回写到原来的RabbitMQ消息队列中, 然后供消费者获取, 当前需要监控的消息队列有44个; CTS文件消息队列主要用于省级向中国气象局传输数据, 当前需要监控的消息队列有11个; DPC文件通知消息队列主要用于CTS和DPC之间的解耦, CTS将文件通知打包消息推送到DPC上游的RabbitMQ消息队列中, 然后进行消息拆分, 随后将拆分的消费转发到下游RabbitMQ的各个消息队列中让每个入库程序进行消费, 当前上游需要监控的消息队列有1个, 下游需要监控的消息队列有372个.

RabbitMQ^[20]提供了HTTP API手册, 通过API接口( http://localhost:15672/api/queues)可以获取每个队列的全部属性, 所有API调用都需要做权限验证, 需在请求头部中加入权限验证信息. 本系统采集RabbitMQ消息队列数据信息的示例代码如下所示.

list_tmp_name = []

list_tmp_num = []

list_tmp_consumer = []

res = requests.get(url=dec(queue_url), auth=HTTPBasicAuth(username=dec(usr), password=dec(pwd)))

if res.status_code == 200:

　　queues = json.loads(res.text)

　　for queue in queues:

　　　　node = queue.get("node")

　　　　state = queue.get("state")

　　　　queue_name = queue.get("name") queue_consumer_count = str(queue.get("consumers"))

　　　　jiya_num = str(queue.get("messages"))

　　　　ready_num = str(queue.get("messages_ready")) unacked_num = str(queue.get("messages_unacknowledged"))

3.2 “天擎” SOD同步、备份、清除监控功能设计

“天擎” SOD (存储与管理系统)中同步任务主要是结构化数据同步, 是对虚谷缓冲库同步到虚谷服务库和Gbase历史分析库的每一张数据表的同步任务进行监控, 采集的信息包括同步任务ID、任务运行状态、任务名称、任务运行失败的时间等; 备份任务是对虚谷缓冲库、Gbase历史分析库、虚谷文件索引库、虚谷结构化数据库_基础库_台站信息表、Gbase历史分析库_基础库_台站信息表中的数据进行备份(包含表结构)、将以上数据周期性的备份到NAS, 采集的信息包括备份任务停止运行的类型、任务名称、备份表名、任务停止运行的时间、四级编码等; 清除任务是对虚谷缓冲库中的数据表进行周期性的删除, 采集的信息包括清除任务运行失败的时间、数据库的类型、清除资料的名称、资料表名、清除时间等. 相关的信息如表3所示.

3.3 上行数据传输状况信息采集 3.3.1 “天擎”国家站和区域站Bufr数据延迟监控功能设计

本监控功能是对81个国家站和1735个区域站小时Bufr数据是否存在传输延迟的情况进行监控, 以2分钟时效作为标准, 通过数据到达数据库时间进行判断. 利用虚谷缓冲库数据表SURF_WEA_CHN_MUL_HOR_TAB的字段D_DATETIME, D_IYMDHM 检索数据, 最终生成数据数据传输延迟报表.

3.3.2 “天擎”自动站土壤水分数据完整性监控功能设计

自动站土壤水分数据完整性监控是通过土壤水分数据中心站和“天擎” API接口进行监控^[21]. API接口资料代码为AGME_CHN_SOIL_HOR, 服务接口为getAgmeEleByTimeAndStaIDAndDepth, 通过不同的土壤深度(20个层级)进行监控. 需要监控的站点有81个, 采集的信息包括站名、站点地址、站点的联系方式、数据开始缺失的时间、连续没有接收到数据的时间、土壤深度、时次等.

3.3.3 “天擎”高空和精细化文件监控功能设计

“天擎”高空数据和精细化数据是气象数据中的核心资料, 是制作天气预报的基础, 气象部门对此类数据有着非常严格的时效考核标准, 高空数据每天有4个考核时次, 包含9个站点. 精细化数据每天有5个考核时次, 包含82个站点, 本系统采用直接访问NAS的方式采集监控对象的相关属性, 采集信息包括站号、考核时次、数据落盘的时间、文件存储的地址、文件名、文件类型、站点数量、文件大小等.

3.4 “天擎”服务器硬件监控功能设计

“天擎”服务器硬件监控是通过管理口获取服务器硬件的运行状态, 比如硬盘、内存、CPU、电压、温度等. 通过网络爬虫的方式提取相关硬件的信息, 当前需要监控的服务器有93台.

3.5 日志信息管理

日志管理是对应用程序及基础设备生成的日志事件进行处理的全过程. 包含日志的收集、解析、分析、存储和归档等, 日志管理的目标是为系统故障溯源及分析提供依据. 本系统采用日志管理方式, 对系统的每个功能模块、关键流程、网络状态、主机运行状态等信息进行实时日志写入. 在每个被监控的主机上采用部署代理的方式进行信息采集, 每个代理都不一样, 都是根据实际业务需求定制开发.

3.6 参数配置

本系统提供了参数配置功能, 可以在系统业务化运行以后根据不同需求进行动态调整, 灵活的参数配置功能不仅使得程序开发和基础配置信息进行了分离, 而且提高了系统的扩展性和通用性. 参数配置项主要包括5个部分: 站点信息配置、站点URL配置、数据库配置、FTP服务配置、微信告警平台^[22]配置(参数包括agentid、corpid及corpsecret)等.

4 系统应用效果

目前该系统已经在实际业务中进行了长时间的应用, 实际应用效果总结如下: (1)系统稳定性方面. 在一年多的时间保持连续不间断的运行, 每个功能和模块都运行稳定可靠, 未出现任何故障; (2)安全性方面. 需要和互联网进行数据交互, 本系统部署在气象业务内网, 和互联网进行数据交互时采用单向访问的方式, 即由气象业务内网将告警数据推送给互联网, 相反, 从互联网是不能访问气象业务内网的, 这样保证了系统的安全性, 也符合相关网络安全的要求和规范; (3)系统扩展性方面. 扩展性是应对业务系统变化的能力, 良好的扩展性可以减少系统开发运维的成本, 本系统的每个模块都是根据实际的业务需求进行开发, 与其他的模块是相互独立的, 对可扩展的部分进行了隔离. 部分模块告警效果图如图4所示.

5 结束语

本文根据“天擎”实际业务的运行和发展需求, 设计了对气象大数据云平台“天擎”的监控告警系统, 实现了对“天擎”各类数据资源的统一告警, 保证了“天擎”系统的稳定运行. 该系统设计的总体思路是对监控资源的集约化管理, 符合全国气象信息标准化、集约化的服务理念. 系统投入业务运行之后, 显著地提高了数据传输的及时性和完整性, 解决了传统的人工监控效率低、自动化管理程度不高的问题, 对气象信息化高质量可持续发展具有重要的保障意义. 后续工作将会利用系统每个模块产生的日志对系统常见故障进行汇总分析, 并加入机器学习和人工智能算法, 对系统运行可能产生的故障进行预测, 给运维人员预留故障处理时间.