feat: add logic of prometheus to rpc server and client interceptor, and http middleware

Author: coderi421

README.md

@@ -62,7 +62,7 @@ Microservice framework implemented based on Golang.
 │  │  └─app.go 这个 app 是 GRPC，服务名称，注册中心等的集合
 │  ├─code 有一些公用的错误码
 │  ├─core 底层共通核心的包
-│  │  ├─metric
+│  │  ├─metric 服务监控相关的逻辑 使用了 prometheus
 │  │  └─trace 链路追踪，采用 opentemlemetry 
 │  ├─registry
 │  │  └─consul 服务注册中心相关逻辑，参考 kratos
@@ -76,9 +76,9 @@ Microservice framework implemented based on Golang.
 │      └─rpcserver rpc 服务的初始化配置
 │          ├─client.go rpc 客户端的初始化配置
 │          ├─server.go rpc 服务端的初始化配置
-│          ├─clientinterceptors 客户端的拦截器：超时连接器
+│          ├─clientinterceptors 客户端的拦截器：超时连接器，监控 prometheus
 │          ├─resolver 服务发现相关的逻辑 解析器
-│          │  ├─direct 直连
+│          │  ├─direct 直连，加权轮询的时候使用
 │          │  └─discovery 服务发现，负载均衡
 │          │      ├─builder.go 服务发现的构建器
 │          │      └─resolver.go 服务发现的解析器,负载均衡的逻辑在这里实现 UpdateState 核心
@@ -89,7 +89,7 @@ Microservice framework implemented based on Golang.
 │          │  ├─p2c  负载均衡 [Power of Two Random Choices] 算法
 │          │  ├─random 负载均衡随机算法
 │          │  └─wrr 负载均衡 加权轮询 [Weighted Round Robin] 算法
-│          └─serverinterceptors 服务端的拦截器：超时，crash恢复
+│          └─serverinterceptors 服务端的拦截器：超时，crash恢复，监控
 ├─pkg
 │  ├─app 包括的项目的启动服务，配置文件的读取，命令行工具，以及其他选项
 │  │  ├─app.go 服务启动：命令行工具，日志，错误包，配置等

app/shop/admin/http.go

@@ -18,6 +18,8 @@ func NewUserHTTPServer(conf *config.Config) (*restserver.Server, error) {
 		restserver.WithPort(conf.Server.HttpPort),
 		restserver.WithEnableProfiling(true),
 		restserver.WithMiddlewares(conf.Server.Middlewares),
+		restserver.WithEnableProfiling(true),
+		restserver.WithMetrics(true),
 	)
 	//_ = tracerProvider()
 

app/user/srv/rpc.go

@@ -29,7 +29,10 @@ func NewUserRPCServer(cfg *config.Config) (*rpcserver.Server, error) {
 	userver := user.NewUserServer(srv)
 
 	rpcAddr := fmt.Sprintf("%s:%d", cfg.Server.Host, cfg.Server.Port)
-	urpcServer := rpcserver.NewServer(rpcserver.WithAddress(rpcAddr))
+	urpcServer := rpcserver.NewServer(
+		rpcserver.WithAddress(rpcAddr),
+		rpcserver.WithEnableMetric(cfg.Server.EnableMetrics),
+	)
 
 	// 注册 user 模块的 rpc 服务
 	upbv1.RegisterUserServer(urpcServer.Server, userver)

configs/user/srv.yaml

@@ -13,6 +13,16 @@ registry:
   address: 127.0.0.1:8500
   scheme: http
 
+server:
+  name: user-srv #服务名，注册到consul使用的名称
+  limit: true
+  host: "127.0.0.1" #host, 默认使用 0.0.0.0
+  port: 8078 # grpc 安全模式的端口号，默认 8081
+  http-port: 8088
+  healthz: true # 是否开启健康检查，如果开启会安装 /healthz 路由，默认 true
+  enable-metrics: true # 开启 metrics, router:  /metrics
+  profiling: true # 开启性能分析, 可以通过 <host>:<port>/debug/pprof/地址查看程序栈、线程等系统信息，默认值为 true
+
 telemetry:
   Name: user-srv
   Endpoint: http://127.0.0.1:14268/api/traces

gmicro/server/restserver/options.go

@@ -55,3 +55,9 @@ func WithCustomMiddlewares(middlewares []gin.HandlerFunc) ServerOption {
 		s.customMiddlewares = middlewares
 	}
 }
+
+func WithMetrics(metrics bool) ServerOption {
+	return func(s *Server) {
+		s.enableMetrics = metrics
+	}
+}

gmicro/server/restserver/server.go

@@ -11,9 +11,9 @@ import (
 	"github.com/CoderI421/gframework/gmicro/server/restserver/validation"
 	"github.com/CoderI421/gframework/pkg/errors"
 	"github.com/CoderI421/gframework/pkg/log"
-
 	"github.com/gin-gonic/gin"
 	ut "github.com/go-playground/universal-translator"
+	"github.com/penglongli/gin-metrics/ginmetrics"
 )
 
 type JwtInfo struct {
@@ -38,6 +38,8 @@ type Server struct {
 	healthz bool
 	//是否开启pprof接口，默认开启,如果开启会自动添加/debug/pprof接口
 	enableProfiling bool
+	//是否开启metrics接口，默认开启，如果开启会自动添加/metrics接口
+	enableMetrics bool
 	//中间件(拦截器)两种用法 1.提前写好，直接配置名称就可以，用起来方便，比rpc自定义的实现弱，2.自定义gin.HandlerFunc
 	middlewares       []string
 	customMiddlewares []gin.HandlerFunc
@@ -112,6 +114,20 @@ func (s *Server) Start(ctx context.Context) error {
 	if s.enableProfiling {
 		pprof.Register(s.Engine)
 	}
+	//根据配置初始化metrics路由
+	if s.enableMetrics {
+		// get global Monitor object
+		m := ginmetrics.GetMonitor()
+		// +optional set metric path, default /debug/metrics
+		m.SetMetricPath("/metrics")
+		// +optional set slow time, default 5s
+		m.SetSlowTime(10)
+		// +optional set request duration, default {0.1, 0.3, 1.2, 5, 10}
+		// used to p95, p99
+		m.SetDuration([]float64{0.1, 0.3, 1.2, 5, 10})
+		//反向注入
+		m.Use(s)
+	}
 
 	// 如果开启了健康检查接口，就添加/health接口
 	if s.healthz {

gmicro/server/rpcserver/clientinterceptors/prometheusinterceptor.go

@@ -0,0 +1,48 @@
+package clientinterceptors
+
+import (
+	"context"
+	"strconv"
+	"time"
+
+	"github.com/CoderI421/gframework/gmicro/core/metric"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/status"
+)
+
+const serverNamespace = "rpc-client"
+
+var (
+	metricServerReqDur = metric.NewHistogramVec(&metric.HistogramVecOpts{
+		VectorOpts: metric.VectorOpts{
+			Namespace: serverNamespace,
+			Subsystem: "requests",
+			Name:      "your_service_name_duration_microseconds",
+			Help:      "rpc client requests duration in microseconds",
+			Labels:    []string{"method"}},
+		Buckets: []float64{5, 10, 25, 50, 100, 250, 500, 1000},
+	})
+
+	metricServerReqCodeTotal = metric.NewCounterVec(&metric.CounterVecOpts{
+		Namespace: serverNamespace,
+		Subsystem: "requests",
+		Name:      "your_service_status_code_total",
+		Help:      "rpc client requests status code count",
+		Labels:    []string{"method", "code"},
+	})
+)
+
+// UnaryClientPrometheusInterceptor prometheus sclient interceptor.
+func UnaryClientPrometheusInterceptor() grpc.UnaryClientInterceptor {
+	return func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn,
+		invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
+		startTime := time.Now()
+		err := invoker(ctx, method, req, reply, cc, opts...)
+		//记录了耗时
+		metricServerReqDur.Observe(int64(time.Since(startTime)/time.Millisecond), method)
+
+		//记录了状态码
+		metricServerReqCodeTotal.Inc(method, strconv.Itoa(int(status.Code(err))))
+		return err
+	}
+}

gmicro/server/rpcserver/server.go

@@ -44,6 +44,9 @@ type Server struct {
 	metadata *apimd.Server
 	//设置监听的ip port
 	endpoint *url.URL
+
+	//是否开启 metric 监测
+	enableMetric bool
 }
 
 func (s *Server) Address() string {
@@ -64,7 +67,9 @@ func NewServer(opts ...ServerOption) *Server {
 	unaryInts := []grpc.UnaryServerInterceptor{
 		srvints.UnaryCrashInterceptor,
 		otelgrpc.UnaryServerInterceptor(), // 设置链路追踪的拦截器
-		//srvints.UnaryTimeoutInterceptor(srv.timeout),
+	}
+	if srv.enableMetric {
+		unaryInts = append(unaryInts, srvints.UnaryServerPrometheusInterceptor)
 	}
 	//timeout可以交给用户设置，不设置就不用此拦截器
 	if srv.timeout > 0 {
@@ -99,34 +104,25 @@ func NewServer(opts ...ServerOption) *Server {
 }
 
 func WithAddress(address string) ServerOption {
-	return func(s *Server) {
-		s.address = address
-	}
+	return func(s *Server) { s.address = address }
 }
 func WithTimeout(timeout time.Duration) ServerOption {
-	return func(s *Server) {
-		s.timeout = timeout
-	}
+	return func(s *Server) { s.timeout = timeout }
 }
 func WithLis(lis net.Listener) ServerOption {
-	return func(s *Server) {
-		s.lis = lis
-	}
+	return func(s *Server) { s.lis = lis }
 }
 func WithUnaryInterceptor(in ...grpc.UnaryServerInterceptor) ServerOption {
-	return func(s *Server) {
-		s.unaryInts = in
-	}
+	return func(s *Server) { s.unaryInts = in }
 }
 func WithStreamInterceptor(in ...grpc.StreamServerInterceptor) ServerOption {
-	return func(s *Server) {
-		s.streamInts = in
-	}
+	return func(s *Server) { s.streamInts = in }
 }
 func WithOptions(opts ...grpc.ServerOption) ServerOption {
-	return func(s *Server) {
-		s.grpcOpts = opts
-	}
+	return func(s *Server) { s.grpcOpts = opts }
+}
+func WithEnableMetric(enable bool) ServerOption {
+	return func(s *Server) { s.enableMetric = enable }
 }
 
 // 完成ip和端口的提取

gmicro/server/rpcserver/serverinterceptors/prometheusinterceptors.go

@@ -1 +1,52 @@
 package serverinterceptors
+
+import (
+	"context"
+	"strconv"
+	"time"
+
+	"github.com/CoderI421/gframework/gmicro/core/metric"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/status"
+)
+
+const serverNamespace = "rpc_server"
+
+var (
+	metricServerReqDur = metric.NewHistogramVec(&metric.HistogramVecOpts{
+		VectorOpts: metric.VectorOpts{
+			// 服务的名称
+			Namespace: serverNamespace,
+			Subsystem: "requests",
+			Name:      "your_service_name_duration_microseconds",
+			Help:      "rpc server requests duration in microseconds",
+			Labels:    []string{"method"},
+		},
+		Buckets: []float64{5, 10, 25, 50, 100, 250, 500, 1000},
+	})
+
+	metricServerReqCodeTotal = metric.NewCounterVec(&metric.CounterVecOpts{
+		Namespace: serverNamespace,
+		Subsystem: "requests",
+		Name:      "your_service_status_code_total",
+		Help:      "rpc server requests status code count",
+		Labels:    []string{"method", "code"},
+	})
+)
+
+// UnaryPrometheusInterceptor prometheus server interceptor.
+func UnaryServerPrometheusInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo,
+	handler grpc.UnaryHandler) (resp interface{}, err error) {
+
+	//进入执行逻辑前
+	startTime := time.Now()
+	// 执行请求的逻辑
+	resp, err = handler(ctx, req)
+
+	metricServerReqDur.Observe(int64(time.Since(startTime)/time.Millisecond), info.FullMethod)
+
+	//记录了状态码记录那个方法，获取了那个状态码
+	metricServerReqCodeTotal.Inc(info.FullMethod, strconv.Itoa(int(status.Code(err))))
+
+	return
+}

go.mod

@@ -45,8 +45,11 @@ require (
 require (
 	github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 // indirect
 	github.com/armon/go-metrics v0.4.0 // indirect
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/bits-and-blooms/bitset v1.2.0 // indirect
 	github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 // indirect
 	github.com/bytedance/sonic v1.8.0 // indirect
+	github.com/cespare/xxhash/v2 v2.2.0 // indirect
 	github.com/chenzhuoyu/base64x v0.0.0-20221115062448-fe3a3abad311 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/fsnotify/fsnotify v1.6.0 // indirect
@@ -74,13 +77,19 @@ require (
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.17 // indirect
 	github.com/mattn/go-runewidth v0.0.14 // indirect
+	github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
 	github.com/mitchellh/go-homedir v1.1.0 // indirect
 	github.com/mitchellh/mapstructure v1.5.0 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/openzipkin/zipkin-go v0.4.1 // indirect
 	github.com/pelletier/go-toml/v2 v2.0.6 // indirect
+	github.com/penglongli/gin-metrics v0.1.10 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/prometheus/client_golang v1.12.0 // indirect
+	github.com/prometheus/client_model v0.2.0 // indirect
+	github.com/prometheus/common v0.32.1 // indirect
+	github.com/prometheus/procfs v0.7.3 // indirect
 	github.com/rivo/uniseg v0.2.0 // indirect
 	github.com/spf13/afero v1.9.3 // indirect
 	github.com/spf13/cast v1.5.0 // indirect