data_monitor/vps_server/src/main.go

package main

import (
	"fmt"
	"github.com/cron"
	"log"
	"net/http"
	qu "qfw/util"
	"strings"
	"sync"
	"time"
)
var (
	sysconfig			map[string]interface{} //配置文件
	port				string
	save_mgo        	*MongodbSim
	idsArr,extArr		[]string
	vpsTmp				map[string]map[string]interface{}
	extTmp				map[string]interface{}
	during,isErr		int64
	test				map[string]interface{}
	updatelock 			sync.Mutex
	save_coll_name		string
)




func init()  {
	//加载配置文件
	log.Println("加载...")
	qu.ReadConfig(&sysconfig)
	port = sysconfig["port"].(string)
	arr1 := sysconfig["vpsIDs"].([]interface{})
	idsArr = qu.ObjArrToStringArr(arr1)
	arr2 := sysconfig["extIDs"].([]interface{})
	extArr = qu.ObjArrToStringArr(arr2)

	resetRecordData()

	during = qu.Int64All(sysconfig["during"])
	isErr = qu.Int64All(sysconfig["isErr"])

	saveconf := sysconfig["save_mgodb"].(map[string]interface{})
	save_coll_name = qu.ObjToString(saveconf["coll"])
	save_mgo = &MongodbSim{
		MongodbAddr: saveconf["addr"].(string),
		DbName:      saveconf["db"].(string),
		Size:        qu.IntAllDef(saveconf["pool"], 5),
	}
	save_mgo.InitPool()
	log.Println("准备完毕...")
}

func main() {

	//http://monitor.spdata.jianyu360.com/，程序端口7811
	addr := ":"+port
	http.HandleFunc("/", handler)
	go http.ListenAndServe(addr, nil)

	//每隔1分钟执行一次：0 */1 * * * ?   每隔5秒执行一次：*/5 * * * * ?

	spec :=fmt.Sprintf("30 */%d * * * ?",during)
	spec_reset := "0 0 0 * * ?"
	spec_extract :=fmt.Sprintf("20 */%d * * * ?",during)
	c := cron.New()
	c.AddFunc(spec, func() { taskFinishing()})
	c.AddFunc(spec_reset, func() { resetRecordData()})
	c.AddFunc(spec_extract, func() { extractRunning()})
	c.Start()
	time.Sleep(99999 * time.Hour)
}

func handler(w http.ResponseWriter, r *http.Request) {
	updatelock.Lock()
	r.ParseForm() //解析参数，默认是不会解析的
	if r.Method == "GET" {
		vpsid ,process,isProMail:= "",int64(0),int64(0)
		extid :=""
		for k, v := range r.Form {
			if k=="id" {
				vpsid = strings.Join(v, "")
				isProMail = qu.Int64All(vpsTmp[vpsid]["isProMail"])
			}else if k=="process" {
				process = qu.Int64All(strings.Join(v, ""))
				if process==0 {
					isProMail = 0
				}
			}else if k=="extract" {
				extid = strings.Join(v, "")
			}else {

			}
		}
		if vpsid!="" {
			vpsTmp[vpsid] = map[string]interface{}{
				"isHeart":1,
				"isErrNum":0,
				"isVpsMail":0,   //收到心跳-vps邮件重置为0，可以发
				"isProcess":process,
				"isProMail":isProMail,
			}
		}

		if extid!="" {
			extTmp[extid] = time.Now().Unix()
		}

	} else if r.Method == "POST" {

	} else {

	}

	updatelock.Unlock()
}

//重置数据
func resetRecordData()  {
	updatelock.Lock()
	vpsTmp = make(map[string]map[string]interface{},0)
	for _,v := range idsArr{
		id := qu.ObjToString(v)
		vpsTmp[id] = map[string]interface{}{
			"isHeart":0,
			"isErrNum":0,
			"isProcess" : 0,
			"isVpsMail":0,
			"isProMail":0,
		}
	}
	log.Println("重置数据vps...",len(vpsTmp))


	extTmp = make(map[string]interface{},0)
	for _,v := range extArr{
		extid := qu.ObjToString(v)
		extTmp[extid] = time.Now().Unix()

	}
	log.Println("重置ext数据...")


	updatelock.Unlock()
}
//不断监听处理
func taskFinishing()  {
	//加锁
	updatelock.Lock()
	log.Println("...处理一次...")
	isVpsMailContent,isProMailContent:= "",""
	for _ , vpsid := range idsArr {
		//此标识-是否正常
		//log.Println("原:",dataTmp[vpsid])
		isHeart,isProcess:= qu.Int64All(vpsTmp[vpsid]["isHeart"]),qu.Int64All(vpsTmp[vpsid]["isProcess"])
		isErrNum := int64(0)
		isVpsMail,isProMail := qu.Int64All(vpsTmp[vpsid]["isVpsMail"]),qu.Int64All(vpsTmp[vpsid]["isProMail"])
		if isVpsMail == 1 { //送过邮件了
			//log.Println("发过vps邮件","心跳:",isHeart,"次数:",isErrNum,"下载器:",isProcess,"vps邮件:",isVpsMail,"pro邮件:",isProMail)
			continue
		}
		if isHeart == 0 { //未接收心跳反应，错误+1
			isErrNum = qu.Int64All(vpsTmp[vpsid]["isErrNum"])
			isErrNum ++
		}
		if isErrNum > isErr { //错误超过一定次数，发邮件vps异常
			isErrNum = 0
			if isVpsMailContent == ""{
				isVpsMailContent = vpsid
			}else {
				isVpsMailContent = isVpsMailContent+","+vpsid
			}
			isVpsMail = 1
		}

		if isProcess == 1  && isProMail==0 {//下载器异常-未发送过下载器异常情况
			if isProMailContent == ""{
				isProMailContent = vpsid
			}else {
				isProMailContent = isProMailContent+","+vpsid
			}
			isProMail = 1
			isProcess = 0
		}

		//log.Println("处理后:","心跳:",0,"次数:",isErrNum,"下载器:",isProcess,"vps邮件:",isVpsMail,"pro邮件:",isProMail)

		vpsTmp[vpsid] = map[string]interface{}{
			"isHeart":0,
			"isErrNum":isErrNum,
			"isProcess":isProcess,
			"isVpsMail":isVpsMail,
			"isProMail" : isProMail,
		}

	}

	//log.Println("处理后",isProMailContent)

	if isVpsMailContent!=""{
		log.Println("发邮件：vps异常...",isVpsMailContent)
		comeintime :=qu.Int64All(time.Now().Unix())//日志记录
		save_mgo.Save(save_coll_name, map[string]interface{}{
			"name":"vps",
			"comeintime":comeintime,
			"date":qu.FormatDateByInt64(&comeintime, qu.DATEFORMAT),
			"detail" : isVpsMailContent,
		})

		sendErrVpsMailApi("vps",isVpsMailContent)
	}else {
		if isProMailContent !="" {
			log.Println("发邮件：下载器异常...",isProMailContent)
			comeintime :=qu.Int64All(time.Now().Unix())//日志记录
			save_mgo.Save(save_coll_name, map[string]interface{}{
				"name":"下载器",
				"comeintime":comeintime,
				"date":qu.FormatDateByInt64(&comeintime, qu.DATEFORMAT),
				"detail" : isProMailContent,
			})

			sendErrVpsMailApi("下载器异常",isProMailContent)
		}
	}

	updatelock.Unlock()

}
//抽取活跃监测
func extractRunning()  {
	log.Println("...抽取活跃处理一次...")
	isSend := false
	for _,v := range extTmp{
		t := qu.Int64All(v)
		now_time := time.Now().Unix()
		if now_time-t > int64(1800) {
			isSend = true//有异常发邮件
			break
		}
	}
	if isSend {
		updatelock.Lock()
		extTmp = make(map[string]interface{},0)
		for _,v := range extArr{
			extid := qu.ObjToString(v)
			extTmp[extid] = time.Now().Unix()

		}
		//发送邮件
		sendErrExtMailApi("抽取程序异常","抽取程序超半小时未响应...清检查")
		updatelock.Unlock()
	}
}