diff --git a/client/grpc/grpc.go b/client/grpc/grpc.go index 3430fbfc..df4163c7 100644 --- a/client/grpc/grpc.go +++ b/client/grpc/grpc.go @@ -59,14 +59,14 @@ func (g *grpcClient) next(request client.Request, opts client.CallOptions) (sele // get proxy address if prx := os.Getenv("MICRO_PROXY_ADDRESS"); len(prx) > 0 { - opts.Address = prx + opts.Address = []string{prx} } // return remote address if len(opts.Address) > 0 { return func() (*registry.Node, error) { return ®istry.Node{ - Address: opts.Address, + Address: opts.Address[0], }, nil }, nil } diff --git a/client/options.go b/client/options.go index 5a0f4100..9f363d74 100644 --- a/client/options.go +++ b/client/options.go @@ -43,8 +43,8 @@ type Options struct { type CallOptions struct { SelectOptions []selector.SelectOption - // Address of remote host - Address string + // Address of remote hosts + Address []string // Backoff func Backoff BackoffFunc // Check if retriable func @@ -245,8 +245,8 @@ func WithExchange(e string) PublishOption { } } -// WithAddress sets the remote address to use rather than using service discovery -func WithAddress(a string) CallOption { +// WithAddress sets the remote addresses to use rather than using service discovery +func WithAddress(a ...string) CallOption { return func(o *CallOptions) { o.Address = a } diff --git a/client/rpc_client.go b/client/rpc_client.go index 36b3dcf7..c724435a 100644 --- a/client/rpc_client.go +++ b/client/rpc_client.go @@ -283,29 +283,36 @@ func (r *rpcClient) next(request Request, opts CallOptions) (selector.Next, erro // get proxy address if prx := os.Getenv("MICRO_PROXY_ADDRESS"); len(prx) > 0 { - opts.Address = prx + opts.Address = []string{prx} } // return remote address if len(opts.Address) > 0 { - address := opts.Address - port := 0 + var nodes []*registry.Node - host, sport, err := net.SplitHostPort(opts.Address) - if err == nil { - address = host - port, _ = strconv.Atoi(sport) - } + for _, addr := range opts.Address { + address := addr + port := 0 - return func() (*registry.Node, error) { - return ®istry.Node{ + host, sport, err := net.SplitHostPort(addr) + if err == nil { + address = host + port, _ = strconv.Atoi(sport) + } + + nodes = append(nodes, ®istry.Node{ Address: address, Port: port, // Set the protocol Metadata: map[string]string{ "protocol": "mucp", }, - }, nil + }) + } + + // crude return method + return func() (*registry.Node, error) { + return nodes[time.Now().Unix()%int64(len(nodes))], nil }, nil } diff --git a/client/selector/router/router.go b/client/selector/router/router.go new file mode 100644 index 00000000..452c9b89 --- /dev/null +++ b/client/selector/router/router.go @@ -0,0 +1,287 @@ +// Package router is a network/router selector +package router + +import ( + "context" + "fmt" + "net" + "os" + "sort" + "strconv" + "sync" + + "github.com/micro/go-micro/client" + "github.com/micro/go-micro/client/selector" + "github.com/micro/go-micro/network/router" + pb "github.com/micro/go-micro/network/router/proto" + "github.com/micro/go-micro/registry" +) + +type routerSelector struct { + opts selector.Options + + // the router + r router.Router + + // the client we have + c client.Client + + // the client for the remote router + rs pb.RouterService + + // name of the router + name string + + // address of the remote router + addr string + + // whether to use the remote router + remote bool +} + +type clientKey struct{} +type routerKey struct{} + +// getRoutes returns the routes whether they are remote or local +func (r *routerSelector) getRoutes(service string) ([]router.Route, error) { + if !r.remote { + // lookup router for routes for the service + return r.r.Table().Lookup(router.NewQuery( + router.QueryDestination(service), + )) + } + + // lookup the remote router + + var addrs []string + + // set the remote address if specified + if len(r.addr) > 0 { + addrs = append(addrs, r.addr) + } else { + // we have a name so we need to check the registry + services, err := r.c.Options().Registry.GetService(r.name) + if err != nil { + return nil, err + } + + for _, service := range services { + for _, node := range service.Nodes { + addr := node.Address + if node.Port > 0 { + addr = fmt.Sprintf("%s:%d", node.Address, node.Port) + } + addrs = append(addrs, addr) + } + } + } + + // no router addresses available + if len(addrs) == 0 { + return nil, selector.ErrNoneAvailable + } + + var pbRoutes *pb.LookupResponse + var err error + + // TODO: implement backoff and retries + for _, addr := range addrs { + // call the router + pbRoutes, err = r.rs.Lookup(context.Background(), &pb.LookupRequest{ + Query: &pb.Query{ + Destination: service, + }, + }, client.WithAddress(addr)) + if err != nil { + continue + } + break + } + + // errored out + if err != nil { + return nil, err + } + + // no routes + if pbRoutes == nil { + return nil, selector.ErrNoneAvailable + } + + var routes []router.Route + + // convert from pb to []*router.Route + for _, r := range pbRoutes.Routes { + routes = append(routes, router.Route{ + Destination: r.Destination, + Gateway: r.Gateway, + Router: r.Router, + Network: r.Network, + Metric: int(r.Metric), + }) + } + + return routes, nil +} + +func (r *routerSelector) Init(opts ...selector.Option) error { + // no op + return nil +} + +func (r *routerSelector) Options() selector.Options { + return r.opts +} + +func (r *routerSelector) Select(service string, opts ...selector.SelectOption) (selector.Next, error) { + // TODO: pull routes asynchronously and cache + routes, err := r.getRoutes(service) + if err != nil { + return nil, err + } + + // no routes return not found error + if len(routes) == 0 { + return nil, selector.ErrNotFound + } + + // TODO: apply filters by pseudo constructing service + + // sort the routes based on metric + sort.Slice(routes, func(i, j int) bool { + return routes[i].Metric < routes[j].Metric + }) + + // roundrobin assuming routes are in metric preference order + var i int + var mtx sync.Mutex + + return func() (*registry.Node, error) { + // get index and increment counter with every call to next + mtx.Lock() + idx := i + i++ + mtx.Unlock() + + // get route based on idx + route := routes[idx%len(routes)] + + // defaults to gateway and no port + address := route.Gateway + port := 0 + + // check if its host:port + host, pr, err := net.SplitHostPort(address) + if err == nil { + pp, _ := strconv.Atoi(pr) + // set port + port = pp + // set address + address = host + } + + // return as a node + return ®istry.Node{ + // TODO: add id and metadata if we can + Address: address, + Port: port, + }, nil + }, nil +} + +func (r *routerSelector) Mark(service string, node *registry.Node, err error) { + // TODO: pass back metrics or information to the router + return +} + +func (r *routerSelector) Reset(service string) { + // TODO: reset the metrics or information at the router + return +} + +func (r *routerSelector) Close() error { + // stop the router advertisements + return r.r.Stop() +} + +func (r *routerSelector) String() string { + return "router" +} + +// NewSelector returns a new router based selector +func NewSelector(opts ...selector.Option) selector.Selector { + options := selector.Options{ + Context: context.Background(), + } + + for _, o := range opts { + o(&options) + } + + // set default registry if not set + if options.Registry == nil { + options.Registry = registry.DefaultRegistry + } + + // try get router from the context + r, ok := options.Context.Value(routerKey{}).(router.Router) + if !ok { + // TODO: Use router.DefaultRouter? + r = router.NewRouter( + router.Registry(options.Registry), + ) + } + + // try get client from the context + c, ok := options.Context.Value(clientKey{}).(client.Client) + if !ok { + c = client.DefaultClient + } + + // get the router from env vars if its a remote service + remote := true + routerName := os.Getenv("MICRO_ROUTER") + routerAddress := os.Getenv("MICRO_ROUTER_ADDRESS") + + // start the router advertisements if we're running it locally + if len(routerName) == 0 && len(routerAddress) == 0 { + go r.Advertise() + remote = false + } + + return &routerSelector{ + opts: options, + // set the internal router + r: r, + // set the client + c: c, + // set the router client + rs: pb.NewRouterService(routerName, c), + // name of the router + name: routerName, + // address of router + addr: routerAddress, + // let ourselves know to use the remote router + remote: remote, + } +} + +// WithClient sets the client for the request +func WithClient(c client.Client) selector.Option { + return func(o *selector.Options) { + if o.Context == nil { + o.Context = context.Background() + } + o.Context = context.WithValue(o.Context, clientKey{}, c) + } +} + +// WithRouter sets the router as an option +func WithRouter(r router.Router) selector.Option { + return func(o *selector.Options) { + if o.Context == nil { + o.Context = context.Background() + } + o.Context = context.WithValue(o.Context, routerKey{}, r) + } +} diff --git a/config/cmd/cmd.go b/config/cmd/cmd.go index 2e7e2a6e..8d7c3c03 100644 --- a/config/cmd/cmd.go +++ b/config/cmd/cmd.go @@ -34,6 +34,7 @@ import ( // selectors "github.com/micro/go-micro/client/selector" "github.com/micro/go-micro/client/selector/dns" + "github.com/micro/go-micro/client/selector/router" "github.com/micro/go-micro/client/selector/static" // transports @@ -196,6 +197,7 @@ var ( "default": selector.NewSelector, "dns": dns.NewSelector, "cache": selector.NewSelector, + "router": router.NewSelector, "static": static.NewSelector, } diff --git a/network/default.go b/network/default.go new file mode 100644 index 00000000..8ee82b04 --- /dev/null +++ b/network/default.go @@ -0,0 +1,95 @@ +package network + +import ( + "sync" + + "github.com/micro/go-micro/config/options" + "github.com/micro/go-micro/network/proxy" + "github.com/micro/go-micro/network/router" +) + +type network struct { + options.Options + + // router + r router.Router + + // proxy + p proxy.Proxy + + // id of this network + id string + + // links maintained for this network + mtx sync.RWMutex + links []Link +} + +type node struct { + *network + + // address of this node + address string +} + +type link struct { + // the embedded node + *node + + // length and weight of the link + mtx sync.RWMutex + length int + weight int +} + +// network methods + +func (n *network) Id() string { + return n.id +} + +func (n *network) Connect() (Node, error) { + return nil, nil +} + +func (n *network) Peer(Network) (Link, error) { + return nil, nil +} + +func (n *network) Links() ([]Link, error) { + n.mtx.RLock() + defer n.mtx.RUnlock() + return n.links, nil +} + +// node methods + +func (n *node) Address() string { + return n.address +} + +func (n *node) Close() error { + return nil +} + +func (n *node) Accept() (*Message, error) { + return nil, nil +} + +func (n *node) Send(*Message) error { + return nil +} + +// link methods + +func (l *link) Length() int { + l.mtx.RLock() + defer l.mtx.RUnlock() + return l.length +} + +func (l *link) Weight() int { + l.mtx.RLock() + defer l.mtx.RUnlock() + return l.weight +} diff --git a/network/network.go b/network/network.go index 29d6653d..098c0807 100644 --- a/network/network.go +++ b/network/network.go @@ -51,6 +51,25 @@ type Message struct { } var ( - // TODO: set default network - DefaultNetwork Network + // The default network ID is local + DefaultNetworkId = "local" + + // just the standard network element + DefaultNetwork = NewNetwork() ) + +// NewNetwork returns a new network +func NewNetwork(opts ...options.Option) Network { + options := options.NewOptions(opts...) + + // get router + + // get proxy + + return &network{ + Options: options, + // fill the blanks + // router: r, + // proxy: p, + } +} diff --git a/network/proxy/mucp/mucp.go b/network/proxy/mucp/mucp.go index 9e8afd39..2d52058a 100644 --- a/network/proxy/mucp/mucp.go +++ b/network/proxy/mucp/mucp.go @@ -7,10 +7,12 @@ import ( "strings" "github.com/micro/go-micro/client" + rselect "github.com/micro/go-micro/client/selector/router" "github.com/micro/go-micro/codec" "github.com/micro/go-micro/codec/bytes" "github.com/micro/go-micro/config/options" "github.com/micro/go-micro/network/proxy" + "github.com/micro/go-micro/network/router" "github.com/micro/go-micro/server" ) @@ -162,5 +164,18 @@ func NewProxy(opts ...options.Option) proxy.Proxy { p.Client = c.(client.Client) } + // get router + r, ok := p.Options.Values().Get("proxy.router") + if ok { + // set the router in the client + p.Client.Init( + // pass new selector as an option to the client + client.Selector(rselect.NewSelector( + // set the router in the selector + rselect.WithRouter(r.(router.Router)), + )), + ) + } + return p } diff --git a/network/proxy/proxy.go b/network/proxy/proxy.go index bb8fcc3b..22d485e8 100644 --- a/network/proxy/proxy.go +++ b/network/proxy/proxy.go @@ -6,6 +6,7 @@ import ( "github.com/micro/go-micro/client" "github.com/micro/go-micro/config/options" + "github.com/micro/go-micro/network/router" "github.com/micro/go-micro/server" ) @@ -29,3 +30,8 @@ func WithEndpoint(e string) options.Option { func WithClient(c client.Client) options.Option { return options.WithValue("proxy.client", c) } + +// WithRouter specifies the router to use +func WithRouter(r router.Router) options.Option { + return options.WithValue("proxy.router", r) +} diff --git a/network/router/default_router.go b/network/router/default_router.go index d600d1da..b342666e 100644 --- a/network/router/default_router.go +++ b/network/router/default_router.go @@ -2,25 +2,14 @@ package router import ( "fmt" - "net" - "strconv" "strings" "sync" - "time" "github.com/micro/go-micro/registry" "github.com/olekukonko/tablewriter" ) -var ( - // AdvertiseTick defines how often in seconds do we scal the local registry - // to advertise the local services to the network registry - AdvertiseTick = 5 * time.Second - // AdvertiseTTL defines network registry TTL in seconds - // NOTE: this is a rather arbitrary picked value subject to change - AdvertiseTTL = 120 * time.Second -) - +// router provides default router implementation type router struct { opts Options exit chan struct{} @@ -29,6 +18,9 @@ type router struct { // newRouter creates new router and returns it func newRouter(opts ...Option) Router { + // TODO: we need to add default GW entry here + // Should default GW be part of router options? + // get default options options := DefaultOptions() @@ -74,41 +66,24 @@ func (r *router) Address() string { // Network returns the address router advertises to the network func (r *router) Network() string { - return r.opts.Advertise + return r.opts.Network } -// Advertise advertises the router routes to the network. -// Advertise is a blocking function. It launches multiple goroutines that watch -// service registries and advertise the router routes to other routers in the network. +// Advertise advertises the routes to the network. It is a blocking function. // It returns error if any of the launched goroutines fail with error. func (r *router) Advertise() error { // add local service routes into the routing table - if err := r.addServiceRoutes(r.opts.Registry, DefaultLocalMetric); err != nil { - return fmt.Errorf("failed adding routes for local services: %v", err) - } - - // add network service routes into the routing table - if err := r.addServiceRoutes(r.opts.Network, DefaultNetworkMetric); err != nil { - return fmt.Errorf("failed adding routes for network services: %v", err) - } - - node, err := r.parseToNode() - if err != nil { - return fmt.Errorf("failed to parse router into service node: %v", err) + if err := r.addServiceRoutes(r.opts.Registry, "local", DefaultLocalMetric); err != nil { + return fmt.Errorf("failed adding routes: %v", err) } localWatcher, err := r.opts.Registry.Watch() if err != nil { - return fmt.Errorf("failed to create local registry watcher: %v", err) - } - - networkWatcher, err := r.opts.Network.Watch() - if err != nil { - return fmt.Errorf("failed to create network registry watcher: %v", err) + return fmt.Errorf("failed to create registry watcher: %v", err) } // error channel collecting goroutine errors - errChan := make(chan error, 3) + errChan := make(chan error, 1) r.wg.Add(1) go func() { @@ -117,102 +92,42 @@ func (r *router) Advertise() error { errChan <- r.manageServiceRoutes(localWatcher, DefaultLocalMetric) }() - r.wg.Add(1) - go func() { - defer r.wg.Done() - // watch network registry and register routes in routine table - errChan <- r.manageServiceRoutes(networkWatcher, DefaultNetworkMetric) - }() - - r.wg.Add(1) - go func() { - defer r.wg.Done() - // watch local registry and advertise local service to the network - errChan <- r.advertiseToNetwork(node) - }() - return <-errChan } // addServiceRoutes adds all services in given registry to the routing table. -// NOTE: this is a one-off operation done when bootstrapping the routing table of the new router. -// It returns error if either the services could not be listed or if the routes could not be added to the routing table. -func (r *router) addServiceRoutes(reg registry.Registry, metric int) error { +// NOTE: this is a one-off operation done when bootstrapping the routing table +// It returns error if either the services failed to be listed or +// if the routes could not be added to the routing table. +func (r *router) addServiceRoutes(reg registry.Registry, network string, metric int) error { services, err := reg.ListServices() if err != nil { return fmt.Errorf("failed to list services: %v", err) } + // add each service node as a separate route; for _, service := range services { - route := Route{ - Destination: service.Name, - Router: r, - Network: r.opts.Advertise, - Metric: metric, - } - if err := r.opts.Table.Add(route); err != nil && err != ErrDuplicateRoute { - return fmt.Errorf("error adding route for service: %s", service.Name) + for _, node := range service.Nodes { + gw := node.Address + if node.Port > 0 { + gw = fmt.Sprintf("%s:%d", node.Address, node.Port) + } + route := Route{ + Destination: service.Name, + Gateway: gw, + Router: r.opts.Address, + Network: r.opts.Network, + Metric: metric, + } + if err := r.opts.Table.Add(route); err != nil && err != ErrDuplicateRoute { + return fmt.Errorf("error adding route for service %s: %s", service.Name, err) + } } } return nil } -// parseToNode parses router into registry.Node and returns the result. -// It returns error if the router network address could not be parsed into host and port. -func (r *router) parseToNode() (*registry.Node, error) { - // split router address to host and port part - addr, portStr, err := net.SplitHostPort(r.opts.Advertise) - if err != nil { - return nil, fmt.Errorf("could not parse router address: %v", err) - } - - // try to parse network port into integer - port, err := strconv.Atoi(portStr) - if err != nil { - return nil, fmt.Errorf("could not parse router network address: %v", err) - } - - node := ®istry.Node{ - Id: r.opts.ID, - Address: addr, - Port: port, - } - - return node, nil -} - -// advertiseToNetwork periodically scans local registry and registers (i.e. advertises) all the local services in the network registry. -// It returns error if either the local services failed to be listed or if it fails to register local service in network registry. -func (r *router) advertiseToNetwork(node *registry.Node) error { - // ticker to periodically scan the local registry - ticker := time.NewTicker(AdvertiseTick) - - for { - select { - case <-r.exit: - return nil - case <-ticker.C: - // list all local services - services, err := r.opts.Registry.ListServices() - if err != nil { - return fmt.Errorf("failed to list local services: %v", err) - } - // loop through all registered local services and register them in the network registry - for _, service := range services { - svc := ®istry.Service{ - Name: service.Name, - Nodes: []*registry.Node{node}, - } - // register the local service in the network registry - if err := r.opts.Network.Register(svc, registry.RegisterTTL(AdvertiseTTL)); err != nil { - return fmt.Errorf("failed to register service %s in network registry: %v", svc.Name, err) - } - } - } - } -} - // manageServiceRoutes watches services in given registry and updates the routing table accordingly. // It returns error if the service registry watcher has stopped or if the routing table failed to be updated. func (r *router) manageServiceRoutes(w registry.Watcher, metric int) error { @@ -240,25 +155,21 @@ func (r *router) manageServiceRoutes(w registry.Watcher, metric int) error { route := Route{ Destination: res.Service.Name, - Router: r, - Network: r.opts.Advertise, + Router: r.opts.Address, + Network: r.opts.Network, Metric: metric, } switch res.Action { case "create": - if len(res.Service.Nodes) > 0 { - // only return error if the route is not duplicate, but something else has failed - if err := r.opts.Table.Add(route); err != nil && err != ErrDuplicateRoute { - return fmt.Errorf("failed to add route for service: %v", res.Service.Name) - } + // only return error if the route is not duplicate, but something else has failed + if err := r.opts.Table.Add(route); err != nil && err != ErrDuplicateRoute { + return fmt.Errorf("failed to add route for service %v: %s", res.Service.Name, err) } case "delete": - if len(res.Service.Nodes) < 1 { - // only return error if the route is present in the table, but something else has failed - if err := r.opts.Table.Delete(route); err != nil && err != ErrRouteNotFound { - return fmt.Errorf("failed to delete route for service: %v", res.Service.Name) - } + // only return error if the route is not in the table, but something else has failed + if err := r.opts.Table.Delete(route); err != nil && err != ErrRouteNotFound { + return fmt.Errorf("failed to delete route for service %v: %s", res.Service.Name, err) } } } @@ -274,30 +185,6 @@ func (r *router) Stop() error { // wait for all goroutines to finish r.wg.Wait() - // NOTE: we need a more efficient way of doing this e.g. network routes - // should ideally be autodeleted when the router stops gossiping - query := NewQuery(QueryRouter(r), QueryNetwork(r.opts.Advertise)) - routes, err := r.opts.Table.Lookup(query) - if err != nil && err != ErrRouteNotFound { - return fmt.Errorf("failed to lookup routes for router %s: %v", r.opts.ID, err) - } - - // parse router to registry.Node - node, err := r.parseToNode() - if err != nil { - return fmt.Errorf("failed to parse router into service node: %v", err) - } - - for _, route := range routes { - service := ®istry.Service{ - Name: route.Destination, - Nodes: []*registry.Node{node}, - } - if err := r.opts.Network.Deregister(service); err != nil { - return fmt.Errorf("failed to deregister service %s from network registry: %v", service.Name, err) - } - } - return nil } @@ -311,7 +198,7 @@ func (r *router) String() string { data := []string{ r.opts.ID, r.opts.Address, - r.opts.Advertise, + r.opts.Network, fmt.Sprintf("%d", r.opts.Table.Size()), } table.Append(data) diff --git a/network/router/default_table.go b/network/router/default_table.go index 04e8f6db..196ca4bf 100644 --- a/network/router/default_table.go +++ b/network/router/default_table.go @@ -171,7 +171,7 @@ func (t *table) Lookup(q Query) ([]Route, error) { } for _, route := range routes { if q.Options().Network == "*" || q.Options().Network == route.Network { - if q.Options().Router.ID() == "*" || q.Options().Router.ID() == route.Router.ID() { + if q.Options().Router == "*" { if route.Metric <= q.Options().Metric { results = append(results, route) } @@ -182,8 +182,8 @@ func (t *table) Lookup(q Query) ([]Route, error) { if q.Options().Destination == "*" { for _, route := range routes { - if q.Options().Network == "*" || q.Options().Network == route.Router.Network() { - if q.Options().Router.ID() == "*" || q.Options().Router.ID() == route.Router.ID() { + if q.Options().Network == "*" || q.Options().Network == route.Network { + if q.Options().Router == "*" { if route.Metric <= q.Options().Metric { results = append(results, route) } @@ -193,7 +193,7 @@ func (t *table) Lookup(q Query) ([]Route, error) { } } - if len(results) == 0 && q.Options().Policy != DiscardNoRoute { + if len(results) == 0 && q.Options().Policy != DiscardIfNone { return nil, ErrRouteNotFound } @@ -205,7 +205,6 @@ func (t *table) Watch(opts ...WatchOption) (Watcher, error) { // by default watch everything wopts := WatchOptions{ Destination: "*", - Network: "*", } for _, o := range opts { @@ -256,13 +255,14 @@ func (t *table) String() string { // create nice table printing structure table := tablewriter.NewWriter(sb) - table.SetHeader([]string{"Destination", "Router", "Network", "Metric"}) + table.SetHeader([]string{"Destination", "Gateway", "Router", "Network", "Metric"}) for _, destRoute := range t.m { for _, route := range destRoute { strRoute := []string{ route.Destination, - route.Router.Address(), + route.Gateway, + route.Router, route.Network, fmt.Sprintf("%d", route.Metric), } @@ -278,12 +278,8 @@ func (t *table) String() string { // hash hashes the route using router gateway and network address func (t *table) hash(r Route) uint64 { - destAddr := r.Destination - routerAddr := r.Router.Address() - netAddr := r.Network - t.h.Reset() - t.h.Write([]byte(destAddr + routerAddr + netAddr)) + t.h.Write([]byte(r.Destination + r.Gateway + r.Router + r.Network)) return t.h.Sum64() } diff --git a/network/router/options.go b/network/router/options.go index ee0c6526..3ede4776 100644 --- a/network/router/options.go +++ b/network/router/options.go @@ -8,8 +8,6 @@ import ( var ( // DefaultAddress is default router address DefaultAddress = ":9093" - // DefaultAdvertise is default address advertised to the network - DefaultAdvertise = ":9094" ) // Options are router options @@ -18,12 +16,10 @@ type Options struct { ID string // Address is router address Address string - // Advertise is the address advertised to the network - Advertise string + // Network is micro network + Network string // Registry is the local registry Registry registry.Registry - // Networkis the network registry - Network registry.Registry // Table is routing table Table Table } @@ -42,10 +38,10 @@ func Address(a string) Option { } } -// Advertise sets the address that is advertise to the network -func Advertise(n string) Option { +// Network sets router network +func Network(n string) Option { return func(o *Options) { - o.Advertise = n + o.Network = n } } @@ -63,22 +59,13 @@ func Registry(r registry.Registry) Option { } } -// Network sets the network registry -func Network(r registry.Registry) Option { - return func(o *Options) { - o.Network = r - } -} - // DefaultOptions returns router default options func DefaultOptions() Options { // NOTE: by default both local and network registies use default registry i.e. mdns return Options{ - ID: uuid.New().String(), - Address: DefaultAddress, - Advertise: DefaultAdvertise, - Registry: registry.DefaultRegistry, - Network: registry.DefaultRegistry, - Table: NewTable(), + ID: uuid.New().String(), + Address: DefaultAddress, + Registry: registry.DefaultRegistry, + Table: NewTable(), } } diff --git a/network/router/proto/router.micro.go b/network/router/proto/router.micro.go new file mode 100644 index 00000000..06736f82 --- /dev/null +++ b/network/router/proto/router.micro.go @@ -0,0 +1,91 @@ +// Code generated by protoc-gen-micro. DO NOT EDIT. +// source: go-micro/network/router/proto/router.proto + +package router + +import ( + fmt "fmt" + proto "github.com/golang/protobuf/proto" + math "math" +) + +import ( + context "context" + client "github.com/micro/go-micro/client" + server "github.com/micro/go-micro/server" +) + +// Reference imports to suppress errors if they are not otherwise used. +var _ = proto.Marshal +var _ = fmt.Errorf +var _ = math.Inf + +// This is a compile-time assertion to ensure that this generated file +// is compatible with the proto package it is being compiled against. +// A compilation error at this line likely means your copy of the +// proto package needs to be updated. +const _ = proto.ProtoPackageIsVersion3 // please upgrade the proto package + +// Reference imports to suppress errors if they are not otherwise used. +var _ context.Context +var _ client.Option +var _ server.Option + +// Client API for Router service + +type RouterService interface { + Lookup(ctx context.Context, in *LookupRequest, opts ...client.CallOption) (*LookupResponse, error) +} + +type routerService struct { + c client.Client + name string +} + +func NewRouterService(name string, c client.Client) RouterService { + if c == nil { + c = client.NewClient() + } + if len(name) == 0 { + name = "router" + } + return &routerService{ + c: c, + name: name, + } +} + +func (c *routerService) Lookup(ctx context.Context, in *LookupRequest, opts ...client.CallOption) (*LookupResponse, error) { + req := c.c.NewRequest(c.name, "Router.Lookup", in) + out := new(LookupResponse) + err := c.c.Call(ctx, req, out, opts...) + if err != nil { + return nil, err + } + return out, nil +} + +// Server API for Router service + +type RouterHandler interface { + Lookup(context.Context, *LookupRequest, *LookupResponse) error +} + +func RegisterRouterHandler(s server.Server, hdlr RouterHandler, opts ...server.HandlerOption) error { + type router interface { + Lookup(ctx context.Context, in *LookupRequest, out *LookupResponse) error + } + type Router struct { + router + } + h := &routerHandler{hdlr} + return s.Handle(s.NewHandler(&Router{h}, opts...)) +} + +type routerHandler struct { + RouterHandler +} + +func (h *routerHandler) Lookup(ctx context.Context, in *LookupRequest, out *LookupResponse) error { + return h.RouterHandler.Lookup(ctx, in, out) +} diff --git a/network/router/proto/router.pb.go b/network/router/proto/router.pb.go new file mode 100644 index 00000000..d0aff457 --- /dev/null +++ b/network/router/proto/router.pb.go @@ -0,0 +1,324 @@ +// Code generated by protoc-gen-go. DO NOT EDIT. +// source: go-micro/network/router/proto/router.proto + +package router + +import ( + context "context" + fmt "fmt" + proto "github.com/golang/protobuf/proto" + grpc "google.golang.org/grpc" + math "math" +) + +// Reference imports to suppress errors if they are not otherwise used. +var _ = proto.Marshal +var _ = fmt.Errorf +var _ = math.Inf + +// This is a compile-time assertion to ensure that this generated file +// is compatible with the proto package it is being compiled against. +// A compilation error at this line likely means your copy of the +// proto package needs to be updated. +const _ = proto.ProtoPackageIsVersion3 // please upgrade the proto package + +// LookupRequest is made to Lookup +type LookupRequest struct { + Query *Query `protobuf:"bytes,1,opt,name=query,proto3" json:"query,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *LookupRequest) Reset() { *m = LookupRequest{} } +func (m *LookupRequest) String() string { return proto.CompactTextString(m) } +func (*LookupRequest) ProtoMessage() {} +func (*LookupRequest) Descriptor() ([]byte, []int) { + return fileDescriptor_fc08514fc6dadd29, []int{0} +} + +func (m *LookupRequest) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_LookupRequest.Unmarshal(m, b) +} +func (m *LookupRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_LookupRequest.Marshal(b, m, deterministic) +} +func (m *LookupRequest) XXX_Merge(src proto.Message) { + xxx_messageInfo_LookupRequest.Merge(m, src) +} +func (m *LookupRequest) XXX_Size() int { + return xxx_messageInfo_LookupRequest.Size(m) +} +func (m *LookupRequest) XXX_DiscardUnknown() { + xxx_messageInfo_LookupRequest.DiscardUnknown(m) +} + +var xxx_messageInfo_LookupRequest proto.InternalMessageInfo + +func (m *LookupRequest) GetQuery() *Query { + if m != nil { + return m.Query + } + return nil +} + +// LookupResponse is returns by Lookup +type LookupResponse struct { + Routes []*Route `protobuf:"bytes,1,rep,name=routes,proto3" json:"routes,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *LookupResponse) Reset() { *m = LookupResponse{} } +func (m *LookupResponse) String() string { return proto.CompactTextString(m) } +func (*LookupResponse) ProtoMessage() {} +func (*LookupResponse) Descriptor() ([]byte, []int) { + return fileDescriptor_fc08514fc6dadd29, []int{1} +} + +func (m *LookupResponse) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_LookupResponse.Unmarshal(m, b) +} +func (m *LookupResponse) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_LookupResponse.Marshal(b, m, deterministic) +} +func (m *LookupResponse) XXX_Merge(src proto.Message) { + xxx_messageInfo_LookupResponse.Merge(m, src) +} +func (m *LookupResponse) XXX_Size() int { + return xxx_messageInfo_LookupResponse.Size(m) +} +func (m *LookupResponse) XXX_DiscardUnknown() { + xxx_messageInfo_LookupResponse.DiscardUnknown(m) +} + +var xxx_messageInfo_LookupResponse proto.InternalMessageInfo + +func (m *LookupResponse) GetRoutes() []*Route { + if m != nil { + return m.Routes + } + return nil +} + +// Query is passed in a LookupRequest +type Query struct { + // destination to lookup + Destination string `protobuf:"bytes,1,opt,name=destination,proto3" json:"destination,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *Query) Reset() { *m = Query{} } +func (m *Query) String() string { return proto.CompactTextString(m) } +func (*Query) ProtoMessage() {} +func (*Query) Descriptor() ([]byte, []int) { + return fileDescriptor_fc08514fc6dadd29, []int{2} +} + +func (m *Query) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_Query.Unmarshal(m, b) +} +func (m *Query) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_Query.Marshal(b, m, deterministic) +} +func (m *Query) XXX_Merge(src proto.Message) { + xxx_messageInfo_Query.Merge(m, src) +} +func (m *Query) XXX_Size() int { + return xxx_messageInfo_Query.Size(m) +} +func (m *Query) XXX_DiscardUnknown() { + xxx_messageInfo_Query.DiscardUnknown(m) +} + +var xxx_messageInfo_Query proto.InternalMessageInfo + +func (m *Query) GetDestination() string { + if m != nil { + return m.Destination + } + return "" +} + +// Route is a service route +type Route struct { + // service for the route + Destination string `protobuf:"bytes,1,opt,name=destination,proto3" json:"destination,omitempty"` + // gateway as the next hop + Gateway string `protobuf:"bytes,2,opt,name=gateway,proto3" json:"gateway,omitempty"` + // the router that advertise this route + Router string `protobuf:"bytes,3,opt,name=router,proto3" json:"router,omitempty"` + // the network for this destination + Network string `protobuf:"bytes,4,opt,name=network,proto3" json:"network,omitempty"` + // the metric / score of this route + Metric int64 `protobuf:"varint,5,opt,name=metric,proto3" json:"metric,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *Route) Reset() { *m = Route{} } +func (m *Route) String() string { return proto.CompactTextString(m) } +func (*Route) ProtoMessage() {} +func (*Route) Descriptor() ([]byte, []int) { + return fileDescriptor_fc08514fc6dadd29, []int{3} +} + +func (m *Route) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_Route.Unmarshal(m, b) +} +func (m *Route) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_Route.Marshal(b, m, deterministic) +} +func (m *Route) XXX_Merge(src proto.Message) { + xxx_messageInfo_Route.Merge(m, src) +} +func (m *Route) XXX_Size() int { + return xxx_messageInfo_Route.Size(m) +} +func (m *Route) XXX_DiscardUnknown() { + xxx_messageInfo_Route.DiscardUnknown(m) +} + +var xxx_messageInfo_Route proto.InternalMessageInfo + +func (m *Route) GetDestination() string { + if m != nil { + return m.Destination + } + return "" +} + +func (m *Route) GetGateway() string { + if m != nil { + return m.Gateway + } + return "" +} + +func (m *Route) GetRouter() string { + if m != nil { + return m.Router + } + return "" +} + +func (m *Route) GetNetwork() string { + if m != nil { + return m.Network + } + return "" +} + +func (m *Route) GetMetric() int64 { + if m != nil { + return m.Metric + } + return 0 +} + +func init() { + proto.RegisterType((*LookupRequest)(nil), "LookupRequest") + proto.RegisterType((*LookupResponse)(nil), "LookupResponse") + proto.RegisterType((*Query)(nil), "Query") + proto.RegisterType((*Route)(nil), "Route") +} + +func init() { + proto.RegisterFile("go-micro/network/router/proto/router.proto", fileDescriptor_fc08514fc6dadd29) +} + +var fileDescriptor_fc08514fc6dadd29 = []byte{ + // 242 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x84, 0x90, 0xc1, 0x4a, 0xc3, 0x40, + 0x10, 0x86, 0x5d, 0x63, 0x56, 0x9c, 0x62, 0x85, 0x3d, 0xc8, 0x22, 0x22, 0x61, 0x4f, 0x55, 0x69, + 0x22, 0x15, 0xdf, 0xc2, 0x8b, 0xfb, 0x06, 0xb1, 0x0e, 0x25, 0x94, 0x66, 0xd2, 0xdd, 0x09, 0xa5, + 0x0f, 0xe1, 0x3b, 0x4b, 0x26, 0x5b, 0x30, 0xa7, 0x1e, 0xbf, 0x99, 0xf9, 0x7e, 0x76, 0x7f, 0x78, + 0xd9, 0xd0, 0x72, 0xd7, 0xac, 0x03, 0x55, 0x2d, 0xf2, 0x81, 0xc2, 0xb6, 0x0a, 0xd4, 0x33, 0x86, + 0xaa, 0x0b, 0xc4, 0x94, 0xa0, 0x14, 0x70, 0x4b, 0xb8, 0xfd, 0x24, 0xda, 0xf6, 0x9d, 0xc7, 0x7d, + 0x8f, 0x91, 0xcd, 0x23, 0xe4, 0xfb, 0x1e, 0xc3, 0xd1, 0xaa, 0x42, 0x2d, 0x66, 0x2b, 0x5d, 0x7e, + 0x0d, 0xe4, 0xc7, 0xa1, 0x7b, 0x83, 0xf9, 0xe9, 0x3c, 0x76, 0xd4, 0x46, 0x34, 0x4f, 0xa0, 0x25, + 0x30, 0x5a, 0x55, 0x64, 0x22, 0xf8, 0x01, 0x7d, 0x9a, 0xba, 0x67, 0xc8, 0x25, 0xc1, 0x14, 0x30, + 0xfb, 0xc1, 0xc8, 0x4d, 0x5b, 0x73, 0x43, 0xad, 0xc4, 0xdf, 0xf8, 0xff, 0x23, 0xf7, 0xab, 0x20, + 0x17, 0xf9, 0xfc, 0xad, 0xb1, 0x70, 0xbd, 0xa9, 0x19, 0x0f, 0xf5, 0xd1, 0x5e, 0xca, 0xf6, 0x84, + 0xe6, 0x3e, 0x3d, 0x28, 0xd8, 0x4c, 0x16, 0x89, 0x06, 0x23, 0xd5, 0x61, 0xaf, 0x46, 0x23, 0xe1, + 0x60, 0xec, 0x90, 0x43, 0xb3, 0xb6, 0x79, 0xa1, 0x16, 0x99, 0x4f, 0xb4, 0xfa, 0x00, 0xed, 0x47, + 0xf7, 0x15, 0xf4, 0xf8, 0x6d, 0x33, 0x2f, 0x27, 0x75, 0x3d, 0xdc, 0x95, 0xd3, 0x3e, 0xdc, 0xc5, + 0xb7, 0x96, 0x66, 0xdf, 0xff, 0x02, 0x00, 0x00, 0xff, 0xff, 0x4d, 0x73, 0x18, 0x9e, 0x87, 0x01, + 0x00, 0x00, +} + +// Reference imports to suppress errors if they are not otherwise used. +var _ context.Context +var _ grpc.ClientConn + +// This is a compile-time assertion to ensure that this generated file +// is compatible with the grpc package it is being compiled against. +const _ = grpc.SupportPackageIsVersion4 + +// RouterClient is the client API for Router service. +// +// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://godoc.org/google.golang.org/grpc#ClientConn.NewStream. +type RouterClient interface { + Lookup(ctx context.Context, in *LookupRequest, opts ...grpc.CallOption) (*LookupResponse, error) +} + +type routerClient struct { + cc *grpc.ClientConn +} + +func NewRouterClient(cc *grpc.ClientConn) RouterClient { + return &routerClient{cc} +} + +func (c *routerClient) Lookup(ctx context.Context, in *LookupRequest, opts ...grpc.CallOption) (*LookupResponse, error) { + out := new(LookupResponse) + err := c.cc.Invoke(ctx, "/Router/Lookup", in, out, opts...) + if err != nil { + return nil, err + } + return out, nil +} + +// RouterServer is the server API for Router service. +type RouterServer interface { + Lookup(context.Context, *LookupRequest) (*LookupResponse, error) +} + +func RegisterRouterServer(s *grpc.Server, srv RouterServer) { + s.RegisterService(&_Router_serviceDesc, srv) +} + +func _Router_Lookup_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(LookupRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(RouterServer).Lookup(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: "/Router/Lookup", + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(RouterServer).Lookup(ctx, req.(*LookupRequest)) + } + return interceptor(ctx, in, info, handler) +} + +var _Router_serviceDesc = grpc.ServiceDesc{ + ServiceName: "Router", + HandlerType: (*RouterServer)(nil), + Methods: []grpc.MethodDesc{ + { + MethodName: "Lookup", + Handler: _Router_Lookup_Handler, + }, + }, + Streams: []grpc.StreamDesc{}, + Metadata: "go-micro/network/router/proto/router.proto", +} diff --git a/network/router/proto/router.proto b/network/router/proto/router.proto new file mode 100644 index 00000000..6f9e7323 --- /dev/null +++ b/network/router/proto/router.proto @@ -0,0 +1,36 @@ +syntax = "proto3"; + +// Router service is used by the proxy to lookup routes +service Router { + rpc Lookup(LookupRequest) returns (LookupResponse) {}; +} + +// LookupRequest is made to Lookup +message LookupRequest { + Query query = 1; +} + +// LookupResponse is returns by Lookup +message LookupResponse { + repeated Route routes = 1; +} + +// Query is passed in a LookupRequest +message Query { + // destination to lookup + string destination = 1; +} + +// Route is a service route +message Route { + // service for the route + string destination = 1; + // gateway as the next hop + string gateway = 2; + // the router that advertise this route + string router = 3; + // the network for this destination + string network = 4; + // the metric / score of this route + int64 metric = 5; +} diff --git a/network/router/query.go b/network/router/query.go index 7c18e075..c479f9db 100644 --- a/network/router/query.go +++ b/network/router/query.go @@ -1,11 +1,18 @@ package router +import ( + "fmt" + "strings" + + "github.com/olekukonko/tablewriter" +) + // LookupPolicy defines query policy type LookupPolicy int const ( - // DiscardNoRoute discards query when no route is found - DiscardNoRoute LookupPolicy = iota + // DiscardIfNone discards query when no route is found + DiscardIfNone LookupPolicy = iota // ClosestMatch returns closest match to supplied query ClosestMatch ) @@ -13,7 +20,7 @@ const ( // String returns human representation of LookupPolicy func (lp LookupPolicy) String() string { switch lp { - case DiscardNoRoute: + case DiscardIfNone: return "DISCARD" case ClosestMatch: return "CLOSEST" @@ -29,10 +36,10 @@ type QueryOption func(*QueryOptions) type QueryOptions struct { // Destination is destination address Destination string + // Router is router address + Router string // Network is network address Network string - // Router is gateway address - Router Router // Metric is route metric Metric int // Policy is query lookup policy @@ -54,7 +61,7 @@ func QueryNetwork(a string) QueryOption { } // QueryRouter sets query gateway address -func QueryRouter(r Router) QueryOption { +func QueryRouter(r string) QueryOption { return func(o *QueryOptions) { o.Router = r } @@ -88,17 +95,14 @@ type query struct { // NewQuery creates new query and returns it func NewQuery(opts ...QueryOption) Query { - // default gateway for wildcard router - r := newRouter(ID("*")) - // default options // NOTE: by default we use DefaultNetworkMetric qopts := QueryOptions{ Destination: "*", + Router: "*", Network: "*", - Router: r, Metric: DefaultNetworkMetric, - Policy: DiscardNoRoute, + Policy: DiscardIfNone, } for _, o := range opts { @@ -114,3 +118,27 @@ func NewQuery(opts ...QueryOption) Query { func (q *query) Options() QueryOptions { return q.opts } + +// String prints routing table query in human readable form +func (q query) String() string { + // this will help us build routing table string + sb := &strings.Builder{} + + // create nice table printing structure + table := tablewriter.NewWriter(sb) + table.SetHeader([]string{"Destination", "Router", "Network", "Metric", "Policy"}) + + strQuery := []string{ + q.opts.Destination, + q.opts.Router, + q.opts.Network, + fmt.Sprintf("%d", q.opts.Metric), + fmt.Sprintf("%s", q.opts.Policy), + } + table.Append(strQuery) + + // render table into sb + table.Render() + + return sb.String() +} diff --git a/network/router/route.go b/network/router/route.go index 88fe5b18..d52c2688 100644 --- a/network/router/route.go +++ b/network/router/route.go @@ -20,7 +20,7 @@ type RoutePolicy int const ( // OverrideIfExists overrides route if it already exists OverrideIfExists RoutePolicy = iota - // IgnoreIfExists does not modify existing route + // IgnoreIfExists instructs to not modify existing route IgnoreIfExists ) @@ -40,8 +40,10 @@ func (p RoutePolicy) String() string { type Route struct { // Destination is destination address Destination string - // Router is the network router - Router Router + // Gateway is route gateway + Gateway string + // Router is the network router address + Router string // Network is micro network address Network string // Metric is the route cost metric @@ -57,11 +59,12 @@ func (r *Route) String() string { // create nice table printing structure table := tablewriter.NewWriter(sb) - table.SetHeader([]string{"Destination", "Router", "Network", "Metric"}) + table.SetHeader([]string{"Destination", "Gateway", "Router", "Network", "Metric"}) strRoute := []string{ r.Destination, - r.Router.Address(), + r.Gateway, + r.Router, r.Network, fmt.Sprintf("%d", r.Metric), } diff --git a/network/router/table_watcher.go b/network/router/table_watcher.go index 17a0971f..97d1f1b7 100644 --- a/network/router/table_watcher.go +++ b/network/router/table_watcher.go @@ -64,22 +64,13 @@ type Watcher interface { type WatchOptions struct { // Specify destination address to watch Destination string - // Specify network to watch - Network string } // WatchDestination sets what destination to watch // Destination is usually microservice name -func WatchDestination(a string) WatchOption { +func WatchDestination(d string) WatchOption { return func(o *WatchOptions) { - o.Destination = a - } -} - -// WatchNetwork sets what network to watch -func WatchNetwork(n string) WatchOption { - return func(o *WatchOptions) { - o.Network = n + o.Destination = d } } @@ -90,19 +81,16 @@ type tableWatcher struct { } // Next returns the next noticed action taken on table -// TODO: this needs to be thought through properly -// we are aiming to provide the same options Query provides +// TODO: this needs to be thought through properly; we only allow watching particular route destination func (w *tableWatcher) Next() (*Event, error) { for { select { case res := <-w.resChan: switch w.opts.Destination { case "*", "": - if w.opts.Network == "*" || w.opts.Network == res.Route.Network { - return res, nil - } - case res.Route.Destination: - if w.opts.Network == "*" || w.opts.Network == res.Route.Network { + return res, nil + default: + if w.opts.Destination == res.Route.Destination { return res, nil } } @@ -132,11 +120,10 @@ func (w *tableWatcher) String() string { sb := &strings.Builder{} table := tablewriter.NewWriter(sb) - table.SetHeader([]string{"Destination", "Network"}) + table.SetHeader([]string{"Destination"}) data := []string{ w.opts.Destination, - w.opts.Network, } table.Append(data)