update micro, add Health/Live/Ready checks
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
This commit is contained in:
parent
28234bcc3c
commit
91c71a3932
19
go.mod
19
go.mod
@ -1,20 +1,19 @@
|
|||||||
module go.unistack.org/micro-server-grpc/v3
|
module go.unistack.org/micro-server-grpc/v3
|
||||||
|
|
||||||
go 1.22
|
go 1.22.7
|
||||||
|
|
||||||
toolchain go1.23.1
|
toolchain go1.23.3
|
||||||
|
|
||||||
require (
|
require (
|
||||||
github.com/golang/protobuf v1.5.4
|
go.unistack.org/micro/v3 v3.10.109
|
||||||
go.unistack.org/micro/v3 v3.10.97
|
golang.org/x/net v0.31.0
|
||||||
golang.org/x/net v0.30.0
|
google.golang.org/grpc v1.68.0
|
||||||
google.golang.org/grpc v1.67.1
|
google.golang.org/protobuf v1.35.2
|
||||||
google.golang.org/protobuf v1.35.1
|
|
||||||
)
|
)
|
||||||
|
|
||||||
require (
|
require (
|
||||||
go.unistack.org/micro-proto/v3 v3.4.1 // indirect
|
go.unistack.org/micro-proto/v3 v3.4.1 // indirect
|
||||||
golang.org/x/sys v0.26.0 // indirect
|
golang.org/x/sys v0.27.0 // indirect
|
||||||
golang.org/x/text v0.19.0 // indirect
|
golang.org/x/text v0.20.0 // indirect
|
||||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20241007155032-5fefd90f89a9 // indirect
|
google.golang.org/genproto/googleapis/rpc v0.0.0-20241118233622-e639e219e697 // indirect
|
||||||
)
|
)
|
||||||
|
28
go.sum
28
go.sum
@ -4,17 +4,17 @@ github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
|
|||||||
github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
|
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 h1:UTjLSRz2YZuaHk9iSlVqqsA50JQNAEK2ZFboGqtEa9Q=
|
||||||
go.unistack.org/micro-proto/v3 v3.4.1/go.mod h1:okx/cnOhzuCX0ggl/vToatbCupi0O44diiiLLsZ93Zo=
|
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.109 h1:WVta6KrkPoFQcH/H+Mum4G35m7mmmK2Gs3yizrzXv+w=
|
||||||
go.unistack.org/micro/v3 v3.10.97/go.mod h1:YzMldzHN9Ei+zy5t/Psu7RUWDZwUfrNYiStSQtTz90g=
|
go.unistack.org/micro/v3 v3.10.109/go.mod h1:YzMldzHN9Ei+zy5t/Psu7RUWDZwUfrNYiStSQtTz90g=
|
||||||
golang.org/x/net v0.30.0 h1:AcW1SDZMkb8IpzCdQUaIq2sP4sZ4zw+55h6ynffypl4=
|
golang.org/x/net v0.31.0 h1:68CPQngjLL0r2AlUKiSxtQFKvzRVbnzLwMUn5SzcLHo=
|
||||||
golang.org/x/net v0.30.0/go.mod h1:2wGyMJ5iFasEhkwi13ChkO/t1ECNC4X4eBKkVFyYFlU=
|
golang.org/x/net v0.31.0/go.mod h1:P4fl1q7dY2hnZFxEk4pPSkDHF+QqjitcnDjUQyMM+pM=
|
||||||
golang.org/x/sys v0.26.0 h1:KHjCJyddX0LoSTb3J+vWpupP9p0oznkqVk/IfjymZbo=
|
golang.org/x/sys v0.27.0 h1:wBqf8DvsY9Y/2P8gAfPDEYNuS30J4lPHJxXSb/nJZ+s=
|
||||||
golang.org/x/sys v0.26.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
|
golang.org/x/sys v0.27.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
|
||||||
golang.org/x/text v0.19.0 h1:kTxAhCbGbxhK0IwgSKiMO5awPoDQ0RpfiVYBfK860YM=
|
golang.org/x/text v0.20.0 h1:gK/Kv2otX8gz+wn7Rmb3vT96ZwuoxnQlY+HlJVj7Qug=
|
||||||
golang.org/x/text v0.19.0/go.mod h1:BuEKDfySbSR4drPmRPG/7iBdf8hvFMuRexcpahXilzY=
|
golang.org/x/text v0.20.0/go.mod h1:D4IsuqiFMhST5bX19pQ9ikHC2GsaKyk/oF+pn3ducp4=
|
||||||
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-20241118233622-e639e219e697 h1:LWZqQOEjDyONlF1H6afSWpAL/znlREo2tHfLoe+8LMA=
|
||||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20241007155032-5fefd90f89a9/go.mod h1:GX3210XPVPUjJbTUbvwI8f2IpZDMZuPJWDzDuebbviI=
|
google.golang.org/genproto/googleapis/rpc v0.0.0-20241118233622-e639e219e697/go.mod h1:5uTbfoYQed2U9p3KIj2/Zzm02PYhndfdmML0qC3q3FU=
|
||||||
google.golang.org/grpc v1.67.1 h1:zWnc1Vrcno+lHZCOofnIMvycFcc0QRGIzm9dhnDX68E=
|
google.golang.org/grpc v1.68.0 h1:aHQeeJbo8zAkAa3pRzrVjZlbz6uSfeOXlJNQM0RAbz0=
|
||||||
google.golang.org/grpc v1.67.1/go.mod h1:1gLDyUQU7CTLJI90u3nXZ9ekeghjeM7pTDZlqFNg2AA=
|
google.golang.org/grpc v1.68.0/go.mod h1:fmSPC5AsjSBCK54MyHRx48kpOti1/jRfOlwEWywNjWA=
|
||||||
google.golang.org/protobuf v1.35.1 h1:m3LfL6/Ca+fqnjnlqQXNpFPABW1UD7mjh8KO2mKFytA=
|
google.golang.org/protobuf v1.35.2 h1:8Ar7bF+apOIoThw1EdZl0p1oWvMqTHmpA2fRTyZO8io=
|
||||||
google.golang.org/protobuf v1.35.1/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
|
google.golang.org/protobuf v1.35.2/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
|
||||||
|
40
grpc.go
40
grpc.go
@ -12,13 +12,13 @@ import (
|
|||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
|
"sync/atomic"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
greflection "google.golang.org/grpc/reflection"
|
greflection "google.golang.org/grpc/reflection"
|
||||||
reflectionv1pb "google.golang.org/grpc/reflection/grpc_reflection_v1"
|
reflectionv1pb "google.golang.org/grpc/reflection/grpc_reflection_v1"
|
||||||
|
|
||||||
// nolint: staticcheck
|
// nolint: staticcheck
|
||||||
oldproto "github.com/golang/protobuf/proto"
|
|
||||||
"go.unistack.org/micro/v3/broker"
|
"go.unistack.org/micro/v3/broker"
|
||||||
"go.unistack.org/micro/v3/codec"
|
"go.unistack.org/micro/v3/codec"
|
||||||
"go.unistack.org/micro/v3/errors"
|
"go.unistack.org/micro/v3/errors"
|
||||||
@ -39,6 +39,7 @@ import (
|
|||||||
"google.golang.org/grpc/peer"
|
"google.golang.org/grpc/peer"
|
||||||
"google.golang.org/grpc/status"
|
"google.golang.org/grpc/status"
|
||||||
"google.golang.org/protobuf/proto"
|
"google.golang.org/protobuf/proto"
|
||||||
|
"google.golang.org/protobuf/protoadapt"
|
||||||
)
|
)
|
||||||
|
|
||||||
const (
|
const (
|
||||||
@ -75,9 +76,12 @@ type Server struct {
|
|||||||
opts server.Options
|
opts server.Options
|
||||||
unknownHandler grpc.StreamHandler
|
unknownHandler grpc.StreamHandler
|
||||||
sync.RWMutex
|
sync.RWMutex
|
||||||
started bool
|
stateLive *atomic.Uint32
|
||||||
registered bool
|
stateReady *atomic.Uint32
|
||||||
reflection bool
|
stateHealth *atomic.Uint32
|
||||||
|
started bool
|
||||||
|
registered bool
|
||||||
|
reflection bool
|
||||||
}
|
}
|
||||||
|
|
||||||
func newServer(opts ...server.Option) *Server {
|
func newServer(opts ...server.Option) *Server {
|
||||||
@ -90,6 +94,9 @@ func newServer(opts ...server.Option) *Server {
|
|||||||
handlers: make(map[string]server.Handler),
|
handlers: make(map[string]server.Handler),
|
||||||
subscribers: make(map[*subscriber][]broker.Subscriber),
|
subscribers: make(map[*subscriber][]broker.Subscriber),
|
||||||
exit: make(chan chan error),
|
exit: make(chan chan error),
|
||||||
|
stateLive: &atomic.Uint32{},
|
||||||
|
stateReady: &atomic.Uint32{},
|
||||||
|
stateHealth: &atomic.Uint32{},
|
||||||
}
|
}
|
||||||
|
|
||||||
g.opts.Meter = g.opts.Meter.Clone(meter.Labels("type", "grpc"))
|
g.opts.Meter = g.opts.Meter.Clone(meter.Labels("type", "grpc"))
|
||||||
@ -438,7 +445,7 @@ func (g *Server) processRequest(ctx context.Context, stream grpc.ServerStream, s
|
|||||||
// user defined error that proto based we can attach it to grpc status
|
// user defined error that proto based we can attach it to grpc status
|
||||||
statusCode = convertCode(appErr)
|
statusCode = convertCode(appErr)
|
||||||
statusDesc = appErr.Error()
|
statusDesc = appErr.Error()
|
||||||
errStatus, err = status.New(statusCode, statusDesc).WithDetails(oldproto.MessageV1(verr))
|
errStatus, err = status.New(statusCode, statusDesc).WithDetails(protoadapt.MessageV1Of(verr))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@ -523,7 +530,7 @@ func (g *Server) processStream(ctx context.Context, stream grpc.ServerStream, se
|
|||||||
// user defined error that proto based we can attach it to grpc status
|
// user defined error that proto based we can attach it to grpc status
|
||||||
statusCode = convertCode(appErr)
|
statusCode = convertCode(appErr)
|
||||||
statusDesc = appErr.Error()
|
statusDesc = appErr.Error()
|
||||||
errStatus, err = status.New(statusCode, statusDesc).WithDetails(oldproto.MessageV1(verr))
|
errStatus, err = status.New(statusCode, statusDesc).WithDetails(protoadapt.MessageV1Of(verr))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@ -834,6 +841,9 @@ func (g *Server) Start() error {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
g.stateLive.Store(1)
|
||||||
|
g.stateReady.Store(1)
|
||||||
|
g.stateHealth.Store(1)
|
||||||
}()
|
}()
|
||||||
|
|
||||||
go func() {
|
go func() {
|
||||||
@ -903,12 +913,18 @@ func (g *Server) Start() error {
|
|||||||
go func() {
|
go func() {
|
||||||
g.srv.GracefulStop()
|
g.srv.GracefulStop()
|
||||||
close(exit)
|
close(exit)
|
||||||
|
g.stateLive.Store(0)
|
||||||
|
g.stateReady.Store(0)
|
||||||
|
g.stateHealth.Store(0)
|
||||||
}()
|
}()
|
||||||
|
|
||||||
select {
|
select {
|
||||||
case <-exit:
|
case <-exit:
|
||||||
case <-time.After(g.opts.GracefulTimeout):
|
case <-time.After(g.opts.GracefulTimeout):
|
||||||
g.srv.Stop()
|
g.srv.Stop()
|
||||||
|
g.stateLive.Store(0)
|
||||||
|
g.stateReady.Store(0)
|
||||||
|
g.stateHealth.Store(0)
|
||||||
}
|
}
|
||||||
|
|
||||||
// close transport
|
// close transport
|
||||||
@ -965,6 +981,18 @@ func (g *Server) GRPCServer() *grpc.Server {
|
|||||||
return g.srv
|
return g.srv
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (g *Server) Live() bool {
|
||||||
|
return g.stateLive.Load() == 1
|
||||||
|
}
|
||||||
|
|
||||||
|
func (g *Server) Ready() bool {
|
||||||
|
return g.stateReady.Load() == 1
|
||||||
|
}
|
||||||
|
|
||||||
|
func (g *Server) Health() bool {
|
||||||
|
return g.stateHealth.Load() == 1
|
||||||
|
}
|
||||||
|
|
||||||
func NewServer(opts ...server.Option) *Server {
|
func NewServer(opts ...server.Option) *Server {
|
||||||
return newServer(opts...)
|
return newServer(opts...)
|
||||||
}
|
}
|
||||||
|
@ -1,489 +0,0 @@
|
|||||||
//go:build ignore
|
|
||||||
// +build ignore
|
|
||||||
|
|
||||||
/*
|
|
||||||
*
|
|
||||||
* Copyright 2016 gRPC authors.
|
|
||||||
*
|
|
||||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
|
||||||
* you may not use this file except in compliance with the License.
|
|
||||||
* You may obtain a copy of the License at
|
|
||||||
*
|
|
||||||
* http://www.apache.org/licenses/LICENSE-2.0
|
|
||||||
*
|
|
||||||
* Unless required by applicable law or agreed to in writing, software
|
|
||||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
|
||||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
||||||
* See the License for the specific language governing permissions and
|
|
||||||
* limitations under the License.
|
|
||||||
*
|
|
||||||
*/
|
|
||||||
|
|
||||||
/*
|
|
||||||
Package reflection implements server reflection service.
|
|
||||||
|
|
||||||
The service implemented is defined in:
|
|
||||||
https://github.com/grpc/grpc/blob/master/src/proto/grpc/reflection/v1alpha/reflection.proto.
|
|
||||||
|
|
||||||
To register server reflection on a gRPC server:
|
|
||||||
|
|
||||||
import "google.golang.org/grpc/reflection"
|
|
||||||
|
|
||||||
s := grpc.NewServer()
|
|
||||||
pb.RegisterYourOwnServer(s, &server{})
|
|
||||||
|
|
||||||
// Register reflection service on gRPC server.
|
|
||||||
reflection.Register(s)
|
|
||||||
|
|
||||||
s.Serve(lis)
|
|
||||||
*/
|
|
||||||
package grpc
|
|
||||||
|
|
||||||
import (
|
|
||||||
"bytes"
|
|
||||||
"compress/gzip"
|
|
||||||
"fmt"
|
|
||||||
"io"
|
|
||||||
"io/ioutil"
|
|
||||||
"reflect"
|
|
||||||
"sort"
|
|
||||||
"sync"
|
|
||||||
|
|
||||||
"github.com/golang/protobuf/proto"
|
|
||||||
dpb "github.com/golang/protobuf/protoc-gen-go/descriptor"
|
|
||||||
"google.golang.org/grpc"
|
|
||||||
"google.golang.org/grpc/codes"
|
|
||||||
rpb "google.golang.org/grpc/reflection/grpc_reflection_v1alpha"
|
|
||||||
"google.golang.org/grpc/status"
|
|
||||||
)
|
|
||||||
|
|
||||||
type serverReflectionServer struct {
|
|
||||||
rpb.UnimplementedServerReflectionServer
|
|
||||||
s *grpc.Server
|
|
||||||
|
|
||||||
initSymbols sync.Once
|
|
||||||
serviceNames []string
|
|
||||||
symbols map[string]*dpb.FileDescriptorProto // map of fully-qualified names to files
|
|
||||||
}
|
|
||||||
|
|
||||||
// Register registers the server reflection service on the given gRPC server.
|
|
||||||
func Register(s *grpc.Server) {
|
|
||||||
rpb.RegisterServerReflectionServer(s, &serverReflectionServer{
|
|
||||||
s: s,
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
// protoMessage is used for type assertion on proto messages.
|
|
||||||
// Generated proto message implements function Descriptor(), but Descriptor()
|
|
||||||
// is not part of interface proto.Message. This interface is needed to
|
|
||||||
// call Descriptor().
|
|
||||||
type protoMessage interface {
|
|
||||||
Descriptor() ([]byte, []int)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (s *serverReflectionServer) getSymbols() (svcNames []string, symbolIndex map[string]*dpb.FileDescriptorProto) {
|
|
||||||
s.initSymbols.Do(func() {
|
|
||||||
serviceInfo := s.s.GetServiceInfo()
|
|
||||||
|
|
||||||
s.symbols = map[string]*dpb.FileDescriptorProto{}
|
|
||||||
s.serviceNames = make([]string, 0, len(serviceInfo))
|
|
||||||
processed := map[string]struct{}{}
|
|
||||||
for svc, info := range serviceInfo {
|
|
||||||
s.serviceNames = append(s.serviceNames, svc)
|
|
||||||
fdenc, ok := parseMetadata(info.Metadata)
|
|
||||||
if !ok {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
fd, err := decodeFileDesc(fdenc)
|
|
||||||
if err != nil {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
s.processFile(fd, processed)
|
|
||||||
}
|
|
||||||
sort.Strings(s.serviceNames)
|
|
||||||
})
|
|
||||||
|
|
||||||
return s.serviceNames, s.symbols
|
|
||||||
}
|
|
||||||
|
|
||||||
func (s *serverReflectionServer) processFile(fd *dpb.FileDescriptorProto, processed map[string]struct{}) {
|
|
||||||
filename := fd.GetName()
|
|
||||||
if _, ok := processed[filename]; ok {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
processed[filename] = struct{}{}
|
|
||||||
|
|
||||||
prefix := fd.GetPackage()
|
|
||||||
|
|
||||||
for _, msg := range fd.MessageType {
|
|
||||||
s.processMessage(fd, prefix, msg)
|
|
||||||
}
|
|
||||||
for _, en := range fd.EnumType {
|
|
||||||
s.processEnum(fd, prefix, en)
|
|
||||||
}
|
|
||||||
for _, ext := range fd.Extension {
|
|
||||||
s.processField(fd, prefix, ext)
|
|
||||||
}
|
|
||||||
for _, svc := range fd.Service {
|
|
||||||
svcName := fqn(prefix, svc.GetName())
|
|
||||||
s.symbols[svcName] = fd
|
|
||||||
for _, meth := range svc.Method {
|
|
||||||
name := fqn(svcName, meth.GetName())
|
|
||||||
s.symbols[name] = fd
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
for _, dep := range fd.Dependency {
|
|
||||||
fdenc := proto.FileDescriptor(dep)
|
|
||||||
fdDep, err := decodeFileDesc(fdenc)
|
|
||||||
if err != nil {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
s.processFile(fdDep, processed)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func (s *serverReflectionServer) processMessage(fd *dpb.FileDescriptorProto, prefix string, msg *dpb.DescriptorProto) {
|
|
||||||
msgName := fqn(prefix, msg.GetName())
|
|
||||||
s.symbols[msgName] = fd
|
|
||||||
|
|
||||||
for _, nested := range msg.NestedType {
|
|
||||||
s.processMessage(fd, msgName, nested)
|
|
||||||
}
|
|
||||||
for _, en := range msg.EnumType {
|
|
||||||
s.processEnum(fd, msgName, en)
|
|
||||||
}
|
|
||||||
for _, ext := range msg.Extension {
|
|
||||||
s.processField(fd, msgName, ext)
|
|
||||||
}
|
|
||||||
for _, fld := range msg.Field {
|
|
||||||
s.processField(fd, msgName, fld)
|
|
||||||
}
|
|
||||||
for _, oneof := range msg.OneofDecl {
|
|
||||||
oneofName := fqn(msgName, oneof.GetName())
|
|
||||||
s.symbols[oneofName] = fd
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func (s *serverReflectionServer) processEnum(fd *dpb.FileDescriptorProto, prefix string, en *dpb.EnumDescriptorProto) {
|
|
||||||
enName := fqn(prefix, en.GetName())
|
|
||||||
s.symbols[enName] = fd
|
|
||||||
|
|
||||||
for _, val := range en.Value {
|
|
||||||
valName := fqn(enName, val.GetName())
|
|
||||||
s.symbols[valName] = fd
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func (s *serverReflectionServer) processField(fd *dpb.FileDescriptorProto, prefix string, fld *dpb.FieldDescriptorProto) {
|
|
||||||
fldName := fqn(prefix, fld.GetName())
|
|
||||||
s.symbols[fldName] = fd
|
|
||||||
}
|
|
||||||
|
|
||||||
func fqn(prefix, name string) string {
|
|
||||||
if prefix == "" {
|
|
||||||
return name
|
|
||||||
}
|
|
||||||
return prefix + "." + name
|
|
||||||
}
|
|
||||||
|
|
||||||
// fileDescForType gets the file descriptor for the given type.
|
|
||||||
// The given type should be a proto message.
|
|
||||||
func (s *serverReflectionServer) fileDescForType(st reflect.Type) (*dpb.FileDescriptorProto, error) {
|
|
||||||
m, ok := reflect.Zero(reflect.PtrTo(st)).Interface().(protoMessage)
|
|
||||||
if !ok {
|
|
||||||
return nil, fmt.Errorf("failed to create message from type: %v", st)
|
|
||||||
}
|
|
||||||
enc, _ := m.Descriptor()
|
|
||||||
|
|
||||||
return decodeFileDesc(enc)
|
|
||||||
}
|
|
||||||
|
|
||||||
// decodeFileDesc does decompression and unmarshalling on the given
|
|
||||||
// file descriptor byte slice.
|
|
||||||
func decodeFileDesc(enc []byte) (*dpb.FileDescriptorProto, error) {
|
|
||||||
raw, err := decompress(enc)
|
|
||||||
if err != nil {
|
|
||||||
return nil, fmt.Errorf("failed to decompress enc: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
fd := new(dpb.FileDescriptorProto)
|
|
||||||
if err := proto.Unmarshal(raw, fd); err != nil {
|
|
||||||
return nil, fmt.Errorf("bad descriptor: %v", err)
|
|
||||||
}
|
|
||||||
return fd, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// decompress does gzip decompression.
|
|
||||||
func decompress(b []byte) ([]byte, error) {
|
|
||||||
r, err := gzip.NewReader(bytes.NewReader(b))
|
|
||||||
if err != nil {
|
|
||||||
return nil, fmt.Errorf("bad gzipped descriptor: %v", err)
|
|
||||||
}
|
|
||||||
out, err := ioutil.ReadAll(r)
|
|
||||||
if err != nil {
|
|
||||||
return nil, fmt.Errorf("bad gzipped descriptor: %v", err)
|
|
||||||
}
|
|
||||||
return out, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func typeForName(name string) (reflect.Type, error) {
|
|
||||||
pt := proto.MessageType(name)
|
|
||||||
if pt == nil {
|
|
||||||
return nil, fmt.Errorf("unknown type: %q", name)
|
|
||||||
}
|
|
||||||
st := pt.Elem()
|
|
||||||
|
|
||||||
return st, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func fileDescContainingExtension(st reflect.Type, ext int32) (*dpb.FileDescriptorProto, error) {
|
|
||||||
m, ok := reflect.Zero(reflect.PtrTo(st)).Interface().(proto.Message)
|
|
||||||
if !ok {
|
|
||||||
return nil, fmt.Errorf("failed to create message from type: %v", st)
|
|
||||||
}
|
|
||||||
|
|
||||||
var extDesc *proto.ExtensionDesc
|
|
||||||
for id, desc := range proto.RegisteredExtensions(m) {
|
|
||||||
if id == ext {
|
|
||||||
extDesc = desc
|
|
||||||
break
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if extDesc == nil {
|
|
||||||
return nil, fmt.Errorf("failed to find registered extension for extension number %v", ext)
|
|
||||||
}
|
|
||||||
|
|
||||||
return decodeFileDesc(proto.FileDescriptor(extDesc.Filename))
|
|
||||||
}
|
|
||||||
|
|
||||||
func (s *serverReflectionServer) allExtensionNumbersForType(st reflect.Type) ([]int32, error) {
|
|
||||||
m, ok := reflect.Zero(reflect.PtrTo(st)).Interface().(proto.Message)
|
|
||||||
if !ok {
|
|
||||||
return nil, fmt.Errorf("failed to create message from type: %v", st)
|
|
||||||
}
|
|
||||||
|
|
||||||
exts := proto.RegisteredExtensions(m)
|
|
||||||
out := make([]int32, 0, len(exts))
|
|
||||||
for id := range exts {
|
|
||||||
out = append(out, id)
|
|
||||||
}
|
|
||||||
return out, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// fileDescWithDependencies returns a slice of serialized fileDescriptors in
|
|
||||||
// wire format ([]byte). The fileDescriptors will include fd and all the
|
|
||||||
// transitive dependencies of fd with names not in sentFileDescriptors.
|
|
||||||
func fileDescWithDependencies(fd *dpb.FileDescriptorProto, sentFileDescriptors map[string]bool) ([][]byte, error) {
|
|
||||||
r := [][]byte{}
|
|
||||||
queue := []*dpb.FileDescriptorProto{fd}
|
|
||||||
for len(queue) > 0 {
|
|
||||||
currentfd := queue[0]
|
|
||||||
queue = queue[1:]
|
|
||||||
if sent := sentFileDescriptors[currentfd.GetName()]; len(r) == 0 || !sent {
|
|
||||||
sentFileDescriptors[currentfd.GetName()] = true
|
|
||||||
currentfdEncoded, err := proto.Marshal(currentfd)
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
r = append(r, currentfdEncoded)
|
|
||||||
}
|
|
||||||
for _, dep := range currentfd.Dependency {
|
|
||||||
fdenc := proto.FileDescriptor(dep)
|
|
||||||
fdDep, err := decodeFileDesc(fdenc)
|
|
||||||
if err != nil {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
queue = append(queue, fdDep)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return r, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// fileDescEncodingByFilename finds the file descriptor for given filename,
|
|
||||||
// finds all of its previously unsent transitive dependencies, does marshalling
|
|
||||||
// on them, and returns the marshalled result.
|
|
||||||
func (s *serverReflectionServer) fileDescEncodingByFilename(name string, sentFileDescriptors map[string]bool) ([][]byte, error) {
|
|
||||||
enc := proto.FileDescriptor(name)
|
|
||||||
if enc == nil {
|
|
||||||
return nil, fmt.Errorf("unknown file: %v", name)
|
|
||||||
}
|
|
||||||
fd, err := decodeFileDesc(enc)
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
return fileDescWithDependencies(fd, sentFileDescriptors)
|
|
||||||
}
|
|
||||||
|
|
||||||
// parseMetadata finds the file descriptor bytes specified meta.
|
|
||||||
// For SupportPackageIsVersion4, m is the name of the proto file, we
|
|
||||||
// call proto.FileDescriptor to get the byte slice.
|
|
||||||
// For SupportPackageIsVersion3, m is a byte slice itself.
|
|
||||||
func parseMetadata(meta interface{}) ([]byte, bool) {
|
|
||||||
// Check if meta is the file name.
|
|
||||||
if fileNameForMeta, ok := meta.(string); ok {
|
|
||||||
return proto.FileDescriptor(fileNameForMeta), true
|
|
||||||
}
|
|
||||||
|
|
||||||
// Check if meta is the byte slice.
|
|
||||||
if enc, ok := meta.([]byte); ok {
|
|
||||||
return enc, true
|
|
||||||
}
|
|
||||||
|
|
||||||
return nil, false
|
|
||||||
}
|
|
||||||
|
|
||||||
// fileDescEncodingContainingSymbol finds the file descriptor containing the
|
|
||||||
// given symbol, finds all of its previously unsent transitive dependencies,
|
|
||||||
// does marshalling on them, and returns the marshalled result. The given symbol
|
|
||||||
// can be a type, a service or a method.
|
|
||||||
func (s *serverReflectionServer) fileDescEncodingContainingSymbol(name string, sentFileDescriptors map[string]bool) ([][]byte, error) {
|
|
||||||
_, symbols := s.getSymbols()
|
|
||||||
fd := symbols[name]
|
|
||||||
if fd == nil {
|
|
||||||
// Check if it's a type name that was not present in the
|
|
||||||
// transitive dependencies of the registered services.
|
|
||||||
if st, err := typeForName(name); err == nil {
|
|
||||||
fd, err = s.fileDescForType(st)
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if fd == nil {
|
|
||||||
return nil, fmt.Errorf("unknown symbol: %v", name)
|
|
||||||
}
|
|
||||||
|
|
||||||
return fileDescWithDependencies(fd, sentFileDescriptors)
|
|
||||||
}
|
|
||||||
|
|
||||||
// fileDescEncodingContainingExtension finds the file descriptor containing
|
|
||||||
// given extension, finds all of its previously unsent transitive dependencies,
|
|
||||||
// does marshalling on them, and returns the marshalled result.
|
|
||||||
func (s *serverReflectionServer) fileDescEncodingContainingExtension(typeName string, extNum int32, sentFileDescriptors map[string]bool) ([][]byte, error) {
|
|
||||||
st, err := typeForName(typeName)
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
fd, err := fileDescContainingExtension(st, extNum)
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
return fileDescWithDependencies(fd, sentFileDescriptors)
|
|
||||||
}
|
|
||||||
|
|
||||||
// allExtensionNumbersForTypeName returns all extension numbers for the given type.
|
|
||||||
func (s *serverReflectionServer) allExtensionNumbersForTypeName(name string) ([]int32, error) {
|
|
||||||
st, err := typeForName(name)
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
extNums, err := s.allExtensionNumbersForType(st)
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
return extNums, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// ServerReflectionInfo is the reflection service handler.
|
|
||||||
func (s *serverReflectionServer) ServerReflectionInfo(stream rpb.ServerReflection_ServerReflectionInfoServer) error {
|
|
||||||
sentFileDescriptors := make(map[string]bool)
|
|
||||||
for {
|
|
||||||
in, err := stream.Recv()
|
|
||||||
if err == io.EOF {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
out := &rpb.ServerReflectionResponse{
|
|
||||||
ValidHost: in.Host,
|
|
||||||
OriginalRequest: in,
|
|
||||||
}
|
|
||||||
switch req := in.MessageRequest.(type) {
|
|
||||||
case *rpb.ServerReflectionRequest_FileByFilename:
|
|
||||||
b, err := s.fileDescEncodingByFilename(req.FileByFilename, sentFileDescriptors)
|
|
||||||
if err != nil {
|
|
||||||
out.MessageResponse = &rpb.ServerReflectionResponse_ErrorResponse{
|
|
||||||
ErrorResponse: &rpb.ErrorResponse{
|
|
||||||
ErrorCode: int32(codes.NotFound),
|
|
||||||
ErrorMessage: err.Error(),
|
|
||||||
},
|
|
||||||
}
|
|
||||||
} else {
|
|
||||||
out.MessageResponse = &rpb.ServerReflectionResponse_FileDescriptorResponse{
|
|
||||||
FileDescriptorResponse: &rpb.FileDescriptorResponse{FileDescriptorProto: b},
|
|
||||||
}
|
|
||||||
}
|
|
||||||
case *rpb.ServerReflectionRequest_FileContainingSymbol:
|
|
||||||
b, err := s.fileDescEncodingContainingSymbol(req.FileContainingSymbol, sentFileDescriptors)
|
|
||||||
if err != nil {
|
|
||||||
out.MessageResponse = &rpb.ServerReflectionResponse_ErrorResponse{
|
|
||||||
ErrorResponse: &rpb.ErrorResponse{
|
|
||||||
ErrorCode: int32(codes.NotFound),
|
|
||||||
ErrorMessage: err.Error(),
|
|
||||||
},
|
|
||||||
}
|
|
||||||
} else {
|
|
||||||
out.MessageResponse = &rpb.ServerReflectionResponse_FileDescriptorResponse{
|
|
||||||
FileDescriptorResponse: &rpb.FileDescriptorResponse{FileDescriptorProto: b},
|
|
||||||
}
|
|
||||||
}
|
|
||||||
case *rpb.ServerReflectionRequest_FileContainingExtension:
|
|
||||||
typeName := req.FileContainingExtension.ContainingType
|
|
||||||
extNum := req.FileContainingExtension.ExtensionNumber
|
|
||||||
b, err := s.fileDescEncodingContainingExtension(typeName, extNum, sentFileDescriptors)
|
|
||||||
if err != nil {
|
|
||||||
out.MessageResponse = &rpb.ServerReflectionResponse_ErrorResponse{
|
|
||||||
ErrorResponse: &rpb.ErrorResponse{
|
|
||||||
ErrorCode: int32(codes.NotFound),
|
|
||||||
ErrorMessage: err.Error(),
|
|
||||||
},
|
|
||||||
}
|
|
||||||
} else {
|
|
||||||
out.MessageResponse = &rpb.ServerReflectionResponse_FileDescriptorResponse{
|
|
||||||
FileDescriptorResponse: &rpb.FileDescriptorResponse{FileDescriptorProto: b},
|
|
||||||
}
|
|
||||||
}
|
|
||||||
case *rpb.ServerReflectionRequest_AllExtensionNumbersOfType:
|
|
||||||
extNums, err := s.allExtensionNumbersForTypeName(req.AllExtensionNumbersOfType)
|
|
||||||
if err != nil {
|
|
||||||
out.MessageResponse = &rpb.ServerReflectionResponse_ErrorResponse{
|
|
||||||
ErrorResponse: &rpb.ErrorResponse{
|
|
||||||
ErrorCode: int32(codes.NotFound),
|
|
||||||
ErrorMessage: err.Error(),
|
|
||||||
},
|
|
||||||
}
|
|
||||||
} else {
|
|
||||||
out.MessageResponse = &rpb.ServerReflectionResponse_AllExtensionNumbersResponse{
|
|
||||||
AllExtensionNumbersResponse: &rpb.ExtensionNumberResponse{
|
|
||||||
BaseTypeName: req.AllExtensionNumbersOfType,
|
|
||||||
ExtensionNumber: extNums,
|
|
||||||
},
|
|
||||||
}
|
|
||||||
}
|
|
||||||
case *rpb.ServerReflectionRequest_ListServices:
|
|
||||||
svcNames, _ := s.getSymbols()
|
|
||||||
serviceResponses := make([]*rpb.ServiceResponse, len(svcNames))
|
|
||||||
for i, n := range svcNames {
|
|
||||||
serviceResponses[i] = &rpb.ServiceResponse{
|
|
||||||
Name: n,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
out.MessageResponse = &rpb.ServerReflectionResponse_ListServicesResponse{
|
|
||||||
ListServicesResponse: &rpb.ListServiceResponse{
|
|
||||||
Service: serviceResponses,
|
|
||||||
},
|
|
||||||
}
|
|
||||||
default:
|
|
||||||
return status.Errorf(codes.InvalidArgument, "invalid MessageRequest: %v", in.MessageRequest)
|
|
||||||
}
|
|
||||||
|
|
||||||
if err := stream.Send(out); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
Loading…
x
Reference in New Issue
Block a user