web_hghl_server/web/service/srv_search/es.go

package srv_search

import (
	"context"
	"errors"
	"fmt"
	"time"
	"videoweb/pub/common/log"
	"videoweb/pub/common/srvdef"
	"videoweb/pub/common/utils"
	"videoweb/pub/esmanager"
	"videoweb/pub/models/esmod"

	"go.mongodb.org/mongo-driver/bson/primitive"
)

var syncIsRuining = false

func StartES() {
	log.Info("srv_search:Start...")
	defer log.Info("srv_search:Start!!!")
	startSyncEs()
	_, _ = skdCron.AddFunc("0 0/17 * * * ?", func() {
		startSyncEs()
	})
	skdCron.Start()
}

func startSyncEs() {
	ctx := context.Background()
	updateEsData(ctx, esmod.GetVideoIndex(), getVideoEsData)
	updateEsData(ctx, esmod.GetPlayletIndex(), getPlayletEsData)
	updateEsData(ctx, esmod.GetActressIndex(), getActressEsData)
}

func StopES() {
	log.Info("srv_search:Stop...")
	defer log.Info("srv_search:Stop!!!")
}

func updateEsData(ctx context.Context, esIndex string, fn func(ctx context.Context, startTime, endTime *time.Time, esIndex string) (esmanager.M, *srvdef.SrvErr)) {
	log.Info("srv_search:do...", log.Any("ESIndex", esIndex))
	defer log.Info("srv_search:do end", log.Any("EsIndex", esIndex))
	if fn == nil {
		return
	}
	esRecord, serr := esmod.GetRecordTimeByESIndex(ctx, esIndex)
	if serr != nil {
		log.Error(fmt.Sprintf("srv_search:startEngine GetRecordByESIndex failed! Index:%v,err:%+v:", esIndex, serr))
		return
	}
	isNew := false
	if esRecord == nil {
		isNew = true
		esRecord = &esmod.ESRecord{
			Id:         primitive.NewObjectID(),
			EsIndex:    esIndex,
			RecordTime: time.Time{},
			Timestamp:  esmod.ESTimestamp{},
			UpdatedAt:  time.Time{},
			CreatedAt:  time.Now(),
		}
	}
	var now = time.Now()
	var esData esmanager.M
	// 跨天凌晨5点重置所有数据
	if isNew || ((now.Hour() >= 5 && esRecord.UpdatedAt.Local().Hour() < 5) || esRecord.UpdatedAt.Local().Day() < now.Day()) {
		log.Info("srv_search:updateEsData 全量同步", log.Any("index", esIndex), log.Any("isNew", isNew))
		esData, serr = fn(ctx, nil, nil, esIndex)
		if serr != nil {
			log.Warn("updateEsData fn return err", log.Any("index", esIndex), log.E(serr))
			return
		}
		log.Info("srv_search:updateEsData 全量数据拉取完成", log.Any("index", esIndex), log.Any("count", len(esData)))
		if len(esData) == 0 {
			if err := esmod.SetOneEsRecordTime(nil, esIndex, now); err != nil {
				return
			}
			return
		}
		isUpdate, err := commit(esData, esIndex, time.Now())
		if err != nil {
			log.Error("Elastic sync video data RunSync commit err! ", log.E(err))
			return
		}
		log.Info("srv_search:updateEsData 全量写入ES完成", log.Any("index", esIndex), log.Any("isUpdate", isUpdate))
		if isUpdate {
			if err := esmod.SetOneEsRecordTime(nil, esIndex, now); err != nil {
				return
			}
		}
	} else {
		timestamp := esRecord.RecordTime
		var recentMinute = utils.RecentMinute(now, esmod.FiveMinuteScale)
		log.Info("srv_search:updateEsData 增量同步", log.Any("index", esIndex),
			log.Any("from", timestamp.Format("2006-01-02 15:04:05")),
			log.Any("to", recentMinute.Format("2006-01-02 15:04:05")))
		totalIncr := 0
		for recentMinute.After(timestamp) {
			var end = timestamp.Add(esmod.FiveMinuteScale * time.Minute)
			if end.After(recentMinute) {
				end = recentMinute
			}
			esData, serr = fn(ctx, &timestamp, &end, esIndex)
			if serr != nil {
				log.Warn("updateEsData fn return err", log.Any("index", esIndex), log.E(serr))
				return
			}
			if esData != nil && len(esData) > 0 {
				totalIncr += len(esData)
				log.Info("srv_search:updateEsData 增量批次",
					log.Any("index", esIndex),
					log.Any("window_start", timestamp.Format("2006-01-02 15:04:05")),
					log.Any("window_end", end.Format("2006-01-02 15:04:05")),
					log.Any("count", len(esData)))
				isUpdate, err := commit(esData, esIndex, time.Now())
				if err != nil {
					log.Error("Elastic sync video data RunSync commit err! ", log.E(err))
					return
				}
				if isUpdate {
					if err := esmod.SetOneEsRecordTime(nil, esIndex, end); err != nil {
						return
					}
				}
			} else {
				if err := esmod.SetOneEsRecordTime(nil, esIndex, end); err != nil {
					return
				}
			}
			timestamp = end
		}
		log.Info("srv_search:updateEsData 增量同步完成", log.Any("index", esIndex), log.Any("totalCount", totalIncr))
	}
}

func commit(data esmanager.M, esIndex string, syncTime time.Time) (bool, error) {
	bulkData := handleBulkCountLimit(data)
	isHaveSuccess := false
	failedBatches := make([]int, 0)
	log.Info("srv_search:commit 开始写入ES", log.Any("index", esIndex),
		log.Any("totalDocs", len(data)), log.Any("batches", len(bulkData)))
	if bulkData != nil && len(bulkData) > 0 {
		for i, doc := range bulkData {
			if len(doc) == 0 {
				continue
			}
			if err := esmanager.Instance.Bulk(esIndex, doc); err != nil {
				log.Warn(fmt.Sprintf("Elastic sync commit video data SyncDataToES failed!, err:%+v\n", err),
					log.Any("index", esIndex), log.Any("batch", i), log.Any("batchSize", len(doc)))
				failedBatches = append(failedBatches, i)
			} else {
				isHaveSuccess = true
				log.Info("srv_search:commit 批次写入成功", log.Any("index", esIndex),
					log.Any("batch", i), log.Any("batchSize", len(doc)))
			}
		}
	}
	if len(failedBatches) > 0 {
		return isHaveSuccess, fmt.Errorf("elastic sync commit failed batches: %v", failedBatches)
	}
	if isHaveSuccess {
		if err := esmanager.Instance.Refresh(esIndex); err != nil {
			return isHaveSuccess, err
		}
		return isHaveSuccess, nil
	} else {
		return isHaveSuccess, errors.New("Elastic sync commit video data UpsertOneTrans failed!")
	}
}

func handleBulkCountLimit(data esmanager.M) []esmanager.M {
	if data == nil || len(data) == 0 {
		return make([]esmanager.M, 0)
	}
	rsl := make([]esmanager.M, 0)
	rsl = append(rsl, make(esmanager.M, batchSize))
	index := 0
	for k, v := range data {
		rsl[index][k] = v
		if len(rsl[index]) >= batchSize {
			index++
			rsl = append(rsl, make(esmanager.M, batchSize))
		}
	}
	return rsl
}