Beats：如何使用 Filebeat 结合机器学习导入日志

2020-12-26 19:00 558人阅读评论(0)

我在很早的文章 “Beats：通过 Filebeat 把日志传入到 Elasticsearch” 已经写给了如何通过 Filebeat 把日志文件写入到 Elasticsearch 中。在那篇文章中，我们没有特别特别指出是如何设计一个 pipeline 来处理数据的。我在文章 “Beats: Filebeat和pipeline processors” 中详述了如何使用 Filebeat 已经 pipeline。在今天的文章中，我将通过一个简单的示例来完整地展示如何运用 Filebeat 并结合机器学习来方便地导入我们的日志。

准备工作

在做今天的练习之前，我希望你已经阅读我之前的文章 “Beats：Beats 入门教程（二）”。在那篇文章中，我已经详述了如何安装 Elastic Stack，并成功地配置，并能运用 Filebeat 导入 system 日志。在今天的文章中，我就不在累述了。

我们接着准备今天的数据。我们首先创建一个叫做 my_app.log 的日志文件。为了方便，我们只创建了几个数据。在实际的应该中，这个文件可能很大，并且可能是实时更新的。我们最终的目的是通过 filebeat 把日志数据导入到 Elasticsearch 中。

my_app.log


  
   
    
     
    
    
     
      192.
      168.
      86.
      255 - 
      2020-
      04-
      15T
      21:
      43:
      32.
      020+
      0000 DEBUG 
      471 - This is log 
      1
     
    
   
    
     
    
    
     
      192.
      168.
      86.
      255 - 
      2020-
      04-
      16T
      21:
      43:
      32.
      020+
      0000 DEBUG 
      472 - This is log 
      2
     
    
   
    
     
    
    
     
      192.
      168.
      86.
      255 - 
      2020-
      04-
      17T
      21:
      43:
      32.
      020+
      0000 DEBUG 
      473 - This is log 
      3
     
    
   
    
     
    
    
     
      192.
      168.
      86.
      255 - 
      2020-
      04-
      18T
      21:
      43:
      32.
      020+
      0000 DEBUG 
      474 - This is log 
      4
     
    
   
    
     
    
    
     
      192.
      168.
      86.
      255 - 
      2020-
      04-
      19T
      21:
      43:
      32.
      020+
      0000 DEBUG 
      475 - This is log 
      5
     
    
   
    
     
    
    
     
      192.
      168.
      86.
      255 - 
      2020-
      04-
      20T
      21:
      43:
      32.
      020+
      0000 DEBUG 
      476 - This is log 
      6
     
    
   
    
     
    
    
     
      192.
      168.
      86.
      255 - 
      2020-
      04-
      21T
      21:
      43:
      32.
      020+
      0000 DEBUG 
      477 - This is log 
      7
     
    
   
    
     
    
    
     
      192.
      168.
      86.
      255 - 
      2020-
      04-
      22T
      21:
      43:
      32.
      020+
      0000 DEBUG 
      478 - This is log 
      8
     
    
   
    
     
    
    
     
      192.
      168.
      86.
      255 - 
      2020-
      04-
      23T
      21:
      43:
      32.
      020+
      0000 DEBUG 
      479 - This is log 
      9
     
    
   
    
     
    
    
     
      192.
      168.
      86.
      255 - 
      2020-
      04-
      24T
      21:
      43:
      32.
      020+
      0000 DEBUG 
      480 - This is log 
      10
     
    
   
    
     
    
    
     
      192.
      168.
      86.
      255 - 
      2020-
      04-
      25T
      21:
      43:
      32.
      020+
      0000 DEBUG 
      481 - This is log 
      11
     
    
   
    
     
    
    
     
      192.
      168.
      86.
      255 - 
      2020-
      04-
      26T
      21:
      43:
      32.
      020+
      0000 DEBUG 
      482 - This is log 
      12

由于这个日志 my_app.log 可能会非常大，并且在实际的生产环境中，它可能还是实时生成的。为此，我们来创建一个比较小的文件 sample.log。它的内容如下：

sample.log


  
   
    
     
    
    
     
      192.
      168.
      86.
      255 - 
      2020-
      04-
      15T
      21:
      43:
      32.
      020+
      0000 DEBUG 
      471 - This is log 
      1
     
    
   
    
     
    
    
     
      192.
      168.
      86.
      255 - 
      2020-
      04-
      16T
      21:
      43:
      32.
      020+
      0000 DEBUG 
      472 - This is log 
      2

这个文件比较小。它是为了让我们建立模型而使用的。在这个文件中，它至少是需要2个以上的日志，否则机器学习将不会认识。

建立 ingest pipeline

接下来的工作我们来使用 Kibana 来导入我们的日志文件 sample.log:

我们选择 sample.log 并进行导入：

从上面我们可以看出来。机器学习帮我们自动完成了一个 Grok pattern 来解析我们的日志。我们如果对这个不太满意的话，我们可以对这个 pattern 进行一些修改。关于这个 Grok pattern, 你可以参照我之前的文章 “Logstash：如何逐步构建自定义 grok 模式”，虽然它是针对 Logstash 的。我们点击上面的 Override settings:

我们把 pattern 修改为如下的格式：

%{IP:ipaddress} .*? %{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:loglevel} %{INT:linenumber} - %{GREEDYDATA:content}

点击 Apply 按钮：

点击 Import 按钮：

我们填入 my_app 作为 index name，并点击 import 按钮：

我们会发现整个的导入速度非常快，比较只有两个文档。我们点击 Create Filebeat configuration 按钮：

我们把用红色框住的那部分拷贝下来，并修改我们 filebeat 的配置文件：

请注意上面的红色框住的部分，这个就是拷贝过来的代码。它被置于 filebeat.inputs：之下。我们同时需要根据自己的 my_app.log 的路径进行修改 paths。为了能够得日志能够导入到 Elasticsearch 中，我们还必须定义 index 以及 pipeline。我们拷贝如下的两行：

把上面的两行粘贴到 filebeat.yml 中：

这样整个 filebeat 被修改的部分如上所示。

我们接下来运行 filebeat，并检查是否成功导入所有在 my_app.log 里的所有数据：

GET _cat/indices/my_app

上面的命令显示：

我们通过如下的命令来查看索引里的文档数：

GET my_app/_count

上面的命令显示：


  
   
    
     
    
    
     
      {
     
    
   
    
     
    
    
       
      "count" : 
      12,
     
    
   
    
     
    
    
       
      "_shards" : {
     
    
   
    
     
    
    
         
      "total" : 
      1,
     
    
   
    
     
    
    
         
      "successful" : 
      1,
     
    
   
    
     
    
    
         
      "skipped" : 
      0,
     
    
   
    
     
    
    
         
      "failed" : 
      0
     
    
   
    
     
    
    
     
        }
     
    
   
    
     
    
    
     
      }