gocommon/pkg/trace/skywalking.go

package trace

import (
	"errors"
	"fmt"
	"path/filepath"
	"runtime"
	"sync/atomic"
	"time"

	"git.shuncheng.lu/bigthing/gocommon/pkg/conf"
	"git.shuncheng.lu/bigthing/gocommon/pkg/internal/util"
	"github.com/SkyAPM/go2sky"
	"github.com/SkyAPM/go2sky/reporter"
)

var (
	//skyWalkingTracer *go2sky.Tracer
	// 防止并发 race问题
	skyWalkingTracer atomic.Value
)

const (
	MysqlComponentID      int32 = 5
	HttpClientComponentID int32 = 2
	HttpServerComponentID int32 = 49
)

/**
main方法手动调用
*/
func Init() error {
	healthInit()
	return nil
}

func loadTracer() {
	// 防止不必要的错误
	defer func() {
		if err := recover(); err != nil {
			fmt.Printf("\033[31m[ERROR]\033[0m init skywaling panic, err=%v\n", err)
		}
	}()
	walkingTracer, err := newSkyWalkingTracer()
	if err != nil {
		fmt.Printf("\033[31m[ERROR]\033[0m init skywaling err, err=%v\n", err.Error())
		return
	}
	fmt.Printf("\033[32m[INFO]\033[0m init skywaling success\n")
	skyWalkingTracer.Store(walkingTracer)
}

func GetSkyWalkingTracer() *go2sky.Tracer {
	tracer, _ := skyWalkingTracer.Load().(*go2sky.Tracer)
	return tracer
}

func getConfig(section, key string, defaultValue ...string) string {
	return conf.GetString(section, key, defaultValue...)
}

/**
创建 tracer
*/
func newSkyWalkingTracer() (*go2sky.Tracer, error) {
	serverAddr := getConfig("trace", "sky_walking_host", "")
	if serverAddr == "" {
		return nil, errors.New("trace.sky_walking_host config not found")
	}
	serverName := getConfig("trace", "application_name", "")
	if serverName == "" {
		return nil, errors.New("trace.application_name config not found")
	}
	/**
	这里就算连接异常，也不会抛出error，因为grcp默认是后台不断重新连接
	*/
	r, err := reporter.NewGRPCReporter(serverAddr)
	if err != nil {
		return nil, err
	}
	currentServerHostName := util.CurrentServerHostName()
	tracer, err := go2sky.NewTracer(serverName, go2sky.WithReporter(r), go2sky.WithInstance(currentServerHostName))
	if err != nil {
		defer func() {
			r.Close() // 防止泄漏连接
		}()
		return nil, err
	}
	fmt.Printf("\033[32m[INFO]\033[0m init skywalking, skywalking_host=%s, application_name=%s\n", serverAddr, serverName)
	return tracer, nil
}

/**
获取 调用 的 方法.line
*/
func GetCallLine(num int) string {
	pc, file, line, ok := runtime.Caller(num)
	if !ok {
		return ""
	}
	pcName := runtime.FuncForPC(pc).Name()
	file, _ = filepath.Abs(file)
	return fmt.Sprintf("[%s.%d] ", pcName, line)
}

/**
健康启动
1、启动成功继续
2、启动失败，定期1分钟重新连接
*/
func healthInit() {
	// 初始化一次
	loadTracer()
	// 如果为空，则定期去连接
	if GetSkyWalkingTracer() == nil {
		go func() {
			retryTime := func() time.Duration {
				return time.Second * 60
			}
			retry := time.NewTimer(retryTime())
			for {
				<-retry.C
				loadTracer() // 初始化
				if GetSkyWalkingTracer() != nil {
					retry.Stop()
					break
				}
				retry.Reset(retryTime())
			}
		}()
	}
}