adm-ods/main.go

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

package main

import (
	"adm-ods/common.in/cache"
	"adm-ods/common.in/mq"
	"adm-ods/impl"
	"adm-ods/impl/analysis"
	"adm-ods/impl/handle"
	"adm-ods/model"
	"adm-ods/pb"
	dutils "adm-ods/utils"
	"context"
	"flag"
	"fmt"
	"log"
	"os"
	"os/signal"
	"strings"
	"syscall"
	"time"
	//"adm-ods/common.in/cache"
	"adm-ods/common.in/clinit"
	"adm-ods/common.in/config"
	"adm-ods/common.in/logger"
	"adm-ods/common.in/utils"

	_ "github.com/go-sql-driver/mysql"
	"gopkg.in/ini.v1"
)

var (
	// 这里可以改默认值
	appConfigFile = flag.String("appconfig", "/etc/adm-ods/app.conf", "app config file location")
	configFile    = flag.String("config", "/etc/adm/common.json", "config file location")
	version       = flag.Bool("version", false, "config file location")
	GitCommit     = "library-import"
	Version       = "library-import"
)

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

func prepare(filename string, etcdAddrs []string, discoveryType string) {
	var conf *config.Configure

	if discoveryType == "k8s" {
		config.SetConfigFile(filename)
		conf = config.GetConfigForK8s()
		if conf == nil {
			fmt.Printf("get conf failed\n\n")
			os.Exit(1)
		}
		//clinit.EctdHandler(conf)
	} else {
		config.SetConfigFile(filename)
		conf = config.GetConfigForK8s()
		if conf == nil {
			fmt.Printf("get conf failed\n\n")
			os.Exit(1)
		}
		clinit.EctdHandler(etcdAddrs)

		// 先行于读配置
		/*clinit.InitEtcd(etcdAddrs)
		conf = config.GetConfig(projectName+"/"+runmode, key, clinit.GetEtcdClient())
		if conf == nil {
			fmt.Printf("get conf failed\n\n")
			os.Exit(1)
		}*/
		//rpc_apis.Init(etcdAddrs, conf)
	}

	// 指定mysql数据库，若无则使用默认数据库
	mysqldb := conf.Rpc.AdmOds.MysqlDb
	if mysqldb == "" {
		mysqldb = conf.Mysql.Db
	}
	// 连接数据库服务器
	clinit.InitMysqlGorm(
		conf.Mysql.User,
		conf.Mysql.Password,
		conf.Mysql.Addr,
		mysqldb,
		conf.Mysql.Charset,
		conf.Mysql.MaxIdle,
		conf.Mysql.MaxConn,
		conf.RunMode != "prod",
	)
	fmt.Println("mysql init finish")
	// 指定redis数据库，若无则使用默认数据库
	redisdb := conf.Rpc.AdmOds.RedisDb
	if redisdb == "" {
		redisdb = conf.Redis.Db
	}
	// 连接redis服务器
	fmt.Println("redis init start")
	cache.InitRedis(&cache.RedisConfig{
		Addrs:        strings.Split(conf.Redis.Addrs, ","),
		Password:     conf.Redis.Password,
		DB:           redisdb,
		PoolSize:     conf.Redis.PoolSize,
		MinIdleConns: conf.Redis.MinIdleConns,
		MaxRetries:   conf.Redis.MaxRetries,
		IsCluster:    conf.Redis.IsCluster,
	})

	fmt.Println("redis init finish")

	// 建立rpc客户端
	conns := pb.SetupClients()
	for _, conn := range conns {
		defer conn.Close()
	}
	fmt.Println("setup client finish")
	// 初始化logger
	ms, _ := conf.Log.MaxSize.Int64()
	mb, _ := conf.Log.MaxBackups.Int64()
	ma, _ := conf.Log.MaxAge.Int64()
	maxSize := int(ms)
	maxBackups := int(mb)
	maxAge := int(ma)

	disableStacktrace := (conf.Log.Stacktrace == "true")
	// 通用logger
	commonLogger := logger.InitLogger(conf.RunMode, fmt.Sprintf("%s/%s.log", conf.Log.Path, conf.Rpc.AdmOds.ServiceName), conf.Rpc.AdmOds.ServiceName, conf.Log.Level,
		maxSize, maxBackups, maxAge, disableStacktrace)
	// 单独设置
	accessLogger := logger.NewInfoLogger(conf.RunMode, fmt.Sprintf("%s/%s-access.log", conf.Log.Path, conf.Rpc.AdmOds.ServiceName), conf.Rpc.AdmOds.ServiceName, conf.Log.Level,
		maxSize, maxBackups, maxAge)

	// 设置需要使用logger的地方
	dutils.SetLogger(commonLogger)
	analysis.SetLogger(commonLogger)
	handle.SetLogger(commonLogger)
	handle.SetAccessLogger(accessLogger)
	model.SetLogger(commonLogger)
	// access日志

	fmt.Println("init rabbitmq")
	// 初始化rabbitmq

	consumerCount, _ := conf.OdsRabbitmq.ConsumerCount.Int64()
	fmt.Println("consumerCount:", consumerCount)
	if consumerCount == 0 {
		consumerCount = 2
	}

	odsMq := mq.InitRabbitmq(
		conf.OdsRabbitmq.Addr,
		conf.OdsRabbitmq.Username,
		conf.OdsRabbitmq.Passwrod,
		conf.OdsRabbitmq.Vhost,
		conf.OdsRabbitmq.ExchangeName,
		conf.OdsRabbitmq.QueueName,
		conf.OdsRabbitmq.RouteBindKey,
		impl.HandleOdsData,
		false,
		int(consumerCount),
	)
	mq.SetOdsMq(odsMq)

	dwsMq := mq.InitRabbitmq(
		conf.DwsRabbitmq.Addr,
		conf.DwsRabbitmq.Username,
		conf.DwsRabbitmq.Passwrod,
		conf.DwsRabbitmq.Vhost,
		conf.DwsRabbitmq.ExchangeName,
		conf.DwsRabbitmq.QueueName,
		conf.DwsRabbitmq.RouteBindKey,
		nil,
		true,
		0,
	)
	mq.SetDwsMq(dwsMq)

	odsMq.StartConsumer()
}

func start() {
	// 优雅关闭服务器
	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)
	// 不管什么行为，都等待5秒退出
	log.Println("Start to shutdown server...")
	_, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	log.Println("Shutdown server finished.")
}

func main() {
	flag.Parse()
	if *version {
		showVersion()
	}
	cfg, err := ini.Load(*appConfigFile)
	if err != nil {
		fmt.Printf("Fail to read file: %v\n\n", err)
		os.Exit(1)
	}

	etcdAddrs := strings.Split(cfg.Section("").Key("etcd_addrs").String(), ",")
	discoveryType := cfg.Section("").Key("discovery_type").String()
	//utils.SetRunmode(runmode)
	prepare(*configFile, etcdAddrs, discoveryType)
	go utils.Free()
	start()
	return
}