package router import ( "fmt" "math" "sort" "strings" "sync" "time" "github.com/google/uuid" "github.com/micro/go-micro/registry" "github.com/micro/go-micro/util/log" ) var ( // AdvertiseEventsTick is time interval in which the router advertises route updates AdvertiseEventsTick = 5 * time.Second // AdvertiseTableTick is time interval in which router advertises all routes found in routing table AdvertiseTableTick = 1 * time.Minute // AdvertiseFlushTick is time the yet unconsumed advertisements are flush i.e. discarded AdvertiseFlushTick = 15 * time.Second // DefaultAdvertTTL is default advertisement TTL DefaultAdvertTTL = 1 * time.Minute // AdvertSuppress is advert suppression threshold AdvertSuppress = 200.0 // AdvertRecover is advert recovery threshold AdvertRecover = 120.0 // Penalty for routes processed multiple times Penalty = 100.0 // PenaltyHalfLife is the time the advert penalty decays to half its value PenaltyHalfLife = 30.0 // MaxSuppressTime defines time after which the suppressed advert is deleted MaxSuppressTime = 5 * time.Minute // PenaltyDecay is a coefficient which controls the speed the advert penalty decays PenaltyDecay = math.Log(2) / PenaltyHalfLife ) // router implements default router type router struct { sync.RWMutex options Options status Status table *table exit chan struct{} errChan chan error eventChan chan *Event advertWg *sync.WaitGroup wg *sync.WaitGroup // advert subscribers sub sync.RWMutex subscribers map[string]chan *Advert } // newRouter creates new router and returns it func newRouter(opts ...Option) Router { // get default options options := DefaultOptions() // apply requested options for _, o := range opts { o(&options) } // set initial status to Stopped status := Status{Code: Stopped, Error: nil} return &router{ options: options, status: status, table: newTable(), advertWg: &sync.WaitGroup{}, wg: &sync.WaitGroup{}, subscribers: make(map[string]chan *Advert), } } // Init initializes router with given options func (r *router) Init(opts ...Option) error { r.Lock() defer r.Unlock() for _, o := range opts { o(&r.options) } return nil } // Options returns router options func (r *router) Options() Options { r.Lock() options := r.options r.Unlock() return options } // Table returns routing table func (r *router) Table() Table { return r.table } // manageRoute applies action on a given route func (r *router) manageRoute(route Route, action string) error { switch action { case "create": if err := r.table.Create(route); err != nil && err != ErrDuplicateRoute { return fmt.Errorf("failed adding route for service %s: %s", route.Service, err) } case "delete": if err := r.table.Delete(route); err != nil && err != ErrRouteNotFound { return fmt.Errorf("failed deleting route for service %s: %s", route.Service, err) } case "update": if err := r.table.Update(route); err != nil { return fmt.Errorf("failed updating route for service %s: %s", route.Service, err) } case "solicit": // nothing to do here return nil default: return fmt.Errorf("failed to manage route for service %s: unknown action %s", route.Service, action) } return nil } // manageServiceRoutes applies action to all routes of the service. // It returns error of the action fails with error. func (r *router) manageServiceRoutes(service *registry.Service, action string) error { // action is the routing table action action = strings.ToLower(action) // take route action on each service node for _, node := range service.Nodes { route := Route{ Service: service.Name, Address: node.Address, Gateway: "", Network: r.options.Network, Router: r.options.Id, Link: DefaultLink, Metric: DefaultLocalMetric, } if err := r.manageRoute(route, action); err != nil { return err } } return nil } // manageRegistryRoutes applies action to all routes of each service found in the registry. // It returns error if either the services failed to be listed or the routing table action fails. func (r *router) manageRegistryRoutes(reg registry.Registry, action string) error { services, err := reg.ListServices() if err != nil { return fmt.Errorf("failed listing services: %v", err) } // add each service node as a separate route for _, service := range services { // get the service to retrieve all its info srvs, err := reg.GetService(service.Name) if err != nil { continue } // manage the routes for all returned services for _, srv := range srvs { if err := r.manageServiceRoutes(srv, action); err != nil { return err } } } return nil } // watchRegistry watches registry and updates routing table based on the received events. // It returns error if either the registry watcher fails with error or if the routing table update fails. func (r *router) watchRegistry(w registry.Watcher) error { exit := make(chan bool) defer func() { // close the exit channel when the go routine finishes close(exit) }() // wait in the background for the router to stop // when the router stops, stop the watcher and exit r.wg.Add(1) go func() { defer w.Stop() defer r.wg.Done() select { case <-r.exit: return case <-exit: return } }() var watchErr error for { res, err := w.Next() if err != nil { if err != registry.ErrWatcherStopped { watchErr = err } break } if err := r.manageServiceRoutes(res.Service, res.Action); err != nil { return err } } return watchErr } // 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 *router) watchTable(w Watcher) error { exit := make(chan bool) defer func() { // close the exit channel when the go routine finishes close(exit) }() // wait in the background for the router to stop // when the router stops, stop the watcher and exit r.wg.Add(1) go func() { defer w.Stop() defer r.wg.Done() select { case <-r.exit: return case <-exit: return } }() var watchErr error for { event, err := w.Next() if err != nil { if err != ErrWatcherStopped { watchErr = err } break } select { case <-r.exit: close(r.eventChan) return nil case r.eventChan <- event: } } // close event channel on error close(r.eventChan) return watchErr } // publishAdvert publishes router advert to advert channel func (r *router) publishAdvert(advType AdvertType, events []*Event) { a := &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() } // advertiseTable advertises the whole routing table to the network func (r *router) advertiseTable() error { // create table advertisement ticker ticker := time.NewTicker(AdvertiseTableTick) defer ticker.Stop() for { select { case <-ticker.C: // do full table flush events, err := r.flushRouteEvents(Update) if err != nil { return fmt.Errorf("failed flushing routes: %s", err) } // advertise routes to subscribers if len(events) > 0 { log.Debugf("Router flushing table with %d events: %s", len(events), r.options.Id) r.advertWg.Add(1) go func() { defer r.advertWg.Done() r.publishAdvert(RouteUpdate, events) }() } case <-r.exit: return nil } } } // advert contains a route event to be advertised type advert struct { // event received from routing table event *Event // lastSeen records the time of the last advert update lastSeen time.Time // penalty is current advert penalty penalty float64 // isSuppressed flags the advert suppression isSuppressed bool // suppressTime records the time interval the advert has been suppressed for suppressTime time.Time } // adverts maintains a map of router adverts type adverts map[uint64]*advert // process processes advert // It updates advert timestamp, increments its penalty and // marks upresses or recovers it if it reaches configured thresholds func (m adverts) process(a *advert) error { // lookup advert in adverts hash := a.event.Route.Hash() a, ok := m[hash] if !ok { return fmt.Errorf("advert not found") } // decay the event penalty delta := time.Since(a.lastSeen).Seconds() // decay advert penalty a.penalty = a.penalty * math.Exp(-delta*PenaltyDecay) service := a.event.Route.Service address := a.event.Route.Address // suppress/recover the event based on its penalty level switch { case a.penalty > AdvertSuppress && !a.isSuppressed: log.Debugf("Router suppressing advert %d %.2f for route %s %s", hash, a.penalty, service, address) a.isSuppressed = true a.suppressTime = time.Now() case a.penalty < AdvertRecover && a.isSuppressed: log.Debugf("Router recovering advert %d %.2f for route %s %s", hash, a.penalty, service, address) a.isSuppressed = false } // if suppressed, checked how long has it been suppressed for if a.isSuppressed { // max suppression time threshold has been reached, delete the advert if time.Since(a.suppressTime) > MaxSuppressTime { delete(m, hash) return nil } } return nil } // advertiseEvents advertises routing table events // It suppresses unhealthy flapping events and advertises healthy events upstream. func (r *router) 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 tableWatcher, err := r.Watch() if err != nil { return fmt.Errorf("failed creating routing table watcher: %v", err) } r.wg.Add(1) go func() { defer r.wg.Done() select { case r.errChan <- r.watchTable(tableWatcher): case <-r.exit: } }() for { select { case <-ticker.C: var events []*Event // collect all events which are not flapping for key, advert := range adverts { // process the advert if err := adverts.process(advert); err != nil { log.Debugf("Router failed processing advert %d: %v", key, err) continue } // if suppressed go to the next advert if advert.isSuppressed { continue } // copy the event and append e := new(Event) // this is ok, because router.Event only contains builtin types // and no references so this creates a deep copy of struct Event *e = *(advert.event) events = append(events, e) // delete the advert from adverts delete(adverts, key) } // advertise events to subscribers if len(events) > 0 { log.Debugf("Router publishing %d events", len(events)) r.advertWg.Add(1) go func() { defer r.advertWg.Done() r.publishAdvert(RouteUpdate, events) }() } case e := <-r.eventChan: // if event is nil, continue if e == nil { continue } now := time.Now() log.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() a, ok := adverts[hash] if !ok { a = &advert{ event: e, penalty: Penalty, lastSeen: now, } adverts[hash] = a continue } // override the route event only if the previous event was different if a.event.Type != e.Type { a.event = e } // process the advert if err := adverts.process(a); err != nil { log.Debugf("Router error processing advert %d: %v", hash, err) continue } // update event penalty and timestamp a.lastSeen = now // increment the penalty a.penalty += Penalty log.Debugf("Router advert %d for route %s %s event penalty: %f", hash, a.event.Route.Service, a.event.Route.Address, a.penalty) case <-r.exit: // first wait for the advertiser to finish r.advertWg.Wait() return nil } } } // close closes exit channels func (r *router) close() { // notify all goroutines to finish close(r.exit) // drain the advertise channel only if advertising if r.status.Code == Advertising { // drain the event channel for range r.eventChan { } r.sub.RLock() // close advert subscribers for id, sub := range r.subscribers { // close the channel close(sub) // delete the subscriber delete(r.subscribers, id) } r.sub.RUnlock() } // mark the router as Stopped and set its Error to nil r.status = Status{Code: Stopped, Error: nil} } // watchErrors watches router errors and takes appropriate actions func (r *router) watchErrors() { var err error select { case <-r.exit: return case err = <-r.errChan: } r.Lock() defer r.Unlock() // if the router is not stopped, stop it if r.status.Code != Stopped { // close all the channels r.close() // set the status error if err != nil { r.status.Error = err } } } // Start starts the router func (r *router) Start() error { r.Lock() defer r.Unlock() // only start if we're stopped if r.status.Code != Stopped { return nil } // add all local service routes into the routing table if err := r.manageRegistryRoutes(r.options.Registry, "create"); err != nil { e := fmt.Errorf("failed adding registry routes: %s", err) r.status = Status{Code: Error, Error: e} return e } // add default gateway into routing table if r.options.Gateway != "" { // note, the only non-default value is the gateway route := Route{ Service: "*", Address: "*", Gateway: r.options.Gateway, Network: "*", Router: r.options.Id, Link: DefaultLink, Metric: DefaultLocalMetric, } if err := r.table.Create(route); err != nil { e := fmt.Errorf("failed adding default gateway route: %s", err) r.status = Status{Code: Error, Error: e} return e } } // create error and exit channels r.errChan = make(chan error, 1) r.exit = make(chan struct{}) // registry watcher regWatcher, err := r.options.Registry.Watch() if err != nil { e := fmt.Errorf("failed creating registry watcher: %v", err) r.status = Status{Code: Error, Error: e} return e } r.wg.Add(1) go func() { defer r.wg.Done() select { case r.errChan <- r.watchRegistry(regWatcher): case <-r.exit: } }() // watch for errors and cleanup r.wg.Add(1) go func() { defer r.wg.Done() r.watchErrors() }() // mark router as Running r.status = Status{Code: Running, Error: nil} 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 *router) Advertise() (<-chan *Advert, error) { r.Lock() defer r.Unlock() switch r.status.Code { case Advertising: advertChan := make(chan *Advert, 128) r.subscribers[uuid.New().String()] = advertChan return advertChan, nil case Running: // list all the routes and pack them into even slice to advertise events, err := r.flushRouteEvents(Create) if err != nil { return nil, fmt.Errorf("failed to flush routes: %s", err) } // create event channels r.eventChan = make(chan *Event) // create advert channel advertChan := make(chan *Advert, 128) r.subscribers[uuid.New().String()] = advertChan // advertise your presence r.advertWg.Add(1) go func() { defer r.advertWg.Done() r.publishAdvert(Announce, events) }() r.wg.Add(1) go func() { defer r.wg.Done() select { case r.errChan <- r.advertiseEvents(): case <-r.exit: } }() r.advertWg.Add(1) go func() { defer r.advertWg.Done() // advertise the whole routing table select { case r.errChan <- r.advertiseTable(): case <-r.exit: } }() // mark router as Running and set its Error to nil r.status = Status{Code: Advertising, Error: nil} log.Debugf("Router starting to advertise") return advertChan, nil case Stopped: return nil, fmt.Errorf("not running") } return nil, fmt.Errorf("error: %s", r.status.Error) } // Process updates the routing table using the advertised values func (r *router) Process(a *Advert) error { // NOTE: event sorting might not be necessary // copy update events intp new slices events := make([]*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) }) log.Debugf("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 { log.Debugf("Router skipping processing its own route: %s", r.options.Id) continue } // create a copy of the route route := event.Route action := event.Type log.Debugf("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, fmt.Sprintf("%s", action)); 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 *router) flushRouteEvents(evType EventType) ([]*Event, error) { // list all routes routes, err := r.table.List() if err != nil { return nil, fmt.Errorf("failed listing routes: %s", err) } if r.options.Advertise == AdvertiseAll { // build a list of events to advertise events := make([]*Event, len(routes)) for i, route := range routes { event := &Event{ Type: evType, Timestamp: time.Now(), Route: route, } events[i] = event } return events, nil } // routeMap stores optimal routes per service bestRoutes := make(map[string]Route) // go through all routes found in the routing table and collapse them to optimal routes for _, route := range routes { routeKey := route.Service + "@" + route.Network optimal, ok := bestRoutes[routeKey] if !ok { bestRoutes[routeKey] = route continue } // if the current optimal route metric is higher than routing table route, replace it if optimal.Metric > route.Metric { bestRoutes[routeKey] = route continue } // if the metrics are the same, prefer advertising your own route if optimal.Metric == route.Metric { if route.Router == r.options.Id { bestRoutes[routeKey] = route continue } } } log.Debugf("Router advertising %d best routes out of %d", len(bestRoutes), len(routes)) // build a list of events to advertise events := make([]*Event, len(bestRoutes)) i := 0 for _, route := range bestRoutes { event := &Event{ Type: evType, Timestamp: time.Now(), Route: route, } events[i] = event i++ } return events, nil } // Solicit advertises all of its routes to the network // It returns error if the router fails to list the routes func (r *router) Solicit() error { events, err := r.flushRouteEvents(Update) if err != nil { return fmt.Errorf("failed solicit routes: %s", err) } // advertise the routes r.advertWg.Add(1) go func() { defer r.advertWg.Done() r.publishAdvert(RouteUpdate, events) }() return nil } // Lookup routes in the routing table func (r *router) Lookup(q ...QueryOption) ([]Route, error) { return r.table.Query(q...) } // Watch routes func (r *router) Watch(opts ...WatchOption) (Watcher, error) { return r.table.Watch(opts...) } // Status returns router status func (r *router) Status() Status { r.RLock() defer r.RUnlock() // make a copy of the status status := r.status return status } // Stop stops the router func (r *router) Stop() error { r.Lock() log.Debugf("Router shutting down") switch r.status.Code { case Stopped, Error: r.Unlock() return r.status.Error case Running, Advertising: // close all the channels // NOTE: close marks the router status as Stopped r.close() } r.Unlock() log.Debugf("Router waiting for all goroutines to finish") // wait for all goroutines to finish r.wg.Wait() log.Debugf("Router successfully stopped") return nil } // String prints debugging information about router func (r *router) String() string { return "memory" }