ELK 到 AWS：輕鬆管理日誌

已發表: 2022-03-11

Elasticsearch 是一個強大的軟件解決方案，旨在快速搜索大量數據中的信息。 與 Logstash 和 Kibana 結合，形成了非正式的“ELK 堆棧”，通常用於收集、臨時存儲、分析和可視化日誌數據。 通常還需要一些其他軟件，例如 Filebeat 將日誌從服務器發送到 Logstash，Elastalert 根據對存儲在 Elasticsearch 中的數據運行的一些分析結果生成警報。

ELK Stack 很強大，但是……

我使用 ELK 管理日誌的經驗非常複雜。 一方面，它非常強大，而且它的能力範圍令人印象深刻。 另一方面，設置起來很棘手，維護起來也很頭疼。

事實是，Elasticsearch 總體來說非常好，可以用在各種各樣的場景中； 它甚至可以用作搜索引擎！ 由於它不是專門用於管理日誌數據，因此需要更多的配置工作來自定義其行為以滿足管理此類數據的特定需求。

設置 ELK 集群非常棘手，需要我嘗試一些參數才能最終啟動並運行。 然後是配置它的工作。 就我而言，我需要配置五種不同的軟件（Filebeat、Logstash、Elasticsearch、Kibana 和 Elastalert）。 這可能是一項相當乏味的工作，因為我必須通讀文檔並調試不與下一個對話的鏈中的一個元素。 即使在您最終啟動並運行集群之後，您仍然需要對其執行日常維護操作：打補丁、升級操作系統包、檢查 CPU、RAM 和磁盤使用情況，根據需要進行細微調整等。

Logstash 更新後，我的整個 ELK 堆棧停止工作。 經過仔細檢查，結果發現，出於某種原因，ELK 開發人員決定更改其配置文件中的關鍵字並將其複數。 那是最後一根稻草，並決定尋找更好的解決方案（至少是針對我的特定需求的更好解決方案）。

我想存儲 Apache 和各種 PHP 和節點應用程序生成的日誌，並解析它們以找到指示軟件錯誤的模式。 我找到的解決方案如下：

  • 在目標上安裝 CloudWatch 代理。
  • 配置 CloudWatch 代理以將日誌發送到 CloudWatch 日誌。
  • 觸發 Lambda 函數的調用以處理日誌。
  • 如果找到模式，Lambda 函數會將消息發佈到 Slack 通道。
  • 在可能的情況下，將篩選器應用於 CloudWatch 日誌組，以避免為每個日誌調用 Lambda 函數（這可能會很快增加成本）。

而且，在高層次上，就是這樣！ 一個 100% 無服務器解決方案，無需任何維護即可正常工作，並且無需任何額外工作即可很好地擴展。 這種無服務器解決方案相對於服務器集群的優勢有很多：

  • 本質上，您將定期在集群服務器上執行的所有日常維護操作現在都由雲提供商負責。 任何底層服務器都會在你不知情的情況下為你打補丁、升級和維護。
  • 您不再需要監控集群，並將所有擴展問題委託給雲提供商。 實際上，如上所述的無服務器設置將自動擴展，您無需執行任何操作！
  • 上述解決方案需要較少的配置，並且云提供商不太可能將重大更改帶入配置格式。
  • 最後，編寫一些 CloudFormation 模板將所有內容作為基礎架構即代碼非常容易。 做同樣的事情來建立一個完整的 ELK 集群需要很多工作。

配置 Slack 警報

所以現在讓我們進入細節！ 讓我們探索一下 CloudFormation 模闆對於此類設置的外觀，以及用於提醒工程師的 Slack webhook。 我們需要先配置所有的 Slack 設置，所以讓我們深入研究一下。

 AWSTemplateFormatVersion: 2010-09-09 Description: Setup log processing Parameters: SlackWebhookHost: Type: String Description: Host name for Slack web hooks Default: hooks.slack.com SlackWebhookPath: Type: String Description: Path part of the Slack webhook URL Default: /services/YOUR/SLACK/WEBHOOK

您需要為此設置 Slack 工作區，請查看此 WebHooks for Slack 指南以獲取更多信息。

創建 Slack 應用程序並配置傳入掛鉤後，掛鉤 URL 將成為 CloudFormation 堆棧的參數。

 Resources: ApacheAccessLogGroup: Type: AWS::Logs::LogGroup Properties: RetentionInDays: 100 # Or whatever is good for you ApacheErrorLogGroup: Type: AWS::Logs::LogGroup Properties: RetentionInDays: 100 # Or whatever is good for you

在這裡，我們創建了兩個日誌組：一個用於Apache 訪問日誌，另一個用於Apache 錯誤日誌。

我沒有為日誌數據配置任何生命週期機制，因為它超出了本文的範圍。 在實踐中，您可能希望縮短保留窗口並設計 S3 生命週期策略以在一段時間後將它們移至 Glacier。

用於處理訪問日誌的 Lambda 函數

現在讓我們實現將處理 Apache 訪問日誌的 Lambda 函數。

 BasicLambdaExecutionRole: Type: AWS::IAM::Role Properties: AssumeRolePolicyDocument: Version: 2012-10-17 Statement: - Effect: Allow Principal: Service: lambda.amazonaws.com Action: sts:AssumeRole ManagedPolicyArns: - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole

在這裡，我們創建了一個將附加到 Lambda 函數的 IAM 角色，以允許它們執行其職責。 實際上， AWSLambdaBasicExecutionRole是（儘管它的名稱）由 AWS 提供的 IAM 策略。 它只允許 Lambda 函數創建其日誌組並在該組中創建日誌流，然後將其自己的日誌發送到 CloudWatch Logs。

 ProcessApacheAccessLogFunction: Type: AWS::Lambda::Function Properties: Handler: index.handler Role: !GetAtt BasicLambdaExecutionRole.Arn Runtime: python3.7 Timeout: 10 Environment: Variables: SLACK_WEBHOOK_HOST: !Ref SlackWebHookHost SLACK_WEBHOOK_PATH: !Ref SlackWebHookPath Code: ZipFile: | import base64 import gzip import json import os from http.client import HTTPSConnection def handler(event, context): tmp = event['awslogs']['data'] # `awslogs.data` is base64-encoded gzip'ed JSON tmp = base64.b64decode(tmp) tmp = gzip.decompress(tmp) tmp = json.loads(tmp) events = tmp['logEvents'] for event in events: raw_log = event['message'] log = json.loads(raw_log) if log['status'][0] == "5": # This is a 5XX status code print(f"Received an Apache access log with a 5XX status code: {raw_log}") slack_host = os.getenv('SLACK_WEBHOOK_HOST') slack_path = os.getenv('SLACK_WEBHOOK_PATH') print(f"Sending Slack post to: host={slack_host}, path={slack_path}, url={url}, content={raw_log}") cnx = HTTPSConnection(slack_host, timeout=5) cnx.request("POST", slack_path, json.dumps({'text': raw_log})) # It's important to read the response; if the cnx is closed too quickly, Slack might not post the msg resp = cnx.getresponse() resp_content = resp.read() resp_code = resp.status assert resp_code == 200

所以在這裡我們定義一個 Lambda 函數來處理 Apache 訪問日誌。 請注意，我沒有使用 Apache 默認的通用日誌格式。 我像這樣配置了訪問日誌格式（您會注意到它本質上會生成格式為 JSON 的日誌，這使得後續處理變得更加容易）：

 LogFormat "{\"vhost\": \"%v:%p\", \"client\": \"%a\", \"user\": \"%u\", \"timestamp\": \"%{%Y-%m-%dT%H:%M:%S}t\", \"request\": \"%r\", \"status\": \"%>s\", \"size\": \"%O\", \"referer\": \"%{Referer}i\", \"useragent\": \"%{User-Agent}i\"}" json

此 Lambda 函數是用 Python 3 編寫的。它採用從 CloudWatch 發送的日誌行並可以搜索模式。 在上面的示例中，它僅檢測導致 5XX 狀態代碼的 HTTP 請求並將消息發佈到 Slack 通道。

您可以在模式檢測方面做任何您喜歡的事情，而且它是一種真正的編程語言 (Python)，而不僅僅是 Logstash 或 Elastalert 配置文件中的正則表達式模式，這為您提供了很多實現複雜模式識別的機會.

修訂控制

關於修訂控制的簡短說明：我發現在 CloudFormation 模板中為小型實用程序 Lambda 函數（例如這個函數）內嵌代碼是完全可以接受和方便的。 當然，對於涉及許多 Lambda 函數和層的大型項目，這很可能不方便，您需要使用 SAM。

 ApacheAccessLogFunctionPermission: Type: AWS::Lambda::Permission Properties: FunctionName: !Ref ProcessApacheAccessLogFunction Action: lambda:InvokeFunction Principal: logs.amazonaws.com SourceArn: !Sub arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:*

以上內容允許 CloudWatch Logs 調用您的 Lambda 函數。 提醒一句：我發現使用SourceAccount屬性會導致與SourceArn發生衝突。

一般來說，當調用 Lambda 函數的服務在同一個 AWS 賬戶中時，我建議不要包含它。 SourceArn都會禁止其他賬戶調用 Lambda 函數。

 ApacheAccessLogSubscriptionFilter: Type: AWS::Logs::SubscriptionFilter DependsOn: ApacheAccessLogFunctionPermission Properties: LogGroupName: !Ref ApacheAccessLogGroup DestinationArn: !GetAtt ProcessApacheAccessLogFunction.Arn FilterPattern: "{$.status = 5*}"

訂閱篩選器資源是 CloudWatch Logs 和 Lambda 之間的鏈接。 在這裡，發送到ApacheAccessLogGroup的日誌將被轉發到我們上面定義的 Lambda 函數，但只有那些通過過濾模式的日誌。 在這裡，過濾器模式需要一些 JSON 作為輸入（過濾器模式以 '{' 開頭並以 '}' 結尾），並且僅當它具有以“5”開頭的字段status時才會匹配日誌條目。

這意味著我們只有在 Apache 返回的 HTTP 狀態碼是 500 碼時才調用 Lambda 函數，這通常意味著正在發生一些非常糟糕的事情。 這樣可以確保我們不會過多地調用 Lambda 函數，從而避免不必要的成本。

有關過濾模式的更多信息，請參閱 Amazon CloudWatch 文檔。 CloudWatch 過濾器模式非常好，但顯然不如 Grok 強大。

請注意DependsOn字段，它確保 CloudWatch Logs 可以在創建訂閱之前實際調用 Lambda 函數。 這只是蛋糕上的一顆櫻桃，它很可能是不必要的，因為在真實情況下，Apache 可能不會在至少幾秒鐘之前收到請求（例如：將 EC2 實例與負載均衡器鏈接，並獲取負載平衡器將 EC2 實例的狀態識別為健康）。

用於處理錯誤日誌的 Lambda 函數

現在讓我們看一下將處理 Apache 錯誤日誌的 Lambda 函數。

 ProcessApacheErrorLogFunction: Type: AWS::Lambda::Function Properties: Handler: index.handler Role: !GetAtt BasicLambdaExecutionRole.Arn Runtime: python3.7 Timeout: 10 Environment: Variables: SLACK_WEBHOOK_HOST: !Ref SlackWebHookHost SLACK_WEBHOOK_PATH: !Ref SlackWebHookPath Code: ZipFile: | import base64 import gzip import json import os from http.client import HTTPSConnection def handler(event, context): tmp = event['awslogs']['data'] # `awslogs.data` is base64-encoded gzip'ed JSON tmp = base64.b64decode(tmp) tmp = gzip.decompress(tmp) tmp = json.loads(tmp) events = tmp['logEvents'] for event in events: raw_log = event['message'] log = json.loads(raw_log) if log['level'] in ["error", "crit", "alert", "emerg"]: # This is a serious error message msg = log['msg'] if msg.startswith("PHP Notice") or msg.startswith("PHP Warning"): print(f"Ignoring PHP notices and warnings: {raw_log}") else: print(f"Received a serious Apache error log: {raw_log}") slack_host = os.getenv('SLACK_WEBHOOK_HOST') slack_path = os.getenv('SLACK_WEBHOOK_PATH') print(f"Sending Slack post to: host={slack_host}, path={slack_path}, url={url}, content={raw_log}") cnx = HTTPSConnection(slack_host, timeout=5) cnx.request("POST", slack_path, json.dumps({'text': raw_log})) # It's important to read the response; if the cnx is closed too quickly, Slack might not post the msg resp = cnx.getresponse() resp_content = resp.read() resp_code = resp.status assert resp_code == 200

第二個 Lambda 函數處理 Apache 錯誤日誌，並且僅在遇到嚴重錯誤時才會向 Slack 發布消息。 在這種情況下，PHP 通知和警告消息的嚴重程度不足以觸發警報。

同樣，此函數要求 Apache 錯誤日誌為 JSON 格式。 所以這是我一直在使用的錯誤日誌格式字符串：

 ErrorLogFormat "{\"vhost\": \"%v\", \"timestamp\": \"%{cu}t\", \"module\": \"%-m\", \"level\": \"%l\", \"pid\": \"%-P\", \"tid\": \"%-T\", \"oserror\": \"%-E\", \"client\": \"%-a\", \"msg\": \"%M\"}"
 ApacheErrorLogFunctionPermission: Type: AWS::Lambda::Permission Properties: FunctionName: !Ref ProcessApacheErrorLogFunction Action: lambda:InvokeFunction Principal: logs.amazonaws.com SourceArn: !Sub arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:* SourceAccount: !Ref AWS::AccountId

此資源授予 CloudWatch Logs 調用您的 Lambda 函數的權限。

 ApacheErrorLogSubscriptionFilter: Type: AWS::Logs::SubscriptionFilter DependsOn: ApacheErrorLogFunctionPermission Properties: LogGroupName: !Ref ApacheErrorLogGroup DestinationArn: !GetAtt ProcessApacheErrorLogFunction.Arn FilterPattern: '{$.msg != "PHP Warning*" && $.msg != "PHP Notice*"}'

最後，我們使用 Apache 錯誤日誌組的訂閱過濾器將 CloudWatch Logs 與 Lambda 函數鏈接起來。 請注意過濾模式，它確保以“PHP 警告”或“PHP 通知”開頭的消息的日誌不會觸發對 Lambda 函數的調用。

最後的想法、定價和可用性

關於成本的最後一句話：這個解決方案比運行 ELK 集群便宜得多。 存儲在 CloudWatch 中的日誌定價與 S3 相同，Lambda 允許每月 100 萬次調用作為其免費套餐的一部分。 對於具有中等到高流量的網站（假設您使用 CloudWatch Logs 過濾器），這可能就足夠了，尤其是如果您編碼得當並且沒有太多錯誤！

此外，請注意 Lambda 函數最多支持 1,000 個並發調用。 在撰寫本文時，這是 AWS 中無法更改的硬限制。 但是，您可以預期對上述函數的調用將持續大約 30-40 毫秒。 這應該足夠快以處理相當大的流量。 如果您的工作量如此之大以至於您達到了這個限制，您可能需要一個基於 Kinesis 的更複雜的解決方案，我可能會在以後的文章中介紹。

進一步閱讀 Toptal 工程博客：

  • 使用 AWS SSM 的 SSH 日誌記錄和會話管理