Go語言Elasticsearch數(shù)據(jù)清理工具思路詳解

更新時(shí)間：2021年10月08日 14:32:56 作者：July_whj

這篇文章主要介紹了Go語言Elasticsearch數(shù)據(jù)清理工具思路詳解,本文給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下

微服務(wù)架構(gòu)中收集通常大家都采用ELK進(jìn)行日志收集，同時(shí)我們還采用了SkyWalking進(jìn)行鏈路跟蹤，而SkyWalking數(shù)據(jù)存儲(chǔ)也用到了ES，SkyWalking每天產(chǎn)生大量的索引數(shù)據(jù)，如下：

WX20211008-104751@2x

這里一天大概產(chǎn)生了700左右個(gè)索引數(shù)據(jù)。對(duì)歷史的鏈路數(shù)據(jù)我們不做過多的保留。

這里我整理了個(gè)小工具，可以定期清理es數(shù)據(jù)。

一、清理思路

可以看到索引數(shù)據(jù)都是以日期結(jié)尾，我們可以根據(jù)日期去匹配索引數(shù)據(jù)，并對(duì)索引進(jìn)行刪除。這里需要考慮一點(diǎn)，有的Es服務(wù)開啟了索引保護(hù)機(jī)制，不能通過*index去刪除，只能通過索引的全名稱去刪除。所以我們整體流程如下：

1、獲取es服務(wù)中全部索引數(shù)據(jù)。

2、根據(jù)當(dāng)前時(shí)間-保留天數(shù)，獲取要?jiǎng)h除的日期。

3、通過字符串匹配，判斷索引中是否包含要?jiǎng)h除的日期，如果包含則進(jìn)行刪除。

4、工具友好性，我們可以通過配置文件配置ES服務(wù)地址、日期格式化類型、保留天數(shù)等信息。

二、代碼實(shí)現(xiàn)

2.1、獲取ES服務(wù)中全部索引數(shù)據(jù)

要獲取Es服務(wù)中全部索引數(shù)據(jù)，我們首先連接Es服務(wù)器，這里我們使用github.com/olivere/elastic/v7庫操作Es。

連接ES:

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),
        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
}

我們通過GetEsClient方法，連接ES,并返回client，供后續(xù)方法使用。這里的Data是包含了ES服務(wù)地址等信息，我們后面會(huì)給出Data的數(shù)據(jù)結(jié)構(gòu)。

獲取全部索引數(shù)據(jù)

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("create index failed, err: %v\n", err)
        return nil
    }
    return result
}

通過client.GetMapping().Index("*")API獲取es服務(wù)中全部的索引數(shù)據(jù)，并返回，數(shù)據(jù)格式如下：

WX20211008-110537@2x

這次我們獲取全部索引完成。

2.2、根據(jù)當(dāng)前時(shí)間-保留天數(shù)，獲取要?jiǎng)h除的日期

我們根據(jù)當(dāng)前時(shí)間-保留天數(shù)，獲取當(dāng)前需要?jiǎng)h除的日期數(shù)據(jù)。我們通過GoLang內(nèi)置的函數(shù)庫time完成該功能的實(shí)現(xiàn)。

currentTime := time.Now()//獲取當(dāng)前時(shí)間
oldTime := currentTime.AddDate(0, 0, data.Day)//通過配置文件獲取保留天數(shù)
format := oldTime.Format(data.IndexFmt)//通過配置文件獲取序列化日期格式

2.3、通過字符串匹配，判斷索引中是否包含要?jiǎng)h除的日期，如果包含則進(jìn)行刪除

這里通過字符串匹配進(jìn)行判斷是否需要?jiǎng)h除索引數(shù)據(jù)。

func delIndex(data Data) {
    currentTime := time.Now()
    oldTime := currentTime.AddDate(0, 0, data.Day)
    format := oldTime.Format(data.IndexFmt)
    index := getIndex(data)//獲取全部索引
    for k := range index {//遍歷索引數(shù)據(jù)
        fmt.Println("key:", k, "format:", format)
        if find := strings.Contains(k, format); find { //判斷索引中是否包含要?jiǎng)h除的日期格式，
            DelIndex(data, k)//如果包含則調(diào)用DelIndex方法刪除
        }
    }
}

// DelIndex 刪除 index
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
}

通過DeleteIndexAPI刪除指定的數(shù)據(jù)。

2.4、通過配置文件靈活配置數(shù)據(jù)

這里我們定義了Config和Data對(duì)象，對(duì)象結(jié)構(gòu)如下：

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

配置文件內(nèi)容如下：

{
  "data": [
    {
      "host": "http://ip1:9200",//服務(wù)IP
      "index_fmt": "20060102",//日期格式化
      "day": -1 //保留天數(shù) 保留1天
    },
    {
      "host": "http://ip2:9200/",
      "index_fmt": "20060102",
      "day": -1
    },
    {
      "host": "http://ip3:32093",
      "index_fmt": "2006.01.02",
      "day": -7  //保留天數(shù) 保留7天
    }
  ]
}

我們通過Init方法加載配置文件到Config;

var config Config
 
func Init() {
    JsonParse := NewJsonStruct()
    //下面使用的是相對(duì)路徑，config.json文件和main.go文件處于同一目錄下
    JsonParse.Load("config/config.json", &config)
}
 
type JsonStruct struct {
}
 
func NewJsonStruct() *JsonStruct {
    return &JsonStruct{}
}
 
func (jst *JsonStruct) Load(filename string, v interface{}) {
    //ReadFile函數(shù)會(huì)讀取文件的全部?jī)?nèi)容，并將結(jié)果以[]byte類型返回
    data, err := ioutil.ReadFile(filename)
    if err != nil {
        return
    }
 
    //讀取的數(shù)據(jù)為json格式，需要進(jìn)行解碼
    err = json.Unmarshal(data, v)
    if err != nil {
        return
    }
}

編寫Main方法運(yùn)行程序：

func main() {
    Init()
    for i, datum := range config.Data {
        fmt.Printf("config data Host is [%s], fmt is [%s]\n", datum.Host, datum.IndexFmt)
        println(i)
        delIndex(datum)
    }
}

這里我們依然遍歷配置文件中的多個(gè)服務(wù)配置?？梢酝瑫r(shí)管理多個(gè)Es服務(wù)。

三、完整代碼

package main
 
import (
    "encoding/json"
    "fmt"
    "io/ioutil"
    "strings"
    "time"
)
 
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()
    //下面使用的是相對(duì)路徑，config.json文件和main.go文件處于同一目錄下
    JsonParse.Load("config/config.json", &config)
}
 
type JsonStruct struct {
}
 
func NewJsonStruct() *JsonStruct {
    return &JsonStruct{}
}
 
func (jst *JsonStruct) Load(filename string, v interface{}) {
    //ReadFile函數(shù)會(huì)讀取文件的全部?jī)?nèi)容，并將結(jié)果以[]byte類型返回
    data, err := ioutil.ReadFile(filename)
    if err != nil {
        return
    }
 
    //讀取的數(shù)據(jù)為json格式，需要進(jìn)行解碼
    err = json.Unmarshal(data, v)
    if err != nil {
        return
    }
}
 
func delIndex(data Data) {
    currentTime := time.Now()
    oldTime := currentTime.AddDate(0, 0, data.Day)
    format := oldTime.Format(data.IndexFmt)
    index := getIndex(data)
    for k := range index {
        fmt.Println("key:", k, "format:", format)
        if find := strings.Contains(k, format); find {
            DelIndex(data, k)
        }
    }
}
 
func main() {
    Init()
    for i, datum := range config.Data {
        fmt.Printf("config data Host is [%s], fmt is [%s]\n", datum.Host, datum.IndexFmt)
        println(i)
        delIndex(datum)
    }
}

package main
 
import (
    "context"
    "fmt"
    "github.com/olivere/elastic/v7"
    "log"
    "os"
    "time"
)
 
// GetEsClient 初始化客戶端
func GetEsClient(data Data) *elastic.Client {
    Init()
    file := "./eslog.log"
    logFile, _ := os.OpenFile(file, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0766) // 應(yīng)該判斷error，此處簡(jiǎn)略
    client, err := elastic.NewClient(
        elastic.SetURL(data.Host),
        elastic.SetSniff(false),
        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
}
 
// IsDocExists 判斷索引是否存儲(chǔ)
func IsDocExists(data Data, id string, index string) bool {
    client := GetEsClient(data)
    defer client.Stop()
    exist, _ := client.Exists().Index(index).Id(id).Do(context.Background())
    if !exist {
        log.Println("ID may be incorrect! ", id)
        return false
    }
    return true
}
 
// PingNode 是否聯(lián)通
func PingNode(data Data) {
    start := time.Now()
    client := GetEsClient(data)
    info, code, err := client.Ping(data.Host).Do(context.Background())
    if err != nil {
        fmt.Printf("ping es failed, err: %v", err)
    }
    duration := time.Since(start)
    fmt.Printf("cost time: %v\n", duration)
    fmt.Printf("Elasticsearch returned with code %d and version %s\n", code, info.Version.Number)
}
 
// GetDoc 獲取文檔
func GetDoc(data Data, id string, index string) (*elastic.GetResult, error) {
    client := GetEsClient(data)
    defer client.Stop()
    if !IsDocExists(data, id, index) {
        return nil, fmt.Errorf("id不存在")
    }
    esResponse, err := client.Get().Index(index).Id(id).Do(context.Background())
    if err != nil {
        return nil, err
    }
    return esResponse, nil
}
 
// CreateIndex 創(chuàng)建 index
func CreateIndex(data Data, index, mapping string) bool {
    client := GetEsClient(data)
    result, err := client.CreateIndex(index).BodyString(mapping).Do(context.Background())
    if err != nil {
        fmt.Printf("create index failed, err: %v\n", err)
        return false
    }
    return result.Acknowledged
}
 
// DelIndex 刪除 index
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
}
 
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("create index failed, err: %v\n", err)
        return nil
    }
    return result
}

代碼已經(jīng)上傳github需要的可自行下載。

到此這篇關(guān)于Go語言Elasticsearch數(shù)據(jù)清理工具的文章就介紹到這了,更多相關(guān)Go Elasticsearch數(shù)據(jù)清理工具內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: