jy-data-extract/monitor/task.go

package main

import (
	"fmt"
	"github.com/cron"
	"go.mongodb.org/mongo-driver/bson"
	"mongodb"
	"net/http"
	"qfw/util"
	"strconv"
	"time"
)

var taskA = 0		// 增量统计跳过

func TimeTask() {
	//StartTask()
	c := cron.New()
	cronstr := "0 0 * * * ?" // 每小时执行一次
	_ = c.AddFunc(cronstr, func() { StartTask() })
	c.Start()
}

func StartTask() {
	util.Debug("Start Task...")
	result := MgoCount()
	result = EsCheck(result)
	if result != "" {
		util.Debug(result)
		SendMail(result)
	}
	util.Debug("Task Over...")
}

// 统计mgo数据
func MgoCount() string {
	defer util.Catch()
	sess := Mgo.GetMgoConn()
	defer Mgo.DestoryMongoConn(sess)
	result := ""
	// biddingback
	count := Mgo.Count("bidding_back", nil)
	util.Debug("bidding_back---", BiddingBackSize, count)
	if int64(count) != BiddingBackSize {
		result = fmt.Sprintf("bindding_back数据异常，统计结果%d，往常数据量%d", count, BiddingBackSize)
		result += "<br>"
	}
	// bidding
	st := time.Now().Unix() - 3600
	currentLastId := fmt.Sprintf("%x0000000000000000", st)
	//	对比内存中存量id的统计结果
	q1 := bson.M{"_id": bson.M{"$lte": mongodb.StringTOBsonId(LastStockId)}}
	count1 := Mgo.Count("bidding", q1)
	util.Debug("bidding---", LastStockSize, count1, q1)
	if int64(count1) != LastStockSize {
		result = fmt.Sprintf("bindding数据存量统计异常，统计条件%v，统计结果%d，上次统计数据量%d", q1, count1, LastStockSize)
		result += "<br>"
	}else {
		//	无异常，替换成当前时间点(当前时间点往前一个小时的id)存量的统计结果
		LastStockId = currentLastId
		q2 := bson.M{"_id": bson.M{"$lte": mongodb.StringTOBsonId(LastStockId)}}
		count2 := Mgo.Count("bidding", q2)
		LastStockSize = int64(count2)
	}
	//	增量
	et := time.Now().Unix()
	currentId := fmt.Sprintf("%x0000000000000000", et)
	q3 := bson.M{"_id": bson.M{"$gt": mongodb.StringTOBsonId(currentLastId), "$lte": mongodb.StringTOBsonId(currentId)}}
	count3 := Mgo.Count("bidding", q3)
	util.Debug("bidding---", count3, q3)
	if int64(count3) == 0 {
		if taskA > 1 {
			result += fmt.Sprintf("bindding数据增量统计异常，统计条件%v，统计结果%d", q3, count3)
			result += "<br><br>"
			taskA = 0
		}else {
			taskA ++
		}
	}
	return result
}

func EsCheck(result string) string {
	client := Es.GetEsConn()
	defer Es.DestoryEsConn(client)
	resp, _ := client.ClusterHealth().Do()
	util.Debug(*resp)
	if resp.Status != "green" {
		result += "<br>" + "检索库异常，异常内容：" + "<br>" +
			"&nbsp;&nbsp;&nbsp;&nbsp;" + "cluster_name：" + resp.ClusterName + "<br>" +
			"&nbsp;&nbsp;&nbsp;&nbsp;" + "status：" + resp.Status + "<br>" +
			"&nbsp;&nbsp;&nbsp;&nbsp;" + "number_of_nodes：" + strconv.Itoa(resp.NumberOfNodes) + "<br>" +
			"&nbsp;&nbsp;&nbsp;&nbsp;" + "number_of_data_nodes：" + strconv.Itoa(resp.NumberOfDataNodes) + "<br>" +
			"&nbsp;&nbsp;&nbsp;&nbsp;" + "number_of_data_nodes：" + strconv.Itoa(resp.ActivePrimaryShards) + "<br>" +
			"&nbsp;&nbsp;&nbsp;&nbsp;" + "active_shards：" + strconv.Itoa(resp.ActiveShards) + "<br>" +
			"&nbsp;&nbsp;&nbsp;&nbsp;" + "relocating_shards：" + strconv.Itoa(resp.RelocatingShards) + "<br>" +
			"&nbsp;&nbsp;&nbsp;&nbsp;" + "initialized_shards：" + strconv.Itoa(resp.InitializedShards) + "<br>" +
			"&nbsp;&nbsp;&nbsp;&nbsp;" + "unassigned_shards：" + strconv.Itoa(resp.UnassignedShards) + "<br>"
	}
	return result
}

func SendMail(report string) {
	_, _ = http.Get(fmt.Sprintf("%s?to=%s&title=%s&body=%s", api, to, "异常报告！", report))
}