홍길동 / onos

Go to a project
Toggle navigation Toggle navigation pinning
Jonathan Hart
Committed by Gerrit Code Review
1ad75f24 1 parent 73b1277f

Modified BGP and FPM route sources to push to new route service. 

Also created an adapter to adapt the new interface to the old one for
backwards compatibilty with existing FIB components.

Change-Id: If8eb2220d9e4e69af135a8f9469ffda567ed4448
Inline Side-by-side
Showing 13 changed files with 795 additions and 604 deletions
......@@ -18,6 +18,7 @@ package org.onosproject.routing;
/**
* A source of route updates.
*/
@Deprecated
public interface RouteSourceService {
/**
......
......@@ -16,6 +16,7 @@
package org.onosproject.routing.bgp;
import org.onlab.packet.IpPrefix;
import org.onosproject.incubator.net.routing.Route;
import org.onosproject.routing.RouteUpdate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
......@@ -45,15 +46,16 @@ class BgpRouteSelector {
* Processes route entry updates: added/updated and deleted route
* entries.
*
* @param bgpSession the BGP session the route entry updates were
* received on
* @param addedBgpRouteEntries the added/updated route entries to process
* @param deletedBgpRouteEntries the deleted route entries to process
*/
synchronized void routeUpdates(BgpSession bgpSession,
synchronized void routeUpdates(
Collection<BgpRouteEntry> addedBgpRouteEntries,
Collection<BgpRouteEntry> deletedBgpRouteEntries) {
Collection<RouteUpdate> routeUpdates = new LinkedList<>();
Collection<Route> updates = new LinkedList<>();
Collection<Route> withdraws = new LinkedList<>();
RouteUpdate routeUpdate;
if (bgpSessionManager.isShutdown()) {
......@@ -61,32 +63,42 @@ class BgpRouteSelector {
}
// Process the deleted route entries
for (BgpRouteEntry bgpRouteEntry : deletedBgpRouteEntries) {
routeUpdate = processDeletedRoute(bgpSession, bgpRouteEntry);
if (routeUpdate != null) {
routeUpdates.add(routeUpdate);
}
routeUpdate = processDeletedRoute(bgpRouteEntry);
convertRouteUpdateToRoute(routeUpdate, updates, withdraws);
}
// Process the added/updated route entries
for (BgpRouteEntry bgpRouteEntry : addedBgpRouteEntries) {
routeUpdate = processAddedRoute(bgpSession, bgpRouteEntry);
if (routeUpdate != null) {
routeUpdates.add(routeUpdate);
routeUpdate = processAddedRoute(bgpRouteEntry);
convertRouteUpdateToRoute(routeUpdate, updates, withdraws);
}
bgpSessionManager.withdraw(withdraws);
bgpSessionManager.update(updates);
}
private void convertRouteUpdateToRoute(RouteUpdate routeUpdate,
Collection<Route> updates,
Collection<Route> withdraws) {
if (routeUpdate != null) {
Route route = new Route(Route.Source.BGP, routeUpdate.routeEntry().prefix(),
routeUpdate.routeEntry().nextHop());
if (routeUpdate.type().equals(RouteUpdate.Type.UPDATE)) {
updates.add(route);
} else if (routeUpdate.type().equals(RouteUpdate.Type.DELETE)) {
withdraws.add(route);
}
}
bgpSessionManager.getRouteListener().update(routeUpdates);
}
/**
* Processes an added/updated route entry.
*
* @param bgpSession the BGP session the route entry update was received on
* @param bgpRouteEntry the added/updated route entry
* @return the result route update that should be forwarded to the
* Route Listener, or null if no route update should be forwarded
*/
private RouteUpdate processAddedRoute(BgpSession bgpSession,
BgpRouteEntry bgpRouteEntry) {
private RouteUpdate processAddedRoute(BgpRouteEntry bgpRouteEntry) {
RouteUpdate routeUpdate;
BgpRouteEntry bestBgpRouteEntry =
bgpSessionManager.findBgpRoute(bgpRouteEntry.prefix());
......@@ -136,13 +148,11 @@ class BgpRouteSelector {
/**
* Processes a deleted route entry.
*
* @param bgpSession the BGP session the route entry update was received on
* @param bgpRouteEntry the deleted route entry
* @return the result route update that should be forwarded to the
* Route Listener, or null if no route update should be forwarded
*/
private RouteUpdate processDeletedRoute(BgpSession bgpSession,
BgpRouteEntry bgpRouteEntry) {
private RouteUpdate processDeletedRoute(BgpRouteEntry bgpRouteEntry) {
RouteUpdate routeUpdate;
BgpRouteEntry bestBgpRouteEntry =
bgpSessionManager.findBgpRoute(bgpRouteEntry.prefix());
......
......@@ -327,6 +327,7 @@ public class BgpSession extends SimpleChannelHandler {
ctx.getChannel().getRemoteAddress(),
ctx.getChannel().getLocalAddress(),
e);
log.debug("Exception:", e.getCause());
processChannelDisconnected();
}
......@@ -350,8 +351,8 @@ public class BgpSession extends SimpleChannelHandler {
BgpRouteSelector bgpRouteSelector =
bgpSessionManager.getBgpRouteSelector();
Collection<BgpRouteEntry> addedRoutes = Collections.emptyList();
bgpRouteSelector.routeUpdates(this, addedRoutes, deletedRoutes4);
bgpRouteSelector.routeUpdates(this, addedRoutes, deletedRoutes6);
bgpRouteSelector.routeUpdates(addedRoutes, deletedRoutes4);
bgpRouteSelector.routeUpdates(addedRoutes, deletedRoutes6);
bgpSessionManager.peerDisconnected(this);
}
......
......@@ -19,6 +19,8 @@ import org.apache.felix.scr.annotations.Activate;
import org.apache.felix.scr.annotations.Component;
import org.apache.felix.scr.annotations.Deactivate;
import org.apache.felix.scr.annotations.Modified;
import org.apache.felix.scr.annotations.Reference;
import org.apache.felix.scr.annotations.ReferenceCardinality;
import org.apache.felix.scr.annotations.Service;
import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.channel.Channel;
......@@ -34,8 +36,8 @@ import org.onlab.packet.Ip4Address;
import org.onlab.packet.Ip4Prefix;
import org.onlab.packet.Ip6Prefix;
import org.onlab.packet.IpPrefix;
import org.onosproject.routing.RouteSourceService;
import org.onosproject.routing.RouteListener;
import org.onosproject.incubator.net.routing.Route;
import org.onosproject.incubator.net.routing.RouteAdminService;
import org.osgi.service.component.ComponentContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
......@@ -48,7 +50,6 @@ import java.util.Dictionary;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import static com.google.common.base.Preconditions.checkNotNull;
import static java.util.concurrent.Executors.newCachedThreadPool;
import static org.onlab.util.Tools.groupedThreads;
......@@ -57,10 +58,13 @@ import static org.onlab.util.Tools.groupedThreads;
*/
@Component(immediate = true, enabled = false)
@Service
public class BgpSessionManager implements BgpInfoService, RouteSourceService {
public class BgpSessionManager implements BgpInfoService {
private static final Logger log =
LoggerFactory.getLogger(BgpSessionManager.class);
@Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
protected RouteAdminService routeService;
boolean isShutdown = true;
private Channel serverChannel; // Listener for incoming BGP connections
private ServerBootstrap serverBootstrap;
......@@ -75,19 +79,19 @@ public class BgpSessionManager implements BgpInfoService, RouteSourceService {
private ConcurrentMap<Ip6Prefix, BgpRouteEntry> bgpRoutes6 =
new ConcurrentHashMap<>();
private RouteListener routeListener;
private static final int DEFAULT_BGP_PORT = 2000;
private int bgpPort;
@Activate
protected void activate(ComponentContext context) {
readComponentConfiguration(context);
start();
log.info("BgpSessionManager started");
}
@Deactivate
protected void deactivate() {
stop();
log.info("BgpSessionManager stopped");
}
......@@ -128,15 +132,6 @@ public class BgpSessionManager implements BgpInfoService, RouteSourceService {
}
/**
* Gets the route listener.
*
* @return the route listener to use
*/
RouteListener getRouteListener() {
return routeListener;
}
/**
* Gets the BGP sessions.
*
* @return the BGP sessions
......@@ -290,13 +285,29 @@ public class BgpSessionManager implements BgpInfoService, RouteSourceService {
return bgpRouteSelector;
}
@Override
public void start(RouteListener routeListener) {
/**
* Sends updates routes to the route service.
*
* @param updates routes to update
*/
void update(Collection<Route> updates) {
routeService.update(updates);
}
/**
* Sends withdrawn routes to the routes service.
*
* @param withdraws routes to withdraw
*/
void withdraw(Collection<Route> withdraws) {
routeService.withdraw(withdraws);
}
public void start() {
log.debug("BGP Session Manager start.");
isShutdown = false;
this.routeListener = checkNotNull(routeListener);
ChannelFactory channelFactory = new NioServerSocketChannelFactory(
newCachedThreadPool(groupedThreads("onos/bgp", "sm-boss-%d")),
newCachedThreadPool(groupedThreads("onos/bgp", "sm-worker-%d")));
......@@ -330,7 +341,6 @@ public class BgpSessionManager implements BgpInfoService, RouteSourceService {
}
}
@Override
public void stop() {
isShutdown = true;
allChannels.close().awaitUninterruptibly();
......
......@@ -154,10 +154,10 @@ final class BgpUpdate {
//
BgpRouteSelector bgpRouteSelector =
bgpSession.getBgpSessionManager().getBgpRouteSelector();
bgpRouteSelector.routeUpdates(bgpSession,
bgpRouteSelector.routeUpdates(
decodedBgpRoutes.addedUnicastRoutes4.values(),
decodedBgpRoutes.deletedUnicastRoutes4.values());
bgpRouteSelector.routeUpdates(bgpSession,
bgpRouteSelector.routeUpdates(
decodedBgpRoutes.addedUnicastRoutes6.values(),
decodedBgpRoutes.deletedUnicastRoutes6.values());
......
......@@ -15,18 +15,10 @@
*/
package org.onosproject.routing.config.impl;
import static org.onosproject.routing.RouteEntry.createBinaryString;
import com.google.common.collect.ImmutableSet;
import com.googlecode.concurrenttrees.radix.node.concrete.DefaultByteArrayNodeFactory;
import com.googlecode.concurrenttrees.radixinverted.ConcurrentInvertedRadixTree;
import com.googlecode.concurrenttrees.radixinverted.InvertedRadixTree;
import java.util.HashSet;
import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;
import org.apache.felix.scr.annotations.Activate;
import org.apache.felix.scr.annotations.Component;
import org.apache.felix.scr.annotations.Deactivate;
......@@ -48,15 +40,22 @@ import org.onosproject.net.config.NetworkConfigListener;
import org.onosproject.net.config.NetworkConfigRegistry;
import org.onosproject.net.config.NetworkConfigService;
import org.onosproject.net.config.basics.SubjectFactories;
import org.onosproject.routing.RoutingService;
import org.onosproject.routing.config.BgpConfig;
import org.onosproject.routing.config.LocalIpPrefixEntry;
import org.onosproject.routing.config.ReactiveRoutingConfig;
import org.onosproject.routing.config.RouterConfig;
import org.onosproject.routing.config.RoutingConfigurationService;
import org.onosproject.routing.impl.Router;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.HashSet;
import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;
import static org.onosproject.routing.RouteEntry.createBinaryString;
/**
* Implementation of RoutingConfigurationService which reads routing
* configuration from a file.
......@@ -165,7 +164,7 @@ public class RoutingConfigurationImpl implements RoutingConfigurationService {
virtualGatewayMacAddress = config.virtualGatewayMacAddress();
// Setup BGP peer connect points
ApplicationId routerAppId = coreService.getAppId(Router.ROUTER_APP_ID);
ApplicationId routerAppId = coreService.getAppId(RoutingService.ROUTER_APP_ID);
if (routerAppId == null) {
log.info("Router application ID is null!");
return;
......
......@@ -15,6 +15,7 @@
*/
package org.onosproject.routing.fpm;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import org.apache.felix.scr.annotations.Activate;
import org.apache.felix.scr.annotations.Component;
......@@ -38,10 +39,8 @@ import org.onlab.packet.IpAddress;
import org.onlab.packet.IpPrefix;
import org.onlab.util.Tools;
import org.onosproject.cfg.ComponentConfigService;
import org.onosproject.routing.RouteEntry;
import org.onosproject.routing.RouteListener;
import org.onosproject.routing.RouteSourceService;
import org.onosproject.routing.RouteUpdate;
import org.onosproject.incubator.net.routing.Route;
import org.onosproject.incubator.net.routing.RouteAdminService;
import org.onosproject.routing.fpm.protocol.FpmHeader;
import org.onosproject.routing.fpm.protocol.Netlink;
import org.onosproject.routing.fpm.protocol.RouteAttribute;
......@@ -55,12 +54,11 @@ import org.slf4j.LoggerFactory;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.Collections;
import java.util.Dictionary;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
import static java.util.concurrent.Executors.newCachedThreadPool;
import static org.onlab.util.Tools.groupedThreads;
......@@ -70,7 +68,7 @@ import static org.onlab.util.Tools.groupedThreads;
*/
@Service
@Component(immediate = true, enabled = false)
public class FpmManager implements RouteSourceService, FpmInfoService {
public class FpmManager implements FpmInfoService {
private final Logger log = LoggerFactory.getLogger(getClass());
private static final int FPM_PORT = 2620;
......@@ -78,15 +76,16 @@ public class FpmManager implements RouteSourceService, FpmInfoService {
@Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
protected ComponentConfigService componentConfigService;
@Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
protected RouteAdminService routeService;
private ServerBootstrap serverBootstrap;
private Channel serverChannel;
private ChannelGroup allChannels = new DefaultChannelGroup();
private Map<SocketAddress, Long> peers = new ConcurrentHashMap<>();
private Map<IpPrefix, RouteEntry> fpmRoutes = new ConcurrentHashMap<>();
private RouteListener routeListener;
private Map<IpPrefix, Route> fpmRoutes = new ConcurrentHashMap<>();
@Property(name = "clearRoutes", boolValue = true,
label = "Whether to clear routes when the FPM connection goes down")
......@@ -96,12 +95,14 @@ public class FpmManager implements RouteSourceService, FpmInfoService {
protected void activate(ComponentContext context) {
componentConfigService.registerProperties(getClass());
modified(context);
startServer();
log.info("Started");
}
@Deactivate
protected void deactivate() {
stopServer();
fpmRoutes.clear();
componentConfigService.unregisterProperties(getClass(), false);
log.info("Stopped");
}
......@@ -165,19 +166,6 @@ public class FpmManager implements RouteSourceService, FpmInfoService {
}
}
@Override
public void start(RouteListener routeListener) {
this.routeListener = routeListener;
startServer();
}
@Override
public void stop() {
fpmRoutes.clear();
stopServer();
}
private void fpmMessage(FpmHeader fpmMessage) {
Netlink netlink = fpmMessage.netlink();
RtNetlink rtNetlink = netlink.rtNetlink();
......@@ -212,51 +200,46 @@ public class FpmManager implements RouteSourceService, FpmInfoService {
IpPrefix prefix = IpPrefix.valueOf(dstAddress, rtNetlink.dstLength());
RouteUpdate routeUpdate = null;
RouteEntry entry;
List<Route> updates = new LinkedList<>();
List<Route> withdraws = new LinkedList<>();
Route route;
switch (netlink.type()) {
case RTM_NEWROUTE:
if (gateway == null) {
// We ignore interface routes with no gateway for now.
return;
}
entry = new RouteEntry(prefix, gateway);
route = new Route(Route.Source.FPM, prefix, gateway);
fpmRoutes.put(entry.prefix(), entry);
fpmRoutes.put(prefix, route);
routeUpdate = new RouteUpdate(RouteUpdate.Type.UPDATE, entry);
updates.add(route);
break;
case RTM_DELROUTE:
RouteEntry existing = fpmRoutes.remove(prefix);
Route existing = fpmRoutes.remove(prefix);
if (existing == null) {
log.warn("Got delete for non-existent prefix");
return;
}
entry = new RouteEntry(prefix, existing.nextHop());
route = new Route(Route.Source.FPM, prefix, existing.nextHop());
routeUpdate = new RouteUpdate(RouteUpdate.Type.DELETE, entry);
withdraws.add(route);
break;
case RTM_GETROUTE:
default:
break;
}
if (routeUpdate == null) {
log.warn("Unsupported FPM message: {}", fpmMessage);
return;
}
routeListener.update(Collections.singletonList(routeUpdate));
routeService.withdraw(withdraws);
routeService.update(updates);
}
private void clearRoutes() {
log.info("Clearing all routes");
List<RouteUpdate> routeUpdates = fpmRoutes.values().stream()
.map(routeEntry -> new RouteUpdate(RouteUpdate.Type.DELETE, routeEntry))
.collect(Collectors.toList());
routeListener.update(routeUpdates);
routeService.withdraw(ImmutableList.copyOf(fpmRoutes.values()));
}
@Override
......
/*
* Copyright 2014-2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.routing.impl;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.Multimaps;
import com.google.common.collect.SetMultimap;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.googlecode.concurrenttrees.common.KeyValuePair;
import com.googlecode.concurrenttrees.radix.node.concrete.DefaultByteArrayNodeFactory;
import com.googlecode.concurrenttrees.radixinverted.ConcurrentInvertedRadixTree;
import com.googlecode.concurrenttrees.radixinverted.InvertedRadixTree;
import org.apache.felix.scr.annotations.Activate;
import org.apache.felix.scr.annotations.Component;
import org.apache.felix.scr.annotations.Deactivate;
import org.apache.felix.scr.annotations.Reference;
import org.apache.felix.scr.annotations.ReferenceCardinality;
import org.apache.felix.scr.annotations.Service;
import org.onlab.packet.Ip4Address;
import org.onlab.packet.Ip6Address;
import org.onlab.packet.IpAddress;
import org.onlab.packet.IpPrefix;
import org.onlab.packet.MacAddress;
import org.onosproject.core.CoreService;
import org.onosproject.net.Host;
import org.onosproject.net.host.HostEvent;
import org.onosproject.net.host.HostListener;
import org.onosproject.net.host.HostService;
import org.onosproject.routing.RouteSourceService;
import org.onosproject.routing.FibEntry;
import org.onosproject.routing.FibListener;
import org.onosproject.routing.FibUpdate;
import org.onosproject.routing.RouteEntry;
import org.onosproject.routing.RouteListener;
import org.onosproject.routing.RouteUpdate;
import org.onosproject.routing.RoutingService;
import org.onosproject.routing.config.RoutingConfigurationService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import static com.google.common.base.Preconditions.checkNotNull;
import static org.onosproject.routing.RouteEntry.createBinaryString;
/**
* This class processes route updates and maintains a Routing Information Base
* (RIB). After route updates have been processed and next hops have been
* resolved, FIB updates are sent to any listening FIB components.
* <p>
* This implementation has been superseded by the RouteService and will be
* removed soon.
* </p>
*/
@Deprecated
@Component(immediate = true, enabled = false)
@Service
public class DefaultRouter implements RoutingService {
private static final Logger log = LoggerFactory.getLogger(DefaultRouter.class);
// Route entries are stored in a radix tree.
// The key in this tree is the binary string of prefix of the route.
private InvertedRadixTree<RouteEntry> ribTable4;
private InvertedRadixTree<RouteEntry> ribTable6;
// Stores all incoming route updates in a queue.
private final BlockingQueue<Collection<RouteUpdate>> routeUpdatesQueue =
new LinkedBlockingQueue<>();
// Next-hop IP address to route entry mapping for next hops pending MAC
// resolution
private SetMultimap<IpAddress, RouteEntry> routesWaitingOnArp;
// The IPv4 address to MAC address mapping
private final Map<IpAddress, MacAddress> ip2Mac = new ConcurrentHashMap<>();
private FibListener fibComponent;
@Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
protected CoreService coreService;
@Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
protected HostService hostService;
@Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
protected RouteSourceService routeSourceService;
@Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
protected RoutingConfigurationService routingConfigurationService;
private ExecutorService bgpUpdatesExecutor;
private final HostListener hostListener = new InternalHostListener();
@Activate
public void activate() {
ribTable4 = new ConcurrentInvertedRadixTree<>(
new DefaultByteArrayNodeFactory());
ribTable6 = new ConcurrentInvertedRadixTree<>(
new DefaultByteArrayNodeFactory());
routesWaitingOnArp = Multimaps.synchronizedSetMultimap(
HashMultimap.create());
coreService.registerApplication(ROUTER_APP_ID);
bgpUpdatesExecutor = Executors.newSingleThreadExecutor(
new ThreadFactoryBuilder()
.setNameFormat("rib-updates-%d").build());
}
@Deactivate
public void deactivate() {
log.debug("Stopped");
}
@Override
public void addFibListener(FibListener fibListener) {
this.fibComponent = checkNotNull(fibListener);
}
@Override
public void start() {
this.hostService.addListener(hostListener);
routeSourceService.start(new InternalRouteListener());
bgpUpdatesExecutor.execute(this::doUpdatesThread);
}
@Override
public void stop() {
routeSourceService.stop();
this.hostService.removeListener(hostListener);
// Stop the thread(s)
bgpUpdatesExecutor.shutdownNow();
synchronized (this) {
// Cleanup all local state
ribTable4 = new ConcurrentInvertedRadixTree<>(
new DefaultByteArrayNodeFactory());
ribTable6 = new ConcurrentInvertedRadixTree<>(
new DefaultByteArrayNodeFactory());
routeUpdatesQueue.clear();
routesWaitingOnArp.clear();
ip2Mac.clear();
}
}
/**
* Entry point for route updates.
*
* @param routeUpdates collection of route updates to process
*/
private void update(Collection<RouteUpdate> routeUpdates) {
try {
routeUpdatesQueue.put(routeUpdates);
} catch (InterruptedException e) {
log.error("Interrupted while putting on routeUpdatesQueue", e);
Thread.currentThread().interrupt();
}
}
/**
* Thread for handling route updates.
*/
private void doUpdatesThread() {
boolean interrupted = false;
try {
while (!interrupted) {
try {
Collection<RouteUpdate> routeUpdates =
routeUpdatesQueue.take();
processRouteUpdates(routeUpdates);
} catch (InterruptedException e) {
log.error("Interrupted while taking from updates queue", e);
interrupted = true;
} catch (Exception e) {
log.error("exception", e);
}
}
} finally {
if (interrupted) {
Thread.currentThread().interrupt();
}
}
}
/**
* Gets all IPv4 routes from the RIB.
*
* @return all IPv4 routes from the RIB
*/
@Override
public Collection<RouteEntry> getRoutes4() {
Iterator<KeyValuePair<RouteEntry>> it =
ribTable4.getKeyValuePairsForKeysStartingWith("").iterator();
List<RouteEntry> routes = new LinkedList<>();
while (it.hasNext()) {
KeyValuePair<RouteEntry> entry = it.next();
routes.add(entry.getValue());
}
return routes;
}
/**
* Gets all IPv6 routes from the RIB.
*
* @return all IPv6 routes from the RIB
*/
@Override
public Collection<RouteEntry> getRoutes6() {
Iterator<KeyValuePair<RouteEntry>> it =
ribTable6.getKeyValuePairsForKeysStartingWith("").iterator();
List<RouteEntry> routes = new LinkedList<>();
while (it.hasNext()) {
KeyValuePair<RouteEntry> entry = it.next();
routes.add(entry.getValue());
}
return routes;
}
/**
* Finds a route in the RIB for a prefix. The prefix can be either IPv4 or
* IPv6.
*
* @param prefix the prefix to use
* @return the route if found, otherwise null
*/
RouteEntry findRibRoute(IpPrefix prefix) {
String binaryString = createBinaryString(prefix);
if (prefix.isIp4()) {
// IPv4
return ribTable4.getValueForExactKey(binaryString);
}
// IPv6
return ribTable6.getValueForExactKey(binaryString);
}
/**
* Adds a route to the RIB. The route can be either IPv4 or IPv6.
*
* @param routeEntry the route entry to use
*/
void addRibRoute(RouteEntry routeEntry) {
if (routeEntry.isIp4()) {
// IPv4
ribTable4.put(createBinaryString(routeEntry.prefix()), routeEntry);
} else {
// IPv6
ribTable6.put(createBinaryString(routeEntry.prefix()), routeEntry);
}
}
/**
* Removes a route for a prefix from the RIB. The prefix can be either IPv4
* or IPv6.
*
* @param prefix the prefix to use
* @return true if the route was found and removed, otherwise false
*/
boolean removeRibRoute(IpPrefix prefix) {
if (prefix.isIp4()) {
// IPv4
return ribTable4.remove(createBinaryString(prefix));
}
// IPv6
return ribTable6.remove(createBinaryString(prefix));
}
/**
* Processes route updates.
*
* @param routeUpdates the route updates to process
*/
void processRouteUpdates(Collection<RouteUpdate> routeUpdates) {
synchronized (this) {
Collection<IpPrefix> withdrawPrefixes = new LinkedList<>();
Collection<FibUpdate> fibUpdates = new LinkedList<>();
Collection<FibUpdate> fibWithdraws = new LinkedList<>();
for (RouteUpdate update : routeUpdates) {
switch (update.type()) {
case UPDATE:
FibEntry fib = processRouteAdd(update.routeEntry(),
withdrawPrefixes);
if (fib != null) {
fibUpdates.add(new FibUpdate(FibUpdate.Type.UPDATE, fib));
}
break;
case DELETE:
processRouteDelete(update.routeEntry(), withdrawPrefixes);
break;
default:
log.error("Unknown update Type: {}", update.type());
break;
}
}
withdrawPrefixes.forEach(p -> fibWithdraws.add(new FibUpdate(
FibUpdate.Type.DELETE, new FibEntry(p, null, null))));
if (!fibUpdates.isEmpty() || !fibWithdraws.isEmpty()) {
fibComponent.update(fibUpdates, fibWithdraws);
}
}
}
/**
* Processes adding a route entry.
* <p>
* The route entry is added to the radix tree. If there was an existing
* next hop for this prefix, but the next hop was different, then the
* old route entry is deleted.
* </p>
* <p>
* NOTE: Currently, we don't handle routes if the next hop is within the
* SDN domain.
* </p>
*
* @param routeEntry the route entry to add
* @param withdrawPrefixes the collection of accumulated prefixes whose
* intents will be withdrawn
* @return the corresponding FIB entry change, or null
*/
private FibEntry processRouteAdd(RouteEntry routeEntry,
Collection<IpPrefix> withdrawPrefixes) {
log.debug("Processing route add: {}", routeEntry);
// Find the old next-hop if we are updating an old route entry
IpAddress oldNextHop = null;
RouteEntry oldRouteEntry = findRibRoute(routeEntry.prefix());
if (oldRouteEntry != null) {
oldNextHop = oldRouteEntry.nextHop();
}
// Add the new route to the RIB
addRibRoute(routeEntry);
if (oldNextHop != null) {
if (oldNextHop.equals(routeEntry.nextHop())) {
return null; // No change
}
//
// Update an existing nexthop for the prefix.
// We need to remove the old flows for this prefix from the
// switches before the new flows are added.
//
withdrawPrefixes.add(routeEntry.prefix());
}
if (routingConfigurationService.isIpPrefixLocal(routeEntry.prefix())) {
// Route originated by local SDN domain
// We don't handle these here, reactive routing APP will handle
// these
log.debug("Own route {} to {}",
routeEntry.prefix(), routeEntry.nextHop());
return null;
}
//
// Find the MAC address of next hop router for this route entry.
// If the MAC address can not be found in ARP cache, then this prefix
// will be put in routesWaitingOnArp queue. Otherwise, generate
// a new route intent.
//
// Monitor the IP address for updates of the MAC address
hostService.startMonitoringIp(routeEntry.nextHop());
// Check if we know the MAC address of the next hop
MacAddress nextHopMacAddress = ip2Mac.get(routeEntry.nextHop());
if (nextHopMacAddress == null) {
Set<Host> hosts = hostService.getHostsByIp(routeEntry.nextHop());
if (!hosts.isEmpty()) {
nextHopMacAddress = hosts.iterator().next().mac();
}
if (nextHopMacAddress != null) {
ip2Mac.put(routeEntry.nextHop(), nextHopMacAddress);
}
}
if (nextHopMacAddress == null) {
routesWaitingOnArp.put(routeEntry.nextHop(), routeEntry);
return null;
}
return new FibEntry(routeEntry.prefix(), routeEntry.nextHop(),
nextHopMacAddress);
}
/**
* Processes the deletion of a route entry.
* <p>
* The prefix for the routing entry is removed from radix tree.
* If the operation is successful, the prefix is added to the collection
* of prefixes whose intents that will be withdrawn.
* </p>
*
* @param routeEntry the route entry to delete
* @param withdrawPrefixes the collection of accumulated prefixes whose
* intents will be withdrawn
*/
private void processRouteDelete(RouteEntry routeEntry,
Collection<IpPrefix> withdrawPrefixes) {
log.debug("Processing route delete: {}", routeEntry);
boolean isRemoved = removeRibRoute(routeEntry.prefix());
if (isRemoved) {
//
// Only withdraw intents if an entry was actually removed from the
// tree. If no entry was removed, the <prefix, nexthop> wasn't
// there so it's probably already been removed and we don't
// need to do anything.
//
withdrawPrefixes.add(routeEntry.prefix());
}
routesWaitingOnArp.remove(routeEntry.nextHop(), routeEntry);
}
/**
* Signals the Router that the MAC to IP mapping has potentially been
* updated. This has the effect of updating the MAC address for any
* installed prefixes if it has changed, as well as installing any pending
* prefixes that were waiting for MAC resolution.
*
* @param ipAddress the IP address that an event was received for
* @param macAddress the most recently known MAC address for the IP address
*/
private void updateMac(IpAddress ipAddress, MacAddress macAddress) {
log.debug("Received updated MAC info: {} => {}", ipAddress,
macAddress);
//
// We synchronize on "this" to prevent changes to the Radix tree
// while we're pushing intents. If the tree changes, the
// tree and the intents could get out of sync.
//
synchronized (this) {
Collection<FibUpdate> submitFibEntries = new LinkedList<>();
Set<RouteEntry> routesToPush =
routesWaitingOnArp.removeAll(ipAddress);
for (RouteEntry routeEntry : routesToPush) {
// These will always be adds
RouteEntry foundRouteEntry = findRibRoute(routeEntry.prefix());
if (foundRouteEntry != null &&
foundRouteEntry.nextHop().equals(routeEntry.nextHop())) {
// We only push FIB updates if the prefix is still in the
// radix tree and the next hop is the same as our entry.
// The prefix could have been removed while we were waiting
// for the ARP, or the next hop could have changed.
submitFibEntries.add(new FibUpdate(FibUpdate.Type.UPDATE,
new FibEntry(routeEntry.prefix(),
ipAddress, macAddress)));
} else {
log.debug("{} has been revoked before the MAC was resolved",
routeEntry);
}
}
if (!submitFibEntries.isEmpty()) {
fibComponent.update(submitFibEntries, Collections.emptyList());
}
ip2Mac.put(ipAddress, macAddress);
}
}
/**
* Listener for host events.
*/
class InternalHostListener implements HostListener {
@Override
public void event(HostEvent event) {
log.debug("Received HostEvent {}", event);
Host host = event.subject();
switch (event.type()) {
case HOST_ADDED:
// FALLTHROUGH
case HOST_UPDATED:
for (IpAddress ipAddress : host.ipAddresses()) {
updateMac(ipAddress, host.mac());
}
break;
case HOST_REMOVED:
for (IpAddress ipAddress : host.ipAddresses()) {
ip2Mac.remove(ipAddress);
}
break;
default:
break;
}
}
}
/**
* Listener for route events.
*/
private class InternalRouteListener implements RouteListener {
@Override
public void update(Collection<RouteUpdate> routeUpdates) {
DefaultRouter.this.update(routeUpdates);
}
}
@Override
public RouteEntry getLongestMatchableRouteEntry(IpAddress ipAddress) {
RouteEntry routeEntry = null;
Iterable<RouteEntry> routeEntries;
if (ipAddress.isIp4()) {
routeEntries = ribTable4.getValuesForKeysPrefixing(
createBinaryString(
IpPrefix.valueOf(ipAddress, Ip4Address.BIT_LENGTH)));
} else {
routeEntries = ribTable6.getValuesForKeysPrefixing(
createBinaryString(
IpPrefix.valueOf(ipAddress, Ip6Address.BIT_LENGTH)));
}
if (routeEntries == null) {
return null;
}
Iterator<RouteEntry> it = routeEntries.iterator();
while (it.hasNext()) {
routeEntry = it.next();
}
return routeEntry;
}
}
......@@ -13,550 +13,110 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.routing.impl;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.Multimaps;
import com.google.common.collect.SetMultimap;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.googlecode.concurrenttrees.common.KeyValuePair;
import com.googlecode.concurrenttrees.radix.node.concrete.DefaultByteArrayNodeFactory;
import com.googlecode.concurrenttrees.radixinverted.ConcurrentInvertedRadixTree;
import com.googlecode.concurrenttrees.radixinverted.InvertedRadixTree;
import org.apache.felix.scr.annotations.Activate;
import org.apache.felix.scr.annotations.Component;
import org.apache.felix.scr.annotations.Deactivate;
import org.apache.felix.scr.annotations.Reference;
import org.apache.felix.scr.annotations.ReferenceCardinality;
import org.apache.felix.scr.annotations.Service;
import org.onlab.packet.Ip4Address;
import org.onlab.packet.Ip6Address;
import org.onlab.packet.IpAddress;
import org.onlab.packet.IpPrefix;
import org.onlab.packet.MacAddress;
import org.onosproject.core.CoreService;
import org.onosproject.net.Host;
import org.onosproject.net.host.HostEvent;
import org.onosproject.net.host.HostListener;
import org.onosproject.net.host.HostService;
import org.onosproject.routing.RouteSourceService;
import org.onosproject.incubator.net.routing.ResolvedRoute;
import org.onosproject.incubator.net.routing.Route;
import org.onosproject.incubator.net.routing.RouteEvent;
import org.onosproject.incubator.net.routing.RouteListener;
import org.onosproject.incubator.net.routing.RouteService;
import org.onosproject.incubator.net.routing.RouteTableId;
import org.onosproject.routing.FibEntry;
import org.onosproject.routing.FibListener;
import org.onosproject.routing.FibUpdate;
import org.onosproject.routing.RouteEntry;
import org.onosproject.routing.RouteListener;
import org.onosproject.routing.RouteUpdate;
import org.onosproject.routing.RoutingService;
import org.onosproject.routing.config.RoutingConfigurationService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import static com.google.common.base.Preconditions.checkNotNull;
import static org.onosproject.routing.RouteEntry.createBinaryString;
import java.util.stream.Collectors;
/**
* This class processes route updates and maintains a Routing Information Base
* (RIB). After route updates have been processed and next hops have been
* resolved, FIB updates are sent to any listening FIB components.
* Adapts new route service interface to old RoutingService interface.
*/
@Component(immediate = true, enabled = false)
@Service
@Component(immediate = true, enabled = false)
public class Router implements RoutingService {
private static final Logger log = LoggerFactory.getLogger(Router.class);
// Route entries are stored in a radix tree.
// The key in this tree is the binary string of prefix of the route.
private InvertedRadixTree<RouteEntry> ribTable4;
private InvertedRadixTree<RouteEntry> ribTable6;
// Stores all incoming route updates in a queue.
private final BlockingQueue<Collection<RouteUpdate>> routeUpdatesQueue =
new LinkedBlockingQueue<>();
// Next-hop IP address to route entry mapping for next hops pending MAC
// resolution
private SetMultimap<IpAddress, RouteEntry> routesWaitingOnArp;
// The IPv4 address to MAC address mapping
private final Map<IpAddress, MacAddress> ip2Mac = new ConcurrentHashMap<>();
private FibListener fibComponent;
@Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
protected CoreService coreService;
@Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
protected HostService hostService;
@Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
protected RouteSourceService routeSourceService;
@Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
protected RoutingConfigurationService routingConfigurationService;
private ExecutorService bgpUpdatesExecutor;
private final HostListener hostListener = new InternalHostListener();
@Activate
public void activate() {
ribTable4 = new ConcurrentInvertedRadixTree<>(
new DefaultByteArrayNodeFactory());
ribTable6 = new ConcurrentInvertedRadixTree<>(
new DefaultByteArrayNodeFactory());
routesWaitingOnArp = Multimaps.synchronizedSetMultimap(
HashMultimap.create());
coreService.registerApplication(ROUTER_APP_ID);
bgpUpdatesExecutor = Executors.newSingleThreadExecutor(
new ThreadFactoryBuilder()
.setNameFormat("rib-updates-%d").build());
}
@Deactivate
public void deactivate() {
log.debug("Stopped");
}
protected RouteService routeService;
@Override
public void addFibListener(FibListener fibListener) {
this.fibComponent = checkNotNull(fibListener);
public void start() {
}
@Override
public void start() {
this.hostService.addListener(hostListener);
routeSourceService.start(new InternalRouteListener());
bgpUpdatesExecutor.execute(this::doUpdatesThread);
public void addFibListener(FibListener fibListener) {
routeService.addListener(new InternalRouteListener(fibListener));
}
@Override
public void stop() {
routeSourceService.stop();
this.hostService.removeListener(hostListener);
// Stop the thread(s)
bgpUpdatesExecutor.shutdownNow();
synchronized (this) {
// Cleanup all local state
ribTable4 = new ConcurrentInvertedRadixTree<>(
new DefaultByteArrayNodeFactory());
ribTable6 = new ConcurrentInvertedRadixTree<>(
new DefaultByteArrayNodeFactory());
routeUpdatesQueue.clear();
routesWaitingOnArp.clear();
ip2Mac.clear();
}
}
/**
* Entry point for route updates.
*
* @param routeUpdates collection of route updates to process
*/
private void update(Collection<RouteUpdate> routeUpdates) {
try {
routeUpdatesQueue.put(routeUpdates);
} catch (InterruptedException e) {
log.error("Interrupted while putting on routeUpdatesQueue", e);
Thread.currentThread().interrupt();
}
}
/**
* Thread for handling route updates.
*/
private void doUpdatesThread() {
boolean interrupted = false;
try {
while (!interrupted) {
try {
Collection<RouteUpdate> routeUpdates =
routeUpdatesQueue.take();
processRouteUpdates(routeUpdates);
} catch (InterruptedException e) {
log.error("Interrupted while taking from updates queue", e);
interrupted = true;
} catch (Exception e) {
log.error("exception", e);
}
}
} finally {
if (interrupted) {
Thread.currentThread().interrupt();
}
}
}
/**
* Gets all IPv4 routes from the RIB.
*
* @return all IPv4 routes from the RIB
*/
@Override
public Collection<RouteEntry> getRoutes4() {
Iterator<KeyValuePair<RouteEntry>> it =
ribTable4.getKeyValuePairsForKeysStartingWith("").iterator();
List<RouteEntry> routes = new LinkedList<>();
while (it.hasNext()) {
KeyValuePair<RouteEntry> entry = it.next();
routes.add(entry.getValue());
}
return routes;
return routeService.getAllRoutes().get(new RouteTableId("ipv4")).stream()
.map(route -> new RouteEntry(route.prefix(), route.nextHop()))
.collect(Collectors.toList());
}
/**
* Gets all IPv6 routes from the RIB.
*
* @return all IPv6 routes from the RIB
*/
@Override
public Collection<RouteEntry> getRoutes6() {
Iterator<KeyValuePair<RouteEntry>> it =
ribTable6.getKeyValuePairsForKeysStartingWith("").iterator();
List<RouteEntry> routes = new LinkedList<>();
while (it.hasNext()) {
KeyValuePair<RouteEntry> entry = it.next();
routes.add(entry.getValue());
}
return routes;
}
/**
* Finds a route in the RIB for a prefix. The prefix can be either IPv4 or
* IPv6.
*
* @param prefix the prefix to use
* @return the route if found, otherwise null
*/
RouteEntry findRibRoute(IpPrefix prefix) {
String binaryString = createBinaryString(prefix);
if (prefix.isIp4()) {
// IPv4
return ribTable4.getValueForExactKey(binaryString);
}
// IPv6
return ribTable6.getValueForExactKey(binaryString);
}
/**
* Adds a route to the RIB. The route can be either IPv4 or IPv6.
*
* @param routeEntry the route entry to use
*/
void addRibRoute(RouteEntry routeEntry) {
if (routeEntry.isIp4()) {
// IPv4
ribTable4.put(createBinaryString(routeEntry.prefix()), routeEntry);
} else {
// IPv6
ribTable6.put(createBinaryString(routeEntry.prefix()), routeEntry);
}
return routeService.getAllRoutes().get(new RouteTableId("ipv6")).stream()
.map(route -> new RouteEntry(route.prefix(), route.nextHop()))
.collect(Collectors.toList());
}
/**
* Removes a route for a prefix from the RIB. The prefix can be either IPv4
* or IPv6.
*
* @param prefix the prefix to use
* @return true if the route was found and removed, otherwise false
*/
boolean removeRibRoute(IpPrefix prefix) {
if (prefix.isIp4()) {
// IPv4
return ribTable4.remove(createBinaryString(prefix));
}
// IPv6
return ribTable6.remove(createBinaryString(prefix));
}
/**
* Processes route updates.
*
* @param routeUpdates the route updates to process
*/
void processRouteUpdates(Collection<RouteUpdate> routeUpdates) {
synchronized (this) {
Collection<IpPrefix> withdrawPrefixes = new LinkedList<>();
Collection<FibUpdate> fibUpdates = new LinkedList<>();
Collection<FibUpdate> fibWithdraws = new LinkedList<>();
for (RouteUpdate update : routeUpdates) {
switch (update.type()) {
case UPDATE:
FibEntry fib = processRouteAdd(update.routeEntry(),
withdrawPrefixes);
if (fib != null) {
fibUpdates.add(new FibUpdate(FibUpdate.Type.UPDATE, fib));
}
break;
case DELETE:
processRouteDelete(update.routeEntry(), withdrawPrefixes);
break;
default:
log.error("Unknown update Type: {}", update.type());
break;
}
}
withdrawPrefixes.forEach(p -> fibWithdraws.add(new FibUpdate(
FibUpdate.Type.DELETE, new FibEntry(p, null, null))));
if (!fibUpdates.isEmpty() || !fibWithdraws.isEmpty()) {
fibComponent.update(fibUpdates, fibWithdraws);
}
}
}
/**
* Processes adding a route entry.
* <p>
* The route entry is added to the radix tree. If there was an existing
* next hop for this prefix, but the next hop was different, then the
* old route entry is deleted.
* </p>
* <p>
* NOTE: Currently, we don't handle routes if the next hop is within the
* SDN domain.
* </p>
*
* @param routeEntry the route entry to add
* @param withdrawPrefixes the collection of accumulated prefixes whose
* intents will be withdrawn
* @return the corresponding FIB entry change, or null
*/
private FibEntry processRouteAdd(RouteEntry routeEntry,
Collection<IpPrefix> withdrawPrefixes) {
log.debug("Processing route add: {}", routeEntry);
// Find the old next-hop if we are updating an old route entry
IpAddress oldNextHop = null;
RouteEntry oldRouteEntry = findRibRoute(routeEntry.prefix());
if (oldRouteEntry != null) {
oldNextHop = oldRouteEntry.nextHop();
}
// Add the new route to the RIB
addRibRoute(routeEntry);
if (oldNextHop != null) {
if (oldNextHop.equals(routeEntry.nextHop())) {
return null; // No change
}
//
// Update an existing nexthop for the prefix.
// We need to remove the old flows for this prefix from the
// switches before the new flows are added.
//
withdrawPrefixes.add(routeEntry.prefix());
}
if (routingConfigurationService.isIpPrefixLocal(routeEntry.prefix())) {
// Route originated by local SDN domain
// We don't handle these here, reactive routing APP will handle
// these
log.debug("Own route {} to {}",
routeEntry.prefix(), routeEntry.nextHop());
return null;
}
//
// Find the MAC address of next hop router for this route entry.
// If the MAC address can not be found in ARP cache, then this prefix
// will be put in routesWaitingOnArp queue. Otherwise, generate
// a new route intent.
//
// Monitor the IP address for updates of the MAC address
hostService.startMonitoringIp(routeEntry.nextHop());
// Check if we know the MAC address of the next hop
MacAddress nextHopMacAddress = ip2Mac.get(routeEntry.nextHop());
if (nextHopMacAddress == null) {
Set<Host> hosts = hostService.getHostsByIp(routeEntry.nextHop());
if (!hosts.isEmpty()) {
nextHopMacAddress = hosts.iterator().next().mac();
}
if (nextHopMacAddress != null) {
ip2Mac.put(routeEntry.nextHop(), nextHopMacAddress);
}
}
if (nextHopMacAddress == null) {
routesWaitingOnArp.put(routeEntry.nextHop(), routeEntry);
return null;
}
return new FibEntry(routeEntry.prefix(), routeEntry.nextHop(),
nextHopMacAddress);
}
/**
* Processes the deletion of a route entry.
* <p>
* The prefix for the routing entry is removed from radix tree.
* If the operation is successful, the prefix is added to the collection
* of prefixes whose intents that will be withdrawn.
* </p>
*
* @param routeEntry the route entry to delete
* @param withdrawPrefixes the collection of accumulated prefixes whose
* intents will be withdrawn
*/
private void processRouteDelete(RouteEntry routeEntry,
Collection<IpPrefix> withdrawPrefixes) {
log.debug("Processing route delete: {}", routeEntry);
boolean isRemoved = removeRibRoute(routeEntry.prefix());
if (isRemoved) {
//
// Only withdraw intents if an entry was actually removed from the
// tree. If no entry was removed, the <prefix, nexthop> wasn't
// there so it's probably already been removed and we don't
// need to do anything.
//
withdrawPrefixes.add(routeEntry.prefix());
@Override
public RouteEntry getLongestMatchableRouteEntry(IpAddress ipAddress) {
Route route = routeService.longestPrefixMatch(ipAddress);
if (route != null) {
return new RouteEntry(route.prefix(), route.nextHop());
}
routesWaitingOnArp.remove(routeEntry.nextHop(), routeEntry);
return null;
}
/**
* Signals the Router that the MAC to IP mapping has potentially been
* updated. This has the effect of updating the MAC address for any
* installed prefixes if it has changed, as well as installing any pending
* prefixes that were waiting for MAC resolution.
*
* @param ipAddress the IP address that an event was received for
* @param macAddress the most recently known MAC address for the IP address
* Internal route listener.
*/
private void updateMac(IpAddress ipAddress, MacAddress macAddress) {
log.debug("Received updated MAC info: {} => {}", ipAddress,
macAddress);
//
// We synchronize on "this" to prevent changes to the Radix tree
// while we're pushing intents. If the tree changes, the
// tree and the intents could get out of sync.
//
synchronized (this) {
Collection<FibUpdate> submitFibEntries = new LinkedList<>();
Set<RouteEntry> routesToPush =
routesWaitingOnArp.removeAll(ipAddress);
for (RouteEntry routeEntry : routesToPush) {
// These will always be adds
RouteEntry foundRouteEntry = findRibRoute(routeEntry.prefix());
if (foundRouteEntry != null &&
foundRouteEntry.nextHop().equals(routeEntry.nextHop())) {
// We only push FIB updates if the prefix is still in the
// radix tree and the next hop is the same as our entry.
// The prefix could have been removed while we were waiting
// for the ARP, or the next hop could have changed.
submitFibEntries.add(new FibUpdate(FibUpdate.Type.UPDATE,
new FibEntry(routeEntry.prefix(),
ipAddress, macAddress)));
} else {
log.debug("{} has been revoked before the MAC was resolved",
routeEntry);
}
}
private class InternalRouteListener implements RouteListener {
if (!submitFibEntries.isEmpty()) {
fibComponent.update(submitFibEntries, Collections.emptyList());
}
private final FibListener fibListener;
ip2Mac.put(ipAddress, macAddress);
/**
* Constructor.
*
* @param fibListener FIB listener
*/
public InternalRouteListener(FibListener fibListener) {
this.fibListener = fibListener;
}
}
/**
* Listener for host events.
*/
class InternalHostListener implements HostListener {
@Override
public void event(HostEvent event) {
log.debug("Received HostEvent {}", event);
public void event(RouteEvent event) {
ResolvedRoute route = event.subject();
FibEntry entry = new FibEntry(route.prefix(), route.nextHop(), route.nextHopMac());
Host host = event.subject();
switch (event.type()) {
case HOST_ADDED:
// FALLTHROUGH
case HOST_UPDATED:
for (IpAddress ipAddress : host.ipAddresses()) {
updateMac(ipAddress, host.mac());
}
case ROUTE_ADDED:
case ROUTE_UPDATED:
fibListener.update(Collections.singleton(new FibUpdate(FibUpdate.Type.UPDATE, entry)),
Collections.emptyList());
break;
case HOST_REMOVED:
for (IpAddress ipAddress : host.ipAddresses()) {
ip2Mac.remove(ipAddress);
}
case ROUTE_REMOVED:
fibListener.update(Collections.emptyList(),
Collections.singleton(new FibUpdate(FibUpdate.Type.DELETE, entry)));
break;
default:
break;
}
}
}
/**
* Listener for route events.
*/
private class InternalRouteListener implements RouteListener {
@Override
public void update(Collection<RouteUpdate> routeUpdates) {
Router.this.update(routeUpdates);
}
}
@Override
public RouteEntry getLongestMatchableRouteEntry(IpAddress ipAddress) {
RouteEntry routeEntry = null;
Iterable<RouteEntry> routeEntries;
if (ipAddress.isIp4()) {
routeEntries = ribTable4.getValuesForKeysPrefixing(
createBinaryString(
IpPrefix.valueOf(ipAddress, Ip4Address.BIT_LENGTH)));
} else {
routeEntries = ribTable6.getValuesForKeysPrefixing(
createBinaryString(
IpPrefix.valueOf(ipAddress, Ip6Address.BIT_LENGTH)));
}
if (routeEntries == null) {
return null;
}
Iterator<RouteEntry> it = routeEntries.iterator();
while (it.hasNext()) {
routeEntry = it.next();
}
return routeEntry;
}
}
......
......@@ -33,8 +33,7 @@ import org.onlab.junit.TestUtils;
import org.onlab.junit.TestUtils.TestUtilsException;
import org.onlab.packet.Ip4Address;
import org.onlab.packet.Ip4Prefix;
import org.onosproject.routing.RouteListener;
import org.onosproject.routing.RouteUpdate;
import org.onosproject.incubator.net.routing.RouteAdminService;
import org.osgi.service.component.ComponentContext;
import java.net.InetAddress;
......@@ -48,6 +47,7 @@ import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import static org.easymock.EasyMock.createMock;
import static org.easymock.EasyMock.createNiceMock;
import static org.easymock.EasyMock.expect;
import static org.easymock.EasyMock.replay;
import static org.hamcrest.Matchers.hasSize;
......@@ -85,6 +85,8 @@ public class BgpSessionManagerTest {
// Timeout waiting for a message to be received
private static final int MESSAGE_TIMEOUT_MS = 5000; // 5s
private RouteAdminService routeService;
// The BGP Session Manager to test
private BgpSessionManager bgpSessionManager;
......@@ -102,19 +104,6 @@ public class BgpSessionManagerTest {
// The socket that the remote peers should connect to
private InetSocketAddress connectToSocket;
private final DummyRouteListener dummyRouteListener =
new DummyRouteListener();
/**
* Dummy implementation for the RouteListener interface.
*/
private class DummyRouteListener implements RouteListener {
@Override
public void update(Collection<RouteUpdate> routeUpdate) {
// Nothing to do
}
}
/**
* A class to capture the state for a BGP peer.
*/
......@@ -238,13 +227,11 @@ public class BgpSessionManagerTest {
}
@SuppressWarnings("unchecked")
private Dictionary
getDictionaryMock(ComponentContext componentContext) {
private void getDictionaryMock(ComponentContext componentContext) {
Dictionary dictionary = createMock(Dictionary.class);
expect(dictionary.get("bgpPort")).andReturn("0");
replay(dictionary);
expect(componentContext.getProperties()).andReturn(dictionary);
return dictionary;
}
@Before
......@@ -262,12 +249,16 @@ public class BgpSessionManagerTest {
// connections.
//
bgpSessionManager = new BgpSessionManager();
routeService = createNiceMock(RouteAdminService.class);
replay(routeService);
bgpSessionManager.routeService = routeService;
// NOTE: We use port 0 to bind on any available port
ComponentContext componentContext = createMock(ComponentContext.class);
Dictionary dictionary = getDictionaryMock(componentContext);
getDictionaryMock(componentContext);
replay(componentContext);
bgpSessionManager.activate(componentContext);
bgpSessionManager.start(dummyRouteListener);
// Get the port number the BGP Session Manager is listening on
Channel serverChannel = TestUtils.getField(bgpSessionManager,
......@@ -547,6 +538,7 @@ public class BgpSessionManagerTest {
withdrawnRoutes.add(Ip4Prefix.valueOf("70.0.0.0/16"));
withdrawnRoutes.add(Ip4Prefix.valueOf("80.0.0.0/24"));
withdrawnRoutes.add(Ip4Prefix.valueOf("90.0.0.0/32"));
// Write the routes
message = peer1.peerChannelHandler.prepareBgpUpdate(
NEXT_HOP1_ROUTER,
......
......@@ -39,7 +39,7 @@ import org.onosproject.net.host.HostListener;
import org.onosproject.net.host.HostService;
import org.onosproject.net.provider.ProviderId;
import org.onosproject.routing.config.RoutingConfigurationService;
import org.onosproject.routing.impl.Router.InternalHostListener;
import org.onosproject.routing.impl.DefaultRouter.InternalHostListener;
import org.onosproject.routing.RouteSourceService;
import org.onosproject.routing.FibEntry;
import org.onosproject.routing.FibListener;
......@@ -74,7 +74,7 @@ public class RouterAsyncArpTest {
DeviceId.deviceId("of:0000000000000003"),
PortNumber.portNumber(1));
private Router router;
private DefaultRouter router;
private InternalHostListener internalHostListener;
@Before
......@@ -90,7 +90,7 @@ public class RouterAsyncArpTest {
fibListener = createMock(FibListener.class);
router = new Router();
router = new DefaultRouter();
router.coreService = createNiceMock(CoreService.class);
router.hostService = hostService;
router.routingConfigurationService = routingConfigurationService;
......
......@@ -90,7 +90,7 @@ public class RouterTest {
private static final ConnectPoint SW6_ETH1 = new ConnectPoint(
DeviceId.deviceId("of:0000000000000006"),
PortNumber.portNumber(1));
private Router router;
private DefaultRouter router;
@Before
public void setUp() throws Exception {
......@@ -105,7 +105,7 @@ public class RouterTest {
fibListener = createMock(FibListener.class);
router = new Router();
router = new DefaultRouter();
router.coreService = createNiceMock(CoreService.class);
router.hostService = hostService;
router.routingConfigurationService = routingConfigurationService;
......
/*
* Copyright 2016-present Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.incubator.net.routing;
import org.onlab.packet.IpAddress;
import java.util.Collection;
import java.util.Map;
import java.util.Set;
/**
* Adapter class for the route service.
*/
public class RouteServiceAdapter implements RouteAdminService {
@Override
public void update(Collection<Route> routes) {
}
@Override
public void withdraw(Collection<Route> routes) {
}
@Override
public Map<RouteTableId, Collection<Route>> getAllRoutes() {
return null;
}
@Override
public Route longestPrefixMatch(IpAddress ip) {
return null;
}
@Override
public Collection<Route> getRoutesForNextHop(IpAddress nextHop) {
return null;
}
@Override
public Set<NextHop> getNextHops() {
return null;
}
@Override
public void addListener(RouteListener listener) {
}
@Override
public void removeListener(RouteListener listener) {
}
}