如何删除ELK中较旧的日志，以便为每个应用程序分配一定的磁盘配额

Question

我尝试在以下场景中使用ELK (Elasticsearch+Logstash+Kibana)堆栈：

我有大约10个应用程序通过Logstash将它们的日志发送到单个Elasticsearch集群。

这些应用程序中的一些自然地比其他应用程序生成更多的日志，并且，有时，它们中的一个可能会因为错误而变得“疯狂”，从而生成比正常情况下更多的日志条目。因此，集群中可用的磁盘空间可能会被单个应用程序的日志不公平地“占用”，从而没有给其他应用程序留下足够的空间。

我目前正在通过Elasticsearch Curator管理可用的磁盘空间。它定期运行，因为它在crontab中，并根据磁盘使用配额删除较旧的索引。当所有索引使用的磁盘空间超过一定的限制时，将逐个删除最旧的索引，直到所有索引使用的磁盘空间之和再次在指定的限制内。

这种方法的第一个问题是Elasticsearch Curator只能删除整个索引。因此，我必须将Logstash配置为每小时创建一个不同的索引，并增加它们的粒度；因此，Curator一次删除较小的日志块。此外，很难决定策展人应该多久运行一次。如果应用程序以更高的速率生成日志，那么即使是一个小时的索引也可能不够。其次，无法为每个不同的应用程序指定磁盘使用配额。

理想情况下，只要索引达到某个磁盘使用限制，Elasticsearch就应该能够自行删除较旧的日志条目。这将消除定义策展人应该多久运行一次的问题。然而，我在Elasticsearch手册中找不到任何类似的功能。

有人会推荐一种不同的方法来解决这些问题吗？

参考资料：http://www.elasticsearch.org https://github.com/elasticsearch/curator

Answer 1

尝试使用索引生命周期管理，这在ELK堆栈6.6更新版本中可用。

请查看此链接：

https://www.elastic.co/guide/en/elasticsearch/reference/6.6/getting-started-index-lifecycle-management.html

当大小超过2 2GB或1d时，这将创建新的索引，并删除1天前的数据。

PUT _ilm/policy/stream_policy
{
  "policy": {
    "phases": {
      "hot": {
        "actions": {
          "rollover": {
            "max_size": "2GB" ,   
            "max_age": "1d"
          }
        }
      },
      "delete": {
        "min_age": "1d",
        "actions": {
          "delete": {} 
        }
      }
    }
  }
}

Answer 2

这就是删除旧日志(本例中为filebeat日志)的方法。

curl -XDELETE 'localhost:9200/filebeat-2016*?pretty'

Answer 3

如果你有很多索引和分片，elasticsearch将很难保持它们的打开(你会得到内存错误)。使用10个应用程序，您将拥有大量的分片。出于这个原因，我更喜欢使用较少的索引。

我认为你可能想要的是一种方法，当你对一个给定的应用程序获得太多的记录时，删除“剩余”记录。

设想一个小脚本，它将运行并按类型和日期计算文档数量，然后删除超出限制的文档。

下面是如何按类型和日期进行计数：

curl -XPOST 'localhost:9200/_search?pretty' -d '
{
  "size": 0,
  "aggs": {
    "by_type": {
      "terms": {
        "field": "_type"
      },
      "aggs": {
        "by_date": {
          "date_histogram": {
            "field": "@timestamp",
            "interval": "day",
            "order": {
              "_key": "desc"
            }
          }
        }
      }
    }
  }
}'

结果如下所示：

  "aggregations" : {
    "by_type" : {
      "buckets" : [ {
        "key" : "type1",
        "doc_count" : 900000,
        "by_date" : {
          "buckets" : [ {
            "key_as_string" : "2015-02-13T00:00:00.000Z",
            "key" : 1423785600000,
            "doc_count" : 300000
          }, {
            "key_as_string" : "2015-02-12T00:00:00.000Z",
            "key" : 1423699200000,
            "doc_count" : 200000
          }, {
            "key_as_string" : "2015-02-11T00:00:00.000Z",
            "key" : 1423612800000,
            "doc_count" : 400000
          }
        }
      }
    }
  }

现在，您可以遍历类型和每个日期的计数，找出您想要删除的日期。在上面的示例中，如果您只需要500,000个"type1“条目，那么您应该知道从2/11/2015删除条目。

您可以使用"delete by query“来删除这些文档。

希望这能有所帮助。