2020-07-27 15:22:00 +03:00
|
|
|
package registry
|
|
|
|
|
|
|
|
import (
|
|
|
|
"sync"
|
|
|
|
"time"
|
|
|
|
|
|
|
|
"github.com/google/uuid"
|
2021-01-16 01:12:18 +03:00
|
|
|
"github.com/unistack-org/micro/v3/logger"
|
|
|
|
"github.com/unistack-org/micro/v3/router"
|
2020-07-27 15:22:00 +03:00
|
|
|
)
|
|
|
|
|
|
|
|
// table is an in-memory routing table
|
|
|
|
type table struct {
|
|
|
|
sync.RWMutex
|
2021-01-16 01:12:18 +03:00
|
|
|
// lookup for a service
|
|
|
|
lookup func(string) ([]router.Route, error)
|
2020-07-27 15:22:00 +03:00
|
|
|
// routes stores service routes
|
2020-08-07 22:53:38 +03:00
|
|
|
routes map[string]map[uint64]*route
|
2020-07-27 15:22:00 +03:00
|
|
|
// watchers stores table watchers
|
|
|
|
watchers map[string]*tableWatcher
|
2021-01-16 01:12:18 +03:00
|
|
|
opts router.Options
|
2020-07-27 15:22:00 +03:00
|
|
|
}
|
|
|
|
|
2020-08-07 22:53:38 +03:00
|
|
|
type route struct {
|
|
|
|
route router.Route
|
|
|
|
updated time.Time
|
|
|
|
}
|
|
|
|
|
2020-07-27 15:22:00 +03:00
|
|
|
// newtable creates a new routing table and returns it
|
2021-01-16 01:12:18 +03:00
|
|
|
func newTable(lookup func(string) ([]router.Route, error), opts ...router.Option) *table {
|
2020-07-27 15:22:00 +03:00
|
|
|
return &table{
|
2021-01-16 01:12:18 +03:00
|
|
|
lookup: lookup,
|
2020-08-07 22:53:38 +03:00
|
|
|
routes: make(map[string]map[uint64]*route),
|
|
|
|
watchers: make(map[string]*tableWatcher),
|
2021-01-16 01:12:18 +03:00
|
|
|
opts: router.NewOptions(opts...),
|
2020-08-07 22:53:38 +03:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// pruneRoutes will prune routes older than the time specified
|
|
|
|
func (t *table) pruneRoutes(olderThan time.Duration) {
|
|
|
|
var routes []router.Route
|
|
|
|
|
|
|
|
t.Lock()
|
|
|
|
|
|
|
|
// search for all the routes
|
|
|
|
for _, routeList := range t.routes {
|
|
|
|
for _, r := range routeList {
|
|
|
|
// if any route is older than
|
|
|
|
if time.Since(r.updated).Seconds() > olderThan.Seconds() {
|
|
|
|
routes = append(routes, r.route)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
t.Unlock()
|
|
|
|
|
|
|
|
// delete the routes we've found
|
|
|
|
for _, route := range routes {
|
|
|
|
t.Delete(route)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// deleteService removes the entire service
|
|
|
|
func (t *table) deleteService(service, network string) {
|
|
|
|
t.Lock()
|
2020-08-09 21:39:21 +03:00
|
|
|
defer t.Unlock()
|
2020-08-07 22:53:38 +03:00
|
|
|
|
|
|
|
routes, ok := t.routes[service]
|
|
|
|
if !ok {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
// delete the routes for the service
|
|
|
|
for hash, rt := range routes {
|
|
|
|
// TODO: check if this causes a problem
|
|
|
|
// with * in the network if that is a thing
|
|
|
|
// or blank strings
|
|
|
|
if rt.route.Network != network {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
delete(routes, hash)
|
|
|
|
}
|
|
|
|
|
|
|
|
// delete the map for the service if its empty
|
|
|
|
if len(routes) == 0 {
|
|
|
|
delete(t.routes, service)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
// save the routes
|
|
|
|
t.routes[service] = routes
|
|
|
|
}
|
|
|
|
|
2020-07-27 15:22:00 +03:00
|
|
|
// sendEvent sends events to all subscribed watchers
|
|
|
|
func (t *table) sendEvent(e *router.Event) {
|
|
|
|
t.RLock()
|
|
|
|
defer t.RUnlock()
|
|
|
|
|
|
|
|
if len(e.Id) == 0 {
|
|
|
|
e.Id = uuid.New().String()
|
|
|
|
}
|
|
|
|
|
|
|
|
for _, w := range t.watchers {
|
|
|
|
select {
|
|
|
|
case w.resChan <- e:
|
|
|
|
case <-w.done:
|
|
|
|
// don't block forever
|
|
|
|
case <-time.After(time.Second):
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// Create creates new route in the routing table
|
|
|
|
func (t *table) Create(r router.Route) error {
|
|
|
|
service := r.Service
|
|
|
|
sum := r.Hash()
|
|
|
|
|
|
|
|
t.Lock()
|
|
|
|
defer t.Unlock()
|
|
|
|
|
|
|
|
// check if there are any routes in the table for the route destination
|
|
|
|
if _, ok := t.routes[service]; !ok {
|
2020-08-07 22:53:38 +03:00
|
|
|
t.routes[service] = make(map[uint64]*route)
|
2020-07-27 15:22:00 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
// add new route to the table for the route destination
|
2020-08-07 22:53:38 +03:00
|
|
|
if _, ok := t.routes[service][sum]; ok {
|
|
|
|
return router.ErrDuplicateRoute
|
|
|
|
}
|
|
|
|
|
|
|
|
// create the route
|
|
|
|
t.routes[service][sum] = &route{r, time.Now()}
|
|
|
|
|
2021-01-16 01:12:18 +03:00
|
|
|
if t.opts.Logger.V(logger.DebugLevel) {
|
|
|
|
t.opts.Logger.Debugf(t.opts.Context, "Router emitting %s for route: %s", router.Create, r.Address)
|
2020-07-27 15:22:00 +03:00
|
|
|
}
|
|
|
|
|
2020-08-07 22:53:38 +03:00
|
|
|
// send a route created event
|
|
|
|
go t.sendEvent(&router.Event{Type: router.Create, Timestamp: time.Now(), Route: r})
|
|
|
|
|
|
|
|
return nil
|
2020-07-27 15:22:00 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
// Delete deletes the route from the routing table
|
|
|
|
func (t *table) Delete(r router.Route) error {
|
|
|
|
service := r.Service
|
|
|
|
sum := r.Hash()
|
|
|
|
|
|
|
|
t.Lock()
|
|
|
|
defer t.Unlock()
|
|
|
|
|
|
|
|
if _, ok := t.routes[service]; !ok {
|
|
|
|
return router.ErrRouteNotFound
|
|
|
|
}
|
|
|
|
|
|
|
|
if _, ok := t.routes[service][sum]; !ok {
|
|
|
|
return router.ErrRouteNotFound
|
|
|
|
}
|
|
|
|
|
2020-08-07 22:53:38 +03:00
|
|
|
// delete the route from the service
|
2020-07-27 15:22:00 +03:00
|
|
|
delete(t.routes[service], sum)
|
2020-08-07 22:53:38 +03:00
|
|
|
|
|
|
|
// delete the whole map if there are no routes left
|
2020-07-27 15:22:00 +03:00
|
|
|
if len(t.routes[service]) == 0 {
|
|
|
|
delete(t.routes, service)
|
|
|
|
}
|
2020-08-07 22:53:38 +03:00
|
|
|
|
2021-01-16 01:12:18 +03:00
|
|
|
if t.opts.Logger.V(logger.DebugLevel) {
|
|
|
|
t.opts.Logger.Debugf(t.opts.Context, "Router emitting %s for route: %s", router.Delete, r.Address)
|
2020-07-27 15:22:00 +03:00
|
|
|
}
|
|
|
|
go t.sendEvent(&router.Event{Type: router.Delete, Timestamp: time.Now(), Route: r})
|
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// Update updates routing table with the new route
|
|
|
|
func (t *table) Update(r router.Route) error {
|
|
|
|
service := r.Service
|
|
|
|
sum := r.Hash()
|
|
|
|
|
|
|
|
t.Lock()
|
|
|
|
defer t.Unlock()
|
|
|
|
|
|
|
|
// check if the route destination has any routes in the table
|
|
|
|
if _, ok := t.routes[service]; !ok {
|
2020-08-07 22:53:38 +03:00
|
|
|
t.routes[service] = make(map[uint64]*route)
|
2020-07-27 15:22:00 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
if _, ok := t.routes[service][sum]; !ok {
|
2020-08-07 22:53:38 +03:00
|
|
|
// update the route
|
|
|
|
t.routes[service][sum] = &route{r, time.Now()}
|
|
|
|
|
2021-01-16 01:12:18 +03:00
|
|
|
if t.opts.Logger.V(logger.DebugLevel) {
|
|
|
|
t.opts.Logger.Debugf(t.opts.Context, "Router emitting %s for route: %s", router.Update, r.Address)
|
2020-07-27 15:22:00 +03:00
|
|
|
}
|
|
|
|
go t.sendEvent(&router.Event{Type: router.Update, Timestamp: time.Now(), Route: r})
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// just update the route, but dont emit Update event
|
2020-08-07 22:53:38 +03:00
|
|
|
t.routes[service][sum] = &route{r, time.Now()}
|
2020-07-27 15:22:00 +03:00
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2021-01-16 01:12:18 +03:00
|
|
|
// List returns a list of all routes in the table
|
|
|
|
func (t *table) List() ([]router.Route, error) {
|
2020-08-21 11:23:01 +03:00
|
|
|
t.RLock()
|
|
|
|
defer t.RUnlock()
|
2020-07-27 15:22:00 +03:00
|
|
|
|
2020-08-21 11:23:01 +03:00
|
|
|
var routes []router.Route
|
2021-01-16 01:12:18 +03:00
|
|
|
for _, rmap := range t.routes {
|
|
|
|
for _, route := range rmap {
|
|
|
|
routes = append(routes, route.route)
|
|
|
|
}
|
|
|
|
}
|
2020-08-07 22:53:38 +03:00
|
|
|
|
2021-01-16 01:12:18 +03:00
|
|
|
return routes, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// isMatch checks if the route matches given query options
|
|
|
|
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 {
|
|
|
|
return true
|
2020-08-23 15:10:48 +03:00
|
|
|
}
|
2021-01-16 01:12:18 +03:00
|
|
|
return false
|
|
|
|
}
|
|
|
|
|
|
|
|
// a simple struct to hold our values
|
|
|
|
type compare struct {
|
|
|
|
a string
|
|
|
|
b string
|
|
|
|
}
|
|
|
|
|
|
|
|
// compare the following values
|
|
|
|
values := []compare{
|
|
|
|
{gateway, route.Gateway},
|
|
|
|
{network, route.Network},
|
|
|
|
{rtr, route.Router},
|
|
|
|
{address, route.Address},
|
|
|
|
{link, route.Link},
|
|
|
|
}
|
|
|
|
|
|
|
|
for _, v := range values {
|
|
|
|
// attempt to match each value
|
|
|
|
if !match(v.a, v.b) {
|
|
|
|
return false
|
2020-08-23 15:10:48 +03:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-01-16 01:12:18 +03:00
|
|
|
return true
|
|
|
|
}
|
|
|
|
|
|
|
|
// filterRoutes finds all the routes for given network and router and returns them
|
|
|
|
func filterRoutes(routes map[uint64]*route, opts router.QueryOptions) []router.Route {
|
|
|
|
address := opts.Address
|
|
|
|
gateway := opts.Gateway
|
|
|
|
network := opts.Network
|
|
|
|
rtr := opts.Router
|
|
|
|
link := opts.Link
|
|
|
|
|
|
|
|
// routeMap stores the routes we're going to advertise
|
|
|
|
routeMap := make(map[string][]router.Route)
|
|
|
|
|
|
|
|
var routeCnt int
|
|
|
|
|
|
|
|
for _, rt := range routes {
|
|
|
|
// get the actual route
|
|
|
|
route := rt.route
|
|
|
|
|
|
|
|
if isMatch(route, address, gateway, network, rtr, link) {
|
|
|
|
// add matchihg route to the routeMap
|
|
|
|
routeKey := route.Service + "@" + route.Network
|
|
|
|
routeMap[routeKey] = append(routeMap[routeKey], route)
|
|
|
|
routeCnt++
|
2020-08-23 15:10:48 +03:00
|
|
|
}
|
2020-07-27 15:22:00 +03:00
|
|
|
}
|
2020-08-07 22:53:38 +03:00
|
|
|
|
2021-01-16 01:12:18 +03:00
|
|
|
results := make([]router.Route, 0, routeCnt)
|
|
|
|
for _, route := range routeMap {
|
|
|
|
results = append(results, route...)
|
|
|
|
}
|
|
|
|
|
|
|
|
return results
|
|
|
|
}
|
|
|
|
|
|
|
|
// Lookup queries routing table and returns all routes that match the lookup query
|
|
|
|
func (t *table) Query(q ...router.QueryOption) ([]router.Route, error) {
|
|
|
|
// create new query options
|
|
|
|
opts := router.NewQuery(q...)
|
|
|
|
|
|
|
|
// create a cwslicelist of query results
|
|
|
|
results := make([]router.Route, 0, len(t.routes))
|
|
|
|
|
|
|
|
// readAndFilter routes for this service under read lock.
|
|
|
|
readAndFilter := func(q router.QueryOptions) ([]router.Route, bool) {
|
|
|
|
t.RLock()
|
|
|
|
defer t.RUnlock()
|
|
|
|
|
|
|
|
routes, ok := t.routes[q.Service]
|
|
|
|
if !ok || len(routes) == 0 {
|
|
|
|
return nil, false
|
|
|
|
}
|
|
|
|
|
|
|
|
return filterRoutes(routes, q), true
|
|
|
|
}
|
|
|
|
|
|
|
|
if opts.Service != "*" {
|
|
|
|
// try and load services from the cache
|
|
|
|
if routes, ok := readAndFilter(opts); ok {
|
|
|
|
return routes, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// lookup the route and try again
|
|
|
|
// TODO: move this logic out of the hot path
|
|
|
|
// being hammered on queries will require multiple lookups
|
|
|
|
routes, err := t.lookup(opts.Service)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
// cache the routes
|
|
|
|
for _, rt := range routes {
|
|
|
|
t.Create(rt)
|
|
|
|
}
|
|
|
|
|
|
|
|
// try again
|
|
|
|
if routes, ok := readAndFilter(opts); ok {
|
|
|
|
return routes, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
return nil, router.ErrRouteNotFound
|
|
|
|
}
|
|
|
|
|
|
|
|
// search through all destinations
|
|
|
|
t.RLock()
|
|
|
|
|
|
|
|
for _, routes := range t.routes {
|
|
|
|
// filter the routes
|
|
|
|
found := filterRoutes(routes, opts)
|
|
|
|
// ensure we don't append zero length routes
|
|
|
|
if len(found) == 0 {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
results = append(results, found...)
|
|
|
|
}
|
|
|
|
|
|
|
|
t.RUnlock()
|
|
|
|
|
|
|
|
return results, nil
|
2020-07-27 15:22:00 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
// Watch returns routing table entry watcher
|
|
|
|
func (t *table) Watch(opts ...router.WatchOption) (router.Watcher, error) {
|
|
|
|
// by default watch everything
|
|
|
|
wopts := router.WatchOptions{
|
|
|
|
Service: "*",
|
|
|
|
}
|
|
|
|
|
|
|
|
for _, o := range opts {
|
|
|
|
o(&wopts)
|
|
|
|
}
|
|
|
|
|
|
|
|
w := &tableWatcher{
|
|
|
|
id: uuid.New().String(),
|
|
|
|
opts: wopts,
|
|
|
|
resChan: make(chan *router.Event, 10),
|
|
|
|
done: make(chan struct{}),
|
|
|
|
}
|
|
|
|
|
|
|
|
// when the watcher is stopped delete it
|
|
|
|
go func() {
|
|
|
|
<-w.done
|
|
|
|
t.Lock()
|
|
|
|
delete(t.watchers, w.id)
|
|
|
|
t.Unlock()
|
|
|
|
}()
|
|
|
|
|
|
|
|
// save the watcher
|
|
|
|
t.Lock()
|
|
|
|
t.watchers[w.id] = w
|
|
|
|
t.Unlock()
|
|
|
|
|
|
|
|
return w, nil
|
|
|
|
}
|