内核堆栈用户堆栈_弹性堆栈介绍

你的名字 2023-02-25 04:58 21阅读 0赞

内核堆栈 用户堆栈

当您运行对公司至关重要的软件时，您将无法拥有仅用于分析一段时间前发生的事情的日志，让客户端告诉您您的应用程序已损坏，而您甚至不知道发生了什么是真实的问题。 解决该问题的方法之一是使用监视和日志记录。

大多数应用程序都将具有日志记录，即使它只是一个文本文件，也需要使用grep或tail等工具来查看发生了什么。 但这是查看日志的一种非常被动的方式，因为您必须采取措施，并且可能只有在出现问题时才查看日志。 您需要的是让日志告诉您正在发生的事情，而今天我们将为此而迈出第一步。

通过标题，您可能已经知道我们将要讨论`Elasticstack` 。 我的这篇文章旨在为您提供开始收集和分析日志的基本方法，而无需您处理我经历的所有困难。

### 弹性堆栈 ###

Elastic Stack在被称为ELK之前，是一组工具，可帮助您`collect` ， `structure` ， `store` ， `analyse`并还可以针对某些情况创建`actions` 。

弹性堆栈由4个组件组成：

*  `Kibana`是一种可视化工具，可从`ElasticSearch`读取数据。 您可以创建仪表板或手动查询`ElasticSearch` 。
 *  `ElasticSearch`是日志的存储。 您可以从`Beats`或`Logstash`发送日志，并将其存储在索引中。 您可以具有多个索引来存储来自多个源的数据。
 *  `Logstash`是负责处理日志的应用程序，您可以将日志解析为更有用的数据，并将其发送到`ElasticSearch` 。
 *  `Beats`或`Filebeat`是一种轻量级的工具，可读取日志并将其发送到`ElasticSearch`或`Logstash` 。 该工具的唯一目的是读取日志文件，它无法执行任何复杂的操作。 如果需要执行复杂的操作，则可以将该日志发送到`Logstash`以便将其解析为所需的信息。

## Logstash ##

我们从`Logstash`开始，因为它是您收集和解析日志的方式（是的，我提到过`Beats`但您无需启动它）。

`Logstash`是日志处理器和检索器。 `Logstash`的主要功能是为非结构化日志文件提供结构，处理日志分为三个步骤：

*  输入–接收或获取日志。
 *  过滤器–处理或过滤。
 *  输出–将日志发送到永久存储。

### 输入项 ###

`Logstash`可以从日志文件中检索数据，而且还可以从多个来源获取数据，例如：

*  文字档
 *  资料库
 *  Shell命令
 *  Queue列
 *  Http请求

如果要查看`Logstash`支持的所有输入插件，请检查[docs][] 。

现在，我一直在使用的输入以及我将解释如何使用的输入是：

*  文字档
 *  资料库
 *  Shell命令

#### **文字文件输入** ####

存储日志的最常见方法之一是位于计算机中某个位置的文本文件，通常是`/var/log` 。 `Logstash`有一个插件可以读取文件，并像`tail -f`一样不断监视新行。

要使用该插件非常简单，只需添加文件的路径，Logstash就会处理其余的事情，不用担心旋转文件，该插件也知道如何处理。

`file`插件的用法如下：

input {
      file {
        path => /var/log/app/realworld.log
        id => realworld
      }
    }

按照我们上面配置Logstash的方式，每一行都是一个日志条目。 但是有时候我们的日志不是那么简单，我们有诸如堆栈跟踪之类的东西，或者我们将JSON写入日志中。 在这种情况下，我们需要它们在一起才有意义，这就是Logstash提供`codecs`的原因，这是扩展输入插件的一种方式。 这些编解码器之一是`Multiline` 。

例如，下面有使用Spring的应用程序的日志，当我们有堆栈跟踪时，我们希望将其分组为一行。

2019-08-18 18:31:53.845 INFO 6724 --- [ main] o.s.t.web.servlet.TestDispatcherServlet : FrameworkServlet '': initialization completed in 17 ms
    com.andre2w.transaction.TransactionTooOldException
      at com.andre2w.transaction.TransactionService.validateTimestamp(TransactionService.java:46)
      at com.andre2w.transaction.TransactionService.add(TransactionService.java:29)
      at com.andre2w.controllers.TransactionController.create(TransactionController.java:42)
      at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
      at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
      at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
      at java.lang.reflect.Method.invoke(Method.java:498) 
    2019-08-18 18:31:53.891 INFO 6724 --- [ main] o.s.b.t.m.w.SpringBootMockServletContext : Initializing Spring FrameworkServlet ''
    2019-08-18 18:31:53.892 INFO 6724 --- [ main] o.s.t.web.servlet.TestDispatcherServlet : FrameworkServlet '': initialization started

因此，如果我们想从日志中捕获堆栈跟踪，我们可以做一个简单的正则表达式来对所有以空间开头的内容进行分组。

input {
      file {
        path => /var/log/app/realworld.log
        id => realworld
        codec => multiline {
          pattern => "^\s"
          what => "previous"
        }
      }
    }

但是在这种情况下，我们可以走得更远。 由于这些日志来自使用Spring的应用程序，并且所有日志都遵循某种格式，因此我们可以将与该格式不匹配的所有内容归为一组，其中包括堆栈跟踪，JSON和对象。

input {
      file {
        path => /var/log/app/realworld.log
        id => realworld
    
        codec => multiline {
          pattern => "\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d{3} \w+ \d+ --- \[\s+\w+]"
          negate => true
          what => "next"
        }
      }
    }

`negate` `what`意思？

“否定”是用于定义是否要将正则表达式匹配或不匹配的行分组的选项。 在第一个示例中，省略了字段`negate` ，因此其默认值为`False` 。 `negate => false`意味着线将被组合在一起时，正则表达式**匹配** 。 `negate => true`表示当**不**匹配正则表达式时，行将被分组在一起。

在`what`是Logstash将如何组的请求。 `previous`意味着匹配的行将与上一行分组，而`next`行则很明显。

#### **数据库** ####

由于Logstash使用Java运行，因此您可以访问`JDBC`支持的任何数据库。 对于Postgres，您可以看到查询表`pg_locks`的锁数。

像这样的配置是：

input {
      jdbc {
        jdbc_connection_string => "jdbc:postgres://localhost:5432/main"
        jdbc_user => "postgres"
        jdbc_password => "postgres"
        jdbc_driver_class => "org.postgresql.Driver"
        statement_filepath => "/etc/logstash/queries/query-locks.sql"
        schedule => "* * * * *"
        tags => ["jdbc","staging-1","query-locks", "main"]
      }
    }

以`jdbc_`开头的字段`jdbc_`用于连接数据库的信息。 然后，我们有`statement_filepath` ，它指向要运行`Logstash`的查询文件，您也可以将`statement`选项用于更多临时查询。 最后，我们有`schedule`选项，这是您要运行查询的频率，该值基于Linux crontab，在此示例中，它将每分钟运行一次。

`tags`帮助您识别正在运行的查询。 它只是一个字符串数组，您可以添加任何内容。

`Logstash`将解析所有字段并将其发送到输出中定义的数据存储。

使用JDBC输入时需要注意的一点是，您需要Logstash类路径中的jar。 在Logstash安装文件夹中，转到`libs/jars`并添加数据库驱动程序的jar。

#### **执行力** ####

有时，您可能希望从`Logstash`不太支持的来源获取数据。 在这种情况下，您可以使用`exec`输入插件来执行命令行应用程序，并将结果作为日志行。

input { 
      exec {
        command => "/usr/bin/retrieve-data.sh"
        schedule => "* * * * *"
        tags => ["retrieve-data","staging"]
      }
    }

此输入非常易于使用，您可以选择：

*  `command` ：您要使用的shell命令。
 *  `schedule` ：您要运行命令的频率与`jdbc-input`计划相同。
 *  `tags` ：用于以后识别结果的信息。

使用此插件，您不必担心结果会多行显示， `Logstash`会将所有内容分组在一行中。

### 过滤 ###

可以说我们正在从所有输入中接收数据，但是我们需要以有用的方式转换数据。 我们不仅希望拥有日志行，还希望统计信息能够实时查看正在发生的事情，并将收到的日志转换为结构正确的数据。

#### **Grok Parser** ####

在大多数情况下，使用Grok Parser插件。 大多数情况下，日志中的数据不是结构化格式，但是我们仍然需要解析这些非结构化数据并为其赋予一些含义。 如果要求和，就不能将数字作为字符串。 Grok不会通过使用正则表达式来构造数据， `Logstash`具有一组涵盖大多数情况的内置正则表达式。

这是过滤器的示例。

我们正在从互联网接收HTTP请求，我们想跟踪哪些端点的请求最多，我们的日志结构如下：

192.168.0.1 GET /索引

因此，我们添加了一个Grok过滤器来结构化：

filter {
      grok {
        match => { 
          "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request}"
        }
      }
    }

该过滤器中发生了什么？

`match`是指检查我们收到的日志的消息字段并将其转换为结构化数据。

因此，我们收到了来自网络服务器的请求：

我希望`Logstash`获取日志并将其转换为结构化数据（这是JSON，就像现在的一切一样）。 我们将匹配器设为：

"%{IP:client} %{WORD:method} %{URIPATHPARAM:request}"

grok的语法为`%{PATTERN:field}`

模式是您要如何匹配该字段以及将在数据中使用哪种正则表达式，而该字段将是要存储该数据的字段（位明显）。 您可以[在此处][Link 1]查看所有可用模式。

如果`Logstash`无法正确解析该模式，它将添加一个名为`_grokparsefailure`的标签。

由于它只是一堆用于解析数据的正则表达式，因此您可以创建自己的模式。 为此，您需要在`Logstash`文件夹中创建一个名为`patterns`的文件夹（您将需要检查安装位置）。 文件名可以命名为任意名称， `Logstash`将读取其中的所有内容。

grok {
      match => {
        "path" => "/%{FILE_VERSION:version}_%{BU_ID:id}\.csv$"
      }
      patterns_dir => ["/elasticsearch/logstash/example/patterns"]
    }

还有一种将文件保存在其他文件夹中，然后在配置文件中声明的方法。

该文件的内容必须遵循`PATTERN_NAME regex`的模式，因此您将以类似以下内容结束：

FILE_VERSION \w+\d{6}(?=_)
    BU_ID \d{3}(?=\.)

如果您不打算重复使用正则表达式并希望内联它，则也可以这样做：

%{DATESTAMP:timestamp} (?<mti_before>\w{46})(?<mti>\w{4})%{GREEDYDATA}

就像正则表达式组一样，您可以使用括号来指定要分组的内容，然后从问号开始通知您将要使用正则表达式。 然后，添加要将数据解析到的字段的名称，最后，您可以添加正则表达式。

Grok也将允许转换为`int`和`float` 。 您只需要在`match`添加一个额外的参数即可。 例如： `%{IP:client:int}`

使用grok插件时，请注意不要通过添加模式来复制正在发送的消息：

grok {
      match { message => "%{GREEDYDATA:message}" }
    }

这会将新消息添加到消息字段，而不是替换。

### JSON格式 ###

您可能处在更好的状况，并且已将日志结构化为JSON之类的格式。 为此， `Logstash`将为您提供免费的解析。

json {
      source => "message"
      target => "payload"
    }

这就是`Logstash`解析所有JSON消息并为字段设置适当的数据类型所需要做的一切。

#### **核心价值** ####

另一个可能有用的过滤器是Key-Value或`kv` 。 它用于基于两个键拆分数据。 因此，如果我们的日志行如下所示：

timestamp=10/09/2019 10:10:50, level=INFO, message=Something wrong might not be right

我们可以这样使用kv过滤器：

filter {
      kv {
        source => "message"
        field_split => ","
        value_split => "="
        target => "summary"
      }
    }

因此，我们可以使用`kv`过滤器解析消息中的数据。 该过滤器的唯一问题是您不能在过滤步骤中设置数据类型。

#### **变异** ####

您可能想要更改正在接收的日志，我不是在讨论对日志行的完整分析，而是在进行一些小的更改。 有一个`mutate`过滤器可以执行此操作，并且可以使用多个命令来更改日志。

您可以使用mutate过滤器进行操作的一些示例：

*  `convert` ：您可能已经解析了一个字段，但是您需要该字段不仅仅是一个字符串。 `convert`命令允许您将其转换为`integer` ， `float` ， `string`或`boolean` 。

filter {
      mutate {
        convert => {
          "quantity" => "integer"
          "is_paid" => "boolean"
        }
      }
    }

*  `remove_field` ：您可能想从日志中删除一些敏感数据，因此可以使用此命令将其删除。

filter {
      mutate {
        remove_field => ["sensitive_data"]
      }
    }

*  `gsub` ：这是一个使用正则表达式替换值的选项，您可能想混淆一些不相关的数据，可以为此使用该选项。

filter {
      mutate {
        gsub => [
          # field                   regex                      result
          "transaction_reference", "\d{4}-\d{4}-\d{4}-\d{4}", "XXXX-XXXX-XXXX-XXXX"
        ]
      }
    }

这将用掩码版本替换所有事务引用。

### 输出量 ###

这是您可以将刚刚解析的日志定向到输出或数据存储的部分。 在我们的例子中，我们将使用NoSQL文档存储的`Elasticsearch` ，但您也可以将其发送到其他地方，例如`CSV` ， `HTTP`甚至是`email` 。

您可以查看`Logstash`的文档以查看所有输出插件。

output {
      elasticsearch {
        hosts => ["192.168.0.15"]
        user => "elastic_user"
        password => "elastic_password"
        index => "application-log"
      }
    }

在此处的示例中，我们将日志发送到托管在另一台计算机上的`Elasticsearch` 。

## 弹性搜索 ##

`Elasticsearch`是一个搜索分析引擎，负责收集和聚合存储的数据。 它还提供对结构化或非结构化文本或数字数据的各种数据的实时搜索。

`Elasticsearch`所有数据都以JSON格式存储，然后进行索引，使您可以立即对其进行搜索。 每个存储的文档都是一组包含数据的键值对，并存储在优化的数据结构中，以帮助以后进行搜索。

### 基本构造 ###

#### **节点数** ####

节点是存储和索引数据的最基本的构造之一。 有多种类型的节点。

*  主节点：这控制集群
 *  数据节点：它保存数据并执行CRUD操作，聚合和搜索。
 *  接收节点：这将在索引之前转换并丰富数据。

#### **指数** ####

索引是具有相似特征的文档的集合，它们类似于关系数据库中的表。

索引比关系数据库更灵活，因为它们是轻量级的，因此可以轻松创建多个索引。 例如，在日志记录中，您可以为每天创建一个索引，并将其类型设置为所拥有的日志类型。

每天都会创建一个新索引，而对于关系数据库则不会这样做。

### 使用Elasticsearch ###

使用`Elasticsearch`时，我们必须注意两点。 它们是`templates`和`policies` 。

#### **范本** ####

可以将模板视为索引的架构， `Elasticsearch`可以设置默认架构，但是如果您希望对所拥有的数据进行汇总和计算，则需要对其进行更多控制。

`Elasticsearch`支持哪些类型？ 支持的主要数据类型有：

*  串
 *  数值（长，整数，短，双精度，浮点）
 *  日期

#### **建筑模板** ####

如何设置模板？ 好吧， `Elasticsearch`有一个类似REST的API，您可以轻松地与之交互。

我们正在为金融科技公司的应用程序添加日志，我们希望监视我们进行的资金转移。 我们用于传输的有效负载是：

{
      "paymentDate": "2019-09-14 11:25:32.321",
      "amount": 100.00,
      "message": "Payment message",
      "origin": {
        "sortCode": "11-11-11",
        "account": "838383"
      },
      "destination": {
        "sortCode": "11-11-11",
        "account": "1313123"
      }
    }

我们开始使用`payment_date`字段构建模板，我们可以将类型设置为日期并提供字段格式：

"payment_date": {
      "type": "date",
      "format": "yyyy-MM-dd HH:mm:ss.SSS"
    }

然后我们有`amount` ，我们与该类型添加该字段`scaled_float`和`scaling_factor`为100所以`Elasticsearch`可以处理在小数部分两位数使生活变得更轻松了后，我们的搜索：

"amount": {
      "type": "scaled_float",
      "scaling_factor": "100"
    }

然后，我们得到的字段`message`只是一个字符串，因此我们将使用`text`类型创建一个为全文搜索建立索引的字段：

"message": {
      "type": "text"
    }

`origin`字段和`destination`字段实际上是相同的，并且它们始终具有相同的格式，因此我们可以使用`keyword`类型。 这种类型适用于少量的半结构化数据，例如邮政编码，地址，电子邮件，分类代码和帐号：

"origin": {
      "properties": {
        "body": {
          "properties": {
            "sort_code": {
              "type": "keyword"
            },
            "account": {
              "type": "keyword"
            }
          }
        }
      }
    }

现在我们有了索引的完整映射，可以将其插入到`Elasticsearch` 。 我们只是向它发出一个`PUT`请求。

curl -X PUT "http://elasticsearch.com/_template/transfers_template" -d @transfers_template.json

{
      "index_patterns": [
        "transfers-*"
      ],
      "mappings": {
        "_meta": {
          "beat": "transfers",
          "version": "7.0.1"
        },
        "date_detection": false,
        "dynamic_templates": [
          {
            "labels": {
              "mapping": {
                "type": "keyword"
              },
              "match_mapping_type": "string",
              "path_match": "labels.*"
            }
          },
          {
            "container.labels": {
              "mapping": {
                "type": "keyword"
              },
              "match_mapping_type": "string",
              "path_match": "container.labels.*"
            }
          },
          {
            "fields": {
              "mapping": {
                "type": "keyword"
              },
              "match_mapping_type": "string",
              "path_match": "fields.*"
            }
          },
          {
            "docker.container.labels": {
              "mapping": {
                "type": "keyword"
              },
              "match_mapping_type": "string",
              "path_match": "docker.container.labels.*"
            }
          },
          {
            "kibana.log.meta": {
              "mapping": {
                "type": "keyword"
              },
              "match_mapping_type": "string",
              "path_match": "kibana.log.meta.*"
            }
          },
          {
            "strings_as_keyword": {
              "mapping": {
                "ignore_above": 1024,
                "type": "keyword"
              },
              "match_mapping_type": "string"
            }
          }
        ],
        "properties": {
          "@timestamp": {
            "type": "date"
          },
          "payment_date": {
            "type": "date",
            "format": "yyyy-MM-ddTHH:mm:ss.SSSSSS"
          },
          "amount": {
            "type": "scaled_float",
            "scaling_factor": "100"
          },
          "message": {
            "type": "text"
          },
          "origin": {
            "properties": {
              "body": {
                "properties": {
                  "sort_code": {
                    "type": "keyword"
                  },
                  "account": {
                    "type": "keyword"
                  }
                }
              }
            }
          },
          "destination": {
            "properties": {
              "body": {
                "properties": {
                  "sort_code": {
                    "type": "keyword"
                  },
                  "account": {
                    "type": "keyword"
                  }
                }
              }
            }
          }
        }
      }
    }

#### **政策规定** ####

此功能仅在`Elasticsearch`的高级版本中可用。

索引将一直被数据轰炸，就像日志文件一样，我们需要一个过渡策略来使磁盘不满。 在`Elasticsearch`的高级版本中，我们拥有索引策略工具来帮助我们进行管理。

首先要知道的是索引可以处于哪些状态。

*  `hot` ：是我们正在写入的索引。
 *  `warm` ：是我们经常查询但未写入的索引。
 *  `cold` ：是我们不再写的索引，并且我们也不经常查询数据。
 *  `delete` ：是不再需要的索引，可以删除。

索引从`hot`状态开始，当我们不想再继续写索引时，可以对`Elasticsearch`说。 我们通过`max_age`和`max_size`选项告诉它开始使用另一个索引。 在下面的示例中，我们每天或在达到`5GB` （任意选择该数字）时创建一个新索引。

{
      "policy": {
        "phases": {
          "hot": {
            "actions": {
              "rollover": {
                "max_age": "1d",
                "max_size": "5GB"
              }
            }
          }
        }
      }
    }

我们不想保持所有索引都`hot` ，因此我们可以开始改变旧索引的状态并使它们`warm` 。 定义将进入`warm`状态的策略的策略始于`min_age`参数，这听起来很明显。 在本例中，我们将`min_age`设置为`7d` ，因此所有7天或更早的`hot`索引都将转换为`warm`索引。

对于`warm`索引，我们有一些以前在`hot`索引中不可用的选项， `actions`部分允许我们在更改索引状态时进行一些更改。

我们可以看到的第一个选项是`forcemerge`此选项（设置为`1` ，它告诉`Elasticsearch`合并从`hot`到`warm`所有索引。 这很有用，因为在`Elastisearch`当您删除文档时，该文档并没有真正被删除，只有标记为已删除。 在合并过程中，标记为已删除的文档将被正确删除，就像您将文件发送到“ `Trash bin`然后在以后从系统中删除它们一样。

然后我们进行`shrink` ，该`shrink`用于减少索引的分片数量。 由于我们不再在该索引中编写数据，因此我们不需要之前分配的所有分片。

最后我们`allocate` 。 在这里，我们可以设置`number_of_replicas` ，以防万一我们需要数据高度可用，它比只有一个分片更安全。

"warm": {
      "min_age": "7d",
      "actions": {
        "forcemerge": {
          "max_num_segments": 1
        },
        "shrink": {
          "number_of_shards": 1
        },
        "allocate": {
          "number_of_replicas": 2,
          "require": {
            "data": "warm"
          }
        }
      }
    }

对于比我们为`warm`状态设置的数据还要旧的数据，以及出于安全原因而仅存储的数据，我们可以将其索引设置为`cold` 。 我们将每个大于6个月的索引设置为`cold`然后将其冻结。 在此步骤中进行冻结的好处是`Elastisearch`不会为冻结的索引使用额外的内存。

"cold": {
      "min_age": "180d",
      "actions": {
        "freeze": {}
      }
    }

最后，删除索引非常简单。

"delete": {
      "min_age": "365d",
      "actions": {
        "delete": {}
      }
    }

我们拥有的最终政策将如下所示：

{
      "policy": {
        "phases": {
          "hot": {
            "actions": {
              "rollover": {
                "max_age": "1d",
                "max_size": "5GB"
              }
            }
          },
          "warm": {
            "min_age": "7d",
            "actions": {
              "forcemerge": {
                "max_num_segments": 1
              },
              "shrink": {
                "number_of_shards": 1
              },
              "allocate": {
                "number_of_replicas": 2,
                "require": {
                  "data": "warm"
                }
              }
            }
          },
          "cold": {
            "min_age": "100d",
            "actions": {
              "freeze": {}
            }
          },
          "delete": {
            "min_age": "365d",
            "actions": {
              "delete": {}
            }
          }
        }
      }
    }

## 节拍 ##

### FileBeat ###

`Filebeat`是一个轻量级的工具，可以读取日志并将其发送到`ElasticSearch`或`Logstash` 。 该工具的唯一目的是读取日志文件，它无法执行任何复杂的操作。 如果要执行复杂的操作，则可以将日志发送到`Logstash`并解析为所需的信息。

如果您有多台服务器，并且由于`Logstash`是一个繁重的应用程序而不想在所有计算机上安装`Logstash` ，则可以使用`Filebeat` （它是用Go语言编写的，本机已编译且非常轻量级的）。

配置`Filebeat`并不`Filebeat`因为它做的并不多。 您具有包含配置的`filebeat.yml` ：

filebeat.config.inputs:
     filebeat:
     inputs:
      - type: log
     paths:
      - "/var/log/applog/*.log"
    
     output:
     logstash:
     hosts: ["logstash:5043"]

我们可以看到`inputs`部分，该部分从文件夹读取日志`Logstash`中托管在其他位置的`output` 。 在此配置文件中，我们正在从文件夹`applog`中读取日志。 您还可以在路径中使用全局模式来获取多个文件或捕获差异，例如大写或小写字母。

#### **多行字符串** ####

使用`Filebeat`您可能需要读取多行字符串。 就像`Logstash`一样，您可以在配置中使用`multiline`选项来读取文件， `Logstash`所有相同字段都应用于`Filebeat` 。

filebeat.config.inputs:
      - type: log
        paths:
          - "/var/log/applog/application.log"
        multiline:
          pattern: "\\[\\d{2}\\/\\d{2}\\/\\d{4} \\d{2}:\\d{2}:\\d{2}]\\[\\w{2}\\]"
          negate: True
          match: after

[Filebeat如何工作][Filebeat] [Filebeat参考\[7.1\] |][Filebeat] [有弹性][Filebeat]

#### **领域** ####

您也可以在正在读取的日志中添加其他信息，以便标识从中获取数据的文件，以备日后使用。

filebeat.config.inputs:
      - type: log
        paths:
          - "/var/log/applog/application.log"
        fields:
          file_name: "application_log"
        multiline:
          pattern: "\\[\\d{2}\\/\\d{2}\\/\\d{4} \\d{2}:\\d{2}:\\d{2}]\\[\\w{2}\\]"
          negate: True
          match: after

#### **自动加载** ####

您还可以从配置文件中拆分输入列表，这样就可以用输入更新文件，而无需重新启动`Filebeat` 。 因此，我们没有将`inputs`直接添加到主配置文件中，而是将其提供给具有输入配置的文件。

filebeat.config.inputs:
      enabled: true
      path: inputs.yml
      reload.enabled: true
      reload.period: 10s

`inputs.yml`将要加载的`Filebeat` ：

- type: log
      paths:
        - "/var/log/haproxy.log"
      fields:
        file_name: "Haproxy"

## 基巴纳 ##

至此，我们正在收集，构建和存储所有这些日志。 现在我们需要从中获得一些价值。 仅仅存储它们不是最佳选择，我们需要可视化它们以获取一些反馈。

为了可视化存储到`Elasticsearch`所有数据，您可以使用`Kibana` 。 它是一个应用程序，允许您从`Elasticsearch`查询数据并基于此创建可视化效果。

在`Kibana`主页中，我们选择了使用一种模式来连接到`Elastisearch`的索引，该模式指定了诸如`logs-*`的索引名称，因此我们可以搜索所有以`logs-`开头的索引，因为我们可能是按天对日志进行了分组，而不是对所有分组在一个索引中。

### 发现 ###

“ `Discovery`区域使您可以可视化和搜索`Elastichsearch`存储的数据。

![format_png][]

#### **1 –过滤器** ####

您有一个可以在其中使用[KQL][]编写查询的栏， [KQL][]是一种非常易于使用的自定义查询语言。 `Kibana`也将帮助您自动完成查询。

因此，如果我们要搜索，您可以输入：

tags: "retrieve-data" and stats1 >= 10000

或者，您可以使用“ `filters`部分，在该部分中，您可以使用更加面向UI的方式进行搜索。

您还可以保存搜索以供以后使用，这样就不必每次都重写。 保存的搜索可以在Kibana UI的其他部分中使用。

#### **2 –日期过滤器** ####

如果要按特定时间段过滤结果，则过滤器允许您使用两个选项：

*  绝对：您可以设置所需的确切日期和时间。
 *  相对：您可以设置所需的日期和时间戳，例如`10 minutes ago` 。 使用此选项，您还可以设置从`Elasticsearch`获取最新数据。

#### **3 –记录** ####

您可以以键值方式展开和查看记录，它还显示字段的类型，可以是三种：

*  `t` –文本字段
 *  `#` –数字字段
 *  时钟符号–日期时间

您还可以通过在左侧的“ `Available Fields`菜单中选择要过滤的字段来进行过滤

### 可视化 ###

我们需要一种可视化所有存储良好的数据的方法，而Visualize允许我们创建多种图形。

#### **Y轴** ####

在此示例中，我们将创建一个`bar graph` 。 在左侧菜单中，您可以设置创建图形的选项。

![format_png 1][]

1.  `Aggregation` ：是您要执行的一种操作，可以是`count` ， `sum`和`average` 。 还有更复杂的操作，例如`Standard Deviation`和使用其他值的操作。
2.  `Field` ：是您要用于进行计算的字段以及要显示的值。 对于简单的聚合，您可以从下拉列表中直接选择字段，对于更复杂的查询，可以使用`QueryDSL`和`JSON`查找值。
3.  `Custom Label` ：您可能不想无意义地显示数据，因此您可以在此处为其添加漂亮的标签。

#### **水桶** ####

![format_png 2][]

1.  `Aggregation` ：您将如何对数据进行分组。 您可能需要一段时间使用“ `Date Histogram`或使用特定字段来对具有端点的端点进行分组。
2.  `Field` ：是您将用于拆分数据的字段。 如果您需要更复杂的聚合，则可以使用`QueryDSL`作为输入。
3.  `Interval` ：如果要按日期分组，可以指定要分割数据的时间段。 对于其他聚合（例如范围），您将获得不同的间隔字段。

### 仪表板 ###

现在，我们可以将构建在`Dashboards`的可视化文件放在一起，以便将所有要查看的数据聚合到一个位置。 您可以添加创建的可视化效果，并调整其大小以适合您所需的屏幕。 您也可以更改对它们的搜索，甚至可以全屏显示它们，从而非常好地显示它们。

![format_png 3][]

### 开发工具 ###

Kibana还为您提供了“ `Dev Tools`部分，其中包含一些有助于您工作的工具。

#### **安慰** ####

我们看到， `Elasticsearch`具有类似REST的API，控制台也提供了与API交互的简便方法。 提供自动完成功能并立即连接到`Elasticsearch`因此您无需手动构建整个请求。 您可以仅通过拥有方法和路径来发出请求，还可以编写带有验证和完成的`JSON` 。

![format_png 4][]

GET _template/transfers_template

PUT /test-template-000001
    {
      "aliases": {
        "test-template-alias": {
          "is_write_index": true
        }
      }
    }

#### **Grok调试器** ####

当我们谈论`Logstash`我们看到了`Grok`插件及其强大功能，但是`Grok`使用正则表达式工作，并且每个人都知道正则表达式非常疯狂并且很容易破解，因此调试器可以帮助我们构建模式来解析我们的代码。日志行。

我们有从服务器获取统计信息的消息，共有四列，但是我们必须拆分为适当的字段，因此我们需要使用`Grok`解析器。

16765 | 10475 | 26017 | 27583

%{NUMBER:stats1} \| %{NUMBER:stats2} \| %{NUMBER:stats3} \| %{NUMBER:stats4}

![format_png 5][]

## 今天就这些 ##

在这篇文章中，我们了解了Elastic Stack的组件，如何开始使用它们以及组织日志的重要性。 另外，您始终可以在[这里][Link 2]使用Elastic Stack文档[，][Link 2]以了解支持的其他功能（我的帖子中未提及）。

> 翻译自: [https://www.javacodegeeks.com/2019/11/elastic-stack-introduction.html][https_www.javacodegeeks.com_2019_11_elastic-stack-introduction.html]

内核堆栈 用户堆栈

[docs]: https://www.elastic.co/guide/en/logstash/current/input-plugins.html
[Link 1]: https://github.com/elastic/logstash/blob/v1.4.2/patterns/grok-patterns
[Filebeat]: https://www.elastic.co/guide/en/beats/filebeat/current/how-filebeat-works.html
[format_png]: /images/20230209/98bbf1de60774de9bdcfff43315d0e49.png
[KQL]: https://www.elastic.co/guide/en/kibana/7.1/kuery-query.html
[format_png 1]: /images/20230209/60037fb4b22f448fba79e6ba87c5888a.png
[format_png 2]: /images/20230209/e11bc2a713d948fbba0b7dfc324ff491.png
[format_png 3]: /images/20230209/3fdf2fd6baac496cbf7ce19a4fe88c1b.png
[format_png 4]: /images/20230209/7a36c90be5804d44a94dd457459c9614.png
[format_png 5]: /images/20230209/a0976fecfc814adc809812d097074ead.png
[Link 2]: https://www.elastic.co/guide/index.html
[https_www.javacodegeeks.com_2019_11_elastic-stack-introduction.html]: https://www.javacodegeeks.com/2019/11/elastic-stack-introduction.html