package consumer

import (
	"encoding/json"
	"git.quantgroup.cn/DevOps/enoch/pkg/end_points"
	"git.quantgroup.cn/DevOps/enoch/service/log"
	"github.com/influxdata/influxdb/client/v2"
	"math/big"
	"net"
	"time"
)

var sysNameIndex = make(map[int64]bool)
var metricsPointSlice = make([]*client.Point, 0, batchSize)
var healthPointSlice = make([]*client.Point, 0, batchSize)

type HealthMessageHandler struct {
}

func (HealthMessageHandler) MsgProcess(msg string) {
	chunkMsg := end_points.ChunkMsg{}
	err := json.Unmarshal([]byte(msg), &chunkMsg)
	if err != nil {
		logger.Error.Println("healthMessageHandler解析json失败:", err)
		logger.Error.Println(msg)
		return
	}
	buildMsg(chunkMsg)
}

func (HealthMessageHandler) Destroy() {
	if len(metricsPointSlice) > 0 {
		logger.Info.Println("metricsMessageHandler 提交本地缓存数据：", len(metricsPointSlice))
		batchWrite(metricsPointSlice)
	}

	if len(healthPointSlice) > 0 {
		logger.Info.Println("HealthMessageHandler 提交本地缓存数据：", len(healthPointSlice))
		batchWrite(healthPointSlice)
	}
}

func buildHealthInfluxMsg(appName string, ip string, timestamp time.Time, submitLimit int, db *[]byte) {
	tags := make(map[string]string, )
	tags["sys_name"] = appName
	tags["host"] = ip
	fields := make(map[string]interface{})
	dbInfo := end_points.DBInfo{}
	err := json.Unmarshal(*db, &dbInfo)
	if err != nil {
		dbDetails := end_points.DBDetail{}
		err = json.Unmarshal(*db, &dbDetails)
		if err == nil {
			fields[dbDetails.Details.Database] = isOK(dbDetails.Status.Code)
		}
	}else {
		for k, v := range dbInfo.Details {
			var fieldName = v.Details.Database + "—" + k
			fields[fieldName] = isOK(v.Status.Code)
		}
	}

	if len(healthPointSlice) >= submitLimit {
		go batchWrite(healthPointSlice)
		healthPointSlice = make([]*client.Point, 0, batchSize)
	}
	point, _ := client.NewPoint("health_info", tags, fields, timestamp)
	healthPointSlice = append(healthPointSlice, point)
}

func buildMetricsInfluxMsg(appName string, ip string, timestamp time.Time, submitLimit int, health end_points.Health, metrics end_points.MetricsInfo) {
	tags := make(map[string]string, )
	fields := make(map[string]interface{})
	tags["sys_name"] = appName
	tags["host"] = ip

	status := health.Status
	fields["sever_status"] = isOK(status.Code)

	redis := health.Details.Redis
	fields["redis_status"] = isOK(redis.Status.Code)
	diskSpace := health.Details.DiskSpace.Details

	fields["disk_tol"] = diskSpace.Total
	fields["disk_free"] = diskSpace.Free
	fields["disk_threshold"] = diskSpace.Threshold

	fields["mem_tol"] = metrics.Mem
	fields["mem_free"] = metrics.MemFree

	fields["heap_tol"] = metrics.Heap
	fields["heap_init"] = metrics.HeapInit
	fields["heap_used"] = metrics.HeapUsed

	fields["nonheap_tol"] = metrics.Nonheap
	fields["nonheap_init"] = metrics.NonheapInit
	fields["nonheap_used"] = metrics.NonheapUsed
	fields["nonheap_commit"] = metrics.NonheapCommitted

	fields["thread_tol"] = metrics.ThreadsTotalStarted
	fields["thread_peak"] = metrics.ThreadsPeak
	fields["thread_daemon"] = metrics.ThreadsDaemon

	fields["class_tol"] = metrics.Classes
	fields["class_loaded"] = metrics.ClassesLoaded
	fields["class_unloaded"] = metrics.ClassesUnloaded

	fields["gc_parnew_count"] = metrics.GcParnewCount
	fields["gc_parnew_time"] = metrics.GcParnewTime
	fields["gc_concurrent_mark_sweep"] = metrics.GcConcurrentmarksweepCount
	fields["gc_concurrent_mark_time"] = metrics.GcConcurrentmarksweepTime

	fields["uptime"] = metrics.Uptime
	fields["instance_uptime"] = metrics.InstanceUptime
	fields["system_load_average"] = metrics.SystemloadAverage
	fields["processors"] = metrics.Processors

	if len(metricsPointSlice) >= submitLimit {
		go batchWrite(metricsPointSlice)
		metricsPointSlice = make([]*client.Point, 0, batchSize)
	}
	point, _ := client.NewPoint("machine_info", tags, fields, timestamp)
	metricsPointSlice = append(metricsPointSlice, point)
}

func buildMsg(chunkMsg end_points.ChunkMsg) {
	var ip = inetAtoN(chunkMsg.Ip)
	sysNameIndex[ip] = true
	var sysNameCount = len(sysNameIndex)
	for _, p := range chunkMsg.EndPoints {

		var appName = chunkMsg.AppName
		var ip = chunkMsg.Ip

		unix := time.Unix(0, p.Timestamp*1000000)
		//metricsInfo
		buildMetricsInfluxMsg(appName, ip, unix, sysNameCount, p.Health, p.Metrics)

		//health_info
		dbByte, _ := json.Marshal(p.Health.Details.Db)
		buildHealthInfluxMsg(appName, ip, unix, sysNameCount, &dbByte)
	}
}

func inetAtoN(ip string) int64 {
	ret := big.NewInt(0)
	ret.SetBytes(net.ParseIP(ip).To4())
	return ret.Int64()
}

func isOK(code string) int {
	if "UP" == code {
		return 1
	}
	return 0
}
