feat: rpc server add the logic open-telemetry,add the comment for rpc client test

Author: coderi421

app/pkg/options/tracing.go

@@ -0,0 +1,42 @@
+package options
+
+import (
+	"github.com/CoderI421/gframework/pkg/errors"
+	"github.com/spf13/pflag"
+)
+
+type TelemetryOptions struct {
+	// Name is the name of the service.
+	Name string `json:"name"`
+	// 连接地址
+	Endpoint string `json:"endpoint"`
+	// SampleRate is the rate at which traces are sampled. 1.0 means all traces are sampled, 0.0 means no traces are sampled.
+	// 意思是采样率，1.0表示所有的都采样，0.0表示不采样
+	Sampler float64 `json:"sampler"`
+	// Batcher is the type of batcher to use for sending traces to the collector.
+	// Batcher是用于将跟踪发送到收集器的批处理程序类型。
+	Batcher string `json:"batcher"`
+}
+
+func NewTelemetryOptions() *TelemetryOptions {
+	return &TelemetryOptions{
+		Name:     "shop",
+		Endpoint: "http://127.0.0.1:14268/api/traces",
+		Sampler:  1.0,
+		Batcher:  "jaeger",
+	}
+}
+
+func (t *TelemetryOptions) Validate() (errs []error) {
+	if t.Batcher != "jaeger" && t.Batcher != "zipkin" {
+		errs = append(errs, errors.New("open-telemetry batcher only supports: jaeger and zipkin"))
+	}
+	return
+}
+
+func (t *TelemetryOptions) AddFlags(fs *pflag.FlagSet) {
+	fs.StringVar(&t.Name, "telemetry.name", t.Name, "open-telemetry name")
+	fs.StringVar(&t.Endpoint, "telemetry.endpoint", t.Endpoint, "open-telemetry endpoint")
+	fs.Float64Var(&t.Sampler, "telemetry.sampler", t.Sampler, "open-telemetry sampler")
+	fs.StringVar(&t.Batcher, "telemetry.batcher", t.Batcher, "open-telemetry batcher,only support jaeger and zipkin")
+}

app/user/client/client.go

@@ -50,7 +50,9 @@ func main() {
 	}
 	defer conn.Close()
 	uc := v1.NewUserClient(conn)
-	// 在终端中，使用 --server.port=*** --server.http-port 命令启动启动多个，这里就会轮询调用，测试负载均衡
+
+	// 测试 grpc 服务的负载均衡用例
+	// 在终端中，使用 --server.port=*** --server.http-port -c configs/user/srv.yaml 命令启动启动多个，这里就会轮询调用，测试负载均衡
 	for {
 		re, err := uc.GetUserList(context.Background(), &v1.PageInfo{})
 		if err != nil {

app/user/srv/config/config.go

@@ -11,9 +11,10 @@ var _ app.CliOptions = &Config{}
 
 func New() *Config {
 	return &Config{
-		Log:      log.NewOptions(),
-		Server:   options.NewServerOptions(),
-		Registry: options.NewRegistryOptions(),
+		Log:       log.NewOptions(),
+		Server:    options.NewServerOptions(),
+		Registry:  options.NewRegistryOptions(),
+		Telemetry: options.NewTelemetryOptions(),
 	}
 }
 
@@ -23,6 +24,8 @@ type Config struct {
 	Server *options.ServerOptions `json:"server" mapstructure:"server"`
 	// 注册中心
 	Registry *options.RegistryOptions `json:"registry" mapstructure:"registry"`
+	//	链路追踪
+	Telemetry *options.TelemetryOptions `json:"telemetry" mapstructure:"telemetry"`
 }
 
 // Flags implements app.CliOptions interface.Add flags to the specified FlagSet object.
@@ -31,6 +34,7 @@ func (c *Config) Flags() (fss cliflag.NamedFlagSets) {
 	c.Log.AddFlags(fss.FlagSet("logs"))
 	c.Server.AddFlags(fss.FlagSet("server"))
 	c.Registry.AddFlags(fss.FlagSet("registry"))
+	c.Telemetry.AddFlags(fss.FlagSet("telemetry"))
 	return fss
 }
 
@@ -40,5 +44,6 @@ func (c *Config) Validate() (errors []error) {
 	errors = append(errors, c.Log.Validate()...)
 	errors = append(errors, c.Server.Validate()...)
 	errors = append(errors, c.Registry.Validate()...)
+	errors = append(errors, c.Telemetry.Validate()...)
 	return
 }

app/user/srv/rpc.go

@@ -3,6 +3,8 @@ package srv
 import (
 	"fmt"
 
+	"github.com/CoderI421/gframework/gmicro/core/trace"
+
 	upbv1 "github.com/CoderI421/gframework/api/user/v1"
 	srv1 "github.com/CoderI421/gframework/app/user/srv/service/v1"
 
@@ -13,6 +15,15 @@ import (
 )
 
 func NewUserRPCServer(cfg *config.Config) (*rpcserver.Server, error) {
+	// 初始化 open-telemetry 的 exporter
+	// 这里会根据 endpoint 为单元注册 trace 服务的实例
+	trace.InitAgent(trace.Options{
+		Name:     cfg.Telemetry.Name,
+		Endpoint: cfg.Telemetry.Endpoint,
+		Sampler:  cfg.Telemetry.Sampler,
+		Batcher:  cfg.Telemetry.Batcher,
+	})
+
 	data := mock.NewUsers()
 	srv := srv1.NewUserService(data)
 	userver := user.NewUserServer(srv)

gmicro/core/trace/agent.go

@@ -22,8 +22,10 @@ const (
 
 var (
 	// struct{}空结构体 不占内存，zerobase
+	// 这个map的作用是，记录已经初始化过的endpoint，避免重复初始化
 	agents = make(map[string]struct{})
-	lock   sync.Mutex
+	// 保证map的并发安全
+	lock sync.Mutex
 )
 
 func InitAgent(o Options) {