如何从Elasticsearch和Kibana导出日志

如何从Elasticsearch和Kibana导出日志

SOC Prime Team
SOC Prime Team linkedin icon 关注

使用简短的 Python 脚本和官方客户端的 helpers.scan,或使用点时(PIT)和 search_after 的 _search API。对于最多 10,000 行的快速拉取,ES|QL 使用 format=json 直接从 Kibana 的开发工具中运行。Kibana Discover 也能生成报告,但不是今天 LogTotal 读取的格式。如果您的集成保留了它,event.original 保持原始日志行。

Elastic 是用于内部安全分析的最常见基础之一,从 Elastic Security 到自建堆栈。Prime Core 为 Elastic 提供了 23,722 条检测规则,映射到 671 个 MITRE ATT&CK 技术。以下是如何免费获取数据,清理和调查它的方法,以供 LogTotal 使用。

免费开始

开始之前

认证。 在中创建一个 API 密钥 堆栈管理 → API 密钥 在您需要的索引上有读取和 view_index_metadata 权限。将其作为 Authorization: ApiKey <key> 发送。在 Elastic Cloud 上,Elasticsearch 端点位于您部署的概览页面上。

知道数据在哪里。 Elastic Agent 集成写入到名为 logs-<dataset>-<namespace> 的数据流。例如,来自系统集成的 Linux 身份验证日志位于 logs-system.auth-default 中。

检查 event.original. 许多集成默认不保留原始行,这是一个“保留原始事件”的选项。如果它缺失,您拥有的就是解析出的字段,对于 JSON 上传来说这没问题。

数据层。 对冷或冻结层的搜索比较慢。数据较旧时将窗口保持较小。

示例遵循当前的 Elastic 文档,检查日期为 2026 年 10 月。

选项 1:使用 helpers.scan 的 Python(推荐)

import json, os
from elasticsearch import Elasticsearch, helpers

es = Elasticsearch("https://my-deployment.es.example.com:443", api_key=os.environ["ES_API_KEY"])
query = {"query": {"range": {"@timestamp": {
    "gte": "2026-10-19T00:00:00Z", "lt": "2026-10-20T00:00:00Z"}}}}

with open("elastic_auth_1019.jsonl", "w") as out:
    for hit in helpers.scan(es, index="logs-system.auth-*", query=query, size=1000):
        out.write(json.dumps(hit["_source"]) + "n")

对于原始行,写入 hit[“_source”].get(“event”, {}).get(“original”, “”) 而不是整个文档。

选项 2:用于快速拉取的 ES|QL

在 Kibana 开发工具中:

POST /_query?format=json
{
  "query": """
    FROM logs-system.auth-*
    | WHERE @timestamp >= "2026-10-19T00:00:00Z" AND @timestamp < "2026-10-20T00:00:00Z"
    | KEEP @timestamp, host.name, user.name, source.ip, message
    | LIMIT 10000
  """
}

两点。ES|QL 返回 1,000 行,除非您设置限制,并且它最多返回 10,000 行。JSON 响应也以列加值的形式,而不是记录的形式返回。要将其转换为每行一个对象,请将查询正文保存为 query.json 并使用 jq 运行它:

curl -sS -H "Authorization: ApiKey $ES_API_KEY" -H 'Content-Type: application/json' 
  "https://my-deployment.es.example.com/_query?format=json" -d @query.json
  | jq -c '.columns as $c | .values[] | [$c, .] | transpose | map({(.[0].name): .[1]}) | add' > esql_1019.jsonl

选项 3:从 Kibana UI(Discover)导出日志

对于较小的调查,Kibana Discover 允许您直接从浏览器查找、过滤和导出日志,而无需编写脚本。

  1. 打开 Discover。 登录 Kibana 并导航到 分析 → Discover,或从主导航中选择 Discover。
  2. 选择您的数据。 选择包含您所需日志的数据视图,例如 logs-system.auth-*。确保它包含相关的 Elasticsearch 索引或数据流。
  3. 设置时间范围。 使用右上角的时间选择器选择调查时间段。缩小窗口以避免导出不必要的事件。
  4. 过滤日志。 使用搜索栏和可用过滤器隔离相关事件。例如,输入 event.category: authentication 聚焦于身份验证活动。
  5. 选择字段。 添加有用的列,如 @timestamp, host.name, user.name, source.ip, 和 message。如果有,请包括 event.original 以保留原始日志内容。
  6. 导出结果。 打开 更多 → 导出 → 选项结果为 CSV (在某些版本中, 分享 → CSV 报告)。生成报告并下载 CSV 文件。如果导出未立即下载,请在 堆栈管理 → 报告.

下找到它。 替代方法:复制日志为 JSON。 对于少量事件,选择 Discover 中的相关行,打开 所选 菜单,并选择复制文档为 JSON

。将复制的内容粘贴到文本编辑器中并保存为 .json 文件。只有当前表选择中包含的字段可能被复制。 限制。

CSV 报告是为中小型导出设计的,其行数和文件大小限值取决于 Kibana 版本和配置。作为 JSON 导出 Discover 会话定义不导出日志事件本身。

对于 LogTotal,请使用受支持的日志格式。当需要 JSONL 时,使用下述 Python 或 Elasticsearch API 方法导出记录,或在上传前将导出数据转换为每行一个 JSON 对象。

  1. 步骤 3:在 LogTotal 中上传、清理和查看 打开 上传。 上传。
  2. 并放入 .jsonl 文件。 清理。
  3. 继续之前查看一下。 

复查。

出错时症状Fix
可能原因API 密钥在索引上缺乏权限添加读取和 view_index_metadata
结果窗口太大在一个 _search 中超过 10,000 个命中使用 helpers.scan 或使用 search_after 的 PIT
ES|QL 精确返回 1,000 行默认限制添加限制(最大 10,000)
没有 event.original 字段集成没保留它改为导出解析后的文档
超时冷或冻结层上的旧数据缩小窗口;分段运行

FAQ

Elastic Security 警报怎么办?

警报位于它们自己的索引中(.alerts-security.alerts-*)。导出警报与导出其背后的日志不同;对于 LogTotal,导出日志。

这在 OpenSearch 上能用吗?

想法相同,客户端不同。

加入 SOC Prime 的 Detection as Code 平台 以提高对与您的业务最相关的威胁的可见性。为了帮助您开始并带来即时价值,现在与 SOC Prime 专家预约会议。