unistack-org/micro
4
0
Fork 2
Code Issues 18 Pull Requests Actions Packages Projects Releases 324 Wiki Activity

cleanup client/selector/lookup (#1937)

Browse Source 
* cleanup client/selector/lookup

* add mdns router, remove registry from client

* fix roundtripper

* remove comment

* fix compile issue

* fix mucp test

* fix api router
This commit is contained in:
Asim Aslam
2020-08-17 22:44:45 +01:00
committed by GitHub
parent 7135787b78
commit 50ec6c748f
17 changed files with 398 additions and 418 deletions
Download Patch File Download Diff File Expand all files Collapse all files

129
client/grpc/grpc.go
View File

@@ -16,7 +16,6 @@ import (
raw "github.com/micro/go-micro/v3/codec/bytes"
"github.com/micro/go-micro/v3/errors"
"github.com/micro/go-micro/v3/metadata"
"github.com/micro/go-micro/v3/registry"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
@@ -70,7 +69,7 @@ func (g *grpcClient) secure(addr string) grpc.DialOption {
return grpc.WithInsecure()
}
func (g *grpcClient) call(ctx context.Context, node *registry.Node, req client.Request, rsp interface{}, opts client.CallOptions) error {
func (g *grpcClient) call(ctx context.Context, addr string, req client.Request, rsp interface{}, opts client.CallOptions) error {
var header map[string]string
header = make(map[string]string)
@@ -103,7 +102,7 @@ func (g *grpcClient) call(ctx context.Context, node *registry.Node, req client.R
grpcDialOptions := []grpc.DialOption{
grpc.WithTimeout(opts.DialTimeout),
g.secure(node.Address),
g.secure(addr),
grpc.WithDefaultCallOptions(
grpc.MaxCallRecvMsgSize(maxRecvMsgSize),
grpc.MaxCallSendMsgSize(maxSendMsgSize),
@@ -114,13 +113,13 @@ func (g *grpcClient) call(ctx context.Context, node *registry.Node, req client.R
grpcDialOptions = append(grpcDialOptions, opts...)
}
cc, err := g.pool.getConn(node.Address, grpcDialOptions...)
cc, err := g.pool.getConn(addr, grpcDialOptions...)
if err != nil {
return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err))
}
defer func() {
// defer execution of release
g.pool.release(node.Address, cc, grr)
g.pool.release(addr, cc, grr)
}()
ch := make(chan error, 1)
@@ -146,7 +145,7 @@ func (g *grpcClient) call(ctx context.Context, node *registry.Node, req client.R
return grr
}
func (g *grpcClient) stream(ctx context.Context, node *registry.Node, req client.Request, rsp interface{}, opts client.CallOptions) error {
func (g *grpcClient) stream(ctx context.Context, addr string, req client.Request, rsp interface{}, opts client.CallOptions) error {
var header map[string]string
if md, ok := metadata.FromContext(ctx); ok {
@@ -186,14 +185,14 @@ func (g *grpcClient) stream(ctx context.Context, node *registry.Node, req client
grpcDialOptions := []grpc.DialOption{
grpc.WithTimeout(opts.DialTimeout),
g.secure(node.Address),
g.secure(addr),
}
if opts := g.getGrpcDialOptions(); opts != nil {
grpcDialOptions = append(grpcDialOptions, opts...)
}
cc, err := grpc.DialContext(dialCtx, node.Address, grpcDialOptions...)
cc, err := grpc.DialContext(dialCtx, addr, grpcDialOptions...)
if err != nil {
return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err))
}
@@ -389,6 +388,34 @@ func (g *grpcClient) Call(ctx context.Context, req client.Request, rsp interface
gcall = callOpts.CallWrappers[i-1](gcall)
}
// use the router passed as a call option, or fallback to the rpc clients router
if callOpts.Router == nil {
callOpts.Router = g.opts.Router
}
if callOpts.Selector == nil {
callOpts.Selector = g.opts.Selector
}
// inject proxy address
// TODO: don't even bother using Lookup/Select in this case
if len(g.opts.Proxy) > 0 {
callOpts.Address = []string{g.opts.Proxy}
}
// lookup the route to send the reques to
// TODO apply any filtering here
routes, err := g.opts.Lookup(ctx, req, callOpts)
if err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
}
// balance the list of nodes
next, err := callOpts.Selector.Select(routes)
if err != nil {
return err
}
// return errors.New("go.micro.client", "request timeout", 408)
call := func(i int) error {
// call backoff first. Someone may want an initial start delay
@@ -402,36 +429,14 @@ func (g *grpcClient) Call(ctx context.Context, req client.Request, rsp interface
time.Sleep(t)
}
// use the router passed as a call option, or fallback to the rpc clients router
if callOpts.Router == nil {
callOpts.Router = g.opts.Router
}
// use the selector passed as a call option, or fallback to the rpc clients selector
if callOpts.Selector == nil {
callOpts.Selector = g.opts.Selector
}
// inject proxy address
if len(g.opts.Proxy) > 0 {
callOpts.Address = []string{g.opts.Proxy}
}
// lookup the route to send the reques to
route, err := client.LookupRoute(req, callOpts)
if err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
}
// pass a node to enable backwards compatability as changing the
// call func would be a breaking change.
// todo v3: change the call func to accept a route
node := &registry.Node{Address: route.Address}
// get the next node
node := next()
// make the call
err = gcall(ctx, node, req, rsp, callOpts)
// record the result of the call to inform future routing decisions
g.opts.Selector.Record(*route, err)
g.opts.Selector.Record(node, err)
// try and transform the error to a go-micro error
if verr, ok := err.(*errors.Error); ok {
@@ -498,6 +503,34 @@ func (g *grpcClient) Stream(ctx context.Context, req client.Request, opts ...cli
gstream = callOpts.CallWrappers[i-1](gstream)
}
// use the router passed as a call option, or fallback to the rpc clients router
if callOpts.Router == nil {
callOpts.Router = g.opts.Router
}
if callOpts.Selector == nil {
callOpts.Selector = g.opts.Selector
}
// inject proxy address
// TODO: don't even bother using Lookup/Select in this case
if len(g.opts.Proxy) > 0 {
callOpts.Address = []string{g.opts.Proxy}
}
// lookup the route to send the reques to
// TODO: move to internal lookup func
routes, err := g.opts.Lookup(ctx, req, callOpts)
if err != nil {
return nil, errors.InternalServerError("go.micro.client", err.Error())
}
// balance the list of nodes
next, err := callOpts.Selector.Select(routes)
if err != nil {
return nil, err
}
call := func(i int) (client.Stream, error) {
// call backoff first. Someone may want an initial start delay
t, err := callOpts.Backoff(ctx, req, i)
@@ -510,44 +543,22 @@ func (g *grpcClient) Stream(ctx context.Context, req client.Request, opts ...cli
time.Sleep(t)
}
// use the router passed as a call option, or fallback to the rpc clients router
if callOpts.Router == nil {
callOpts.Router = g.opts.Router
}
// use the selector passed as a call option, or fallback to the rpc clients selector
if callOpts.Selector == nil {
callOpts.Selector = g.opts.Selector
}
// inject proxy address
if len(g.opts.Proxy) > 0 {
callOpts.Address = []string{g.opts.Proxy}
}
// lookup the route to send the reques to
route, err := client.LookupRoute(req, callOpts)
if err != nil {
return nil, errors.InternalServerError("go.micro.client", err.Error())
}
// pass a node to enable backwards compatability as changing the
// call func would be a breaking change.
// todo v3: change the call func to accept a route
node := &registry.Node{Address: route.Address}
// get the next node
node := next()
// make the call
stream := &grpcStream{}
err = g.stream(ctx, node, req, stream, callOpts)
// record the result of the call to inform future routing decisions
g.opts.Selector.Record(*route, err)
g.opts.Selector.Record(node, err)
// try and transform the error to a go-micro error
if verr, ok := err.(*errors.Error); ok {
return nil, verr
}
g.opts.Selector.Record(*route, err)
g.opts.Selector.Record(node, err)
return stream, err
}

26
client/util.go → client/lookup.go
View File

@@ -1,21 +1,20 @@
package client
import (
"math/rand"
"context"
"github.com/micro/go-micro/v3/errors"
"github.com/micro/go-micro/v3/router"
"github.com/micro/go-micro/v3/selector"
)
// LookupFunc is used to lookup routes for a service
type LookupFunc func(context.Context, Request, CallOptions) ([]string, error)
// LookupRoute for a request using the router and then choose one using the selector
func LookupRoute(req Request, opts CallOptions) (*router.Route, error) {
func LookupRoute(ctx context.Context, req Request, opts CallOptions) ([]string, error) {
// check to see if an address was provided as a call option
if len(opts.Address) > 0 {
return &router.Route{
Service: req.Service(),
Address: opts.Address[rand.Int()%len(opts.Address)],
}, nil
return opts.Address, nil
}
// construct the router query
@@ -35,12 +34,11 @@ func LookupRoute(req Request, opts CallOptions) (*router.Route, error) {
return nil, errors.InternalServerError("go.micro.client", "error getting next %s node: %s", req.Service(), err.Error())
}
// select the route to use for the request
if route, err := opts.Selector.Select(routes, opts.SelectOptions...); err == selector.ErrNoneAvailable {
return nil, errors.InternalServerError("go.micro.client", "service %s: %s", req.Service(), err.Error())
} else if err != nil {
return nil, errors.InternalServerError("go.micro.client", "error getting next %s node: %s", req.Service(), err.Error())
} else {
return route, nil
var addrs []string
for _, route := range routes {
addrs = append(addrs, route.Address)
}
return addrs, nil
}

155
client/mucp/mucp.go
View File

@@ -14,17 +14,11 @@ import (
raw "github.com/micro/go-micro/v3/codec/bytes"
"github.com/micro/go-micro/v3/errors"
"github.com/micro/go-micro/v3/metadata"
"github.com/micro/go-micro/v3/registry"
"github.com/micro/go-micro/v3/transport"
"github.com/micro/go-micro/v3/util/buf"
"github.com/micro/go-micro/v3/util/pool"
)
// NewClient returns a new micro client interface
func NewClient(opts ...client.Option) client.Client {
return newClient(opts...)
}
type rpcClient struct {
once atomic.Value
opts client.Options
@@ -32,7 +26,8 @@ type rpcClient struct {
seq uint64
}
func newClient(opt ...client.Option) client.Client {
// NewClient returns a new micro client interface
func NewClient(opt ...client.Option) client.Client {
opts := client.NewOptions(opt...)
p := pool.NewPool(
@@ -68,7 +63,7 @@ func (r *rpcClient) newCodec(contentType string) (codec.NewCodec, error) {
return nil, fmt.Errorf("Unsupported Content-Type: %s", contentType)
}
func (r *rpcClient) call(ctx context.Context, node *registry.Node, req client.Request, resp interface{}, opts client.CallOptions) error {
func (r *rpcClient) call(ctx context.Context, addr string, req client.Request, resp interface{}, opts client.CallOptions) error {
msg := &transport.Message{
Header: make(map[string]string),
}
@@ -92,16 +87,9 @@ func (r *rpcClient) call(ctx context.Context, node *registry.Node, req client.Re
// set the accept header
msg.Header["Accept"] = req.ContentType()
// setup old protocol
cf := setupProtocol(msg, node)
// no codec specified
if cf == nil {
var err error
cf, err = r.newCodec(req.ContentType())
if err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
}
cf, err := r.newCodec(req.ContentType())
if err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
}
dOpts := []transport.DialOption{
@@ -112,7 +100,7 @@ func (r *rpcClient) call(ctx context.Context, node *registry.Node, req client.Re
dOpts = append(dOpts, transport.WithTimeout(opts.DialTimeout))
}
c, err := r.pool.Get(node.Address, dOpts...)
c, err := r.pool.Get(addr, dOpts...)
if err != nil {
return errors.InternalServerError("go.micro.client", "connection error: %v", err)
}
@@ -185,7 +173,7 @@ func (r *rpcClient) call(ctx context.Context, node *registry.Node, req client.Re
return nil
}
func (r *rpcClient) stream(ctx context.Context, node *registry.Node, req client.Request, opts client.CallOptions) (client.Stream, error) {
func (r *rpcClient) stream(ctx context.Context, addr string, req client.Request, opts client.CallOptions) (client.Stream, error) {
msg := &transport.Message{
Header: make(map[string]string),
}
@@ -206,16 +194,9 @@ func (r *rpcClient) stream(ctx context.Context, node *registry.Node, req client.
// set the accept header
msg.Header["Accept"] = req.ContentType()
// set old codecs
cf := setupProtocol(msg, node)
// no codec specified
if cf == nil {
var err error
cf, err = r.newCodec(req.ContentType())
if err != nil {
return nil, errors.InternalServerError("go.micro.client", err.Error())
}
cf, err := r.newCodec(req.ContentType())
if err != nil {
return nil, errors.InternalServerError("go.micro.client", err.Error())
}
dOpts := []transport.DialOption{
@@ -226,7 +207,7 @@ func (r *rpcClient) stream(ctx context.Context, node *registry.Node, req client.
dOpts = append(dOpts, transport.WithTimeout(opts.DialTimeout))
}
c, err := r.opts.Transport.Dial(node.Address, dOpts...)
c, err := r.opts.Transport.Dial(addr, dOpts...)
if err != nil {
return nil, errors.InternalServerError("go.micro.client", "connection error: %v", err)
}
@@ -356,6 +337,34 @@ func (r *rpcClient) Call(ctx context.Context, request client.Request, response i
rcall = callOpts.CallWrappers[i-1](rcall)
}
// use the router passed as a call option, or fallback to the rpc clients router
if callOpts.Router == nil {
callOpts.Router = r.opts.Router
}
if callOpts.Selector == nil {
callOpts.Selector = r.opts.Selector
}
// inject proxy address
// TODO: don't even bother using Lookup/Select in this case
if len(r.opts.Proxy) > 0 {
callOpts.Address = []string{r.opts.Proxy}
}
// lookup the route to send the reques to
// TODO apply any filtering here
routes, err := r.opts.Lookup(ctx, request, callOpts)
if err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
}
// balance the list of nodes
next, err := callOpts.Selector.Select(routes)
if err != nil {
return err
}
// return errors.New("go.micro.client", "request timeout", 408)
call := func(i int) error {
// call backoff first. Someone may want an initial start delay
@@ -369,36 +378,14 @@ func (r *rpcClient) Call(ctx context.Context, request client.Request, response i
time.Sleep(t)
}
// use the router passed as a call option, or fallback to the rpc clients router
if callOpts.Router == nil {
callOpts.Router = r.opts.Router
}
// use the selector passed as a call option, or fallback to the rpc clients selector
if callOpts.Selector == nil {
callOpts.Selector = r.opts.Selector
}
// inject proxy address
if len(r.opts.Proxy) > 0 {
callOpts.Address = []string{r.opts.Proxy}
}
// lookup the route to send the request via
route, err := client.LookupRoute(request, callOpts)
if err != nil {
return err
}
// pass a node to enable backwards comparability as changing the
// call func would be a breaking change.
// todo v3: change the call func to accept a route
node := &registry.Node{Address: route.Address, Metadata: route.Metadata}
// get the next node
node := next()
// make the call
err = rcall(ctx, node, request, response, callOpts)
// record the result of the call to inform future routing decisions
r.opts.Selector.Record(*route, err)
r.opts.Selector.Record(node, err)
return err
}
@@ -458,6 +445,34 @@ func (r *rpcClient) Stream(ctx context.Context, request client.Request, opts ...
default:
}
// use the router passed as a call option, or fallback to the rpc clients router
if callOpts.Router == nil {
callOpts.Router = r.opts.Router
}
if callOpts.Selector == nil {
callOpts.Selector = r.opts.Selector
}
// inject proxy address
// TODO: don't even bother using Lookup/Select in this case
if len(r.opts.Proxy) > 0 {
callOpts.Address = []string{r.opts.Proxy}
}
// lookup the route to send the reques to
// TODO apply any filtering here
routes, err := r.opts.Lookup(ctx, request, callOpts)
if err != nil {
return nil, errors.InternalServerError("go.micro.client", err.Error())
}
// balance the list of nodes
next, err := callOpts.Selector.Select(routes)
if err != nil {
return nil, err
}
call := func(i int) (client.Stream, error) {
// call backoff first. Someone may want an initial start delay
t, err := callOpts.Backoff(ctx, request, i)
@@ -470,36 +485,14 @@ func (r *rpcClient) Stream(ctx context.Context, request client.Request, opts ...
time.Sleep(t)
}
// use the router passed as a call option, or fallback to the rpc clients router
if callOpts.Router == nil {
callOpts.Router = r.opts.Router
}
// use the selector passed as a call option, or fallback to the rpc clients selector
if callOpts.Selector == nil {
callOpts.Selector = r.opts.Selector
}
// inject proxy address
if len(r.opts.Proxy) > 0 {
callOpts.Address = []string{r.opts.Proxy}
}
// lookup the route to send the request via
route, err := client.LookupRoute(request, callOpts)
if err != nil {
return nil, err
}
// pass a node to enable backwards compatability as changing the
// call func would be a breaking change.
// todo v3: change the call func to accept a route
node := &registry.Node{Address: route.Address, Metadata: route.Metadata}
// get the next node
node := next()
// perform the call
stream, err := r.stream(ctx, node, request, callOpts)
// record the result of the call to inform future routing decisions
r.opts.Selector.Record(*route, err)
r.opts.Selector.Record(node, err)
return stream, err
}

39
client/mucp/mucp_test.go
View File

@@ -9,10 +9,13 @@ import (
"github.com/micro/go-micro/v3/errors"
"github.com/micro/go-micro/v3/registry"
"github.com/micro/go-micro/v3/registry/memory"
"github.com/micro/go-micro/v3/router"
regRouter "github.com/micro/go-micro/v3/router/registry"
)
func newTestRegistry() registry.Registry {
return memory.NewRegistry(memory.Services(testData))
func newTestRouter() router.Router {
reg := memory.NewRegistry(memory.Services(testData))
return regRouter.NewRouter(router.Registry(reg))
}
func TestCallAddress(t *testing.T) {
@@ -22,7 +25,7 @@ func TestCallAddress(t *testing.T) {
address := "10.1.10.1:8080"
wrap := func(cf client.CallFunc) client.CallFunc {
return func(ctx context.Context, node *registry.Node, req client.Request, rsp interface{}, opts client.CallOptions) error {
return func(ctx context.Context, node string, req client.Request, rsp interface{}, opts client.CallOptions) error {
called = true
if req.Service() != service {
@@ -33,8 +36,8 @@ func TestCallAddress(t *testing.T) {
return fmt.Errorf("expected service: %s got %s", endpoint, req.Endpoint())
}
if node.Address != address {
return fmt.Errorf("expected address: %s got %s", address, node.Address)
if node != address {
return fmt.Errorf("expected address: %s got %s", address, node)
}
// don't do the call
@@ -42,9 +45,10 @@ func TestCallAddress(t *testing.T) {
}
}
r := newTestRegistry()
r := newTestRouter()
c := NewClient(
client.Registry(r),
client.Router(r),
client.WrapCall(wrap),
)
@@ -69,7 +73,7 @@ func TestCallRetry(t *testing.T) {
var called int
wrap := func(cf client.CallFunc) client.CallFunc {
return func(ctx context.Context, node *registry.Node, req client.Request, rsp interface{}, opts client.CallOptions) error {
return func(ctx context.Context, node string, req client.Request, rsp interface{}, opts client.CallOptions) error {
called++
if called == 1 {
return errors.InternalServerError("test.error", "retry request")
@@ -80,9 +84,9 @@ func TestCallRetry(t *testing.T) {
}
}
r := newTestRegistry()
r := newTestRouter()
c := NewClient(
client.Registry(r),
client.Router(r),
client.WrapCall(wrap),
)
@@ -107,7 +111,7 @@ func TestCallWrapper(t *testing.T) {
address := "10.1.10.1:8080"
wrap := func(cf client.CallFunc) client.CallFunc {
return func(ctx context.Context, node *registry.Node, req client.Request, rsp interface{}, opts client.CallOptions) error {
return func(ctx context.Context, node string, req client.Request, rsp interface{}, opts client.CallOptions) error {
called = true
if req.Service() != service {
@@ -118,8 +122,8 @@ func TestCallWrapper(t *testing.T) {
return fmt.Errorf("expected service: %s got %s", endpoint, req.Endpoint())
}
if node.Address != address {
return fmt.Errorf("expected address: %s got %s", address, node.Address)
if node != address {
return fmt.Errorf("expected address: %s got %s", address, node)
}
// don't do the call
@@ -127,22 +131,19 @@ func TestCallWrapper(t *testing.T) {
}
}
r := newTestRegistry()
r := newTestRouter()
c := NewClient(
client.Registry(r),
client.Router(r),
client.WrapCall(wrap),
)
r.Register(&registry.Service{
r.Options().Registry.Register(&registry.Service{
Name: service,
Version: "latest",
Nodes: []*registry.Node{
{
Id: id,
Address: address,
Metadata: map[string]string{
"protocol": "mucp",
},
},
},
})

22
client/options.go
View File

@@ -7,10 +7,10 @@ import (
"github.com/micro/go-micro/v3/broker"
"github.com/micro/go-micro/v3/broker/http"
"github.com/micro/go-micro/v3/codec"
"github.com/micro/go-micro/v3/registry"
"github.com/micro/go-micro/v3/router"
regRouter "github.com/micro/go-micro/v3/router/registry"
"github.com/micro/go-micro/v3/selector"
"github.com/micro/go-micro/v3/selector/random"
"github.com/micro/go-micro/v3/transport"
thttp "github.com/micro/go-micro/v3/transport/http"
)
@@ -28,6 +28,9 @@ type Options struct {
Selector selector.Selector
Transport transport.Transport
// Lookup used for looking up routes
Lookup LookupFunc
// Connection Pool
PoolSize int
PoolTTL time.Duration
@@ -116,11 +119,12 @@ func NewOptions(options ...Option) Options {
RequestTimeout: DefaultRequestTimeout,
DialTimeout: transport.DefaultDialTimeout,
},
Lookup: LookupRoute,
PoolSize: DefaultPoolSize,
PoolTTL: DefaultPoolTTL,
Broker: http.NewBroker(),
Router: regRouter.NewRouter(),
Selector: selector.DefaultSelector,
Selector: random.NewSelector(),
Transport: thttp.NewTransport(),
}
@@ -216,6 +220,13 @@ func Backoff(fn BackoffFunc) Option {
}
}
// Lookup sets the lookup function to use for resolving service names
func Lookup(l LookupFunc) Option {
return func(o *Options) {
o.Lookup = l
}
}
// Number of retries when making the request.
// Should this be a Call Option?
func Retries(i int) Option {
@@ -231,13 +242,6 @@ func Retry(fn RetryFunc) Option {
}
}
// Registry sets the routers registry
func Registry(r registry.Registry) Option {
return func(o *Options) {
o.Router.Init(router.Registry(r))
}
}
// The request timeout.
// Should this be a Call Option?
func RequestTimeout(d time.Duration) Option {

4
client/wrapper.go
View File

@@ -2,12 +2,10 @@ package client
import (
"context"
"github.com/micro/go-micro/v3/registry"
)
// CallFunc represents the individual call func
type CallFunc func(ctx context.Context, node *registry.Node, req Request, rsp interface{}, opts CallOptions) error
type CallFunc func(ctx context.Context, addr string, req Request, rsp interface{}, opts CallOptions) error
// CallWrapper is a low level wrapper for the CallFunc
type CallWrapper func(CallFunc) CallFunc