diff --git a/network/mucp/mucp.go b/network/mucp/mucp.go index e144c351..442ec957 100644 --- a/network/mucp/mucp.go +++ b/network/mucp/mucp.go @@ -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 } diff --git a/router/dns/dns.go b/router/dns/dns.go index 151c9c23..dfbd7b40 100644 --- a/router/dns/dns.go +++ b/router/dns/dns.go @@ -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...) } diff --git a/router/options.go b/router/options.go index 0b82c673..ff04f7e1 100644 --- a/router/options.go +++ b/router/options.go @@ -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(), } } diff --git a/router/query.go b/router/query.go index ec079887..39efd5c2 100644 --- a/router/query.go +++ b/router/query.go @@ -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 { diff --git a/router/registry/registry.go b/router/registry/registry.go index 11571a19..c762d7bc 100644 --- a/router/registry/registry.go +++ b/router/registry/registry.go @@ -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 diff --git a/router/registry/registry_test.go b/router/registry/registry_test.go index 0c1af956..b8886dfb 100644 --- a/router/registry/registry_test.go +++ b/router/registry/registry_test.go @@ -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) - } -} diff --git a/router/registry/table.go b/router/registry/table.go index 7e4b0691..cefd723c 100644 --- a/router/registry/table.go +++ b/router/registry/table.go @@ -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() diff --git a/router/registry/table_test.go b/router/registry/table_test.go index 4fa4131c..1cb5fc5e 100644 --- a/router/registry/table_test.go +++ b/router/registry/table_test.go @@ -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") } } diff --git a/router/router.go b/router/router.go index 66bf9089..c68d8b68 100644 --- a/router/router.go +++ b/router/router.go @@ -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 diff --git a/router/static/static.go b/router/static/static.go index 439b2c8d..f7822b19 100644 --- a/router/static/static.go +++ b/router/static/static.go @@ -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...) }