Beats：使用 Elastic Stack 记录 Python 应用日志

文章转载自：https://elasticstack.blog.csdn.net/article/details/112259500

日志记录实际上是每个应用程序都必须具备的功能。无论你选择基于哪种技术，都需要监视应用程序的运行状况和操作。随着应用程序扩展，这变得越来越困难，你需要查看不同的文件，文件夹甚至服务器来查找所需的信息。虽然你可以使用内置功能从应用程序本身编写 Python 日志，但应将这些日志集中在 Elastic Stack 之类的工具中。

借助 Elasticsearch 筛选大量数据的效率，应用程序开发人员可以快速缩小最重要的日志的范围。仪表板可发送给运营团队，使他们能够在检测到异常行为时迅速做出反应。

本文将重点介绍为Python应用程序构建健壮的应用程序日志记录基础结构。 Python 是一种非常流行且易于使用的通用编程语言。从学习到编程，再到实施复杂的机器学习解决方案，这是各种活动的绝佳选择。

在我之前的文章 “Beats: 使用 Filebeat 进行日志结构化 - Python”，我使用了另外一种方法来记录 Python 日志。

Python 日志概述

Python 随附了一个非常灵活且易于使用的日志记录模块。 与许多日志记录库一样，Python 可以在多个级别进行日志记录（例如，INFO 或 ERROR），以各种方式格式化日志输出，并写入不同的目标位置（例如，控制台或文件）。 实际上，将某些内容记录到文件中很简单：

main.py

import logging
 
logging.basicConfig(filename="app.log", level=logging.DEBUG)
 
logging.info('Application running!')

当我们运行上面的 Python 应用后，就会在应用当前的目录里生成一个叫做 app.log 的文件：

app.log

INFO:root:Application running!

当你向日志中添加更多信息时，有必要以人类可读和机器可解析的格式编写日志。 这称为结构化日志记录。 JSON结构的日志特别容易传送到 Elastic Stack 中。

将 JSON 格式的日志结构化过程与 Python 的日志记录模块集成在一起很容易，该模块提供了处理程序和格式化程序，以分离处理输出目标和格式化日志本身的问题。 通过这种分离，你可以自定义日志从应用程序代码到其目的地的旅程的任何部分。 实际上，python-json-logger 是一个免费的 Python JSON 记录器。 要进行设置，请先通过 pip 安装它：

pip install python-json-logger

或者针对 Python3 的安装：

pip3 install python-json-logger

接下来，你可以使用具有以下结构的配置文件 logging.conf 来设置 JSON 日志记录。这个文件可以放置于项目的当前目录中：

logging.conf

[loggers]
keys = root
 
[logger_root]
level = INFO
handlers = root
 
[handlers]
keys = root
 
[handler_root]
class = FileHandler
level = INFO
formatter = json
args = ('application.log',)
 
[formatters]
keys = json
 
[formatter_json]
class = __main__.ElkJsonFormatter

最后，以下代码允许你编写 JSON 日志：

main.py

import logging
import logging.config
from pythonjsonlogger import jsonlogger
from datetime import datetime;
 
class ElkJsonFormatter(jsonlogger.JsonFormatter):
    def add_fields(self, log_record, record, message_dict):
        super(ElkJsonFormatter, self).add_fields(log_record, record, message_dict)
        now = datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%S.%fZ')
        log_record['@timestamp'] = now
        log_record['level'] = record.levelname
        log_record['logger'] = record.name
 
logging.config.fileConfig('logging.conf')
logger = logging.getLogger("MainLogger")
 
logging.info('Application running!')

此代码加载之前在 logging.config 配置中定义的 ElkJsonFormatter 类。 我们本可以直接使用 JsonFormatte r类（来自 python-json-logger）来生成 JSON 日志。 但是，在这种情况下，我们将设置特定的字段（尤其是 @timestamp），这将使将日志更轻松地发送到 Elasticsearch 产生以下结构：

application.log

{"message": "Application running!", "@timestamp": "2021-01-06T03:39:12.846837Z", "level": "INFO", "logger": "root"}

重新运行我们的应用。我们可以在项目当前目录下发现一个叫做 application.log 的文件。

将 Python 日志传送到 Elastic Stack

一旦我们的日志处于我们可以推理的结构中，就可以将其发送到 Elasticsearch 进行处理并生成我们所需的见解。 当日志为 JSON 格式时，这是最容易设置的，但也可以与其他非 JSON 日志一起使用，只要它们具有足够清晰的结构即可解析。

安装 Elastic Stack

如果你还没有安装好自己的 Elastic Stack，那么请参照我之前的文章 “Elastic：菜鸟上手指南” 安装好自己的 Elasticsearch 以及 Kibana。在我们今天的练习中，我将使用 Filebeat 来把数据导入到 Elastic Stack 中。如果你还没有安装好自己的 Filebeat，请阅读之前的介绍文章：

• Beats：Beats 入门教程 （一）
• Beats：Beats 入门教程 （二）

在上面的两篇文章中，它详述了如何安装 Filebeat。

使用 Filebeat 传送 JSON 日志

由于 Filebeat 天生就具有 JSON 处理器，Filebeat 能够很轻松地将 JSON 日志直接传送到 Elasticsearch 中。 为此，我们只需要创建一个 Filebeat 的配置文件：

filebeat_python_logging.yml

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /Users/liuxg/python/python_logging/application.log
  json:
    keys_under_root: true
    overwrite_keys: true
    message_key: 'message'
    add_error_key: true
 
output.elasticsearch:
  hosts: ["localhost:9200"]
  index: "python_logs"
 
processors:
 - decode_json_fields:
     fields: ['message']
     target: json
 - drop_fields:
     fields: ["ecs", "agent", "log", "input", "host"]
 
setup.ilm.enabled: false
setup.template.name: python_logs
setup.template.pattern: python_logs

上面的 paths 路径需要依据你自己的 log 路径改变而改变。在上面我定义了一个特定的索引名称 python_logs。在 filebeat 的安装目录中，我们运行如下的命令：

./filebeat -e -c filebeat_python_logging.yml 

运行完上面的命令后，我们可以在 Kibana 中使用如下的命令进行查看：

GET _cat/indices
yellow open twitter                         ztCFdrdbTHuooTkIdaDcjw 1 1     6   0  10.3kb  10.3kb
green  open .apm-custom-link                p18qhpfTRP6I0xDBjW9Asw 1 0     0   0    208b    208b
green  open .kibana_task_manager_1          nvc7Qkt4RdaCaqClux2G1Q 1 0     5 310 403.5kb 403.5kb
yellow open python_logs                     RihQ0eqIS1ivAtHvC5B_DQ 1 1     1   0   4.5kb   4.5kb
green  open .apm-agent-configuration        hC534SjHSIKdmyH910AVhA 1 0     0   0    208b    208b
green  open kibana_sample_data_logs         QtVaJBiLRc-qj_A0pqlzKw 1 0 14074   0  10.4mb  10.4mb
green  open .kibana-event-log-7.10.0-000001 16sjjPW9QaKsxGw52Snstw 1 0     3   0  16.4kb  16.4kb
green  open .async-search                   i7qc9VdhTlOvBjAbgYTuKQ 1 0     0   0    228b    228b
green  open .kibana_1                       VTuPMA6TRr-vLIRqQFh64g 1 0    97  25  10.4mb  10.4mb

我们可以发现一个新生成的 python_logs 文件已经生成了。我们可以通过如下的命令来查看文档的内容：

GET python_logs/_search
{
  "took" : 1,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 1,
      "relation" : "eq"
    },
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "python_logs",
        "_type" : "_doc",
        "_id" : "oOTH1XYBli_HPhP3ZESw",
        "_score" : 1.0,
        "_source" : {
          "@timestamp" : "2021-01-06T03:39:12.846Z",
          "level" : "INFO",
          "logger" : "root",
          "message" : "Application running!"
        }
      }
    ]
  }
}

显然，我们把之前的 application.log 的内容已经成功地导入到 Elasticsearch 之中了。