Added router ID. Deregister remote services when router is stopped.

Added ID function to router interface.

Network registry addresses are deregistered when the router is stopped.

Query has been updated to search for particular GW in lookups.
This commit is contained in:
Milos Gajdos 2019-06-16 23:09:59 +01:00
parent 322eaae529
commit f62fcaad76
No known key found for this signature in database
GPG Key ID: 8B31058CC55DFD4F
4 changed files with 53 additions and 7 deletions

View File

@ -47,6 +47,11 @@ func (r *router) Options() Options {
return r.opts return r.opts
} }
// ID returns router ID
func (r *router) ID() string {
return r.opts.ID
}
// Table returns routing table // Table returns routing table
func (r *router) Table() Table { func (r *router) Table() Table {
return r.opts.Table return r.opts.Table
@ -90,7 +95,7 @@ func (r *router) Start() error {
node, err := r.parseToNode() node, err := r.parseToNode()
if err != nil { if err != nil {
return fmt.Errorf("failed to parse router into node: %v", err) return fmt.Errorf("failed to parse router into service node: %v", err)
} }
for _, route := range localRoutes { for _, route := range localRoutes {
@ -147,7 +152,7 @@ func (r *router) addServiceRoutes(reg registry.Registry, network string, metric
Network(network), Network(network),
Metric(metric), Metric(metric),
) )
if err := r.opts.Table.Add(route); err != nil { if err := r.opts.Table.Add(route); err != nil && err != ErrDuplicateRoute {
return fmt.Errorf("error adding route for service: %s", service.Name) return fmt.Errorf("error adding route for service: %s", service.Name)
} }
} }
@ -214,12 +219,12 @@ func (r *router) manageServiceRoutes(w registry.Watcher, network string, metric
switch res.Action { switch res.Action {
case "create": case "create":
if len(res.Service.Nodes) > 0 { if len(res.Service.Nodes) > 0 {
if err := r.opts.Table.Add(route); err != nil { if err := r.opts.Table.Add(route); err != nil && err != ErrDuplicateRoute {
return fmt.Errorf("failed to add route for service: %v", res.Service.Name) return fmt.Errorf("failed to add route for service: %v", res.Service.Name)
} }
} }
case "delete": case "delete":
if err := r.opts.Table.Remove(route); err != nil { if err := r.opts.Table.Remove(route); err != nil && err != ErrRouteNotFound {
return fmt.Errorf("failed to remove route for service: %v", res.Service.Name) return fmt.Errorf("failed to remove route for service: %v", res.Service.Name)
} }
} }
@ -280,6 +285,29 @@ func (r *router) watchTable(w Watcher) error {
// Stop stops the router // Stop stops the router
func (r *router) Stop() error { func (r *router) Stop() error {
// NOTE: we need a more efficient way of doing this e.g. network routes should be autoremoved when router stops gossiping
// deregister all services advertised by this router from remote registry
query := NewQuery(QueryGateway(r), QueryNetwork(r.opts.NetworkAddress))
routes, err := r.opts.Table.Lookup(query)
if err != nil && err != ErrRouteNotFound {
return fmt.Errorf("failed to lookup routes for router %s: %v", r.opts.ID, err)
}
node, err := r.parseToNode()
if err != nil {
return fmt.Errorf("failed to parse router into service node: %v", err)
}
for _, route := range routes {
service := &registry.Service{
Name: route.Options().DestAddr,
Nodes: []*registry.Node{node},
}
if err := r.opts.NetworkRegistry.Deregister(service); err != nil {
return fmt.Errorf("failed to deregister service %s from network registry: %v", service.Name, err)
}
}
// notify all goroutines to finish // notify all goroutines to finish
close(r.exit) close(r.exit)

View File

@ -64,7 +64,6 @@ func (t *table) Options() TableOptions {
// Add adds a route to the routing table // Add adds a route to the routing table
func (t *table) Add(r Route) error { func (t *table) Add(r Route) error {
destAddr := r.Options().DestAddr destAddr := r.Options().DestAddr
sum := t.hash(r) sum := t.hash(r)
@ -144,19 +143,23 @@ func (t *table) Lookup(q Query) ([]Route, error) {
} }
for _, route := range routes { for _, route := range routes {
if q.Options().Network == "*" || q.Options().Network == route.Options().Network { if q.Options().Network == "*" || q.Options().Network == route.Options().Network {
if q.Options().Gateway.ID() == "*" || q.Options().Gateway.ID() == route.Options().Gateway.ID() {
results = append(results, route) results = append(results, route)
} }
} }
} }
}
if q.Options().DestAddr == "*" { if q.Options().DestAddr == "*" {
for _, route := range routes { for _, route := range routes {
if q.Options().Network == "*" || q.Options().Network == route.Options().Network { if q.Options().Network == "*" || q.Options().Network == route.Options().Network {
if q.Options().Gateway.ID() == "*" || q.Options().Gateway.ID() == route.Options().Gateway.ID() {
results = append(results, route) results = append(results, route)
} }
} }
} }
} }
}
if len(results) == 0 && q.Options().Policy != DiscardNoRoute { if len(results) == 0 && q.Options().Policy != DiscardNoRoute {
return nil, ErrRouteNotFound return nil, ErrRouteNotFound

View File

@ -19,6 +19,8 @@ type QueryOptions struct {
DestAddr string DestAddr string
// NetworkAddress is network address // NetworkAddress is network address
Network string Network string
// Gateway is gateway address
Gateway Router
// Policy is query lookup policy // Policy is query lookup policy
Policy LookupPolicy Policy LookupPolicy
} }
@ -37,6 +39,13 @@ func QueryNetwork(a string) QueryOption {
} }
} }
// QueryGateway sets query gateway address
func QueryGateway(r Router) QueryOption {
return func(o *QueryOptions) {
o.Gateway = r
}
}
// QueryPolicy sets query policy // QueryPolicy sets query policy
func QueryPolicy(p LookupPolicy) QueryOption { func QueryPolicy(p LookupPolicy) QueryOption {
return func(o *QueryOptions) { return func(o *QueryOptions) {
@ -57,10 +66,14 @@ type query struct {
// NewQuery creates new query and returns it // NewQuery creates new query and returns it
func NewQuery(opts ...QueryOption) Query { func NewQuery(opts ...QueryOption) Query {
// default gateway for wildcard router
r := newRouter(ID("*"))
// default options // default options
qopts := QueryOptions{ qopts := QueryOptions{
DestAddr: "*", DestAddr: "*",
Network: "*", Network: "*",
Gateway: r,
Policy: DiscardNoRoute, Policy: DiscardNoRoute,
} }

View File

@ -7,6 +7,8 @@ type Router interface {
Init(...Option) error Init(...Option) error
// Options returns the router options // Options returns the router options
Options() Options Options() Options
// ID returns id of the router
ID() string
// Table returns the router routing table // Table returns the router routing table
Table() Table Table() Table
// Address returns the router adddress // Address returns the router adddress