package main

import (
	"git.quantgroup.cn/DevOps/enoch/pkg/dao"
	"git.quantgroup.cn/DevOps/enoch/pkg/global"
	"git.quantgroup.cn/DevOps/enoch/pkg/glog"
	"git.quantgroup.cn/DevOps/enoch/pkg/points"
	"github.com/Shopify/sarama"
	_ "github.com/mkevac/debugcharts"
	"net/http"
	_ "net/http/pprof"
	"os"
	"os/signal"
	"syscall"
	"time"
)

func handlerKafkaMsg() {
	db := dao.New(global.BatchSize, time.Second*60, global.InfluxDbAddress, global.DaoFileCacheDir)

	//处理调用链条信息
	braveTopic := global.Config.GetOrDefault(global.NamespaceTechSleuth, "tech.brave.kafkaTopic", "")
	braveConsumer, err := global.KafkaRecver.NewConsumer(global.KafkaGroup, []string{braveTopic}, func(msg *sarama.ConsumerMessage) {
		pointList, err := points.TraceBaseInfoToPoint(msg.Value)
		if err != nil {
			glog.Errorf("can not make trace point:", err)
			return
		}
		for _, point := range pointList {
			db.MsgProcess(point)
		}
	})
	if err != nil {
		glog.Fatal("监听kafka recver失败!")
	}

	//处理基本信息
	healthTopic := global.Config.GetOrDefault(global.NamespaceTechSleuth, "tech.enoch.kafkaTopic", "")
	healthConsumer, err := global.KafkaRecver.NewConsumer(global.KafkaGroup, []string{healthTopic}, func(msg *sarama.ConsumerMessage) {
		pointList, err := points.HostBaseInfoToPoint(msg.Value)
		if err != nil {
			glog.Errorf("can not make health point:", err)
			return
		}
		for _, point := range pointList {
			db.MsgProcess(point)
		}
	})
	if err != nil {
		glog.Fatal("监听kafka recver失败!")
	}

	glog.Info(global.AppName + "启动")
	//平滑退出
	sigterm := make(chan os.Signal, 1)
	signal.Notify(sigterm, syscall.SIGINT, syscall.SIGTERM)
	select {
	case <-sigterm:
		if braveConsumer != nil {
			braveConsumer.Close()
			glog.Info("braceConsumer平滑退出")
		}
		if healthConsumer != nil {
			healthConsumer.Close()
			glog.Info("healthConsumer平滑退出")
		}
		if db != nil {
			db.Close()
			glog.Info("dao平滑退出")
		}
		glog.Info(global.AppName + "平滑退出")
	}
}

func main() {
	//性能监控
	go func() {
		defer func() {
			if err := recover(); err != nil {
				glog.Error(err)
			}
		}()
		if err := http.ListenAndServe(":9999", nil); err != nil {
			glog.Error(err)
		}
	}()

	//初始化数据库（创建DB，创建连续查询）
	dao.DbInit()

	//处理消息
	handlerKafkaMsg()
	/*
		if quartz {
			log.Println("启动定时任务")
			job.AutoEmailPerformInfo()
		}

		//开启服务状态监控，当服务状态异常时，调用web-hook函数
		if denv == "dev" {		//开发环境
			go node_check.NodeCheck()
			job.AutoAlarm()
		} else if job.CheckIp("172.30.12.22") {  //生产环境，只有172.30.12.22执行
			go node_check.NodeCheck()
			job.AutoAlarm()
			continuous_queries.Load()		//连续查询设置只在生产环境上执行
		}

		go func() {
			_ = http.ListenAndServe("0.0.0.0:"+strconv.Itoa(intPort+1), nil)
		}()

		http.HandleFunc("/duration", service.DurationInterface)
		http.HandleFunc("/tech/health/check", func(writer http.ResponseWriter, request *http.Request) {
			writer.WriteHeader(http.StatusOK)
		})
		http.HandleFunc("/counter", service.CounterInterface)

		err = http.ListenAndServe(":"+port, nil)
		if err != nil {
			log.Fatalln("服务启动失败", err)
		}
	*/
}
