加载中

使用 Filebeat 从 Python 应用程序摄取日志

在本指南中，我们将向您展示如何提取 Python 应用程序的日志，并将其安全地传输到 Elastic Cloud Hosted 部署中。您将设置 Filebeat 以监控采用 Elastic Common Schema (ECS) 格式的 JSON 结构化日志文件。然后，您将在 Kibana 中实时查看这些日志事件的可视化效果。

虽然我们在本示例中使用 Python，但您可以将相同的方法应用于监控多种客户端类型的日志输出。查看可用 ECS 日志插件的列表。我们还使用 Elastic Cloud Hosted 作为日志的目标 Elastic Stack 目的地，但通过一些小的修改，您可以将本指南中的步骤应用于其他部署，例如自托管的 Elastic Stack 和 Elastic Cloud Enterprise。

在本指南中，您将

所需时间：1 小时

要完成本指南中的步骤，您需要具备

  • 一个具有 Elastic for Observability 解决方案视图的 Elastic Cloud Hosted 部署，以及部署创建时提供的超级用户凭据。有关更多详细信息，请参阅创建 Elastic Cloud Hosted 部署。
  • 已安装与 Python 的 ECS 日志库兼容的 Python 版本。有关兼容 Python 版本的列表，请查看该库的 README。
  • 已安装 Python 的 ECS 日志库。

要安装 Python 的 ECS 日志库，请运行

python -m pip install ecs-logging
注意

根据您使用的 Python 版本，您可能需要在 Python 虚拟环境中安装该库。

在此步骤中，您将创建一个使用 Python 标准 logging 模块以 JSON 格式生成日志的 Python 脚本。

  1. 在本地目录中，创建一个名为 elvis.py 的新文件，并保存以下内容

    #!/usr/bin/python

import logging
import ecs_logging
import time
from random import randint

#logger = logging.getLogger(__name__)
logger = logging.getLogger("app")
logger.setLevel(logging.DEBUG)
handler = logging.FileHandler('elvis.json')
handler.setFormatter(ecs_logging.StdlibFormatter())
logger.addHandler(handler)

print("Generating log entries...")

messages = [
    "Elvis has left the building.",#
    "Elvis has left the oven on.",
    "Elvis has two left feet.",
    "Elvis was left out in the cold.",
    "Elvis was left holding the baby.",
    "Elvis left the cake out in the rain.",
    "Elvis came out of left field.",
    "Elvis exited stage left.",
    "Elvis took a left turn.",
    "Elvis left no stone unturned.",
    "Elvis picked up where he left off.",
    "Elvis's train has left the station."
    ]

while True:
    random1 = randint(0,15)
    random2 = randint(1,10)
    if random1 > 11:
        random1 = 0
    if(random1<=4):
        logger.info(messages[random1], extra={"http.request.body.content": messages[random1]})
    elif(random1>=5 and random1<=8):
        logger.warning(messages[random1], extra={"http.request.body.content": messages[random1]})
    elif(random1>=9 and random1<=10):
        logger.error(messages[random1], extra={"http.request.body.content": messages[random1]})
    else:
        logger.critical(messages[random1], extra={"http.request.body.content": messages[random1]})
    time.sleep(random2)

    该 Python 脚本会持续随机生成十二条日志消息中的一条，随机间隔为 1 到 10 秒。日志消息被写入 elvis.json 文件，每条消息都包含时间戳、info、warning、error 或 critical 的日志级别以及其他数据。为了增加日志数据的差异性，将 info 消息“Elvis has left the building”（猫王已离开大楼）设置为最可能的日志事件。

    为简单起见，这里只有一个日志文件 (elvis.json)，它被写入 elvis.py 所在的本地目录。在生产环境中，您可能有多个与不同模块和记录器关联的日志文件，并且很可能存储在 /var/log 或类似目录中。要了解有关在 Python 中配置日志记录的更多信息，请查看 Python 的日志记录工具。

    将日志以带有 ECS 字段的 JSON 格式写入，可以方便解析和分析，并实现与其他应用程序的标准化。随着日志中捕获的数据量和类型的增加，一种标准且易于解析的格式变得越来越重要。

    除了每个日志条目包含的标准字段外，还有一个额外的 http.request.body.content 字段。这个额外的字段旨在为您提供一些有趣的附加数据以供处理，并演示如何向日志数据添加可选字段。查看 ECS 字段参考以获取可用字段的完整列表。

  2. 让我们测试一下这个 Python 脚本。在保存 elvis.py 的位置打开一个终端实例，并运行以下命令

    python elvis.py

    脚本运行约 15 秒后，输入 CTRL + C 停止它。查看新生成的 elvis.json 文件。它应该包含一个或多个像这样的条目

    {"@timestamp":"2025-06-16T02:19:34.687Z","log.level":"info","message":"Elvis has left the building.","ecs":{"version":"1.6.0"},"http":{"request":{"body":{"content":"Elvis has left the building."}}},"log":{"logger":"app","origin":{"file":{"line":39,"name":"elvis.py"},"function":"<module>"},"original":"Elvis has left the building."},"process":{"name":"MainProcess","pid":3044,"thread":{"id":4444857792,"name":"MainThread"}}}

  3. 在确认 elvis.py 运行符合预期后，您可以删除 elvis.json。

要连接到您的 Elastic Cloud Hosted 部署、流式传输数据并发出查询，您必须使用部署的 Cloud ID 指定连接详细信息，并且必须使用基本身份验证 (basic authentication) 或 API 密钥 (API key) 进行身份验证。

要查找您部署的 Cloud ID，请转到 Kibana 主菜单，然后选择 Management（管理） → Integrations（集成） → Connection details（连接详细信息）。Cloud ID 值格式为 deployment-name:hash。保存此值以便稍后使用。

要进行身份验证并将数据发送到 Elastic Cloud Hosted，您可以使用创建部署时保存的用户名和密码。我们在 配置 Filebeat 以访问 Elastic Cloud Hosted 部分中使用此方法来设置 Filebeat 连接。

您还可以通过 Elastic Cloud Hosted 控制台生成 API 密钥，并配置 Filebeat 使用新密钥安全地连接到您的部署。API 密钥是连接到生产环境的首选方法。

为 Filebeat 创建 API 密钥

  1. 登录 Elastic Cloud Console，然后选择您的部署。

  2. 在主菜单中，转到 Developer tools（开发工具）。

  3. 输入以下请求

    POST /_security/api_key
{
 "name": "filebeat-api-key",
 "role_descriptors": {
   "logstash_read_write": {
     "cluster": ["manage_index_templates", "monitor"],
     "index": [
       {
         "names": ["filebeat-*"],
         "privileges": ["create_index", "write", "read", "manage"]
       }
     ]
   }
 }
}

    此请求创建了一个具有 monitor 集群权限的 API 密钥，该权限提供用于确定集群状态的只读访问权限，以及 manage_index_templates 权限，该权限允许对索引模板进行所有操作。其他权限允许对指定索引 (filebeat-*) 进行 create_index、write 和 manage 操作。添加索引 manage 权限是为了启用索引刷新。

  4. 单击 ▶ 运行请求。输出应类似于以下内容

    {
  "api_key": "tV1dnfF-GHI59ykgv4N0U3",
  "id": "2TBR42gBabmINotmvZjv",
  "name": "filebeat-api-key"
}

在 配置 API 密钥 部分了解如何设置 Filebeat 配置中的 API 密钥。

Filebeat 提供了一种直接、易于配置的方法来监控您的 Python 日志文件，并将日志数据移植到您的部署中。

下载 Filebeat，然后在您创建 elvis.py 脚本的机器上解压它。

转到解压 Filebeat 的目录，并打开 filebeat.yml 配置文件。在 Elastic Cloud 部分，进行以下修改以设置基本身份验证

# =============================== Elastic Cloud ================================

# These settings simplify using Filebeat with the Elastic Cloud (https://cloud.elastic.co/).

# The cloud.id setting overwrites the `output.elasticsearch.hosts` and
# `setup.kibana.host` options.
# You can find the `cloud.id` in the Elastic Cloud web UI.
cloud.id: deployment-name:hash

# The cloud.auth setting overwrites the `output.elasticsearch.username` and
# `output.elasticsearch.password` settings. The format is `<user>:<pass>`.
cloud.auth: username:password
  1. 取消注释 cloud.id 行，并将部署的 Cloud ID 添加为键的值。cloud.id 值格式为 deployment-name:hash。通过转到 Kibana 主菜单并选择 Management → Integrations → Connection details 来查找您的 Cloud ID。
  2. 取消注释 cloud.auth 行，并以 username:password 的格式添加部署的用户名和密码。例如，cloud.auth: elastic:57ugj782kvkwmSKg8uVe。
注意

作为使用 cloud.id 和 cloud.auth 配置连接的替代方法，您可以直接在 Elasticsearch 输出中指定 Elasticsearch URL 和身份验证详细信息。这在连接到不同部署类型（例如自托管集群）时非常有用。

要使用 API 密钥 进行身份验证，请保持 cloud.auth 行的注释状态，因为 Filebeat 将使用 API 密钥而不是部署凭据进行身份验证。

在 filebeat.yml 的 output.elasticsearch 部分，取消注释 api_key 行，并添加您为 Filebeat 创建的 API 密钥。该值的格式为 id:api_key，其中 id 和 api_key 是 创建 API 密钥 API 返回的值。

使用我们之前使用的 POST 请求返回的示例值，API 密钥身份验证的配置如下所示

cloud.id: my-deployment:yTMtd5VzdKEuP2NwPbNsb3VkLtKzLmldJDcyMzUyNjBhZGP7MjQ4OTZiNTIxZTQyOPY2C2NeOGQwJGQ2YWQ4M5FhNjIyYjQ9ODZhYWNjKDdlX2Yz4ELhRYJ7
#cloud.auth:

output.elasticsearch:
  ...
  api_key: "2TBR42gBabmINotmvZjv:tV1dnfF-GHI59ykgv4N0U3"

Filebeat 有多种收集日志的方法。在此示例中，您将手动配置日志收集。在 filebeat.yml 的 filebeat.inputs 部分

filebeat.inputs:

# Each - is an input. Most options can be set at the input level, so
# you can use different inputs for various configurations.
# Below are the input-specific configurations.

# filestream is an input for collecting log messages from files.
- type: filestream

  # Unique ID among all inputs, an ID is required.
  id: my-filestream-id

  # Change to true to enable this input configuration.
  enabled: true

  # Paths that should be crawled and fetched. Glob based paths.
  paths:
    - /path/to/log/files/*.json
  1. 将 enabled 设置为 true。
  2. 将 paths 设置为您的日志文件位置。在此示例中，将 paths 设置为您保存 elvis.py 的目录。

您可以使用通配符 (*) 字符来指示应读取指定目录中的所有日志文件。您还可以使用通配符从多个目录读取日志。例如，/var/log/*/*.log。

Filebeat 的 filestream 输入配置包括用于解码 JSON 消息格式日志的多个选项。您可以在 parsers.ndjson 中设置这些选项。Filebeat 逐行处理日志，因此它们每行包含一个 JSON 对象非常重要。

在此示例中，将 Filebeat 设置为使用带有以下解码选项的 ndjson 解析器

parsers:
  - ndjson:
      target: ""
      overwrite_keys: true
      expand_keys: true
      add_error_key: true
      message_key: msg

要了解有关这些设置的更多信息，请查看 Filebeat 参考中的 ndjson 解析器配置选项和 解码 JSON 字段。

将带有设置的解码选项的 parsers.ndjson 追加到 filebeat.yml 的 filebeat.inputs 部分，以便该部分现在看起来像这样

# ============================== Filebeat inputs ===============================

filebeat.inputs:

# Each - is an input. Most options can be set at the input level, so
# you can use different inputs for various configurations.
# Below are the input-specific configurations.

# filestream is an input for collecting log messages from files.
- type: filestream

  # Unique ID among all inputs, an ID is required.
  id: my-filestream-id

  # Change to true to enable this input configuration.
  enabled: true

  # Paths that should be crawled and fetched. Glob based paths.
  paths:
    - /path/to/log/files/*.json
    #- c:\programdata\elasticsearch\logs\*

  parsers:
    - ndjson:
        target: ""
        overwrite_keys: true
        expand_keys: true
        add_error_key: true
        message_key: msg

Filebeat 附带用于解析、索引和可视化数据的预定义资产。要将这些资产加载到您的 Elastic Cloud Hosted 部署上的 Kibana 中，请从 Filebeat 安装目录运行以下命令

./filebeat setup -e
重要提示

根据安装位置、环境和本地权限等变量，您可能需要 更改 filebeat.yml 的所有权。您也可以尝试以 root 用户身份运行命令：sudo ./filebeat setup -e*，或者可以通过运行带有 --strict.perms=false 选项的命令来禁用严格权限检查。

设置过程需要几分钟。如果设置成功，您应该会收到确认消息

Loaded Ingest pipelines

Filebeat 数据视图（以前称为索引模式）现在在 Elasticsearch 中可用。要验证

注意

从 Elastic Stack 8.0 版本开始，Kibana index patterns（索引模式）已重命名为 data views（数据视图）。要了解更多信息，请查看 Kibana 8.0 新特性页面。

  1. 登录 Elastic Cloud Console，然后选择您的部署。
  2. 在主菜单中，选择 Management → Stack Management → Data Views。
  3. 在搜索栏中，搜索 filebeat。您应该在搜索结果中看到 filebeat-*。

Filebeat 现在已设置为收集日志消息并将其流式传输到您的部署。

是时候将一些日志数据发送到 Elasticsearch 了。

在新终端中

  1. 导航到您创建 elvis.py Python 脚本的目录，然后运行它

    python elvis.py

    让脚本运行几分钟，然后确保生成了 elvis.json 文件并填充了多个日志条目。

  2. 通过从 Filebeat 安装目录运行以下命令来启动 Filebeat

    ./filebeat -c filebeat.yml -e

    在此命令中

    • -e 标志将输出发送到标准错误而不是配置的日志输出。
    • -c 标志指定 Filebeat 配置文件的路径。
    注意

    如果该命令未按预期工作，请查看 Filebeat 快速入门以获取适用于您的操作系统的详细命令语法。您也可以尝试以 root 用户身份运行命令：sudo ./filebeat -c filebeat.yml -e。

    Filebeat 现在应该正在运行并监控 elvis.json 文件的内容。

要确认日志数据已成功发送到您的部署

  1. 登录 Elastic Cloud Console，然后选择您的部署。
  2. 在主菜单中，选择 Management → Stack Management → Data Views。
  3. 在搜索栏中，搜索 filebeat，然后选择 filebeat-*。

filebeat 数据视图显示了字段列表及其详细信息。

现在您可以根据 Python 应用程序日志数据创建可视化

  1. 在主菜单中，选择 Dashboards → Create dashboard。

  2. 选择 Create visualization。 Lens 可视化编辑器将打开。

  3. 在 Data view 下拉框中，如果尚未选择，请选择 filebeat-*。

  4. 在设置可视化类型的菜单中，选择 Bar（柱状图）和 Stacked（堆叠），如果尚未选择。

  5. 检查 时间过滤器是否设置为 Last 15 minutes（过去 15 分钟）。

  6. 从 Available fields（可用字段）列表中，将 @timestamp 字段拖放到可视化构建器上。

  7. 将 log.level 字段拖放到可视化构建器上。

  8. 在 Breakdown（细分）下的图表设置区域中，选择 Top values of log.level。

  9. 将 Number of values（值数量）字段设置为 4，以在图表图例中显示所有四种严重级别。

  10. 选择 Refresh（刷新）。堆叠柱状图现在显示了四种日志严重级别随时间变化的相对频率。

    A screen capture of the Kibana "Bar vertical stacked" visualization with several bars. The X axis shows "Count of records" and the Y axis shows "@timestamp per 30 seconds". Each bar is divided into the four log severity levels.

  11. 选择 Save and return（保存并返回）将此可视化添加到您的仪表板。

让我们创建第二个可视化

  1. 选择 Create visualization。

  2. 在设置可视化类型的菜单中，选择 Bar（柱状图）和 Stacked（堆叠），如果尚未选择。

  3. 从 Available fields（可用字段）列表中，将 @timestamp 字段拖放到可视化构建器上。

  4. 将 http.request.body.content 字段拖放到可视化构建器上。

  5. 在 Breakdown 下的图表设置区域中，选择 Top values of http.request.body.content。

  6. 将 Number of values 设置为 12，以在图表图例中显示所有十二条日志消息。

  7. 选择 Refresh。堆叠柱状图现在显示了每条日志消息随时间变化的相对频率。

    A screen capture of the visualization builder

  8. 选择 Save and return（保存并返回）将此可视化添加到您的仪表板。

现在，创建最后一个可视化

  1. 选择 Create visualization。

  2. 在设置可视化类型的菜单中，选择 Pie（饼图）。

  3. 从 Available fields 列表中，将 log.level 字段拖放到可视化构建器上。出现一个饼图。

    A screen capture of a pie chart divided into four sections

  4. 选择 Save and return（保存并返回）将此可视化添加到您的仪表板。

  5. 选择 Save 并添加标题以保存您的新仪表板。

您现在拥有一个包含三个可视化的 Kibana 仪表板：一个显示每个日志严重级别随时间变化的频率的堆叠柱状图，另一个显示各种消息字符串随时间变化的频率的堆叠柱状图（来自添加的 http.request.body.content 参数），以及一个显示每个日志严重类型的相对频率的饼图。

您可以为可视化添加标题，根据需要调整它们的大小和位置，然后保存您的更改。

  1. 在 Kibana 仪表板上选择 Refresh。由于 elvis.py 继续运行并生成日志数据，您的 Kibana 可视化会随每次刷新而更新。

    A screen capture of the completed Kibana dashboard

  2. 最后一步，记得停止 Filebeat 和 Python 脚本。在 Filebeat 终端和 elvis.py 终端中输入 CTRL + C。

您现在了解了如何监控 Python 应用程序的日志文件、如何将日志事件数据安全地传输到 Elastic Cloud Hosted 部署，以及如何在 Kibana 中实时可视化结果。查阅 Filebeat 文档以了解有关可用于您的数据的 Filebeat 提取和处理选项的更多信息。您还可以浏览我们的 文档以了解有关使用其他工具提取数据的更多信息。

© . This website operates independently and is not affiliated with or endorsed by Elasticsearch B.V. All brand names, logos, and trademarks are the property of their respective owners.