CORD-354 OF-DPA support for link-failures.

Bug fix in flowObjectives store. Adding a removeNextGroup API to the store.

Change-Id: I5890411e5b4eabdc057402687ada26e539500f8f

@@ -597,11 +597,20 @@ public class SegmentRoutingManager implements SegmentRoutingService {
                     } else if (event.type() == DeviceEvent.Type.DEVICE_ADDED ||
                             event.type() == DeviceEvent.Type.DEVICE_AVAILABILITY_CHANGED ||
                             event.type() == DeviceEvent.Type.DEVICE_UPDATED) {
-                        if (deviceService.isAvailable(((Device) event.subject()).id())) {
+                        DeviceId deviceId = ((Device) event.subject()).id();
+                        if (deviceService.isAvailable(deviceId)) {
                             log.info("Processing device event {} for available device {}",
                                      event.type(), ((Device) event.subject()).id());
                             processDeviceAdded((Device) event.subject());
+                        } /* else {
+                            if (event.type() == DeviceEvent.Type.DEVICE_AVAILABILITY_CHANGED) {
+                                // availability changed and not available - dev gone
+                                DefaultGroupHandler groupHandler = groupHandlerMap.get(deviceId);
+                                if (groupHandler != null) {
+                                    groupHandler.removeAllGroups();
+                                }
                             }
+                        }*/
                     } else if (event.type() == DeviceEvent.Type.PORT_REMOVED) {
                         processPortRemoved((Device) event.subject(),
                                            ((DeviceEvent) event).port());
@@ -655,7 +664,8 @@ public class SegmentRoutingManager implements SegmentRoutingService {
         log.debug("A link {} was removed", link.toString());
         DefaultGroupHandler groupHandler = groupHandlerMap.get(link.src().deviceId());
         if (groupHandler != null) {
-            groupHandler.portDown(link.src().port());
+            groupHandler.portDown(link.src().port(),
+                                  mastershipService.isLocalMaster(link.src().deviceId()));
         }
         log.trace("Starting optimized route population process");
         defaultRoutingHandler.populateRoutingRulesForLinkStatusChange(link);
@@ -711,7 +721,8 @@ public class SegmentRoutingManager implements SegmentRoutingService {
         log.debug("Port {} was removed", port.toString());
         DefaultGroupHandler groupHandler = groupHandlerMap.get(device.id());
         if (groupHandler != null) {
-            groupHandler.portDown(port.number());
+            groupHandler.portDown(port.number(),
+                                  mastershipService.isLocalMaster(device.id()));
         }
     }
 

@@ -32,11 +32,13 @@ import org.onlab.packet.Ip4Prefix;
 import org.onlab.packet.IpPrefix;
 import org.onlab.packet.MacAddress;
 import org.onlab.packet.MplsLabel;
+import org.onlab.packet.VlanId;
 import org.onlab.util.KryoNamespace;
 import org.onosproject.core.ApplicationId;
 import org.onosproject.net.DeviceId;
 import org.onosproject.net.Link;
 import org.onosproject.net.PortNumber;
+import org.onosproject.net.flow.DefaultTrafficSelector;
 import org.onosproject.net.flow.DefaultTrafficTreatment;
 import org.onosproject.net.flow.TrafficSelector;
 import org.onosproject.net.flow.TrafficTreatment;
@@ -49,6 +51,7 @@ import org.onosproject.net.flowobjective.ObjectiveError;
 import org.onosproject.net.group.DefaultGroupKey;
 import org.onosproject.net.group.GroupKey;
 import org.onosproject.net.link.LinkService;
+import org.onosproject.segmentrouting.SegmentRoutingManager;
 import org.onosproject.segmentrouting.config.DeviceConfigNotFoundException;
 import org.onosproject.segmentrouting.config.DeviceProperties;
 import org.onosproject.store.service.EventuallyConsistentMap;
@@ -71,9 +74,11 @@ public class DefaultGroupHandler {
     protected MacAddress nodeMacAddr = null;
     protected LinkService linkService;
     protected FlowObjectiveService flowObjectiveService;
-
+    // local store for neighbor-device-ids and the set of ports on this device
+    // that connect to the same neighbor
     protected ConcurrentHashMap<DeviceId, Set<PortNumber>> devicePortMap =
             new ConcurrentHashMap<>();
+    //local store for ports on this device connected to neighbor-device-id
     protected ConcurrentHashMap<PortNumber, DeviceId> portDeviceMap =
             new ConcurrentHashMap<>();
     protected EventuallyConsistentMap<
@@ -225,6 +230,9 @@ public class DefaultGroupHandler {
                 deviceId,
                 nsSet);
         for (NeighborSet ns : nsSet) {
+            Integer nextId = nsNextObjStore.
+                    get(new NeighborSetNextObjectiveStoreKey(deviceId, ns));
+            if (nextId != null && isMaster) {
                 // Create the new bucket to be updated
                 TrafficTreatment.Builder tBuilder =
                         DefaultTrafficTreatment.builder();
@@ -236,15 +244,19 @@ public class DefaultGroupHandler {
                         .copyTtlOut()
                         .setMpls(MplsLabel.mplsLabel(ns.getEdgeLabel()));
                 }
-
-            Integer nextId = nsNextObjStore.
-                    get(new NeighborSetNextObjectiveStoreKey(deviceId, ns));
-            if (nextId != null && isMaster) {
-                NextObjective.Builder nextObjBuilder = DefaultNextObjective
-                        .builder().withId(nextId)
-                        .withType(NextObjective.Type.HASHED).fromApp(appId);
-
-                nextObjBuilder.addTreatment(tBuilder.build());
+                // setup metadata to pass to nextObjective - indicate the vlan on egress
+                // if needed by the switch pipeline. Since hashed next-hops are always to
+                // other neighboring routers, there is no subnet assigned on those ports.
+                TrafficSelector.Builder metabuilder = DefaultTrafficSelector.builder();
+                metabuilder.matchVlanId(
+                    VlanId.vlanId(SegmentRoutingManager.ASSIGNED_VLAN_NO_SUBNET));
+
+                NextObjective.Builder nextObjBuilder = DefaultNextObjective.builder()
+                        .withId(nextId)
+                        .withType(NextObjective.Type.HASHED)
+                        .addTreatment(tBuilder.build())
+                        .withMeta(metabuilder.build())
+                        .fromApp(appId);
                 log.info("**linkUp in device {}: Adding Bucket "
                         + "with Port {} to next object id {}",
                         deviceId,
@@ -253,6 +265,18 @@ public class DefaultGroupHandler {
                 NextObjective nextObjective = nextObjBuilder.
                         addToExisting(new SRNextObjectiveContext(deviceId));
                 flowObjectiveService.next(deviceId, nextObjective);
+
+                // the addition of a bucket may actually change the neighborset
+                // update the global store
+                /*
+                Set<DeviceId> neighbors = new HashSet<DeviceId>(ns.getDeviceIds());
+                boolean newadd = neighbors.add(newLink.dst().deviceId());
+                if (newadd) {
+                    NeighborSet nsnew = new NeighborSet(neighbors, ns.getEdgeLabel());
+                    nsNextObjStore.put(new NeighborSetNextObjectiveStoreKey(deviceId, nsnew),
+                                       nextId);
+                    nsNextObjStore.remove(new NeighborSetNextObjectiveStoreKey(deviceId, ns));
+                }*/
             } else if (isMaster) {
                 log.warn("linkUp in device {}, but global store has no record "
                         + "for neighbor-set {}", deviceId, ns);
@@ -265,7 +289,7 @@ public class DefaultGroupHandler {
      *
      * @param port port number that has gone down
      */
-    public void portDown(PortNumber port) {
+    public void portDown(PortNumber port, boolean isMaster) {
         if (portDeviceMap.get(port) == null) {
             log.warn("portDown: unknown port");
             return;
@@ -292,10 +316,17 @@ public class DefaultGroupHandler {
                 .filter((ns) -> (ns.getDeviceIds()
                         .contains(portDeviceMap.get(port))))
                 .collect(Collectors.toSet());
-        log.trace("portDown: nsNextObjStore contents for device {}:",
-                  deviceId,
-                  nsSet);
+        log.debug("portDown: nsNextObjStore contents for device {}:{}",
+                  deviceId, nsSet);
         for (NeighborSet ns : nsSet) {
+            Integer nextId = nsNextObjStore.
+                    get(new NeighborSetNextObjectiveStoreKey(deviceId, ns));
+            if (nextId != null && isMaster) {
+                log.info("**portDown in device {}: Removing Bucket "
+                        + "with Port {} to next object id {}",
+                        deviceId,
+                        port,
+                        nextId);
                 // Create the bucket to be removed
                 TrafficTreatment.Builder tBuilder = DefaultTrafficTreatment
                         .builder();
@@ -307,25 +338,28 @@ public class DefaultGroupHandler {
                         .copyTtlOut()
                         .setMpls(MplsLabel.mplsLabel(ns.getEdgeLabel()));
                 }
-
-            Integer nextId = nsNextObjStore.
-                    get(new NeighborSetNextObjectiveStoreKey(deviceId, ns));
-            if (nextId != null) {
                 NextObjective.Builder nextObjBuilder = DefaultNextObjective
-                        .builder().withType(NextObjective.Type.SIMPLE).withId(nextId).fromApp(appId);
+                        .builder()
+                        .withType(NextObjective.Type.HASHED) //same as original
+                        .withId(nextId)
+                        .fromApp(appId)
+                        .addTreatment(tBuilder.build());
+                NextObjective nextObjective = nextObjBuilder.
+                        removeFromExisting(new SRNextObjectiveContext(deviceId));
 
-                nextObjBuilder.addTreatment(tBuilder.build());
+                flowObjectiveService.next(deviceId, nextObjective);
 
-                log.info("**portDown in device {}: Removing Bucket "
-                        + "with Port {} to next object id {}",
-                        deviceId,
-                        port,
+                // the removal of a bucket may actually change the neighborset
+                // update the global store
+                /*
+                Set<DeviceId> neighbors = new HashSet<DeviceId>(ns.getDeviceIds());
+                boolean removed = neighbors.remove(portDeviceMap.get(port));
+                if (removed) {
+                    NeighborSet nsnew = new NeighborSet(neighbors, ns.getEdgeLabel());
+                    nsNextObjStore.put(new NeighborSetNextObjectiveStoreKey(deviceId, nsnew),
                                        nextId);
-                // should do removefromexisting and only if master
-                /*NextObjective nextObjective = nextObjBuilder.
-                        remove(new SRNextObjectiveContext(deviceId));
-
-                flowObjectiveService.next(deviceId, nextObjective);*/
+                    nsNextObjStore.remove(new NeighborSetNextObjectiveStoreKey(deviceId, ns));
+                }*/
             }
 
         }
@@ -718,6 +752,22 @@ public class DefaultGroupHandler {
         return false;
     }
 
+    public void removeAllGroups() {
+        for (Map.Entry<NeighborSetNextObjectiveStoreKey, Integer> entry:
+                nsNextObjStore.entrySet()) {
+            removeGroup(entry.getValue());
+        }
+        for (Map.Entry<PortNextObjectiveStoreKey, Integer> entry:
+                portNextObjStore.entrySet()) {
+            removeGroup(entry.getValue());
+        }
+        for (Map.Entry<SubnetNextObjectiveStoreKey, Integer> entry:
+                subnetNextObjStore.entrySet()) {
+            removeGroup(entry.getValue());
+        }
+        // should probably clean local stores port-neighbor
+    }
+
     protected static class SRNextObjectiveContext implements ObjectiveContext {
         final DeviceId deviceId;
 

@@ -27,7 +27,8 @@ public interface FlowObjectiveStore
         extends Store<ObjectiveEvent, FlowObjectiveStoreDelegate> {
 
     /**
-     * Adds a NextGroup to the store.
+     * Adds a NextGroup to the store, by mapping it to the nextId as key,
+     * and replacing any previous mapping.
      *
      * @param nextId an integer
      * @param group a next group opaque object
@@ -36,12 +37,22 @@ public interface FlowObjectiveStore
 
     /**
      * Fetch a next group from the store.
-     * @param nextId an integer
-     * @return a next group
+     *
+     * @param nextId an integer used as key
+     * @return a next group, or null if group was not found
      */
     NextGroup getNextGroup(Integer nextId);
 
     /**
+     * Remove a next group mapping from the store.
+     *
+     * @param nextId  the key to remove from the store.
+     * @return the next group which mapped to the nextId and is now removed, or
+     *          null if no group mapping existed in the store
+     */
+    NextGroup removeNextGroup(Integer nextId);
+
+    /**
      * Allocates a next objective id. This id is globally unique
      *
      * @return an integer

@@ -48,6 +48,7 @@ import org.onosproject.net.flowobjective.NextObjective;
 import org.onosproject.net.flowobjective.Objective;
 import org.onosproject.net.flowobjective.ObjectiveError;
 import org.onosproject.net.flowobjective.ObjectiveEvent;
+import org.onosproject.net.flowobjective.ObjectiveEvent.Type;
 import org.onosproject.net.group.GroupService;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -381,6 +382,7 @@ public class FlowObjectiveManager implements FlowObjectiveService {
     private class InternalStoreDelegate implements FlowObjectiveStoreDelegate {
         @Override
         public void notify(ObjectiveEvent event) {
+            if (event.type() == Type.ADD) {
                 log.debug("Received notification of obj event {}", event);
                 Set<PendingNext> pending = pendingForwards.remove(event.subject());
 
@@ -390,10 +392,9 @@ public class FlowObjectiveManager implements FlowObjectiveService {
                 }
 
                 log.debug("Processing pending forwarding objectives {}", pending.size());
-
                 pending.forEach(p -> getDevicePipeliner(p.deviceId())
                                 .forward(p.forwardingObjective()));
-
+            }
         }
     }
 

@@ -79,10 +79,9 @@ public class DistributedFlowObjectiveStore
         log.info("Stopped");
     }
 
-
     @Override
     public void putNextGroup(Integer nextId, NextGroup group) {
-        nextGroups.putIfAbsent(nextId, group.data());
+        nextGroups.put(nextId, group.data());
         notifyDelegate(new ObjectiveEvent(ObjectiveEvent.Type.ADD, nextId));
     }
 
@@ -96,6 +95,16 @@ public class DistributedFlowObjectiveStore
     }
 
     @Override
+    public NextGroup removeNextGroup(Integer nextId) {
+        Versioned<byte[]> versionGroup = nextGroups.remove(nextId);
+        if (versionGroup != null) {
+            notifyDelegate(new ObjectiveEvent(ObjectiveEvent.Type.REMOVE, nextId));
+            return new DefaultNextGroup(versionGroup.value());
+        }
+        return null;
+    }
+
+    @Override
     public int allocateNextId() {
         return (int) nextIds.incrementAndGet();
     }

@@ -71,7 +71,6 @@ public class CpqdOFDPA2Pipeline extends OFDPA2Pipeline {
      * (non-Javadoc)
      * @see org.onosproject.driver.pipeline.OFDPA2Pipeline#processVlanIdFilter
      */
-
     @Override
     protected List<FlowRule> processVlanIdFilter(PortCriterion portCriterion,
                                                  VlanIdCriterion vidCriterion,
@@ -267,7 +266,8 @@ public class CpqdOFDPA2Pipeline extends OFDPA2Pipeline {
         }
 
         if (fwd.nextId() != null) {
-            NextGroup next = flowObjectiveStore.getNextGroup(fwd.nextId());
+            NextGroup next = getGroupForNextObjective(fwd.nextId());
+            if (next != null) {
                 List<Deque<GroupKey>> gkeys = appKryo.deserialize(next.data());
                 // we only need the top level group's key to point the flow to it
                 Group group = groupService.getGroup(deviceId, gkeys.get(0).peekFirst());
@@ -278,6 +278,7 @@ public class CpqdOFDPA2Pipeline extends OFDPA2Pipeline {
                 }
                 tb.deferred().group(group.id());
             }
+        }
         tb.transition(ACL_TABLE);
         FlowRule.Builder ruleBuilder = DefaultFlowRule.builder()
                 .fromApp(fwd.appId())