adm-task/main.go

// Copyright 2019 getensh.com. All rights reserved.
// Use of this source code is governed by getensh.com.

package main

import (
	"adm-task/impl/v1/check"
	"adm-task/utils"
	"context"
	"flag"
	"fmt"
	"log"
	"net"
	"os"
	"os/signal"
	"strings"
	"syscall"
	"time"

	"gopkg.in/ini.v1"

	"adm-task/impl"
	"adm-task/parser"

	"git.getensh.com/common/gopkgsv2/database"
	"git.getensh.com/common/gopkgsv2/tasker"

	"google.golang.org/grpc"
)

var (
	appConfigFile = flag.String("appconfig", "/etc/adm-task/app.conf", "app config file location")
	configFile    = flag.String("config", "/etc/adm/common.json", "config file location")
	version       = flag.Bool("version", false, "print the version")
	GitCommit     = "library-import"
	Version       = "library-import"
)

func showVersion() {
	fmt.Println("Version:  ", Version)
	fmt.Println("GitCommit:", GitCommit)
}

func prepare(filename string, discoveryType string) {
	// 加载配置

	if discoveryType == "k8s" {
		err := parser.LoadConfig(filename)
		if err != nil {
			fmt.Printf("get conf failed, err: %+v\n\n", err)
			os.Exit(1)
		}
	} else {
		// TODO 从etcd读取配置
		err := parser.LoadConfig(filename)
		if err != nil {
			fmt.Printf("get conf failed, err: %+v\n\n", err)
			os.Exit(1)
		}
	}

	/*err = parser.LoadConfig("conf/app.yaml")
	if err != nil {
		fmt.Printf("get conf failed, err: %+v\n\n", err)
		os.Exit(1)
	}*/
	check.SetRunmode(parser.Conf.RunMode)
	// 注册处理函数
	// parser.Register(parser.MysqlHandler, parser.RedisHandler, parser.LoggerHandler)
	parser.Register(parser.LoggerHandler, parser.MysqlHandler)
	// 执行注册的处理函数
	parser.Handle()

	// 加载字段管理，启动定时任务
	utils.FieldLoad()
}

func run(etcdAddrs []string, serveAddr string) {
	// 开始监听
	parser.EctdHandler(etcdAddrs, serveAddr)
	//serveAddr := fmt.Sprintf("%s:%d", parser.Conf.Rpc.AdmTask.ServiceIp, parser.Conf.Rpc.AdmTask.ServicePort)
	log.Printf("Listening and serving TCP on %s\n", serveAddr)
	lis, err := net.Listen("tcp", serveAddr)
	if err != nil {
		os.Exit(1)
	}
	s := grpc.NewServer()
	impl.Register(s)

	go func() {
		if err := s.Serve(lis); err != nil {
			log.Fatalf("HTTP server listen failed. err: %s\n", err.Error())
		}
	}()

	// 优雅关闭服务器
	sigChan := make(chan os.Signal, 1)
	// 捕获信号
	signal.Notify(sigChan, syscall.SIGQUIT, syscall.SIGTERM, syscall.SIGINT, syscall.SIGKILL)
	sigValue := <-sigChan
	log.Printf("Got a signal:%v", sigValue)

	// 让tasker安全退出
	tasker.SignalNotify(sigValue)

	// 不管什么行为，都等待5秒退出
	log.Println("Start to shutdown server...")
	_, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	s.Stop()
	log.Println("Shutdown server finished.")
}

func runPeerToPeer() {
	// TODO 支持peer to peer 模式
}

func main() {
	defer func() {
		database.Close()
	}()

	flag.Parse()
	if *version {
		showVersion()
		os.Exit(1)
	}

	cfg, err := ini.Load(*appConfigFile)
	if err != nil {
		fmt.Printf("Fail to read app config file: %v\n\n", err)
		os.Exit(1)
	}
	etcdAddrs := strings.Split(cfg.Section("").Key("etcd_addrs").String(), ",")
	serveAddr := cfg.Section("").Key("serve_addr").String()
	discoveryType := cfg.Section("").Key("discovery_type").String()

	prepare(*configFile, discoveryType)

	// go utils.Free()
	if discoveryType == "k8s" {
		runPeerToPeer()
	} else {
		run(etcdAddrs, serveAddr)
	}

	return
}