logstash：如何包含输入文件行号

Question

我正在尝试创建一种导航我的日志文件的方法，我需要的主要功能是：

在日志文件中搜索字符串（并返回出现的行）。
从第 x 行到第 y 行的分页。

现在我正在检查 Logstash，它看起来非常适合我的第一个功能（搜索），但第二个功能却不是那么好。我的想法是我可以以某种方式索引文件行号以及每条记录的日志信息，但我似乎找不到办法。

是否有某种 Logstash 过滤器可以执行此操作？还是 Filebeat 处理器？我做不到。

我在想也许我可以创建一种方法让我的所有进程使用处理过的信息登录到数据库中，但这也是不可能的（或非常困难的），因为日志处理程序也不知道是什么当前日志行。

最后我能做的是，为了提供一种对我的日志文件进行分页的方法（通过服务），将实际打开它，导航到特定的行并在不是很方便的服务中显示它最优，因为文件可能非常大，我已经将其索引到 Elasticsearch（使用 Logstash）中。

我现在的配置很简单：

Filebeat

filebeat.prospectors:
- type: log
  paths:
    - /path/of/logs/*.log
output.logstash:
  hosts: ["localhost:5044"]

Logstash

input {
    beats {
        port => "5044"
    }
}
output {
  elasticsearch {
        hosts => [ "localhost:9200" ]
    }
}

例如，现在我收到了这样的商品：

    {
      "beat": {
        "hostname": "my.local",
        "name": "my.local",
        "version": "6.2.2"
      },
      "@timestamp": "2018-02-26T04:25:16.832Z",
      "host": "my.local",
      "tags": [
        "beats_input_codec_plain_applied",
      ],
      "prospector": {
        "type": "log"
      },
      "@version": "1",
      "message": "2018-02-25 22:37:55 [mylibrary] INFO: this is an example log line",
      "source": "/path/of/logs/example.log",
      "offset": 1124
    }

如果我能以某种方式在该项目中包含一个像 line_number: 1 这样的字段，那就太好了，因为我可以使用 Elasticsearch 过滤器实际浏览整个日志。

如果你们对存储我的日志（和导航）的不同方式有想法，也请告诉我

Answer 1

目前，我认为这里没有任何解决方案。 Logstash、Beats、Kibana 都具有随时间变化的事件概念，这基本上就是事物的排序方式。行号更像是一种文本编辑器的功能。

在某种程度上，Kibana 可以向您显示文件中的事件。它不会为您提供逐页列表，您可以在其中实际单击页码，但使用时间范围理论上您可以查看整个文件。

Beats and Logstash 有类似的请求（增强）。

Answer 2

日志文件是你生成的吗？或者你能改变日志结构吗？然后你可以添加一个计数器作为前缀并用logstash过滤掉它。

例如

12345 2018-02-25 22:37:55 [mylibrary] INFO: this is an example log line

您的过滤器必须如下所示：

filter {
   grok {
     match => {"message" => "%{INT:count} %{GREEDYDATA:message}"
     overwrite => ["message"]
   }
}

将创建新字段 "count"。然后您可以将它用于您的目的。

Answer 3

首先让我给出 Filebeat 还没有行号字段的主要原因可能是什么。当 Filebeat 恢复读取文件时（比如重启后），它会执行 fseek 从最后记录的偏移量恢复。如果它必须报告行号，则需要将此状态存储在其注册表中或 re-read 文件并计算换行符直到偏移量。

如果您想提供一项服务，允许您对 Elasticsearch 支持的日志进行分页，您可以使用 scroll API with a query for the file. You must sort @timestamp 然后 offset 结果.您的服务将使用滚动查询来获取第一页结果。

POST /filebeat-*/_search?scroll=1m
{
  "size": 10,
  "query": {
    "match": {
      "source": "/var/log/messages"
    }
  },
  "sort": [
    {
      "@timestamp": {
        "order": "asc"
      }
    },
    {
      "offset": "asc"
    }
  ]
}

然后要获取所有未来的页面，您可以使用从第一个查询返回的 scroll_id。

POST  /_search/scroll
{
    "scroll" : "1m",
    "scroll_id" : "DnF1ZXJ5VGhlbkZldGNoBwAAAAAAPXDOFk12OEYw="
}

这将为您提供给定文件名的所有日志数据，甚至可以跨轮转跟踪它。如果行号很重要，您可以通过从具有 offset == 0 的第一个事件开始计算事件来综合生成它们，但我避免这样做，因为它很容易出错，特别是如果您添加任何过滤或多行分组。

logstash：如何包含输入文件行号

logstash: how to include input file line number

logging

elasticsearch

logstash

filebeat