Elastic：创建你的第一个Elastic watcher

在 Elasticsearch 可以提供给我们数据的存储及快速的搜索，但是如果能够实时地根据一些条件来发送警报则对我们的应用来说将会非常地有用。在今天的文章中，我们将讲述如何使用 API 的方法利用 Elasticsearch 来生成警报。在之前我有两个教程来讲述如何在 Elastic 中发送警报：

• Elastic：使用 ElastAlert 发送通知
• X-Pack：创建阈值检查警报 

我们今天的配置环境如下：

在上面，我们使用一个 MacOS 来安装我们的 Elasticsearch 及 Kibana，同时在一个虚机上安装我们的 Filebeat。在下面的配置中，我们将讲述该如何进行安装。

安装

Elasticsearch

我们可参考我之前的文章 “如何在 Linux，MacOS 及 Windows上进行安装 Elasticsearch” 来安装我们的 Elasticsearch。在安装完后，我们需要做如下的调整。打开 config/elasticsearch.yml 文件，并修改：

cluster.name: elasticsearch
network.host: 0.0.0.0
discovery.type: single-node

在这里我们配置一个节点的 Elasticsearch，同时我们配置 network.host: 0.0.0.0，这样保证我们的 Elasticsearch 绑定于每一个网路接口上。如果你还不知道自己的IP地址，你可以打入如下的命令来找到自己的 IP 地址：

ifconfig

修改完上面的配置后，我们重新启动 Elasticsearch：

我们可以在 Elasticsearch 启动的 Log 里找到相应的启动地址。我们在浏览器中打入上面显示的地址：192.168.0.100:9200，

如果我们能够看到上面的输出，那么则表明我们的 Elasticsearch 安装是成功的。

Kibana 安装

我们接下来安装 Kibana。我们可以参考我之前的文章 “如何在 Linux，MacOS 及 Windows 上安装 Elastic 栈中的 Kibana” 来进行我们的安装。为了能够使得我们的 Kibana 能访问 Elasticsearch 及被我们的 Beats 所访问到，我们必须修改 config/kibana.yml 的相应的配置：

server.host: "192.168.0.100"
elasticsearch.hosts: ["http://192.168.0.100:9200"]

请根据自己的电脑上的 IP 地址进行上面的修改。你可以通过上面介绍的 ifconfig 来得到自己电脑上的IP地址。

我们在自己的浏览器中打入如下地址：

如果你能发现上面的界面，则表明你的 Kibana 的安装是成功的。

为了能够实现对 log 的监控，我们必须使用 Start a 30-day trial 授权：

Filebeat 安装

就如同我们上面介绍的那样，我们的 Filebeat 运行于 Ubuntu OS 18.04 中。如果大家还不知道如何来进行安装的话，请参阅我之前的文章 “Vagrant 入门教程”。那个教程非常简单直接。一旦我们的 Ubuntu 已经安装起来后，那么我们该如何安装 Filebeat 呢？

我们打开我们的 Kibana 界面：

我们点击 Add log data:

我们找到 System logs：

Ubuntu OS 是 Debian 系统。我们选择 DEB 平台，让后严格地安照上面的信息来进行安装。在这里我必须指出的是：我们必须修改 /etc/filebeat/filebeat.yml 文件的配置以使得 filebeat 能够连接上我们的 Kibana 及 Elasticsearch。为此，我们打开filebeat.yml配置文件：
 

sudo vi /etc/filebeat/filebeat.yml

 我们对该文件进行如下的修改：

及

在这里，我们必须注意的是：替换上面的IP地址为你自己电脑上的 IP 地址。这个 IP 地址是你的 Elasticsearch 及 Kibana 的运行 IP 地址。你可以通过 ifconfig 命令来获得。

因为如下的练习将使用到 Syslog，所以我们必须启动 system 模块：

sudo filebeat modules enable system

一旦完成上面的配置后，我们按照 Kibana 中所指出的命令：

sudo service filebeat start

来启动 filebeat 这个服务。如果我们安装正确的话，我们可以在 Kibana 中看到数据进来：

如果我们能够看到有数据在上面的 Discover 中来，则表明我们的 Filebeat 的安装是成功的。

我们可以在 Ubuntu 的 terminal 中打入如下的命令：

sudo logger -t JUSTME this is message 2

等过一点时间，我们通过 Kibana 来查询我们的数据：

上面的命令生成一个 process.name 为 JUSTME 的 log 信息。我们可以在 Kibana 中查询到这个信息。

Watcher API

在今天的练习中，我们将使用 Watcher API 来进行监控。请大家参考 Elastic 的官方文档Watcher APIs | Elasticsearch Guide [8.2] | Elastic。简单地说，为了配置一个 Watcher，我们必须配置如下的几个：

PUT _watcher/watch/my-justme-watcher
{
  "trigger": {},
  "input": {},
  "condition": {},
  "actions": {}
}

下图展示了 watch 的执行过程：

 我们下面来分别进行描述:

trigger

这个定义长时间 watcher 运行一次。比如我们可以定义如下：

  "trigger": {
    "schedule" : {
      "interval": "1h"
    }
  },

上面表明，我们希望 watcher 每一个小时运行一次。针对我们的情况，我们定义为：

  "trigger": {
    "schedule": {
      "cron": "0 0/1 * * * ?"
    }
  },

上面表明，我们希望每一分钟运行一次。

input

这个是定义 alert 想搜索的 index，及针对多长时间的 index 来进行的搜索的：

  "input": {
    "search": {
      "request": {
        "indices": [
          "filebeat-*"
        ],
        "body": {
          "query": {
            "bool": {
              "must": {
                "match": {
                  "process.name": "JUSTME"
                }
              },
              "filter": {
                "range": {
                  "@timestamp": {
                    "from": "{{ctx.trigger.scheduled_time}}||-5m",
                    "to": "{{ctx.trigger.triggered_time}}"
                  }
                }
              }
            }
          }
        }
      }
    }
  },

在上面，我们想针对 filebeat-* 的索引来进行搜索。如果 process.name 含有 JUSTME，则表明有一个匹配。同时我们针对在过去5分钟内发生的事件来进行查询。

condition

我们定义如下：

"condition" : {
    "compare" : { "ctx.payload.hits.total" : { "gt" : 0 }}
  },

上面显示如果在查询结果中，有至少一个的结果，那么就会发送出 alert。

actions

这个是定义当上面的 condition 被满足时，需要采取的一些行动。这些行动可以在地址找到。目前 Watcher 支持的 action 为: email，webhook，index，logging，slack 及pagerduty。

在今天的练习中，我们使用最简单的 logging：

  "actions": {
    "log": {
      "logging": {
        "text": "WARNING: find JUSTME(s) in filebeat in the last 5 minutes"
      }
    }
  }

如果大家想如何实现 Slack 的通知，请参考我之前的文章 X-Pack：创建阈值检查警报 。

经过上面的配置后，我们 alert 配置如下：

PUT _watcher/watch/my-justme-watcher
{
  "trigger": {
    "schedule": {
      "cron": "0 0/1 * * * ?"
    }
  },
  "input": {
    "search": {
      "request": {
        "indices": [
          "filebeat-*"
        ],
        "body": {
          "query": {
            "bool": {
              "must": {
                "match": {
                  "process.name": "JUSTME"
                }
              },
              "filter": {
                "range": {
                  "@timestamp": {
                    "from": "{{ctx.trigger.scheduled_time}}||-5m",
                    "to": "{{ctx.trigger.triggered_time}}"
                  }
                }
              }
            }
          }
        }
      }
    }
  },
  "condition": {
    "compare": {
      "ctx.payload.hits.total": {
        "gt": 0
      }
    }
  },
  "actions": {
    "log": {
      "logging": {
        "text": "WARNING: find JUSTME(s) in filebeat in the last 5 minutes"
      }
    }
  }
}

执行上面的命令。显示结果：

{
  "_id" : "my-justme-watcher",
  "_version" : 1,
  "_seq_no" : 6,
  "_primary_term" : 1,
  "created" : true
}

测试 Watcher

在很多情况下，如果我们设置的时间比较久，我们可能等不了那么久才做一个测试。Elastic 为了方便我们的测试，提供了 _execute 接口：

PUT _watcher/watch/my-justme-watcher/_execute

通过上面的接口，我们不需要等在 schedule 中设置的那么久的时间才运行一下我们的 watcher。上面的命令可以直接帮我们运行 watcher。

针对我们的情况，我们在 Ubuntu OS 的 terminal 中运行如下的命令:

sudo logger -t JUSTME this is message 3

我们切换到 Elasticsearch 的运行 terminal 中，等一小会，我们可以看到：

上面显示，我们已经有一个事情发生了：在 filebeat 的 log 里找到含有 JUSTME 的事件。上面的信息虽然显示了有 JUSTME 的事件发生了，但是它并没有显示有多少个事件发生了。为此，我们需要修改一下我们的watcher：

PUT _watcher/watch/my-justme-watcher
{
  "trigger": {
    "schedule": {
      "cron": "0 0/1 * * * ?"
    }
  },
  "input": {
    "search": {
      "request": {
        "indices": [
          "filebeat-*"
        ],
        "body": {
          "query": {
            "bool": {
              "must": {
                "match": {
                  "process.name": "JUSTME"
                }
              },
              "filter": {
                "range": {
                  "@timestamp": {
                    "from": "{{ctx.trigger.scheduled_time}}||-5m",
                    "to": "{{ctx.trigger.triggered_time}}"
                  }
                }
              }
            }
          }
        }
      }
    }
  },
  "condition": {
    "compare": {
      "ctx.payload.hits.total": {
        "gt": 0
      }
    }
  },
  "actions": {
    "log": {
      "logging": {
        "text": "WARNING: find {{ctx.payload.hits.total}} JUSTME(s) in filebeat in the last 5 minutes"
      }
    }
  }
}

在上面，我们使用 ctx.payload.hits.total 来展示发生了几次。再次我们在 Ubuntu 的 terminal 中连续打入如下的几个 log 信息：

我们连续地生成5个含有 JUSTME 的事件，我们再次回到 Elasticsearch 运行的 terminal 中查看：

在这一次，我们正确地显示了有5个这样的事件。

如果大家想知道如何发送一个邮件的通知信息，请参阅我的另外一篇文章 “Elastic：创建你的第一个Elastic wacther - 邮件通知”。