Go語(yǔ)言Elasticsearch數(shù)據(jù)清理工具思路詳解
更新時(shí)間：2021年10月08日 14:32:56 作者：July_whj
這篇文章主要介紹了Go語(yǔ)言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ù)我們不做過(guò)多的保留。
這里我整理了個(gè)小工具，可以定期清理es數(shù)據(jù)。
一、清理思路
可以看到索引數(shù)據(jù)都是以日期結(jié)尾，我們可以根據(jù)日期去匹配索引數(shù)據(jù)，并對(duì)索引進(jìn)行刪除。這里需要考慮一點(diǎn)，有的Es服務(wù)開(kāi)啟了索引保護(hù)機(jī)制，不能通過(guò)*index去刪除，只能通過(guò)索引的全名稱去刪除。所以我們整體流程如下：
1、獲取es服務(wù)中全部索引數(shù)據(jù)。
2、根據(jù)當(dāng)前時(shí)間-保留天數(shù)，獲取要?jiǎng)h除的日期。
3、通過(guò)字符串匹配，判斷索引中是否包含要?jiǎng)h除的日期，如果包含則進(jìn)行刪除。
4、工具友好性，我們可以通過(guò)配置文件配置ES服務(wù)地址、日期格式化類型、保留天數(shù)等信息。
二、代碼實(shí)現(xiàn)
2.1、獲取ES服務(wù)中全部索引數(shù)據(jù)
要獲取Es服務(wù)中全部索引數(shù)據(jù)，我們首先連接Es服務(wù)器，這里我們使用github.com/olivere/elastic/v7庫(kù)操作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
}
我們通過(guò)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
}
通過(guò)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ù)。我們通過(guò)GoLang內(nèi)置的函數(shù)庫(kù)time完成該功能的實(shí)現(xiàn)。
currentTime := time.Now()//獲取當(dāng)前時(shí)間 oldTime := currentTime.AddDate(0, 0, data.Day)//通過(guò)配置文件獲取保留天數(shù) format := oldTime.Format(data.IndexFmt)//通過(guò)配置文件獲取序列化日期格式
2.3、通過(guò)字符串匹配，判斷索引中是否包含要?jiǎng)h除的日期，如果包含則進(jìn)行刪除
這里通過(guò)字符串匹配進(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
}
通過(guò)DeleteIndexAPI刪除指定的數(shù)據(jù)。
2.4、通過(guò)配置文件靈活配置數(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天
}
]
}
我們通過(guò)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
}
}
編寫(xiě)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語(yǔ)言Elasticsearch數(shù)據(jù)清理工具的文章就介紹到這了,更多相關(guān)Go Elasticsearch數(shù)據(jù)清理工具內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
這篇文章主要介紹了Go標(biāo)準(zhǔn)庫(kù)template模板用法詳解;包括GO模板注釋,作用域，語(yǔ)法，函數(shù)等知識(shí)，需要的朋友可以參考下2022-04-04
GORM 是一個(gè)用于 Go 語(yǔ)言的 ORM（對(duì)象關(guān)系映射）庫(kù),它提供了一種簡(jiǎn)單而強(qiáng)大的方式來(lái)與數(shù)據(jù)庫(kù)進(jìn)行交互,GORM 支持多種數(shù)據(jù)庫(kù),并且提供了豐富的功能,如自動(dòng)遷移、預(yù)加載、事務(wù)管理等,文中通過(guò)代碼示例講解的非常詳細(xì),需要的朋友可以參考下2024-11-11
Golang unsafe.Sizeof函數(shù)代碼示例使用解析
這篇文章主要為大家介紹了Golang unsafe.Sizeof函數(shù)代碼示例使用解析,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2023-12-12
這篇文章主要為大家介紹了Go常量基礎(chǔ)概念包括常量的聲明更改只讀,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2023-08-08
使用Golang的gomail庫(kù)實(shí)現(xiàn)郵件發(fā)送功能
本篇博客詳細(xì)介紹了如何使用Golang語(yǔ)言中的gomail庫(kù)來(lái)實(shí)現(xiàn)郵件發(fā)送的功能,首先,需要準(zhǔn)備工作,包括安裝Golang環(huán)境、gomail庫(kù),以及申請(qǐng)126郵箱的SMTP服務(wù)和獲取授權(quán)碼,其次,介紹了在config文件中配置SMTP服務(wù)器信息的步驟2024-10-10
在?Go?語(yǔ)言中處理多行文本的正則表達(dá)式時(shí),換行符常常會(huì)導(dǎo)致意外的匹配結(jié)果,本文為大家整理了一些解決方法,有需要的小伙伴可以參考一下2024-11-11
使用Go語(yǔ)言實(shí)現(xiàn)簡(jiǎn)單聊天系統(tǒng)
本文介紹了如何使用Go語(yǔ)言和WebSocket技術(shù)構(gòu)建一個(gè)簡(jiǎn)單的多人聊天室系統(tǒng),包括客戶端連接管理、消息廣播和并發(fā)處理,最后,通過(guò)編寫(xiě)main.go、hub.go和client.go等核心代碼模塊,具有一定的參考價(jià)值,感興趣的可以了解一下2024-10-10
Go語(yǔ)言實(shí)現(xiàn)運(yùn)算符重載的方法詳解
這篇文章主要為大家詳細(xì)介紹了如何利用Go語(yǔ)言實(shí)現(xiàn)運(yùn)算符重載的方法，文中的示例代碼講解詳細(xì)，感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下2022-09-09
最近在做性能優(yōu)化,有個(gè)函數(shù)里面的耗時(shí)特別長(zhǎng),看里面的操作大多是一些字符串拼接的操作,而字符串拼接在 golang 里面其實(shí)有很多種實(shí)現(xiàn),下面這篇文章主要給大家介紹了關(guān)于Golang語(yǔ)言如何高效拼接字符串的相關(guān)資料,需要的朋友可以參考下2021-11-11
golang float和科學(xué)計(jì)數(shù)法轉(zhuǎn)字符串的實(shí)現(xiàn)方式
這篇文章主要介紹了golang float和科學(xué)計(jì)數(shù)法轉(zhuǎn)字符串的實(shí)現(xiàn)方式,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2025-05-05