嘿，刚好我之前处理过类似的场景，给你梳理下怎么搞定这个需求——既要按索引名称的时间删旧索引，又要根据日志里的字段（比如你示例里的container_id）应用不同的删除规则。

一、优先用官方工具：Elasticsearch Curator #

这是Elastic官方出的索引/快照管理工具，完美适配你的需求：既能按索引名称的时间模式筛选，还能结合查询条件做文档级别的删除，比自己写脚本靠谱多了。

1. 先装Curator #

如果还没装，直接用pip装就行：

pip install elasticsearch-curator

2. 写Curator的操作配置文件 #

Curator用YAML配置文件定义要执行的操作，我们可以分两步来处理你的需求：

• 第一步：直接删除那些超过指定天数、且没有特殊字段保留需求的整个索引
• 第二步：对需要按字段保留的索引，先筛出时间符合条件的，再删掉里面不符合保留规则的文档

给你个示例配置文件（比如叫cleanup_fluentd_indices.yml）：

actions:
  # 操作1：删除超过30天的普通日志索引（排除需要特殊保留的容器日志所在索引）
  1:
    action: delete_indices
    description: "删掉所有超过30天、且不含指定容器日志的fluentd索引"
    options:
      ignore_empty_list: True  # 没有符合条件的索引时不报错
      disable_action: False    # 改成True可以先测试，不执行实际删除
    filters:
      - filtertype: pattern
        kind: prefix
        value: fluentd-  # 只处理fluentd开头的索引
      - filtertype: age
        source: name
        direction: older
        timestring: '%Y.%m.%d'  # 索引名称里的时间格式，对应你的fluentd-2018.03.28
        unit: days
        unit_count: 30  # 保留30天内的，超过的处理
      - filtertype: query
        # 排除包含指定container_id日志的索引，这些索引留到下一步处理
        query: '{"bool": {"must_not": [{"term": {"container_id": "bbd72ec5e46921ab8896a05684a7672ef113a79e842285d93"}}]}}'

  # 操作2：在超过15天的索引里，只删除非指定容器的日志文档
  2:
    action: delete_by_query
    description: "清理15天以上索引里的非指定容器日志，保留目标容器的日志"
    options:
      ignore_empty_list: True
      disable_action: False
      query:
        # 删掉所有不是目标container_id的文档
        bool:
          must_not:
            - term:
                container_id: "bbd72ec5e46921ab8896a05684a7672ef113a79e842285d93"
    filters:
      - filtertype: pattern
        kind: prefix
        value: fluentd-
      - filtertype: age
        source: name
        direction: older
        timestring: '%Y.%m.%d'
        unit: days
        unit_count: 15  # 可根据实际需求调整目标容器日志的保留天数

3. 执行Curator命令 #

把配置文件路径填对，运行就行：

curator --config /path/to/your/curator/config.yml cleanup_fluentd_indices.yml

（Curator的主配置文件config.yml主要是填ES的地址、认证信息这些，写法很简单，查官方文档就能搞定）

二、如果不想用Curator，自己写Python脚本也行 #

要是你喜欢自定义逻辑，用Python结合Elasticsearch API也能搞定，给你个示例代码：

from elasticsearch import Elasticsearch
from datetime import datetime, timedelta

# 初始化ES客户端，填你的ES地址
es = Elasticsearch(["http://your-es-host:9200"])

# 定义时间阈值，比如30天前和15天前的日期
threshold_30d = (datetime.now() - timedelta(days=30)).strftime("%Y.%m.%d")
threshold_15d = (datetime.now() - timedelta(days=15)).strftime("%Y.%m.%d")
# 要特殊保留的容器ID
target_container = "bbd72ec5e46921ab8896a05684a7672ef113a79e842285d93"

# 获取所有fluentd前缀的索引
indices = es.cat.indices(index="fluentd-*", h="index").split()

for index in indices:
    # 从索引名里提取日期部分，比如fluentd-2018.03.28提取出2018.03.28
    index_date = index.split("-")[-1]
    if index_date < threshold_30d:
        # 检查这个索引里有没有目标容器的日志
        has_target_docs = es.count(index=index, q=f"container_id:{target_container}")["count"] > 0
        if not has_target_docs:
            # 没有的话直接删整个索引
            es.indices.delete(index=index)
            print(f"已删除索引: {index}")
        else:
            # 有的话只删掉非目标容器的文档
            es.delete_by_query(
                index=index,
                body={
                    "query": {
                        "bool": {
                            "must_not": [{"term": {"container_id": target_container}}]
                        }
                    }
                }
            )
            print(f"已清理索引 {index} 中非目标容器的日志")
    elif index_date < threshold_15d:
        # 针对15-30天的索引，同样只清理非目标容器的日志
        es.delete_by_query(
            index=index,
            body={
                "query": {
                    "bool": {
                        "must_not": [{"term": {"container_id": target_container}}]
                    }
                }
            }
        )
        print(f"已清理索引 {index} 中非目标容器的日志")

三、几个重要的注意事项 #

• 先测试再执行！：Curator里可以把disable_action改成True，脚本里可以先把删除操作改成打印，确认筛选逻辑完全符合你的预期再实际执行，删错了可就找不回来了。
• 要是日志量特别大，delete_by_query会占用不少ES资源，建议在业务低峰期执行，或者分批处理。
• 可以把Curator命令或者脚本加到定时任务里（比如Linux的crontab），实现自动定期清理，省得手动跑。

内容的提问来源于stack exchange，提问作者BlackHoleGalaxy