unistack-org/micro
4
0
Fork 3
Code Issues 20 Pull Requests 1 Actions Packages Projects Releases 280 Wiki Activity

Strip Advertise/Process from router

Browse Source
This commit is contained in:
Asim Aslam 2020-08-14 23:51:52 +01:00
parent 5238a8a85f
commit 55d62fc1a5
10 changed files with 129 additions and 696 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

125
network/mucp/mucp.go
View File

@ -317,44 +317,43 @@ func (n *mucpNetwork) maskRoute(r *pb.Route) {
}
// advertise advertises routes to the network
func (n *mucpNetwork) advertise(advertChan <-chan *router.Advert) {
func (n *mucpNetwork) advertise(eventChan <-chan *router.Event) {
rnd := rand.New(rand.NewSource(time.Now().UnixNano()))
for {
select {
// process local adverts and randomly fire them at other nodes
case advert := <-advertChan:
// process local events and randomly fire them at other nodes
case event := <-eventChan:
// create a proto advert
var events []*pb.Event
var pbEvents []*pb.Event
for _, event := range advert.Events {
// make a copy of the route
route := &pb.Route{
Service: event.Route.Service,
Address: event.Route.Address,
Gateway: event.Route.Gateway,
Network: event.Route.Network,
Router: event.Route.Router,
Link: event.Route.Link,
Metric: event.Route.Metric,
}
// override the various values
n.maskRoute(route)
e := &pb.Event{
Type: pb.EventType(event.Type),
Timestamp: event.Timestamp.UnixNano(),
Route: route,
}
events = append(events, e)
// make a copy of the route
route := &pb.Route{
Service: event.Route.Service,
Address: event.Route.Address,
Gateway: event.Route.Gateway,
Network: event.Route.Network,
Router: event.Route.Router,
Link: event.Route.Link,
Metric: event.Route.Metric,
}
// override the various values
n.maskRoute(route)
e := &pb.Event{
Type: pb.EventType(event.Type),
Timestamp: event.Timestamp.UnixNano(),
Route: route,
}
pbEvents = append(pbEvents, e)
msg := &pb.Advert{
Id: advert.Id,
Type: pb.AdvertType(advert.Type),
Timestamp: advert.Timestamp.UnixNano(),
Events: events,
Id: n.Id(),
Type: pb.AdvertType(event.Type),
Timestamp: event.Timestamp.UnixNano(),
Events: pbEvents,
}
// get a list of node peers
@ -673,9 +672,10 @@ func (n *mucpNetwork) processCtrlChan(listener tunnel.Listener) {
}
// don't process your own messages
if pbAdvert.Id == n.options.Id {
if pbAdvert.Id == n.Id() {
continue
}
if logger.V(logger.DebugLevel, logger.DefaultLogger) {
logger.Debugf("Network received advert message from: %s", pbAdvert.Id)
}
@ -690,8 +690,6 @@ func (n *mucpNetwork) processCtrlChan(listener tunnel.Listener) {
continue
}
var events []*router.Event
for _, event := range pbAdvert.Events {
// for backwards compatibility reasons
if event == nil || event.Route == nil {
@ -736,38 +734,11 @@ func (n *mucpNetwork) processCtrlChan(listener tunnel.Listener) {
route.Metric = d
}
// create router event
e := &router.Event{
Type: router.EventType(event.Type),
Timestamp: time.Unix(0, pbAdvert.Timestamp),
Route: route,
}
events = append(events, e)
}
// if no events are eligible for processing continue
if len(events) == 0 {
if logger.V(logger.TraceLevel, logger.DefaultLogger) {
logger.Tracef("Network no events to be processed by router: %s", n.options.Id)
}
continue
}
// create an advert and process it
advert := &router.Advert{
Id: pbAdvert.Id,
Type: router.AdvertType(pbAdvert.Type),
Timestamp: time.Unix(0, pbAdvert.Timestamp),
TTL: time.Duration(pbAdvert.Ttl),
Events: events,
}
if logger.V(logger.TraceLevel, logger.DefaultLogger) {
logger.Tracef("Network router %s processing advert: %s", n.Id(), advert.Id)
}
if err := n.router.Process(advert); err != nil {
if logger.V(logger.DebugLevel, logger.DefaultLogger) {
logger.Debugf("Network failed to process advert %s: %v", advert.Id, err)
// update the local table
if err := n.router.Table().Update(route); err != nil {
if logger.V(logger.DebugLevel, logger.DefaultLogger) {
logger.Debugf("Network failed to process advert %s: %v", event.Id, err)
}
}
}
}
@ -1024,13 +995,9 @@ func (n *mucpNetwork) processNetChan(listener tunnel.Listener) {
route.Metric = d
}
/////////////////////////////////////////////////////////////////////
// maybe we should not be this clever ¯\_(ツ)_/¯ //
/////////////////////////////////////////////////////////////////////
// lookup best routes for the services in the just received route
q := []router.QueryOption{
router.QueryService(route.Service),
router.QueryStrategy(n.router.Options().Advertise),
router.QueryLink(route.Link),
}
routes, err := n.router.Table().Query(q...)
@ -1068,8 +1035,6 @@ func (n *mucpNetwork) processNetChan(listener tunnel.Listener) {
if bestRoute.Metric <= route.Metric {
continue
}
///////////////////////////////////////////////////////////////////////
///////////////////////////////////////////////////////////////////////
// add route to the routing table
if err := n.router.Table().Create(route); err != nil && err != router.ErrDuplicateRoute {
@ -1168,6 +1133,7 @@ func (n *mucpNetwork) prunePeerRoutes(peer *node) error {
// lookup all routes originated by router
q := []router.QueryOption{
router.QueryRouter(peer.id),
router.QueryLink("*"),
}
if err := n.pruneRoutes(q...); err != nil {
return err
@ -1176,6 +1142,7 @@ func (n *mucpNetwork) prunePeerRoutes(peer *node) error {
// lookup all routes routable via gw
q = []router.QueryOption{
router.QueryGateway(peer.address),
router.QueryLink("*"),
}
if err := n.pruneRoutes(q...); err != nil {
return err
@ -1417,12 +1384,7 @@ func (n *mucpNetwork) manage() {
// based on the advertisement strategy encoded in protobuf
// It returns error if the routes failed to be retrieved from the routing table
func (n *mucpNetwork) getProtoRoutes() ([]*pb.Route, error) {
// get a list of the best routes for each service in our routing table
q := []router.QueryOption{
router.QueryStrategy(n.router.Options().Advertise),
}
routes, err := n.router.Table().Query(q...)
routes, err := n.router.Table().List()
if err != nil && err != router.ErrRouteNotFound {
return nil, err
}
@ -1768,7 +1730,12 @@ func (n *mucpNetwork) Connect() error {
n.closed = make(chan bool)
// start advertising routes
advertChan, err := n.options.Router.Advertise()
watcher, err := n.options.Router.Watch()
if err != nil {
return err
}
advertChan, err := watcher.Chan()
if err != nil {
return err
}

8
router/dns/dns.go
View File

@ -41,14 +41,6 @@ func (d *dns) Table() router.Table {
return d.table
}
func (d *dns) Advertise() (<-chan *router.Advert, error) {
return nil, nil
}
func (d *dns) Process(*router.Advert) error {
return nil
}
func (d *dns) Lookup(opts ...router.QueryOption) ([]router.Route, error) {
return d.table.Query(opts...)
}

20
router/options.go
View File

@ -20,8 +20,6 @@ type Options struct {
Network string
// Registry is the local registry
Registry registry.Registry
// Advertise is the advertising strategy
Advertise Strategy
// Context for additional options
Context context.Context
// Precache routes
@ -63,13 +61,6 @@ func Registry(r registry.Registry) Option {
}
}
// Advertise sets route advertising strategy
func Advertise(a Strategy) Option {
return func(o *Options) {
o.Advertise = a
}
}
// Precache the routes
func Precache() Option {
return func(o *Options) {
@ -80,11 +71,10 @@ func Precache() Option {
// DefaultOptions returns router default options
func DefaultOptions() Options {
return Options{
Id: uuid.New().String(),
Address: DefaultAddress,
Network: DefaultNetwork,
Registry: mdns.NewRegistry(),
Advertise: AdvertiseLocal,
Context: context.Background(),
Id: uuid.New().String(),
Address: DefaultAddress,
Network: DefaultNetwork,
Registry: mdns.NewRegistry(),
Context: context.Background(),
}
}

23
router/query.go
View File

@ -4,6 +4,7 @@ package router
type QueryOption func(*QueryOptions)
// QueryOptions are routing table query options
// TODO replace with Filter(Route) bool
type QueryOptions struct {
// Service is destination service name
Service string
@ -15,8 +16,8 @@ type QueryOptions struct {
Network string
// Router is router id
Router string
// Strategy is routing strategy
Strategy Strategy
// Link to query
Link string
}
// QueryService sets service to query
@ -54,10 +55,10 @@ func QueryRouter(r string) QueryOption {
}
}
// QueryStrategy sets strategy to query
func QueryStrategy(s Strategy) QueryOption {
// QueryLink sets the link to query
func QueryLink(link string) QueryOption {
return func(o *QueryOptions) {
o.Strategy = s
o.Link = link
}
}
@ -65,12 +66,12 @@ func QueryStrategy(s Strategy) QueryOption {
func NewQuery(opts ...QueryOption) QueryOptions {
// default options
qopts := QueryOptions{
Service: "*",
Address: "*",
Gateway: "*",
Network: "*",
Router: "*",
Strategy: AdvertiseAll,
Service: "*",
Address: "*",
Gateway: "*",
Network: "*",
Router: "*",
Link: DefaultLink,
}
for _, o := range opts {

376
router/registry/registry.go
View File

@ -2,12 +2,10 @@ package registry
import (
"fmt"
"sort"
"strings"
"sync"
"time"
"github.com/google/uuid"
"github.com/micro/go-micro/v3/logger"
"github.com/micro/go-micro/v3/registry"
"github.com/micro/go-micro/v3/router"
@ -18,26 +16,17 @@ var (
RefreshInterval = time.Second * 120
// PruneInterval is how often we prune the routing table
PruneInterval = time.Second * 10
// AdvertiseEventsTick is time interval in which the router advertises route updates
AdvertiseEventsTick = 10 * time.Second
// DefaultAdvertTTL is default advertisement TTL
DefaultAdvertTTL = 2 * time.Minute
)
// rtr implements router interface
type rtr struct {
sync.RWMutex
running bool
table *table
options router.Options
exit chan bool
initChan chan bool
eventChan chan *router.Event
// advert subscribers
sub sync.RWMutex
subscribers map[string]chan *router.Advert
running bool
table *table
options router.Options
exit chan bool
initChan chan bool
}
// NewRouter creates new router and returns it
@ -52,9 +41,8 @@ func NewRouter(opts ...router.Option) router.Router {
// construct the router
r := &rtr{
options: options,
initChan: make(chan bool),
subscribers: make(map[string]chan *router.Advert),
options: options,
initChan: make(chan bool),
}
// create the new table, passing the fetchRoute method in as a fallback if
@ -327,214 +315,6 @@ func (r *rtr) watchRegistry(w registry.Watcher) error {
return nil
}
// watchTable watches routing table entries and either adds or deletes locally registered service to/from network registry
// It returns error if the locally registered services either fails to be added/deleted to/from network registry.
func (r *rtr) watchTable(w router.Watcher) error {
exit := make(chan bool)
defer func() {
close(exit)
}()
// wait in the background for the router to stop
// when the router stops, stop the watcher and exit
go func() {
defer w.Stop()
select {
case <-r.exit:
return
case <-exit:
return
}
}()
for {
event, err := w.Next()
if err != nil {
if err != router.ErrWatcherStopped {
return err
}
break
}
select {
case <-r.exit:
return nil
case r.eventChan <- event:
// process event
}
}
return nil
}
// publishAdvert publishes router advert to advert channel
func (r *rtr) publishAdvert(advType router.AdvertType, events []*router.Event) {
a := &router.Advert{
Id: r.options.Id,
Type: advType,
TTL: DefaultAdvertTTL,
Timestamp: time.Now(),
Events: events,
}
r.sub.RLock()
for _, sub := range r.subscribers {
// now send the message
select {
case sub <- a:
case <-r.exit:
r.sub.RUnlock()
return
}
}
r.sub.RUnlock()
}
// adverts maintains a map of router adverts
type adverts map[uint64]*router.Event
// advertiseEvents advertises routing table events
// It suppresses unhealthy flapping events and advertises healthy events upstream.
func (r *rtr) advertiseEvents() error {
// ticker to periodically scan event for advertising
ticker := time.NewTicker(AdvertiseEventsTick)
defer ticker.Stop()
// adverts is a map of advert events
adverts := make(adverts)
// routing table watcher
w, err := r.Watch()
if err != nil {
return err
}
defer w.Stop()
go func() {
var err error
for {
select {
case <-r.exit:
return
default:
if w == nil {
// routing table watcher
w, err = r.Watch()
if err != nil {
if logger.V(logger.DebugLevel, logger.DefaultLogger) {
logger.Debugf("Error creating watcher: %v", err)
}
time.Sleep(time.Second)
continue
}
}
if err := r.watchTable(w); err != nil {
if logger.V(logger.DebugLevel, logger.DefaultLogger) {
logger.Debugf("Error watching table: %v", err)
}
time.Sleep(time.Second)
}
if w != nil {
// reset
w.Stop()
w = nil
}
}
}
}()
for {
select {
case <-ticker.C:
// If we're not advertising any events then sip processing them entirely
if r.options.Advertise == router.AdvertiseNone {
continue
}
var events []*router.Event
// collect all events which are not flapping
for key, event := range adverts {
// if we only advertise local routes skip processing anything not link local
if r.options.Advertise == router.AdvertiseLocal && event.Route.Link != "local" {
continue
}
// copy the event and append
e := new(router.Event)
// this is ok, because router.Event only contains builtin types
// and no references so this creates a deep copy of struct Event
*e = *event
events = append(events, e)
// delete the advert from adverts
delete(adverts, key)
}
// advertise events to subscribers
if len(events) > 0 {
if logger.V(logger.DebugLevel, logger.DefaultLogger) {
logger.Debugf("Router publishing %d events", len(events))
}
go r.publishAdvert(router.RouteUpdate, events)
}
case e := <-r.eventChan:
// if event is nil, continue
if e == nil {
continue
}
// If we're not advertising any events then skip processing them entirely
if r.options.Advertise == router.AdvertiseNone {
continue
}
// if we only advertise local routes skip processing anything not link local
if r.options.Advertise == router.AdvertiseLocal && e.Route.Link != "local" {
continue
}
if logger.V(logger.DebugLevel, logger.DefaultLogger) {
logger.Debugf("Router processing table event %s for service %s %s", e.Type, e.Route.Service, e.Route.Address)
}
// check if we have already registered the route
hash := e.Route.Hash()
ev, ok := adverts[hash]
if !ok {
ev = e
adverts[hash] = e
continue
}
// override the route event only if the previous event was different
if ev.Type != e.Type {
ev = e
}
case <-r.exit:
if w != nil {
w.Stop()
}
return nil
}
}
}
// drain all the events, only called on Stop
func (r *rtr) drain() {
for {
select {
case <-r.eventChan:
default:
return
}
}
}
// start the router. Should be called under lock.
func (r *rtr) start() error {
if r.running {
@ -621,130 +401,6 @@ func (r *rtr) start() error {
return nil
}
// Advertise stars advertising the routes to the network and returns the advertisements channel to consume from.
// If the router is already advertising it returns the channel to consume from.
// It returns error if either the router is not running or if the routing table fails to list the routes to advertise.
func (r *rtr) Advertise() (<-chan *router.Advert, error) {
r.Lock()
defer r.Unlock()
// we're mutating the subscribers so they need to be locked also
r.sub.Lock()
defer r.sub.Unlock()
// already advertising
if r.eventChan != nil {
advertChan := make(chan *router.Advert, 128)
r.subscribers[uuid.New().String()] = advertChan
return advertChan, nil
}
// list all the routes and pack them into even slice to advertise
events, err := r.flushRouteEvents(router.Create)
if err != nil {
return nil, fmt.Errorf("failed to flush routes: %s", err)
}
// create event channels
r.eventChan = make(chan *router.Event)
// create advert channel
advertChan := make(chan *router.Advert, 128)
r.subscribers[uuid.New().String()] = advertChan
// advertise your presence
go r.publishAdvert(router.Announce, events)
go func() {
select {
case <-r.exit:
return
default:
if err := r.advertiseEvents(); err != nil {
if logger.V(logger.DebugLevel, logger.DefaultLogger) {
logger.Debugf("Error adveritising events: %v", err)
}
}
}
}()
return advertChan, nil
}
// Process updates the routing table using the advertised values
func (r *rtr) Process(a *router.Advert) error {
// NOTE: event sorting might not be necessary
// copy update events intp new slices
events := make([]*router.Event, len(a.Events))
copy(events, a.Events)
// sort events by timestamp
sort.Slice(events, func(i, j int) bool {
return events[i].Timestamp.Before(events[j].Timestamp)
})
if logger.V(logger.TraceLevel, logger.DefaultLogger) {
logger.Tracef("Router %s processing advert from: %s", r.options.Id, a.Id)
}
for _, event := range events {
// skip if the router is the origin of this route
if event.Route.Router == r.options.Id {
if logger.V(logger.TraceLevel, logger.DefaultLogger) {
logger.Tracef("Router skipping processing its own route: %s", r.options.Id)
}
continue
}
// create a copy of the route
route := event.Route
action := event.Type
if logger.V(logger.TraceLevel, logger.DefaultLogger) {
logger.Tracef("Router %s applying %s from router %s for service %s %s", r.options.Id, action, route.Router, route.Service, route.Address)
}
if err := r.manageRoute(route, action.String()); err != nil {
return fmt.Errorf("failed applying action %s to routing table: %s", action, err)
}
}
return nil
}
// flushRouteEvents returns a slice of events, one per each route in the routing table
func (r *rtr) flushRouteEvents(evType router.EventType) ([]*router.Event, error) {
// get a list of routes for each service in our routing table
// for the configured advertising strategy
q := []router.QueryOption{
router.QueryStrategy(r.options.Advertise),
}
routes, err := r.table.Query(q...)
if err != nil && err != router.ErrRouteNotFound {
return nil, err
}
if logger.V(logger.DebugLevel, logger.DefaultLogger) {
logger.Debugf("Router advertising %d routes with strategy %s", len(routes), r.options.Advertise)
}
// build a list of events to advertise
events := make([]*router.Event, len(routes))
var i int
for _, route := range routes {
event := &router.Event{
Type: evType,
Timestamp: time.Now(),
Route: route,
}
events[i] = event
i++
}
return events, nil
}
// Lookup routes in the routing table
func (r *rtr) Lookup(q ...router.QueryOption) ([]router.Route, error) {
return r.Table().Query(q...)
@ -769,24 +425,6 @@ func (r *rtr) Close() error {
}
close(r.exit)
// extract the events
r.drain()
r.sub.Lock()
// close advert subscribers
for id, sub := range r.subscribers {
// close the channel
close(sub)
// delete the subscriber
delete(r.subscribers, id)
}
r.sub.Unlock()
}
// close and remove event chan
if r.eventChan != nil {
close(r.eventChan)
r.eventChan = nil
}
r.running = false

108
router/registry/registry_test.go
View File

@ -1,11 +1,8 @@
package registry
import (
"fmt"
"os"
"sync"
"testing"
"time"
"github.com/micro/go-micro/v3/registry/memory"
"github.com/micro/go-micro/v3/router"
@ -19,11 +16,6 @@ func routerTestSetup() router.Router {
func TestRouterClose(t *testing.T) {
r := routerTestSetup()
_, err := r.Advertise()
if err != nil {
t.Errorf("failed to start advertising: %v", err)
}
if err := r.Close(); err != nil {
t.Errorf("failed to stop router: %v", err)
}
@ -31,103 +23,3 @@ func TestRouterClose(t *testing.T) {
t.Logf("TestRouterStartStop STOPPED")
}
}
func TestRouterAdvertise(t *testing.T) {
r := routerTestSetup()
// lower the advertise interval
AdvertiseEventsTick = 500 * time.Millisecond
ch, err := r.Advertise()
if err != nil {
t.Errorf("failed to start advertising: %v", err)
}
// receive announce event
ann := <-ch
if len(os.Getenv("IN_TRAVIS_CI")) == 0 {
t.Logf("received announce advert: %v", ann)
}
// Generate random unique routes
nrRoutes := 5
routes := make([]router.Route, nrRoutes)
route := router.Route{
Service: "dest.svc",
Address: "dest.addr",
Gateway: "dest.gw",
Network: "dest.network",
Router: "src.router",
Link: "local",
Metric: 10,
}
for i := 0; i < nrRoutes; i++ {
testRoute := route
testRoute.Service = fmt.Sprintf("%s-%d", route.Service, i)
routes[i] = testRoute
}
var advertErr error
createDone := make(chan bool)
errChan := make(chan error)
var wg sync.WaitGroup
wg.Add(1)
go func() {
wg.Done()
defer close(createDone)
for _, route := range routes {
if len(os.Getenv("IN_TRAVIS_CI")) == 0 {
t.Logf("Creating route %v", route)
}
if err := r.Table().Create(route); err != nil {
if len(os.Getenv("IN_TRAVIS_CI")) == 0 {
t.Logf("Failed to create route: %v", err)
}
errChan <- err
return
}
}
}()
var adverts int
readDone := make(chan bool)
wg.Add(1)
go func() {
defer func() {
wg.Done()
readDone <- true
}()
for advert := range ch {
select {
case advertErr = <-errChan:
t.Errorf("failed advertising events: %v", advertErr)
default:
// do nothing for now
if len(os.Getenv("IN_TRAVIS_CI")) == 0 {
t.Logf("Router advert received: %v", advert)
}
adverts += len(advert.Events)
}
return
}
}()
// done adding routes to routing table
<-createDone
// done reading adverts from the routing table
<-readDone
if adverts != nrRoutes {
t.Errorf("Expected %d adverts, received: %d", nrRoutes, adverts)
}
wg.Wait()
if err := r.Close(); err != nil {
t.Errorf("failed to stop router: %v", err)
}
}

44
router/registry/table.go
View File

@ -217,7 +217,7 @@ func (t *table) List() ([]router.Route, error) {
}
// isMatch checks if the route matches given query options
func isMatch(route router.Route, address, gateway, network, rtr string, strategy router.Strategy) bool {
func isMatch(route router.Route, address, gateway, network, rtr, link string) bool {
// matches the values provided
match := func(a, b string) bool {
if a == "*" || b == "*" || a == b {
@ -232,13 +232,6 @@ func isMatch(route router.Route, address, gateway, network, rtr string, strategy
b string
}
// by default assume we are querying all routes
link := "*"
// if AdvertiseLocal change the link query accordingly
if strategy == router.AdvertiseLocal {
link = "local"
}
// compare the following values
values := []compare{
{gateway, route.Gateway},
@ -264,7 +257,7 @@ func filterRoutes(routes map[uint64]*route, opts router.QueryOptions) []router.R
gateway := opts.Gateway
network := opts.Network
rtr := opts.Router
strategy := opts.Strategy
link := opts.Link
// routeMap stores the routes we're going to advertise
routeMap := make(map[string][]router.Route)
@ -273,37 +266,15 @@ func filterRoutes(routes map[uint64]*route, opts router.QueryOptions) []router.R
// get the actual route
route := rt.route
if isMatch(route, address, gateway, network, rtr, strategy) {
if isMatch(route, address, gateway, network, rtr, link) {
// add matchihg route to the routeMap
routeKey := route.Service + "@" + route.Network
// append the first found route to routeMap
_, ok := routeMap[routeKey]
if !ok {
routeMap[routeKey] = append(routeMap[routeKey], route)
continue
}
// if AdvertiseAll, keep appending
if strategy == router.AdvertiseAll || strategy == router.AdvertiseLocal {
routeMap[routeKey] = append(routeMap[routeKey], route)
continue
}
// now we're going to find the best routes
if strategy == router.AdvertiseBest {
// if the current optimal route metric is higher than routing table route, replace it
if len(routeMap[routeKey]) > 0 {
// NOTE: we know that when AdvertiseBest is set, we only ever have one item in current
if routeMap[routeKey][0].Metric > route.Metric {
routeMap[routeKey][0] = route
continue
}
}
}
routeMap[routeKey] = append(routeMap[routeKey], route)
}
}
var results []router.Route
for _, route := range routeMap {
results = append(results, route...)
}
@ -319,11 +290,6 @@ func (t *table) Query(q ...router.QueryOption) ([]router.Route, error) {
// create a cwslicelist of query results
results := make([]router.Route, 0, len(t.routes))
// if No routes are queried, return early
if opts.Strategy == router.AdvertiseNone {
return results, nil
}
// readAndFilter routes for this service under read lock.
readAndFilter := func(q router.QueryOptions) ([]router.Route, bool) {
t.RLock()

109
router/registry/table_test.go
View File

@ -28,19 +28,19 @@ func TestCreate(t *testing.T) {
table, route := testSetup()
if err := table.Create(route); err != nil {
t.Errorf("error adding route: %s", err)
t.Fatalf("error adding route: %s", err)
}
// adds new route for the original destination
route.Gateway = "dest.gw2"
if err := table.Create(route); err != nil {
t.Errorf("error adding route: %s", err)
t.Fatalf("error adding route: %s", err)
}
// adding the same route under Insert policy must error
if err := table.Create(route); err != router.ErrDuplicateRoute {
t.Errorf("error adding route. Expected error: %s, found: %s", router.ErrDuplicateRoute, err)
t.Fatalf("error adding route. Expected error: %s, found: %s", router.ErrDuplicateRoute, err)
}
}
@ -48,7 +48,7 @@ func TestDelete(t *testing.T) {
table, route := testSetup()
if err := table.Create(route); err != nil {
t.Errorf("error adding route: %s", err)
t.Fatalf("error adding route: %s", err)
}
// should fail to delete non-existant route
@ -56,14 +56,14 @@ func TestDelete(t *testing.T) {
route.Service = "randDest"
if err := table.Delete(route); err != router.ErrRouteNotFound {
t.Errorf("error deleting route. Expected: %s, found: %s", router.ErrRouteNotFound, err)
t.Fatalf("error deleting route. Expected: %s, found: %s", router.ErrRouteNotFound, err)
}
// we should be able to delete the existing route
route.Service = prevSvc
if err := table.Delete(route); err != nil {
t.Errorf("error deleting route: %s", err)
t.Fatalf("error deleting route: %s", err)
}
}
@ -71,21 +71,21 @@ func TestUpdate(t *testing.T) {
table, route := testSetup()
if err := table.Create(route); err != nil {
t.Errorf("error adding route: %s", err)
t.Fatalf("error adding route: %s", err)
}
// change the metric of the original route
route.Metric = 200
if err := table.Update(route); err != nil {
t.Errorf("error updating route: %s", err)
t.Fatalf("error updating route: %s", err)
}
// this should add a new route
route.Service = "rand.dest"
if err := table.Update(route); err != nil {
t.Errorf("error updating route: %s", err)
t.Fatalf("error updating route: %s", err)
}
}
@ -97,17 +97,17 @@ func TestList(t *testing.T) {
for i := 0; i < len(svc); i++ {
route.Service = svc[i]
if err := table.Create(route); err != nil {
t.Errorf("error adding route: %s", err)
t.Fatalf("error adding route: %s", err)
}
}
routes, err := table.List()
if err != nil {
t.Errorf("error listing routes: %s", err)
t.Fatalf("error listing routes: %s", err)
}
if len(routes) != len(svc) {
t.Errorf("incorrect number of routes listed. Expected: %d, found: %d", len(svc), len(routes))
t.Fatalf("incorrect number of routes listed. Expected: %d, found: %d", len(svc), len(routes))
}
}
@ -124,17 +124,19 @@ func TestQuery(t *testing.T) {
route.Network = net[i]
route.Gateway = gw[i]
route.Router = rtr[i]
route.Link = router.DefaultLink
if err := table.Create(route); err != nil {
t.Errorf("error adding route: %s", err)
t.Fatalf("error adding route: %s", err)
}
}
// return all routes
routes, err := table.Query()
if err != nil {
t.Errorf("error looking up routes: %s", err)
t.Fatalf("error looking up routes: %s", err)
} else if len(routes) == 0 {
t.Errorf("error looking up routes: not found")
t.Fatalf("error looking up routes: not found")
}
// query routes particular network
@ -142,16 +144,16 @@ func TestQuery(t *testing.T) {
routes, err = table.Query(router.QueryNetwork(network))
if err != nil {
t.Errorf("error looking up routes: %s", err)
t.Fatalf("error looking up routes: %s", err)
}
if len(routes) != 2 {
t.Errorf("incorrect number of routes returned. Expected: %d, found: %d", 2, len(routes))
t.Fatalf("incorrect number of routes returned. Expected: %d, found: %d", 2, len(routes))
}
for _, route := range routes {
if route.Network != network {
t.Errorf("incorrect route returned. Expected network: %s, found: %s", network, route.Network)
t.Fatalf("incorrect route returned. Expected network: %s, found: %s", network, route.Network)
}
}
@ -160,15 +162,15 @@ func TestQuery(t *testing.T) {
routes, err = table.Query(router.QueryGateway(gateway))
if err != nil {
t.Errorf("error looking up routes: %s", err)
t.Fatalf("error looking up routes: %s", err)
}
if len(routes) != 1 {
t.Errorf("incorrect number of routes returned. Expected: %d, found: %d", 1, len(routes))
t.Fatalf("incorrect number of routes returned. Expected: %d, found: %d", 1, len(routes))
}
if routes[0].Gateway != gateway {
t.Errorf("incorrect route returned. Expected gateway: %s, found: %s", gateway, routes[0].Gateway)
t.Fatalf("incorrect route returned. Expected gateway: %s, found: %s", gateway, routes[0].Gateway)
}
// query routes for particular router
@ -176,15 +178,15 @@ func TestQuery(t *testing.T) {
routes, err = table.Query(router.QueryRouter(rt))
if err != nil {
t.Errorf("error looking up routes: %s", err)
t.Fatalf("error looking up routes: %s", err)
}
if len(routes) != 1 {
t.Errorf("incorrect number of routes returned. Expected: %d, found: %d", 1, len(routes))
t.Fatalf("incorrect number of routes returned. Expected: %d, found: %d", 1, len(routes))
}
if routes[0].Router != rt {
t.Errorf("incorrect route returned. Expected router: %s, found: %s", rt, routes[0].Router)
t.Fatalf("incorrect route returned. Expected router: %s, found: %s", rt, routes[0].Router)
}
// query particular gateway and network
@ -196,57 +198,57 @@ func TestQuery(t *testing.T) {
routes, err = table.Query(query...)
if err != nil {
t.Errorf("error looking up routes: %s", err)
t.Fatalf("error looking up routes: %s", err)
}
if len(routes) != 1 {
t.Errorf("incorrect number of routes returned. Expected: %d, found: %d", 1, len(routes))
t.Fatalf("incorrect number of routes returned. Expected: %d, found: %d", 1, len(routes))
}
if routes[0].Gateway != gateway {
t.Errorf("incorrect route returned. Expected gateway: %s, found: %s", gateway, routes[0].Gateway)
t.Fatalf("incorrect route returned. Expected gateway: %s, found: %s", gateway, routes[0].Gateway)
}
if routes[0].Network != network {
t.Errorf("incorrect network returned. Expected network: %s, found: %s", network, routes[0].Network)
t.Fatalf("incorrect network returned. Expected network: %s, found: %s", network, routes[0].Network)
}
if routes[0].Router != rt {
t.Errorf("incorrect route returned. Expected router: %s, found: %s", rt, routes[0].Router)
t.Fatalf("incorrect route returned. Expected router: %s, found: %s", rt, routes[0].Router)
}
// non-existen route query
routes, err = table.Query(router.QueryService("foobar"))
if err != router.ErrRouteNotFound {
t.Errorf("error looking up routes. Expected: %s, found: %s", router.ErrRouteNotFound, err)
t.Fatalf("error looking up routes. Expected: %s, found: %s", router.ErrRouteNotFound, err)
}
if len(routes) != 0 {
t.Errorf("incorrect number of routes returned. Expected: %d, found: %d", 0, len(routes))
t.Fatalf("incorrect number of routes returned. Expected: %d, found: %d", 0, len(routes))
}
// query NO routes
query = []router.QueryOption{
router.QueryGateway(gateway),
router.QueryNetwork(network),
router.QueryStrategy(router.AdvertiseNone),
router.QueryLink("network"),
}
routes, err = table.Query(query...)
if err != nil {
t.Errorf("error looking up routes: %s", err)
t.Fatalf("error looking up routes: %s", err)
}
if len(routes) > 0 {
t.Errorf("incorrect number of routes returned. Expected: %d, found: %d", 0, len(routes))
t.Fatalf("incorrect number of routes returned. Expected: %d, found: %d", 0, len(routes))
}
// insert local routes to query
for i := 0; i < 2; i++ {
route.Link = "local"
route.Link = "foobar"
route.Address = fmt.Sprintf("local.route.address-%d", i)
if err := table.Create(route); err != nil {
t.Errorf("error adding route: %s", err)
t.Fatalf("error adding route: %s", err)
}
}
@ -254,16 +256,16 @@ func TestQuery(t *testing.T) {
query = []router.QueryOption{
router.QueryGateway("*"),
router.QueryNetwork("*"),
router.QueryStrategy(router.AdvertiseLocal),
router.QueryLink("foobar"),
}
routes, err = table.Query(query...)
if err != nil {
t.Errorf("error looking up routes: %s", err)
t.Fatalf("error looking up routes: %s", err)
}
if len(routes) != 2 {
t.Errorf("incorrect number of routes returned. Expected: %d, found: %d", 2, len(routes))
t.Fatalf("incorrect number of routes returned. Expected: %d, found: %d", 2, len(routes))
}
// add two different routes for svcX with different metric
@ -271,32 +273,30 @@ func TestQuery(t *testing.T) {
route.Service = "svcX"
route.Address = fmt.Sprintf("svcX.route.address-%d", i)
route.Metric = int64(100 + i)
route.Link = router.DefaultLink
if err := table.Create(route); err != nil {
t.Errorf("error adding route: %s", err)
t.Fatalf("error adding route: %s", err)
}
}
// query best routes for svcX
query = []router.QueryOption{
router.QueryService("svcX"),
router.QueryStrategy(router.AdvertiseBest),
}
routes, err = table.Query(query...)
if err != nil {
t.Errorf("error looking up routes: %s", err)
t.Fatalf("error looking up routes: %s", err)
}
if len(routes) != 1 {
t.Errorf("incorrect number of routes returned. Expected: %d, found: %d", 1, len(routes))
if len(routes) != 2 {
t.Fatalf("incorrect number of routes returned. Expected: %d, found: %d", 1, len(routes))
}
}
func TestFallback(t *testing.T) {
r := &rtr{
subscribers: make(map[string]chan *router.Advert),
options: router.DefaultOptions(),
options: router.DefaultOptions(),
}
route := router.Route{
Service: "go.micro.service.foo",
@ -311,31 +311,30 @@ func TestFallback(t *testing.T) {
rts, err := r.Lookup(router.QueryService("go.micro.service.foo"))
if err != nil {
t.Errorf("error looking up service %s", err)
t.Fatalf("error looking up service %s", err)
}
if len(rts) != 1 {
t.Errorf("incorrect number of routes returned %d", len(rts))
t.Fatalf("incorrect number of routes returned %d", len(rts))
}
// deleting from the table but the next query should invoke the fallback that we passed during new table creation
if err := r.table.Delete(route); err != nil {
t.Errorf("error deleting route %s", err)
t.Fatalf("error deleting route %s", err)
}
rts, err = r.Lookup(router.QueryService("go.micro.service.foo"))
if err != nil {
t.Errorf("error looking up service %s", err)
t.Fatalf("error looking up service %s", err)
}
if len(rts) != 1 {
t.Errorf("incorrect number of routes returned %d", len(rts))
t.Fatalf("incorrect number of routes returned %d", len(rts))
}
}
func TestFallbackError(t *testing.T) {
r := &rtr{
subscribers: make(map[string]chan *router.Advert),
options: router.DefaultOptions(),
options: router.DefaultOptions(),
}
r.table = newTable(func(s string) ([]router.Route, error) {
return nil, fmt.Errorf("ERROR")
@ -343,7 +342,7 @@ func TestFallbackError(t *testing.T) {
r.start()
_, err := r.Lookup(router.QueryService("go.micro.service.foo"))
if err == nil {
t.Errorf("expected error looking up service but none returned")
t.Fatalf("expected error looking up service but none returned")
}
}

4
router/router.go
View File

@ -27,10 +27,6 @@ type Router interface {
Options() Options
// The routing table
Table() Table
// Advertise advertises routes
Advertise() (<-chan *Advert, error)
// Process processes incoming adverts
Process(*Advert) error
// Lookup queries routes in the routing table
Lookup(...QueryOption) ([]Route, error)
// Watch returns a watcher which tracks updates to the routing table

8
router/static/static.go
View File

@ -33,14 +33,6 @@ func (s *static) Table() router.Table {
return nil
}
func (s *static) Advertise() (<-chan *router.Advert, error) {
return nil, nil
}
func (s *static) Process(*router.Advert) error {
return nil
}
func (s *static) Lookup(opts ...router.QueryOption) ([]router.Route, error) {
return s.table.Query(opts...)
}