minimize allocations (#1472)

* server: minimize allocations on re-register

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

* server: stop old instance before Init()

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

* client/grpc: fix allocations in protobuf marshal

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

* codec/json: fix allocations in protobuf marshal

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

* remove stop from init

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

* codec/grpc: expose MaxMessageSize

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

* codec: use buffer pool

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

* metadata: minimize reallocations

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

* util/wrapper: use metadata helper

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

* registry/cache: move logs to debug level

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

* server: move logs to debug level

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

* server: cache service only when Advertise is ip addr

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

* server: use metadata.Copy

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
This commit is contained in:
Василий Толстов 2020-04-08 12:50:19 +03:00 committed by GitHub
parent 98fc3dfbad
commit 1fbc056dd4
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 220 additions and 83 deletions

View File

@ -11,6 +11,7 @@ import (
"github.com/golang/protobuf/proto" "github.com/golang/protobuf/proto"
"github.com/micro/go-micro/v2/codec" "github.com/micro/go-micro/v2/codec"
"github.com/micro/go-micro/v2/codec/bytes" "github.com/micro/go-micro/v2/codec/bytes"
"github.com/oxtoacart/bpool"
"google.golang.org/grpc" "google.golang.org/grpc"
"google.golang.org/grpc/encoding" "google.golang.org/grpc/encoding"
) )
@ -23,6 +24,9 @@ type wrapCodec struct{ encoding.Codec }
var jsonpbMarshaler = &jsonpb.Marshaler{} var jsonpbMarshaler = &jsonpb.Marshaler{}
var useNumber bool var useNumber bool
// create buffer pool with 16 instances each preallocated with 256 bytes
var bufferPool = bpool.NewSizedBufferPool(16, 256)
var ( var (
defaultGRPCCodecs = map[string]encoding.Codec{ defaultGRPCCodecs = map[string]encoding.Codec{
"application/json": jsonCodec{}, "application/json": jsonCodec{},
@ -106,14 +110,19 @@ func (bytesCodec) Name() string {
} }
func (jsonCodec) Marshal(v interface{}) ([]byte, error) { func (jsonCodec) Marshal(v interface{}) ([]byte, error) {
if pb, ok := v.(proto.Message); ok {
s, err := jsonpbMarshaler.MarshalToString(pb)
return []byte(s), err
}
if b, ok := v.(*bytes.Frame); ok { if b, ok := v.(*bytes.Frame); ok {
return b.Data, nil return b.Data, nil
} }
if pb, ok := v.(proto.Message); ok {
buf := bufferPool.Get()
defer bufferPool.Put(buf)
if err := jsonpbMarshaler.Marshal(buf, pb); err != nil {
return nil, err
}
return buf.Bytes(), nil
}
return json.Marshal(v) return json.Marshal(v)
} }

View File

@ -1,7 +1,7 @@
package bytes package bytes
import ( import (
"errors" "github.com/micro/go-micro/v2/codec"
) )
type Marshaler struct{} type Marshaler struct{}
@ -20,7 +20,7 @@ func (n Marshaler) Marshal(v interface{}) ([]byte, error) {
case *Message: case *Message:
return ve.Body, nil return ve.Body, nil
} }
return nil, errors.New("invalid message") return nil, codec.ErrInvalidMessage
} }
func (n Marshaler) Unmarshal(d []byte, v interface{}) error { func (n Marshaler) Unmarshal(d []byte, v interface{}) error {
@ -30,7 +30,7 @@ func (n Marshaler) Unmarshal(d []byte, v interface{}) error {
case *Message: case *Message:
ve.Body = d ve.Body = d
} }
return errors.New("invalid message") return codec.ErrInvalidMessage
} }
func (n Marshaler) String() string { func (n Marshaler) String() string {

View File

@ -7,7 +7,7 @@ import (
) )
var ( var (
maxMessageSize = 1024 * 1024 * 4 MaxMessageSize = 1024 * 1024 * 4 // 4Mb
maxInt = int(^uint(0) >> 1) maxInt = int(^uint(0) >> 1)
) )
@ -34,8 +34,8 @@ func decode(r io.Reader) (uint8, []byte, error) {
if int64(length) > int64(maxInt) { if int64(length) > int64(maxInt) {
return cf, nil, fmt.Errorf("grpc: received message larger than max length allowed on current machine (%d vs. %d)", length, maxInt) return cf, nil, fmt.Errorf("grpc: received message larger than max length allowed on current machine (%d vs. %d)", length, maxInt)
} }
if int(length) > maxMessageSize { if int(length) > MaxMessageSize {
return cf, nil, fmt.Errorf("grpc: received message larger than max (%d vs. %d)", length, maxMessageSize) return cf, nil, fmt.Errorf("grpc: received message larger than max (%d vs. %d)", length, MaxMessageSize)
} }
msg := make([]byte, int(length)) msg := make([]byte, int(length))

View File

@ -1,21 +1,36 @@
package json package json
import ( import (
"bytes"
"encoding/json" "encoding/json"
"github.com/golang/protobuf/jsonpb" "github.com/golang/protobuf/jsonpb"
"github.com/golang/protobuf/proto" "github.com/golang/protobuf/proto"
"github.com/oxtoacart/bpool"
) )
var jsonpbMarshaler = &jsonpb.Marshaler{}
// create buffer pool with 16 instances each preallocated with 256 bytes
var bufferPool = bpool.NewSizedBufferPool(16, 256)
type Marshaler struct{} type Marshaler struct{}
func (j Marshaler) Marshal(v interface{}) ([]byte, error) { func (j Marshaler) Marshal(v interface{}) ([]byte, error) {
if pb, ok := v.(proto.Message); ok {
buf := bufferPool.Get()
defer bufferPool.Put(buf)
if err := jsonpbMarshaler.Marshal(buf, pb); err != nil {
return nil, err
}
return buf.Bytes(), nil
}
return json.Marshal(v) return json.Marshal(v)
} }
func (j Marshaler) Unmarshal(d []byte, v interface{}) error { func (j Marshaler) Unmarshal(d []byte, v interface{}) error {
if pb, ok := v.(proto.Message); ok { if pb, ok := v.(proto.Message); ok {
return jsonpb.UnmarshalString(string(d), pb) return jsonpb.Unmarshal(bytes.NewReader(d), pb)
} }
return json.Unmarshal(d, v) return json.Unmarshal(d, v)
} }

View File

@ -1,17 +1,45 @@
package proto package proto
import ( import (
"bytes"
"github.com/golang/protobuf/proto" "github.com/golang/protobuf/proto"
"github.com/micro/go-micro/v2/codec"
"github.com/oxtoacart/bpool"
) )
// create buffer pool with 16 instances each preallocated with 256 bytes
var bufferPool = bpool.NewSizedBufferPool(16, 256)
type Marshaler struct{} type Marshaler struct{}
func (Marshaler) Marshal(v interface{}) ([]byte, error) { func (Marshaler) Marshal(v interface{}) ([]byte, error) {
return proto.Marshal(v.(proto.Message)) pb, ok := v.(proto.Message)
if !ok {
return nil, codec.ErrInvalidMessage
}
// looks not good, but allows to reuse underlining bytes
buf := bufferPool.Get()
pbuf := proto.NewBuffer(buf.Bytes())
defer func() {
bufferPool.Put(bytes.NewBuffer(pbuf.Bytes()))
}()
if err := pbuf.Marshal(pb); err != nil {
return nil, err
}
return pbuf.Bytes(), nil
} }
func (Marshaler) Unmarshal(data []byte, v interface{}) error { func (Marshaler) Unmarshal(data []byte, v interface{}) error {
return proto.Unmarshal(data, v.(proto.Message)) pb, ok := v.(proto.Message)
if !ok {
return codec.ErrInvalidMessage
}
return proto.Unmarshal(data, pb)
} }
func (Marshaler) String() string { func (Marshaler) String() string {

View File

@ -34,7 +34,7 @@ func (md Metadata) Delete(key string) {
// Copy makes a copy of the metadata // Copy makes a copy of the metadata
func Copy(md Metadata) Metadata { func Copy(md Metadata) Metadata {
cmd := make(Metadata) cmd := make(Metadata, len(md))
for k, v := range md { for k, v := range md {
cmd[k] = v cmd[k] = v
} }
@ -86,7 +86,7 @@ func FromContext(ctx context.Context) (Metadata, bool) {
} }
// capitalise all values // capitalise all values
newMD := make(map[string]string, len(md)) newMD := make(Metadata, len(md))
for k, v := range md { for k, v := range md {
newMD[strings.Title(k)] = v newMD[strings.Title(k)] = v
} }
@ -105,7 +105,7 @@ func MergeContext(ctx context.Context, patchMd Metadata, overwrite bool) context
ctx = context.Background() ctx = context.Background()
} }
md, _ := ctx.Value(MetadataKey{}).(Metadata) md, _ := ctx.Value(MetadataKey{}).(Metadata)
cmd := make(Metadata) cmd := make(Metadata, len(md))
for k, v := range md { for k, v := range md {
cmd[k] = v cmd[k] = v
} }

View File

@ -339,8 +339,8 @@ func (c *cache) run() {
c.setStatus(err) c.setStatus(err)
if a > 3 { if a > 3 {
if logger.V(logger.InfoLevel, logger.DefaultLogger) { if logger.V(logger.DebugLevel, logger.DefaultLogger) {
logger.Info("rcache: ", err, " backing off ", d) logger.Debug("rcache: ", err, " backing off ", d)
} }
a = 0 a = 0
} }
@ -364,8 +364,8 @@ func (c *cache) run() {
c.setStatus(err) c.setStatus(err)
if b > 3 { if b > 3 {
if logger.V(logger.InfoLevel, logger.DefaultLogger) { if logger.V(logger.DebugLevel, logger.DefaultLogger) {
logger.Info("rcache: ", err, " backing off ", d) logger.Debug("rcache: ", err, " backing off ", d)
} }
b = 0 b = 0
} }

View File

@ -59,6 +59,9 @@ type grpcServer struct {
started bool started bool
// used for first registration // used for first registration
registered bool registered bool
// registry service instance
rsvc *registry.Service
} }
func init() { func init() {
@ -102,6 +105,9 @@ func (r grpcRouter) ServeRequest(ctx context.Context, req server.Request, rsp se
} }
func (g *grpcServer) configure(opts ...server.Option) { func (g *grpcServer) configure(opts ...server.Option) {
g.Lock()
defer g.Unlock()
// Don't reprocess where there's no config // Don't reprocess where there's no config
if len(opts) == 0 && g.srv != nil { if len(opts) == 0 && g.srv != nil {
return return
@ -127,6 +133,7 @@ func (g *grpcServer) configure(opts ...server.Option) {
gopts = append(gopts, opts...) gopts = append(gopts, opts...)
} }
g.rsvc = nil
g.srv = grpc.NewServer(gopts...) g.srv = grpc.NewServer(gopts...)
} }
@ -559,11 +566,24 @@ func (g *grpcServer) Subscribe(sb server.Subscriber) error {
} }
func (g *grpcServer) Register() error { func (g *grpcServer) Register() error {
g.RLock()
rsvc := g.rsvc
config := g.opts
g.RUnlock()
// if service already filled, reuse it and return early
if rsvc != nil {
rOpts := []registry.RegisterOption{registry.RegisterTTL(config.RegisterTTL)}
if err := config.Registry.Register(rsvc, rOpts...); err != nil {
return err
}
return nil
}
var err error var err error
var advt, host, port string var advt, host, port string
var cacheService bool
// parse address for host, port
config := g.opts
// check the advertise address first // check the advertise address first
// if it exists then use it, otherwise // if it exists then use it, otherwise
@ -584,16 +604,17 @@ func (g *grpcServer) Register() error {
host = advt host = advt
} }
if ip := net.ParseIP(host); ip != nil {
cacheService = true
}
addr, err := addr.Extract(host) addr, err := addr.Extract(host)
if err != nil { if err != nil {
return err return err
} }
// make copy of metadata // make copy of metadata
md := make(meta.Metadata) md := meta.Copy(config.Metadata)
for k, v := range config.Metadata {
md[k] = v
}
// register service // register service
node := &registry.Node{ node := &registry.Node{
@ -646,13 +667,13 @@ func (g *grpcServer) Register() error {
Endpoints: endpoints, Endpoints: endpoints,
} }
g.Lock() g.RLock()
registered := g.registered registered := g.registered
g.Unlock() g.RUnlock()
if !registered { if !registered {
if logger.V(logger.InfoLevel, logger.DefaultLogger) { if logger.V(logger.DebugLevel, logger.DefaultLogger) {
logger.Infof("Registry [%s] Registering node: %s", config.Registry.String(), node.Id) logger.Debugf("Registry [%s] Registering node: %s", config.Registry.String(), node.Id)
} }
} }
@ -671,6 +692,9 @@ func (g *grpcServer) Register() error {
g.Lock() g.Lock()
defer g.Unlock() defer g.Unlock()
if cacheService {
g.rsvc = service
}
g.registered = true g.registered = true
for sb := range g.subscribers { for sb := range g.subscribers {
@ -688,8 +712,8 @@ func (g *grpcServer) Register() error {
opts = append(opts, broker.DisableAutoAck()) opts = append(opts, broker.DisableAutoAck())
} }
if logger.V(logger.InfoLevel, logger.DefaultLogger) { if logger.V(logger.DebugLevel, logger.DefaultLogger) {
logger.Infof("Subscribing to topic: %s", sb.Topic()) logger.Debug("Subscribing to topic: %s", sb.Topic())
} }
sub, err := config.Broker.Subscribe(sb.Topic(), handler, opts...) sub, err := config.Broker.Subscribe(sb.Topic(), handler, opts...)
if err != nil { if err != nil {
@ -705,7 +729,9 @@ func (g *grpcServer) Deregister() error {
var err error var err error
var advt, host, port string var advt, host, port string
g.RLock()
config := g.opts config := g.opts
g.RUnlock()
// check the advertise address first // check the advertise address first
// if it exists then use it, otherwise // if it exists then use it, otherwise
@ -742,14 +768,15 @@ func (g *grpcServer) Deregister() error {
Nodes: []*registry.Node{node}, Nodes: []*registry.Node{node},
} }
if logger.V(logger.InfoLevel, logger.DefaultLogger) { if logger.V(logger.DebugLevel, logger.DefaultLogger) {
logger.Infof("Deregistering node: %s", node.Id) logger.Debugf("Deregistering node: %s", node.Id)
} }
if err := config.Registry.Deregister(service); err != nil { if err := config.Registry.Deregister(service); err != nil {
return err return err
} }
g.Lock() g.Lock()
g.rsvc = nil
if !g.registered { if !g.registered {
g.Unlock() g.Unlock()
@ -760,8 +787,8 @@ func (g *grpcServer) Deregister() error {
for sb, subs := range g.subscribers { for sb, subs := range g.subscribers {
for _, sub := range subs { for _, sub := range subs {
if logger.V(logger.InfoLevel, logger.DefaultLogger) { if logger.V(logger.DebugLevel, logger.DefaultLogger) {
logger.Infof("Unsubscribing from topic: %s", sub.Topic()) logger.Debugf("Unsubscribing from topic: %s", sub.Topic())
} }
sub.Unsubscribe() sub.Unsubscribe()
} }
@ -819,11 +846,14 @@ func (g *grpcServer) Start() error {
if len(g.subscribers) > 0 { if len(g.subscribers) > 0 {
// connect to the broker // connect to the broker
if err := config.Broker.Connect(); err != nil { if err := config.Broker.Connect(); err != nil {
if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
logger.Errorf("Broker [%s] connect error: %v", config.Broker.String(), err)
}
return err return err
} }
if logger.V(logger.InfoLevel, logger.DefaultLogger) { if logger.V(logger.DebugLevel, logger.DefaultLogger) {
logger.Infof("Broker [%s] Connected to %s", config.Broker.String(), config.Broker.Address()) logger.Debugf("Broker [%s] Connected to %s", config.Broker.String(), config.Broker.Address())
} }
} }
@ -900,11 +930,15 @@ func (g *grpcServer) Start() error {
// close transport // close transport
ch <- nil ch <- nil
if logger.V(logger.InfoLevel, logger.DefaultLogger) { if logger.V(logger.DebugLevel, logger.DefaultLogger) {
logger.Infof("Broker [%s] Disconnected from %s", config.Broker.String(), config.Broker.Address()) logger.Debugf("Broker [%s] Disconnected from %s", config.Broker.String(), config.Broker.Address())
} }
// disconnect broker // disconnect broker
config.Broker.Disconnect() if err := config.Broker.Disconnect(); err != nil {
if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
logger.Errorf("Broker [%s] disconnect error: %v", config.Broker.String(), err)
}
}
}() }()
// mark the server as started // mark the server as started
@ -930,6 +964,7 @@ func (g *grpcServer) Stop() error {
select { select {
case err = <-ch: case err = <-ch:
g.Lock() g.Lock()
g.rsvc = nil
g.started = false g.started = false
g.Unlock() g.Unlock()
} }

View File

@ -40,6 +40,8 @@ type rpcServer struct {
subscriber broker.Subscriber subscriber broker.Subscriber
// graceful exit // graceful exit
wg *sync.WaitGroup wg *sync.WaitGroup
rsvc *registry.Service
} }
func newRpcServer(opts ...Option) Server { func newRpcServer(opts ...Option) Server {
@ -459,10 +461,11 @@ func (s *rpcServer) Options() Options {
func (s *rpcServer) Init(opts ...Option) error { func (s *rpcServer) Init(opts ...Option) error {
s.Lock() s.Lock()
defer s.Unlock()
for _, opt := range opts { for _, opt := range opts {
opt(&s.opts) opt(&s.opts)
} }
// update router if its the default // update router if its the default
if s.opts.Router == nil { if s.opts.Router == nil {
r := newRpcRouter() r := newRpcRouter()
@ -472,7 +475,8 @@ func (s *rpcServer) Init(opts ...Option) error {
s.router = r s.router = r
} }
s.Unlock() s.rsvc = nil
return nil return nil
} }
@ -510,11 +514,24 @@ func (s *rpcServer) Subscribe(sb Subscriber) error {
} }
func (s *rpcServer) Register() error { func (s *rpcServer) Register() error {
s.RLock()
rsvc := s.rsvc
config := s.Options()
s.RUnlock()
if rsvc != nil {
rOpts := []registry.RegisterOption{registry.RegisterTTL(config.RegisterTTL)}
if err := config.Registry.Register(rsvc, rOpts...); err != nil {
return err
}
return nil
}
var err error var err error
var advt, host, port string var advt, host, port string
var cacheService bool
// parse address for host, port
config := s.Options()
// check the advertise address first // check the advertise address first
// if it exists then use it, otherwise // if it exists then use it, otherwise
@ -535,16 +552,17 @@ func (s *rpcServer) Register() error {
host = advt host = advt
} }
if ip := net.ParseIP(host); ip != nil {
cacheService = true
}
addr, err := addr.Extract(host) addr, err := addr.Extract(host)
if err != nil { if err != nil {
return err return err
} }
// make copy of metadata // make copy of metadata
md := make(metadata.Metadata) md := metadata.Copy(config.Metadata)
for k, v := range config.Metadata {
md[k] = v
}
// mq-rpc(eg. nats) doesn't need the port. its addr is queue name. // mq-rpc(eg. nats) doesn't need the port. its addr is queue name.
if port != "" { if port != "" {
@ -612,7 +630,9 @@ func (s *rpcServer) Register() error {
s.RUnlock() s.RUnlock()
if !registered { if !registered {
log.Infof("Registry [%s] Registering node: %s", config.Registry.String(), node.Id) if logger.V(logger.DebugLevel, logger.DefaultLogger) {
log.Debugf("Registry [%s] Registering node: %s", config.Registry.String(), node.Id)
}
} }
// create registry options // create registry options
@ -630,6 +650,9 @@ func (s *rpcServer) Register() error {
s.Lock() s.Lock()
defer s.Unlock() defer s.Unlock()
if cacheService {
s.rsvc = service
}
s.registered = true s.registered = true
// set what we're advertising // set what we're advertising
s.opts.Advertise = addr s.opts.Advertise = addr
@ -665,8 +688,9 @@ func (s *rpcServer) Register() error {
if err != nil { if err != nil {
return err return err
} }
log.Infof("Subscribing to topic: %s", sub.Topic()) if logger.V(logger.DebugLevel, logger.DefaultLogger) {
log.Debugf("Subscribing to topic: %s", sub.Topic())
}
s.subscribers[sb] = []broker.Subscriber{sub} s.subscribers[sb] = []broker.Subscriber{sub}
} }
@ -677,7 +701,9 @@ func (s *rpcServer) Deregister() error {
var err error var err error
var advt, host, port string var advt, host, port string
s.RLock()
config := s.Options() config := s.Options()
s.RUnlock()
// check the advertise address first // check the advertise address first
// if it exists then use it, otherwise // if it exists then use it, otherwise
@ -719,12 +745,15 @@ func (s *rpcServer) Deregister() error {
Nodes: []*registry.Node{node}, Nodes: []*registry.Node{node},
} }
log.Infof("Registry [%s] Deregistering node: %s", config.Registry.String(), node.Id) if logger.V(logger.DebugLevel, logger.DefaultLogger) {
log.Debugf("Registry [%s] Deregistering node: %s", config.Registry.String(), node.Id)
}
if err := config.Registry.Deregister(service); err != nil { if err := config.Registry.Deregister(service); err != nil {
return err return err
} }
s.Lock() s.Lock()
s.rsvc = nil
if !s.registered { if !s.registered {
s.Unlock() s.Unlock()
@ -741,7 +770,9 @@ func (s *rpcServer) Deregister() error {
for sb, subs := range s.subscribers { for sb, subs := range s.subscribers {
for _, sub := range subs { for _, sub := range subs {
log.Infof("Unsubscribing %s from topic: %s", node.Id, sub.Topic()) if logger.V(logger.DebugLevel, logger.DefaultLogger) {
log.Debugf("Unsubscribing %s from topic: %s", node.Id, sub.Topic())
}
sub.Unsubscribe() sub.Unsubscribe()
} }
s.subscribers[sb] = nil s.subscribers[sb] = nil
@ -767,7 +798,9 @@ func (s *rpcServer) Start() error {
return err return err
} }
log.Infof("Transport [%s] Listening on %s", config.Transport.String(), ts.Addr()) if logger.V(logger.DebugLevel, logger.DefaultLogger) {
log.Debugf("Transport [%s] Listening on %s", config.Transport.String(), ts.Addr())
}
// swap address // swap address
s.Lock() s.Lock()
@ -775,22 +808,31 @@ func (s *rpcServer) Start() error {
s.opts.Address = ts.Addr() s.opts.Address = ts.Addr()
s.Unlock() s.Unlock()
bname := config.Broker.String()
// connect to the broker // connect to the broker
if err := config.Broker.Connect(); err != nil { if err := config.Broker.Connect(); err != nil {
if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
log.Errorf("Broker [%s] connect error: %v", bname, err)
}
return err return err
} }
bname := config.Broker.String() if logger.V(logger.DebugLevel, logger.DefaultLogger) {
log.Debugf("Broker [%s] Connected to %s", bname, config.Broker.Address())
log.Infof("Broker [%s] Connected to %s", bname, config.Broker.Address()) }
// use RegisterCheck func before register // use RegisterCheck func before register
if err = s.opts.RegisterCheck(s.opts.Context); err != nil { if err = s.opts.RegisterCheck(s.opts.Context); err != nil {
log.Errorf("Server %s-%s register check error: %s", config.Name, config.Id, err) if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
log.Errorf("Server %s-%s register check error: %s", config.Name, config.Id, err)
}
} else { } else {
// announce self to the world // announce self to the world
if err = s.Register(); err != nil { if err = s.Register(); err != nil {
log.Errorf("Server %s-%s register error: %s", config.Name, config.Id, err) if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
log.Errorf("Server %s-%s register error: %s", config.Name, config.Id, err)
}
} }
} }
@ -811,7 +853,9 @@ func (s *rpcServer) Start() error {
// check the error and backoff // check the error and backoff
default: default:
if err != nil { if err != nil {
log.Errorf("Accept error: %v", err) if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
log.Errorf("Accept error: %v", err)
}
time.Sleep(time.Second) time.Sleep(time.Second)
continue continue
} }
@ -844,17 +888,25 @@ func (s *rpcServer) Start() error {
s.RUnlock() s.RUnlock()
rerr := s.opts.RegisterCheck(s.opts.Context) rerr := s.opts.RegisterCheck(s.opts.Context)
if rerr != nil && registered { if rerr != nil && registered {
log.Errorf("Server %s-%s register check error: %s, deregister it", config.Name, config.Id, err) if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
log.Errorf("Server %s-%s register check error: %s, deregister it", config.Name, config.Id, err)
}
// deregister self in case of error // deregister self in case of error
if err := s.Deregister(); err != nil { if err := s.Deregister(); err != nil {
log.Errorf("Server %s-%s deregister error: %s", config.Name, config.Id, err) if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
log.Errorf("Server %s-%s deregister error: %s", config.Name, config.Id, err)
}
} }
} else if rerr != nil && !registered { } else if rerr != nil && !registered {
log.Errorf("Server %s-%s register check error: %s", config.Name, config.Id, err) if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
log.Errorf("Server %s-%s register check error: %s", config.Name, config.Id, err)
}
continue continue
} }
if err := s.Register(); err != nil { if err := s.Register(); err != nil {
log.Errorf("Server %s-%s register error: %s", config.Name, config.Id, err) if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
log.Errorf("Server %s-%s register error: %s", config.Name, config.Id, err)
}
} }
// wait for exit // wait for exit
case ch = <-s.exit: case ch = <-s.exit:
@ -870,7 +922,9 @@ func (s *rpcServer) Start() error {
if registered { if registered {
// deregister self // deregister self
if err := s.Deregister(); err != nil { if err := s.Deregister(); err != nil {
log.Errorf("Server %s-%s deregister error: %s", config.Name, config.Id, err) if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
log.Errorf("Server %s-%s deregister error: %s", config.Name, config.Id, err)
}
} }
} }
@ -886,9 +940,15 @@ func (s *rpcServer) Start() error {
// close transport listener // close transport listener
ch <- ts.Close() ch <- ts.Close()
log.Infof("Broker [%s] Disconnected from %s", bname, config.Broker.Address()) if logger.V(logger.DebugLevel, logger.DefaultLogger) {
log.Debugf("Broker [%s] Disconnected from %s", bname, config.Broker.Address())
}
// disconnect the broker // disconnect the broker
config.Broker.Disconnect() if err := config.Broker.Disconnect(); err != nil {
if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
log.Errorf("Broker [%s] Disconnect error: %v", bname, err)
}
}
// swap back address // swap back address
s.Lock() s.Lock()

View File

@ -35,18 +35,8 @@ var (
) )
func (c *clientWrapper) setHeaders(ctx context.Context) context.Context { func (c *clientWrapper) setHeaders(ctx context.Context) context.Context {
// copy metadata // don't overwrite keys
mda, _ := metadata.FromContext(ctx) return metadata.MergeContext(ctx, c.headers, false)
md := metadata.Copy(mda)
// set headers
for k, v := range c.headers {
if _, ok := md[k]; !ok {
md[k] = v
}
}
return metadata.NewContext(ctx, md)
} }
func (c *clientWrapper) Call(ctx context.Context, req client.Request, rsp interface{}, opts ...client.CallOption) error { func (c *clientWrapper) Call(ctx context.Context, req client.Request, rsp interface{}, opts ...client.CallOption) error {