In this commit:
Bug fix where filtering objectives are not installed due to available ports becoming enabled later.
Bug fix where flow objective store had no listener for notifications from drivers across multiple instances of the controller.
NPE fix in ofdpa driver for non-existing groups.
Preventing ofdpa driver from sending spurious pass notification to app.
Incrementing retry filter timer from 1 to 5 secs in default routing handler.
Made several debug messages clearer.
Change-Id: I828671ee4c8bcfe03c946d051e1d1aac9d8f68dd
diff --git a/apps/segmentrouting/src/main/java/org/onosproject/segmentrouting/DefaultRoutingHandler.java b/apps/segmentrouting/src/main/java/org/onosproject/segmentrouting/DefaultRoutingHandler.java
index 1324795..82f991a 100644
--- a/apps/segmentrouting/src/main/java/org/onosproject/segmentrouting/DefaultRoutingHandler.java
+++ b/apps/segmentrouting/src/main/java/org/onosproject/segmentrouting/DefaultRoutingHandler.java
@@ -47,7 +47,7 @@
* routing rule population.
*/
public class DefaultRoutingHandler {
- private static final int MAX_RETRY_ATTEMPTS = 5;
+ private static final int MAX_RETRY_ATTEMPTS = 25;
private static final String ECMPSPG_MISSING = "ECMP shortest path graph not found";
private static Logger log = LoggerFactory.getLogger(DefaultRoutingHandler.class);
@@ -212,11 +212,11 @@
log.trace("repopulateRoutingRulesForRoutes: running ECMP graph for device {}", link.get(0));
EcmpShortestPathGraph ecmpSpg = new EcmpShortestPathGraph(link.get(0), srManager);
if (populateEcmpRoutingRules(link.get(0), ecmpSpg, ImmutableSet.of())) {
- log.debug("Populating flow rules from {} to all is successful",
+ log.debug("Populating flow rules from all to dest:{} is successful",
link.get(0));
currentEcmpSpgMap.put(link.get(0), ecmpSpg);
} else {
- log.warn("Failed to populate the flow rules from {} to all", link.get(0));
+ log.warn("Failed to populate the flow rules from all to dest:{}", link.get(0));
return false;
}
} else {
@@ -463,9 +463,9 @@
/**
* Populate ECMP rules for subnets from target to destination via nexthops.
*
- * @param targetSw Device ID of target switch
- * @param destSw Device ID of destination switch
- * @param nextHops List of next hops
+ * @param targetSw Device ID of target switch in which rules will be programmed
+ * @param destSw Device ID of final destination switch to which the rules will forward
+ * @param nextHops List of next hops via which destSw will be reached
* @param subnets Subnets to be populated. If empty, populate all configured subnets.
* @return true if succeed
*/
@@ -647,6 +647,8 @@
@Override
public void run() {
+ log.info("RETRY FILTER ATTEMPT# {} for dev:{}",
+ MAX_RETRY_ATTEMPTS - attempts, devId);
boolean success = rulePopulator.populateRouterMacVlanFilters(devId);
if (!success && --attempts > 0) {
executorService.schedule(this, 200, TimeUnit.MILLISECONDS);
diff --git a/apps/segmentrouting/src/main/java/org/onosproject/segmentrouting/RoutingRulePopulator.java b/apps/segmentrouting/src/main/java/org/onosproject/segmentrouting/RoutingRulePopulator.java
index 0ddffa4..3542ec9 100644
--- a/apps/segmentrouting/src/main/java/org/onosproject/segmentrouting/RoutingRulePopulator.java
+++ b/apps/segmentrouting/src/main/java/org/onosproject/segmentrouting/RoutingRulePopulator.java
@@ -244,7 +244,8 @@
}
/**
- * Populates IP flow rules for the router IP address.
+ * Populates IP flow rules for an IP prefix in the target device. The prefix
+ * is reachable via destination device.
*
* @param deviceId target device ID to set the rules
* @param ipPrefix the IP address of the destination router
@@ -312,9 +313,11 @@
ipPrefix,
deviceId);
ObjectiveContext context = new DefaultObjectiveContext(
- (objective) -> log.debug("IP rule for router {} populated", ipPrefix),
+ (objective) -> log.debug("IP rule for router {} populated in dev:{}",
+ ipPrefix, deviceId),
(objective, error) ->
- log.warn("Failed to populate IP rule for router {}: {}", ipPrefix, error));
+ log.warn("Failed to populate IP rule for router {}: {} in dev:{}",
+ ipPrefix, error, deviceId));
srManager.flowObjectiveService.forward(deviceId, fwdBuilder.add(context));
rulePopulationCounter.incrementAndGet();
@@ -356,14 +359,15 @@
}
/**
- * Populates MPLS flow rules to all routers.
+ * Populates MPLS flow rules in the target device to point towards the
+ * destination device.
*
- * @param deviceId target device ID of the switch to set the rules
+ * @param targetSwId target device ID of the switch to set the rules
* @param destSwId destination switch device ID
* @param nextHops next hops switch ID list
* @return true if all rules are set successfully, false otherwise
*/
- public boolean populateMplsRule(DeviceId deviceId, DeviceId destSwId,
+ public boolean populateMplsRule(DeviceId targetSwId, DeviceId destSwId,
Set<DeviceId> nextHops) {
int segmentId;
try {
@@ -392,11 +396,11 @@
// If the next hop is the destination router for the segment, do pop
if (nextHops.size() == 1 && destSwId.equals(nextHops.toArray()[0])) {
log.debug("populateMplsRule: Installing MPLS forwarding objective for "
- + "label {} in switch {} with pop", segmentId, deviceId);
+ + "label {} in switch {} with pop", segmentId, targetSwId);
// bos pop case (php)
ForwardingObjective.Builder fwdObjBosBuilder =
- getMplsForwardingObjective(deviceId,
+ getMplsForwardingObjective(targetSwId,
nextHops,
true,
true,
@@ -416,13 +420,13 @@
} else {
// next hop is not destination, SR CONTINUE case (swap with self)
log.debug("Installing MPLS forwarding objective for "
- + "label {} in switch {} without pop", segmentId, deviceId);
+ + "label {} in switch {} without pop", segmentId, targetSwId);
// continue case with bos - this does get triggered in edge routers
// and in core routers - driver can handle depending on availability
// of MPLS ECMP or not
ForwardingObjective.Builder fwdObjBosBuilder =
- getMplsForwardingObjective(deviceId,
+ getMplsForwardingObjective(targetSwId,
nextHops,
false,
true,
@@ -441,17 +445,23 @@
false); */
}
-
+ // XXX when other cases above are implemented check for validity of
+ // debug messages below
for (ForwardingObjective.Builder fwdObjBuilder : fwdObjBuilders) {
((Builder) ((Builder) fwdObjBuilder.fromApp(srManager.appId)
.makePermanent()).withSelector(selector)
.withPriority(SegmentRoutingService.DEFAULT_PRIORITY))
.withFlag(ForwardingObjective.Flag.SPECIFIC);
ObjectiveContext context = new DefaultObjectiveContext(
- (objective) -> log.debug("MPLS rule for SID {} populated", segmentId),
+ (objective) -> log.debug("MPLS rule {} for SID {} populated in dev:{} ",
+ objective.id(), segmentId, targetSwId),
(objective, error) ->
- log.warn("Failed to populate MPLS rule for SID {}: {}", segmentId, error));
- srManager.flowObjectiveService.forward(deviceId, fwdObjBuilder.add(context));
+ log.warn("Failed to populate MPLS rule {} for SID {}: {} in dev:{}",
+ objective.id(), segmentId, error, targetSwId));
+ ForwardingObjective fob = fwdObjBuilder.add(context);
+ log.debug("Sending MPLS fwd obj {} for SID {}-> next {} in sw: {}",
+ fob.id(), segmentId, fob.nextId(), targetSwId);
+ srManager.flowObjectiveService.forward(targetSwId, fob);
rulePopulationCounter.incrementAndGet();
}
@@ -493,13 +503,16 @@
// packets will be hashed or not.
fwdBuilder.withTreatment(tbuilder.build());
NeighborSet ns = new NeighborSet(nextHops);
- log.debug("Trying to get a nextObjid for mpls rule on device:{} to ns:{}",
+ log.debug("Trying to get a nextObjId for mpls rule on device:{} to ns:{}",
deviceId, ns);
int nextId = srManager.getNextObjectiveId(deviceId, ns, meta);
if (nextId <= 0) {
log.warn("No next objective in {} for ns: {}", deviceId, ns);
return null;
+ } else {
+ log.debug("nextObjId found:{} for mpls rule on device:{} to ns:{}",
+ nextId, deviceId, ns);
}
fwdBuilder.nextStep(nextId);
@@ -537,40 +550,54 @@
deviceId);
return false;
}
-
+ int disabledPorts = 0, suppressedPorts = 0, filteredPorts = 0;
for (Port port : devPorts) {
ConnectPoint connectPoint = new ConnectPoint(deviceId, port.number());
// TODO: Handles dynamic port events when we are ready for dynamic config
SegmentRoutingAppConfig appConfig = srManager.cfgService
.getConfig(srManager.appId, SegmentRoutingAppConfig.class);
- if ((appConfig == null || !appConfig.suppressSubnet().contains(connectPoint)) &&
- port.isEnabled()) {
- Ip4Prefix portSubnet = config.getPortSubnet(deviceId, port.number());
- VlanId assignedVlan = (portSubnet == null)
- ? VlanId.vlanId(SegmentRoutingManager.ASSIGNED_VLAN_NO_SUBNET)
- : srManager.getSubnetAssignedVlanId(deviceId, portSubnet);
+ if (!port.isEnabled()) {
+ disabledPorts++;
+ continue;
+ }
+ if (appConfig != null && appConfig.suppressSubnet().contains(connectPoint)) {
+ suppressedPorts++;
+ continue;
+ }
+ Ip4Prefix portSubnet = config.getPortSubnet(deviceId, port.number());
+ VlanId assignedVlan = (portSubnet == null)
+ ? VlanId.vlanId(SegmentRoutingManager.ASSIGNED_VLAN_NO_SUBNET)
+ : srManager.getSubnetAssignedVlanId(deviceId, portSubnet);
- FilteringObjective.Builder fob = DefaultFilteringObjective.builder();
- fob.withKey(Criteria.matchInPort(port.number()))
+ FilteringObjective.Builder fob = DefaultFilteringObjective.builder();
+ fob.withKey(Criteria.matchInPort(port.number()))
.addCondition(Criteria.matchEthDst(deviceMac))
.addCondition(Criteria.matchVlanId(VlanId.NONE))
.withPriority(SegmentRoutingService.DEFAULT_PRIORITY);
- // vlan assignment is valid only if this instance is master
- if (srManager.mastershipService.isLocalMaster(deviceId)) {
- TrafficTreatment tt = DefaultTrafficTreatment.builder()
- .pushVlan().setVlanId(assignedVlan).build();
- fob.withMeta(tt);
- }
- fob.permit().fromApp(srManager.appId);
- log.debug("Sending filtering objective for dev/port:{}/{}", deviceId, port);
- ObjectiveContext context = new DefaultObjectiveContext(
- (objective) -> log.debug("Filter for {} populated", connectPoint),
- (objective, error) ->
- log.warn("Failed to populate filter for {}: {}", connectPoint, error));
- srManager.flowObjectiveService.filter(deviceId, fob.add(context));
+ // vlan assignment is valid only if this instance is master
+ if (srManager.mastershipService.isLocalMaster(deviceId)) {
+ TrafficTreatment tt = DefaultTrafficTreatment.builder()
+ .pushVlan().setVlanId(assignedVlan).build();
+ fob.withMeta(tt);
}
+ fob.permit().fromApp(srManager.appId);
+ log.debug("Sending filtering objective for dev/port:{}/{}", deviceId, port);
+ filteredPorts++;
+ ObjectiveContext context = new DefaultObjectiveContext(
+ (objective) -> log.debug("Filter for {} populated", connectPoint),
+ (objective, error) ->
+ log.warn("Failed to populate filter for {}: {}", connectPoint, error));
+ srManager.flowObjectiveService.filter(deviceId, fob.add(context));
}
- return true;
+ log.info("Filtering on dev:{}, disabledPorts:{}, suppressedPorts:{}, filteredPorts:{}",
+ deviceId, disabledPorts, suppressedPorts, filteredPorts);
+ // XXX With this check, there is a chance that not all the ports that
+ // should be filtered actually get filtered as long as one of them does.
+ // Note there is no PORT_UPDATED event that makes the port go from disabled
+ // to enabled state, because the ports comes enabled from the switch.
+ // Check ONOS core, where the port becoming available and being declared
+ // enabled is possibly not atomic.
+ return (filteredPorts > 0) ? true : false;
}
/**
diff --git a/core/net/src/main/java/org/onosproject/net/flowobjective/impl/FlowObjectiveManager.java b/core/net/src/main/java/org/onosproject/net/flowobjective/impl/FlowObjectiveManager.java
index b6c748a..6933c07 100644
--- a/core/net/src/main/java/org/onosproject/net/flowobjective/impl/FlowObjectiveManager.java
+++ b/core/net/src/main/java/org/onosproject/net/flowobjective/impl/FlowObjectiveManager.java
@@ -239,7 +239,8 @@
private boolean queueObjective(DeviceId deviceId, ForwardingObjective fwd) {
if (fwd.nextId() != null &&
flowObjectiveStore.getNextGroup(fwd.nextId()) == null) {
- log.trace("Queuing forwarding objective for nextId {}", fwd.nextId());
+ log.debug("Queuing forwarding objective {} for nextId {} meant for device {}",
+ fwd.id(), fwd.nextId(), deviceId);
// TODO: change to computeIfAbsent
Set<PendingNext> newset = Collections.newSetFromMap(
new ConcurrentHashMap<PendingNext, Boolean>());
@@ -398,7 +399,7 @@
Set<PendingNext> pending = pendingForwards.remove(event.subject());
if (pending == null) {
- log.warn("Nothing pending for this obj event {}", event);
+ log.debug("Nothing pending for this obj event {}", event);
return;
}
@@ -457,7 +458,7 @@
public List<String> getNextMappings() {
List<String> mappings = new ArrayList<>();
Map<Integer, NextGroup> allnexts = flowObjectiveStore.getAllGroups();
- // XXX if the NextGroup upon decoding stored info of the deviceId
+ // XXX if the NextGroup after de-serialization actually stored info of the deviceId
// then info on any nextObj could be retrieved from one controller instance.
// Right now the drivers on one instance can only fetch for next-ids that came
// to them.
diff --git a/core/store/dist/src/main/java/org/onosproject/store/flowobjective/impl/DistributedFlowObjectiveStore.java b/core/store/dist/src/main/java/org/onosproject/store/flowobjective/impl/DistributedFlowObjectiveStore.java
index 957523c..b5a57cd 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/flowobjective/impl/DistributedFlowObjectiveStore.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/flowobjective/impl/DistributedFlowObjectiveStore.java
@@ -30,15 +30,22 @@
import org.onosproject.store.AbstractStore;
import org.onosproject.store.service.AtomicCounter;
import org.onosproject.store.service.ConsistentMap;
+import org.onosproject.store.service.MapEvent;
+import org.onosproject.store.service.MapEventListener;
import org.onosproject.store.service.Serializer;
import org.onosproject.store.service.StorageService;
import org.onosproject.store.service.Versioned;
import org.slf4j.Logger;
+import static org.onlab.util.Tools.groupedThreads;
import static org.slf4j.LoggerFactory.getLogger;
import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
/**
* Manages the inventory of created next groups.
@@ -57,9 +64,16 @@
protected StorageService storageService;
private AtomicCounter nextIds;
+ private MapEventListener<Integer, byte[]> mapListener = new NextGroupListener();
+ // event queue to separate map-listener threads from event-handler threads (tpool)
+ private BlockingQueue<ObjectiveEvent> eventQ;
+ private ExecutorService tpool;
@Activate
public void activate() {
+ tpool = Executors.newFixedThreadPool(4, groupedThreads("onos/flobj-notifier", "%d", log));
+ eventQ = new LinkedBlockingQueue<ObjectiveEvent>();
+ tpool.execute(new FlowObjectiveNotifier());
nextGroups = storageService.<Integer, byte[]>consistentMapBuilder()
.withName("flowobjective-groups")
.withSerializer(Serializer.using(
@@ -68,7 +82,7 @@
.register(Versioned.class)
.build("DistributedFlowObjectiveStore")))
.build();
-
+ nextGroups.addListener(mapListener);
nextIds = storageService.getAtomicCounter("next-objective-counter");
log.info("Started");
}
@@ -76,6 +90,7 @@
@Deactivate
public void deactivate() {
+ tpool.shutdown();
log.info("Stopped");
}
@@ -120,4 +135,37 @@
public int allocateNextId() {
return (int) nextIds.incrementAndGet();
}
+
+ private class FlowObjectiveNotifier implements Runnable {
+ @Override
+ public void run() {
+ try {
+ while (!Thread.currentThread().isInterrupted()) {
+ notifyDelegate(eventQ.take());
+ }
+ } catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+
+ private class NextGroupListener implements MapEventListener<Integer, byte[]> {
+ @Override
+ public void event(MapEvent<Integer, byte[]> event) {
+ switch (event.type()) {
+ case INSERT:
+ eventQ.add(new ObjectiveEvent(ObjectiveEvent.Type.ADD, event.key()));
+ break;
+ case REMOVE:
+ eventQ.add(new ObjectiveEvent(ObjectiveEvent.Type.REMOVE, event.key()));
+ break;
+ case UPDATE:
+ // TODO Introduce UPDATE ObjectiveEvent when the map is being updated
+ break;
+ default:
+ break;
+ }
+ }
+ }
+
}
diff --git a/drivers/default/src/main/java/org/onosproject/driver/pipeline/Ofdpa2Pipeline.java b/drivers/default/src/main/java/org/onosproject/driver/pipeline/Ofdpa2Pipeline.java
index 6f42dbc..d1b285b 100644
--- a/drivers/default/src/main/java/org/onosproject/driver/pipeline/Ofdpa2Pipeline.java
+++ b/drivers/default/src/main/java/org/onosproject/driver/pipeline/Ofdpa2Pipeline.java
@@ -182,6 +182,12 @@
FlowRuleOperations.Builder flowOpsBuilder = FlowRuleOperations.builder();
rules = processForward(fwd);
+ if (rules == null || rules.isEmpty()) {
+ // Assumes fail message has already been generated to the objective
+ // context. Returning here prevents spurious pass message to be
+ // generated by FlowRule service for empty flowOps.
+ return;
+ }
switch (fwd.op()) {
case ADD:
rules.stream()
@@ -748,7 +754,7 @@
* returned if there is an issue in processing the objective.
*/
protected Collection<FlowRule> processSpecific(ForwardingObjective fwd) {
- log.trace("Processing specific fwd objective:{} in dev:{} with next:{}",
+ log.debug("Processing specific fwd objective:{} in dev:{} with next:{}",
fwd.id(), deviceId, fwd.nextId());
boolean isEthTypeObj = isSupportedEthTypeObjective(fwd);
boolean isEthDstObj = isSupportedEthDstObjective(fwd);
@@ -885,8 +891,8 @@
if (fwd.nextId() != null) {
if (forTableId == MPLS_TABLE_1 && !popMpls) {
log.warn("SR CONTINUE case cannot be handled as MPLS ECMP "
- + "is not implemented in OF-DPA yet. Aborting this flow "
- + "in this device {}", deviceId);
+ + "is not implemented in OF-DPA yet. Aborting this flow {} -> next:{}"
+ + "in this device {}", fwd.id(), fwd.nextId(), deviceId);
// XXX We could convert to forwarding to a single-port, via a
// MPLS interface, or a MPLS SWAP (with-same) but that would
// have to be handled in the next-objective. Also the pop-mpls
@@ -907,6 +913,11 @@
return Collections.emptySet();
}
tb.deferred().group(group.id());
+ } else {
+ log.warn("Cannot find group for nextId:{} in dev:{}. Aborting fwd:{}",
+ fwd.nextId(), deviceId, fwd.id());
+ fail(fwd, ObjectiveError.FLOWINSTALLATIONFAILED);
+ return Collections.emptySet();
}
}
tb.transition(ACL_TABLE);
@@ -1063,7 +1074,7 @@
for (GroupKey gk : gkd) {
Group g = groupService.getGroup(deviceId, gk);
if (g == null) {
- gchain.append(" ERROR").append(" -->");
+ gchain.append(" NoGrp").append(" -->");
continue;
}
gchain.append(" 0x").append(Integer.toHexString(g.id().id()))
@@ -1071,7 +1082,12 @@
lastGroup = g;
}
// add port information for last group in group-chain
- for (Instruction i: lastGroup.buckets().buckets().get(0).treatment().allInstructions()) {
+ List<Instruction> lastGroupIns = new ArrayList<Instruction>();
+ if (gchain != null) {
+ lastGroupIns = lastGroup.buckets().buckets().get(0)
+ .treatment().allInstructions();
+ }
+ for (Instruction i: lastGroupIns) {
if (i instanceof OutputInstruction) {
gchain.append(" port:").append(((OutputInstruction) i).port());
}