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

實時運維平臺整體架構

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

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

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

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

業務服務層：基于存儲層，后續的整體業務應用涵蓋了APM、網絡監控、拓撲、告警、工單、CMDB等。

整體系統涉及的主要開源框架情況如下：

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

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

Flume+Logstash日志收集

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

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

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

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

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

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

Output(輸出)插件用于將處理后的事件數據發送到指定目的地，指定了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信息會發送至Kafka集群中進行消息的緩存。

Kafka日志緩沖

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

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

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

Kafka集群名為syslog-kafka的topic：

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

部分日志樣例如下：

Spark日志處理邏輯

Spark是一個為大規模數據處理而生的快速、通用的引擎，在速度、效率及通用性上表現極為優異。

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

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

初始化SparkContext并指定Application的參數;

創建基于Kafka topic “syslog-kafka” 的DirectStream;

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

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

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

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

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

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

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

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

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

***，諸如APM、網絡監控或者告警等業務應用便可以基于MySQL做可視化展示。

總結

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