logstash -> elasticsearch：如何在输出新数据之前删除所有数据

Question

logstash 每周获取多个事件，然后将这些事件转发给 elasticsearch，

如何配置 logstash 让它告诉 elasticsearch 删除旧事件？

编辑 2018-03-28：

输入：

{host:"host1", type:"packages", records: [{name:"pkg1", ver: "1"}, {name: "pkg2", ver: "2"},...]
{host:"host1", type:"mounts", records: [{path:"path1", dev: "dev1"}, {path:"path2", dev: "dev2"},...]
{host:"host1", type:"???", records: [{???}, {???},...]
...
{host:"host2", type:"packages, records: [{name:"pkg1", version: "1"}, {name: "pkg2", ver: "2"},...]
{host:"host2", type: "mounts", records: [{path:"path1", dev: "dev1"}, {path:"path2", dev: "dev2"},...]
{host:"host2", type:"???", records: [{???}, {???},...]

这是每个主机的各种事件。每个事件都由一组 undeterminable 架构组成。

为了能够精确搜索数组中的字段，我必须将数组拆分为多个elasticsearch文档。

（我知道有一些方法可以不拆分但能够在数组内部搜索。那是另一个故事：Nested Object。在我的情况下，内部对象不是固定模式，所以我不能提供每个内部字段定义事先）

输出：

{host: "host1", type:"packages", record: {name: "pkg1", ver: "1"}}
{host: "host1", type:"packages", record: {name: "pkg2", ver: "2"}}
{host: "host1", type:"mounts", record: {path: "path1", dev: "dev1"}}
{host: "host1", type:"???", record: {???}
{host: "host1", type:"???", record: {???}
{host: "host1", type:"mounts", record: {path: "path2", dev: "dev2"}}
{host: "host2", type:"packages", record: {name: "pkg1", ver: "1"}}
{host: "host2", type:"packages", record: {name: "pkg2", ver: "2"}}
{host: "host2", type:"mounts", record: {path: "path1", dev: "dev1"}}
{host: "host2", type:"mounts", record: {path: "path2", dev: "dev2"}}
{host: "host2", type:"???", record: {???}
{host: "host2", type:"???", record: {???}
...

logstash.conf：

input { ... }

filter {
    split {
      # split array and save them into new multiple events
      field => "records"
    }
    mutate {
      rename => { "records" => "record" }
    }
}

output {
  elasticsearch {
    hosts => ["ELASTIC_IP:PORT"]
    index => "packages-%{+YYYY.MM.dd}"
  }
}

-

问题是：Elasticsearch 会为每种类型的每个主机填充越来越多的旧事件。

所以我想在获得新数据后删除主机的旧数据。

注意一些失败的尝试：

• Overwrite existing document by specify document id in the logstash.conf

因为输出的是多个文档，而不是单个文档，有时多，有时少，所以不是简单的更新。它必须是全部删除并添加。

我知道有一些方法可以不拆分但能够在数组内搜索。那是另一个故事：Nested Object。就我而言，内部对象不是固定模式，所以我不能事先提供每个内部字段定义

Answer 1

好吧，我已经确认可以通过 ruby​​ 过滤器来删除旧索引。

input { ... }

filter {
  split {
    # split array and save them into new multiple events
    field => "records"
  }
  mutate {
    rename => { "records" => "record" }
  }

  ruby {
    init => "
       require 'net/http'
       require 'uri'
     "
    code => "
      uri = URI.parse('http://docker.for.mac.localhost:19200/inventory-' + event.get('type') + '@' + event.get('host'))
      http = Net::HTTP.new(uri.host, uri.port)
      req = Net::HTTP::Delete.new(uri.request_uri)
      req.basic_auth 'elastic', 'changeme'
      res = http.request(req)
    "
  }
}

output {
  elasticsearch {
    hosts => ["ELASTIC_IP:PORT"]
    index => "inventory-%{type}@%{host}"
  }
}

重要的是为每个主机和类型组合指定索引，以便在删除时轻松定位。

index => "inventory-%{type}@%{host}"