114 lines
2.1 KiB
Go
114 lines
2.1 KiB
Go
|
package esLogUtil
|
|||
|
|
|
|||
|
|
import (
|
|||
|
|
"bytes"
|
|||
|
|
"encoding/json"
|
|||
|
|
"sync"
|
|||
|
|
"time"
|
|||
|
|
|
|||
|
|
"github.com/elastic/go-elasticsearch/v8/esutil"
|
|||
|
|
"golang.org/x/net/context"
|
|||
|
|
"goutil/logUtil"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
var (
|
|||
|
|
indexMutex sync.Mutex
|
|||
|
|
indexer esutil.BulkIndexer
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
func timedReindex() {
|
|||
|
|
go func() {
|
|||
|
|
defer func() {
|
|||
|
|
if r := recover(); r != nil {
|
|||
|
|
logUtil.LogUnknownError(r)
|
|||
|
|
|
|||
|
|
timedReindex()
|
|||
|
|
time.Sleep(time.Second * 1)
|
|||
|
|
}
|
|||
|
|
}()
|
|||
|
|
|
|||
|
|
currentIndexName := getIndexName()
|
|||
|
|
for {
|
|||
|
|
//设置休眠
|
|||
|
|
time.Sleep(time.Second * 1)
|
|||
|
|
|
|||
|
|
newIndexName := getIndexName()
|
|||
|
|
if currentIndexName != getIndexName() {
|
|||
|
|
newIndexer(newIndexName)
|
|||
|
|
currentIndexName = newIndexName
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}()
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func newIndexer(newIndexName string) {
|
|||
|
|
if indexer != nil {
|
|||
|
|
closeIndex()
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
//wait until get the lock
|
|||
|
|
indexMutex.Lock()
|
|||
|
|
defer indexMutex.Unlock()
|
|||
|
|
|
|||
|
|
var err error
|
|||
|
|
indexer, err = esutil.NewBulkIndexer(esutil.BulkIndexerConfig{
|
|||
|
|
Index: getIndexName(), // The default index name
|
|||
|
|
Client: esClient, // The Elasticsearch client
|
|||
|
|
FlushInterval: time.Second, // The periodic flush interval
|
|||
|
|
})
|
|||
|
|
if err != nil {
|
|||
|
|
logUtil.ErrorLog("[%s]: Creating the indexer err: %s", serverModuleName, err)
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func closeIndex() {
|
|||
|
|
//wait until get the lock
|
|||
|
|
indexMutex.Lock()
|
|||
|
|
defer indexMutex.Unlock()
|
|||
|
|
|
|||
|
|
if indexer == nil {
|
|||
|
|
return
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
err := indexer.Close(context.Background())
|
|||
|
|
if err != nil {
|
|||
|
|
logUtil.ErrorLog("[%s]:Close err：%s", serverModuleName, err.Error())
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
indexer = nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// 批量保存到在线日志系统
|
|||
|
|
// 参数:
|
|||
|
|
//
|
|||
|
|
// 数量
|
|||
|
|
//
|
|||
|
|
// 返回值：
|
|||
|
|
//
|
|||
|
|
// 日志列表对象
|
|||
|
|
func bulkSendHandler(logObj EsLog) {
|
|||
|
|
if esClient == nil || indexer == nil {
|
|||
|
|
return
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
//try to get the lock in 10000 milliseconds,if cant obtain it,return false
|
|||
|
|
indexMutex.Lock()
|
|||
|
|
defer indexMutex.Unlock()
|
|||
|
|
|
|||
|
|
message, err := json.Marshal(logObj)
|
|||
|
|
if err != nil {
|
|||
|
|
logUtil.ErrorLog("[%s]: Marshal failed. Err:%s", serverModuleName, err)
|
|||
|
|
return
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
err = indexer.Add(
|
|||
|
|
context.Background(),
|
|||
|
|
esutil.BulkIndexerItem{
|
|||
|
|
Action: "index",
|
|||
|
|
Body: bytes.NewReader(message),
|
|||
|
|
})
|
|||
|
|
if err != nil {
|
|||
|
|
logUtil.ErrorLog("[%s]: Add data err：%s", serverModuleName, err.Error())
|
|||
|
|
}
|
|||
|
|
}
|