Go语言实现Elasticsearch索引定期清理工具

晚来风

发布时间：2026-10-13 | 6人浏览过

原创

在微服务架构中，Elasticsearch（ES）常用于集中存储日志与链路追踪数据。以SkyWalking为例，每日会产生大量以日期命名的索引，若不及时清理将迅速耗尽磁盘空间。本文将展示如何使用Go语言编写一个轻量级工具，通过配置文件灵活指定保留策略，自动扫描并删除超过保留天数的过期索引。

清理思路

常见的链路追踪和日志收集方案会占用大量的ES存储空间。例如SkyWalking每天可能产生数百个新的索引。对历史数据的留存通常遵循严格的时效性要求。

基于上述背景，我们的核心处理流程分为以下四步：

全量获取：连接目标ES集群，拉取当前所有的索引名称列表。
计算阈值：根据系统当前时间与配置文件中的“保留天数”，计算出需要截断的基准日期。
匹配过滤：由于部分ES集群开启了索引保护机制，不支持使用通配符（如*index）直接进行破坏性操作，必须使用精确的全名进行删除。因此，程序遍历所有索引，通过字符串包含关系匹配出需要剔除的索引名。
安全执行：调用删除API移除匹配的索引，确保配置化管理，支持多节点/多集群扩展。

第1步：连接ES并初始化客户端

首先建立与Elasticsearch服务的通信连接。我们使用广泛采用的github.com/olivere/elastic/v7第三方库来简化HTTP请求的封装。该步骤不仅要建立连接，还需设置合理的日志输出以便排查网络或权限问题。

func GetEsClient(data Data) *elastic.Client {
	Init()
	file := "./eslog.log"
	// 打开日志文件用于记录连接状态
	logFile, _ := os.OpenFile(file, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0766)
	
	client, err := elastic.NewClient(
		elastic.SetURL(data.Host),
		elastic.SetSniff(false), // 关闭嗅探功能，强制使用配置的URL
		elastic.SetInfoLog(log.New(logFile, "ES-INFO: ", 0)),
		elastic.SetTraceLog(log.New(logFile, "ES-TRACE: ", 0)),
		elastic.SetErrorLog(log.New(logFile, "ES-ERROR: ", 0)),
	)
	if err != nil {
		return nil
	}
	return client
}

这里定义了Data结构体作为参数传递ES连接地址，后续的API调用将依赖此Client实例。

第2步：获取全部索引数据

我们需要知道集群里有哪些索引。利用client.GetMapping().Index("*") API可以一次性获取所有索引元数据。注意：在实际的生产环境中，对于拥有数十万级别索引的极端场景，建议使用专门的查询接口代替GetMappings以提升效率，但在常规监控场景下此方法足够且稳定。

func getIndex(data Data) map[string]interface{} {
	client := GetEsClient(data)
	mapping := client.GetMapping()
	service := mapping.Index("*")
	result, err := service.Do(context.Background())
	if err != nil {
		fmt.Printf("fetch indices failed, err: %v\n", err)
		return nil
	}
	return result
}

返回的数据是一个嵌套的Map结构，其中Key即为索引的名称（Index Name）。接下来我们将基于这些Key进行处理。

第3步：计算待删除日期与字符串匹配

结合Go内置的time包，我们根据配置文件中设定的保留天数反推一个时间点，并将其格式化为用户自定义的日期字符串（例如：2006.01.02）。

currentTime := time.Now()
// data.Day为负数表示过去，例如 -7 表示7天前
oldTime := currentTime.AddDate(0, 0, data.Day)
format := oldTime.Format(data.IndexFmt)

随后在主循环中使用strings.Contains检查每个索引是否包含该时间片段，一旦命中，便触发删除逻辑。这比解析索引名中的时间戳再比较更为通用，因为它兼容不同的命名规范。

第4步：精确执行删除操作

确认要删除的索引后，必须通过其确切的全名执行DeleteIndex操作，以此规避通配符误删的风险。

func DelIndex(data Data, index ...string) bool {
	client := GetEsClient(data)
	response, err := client.DeleteIndex(index...).Do(context.Background())
	if err != nil {
		fmt.Printf("delete index failed, err: %v\n", err)
		return false
	}
	return response.Acknowledged
}

完整代码实现与配置

为了使该工具具备实际可用性，我们通过JSON配置文件定义多个ES集群的连接信息及对应的保留策略。下面展示了完整的可运行代码逻辑框架，以及配套的config.json示例。

配置文件示例 (`config.json`)

支持同时定义多个Host，分别指定各自的日期格式和保留时长。

{
  "data": [
    {
      "host": "http://localhost:9200",
      "index_fmt": "20060102",
      "day": -1   // 仅保留最新1天以内的索引
    },
    {
      "host": "http://k8s-es-node:9200",
      "index_fmt": "2006.01.02",
      "day": -7   // 保留近7天
    }
  ]
}

完整逻辑代码

package main

import (
	"encoding/json"
	"fmt"
	"io/ioutil"
	"strings"
	"time"

	"context"
	"github.com/olivere/elastic/v7"
	"log"
	"os"
)

// 定义数据结构，便于JSON解析映射
type Config struct {
	Data []Data `json:"data"`
}

type Data struct {
	Host     string `json:"host"`
	IndexFmt string `json:"index_fmt"`
	Day      int    `json:"day"`
}

var config Config

// 加载配置文件
func Init() {
	JsonParse := NewJsonStruct()
	JsonParse.Load("config.json", &config)
}

type JsonStruct struct{}

func NewJsonStruct() *JsonStruct {
	return &JsonStruct{}
}

func (jst *JsonStruct) Load(filename string, v interface{}) {
	data, err := ioutil.ReadFile(filename)
	if err != nil {
		return
	}
	err = json.Unmarshal(data, v)
	if err != nil {
		return
	}
}

// 获取全部索引列表
func getIndex(data Data) map[string]interface{} {
	client := GetEsClient(data)
	mapping := client.GetMapping()
	service := mapping.Index("*")
	result, err := service.Do(context.Background())
	if err != nil {
		fmt.Printf("fetch indices failed, err: %v\n", err)
		return nil
	}
	return result
}

// 核心清理逻辑
func delIndex(data Data) {
	currentTime := time.Now()
	// 计算截止日期
	oldTime := currentTime.AddDate(0, 0, data.Day)
	format := oldTime.Format(data.IndexFmt)

	index := getIndex(data)
	if index == nil {
		return
	}

	for k := range index {
		fmt.Printf("Scanning index: %s (Target format: %s)\n", k, format)
		if find := strings.Contains(k, format); find {
			// 如果包含指定的日期片段，则执行删除
			if success := DelIndex(data, k); !success {
				fmt.Println("Failed to delete index:", k)
			} else {
				fmt.Println("Successfully deleted index:", k)
			}
		}
	}
}

// 省略了GetEsClient和DelIndex的具体实现，请参考上文独立函数块。
// 注意：在生产环境使用时，请集成前面的GetEsClient与DelIndex定义。

func main() {
	Init()
	// 遍历所有配置好的集群
	for i, datum := range config.Data {
		fmt.Printf("Starting cleanup for Cluster[%d] Host: [%s]\n", i, datum.Host)
		delIndex(datum)
	}
	fmt.Println("Cleanup process finished.")
}

排错与注意事项

SSL/TLS验证：如果你的ES集群开启了HTTPS并且使用自签名证书，需要在GetEsClient中添加elastic.SetHeader("X-XSRF-TOKEN", "...")或使用SetHttpClient传入忽略证书的自定义HTTP Client。
超时设置：当索引数量非常庞大时，获取全量索引的过程可能会耗时较长，建议在elastic.NewClient中增加elastic.SetGzip(true)提升网络传输速度，并适当延长上下文超时时限。
权限校验：确保用于连接的账号具备manage_index_states和delete权限。