公安行業(yè)存在數(shù)以萬計的前后端設備，前端設備包括相機、檢測器及感應器，后端設備包括各級中心機房中的服務器、應用服務器、網(wǎng)絡設備及機房動力系統(tǒng)，數(shù)量巨大、種類繁多的設備給公安內部運維管理帶來了巨大挑戰(zhàn)。傳統(tǒng)通過ICMP/SNMP、Trap/Syslog等工具對設備進行診斷分析的方式已不能滿足實際要求，由于公安內部運維管理的特殊性，現(xiàn)行通過ELK等架構的方式同樣也滿足不了需要。為尋求合理的方案，我們將目光轉向開源架構，構建了一套適用于公安行業(yè)的實時運維管理平臺。

實時運維平臺整體架構

數(shù)據(jù)采集層：Logstash+Flume，負責在不同場景下收集、過濾各類前后端硬件設備輸出的Snmp Trap、Syslog日志信息以及應用服務器自身產(chǎn)生的系統(tǒng)和業(yè)務日志;

數(shù)據(jù)傳輸層：采用高吞吐的分布式消息隊列Kafka集群，保證匯聚的日志、消息的可靠傳輸;

數(shù)據(jù)處理層：由Spark實時Pull Kafka數(shù)據(jù)，通過Spark Streaming以及RDD操作進行數(shù)據(jù)流的處理以及邏輯分析;

數(shù)據(jù)存儲層：實時數(shù)據(jù)存入MySQL中便于實時的業(yè)務應用和展示;全量數(shù)據(jù)存入ES以及HBase中便于后續(xù)的檢索分析;

業(yè)務服務層：基于存儲層，后續(xù)的整體業(yè)務應用涵蓋了APM、網(wǎng)絡監(jiān)控、拓撲、告警、工單、CMDB等。

整體系統(tǒng)涉及的主要開源框架情況如下：

另外，整體環(huán)境基于JDK 8以及Scala 2.10.4。公安系統(tǒng)設備種類繁多，接下來將以交換機Syslog日志為例，詳細介紹日志處理分析的整體流程。

圖1 公安實時運維平臺整體架構

Flume+Logstash日志收集

Flume是Cloudera貢獻的一個分布式、可靠及高可用的海量日志采集系統(tǒng)，支持定制各類Source(數(shù)據(jù)源)用于數(shù)據(jù)收集，同時提供對數(shù)據(jù)的簡單處理以及通過緩存寫入Sink(數(shù)據(jù)接收端)的能力。

Flume中，Source、Channel及Sink的配置如下：

該配置通過syslog source配置localhost tcp 5140端口來接收網(wǎng)絡設備發(fā)送的Syslog信息，event緩存在內存中，再通過KafkaSink將日志發(fā)送到kafka集群中名為“syslog-kafka”的topic中。

Logstash來自Elastic公司，專為收集、分析和傳輸各類日志、事件以及非結構化的數(shù)據(jù)所設計。它有三個主要功能：事件輸入(Input)、事件過濾器(Filter)以及事件輸出(Output)，在后綴為.conf的配置文件中設置，本例中Syslog配置如下：

Input(輸入)插件用于指定各種數(shù)據(jù)源，本例中的Logstash通過udp 514端口接收Syslog信息;

Filter(過濾器)插件雖然在本例中不需要配置，但它的功能非常強大，可以進行復雜的邏輯處理，包括正則表達式處理、編解碼、k/v切分以及各種數(shù)值、時間等數(shù)據(jù)處理，具體可根據(jù)實際場景設置;

Output(輸出)插件用于將處理后的事件數(shù)據(jù)發(fā)送到指定目的地，指定了Kafka的位置、topic以及壓縮類型。在***的Codec編碼插件中，指定來源主機的IP地址(host)、Logstash處理的時間戳(@timestamp)作為前綴并整合原始的事件消息(message)，方便在事件傳輸過程中判斷Syslog信息來源。單條原始Syslog信息流樣例如下：

147>12164: Oct 9 18:04:10.735: %LINK-3-UPDOWN: Interface GigabitEthernet0/16, changed state to down

Logstash Output插件處理后的信息流變成為：

19.1.1.12 2016-10-13T10:04:54.520Z <147>12164: Oct 9 18:04:10.735: %LINK-3-UPDOWN: Interface GigabitEthernet0/16, changed state to down

其中紅色字段就是codec編碼插件植入的host以及timestamp信息。處理后的Syslog信息會發(fā)送至Kafka集群中進行消息的緩存。

Kafka日志緩沖

Kafka是一個高吞吐的分布式消息隊列，也是一個訂閱/發(fā)布系統(tǒng)。Kafka集群中每個節(jié)點都有一個被稱為broker的實例，負責緩存數(shù)據(jù)。Kafka有兩類客戶端，Producer(消息生產(chǎn)者的)和Consumer(消息消費者)。Kafka中不同業(yè)務系統(tǒng)的消息可通過topic進行區(qū)分，每個消息都會被分區(qū)，用以分擔消息讀寫負載，每個分區(qū)又可以有多個副本來防止數(shù)據(jù)丟失。消費者在具體消費某個topic消息時，指定起始偏移量。Kafka通過Zero-Copy、Exactly Once等技術語義保證了消息傳輸?shù)膶崟r、高效、可靠以及容錯性。

Kafka集群中某個broker的配置文件server.properties的部分配置如下：

其中需指定集群里不同broker的id，此臺broker的id為1，默認監(jiān)聽9092端口，然后配置Zookeeper(后續(xù)簡稱zk)集群，再啟動broker即可。

Kafka集群名為syslog-kafka的topic：

Kafka集群的topic以及partition等信息也可以通過登錄zk來觀察。然后再通過下列命令查看Kafka接收到的所有交換機日志信息：

部分日志樣例如下：

Spark日志處理邏輯

Spark是一個為大規(guī)模數(shù)據(jù)處理而生的快速、通用的引擎，在速度、效率及通用性上表現(xiàn)極為優(yōu)異。

在Spark主程序中，通過Scala的正則表達式解析Kafka Source中名為“syslog-kafka” 的topic中的所有Syslog信息，再將解析后的有效字段封裝為結果對象，***通過MyBatis近實時地寫入MySQL中，供前端應用進行實時地可視化展示。另外，全量數(shù)據(jù)存儲進入HBase及ES中，為后續(xù)海量日志的檢索分析及其它更高級的應用提供支持。主程序示例代碼如下：

整體的處理分析主要分為4步：

初始化SparkContext并指定Application的參數(shù);

創(chuàng)建基于Kafka topic “syslog-kafka” 的DirectStream;

將獲取的每一行數(shù)據(jù)映射為Syslog對象，調用Service進行對象封裝并返回;

遍歷RDD，記錄不為空時保存或者更新Syslog信息到MySQL中。

Syslog POJO的部分基本屬性如下：

SwSyslog實體中的基本屬性對應Syslog中的接口信息，注解中的name對應MySQL中的表sw_syslog 以及各個字段，MyBatis完成成員屬性和數(shù)據(jù)庫結構的ORM(對象關系映射)。

程序中的SwSyslogService有兩個主要功能：

encapsulateSwSyslog()將Spark處理后的每一行Syslog通過Scala的正則表達式解析為不同的字段，然后封裝并返回Syslog對象;遍歷RDD分區(qū)生成的每一個Syslog對象中都有ip以及接口信息，saveSwSyslog()會據(jù)此判斷該插入還是更新Syslog信息至數(shù)據(jù)庫。另外，封裝好的Syslog對象通過ORM工具MyBatis與MySQL進行互操作。

獲取到的每一行Syslog信息如之前所述：

這段信息需解析為設備ip、服務器時間、信息序號、設備時間、Syslog類型、屬性、設備接口、接口狀態(tài)等字段。Scala正則解析邏輯如下：

通過正則過濾、Syslog封裝以及MyBatis持久層映射，Syslog接口狀態(tài)信息最終解析如下：

***，諸如APM、網(wǎng)絡監(jiān)控或者告警等業(yè)務應用便可以基于MySQL做可視化展示。

總結

本文首先對公安運維管理現(xiàn)狀做了簡要介紹，然后介紹公安實時運維平臺的整體架構，再以交換機Syslog信息為例，詳細介紹如何使用Flume+Logstash+Kafka+Spark Streaming進行實時日志處理分析，對處理過程中大量的技術細節(jié)進行了描述并通過代碼詳細地介紹整體處理步驟。本文中的示例實時地將數(shù)據(jù)寫入MySQL存在一定的性能瓶頸，后期會對包含本例的相關代碼重構，數(shù)據(jù)將會實時寫入HBase來提高性能。