Beats：运用 Logstash 来加工 Metricbeat 数据

Posted 2021-05-18 Elastic 中国社区官方博客

在我之前的教程 “Logstash：Logstash 入门教程 （二）”，我详述了如何使用 Logstash 来处理数据。在那个教程的最后的部分，我也讲述了如何使用 Filebeat 把数据送到 Logstash 并对数据进行加工。在今天的教程中，我将使用一个简单的例子来展示如何使用 Logstash 来对 Metricbeat 的数据进行加工。

 

导入 Metricbeat 数据

我们可以参考文章 “Beats：Beats 入门教程 （二）” 来安装我们的 Metricbeat。我们只启动 system 模块，并收集自己电脑中的 Metricbeat 数据。如果我们直接把 Metricbeat 的数据传入到 Elasticsearch 的话，我们可以在 Kibana 中的 Metrics 应用中看到：

我们可以看到一大堆没有用的 IP 地址。我们想，我们是不是可以考虑使用 Logstash 来对这个 host.ip 里的数据进行处理从而得到一个只有我们想要的 IP 地址字段呢？

 

通过 Logstash 来处理 Metricbeat 数据

我们首先安装好 Logstash，并创建如下的 配置文件：

logstash.conf

input {
    beats {
        port => "5044"
    }
}

filter {
    if [metricset][name] == "network" {
        ruby {
            code => "
                filteredIps = []
                cidr = IPAddr.new('192.168.0.0/24')
                existingIps = event.get('[host][ip]')
                existingIps.each { |ip|
                    if cidr.include?(ip)
                        filteredIps.push(ip)
                    end
                }
                event.set('host.ips', filteredIps)
            "
        }
        mutate {
            add_tag => ["processed_by_logstash"]
        }
    }
}

output {
    stdout {
        codec => rubydebug 
    }
    elasticsearch {
        hosts => "localhost:9200"
        index => "%{[@metadata][beat]}-%{[@metadata][version]}"
    }
}

在上面的 filter 部分，我们使用 ruby filter 来对 IP 地址进行过滤。如果 IP 地址是在 192.168.0.0/24 范围之内，那么我们就把它放到一个数组里，并最终放到 event 中的  host.ips 字段中。同时我们添加了一个叫做 processed_by_logstash 的 tag。这个用于我们以后便于搜索。接着我们可以使用如下的命令来启动 Logstash：

 ./bin/logstash -f logstash.conf

为了能够使得 Logstash 能够处理来自 Metricbeat 的数据，我们修改 Metricbeat 中的配置文件 metricbeat.yml 文件。首先，我们除去向 Elasticsearch 写的操作：

 

同时，我们打开想 Logstash 写的操作：

经过这样的操作后，我们重新启动 Metricbeat:

./metricbeat -e

这个时候，我们去 Logstash 的 terminal 中进行查看：

针对 metricset.name 是 network 的情况。我们添加了一个新的字段 host.ips。同时它也添加了一个新的 tag：processed_by_logstash。

我们可以在 Kibana 中看到我们想要的被添加的字段 host.ips：

我们也可以直接搜索 processed_by_logstash tag: