1 背景和問題

隨著雲端計算、PaaS平臺的普及，虛擬化、容器化等技術的應用，例如Docker等技術，越來越多的服務會部署在雲端。通常，我們需要需要獲取日誌，來進行監控、分析、預測、統計等工作，但是雲端的服務不是物理的固定資源，日誌獲取的難度增加了，以往可以SSH登陸的或者FTP獲取的，現在可不那麼容易獲得，但這又是工程師迫切需要的，最典型的場景便是：上線過程中，一切都在GUI化的PaaS平臺點點滑鼠完成，但是我們需要結合tail -F、grep等命令來觀察日誌，判斷是否上線成功。當然這是一種情況，完善的PaaS平臺會為我們完成這個工作，但是還有非常多的ad-hoc的需求，PaaS平臺無法滿足我們，我們需要日誌。本文就給出了在分散式環境下，容器化的服務中的分散日誌，如何集中收集的一種方法。

2 設計約束和需求描述

做任何設計之前，都需要明確應用場景、功能需求和非功能需求。

2.1 應用場景

分散式環境下可承載百臺伺服器產生的日誌，單條資料日誌小於1k，最大不超過50k，日誌總大小每天小於500G。

2.2 功能需求

1）集中收集所有服務日誌。

2）可區分來源，按服務、模組和天粒度切分。

2.3 非功能需求

1）不侵入服務程序，收集日誌功能需獨立部署，佔用系統資源可控。

2）實時性，低延遲，從產生日誌到集中儲存延遲小於4s。

3）持久化，保留最近N天。

4）儘量遞送日誌即可，不要求不丟不重，但比例應該不超過一個閾值（例如萬分之一）。

4）可以容忍不嚴格有序。

5）收集服務屬於線下離線功能，可用性要求不高，全年滿足3個9即可。

3 實現架構

一種方案實現的架構如下圖所示：

3.1 Producer層分析

PaaS平臺內的服務假設部署在Docker容器內，那麼為了滿足非功能需求#1，獨立另外一個程序負責收集日誌，因此不侵入服務框架和程序。採用Flume NG來進行日誌的收集，這個開源的元件非常強大，可以看做一種監控、生產增量，並且可以釋出、消費的模型，Source就是源，是增量源，Channel是緩衝通道，這裡使用記憶體佇列緩衝區，Sink就是槽，是個消費的地方。容器內的Source就是執行tail -F這個命令的去利用linux的標準輸出讀取增量日誌，Sink是一個Kafka的實現，用於推送訊息到分散式訊息中介軟體。

3.2 Broker層分析

PaaS平臺內的多個容器，會存在多個Flume NG的客戶端去推送訊息到Kafka訊息中介軟體。Kafka是一個吞吐量、效能非常高的訊息中介軟體，採用單個分割槽按照順序的寫入的方式工作，並且支援按照offset偏移量隨機讀取的特性，因此非常適合做topic釋出訂閱模型的實現。這裡圖中有多個Kafka，是因為支援叢集特性，容器內的Flume NG客戶端可以連線若干個Kafka的broker釋出日誌，也可以理解為連線若干個topic下的分割槽，這樣可以實現高吞吐，一來可以在Flume NG內部做打包批量傳送來減輕QPS壓力，二來可以分散到多個分割槽寫入，同時Kafka還會指定replica備份個數，保證寫入某個master後還需要寫入N個備份，這裡設定為2，沒有采用常用的分散式系統的3，是因為儘量保證高併發特性，滿足非功能需求中的#4。

3.3 Consumer層分析

消費Kafka增量的也是一個Flume NG，可以看出它的強大之處，在於可以接入任意的資料來源，都是可插拔的實現，通過少量配置即可。這裡使用Kafka Source訂閱topic，收集過來的日誌同樣先入記憶體緩衝區，之後使用一個File Sink寫入檔案，為了滿足功能需求#2，可區分來源，按服務、模組和天粒度切分，我自己實現了一個Sink，叫做RollingByTypeAndDayFileSink，原始碼放到了github上，可以從這個頁面下載jar，直接放到flume的lib目錄即可。

4 實踐方法

4.1 容器內配置

Dockerfile

Dockerfile是容器內程式的執行指令碼，裡面會含有不少docker自帶的命令，下面是要典型的Dockerfile，BASE_IMAGE是一個包含了執行程式以及flume bin的映象，比較重要的就是ENTRYPOINT，主要利用supervisord來保證容器內程序的高可用。

 FROM ${BASE_IMAGE}
MAINTAINER ${MAINTAINER}
ENV REFRESH_AT ${REFRESH_AT}
RUN mkdir -p /opt/${MODULE_NAME}
ADD ${PACKAGE_NAME} /opt/${MODULE_NAME}/
COPY service.supervisord.conf /etc/supervisord.conf.drvice.supervisord.conf
COPY supervisor-msoa-wrapper.sh /opt/${MODULE_NAME}/supervisor-msoa-wrapper.sh
RUN chmod +x /opt/${MODULE_NAME}/supervisor-msoa-wrapper.sh
RUN chmod +x /opt/${MODULE_NAME}/*.sh
EXPOSE
ENTRYPOINT ['/usr/bin/supervisord', '-c', '/etc/supervisord.conf'] 

下面是supervisord的配置檔案，執行supervisor-msoa-wrapper.sh指令碼。

[program:${MODULE_NAME}]

command=/opt/${MODULE_NAME}/supervisor-msoa-wrapper.sh

下面是supervisor-msoa-wrapper.sh，這個指令碼內的start.sh或者stop.sh就是應用程式的啟動和停止指令碼，這裡的背景是我們的啟停的指令碼都是在後臺執行的，因此不會阻塞當前程序，因此直接退出了，Docker就會認為程式結束，因此應用生命週期也結束，這裡使用wait命令來進行一個阻塞，這樣就可以保證即使後臺執行的程序，我們可以看似是前臺跑的。

這裡加入了flume的執行命令，–conf後面的引數標示會去這個資料夾下面尋找flume-env.sh，裡面可以定義JAVA_HOME和JAVA_OPTS。–conf-file指定flume實際的source、channel、sink等的配置。

#! /bin/bash

function shutdown()

{

    date

    echo 'Shutting down Service'

    unset SERVICE_PID # Necessary in some cases

    cd /opt/${MODULE_NAME}

    source stop.sh

}

## 停止程序

cd /opt/${MODULE_NAME}

echo 'Stopping Service'

source stop.sh

## 啟動程序

echo 'Starting Service'

source start.sh

export SERVICE_PID=$!

## 啟動Flume NG agent，等待4s日誌由start.sh生成

sleep 4 

nohup /opt/apache-flume-1.6.0-bin/bin/flume-ng agent --conf /opt/apache-flume-1.6.0-bin/conf --conf-file /opt/apache-flume-1.6.0-bin/conf/logback-to-kafka.conf --name a1 -Dflume.root.logger=INFO,console &

# Allow any signal which would kill a process to stop Service

trap shutdown HUP INT QUIT ABRT KILL ALRM TERM TSTP

echo 'Waiting for $SERVICE_PID'

wait $SERVICE_PID

Flume配置

source本應該採用exec source，執行tailf -F日誌檔案即可。但是這裡使用了一個自行開發的StaticLinePrefixExecSource，原始碼可以在github上找到。之所以採用自定義的，是因為需要將一些固定的資訊傳遞下去，例如服務/模組的名稱以及分散式服務所在容器的hostname，便於收集方根據這個標記來區分日誌。如果這裡你發現為什麼不用flume的攔截器interceptor來做這個工作，加入header中一些KV不就OK了嗎？這是個小坑，我後續會解釋一下。

例如原來日誌的一行為：

[INFO]  2016-03-18 12:59:31,080 [main]  fountain.runner.CustomConsumerFactoryPostProcessor      (CustomConsumerFactoryPostProcessor.java:91)    -Start to init IoC container by loading XML bean definitions from classpath:fountain-consumer-stdout.xml

按照如下配置，那麼實際傳遞給Channel的日誌為：

service1##$$##m1-ocean-1004.cp  [INFO]  2016-03-18 12:59:31,080 [main]  fountain.runner.CustomConsumerFactoryPostProcessor      (CustomConsumerFactoryPostProcessor.java:91)    -Start to init IoC container by loading XML bean definitions from classpath:fountain-consumer-stdout.xml

channel使用記憶體緩衝佇列，大小標識可容乃的日誌條數（event size），事務可以控制一次性從source以及一次性給sink的批量日誌條數，實際內部有個timeout超時，可通過keepAlive引數設定，超時後仍然會推送過去，預設為3s。

sink採用Kafka sink，配置broker的list列表以及topic的名稱，需要ACK與否，以及一次性批量傳送的日誌大小，預設5條一個包，如果併發很大可以把這個值擴大，加大吞吐。

4.2 Broker配置

參考Kafka官方的教程，這裡新建一個名稱叫做keplerlog的topic，備份數量為2，分割槽為4。

 > binfka-topics.sh --create --zookeeper localhost:2181 --replication-factor 2 --partitions 4 --topic keplerlog

製造一些增量資訊，例如如下指令碼，在終端內可以隨便輸入一些字串:

  > binfka-console-producer.sh --broker-list localhost:9092 --topic keplerlog

開啟另外一個終端，訂閱topic，確認可以看到producer的輸入的字串即可，即表示聯通了。

 > binfka-console-consumer.sh --zookeeper localhost:2181 --topic keplerlog --from-beginning

4.3 集中接收日誌配置

Flume配置

首先source採用flume官方提供的KafkaSource，配置好zookeeper的地址，會去找可用的broker list進行日誌的訂閱接收。channel採用記憶體快取佇列。sink由於我們的需求是按照服務名稱和日期切分日誌，而官方提供的預設file roll sink，只能按照時間戳，和時間interval來切分。

定製版RollingByTypeAndDayFileSink

原始碼見github。RollingByTypeAndDayFileSink使用有兩個條件：

1）Event header中必須有timestamp，否則會忽略事件，並且會丟擲{@link InputNotSpecifiedException} 

2）Event body如果是按照##$$##分隔的，那麼把分隔之前的字串當做模組名稱（module name）來處理；如果沒有則預設為default檔名。

輸出到本地檔案，首先要設定一個跟目錄，通過sink.directory設定。其次根據條件#2中提取出來的module name作為檔名稱字首，timestamp日誌作為檔名稱字尾，例如檔名為portal.20150606或者default.20150703。

規整完的一個檔案目錄形式如下，可以看出彙集了眾多服務的日誌，並且按照服務名稱、時間進行了區分：

~/data/kepler-log$ lsauthorization.20160512  default.20160513  default.20160505 portal.20160512       portal.20160505   portal.20160514

不得不提的兩個坑

坑1

回到前兩節提到的自定義了一個StaticLinePrefixExecSource來進行新增一些字首的工作。由於要區分來源的服務/模組名稱，並且按照時間來切分，根據官方flume文件，完全可以採用如下的Source攔截器配置。例如i1表示時間戳，i2表示預設的靜態變數KV，key=module，value=portal。

但是flume官方預設的KafkaSource（v1.6.0）的實現:

可以看出自己重寫了Event header中的KV，丟棄了傳送過來的header，因為這個坑的存在因此，tailf -F在event body中在前面指定模組/服務名稱，然後RollingByTypeAndDayFileSink會按照分隔符切分。否則下游無法能達到KV。

坑2

exec source需要執行tail -F命令來通過標準輸出和標準錯誤一行一行的讀取，但是如果把tail -F封裝在一個指令碼中，指令碼中再執行一些管道命令，例如tail -F logback.log | awk ‘{print 'portal##$$##'$0}’，那麼exec source總是會把最近的輸出丟棄掉，導致追加到檔案末尾的日誌有一些無法總是“姍姍來遲”，除非有新的日誌追加，他們才會被“擠”出來。這個問題比較詭異。暫時沒有細緻研究。以示後人不要採坑。

5 結語

從這個分散式服務分散日誌的集中收集方法，可以看出利用一些開源元件，可以非常方便的解決我們日常工作中所發現的問題，而這個發現問題和解決問題的能力才是工程師的基本素質要求。對於其不滿足需求的，需要具備有鑽研精神，知其然還要知其所以然的去做一些ad-hoc工作，才可以更加好的leverage這些元件。

另外，日誌的收集只是起點，利用寶貴的資料，後面的使用場景和想象空間都會非常大，例如

1）利用Spark streaming在一個時間視窗內計算日誌，做流量控制和訪問限制。

2）使用awk指令碼、scala語言的高階函式做單機的訪問統計分析，或者Hadoop、Spark做大資料的統計分析。

3）除了埠存活和語義監控，利用實時計算處理日誌，做ERROR、異常等資訊的過濾，實現服務真正的健康保障和預警監控。

4）收集的日誌可以通過logstash匯入Elastic Search，使用ELK方式做日誌查詢使用。