unistack-org/micro-server-grpc
4
0
Fork 2
Code Issues 3 Pull Requests Actions Packages Projects Releases 50 Wiki Activity

use meter directly
Some checks failed
codeql / analyze (go) (push) Failing after 1m55s
Details
build / test (push) Failing after 2m8s
Details
build / lint (push) Successful in 9m19s
Details

Browse Source 
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
This commit is contained in:
Василий Толстов 2024-04-06 23:10:51 +03:00
parent 3941a76333
commit 241146cb6f
1 changed files with 22 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
grpc.go
View File

@ -21,7 +21,9 @@ import (
"go.unistack.org/micro/v3/logger" "go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/metadata" "go.unistack.org/micro/v3/metadata"
"go.unistack.org/micro/v3/register" "go.unistack.org/micro/v3/register"
"go.unistack.org/micro/v3/semconv"
"go.unistack.org/micro/v3/server" "go.unistack.org/micro/v3/server"
msync "go.unistack.org/micro/v3/sync"
"golang.org/x/net/netutil" "golang.org/x/net/netutil"
"google.golang.org/grpc" "google.golang.org/grpc"
"google.golang.org/grpc/codes" "google.golang.org/grpc/codes"
@ -47,7 +49,7 @@ type Server struct {
handlers map[string]server.Handler handlers map[string]server.Handler
srv *grpc.Server srv *grpc.Server
exit chan chan error exit chan chan error
wg *sync.WaitGroup wg *msync.WaitGroup
rsvc *register.Service rsvc *register.Service
subscribers map[*subscriber][]broker.Subscriber subscribers map[*subscriber][]broker.Subscriber
rpc *rServer rpc *rServer
@ -192,12 +194,30 @@ func (g *Server) getGrpcOptions() []grpc.ServerOption {
return opts return opts
} }
func (g *Server) handler(srv interface{}, stream grpc.ServerStream) (err error) { func (g *Server) handler(srv interface{}, stream grpc.ServerStream) error {
var err error
fullMethod, ok := grpc.MethodFromServerStream(stream) fullMethod, ok := grpc.MethodFromServerStream(stream)
if !ok { if !ok {
return status.Errorf(codes.Internal, "method does not exist in context") return status.Errorf(codes.Internal, "method does not exist in context")
} }
ts := time.Now()
g.opts.Meter.Counter(semconv.ServerRequestInflight, "endpoint", fullMethod).Inc()
defer func() {
te := time.Since(ts)
g.opts.Meter.Summary(semconv.ServerRequestLatencyMicroseconds, "endpoint", fullMethod).Update(te.Seconds())
g.opts.Meter.Histogram(semconv.ServerRequestDurationSeconds, "endpoint", fullMethod).Update(te.Seconds())
g.opts.Meter.Counter(semconv.ServerRequestInflight, "endpoint", fullMethod).Dec()
st := status.Convert(err)
if st == nil || st.Code() == codes.OK {
g.opts.Meter.Counter(semconv.ServerRequestTotal, "endpoint", fullMethod, "status", "success", "code", strconv.Itoa(int(codes.OK))).Inc()
} else {
g.opts.Meter.Counter(semconv.ServerRequestTotal, "endpoint", fullMethod, "status", "failure", "code", strconv.Itoa(int(st.Code()))).Inc()
}
}()
serviceName, methodName, err := serviceMethod(fullMethod) serviceName, methodName, err := serviceMethod(fullMethod)
if err != nil { if err != nil {
return status.New(codes.InvalidArgument, err.Error()).Err() return status.New(codes.InvalidArgument, err.Error()).Err()