基于流式处理架构的日志采集系统的设计与实现

邵旭东，樊志杰，，张敬锋，曹志威，周明富，熊已兴，张 林

(1.公安部第三研究所 信息安全技术部，上海 200031；2.安徽省公安厅 科技信息化处，合肥 230061；3.上海辰锐信息科技有限公司 研发中心，上海 200031)

0 引言

随着世界范围内围绕信息的获取、使用、控制的斗争愈演愈烈，全球网络攻击、网络窃密和网络犯罪等问题日渐突出，网络安全问题已经成为与政治安全、经济安全、文化安全同等重要，事关国家安全的重大战略要害问题。泛政府行业大数据中心一旦发生安全问题，造成系统崩溃、网络瘫痪、病毒爆发或重要数据丢失、泄漏，势必导致灾难性后果，给泛政务业务系统正常运转、甚至国家信息体系完整性带来重大损失。为此，亟需建立以数据安全为核心的大数据安全保障体系，建立先进的安全运行管理平台、专业的安全技术团队、全面的安全策略以及高效的运行管理机制来保护大数据的安全。

安全运行管理平台汇聚的安全信息包括数据服务、应用、云平台、终端、边界、网络和安全保障设施等系统运行记录、操作日志、告警信息，各类系统运行信息格式不一、支持的传输协议种类繁多，因此需寻求技术手段，对各类异构的数据源进行统一采集和预处理，通过标准协议上报至安全运行管理平台[1-5]。

当前，行业内存在集中监管与审计系统，该系统为了保障泛政府行业信息安全，对接入泛政府行业内网的业务和用户进行监控，一旦发现安全隐患或攻击行为，立刻采取措施，保障泛政府行业内网的安全。该系统支持对以TBSG可信边界接入网关为隔离设备的边界接入平台、通过VPN网关接入的移动接入平台、通过视频接入专用设备的视频接入平台和以单向光闸为隔离设备的公网接入平台等多种类型平台上的通用网络设备、通用安全设备和专用安全设备的监控，以及设备状态、业务流量和用户访问日志等信息的采集，使得泛政府行业建设单位的管理人员可以方便地对接入平台进行管理，保证接入平台安全稳定地运行。但是该系统也存在一些不足之处，无法满足大数据安全体系下的监管要求，具体来说，存在的主要问题包括：

1)支持的协议类型较少，只支持SNMP和SYSLOG等少数日志采集协议，且扩展难度大；

2)系统将采集到的数据首先保存在关系数据库中，上报数据到其他平台时需要再次从数据库中读取、解析，时延大、执行效率低；

3)采集的日志只支持上报到单一的平台上，数据上报协议固定。为了支持多平台、多协议格式上报，扩展难度大；

4)系统为采用模块化的设计，模块之间耦合大，不方便扩展功能。

鉴于此，本文提出统一日志采集技术并研制系统，可解决集中监管与审计系统存在的不足：

1)采用标准化接口和插件技术，可灵活支持各种日志采集协议和数据上报协议，实现系统整体架构的高度稳定性和可扩展性；

2)采用基于消息队列的流式处理架构，实现日志采集、日志处理、日志上报等各个环节的解耦，并支持灵活的功能扩展；

3)通过消息队列的流量削峰保证了日志传输的可靠性；

4)根据日志流量特征，系统支持动态调整消费组规模，满足系统的高性能要求。

1 系统结构及原理

本文研制的统一日志采集系统，软件层面可以分为数据面与控制面两个部分，如图1所示。

图1 系统结构图

数据面主要包括数据采集、数据发布、数据存储与数据上报等功能，这些功能之间通过消息队列实现模块功能解耦和数据流式处理，方便功能的扩展。

控制面主要包括日志源管理、策略管理、Agent管理、系统管理等功能，实现对数据面功能的策略配置与管控。

2 系统软件设计

本文所设计的统一日志采集系统主要由日志采集、数据上报、数据管理、系统管理、策略管理、Agent管理、日志源管理模块和日志采集代理(Agent)子系统组成[6-10]，如图2所示。

图2 软件框架图

日志采集模块通过各种采集协议实现对各种数据源的日志采集功能。

数据上报模块支持通过kafka、消息队列、SFTP、FTP等协议将日志上报到各类平台。

数据管理模块实现日志数据的本地存储和查询功能。

图3 日志采集模块结构图

系统管理模块实现系统登录、用户角色管理、权限管理、版本管理、备份恢复等功能。

策略管理模块实现日志采集策略的增删改查、同步等功能。

Agent管理模块实现Agent策略配置、Agent状态监控等功能。日志采集代理(Agent)可部署在Windows或Linux系统中，实现特定日志数据的采集上报，并接受统一日志采集系统的集中管理(如：策略配置、状态管理等)。

日志源管理模块实现日志源类型和采集协议管理以及日志源自动发现等功能。

2.1 日志采集

日志采集模块由任务管理、任务调度、采集器和过滤器等组成，日志采集流程如图3所示。

任务管理模块负责生成采集任务，根据系统配置的日志源、日志采集策略和清洗策略，创建相应的采集器和过滤器，放到任务队列中等待执行。

任务调度模块负责读取采集任务队列，分配工作线程，调用采集器完成相应类型日志的采集，采集器通过调用过滤器完成日志数据的清洗，清洗后的日志数据按照主题发布到日志队列中。

采集器根据采集方式不同可以分为主动采集和被动采集。主动采集主要包括FTP、SFTP、SNMP Get、WMI、ODBC/JDBC等协议的日志采集。被动采集主要包括SYSLOG、HTTP、SNMP Trap等协议的日志采集。

对于主动日志采集是通过周期轮询的方式实现日志的采集，任务管理模块负责周期生成日志采集任务，并写入采集任务队列，等待任务调度模块调度执行。

对于被动日志采集，采集器采用监听服务，实现日志数据的持续接收，处理流程如图4所示。

图4 被动采集流程图

过滤器负责根据系统配置的数据清洗策略对采集器获取到的日志数据进行过滤处理。根据不同的日志源类型和数据清洗策略，系统可实现多种类型的过滤器，多个过滤器可以通过组合模式形成新的过滤器。

本系统通过制定标准化的采集器接口和过滤器接口，对于各种数据源和日志类型，按照标准接口实现各类采集器和过滤器，并以插件的方式注入系统中，实现日志采集系统整体架构的稳定性和可扩展性，灵活支持项目的各种实际需求，如图5所示。

图5 采集器和过滤器插件工作原理

2.2 日志数据发布

采集器通过各种采集协议采集到原始日志后，经过清洗、转换等处理后按照主题发布到日志队列中。

主题可采用如下格式：<日志源类型>.<协议类型>.<日志级别>.<日志源IP>，方便数据存储、数据上报等模块按需进行订阅。

2.3 数据上报

数据上报模块支持通过kafka、MQ、SFTP、FTP等协议将日志上报到安全运行管理平台。

本系统通过制定标准化的数据上报引擎接口，对于各种上报协议方式，按照标准接口实现各类数据上报引擎，并以插件的方式注入系统中，实现日志采集系统整体架构的稳定性和可扩展性，如图6所示。

图6 数据上报引擎工作原理

数据上报模块根据系统配置策略，加载指定的数据上报引擎，并创建独立的工作线程，按照主题从日志队列中订阅日志消息，将日志数据上报到安全运行管理平台。

系统通过监测队列中的积压消息数，动态调整工作线程数，实现日志消息传输的低延迟和可靠性。

对于MQ、kafka等协议上报方式，当接收端异常无法上报时，系统将产生告警日志，如果故障未恢复且RabbitMQ中的积压消息达到配置的最大门限值，为了防止内存耗尽，影响系统的整体性能和运行稳定性，系统将启动本地缓存机制，将日志接收并保存到本地文件中。

在上报故障期间，上报线程定期尝试将收到的新数据发送到MQ或kafka，以检测其是否恢复故障，如果未恢复，则继续保存到本地文件中。

如果检测到接收端故障恢复，数据上报正常，则原上报线程将接收的新数据直接上报给MQ、kafka。同时启动独立的线程负责读取保存在本地缓存中的日志信息并上报，上报成功后删除本地缓存文件。

对于FTP、SFTP等协议的处理方式，采集日志数据后先保存到本地文件中，然后通过另外的上传线程负责将本地文件上传到FTP服务器上。

2.4 数据管理

数据管理模块负责实现日志消息的本地存储、归档和检索等功能。

数据管理模块按照主题订阅日志，从日志队列中获取日志数据进行本地存储或者保存到全文搜索引擎中。

2.4.1 日志存储

对于日志存储，日志文件可按照年、月、日分级存储，即目录结构为：<根目录><年><月><日><日志文件>，这样既方便查找日志文件，又避免同一级目录下文件数过多，影响查询速度。

日志存储模块执行流程如下所示：

1)查询数据表获取系统配置的所有已使能的日志存储策略；

2)对于每条存储策略，创建一个线程，线程函数的执行逻辑如下：

(1)根据配置的订阅主题信息，依次订阅这些主题的日志消息；

(2)接收日志消息并保存日志到系统配置的文件存储路径下；

(3)根据系统配置的文件切割策略，完成日志文件的切割。

2.4.2 日志归档

日志归档模块根据系统配置的策略，将历史日志文件压缩后归档。归档日志可根据系统配置的保存时间定期清理。

日志归档模块执行流程如下：

1)启动周期定时器，每天在指定时间(如：凌晨3点)触发归档操作；

2)归档步骤如下：

(1)查询数据表，获取系统配置的所有已使用的日志存储策略；

(2)遍历每条日志存储策略，执行如下流程：

图7 日志归档流程图

2.4.3 日志查询

日志查询模块通过订阅机制，获取日志数据保存到全文搜索引擎中，用于支持本地数据查询功能。

日志查询模块可根据系统配置的策略，定期清除过期数据。对于已过期的日志，系统提供日志文件下载功能。

日志查询模块执行流程如下：

1)查询数据表获取系统配置的所有已使能的日志查询策略；

2)对于每条策略，创建一个线程，线程执行逻辑如下：

(1)根据配置的订阅主题信息，依次订阅这些主题的日志消息；

(2)接收日志消息并保存日志到全文搜索引擎中，并制定相应参数，实现过期日志的自动删除功能。

2.5 策略管理

提供日志源、数据上报策略、日志存储策略和日志归档策略等管理功能，将配置信息保存到MySQL数据库，然后发布配置更新消息到RabbitMQ消息队列中，日志采集、数据上报和数据管理等模块通过订阅相应的配置更新消息，并按照新策略执行，流程如图8所示。

图8 策略管理模块结构图

2.6 系统管理

系统管理模块提供用户管理、角色管理、系统登录、版本管理、备份恢复、系统状态监控等功能。

2.6.1 用户管理

用户管理主要完成用户的查询、新增、修改、删除、导入、登录IP限制和授权等操作。

2.6.2 角色管理

角色管理主要完成角色的查询、新增、修改、删除和授权等操作。

2.6.3 系统登录

平台提供登录页面，支持使用数字证书或用户名密码方式登录系统，平台提供配置方法，可以选择支持其中的一种或两种方式登录系统。为了提高系统的安全性，使用用户名密码登录时，支持字符验证码或手机短信动态验证码，使用哪种方式支持可配置。

用户登录流程如下：

图9 用户登录流程图

1)用户打开登录页面，向服务器请求一个验证码图片；

2)服务器生成一个随机验证码图片返回，并将验证码保存到用户Session中；

3)用户在浏览器中输入用户名、密码和验证码信息，并点击登录按钮；

4)服务器验证收到的用户名、密码和验证码是否正确，如果不正确提示登录失败并返回登录页面；等保测试下如果用户连续登录失败达到一定次数后，应锁定账户，记录安全审计事件；

5)如果当前为等保测试状态且设置了密码的最长使用期限，则判断密码是否过期，如果密码过期则跳转到用户密码修改页面；否则表示登录成功，跳转到首页。

2.6.4 版本管理

提供系统版本升级功能，为了简化升级流程，最大程度支持向下兼容，系统对软件部署和升级方式做如下约束：

1)只能从低版本升级到更高版本，支持跨版本升级，任何高版本升级包都能够升级低版本；

2)除了用到的一些系统级的配置文件，软件安装部署后要求其所有的程序文件、数据文件和配置文件等都在同一个根目录下，根目录下可以包含子目录。

系统升级流程如图10所示。

2.6.5 备份恢复

系统备份主要包括配置文件、数据文件和数据库的备份，其中对于数据库备份目前先实现全量备份。

备份时可以选择备份的内容：配置文件、数据文件和/或数据库，如果输入的备份密码为空，则表示备份包不做加密。

系统备份的执行流程如图11所示。

图10 系统升级流程图

图11 系统备份流程图

2.6.6 系统状态监控

系统状态监控模块通过向设备状态获取线程订阅的方式，获取服务器的CPU、内存、磁盘利用率以及各个网卡接口的数据流量等信息，经过处理后保存到数据库中。

以图表的形式展现服务器的内存、磁盘、CPU占用率以及网络接口流量信息。在首页和性能监控页面中都可以查看设备的状态，首页由于展示区域有限应以紧凑的形式进行展示，对于性能监控页面可以详细地分开展示设备的各种状态。

2.7 日志源管理

日志源管理模块负责管理维护各类日志源属性信息(如：系统名称、IP地址、采集协议、认证凭据、日志源类型等)，并实现日志源自动发现功能。

系统支持日志源自动发现功能。一旦被监控设备或系统有信息发送到统一日志采集系统，系统会根据信息类型和发送端IP地址等进行匹配，一旦发现新的IP地址在监控范围内，且监控设备数量不超过授权许可数量时，自动产生不同类型的新设备。新发现的IP如果发送的是Agent代理发送的事件信息，系统会自动生成一个主机类型日志源；新发现的IP如果发送的是其他类型信息，系统会自动生成一个未分类新日志源，需要管理员定义日志源信息。

3 系统主要功能

本文所研究统一日志采集系统的主要功能模块包括：日志信息采集、数据管理、数据上报、Agent管理、日志源管理等功能[11-15]。

3.1 日志采集功能

支持各种数据源的运行信息和日志数据采集，按数据请求方向分为主动采集和被动采集两类。

主动采集：由统一日志采集系统主动向日志数据源请求获取日志数据，日志源设备被动响应请求回应数据，采集协议包括SFTP、SNMP Get、WMI、HTTPS等。

被动采集：统一日志采集系统支持特定协议监听，被动等待日志数据源传入数据，支持的采集协议有SYSLOG、HTTPS、SNMP Trap等。

系统支持的主要采集协议参见表1。

3.2 数据管理功能

支持采集的数据本地分类存储，提供全文查询和文件检索。提供数据清洗、压缩、归档等数据管理维护功能。

3.3 数据上报功能

支持通过kafaka、消息队列、SFTP、FTP等协议将数据上报到安全数据系统、安全审计中心等外部日志分析处理平台。

3.3.1 本地缓存

在上报故障期间，上报线程应定期尝试将收到的新数据发送到MQ、kafka，以检测其是否恢复故障，如果未恢复，则继续保存到本地文件中。

表1 采集协议

如果检测到接收端故障恢复，数据上报正常，则原上报线程将接收的新数据直接上报给MQ、kafka。同时启动独立的线程负责读取保存在本地缓存中的日志信息并上报，上报成功后删除本地缓存文件。

3.3.2 跨边界级联

在跨各类边界进行日志信息采集时，边界外侧部署一套统一日志采集系统1，采集各类设备运行日志信息，通过FTP协议将采集的数据通过边界设备传到边界内测，该功能同属数据上报模块，只需要新增一个FTP级联上报协议；边界内侧部署另一套统一日志采集系统2，通过FTP协议采集外侧传入的文件并解析处理，该功能同属日志采集模块，只需要新增一个FTP级联数据采集协议。

3.4 策略管理功能

支持数据采集、数据处理、数据上报等策略的配置管理。

3.5 日志源管理

支持对各类日志源信息(日志源类型、系统名称、系统品牌、系统型号、软件版本、IPv4地址、IPv6地址、MAC地址、采集协议、协议版本、认证凭证)的增加、删除、修改、查询、导入、导出等功能，支持日志源自动发现。

3.6 Agent管理

设置Agent配置策略，包括系统的基本参数，如：日志采集间隔、本地日志保存时间、文件大小设置、系统登录密码、卸载密码等参数，服务设置了相关参数后，主机Agent自动同步该参数。

监控各主机Agent信息及运行状态，提供Agent管理操作日志、各模块的启停等运行信息查询、导出、归档等操作。

4 实施方式与应用案例

4.1 实施方式

日志采集、数据管理、数据上报和策略管理作为独立的模块进行部署。此外，系统还需要部署RabbitMQ、ElasticSearch和MySQL[16-19]。统一日志采集系统部署框架如图12所示。

图12 统一日志采集系统部署图

图12中的箭头表示它们之间的数据流向。日志采集模块从MySQL数据库中读取日志源、采集策略、清洗策略等配置信息，完成日志的采集后发布到RabbitMQ消息队列中。

数据上报模块从MySQL数据库中读取上报策略信息，然后从RabbitMQ消息队列中订阅指定主题的日志消息，按照相应协议完成日志上报。

数据管理模块从MySQL数据库中读取日志查询、日志存储和日志归档等策略信息，从RabbitMQ消息队列中订阅指定主题的日志消息，将日志数据保存到ElasticSearch全文搜索引擎或本地文件中，并对本地存储的日志文件进行归档处理。

策略管理模块负责完成日志源管理、数据上报策略配置、日志存储和日志归档策略配置等，将配置信息保存到MySQL数据库。系统管理模块通过查询ElasticSearch全文搜索引擎提供日志数据的查询功能。

4.2 应用案例

本文所设计系统已在多地泛政府行业部门进行实际验证，用于泛政府行业信息内网和新一代移动警务平台中应用、云平台、终端、边界、网络、安全基础设施的全面日志采集，具体方案如下：

图13 某省泛政府行业信息网统一日志采集系统部署拓扑

图14 某省移动警务平台统一日志采集系统部署拓扑

4.2.1 新一代泛政府行业信息网部署

如图13所示，某省泛政府行业信息网按业务功能划分为用户接入和数据接入两大子网，以及子网和外部网络之间的用户访问和数据交换通道，为实现全面日志采集，需要在三个不同网域分别部署一套统一日志采集系统。其中数据中心采用云化部署模式，采集数据中心内主机、网络、安全设备和应用系统日志；用户接入区采用软硬一体部署模式，采集用户接入区内终端、网络、安全设备和应用系统日志；安全访问与数据交换通道采用软硬一体部署模式，采集网络、安全设备和应用系统日志[20-22]。

4.2.2 新一代移动警务平台部署

如图14所示，某省新一移动警务平台按业务功能划分移动互联网服务子平台(I区)、联网服务子平台(II区)、泛政府行业信息网服务子平台(III区)，以及移动安全接入子平台、移动互联网隔离交换区。为实现全过程日志采集，需要在三个不同子区域分别部署一套统一日志采集系统，均采用软硬一体化部署模式。其中I区统一采集系统负责，采集I区内主机、网络、安全设备和应用系统日志；II区统一日志采集系统负责采集II区内终端、网络、安全设备和应用系统日志，同时采集两个隔离交换区内安全设备日志；III区统一日志采集III区内网络、安全设备和应用系统日志[23-25]。

5 结束语

当前，全国泛政府行业均以习近平总书记“以安全保发展、以发展促安全”新时代网络安全思想为指引，牢固确立以人民为中心的发展思想，全面贯彻总体国家安全观，全面实施泛政府行业大数据战略，坚持大数据建设应用与安全防护同步规划实施，加快构建大数据安全保障体系。本文提出的统一日志采集系统，采用标准化接口和插件技术，全面提升泛政府行业大数据安全运行管理平台中日志采集的稳定性和可扩展性。同时，采用基于消息队列的流式处理架构，实现日志采集、日志处理、日志上报等各个环节的解耦，并支持灵活的功能扩展；通过消息队列的流量削峰，保证日志传输的可靠性，有效解决不同网域间海量日志数据的统一采集，进而为安全运行管理平台提供数据支撑、安全分析和智能决策。