Compare commits

...

4 Commits
v3.10.18 ... v3

Author SHA1 Message Date
f6ec5ae624 update for latest micro logger changes
Some checks failed
build / test (push) Failing after 4m56s
build / lint (push) Successful in 9m32s
codeql / analyze (go) (push) Failing after 1m43s
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2024-10-12 13:18:36 +03:00
295f26dd2c #348 add check method in should be skipped (#175)
Some checks failed
build / test (push) Failing after 7s
build / lint (push) Failing after 7s
codeql / analyze (go) (push) Failing after 13m47s
Co-authored-by: Gorbunov Kirill Andreevich <kgorbunov@mtsbank.ru>
Reviewed-on: #175
Reviewed-by: Василий Толстов <v.tolstov@unistack.org>
Co-authored-by: Кирилл Горбунов <kirya_gorbunov_2015@mail.ru>
Co-committed-by: Кирилл Горбунов <kirya_gorbunov_2015@mail.ru>
2024-09-20 17:43:08 +03:00
7b5ce8c49a update to latest micro
Some checks failed
build / test (push) Failing after 9s
build / lint (push) Failing after 9s
codeql / analyze (go) (push) Failing after 10s
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2024-09-17 12:52:41 +03:00
3fc05ae291 fix metadata extract and trace span creating
Some checks failed
build / lint (push) Successful in 27s
build / test (push) Failing after 33s
codeql / analyze (go) (push) Failing after 10m56s
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2024-07-18 11:09:05 +03:00
6 changed files with 79 additions and 102 deletions

View File

@ -1,8 +1,6 @@
package grpc
import (
"io"
"go.unistack.org/micro/v3/codec"
"google.golang.org/grpc/encoding"
)
@ -49,29 +47,3 @@ func (w *wrapGrpcCodec) Unmarshal(d []byte, v interface{}, opts ...codec.Option)
}
return w.Codec.Unmarshal(d, v)
}
func (w *wrapGrpcCodec) ReadHeader(conn io.Reader, m *codec.Message, mt codec.MessageType) error {
return nil
}
func (w *wrapGrpcCodec) ReadBody(conn io.Reader, v interface{}) error {
if m, ok := v.(*codec.Frame); ok {
_, err := conn.Read(m.Data)
return err
}
return codec.ErrInvalidMessage
}
func (w *wrapGrpcCodec) Write(conn io.Writer, m *codec.Message, v interface{}) error {
// if we don't have a body
if v != nil {
b, err := w.Marshal(v)
if err != nil {
return err
}
m.Body = b
}
// write the body using the framing codec
_, err := conn.Write(m.Body)
return err
}

19
go.mod
View File

@ -1,17 +1,20 @@
module go.unistack.org/micro-server-grpc/v3
go 1.20
go 1.22
toolchain go1.23.1
require (
github.com/golang/protobuf v1.5.4
go.unistack.org/micro/v3 v3.10.66
golang.org/x/net v0.24.0
google.golang.org/grpc v1.63.2
google.golang.org/protobuf v1.33.0
go.unistack.org/micro/v3 v3.10.97
golang.org/x/net v0.30.0
google.golang.org/grpc v1.67.1
google.golang.org/protobuf v1.35.1
)
require (
golang.org/x/sys v0.19.0 // indirect
golang.org/x/text v0.14.0 // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20240415180920-8c6c420018be // indirect
go.unistack.org/micro-proto/v3 v3.4.1 // indirect
golang.org/x/sys v0.26.0 // indirect
golang.org/x/text v0.19.0 // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20241007155032-5fefd90f89a9 // indirect
)

31
go.sum
View File

@ -1,17 +1,20 @@
github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
go.unistack.org/micro/v3 v3.10.66 h1:tiG8HnyTC71IZWSC2qT/DmLhJinZJL9qvw+4Fvpm3d4=
go.unistack.org/micro/v3 v3.10.66/go.mod h1:erMgt3Bl7vQQ0e9UpQyR5NlLiZ9pKeEJ9+1tfYFaqUg=
golang.org/x/net v0.24.0 h1:1PcaxkF854Fu3+lvBIx5SYn9wRlBzzcnHZSiaFFAb0w=
golang.org/x/net v0.24.0/go.mod h1:2Q7sJY5mzlzWjKtYUEXSlBWCdyaioyXzRB2RtU8KVE8=
golang.org/x/sys v0.19.0 h1:q5f1RH2jigJ1MoAWp2KTp3gm5zAGFUTarQZ5U386+4o=
golang.org/x/sys v0.19.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=
golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
google.golang.org/genproto/googleapis/rpc v0.0.0-20240415180920-8c6c420018be h1:LG9vZxsWGOmUKieR8wPAUR3u3MpnYFQZROPIMaXh7/A=
google.golang.org/genproto/googleapis/rpc v0.0.0-20240415180920-8c6c420018be/go.mod h1:WtryC6hu0hhx87FDGxWCDptyssuo68sk10vYjF+T9fY=
google.golang.org/grpc v1.63.2 h1:MUeiw1B2maTVZthpU5xvASfTh3LDbxHd6IJ6QQVU+xM=
google.golang.org/grpc v1.63.2/go.mod h1:WAX/8DgncnokcFUldAxq7GeB5DXHDbMF+lLvDomNkRA=
google.golang.org/protobuf v1.33.0 h1:uNO2rsAINq/JlFpSdYEKIZ0uKD/R9cpdv0T+yoGwGmI=
google.golang.org/protobuf v1.33.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
go.unistack.org/micro-proto/v3 v3.4.1 h1:UTjLSRz2YZuaHk9iSlVqqsA50JQNAEK2ZFboGqtEa9Q=
go.unistack.org/micro-proto/v3 v3.4.1/go.mod h1:okx/cnOhzuCX0ggl/vToatbCupi0O44diiiLLsZ93Zo=
go.unistack.org/micro/v3 v3.10.97 h1:8l7fv+i06/PjPrBBhRC/ZQkWGIOuHPg3jJN0vktYE78=
go.unistack.org/micro/v3 v3.10.97/go.mod h1:YzMldzHN9Ei+zy5t/Psu7RUWDZwUfrNYiStSQtTz90g=
golang.org/x/net v0.30.0 h1:AcW1SDZMkb8IpzCdQUaIq2sP4sZ4zw+55h6ynffypl4=
golang.org/x/net v0.30.0/go.mod h1:2wGyMJ5iFasEhkwi13ChkO/t1ECNC4X4eBKkVFyYFlU=
golang.org/x/sys v0.26.0 h1:KHjCJyddX0LoSTb3J+vWpupP9p0oznkqVk/IfjymZbo=
golang.org/x/sys v0.26.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/text v0.19.0 h1:kTxAhCbGbxhK0IwgSKiMO5awPoDQ0RpfiVYBfK860YM=
golang.org/x/text v0.19.0/go.mod h1:BuEKDfySbSR4drPmRPG/7iBdf8hvFMuRexcpahXilzY=
google.golang.org/genproto/googleapis/rpc v0.0.0-20241007155032-5fefd90f89a9 h1:QCqS/PdaHTSWGvupk2F/ehwHtGc0/GYkT+3GAcR1CCc=
google.golang.org/genproto/googleapis/rpc v0.0.0-20241007155032-5fefd90f89a9/go.mod h1:GX3210XPVPUjJbTUbvwI8f2IpZDMZuPJWDzDuebbviI=
google.golang.org/grpc v1.67.1 h1:zWnc1Vrcno+lHZCOofnIMvycFcc0QRGIzm9dhnDX68E=
google.golang.org/grpc v1.67.1/go.mod h1:1gLDyUQU7CTLJI90u3nXZ9ekeghjeM7pTDZlqFNg2AA=
google.golang.org/protobuf v1.35.1 h1:m3LfL6/Ca+fqnjnlqQXNpFPABW1UD7mjh8KO2mKFytA=
google.golang.org/protobuf v1.35.1/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=

68
grpc.go
View File

@ -7,6 +7,7 @@ import (
"fmt"
"net"
"reflect"
"slices"
"sort"
"strconv"
"strings"
@ -177,12 +178,9 @@ func (g *Server) configure(opts ...server.Option) error {
}
func (g *Server) getMaxMsgSize() int {
if g.opts.Context == nil {
return codec.DefaultMaxMsgSize
}
s, ok := g.opts.Context.Value(maxMsgSizeKey{}).(int)
if !ok {
return codec.DefaultMaxMsgSize
return 4 * 1024 * 1024
}
return s
}
@ -210,45 +208,22 @@ func (g *Server) handler(srv interface{}, stream grpc.ServerStream) error {
return status.Errorf(codes.Internal, "method does not exist in context")
}
ts := time.Now()
var sp tracer.Span
if !slices.Contains(tracer.DefaultSkipEndpoints, fullMethod) {
ctx, sp = g.opts.Tracer.Start(ctx, fullMethod+" rpc-server",
tracer.WithSpanKind(tracer.SpanKindServer),
tracer.WithSpanLabels(
"endpoint", fullMethod,
),
)
stream = &streamWrapper{ctx, stream}
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 {
if st != nil || st.Code() != codes.OK {
sp.SetStatus(tracer.SpanStatusError, err.Error())
g.opts.Meter.Counter(semconv.ServerRequestTotal, "endpoint", fullMethod, "status", "failure", "code", strconv.Itoa(int(st.Code()))).Inc()
}
sp.Finish()
}()
var serviceName, methodName string
serviceName, methodName, err = serviceMethod(fullMethod)
if err != nil {
err = status.New(codes.InvalidArgument, err.Error()).Err()
return err
}
if g.opts.Wait != nil {
g.opts.Wait.Add(1)
defer g.opts.Wait.Done()
}
// get grpc metadata
@ -257,6 +232,13 @@ func (g *Server) handler(srv interface{}, stream grpc.ServerStream) error {
gmd = gmetadata.MD{}
}
var serviceName, methodName string
serviceName, methodName, err = serviceMethod(fullMethod)
if err != nil {
err = status.New(codes.InvalidArgument, err.Error()).Err()
return err
}
md := metadata.New(len(gmd))
for k, v := range gmd {
md.Set(k, strings.Join(v, ", "))
@ -304,6 +286,30 @@ func (g *Server) handler(srv interface{}, stream grpc.ServerStream) error {
// create new context
ctx = metadata.NewIncomingContext(ctx, md)
stream = &streamWrapper{ctx, stream}
if !slices.Contains(meter.DefaultSkipEndpoints, fullMethod) {
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()
}
}()
}
if g.opts.Wait != nil {
g.opts.Wait.Add(1)
defer g.opts.Wait.Done()
}
// get peer from context
if p, ok := peer.FromContext(ctx); ok {
md.Set("Remote", p.Addr.String())
@ -879,7 +885,7 @@ func (g *Server) Start() error {
// deregister self
if err = g.Deregister(); err != nil {
if config.Logger.V(logger.ErrorLevel) {
config.Logger.Errorf(config.Context, "Server deregister error: %v", err)
config.Logger.Error(config.Context, "Server deregister error", err)
}
}

View File

@ -47,11 +47,7 @@ func (r *rpcRequest) Header() metadata.Metadata {
}
func (r *rpcRequest) Read() ([]byte, error) {
f := &codec.Frame{}
if err := r.codec.ReadBody(r.rw, f); err != nil {
return nil, err
}
return f.Data, nil
return nil, nil
}
func (r *rpcRequest) Stream() bool {

View File

@ -27,8 +27,5 @@ func (r *rpcResponse) WriteHeader(hdr metadata.Metadata) {
}
func (r *rpcResponse) Write(b []byte) error {
return r.codec.Write(r.rw, &codec.Message{
Header: r.header,
Body: b,
}, nil)
return nil
}