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

@@ -47,7 +47,7 @@ import static com.google.common.base.Preconditions.checkNotNull;
  * 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 @@ public class DefaultRoutingHandler {
                 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 @@ public class DefaultRoutingHandler {
     /**
      * 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 @@ public class DefaultRoutingHandler {
 
         @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);

@@ -244,7 +244,8 @@ public class RoutingRulePopulator {
     }
 
     /**
-     * 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 @@ public class RoutingRulePopulator {
                 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 @@ public class RoutingRulePopulator {
     }
 
     /**
-     * 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 @@ public class RoutingRulePopulator {
         // 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 @@ public class RoutingRulePopulator {
         } 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 @@ public class RoutingRulePopulator {
                                                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 @@ public class RoutingRulePopulator {
         // 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 @@ public class RoutingRulePopulator {
                      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;
     }
 
     /**

@@ -239,7 +239,8 @@ public class FlowObjectiveManager implements FlowObjectiveService {
     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 @@ public class FlowObjectiveManager implements FlowObjectiveService {
                 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 class FlowObjectiveManager implements FlowObjectiveService {
     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.

@@ -30,15 +30,22 @@ import org.onosproject.net.flowobjective.ObjectiveEvent;
 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 @@ public class DistributedFlowObjectiveStore
     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 @@ public class DistributedFlowObjectiveStore
                                 .register(Versioned.class)
                                 .build("DistributedFlowObjectiveStore")))
                 .build();
-
+        nextGroups.addListener(mapListener);
         nextIds = storageService.getAtomicCounter("next-objective-counter");
         log.info("Started");
     }
@@ -76,6 +90,7 @@ public class DistributedFlowObjectiveStore
 
     @Deactivate
     public void deactivate() {
+        tpool.shutdown();
         log.info("Stopped");
     }
 
@@ -120,4 +135,37 @@ public class DistributedFlowObjectiveStore
     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;
+            }
+        }
+    }
+
 }

@@ -182,6 +182,12 @@ public class Ofdpa2Pipeline extends AbstractHandlerBehaviour implements Pipeline
         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 @@ public class Ofdpa2Pipeline extends AbstractHandlerBehaviour implements Pipeline
      *            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 @@ public class Ofdpa2Pipeline extends AbstractHandlerBehaviour implements Pipeline
         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 @@ public class Ofdpa2Pipeline extends AbstractHandlerBehaviour implements Pipeline
                     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 @@ public class Ofdpa2Pipeline extends AbstractHandlerBehaviour implements Pipeline
             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 @@ public class Ofdpa2Pipeline extends AbstractHandlerBehaviour implements Pipeline
                 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());
                 }