[CORD-2607] Mcast buckets correction

Change-Id: Ib47b2d8e40babdbb2ccdba61b48365a141752016

apps/segmentrouting/src/main/java/org/onosproject/segmentrouting/McastHandler.java

@@ -50,6 +50,7 @@ import org.onosproject.net.flowobjective.ForwardingObjective;
 import org.onosproject.net.flowobjective.NextObjective;
 import org.onosproject.net.flowobjective.ObjectiveContext;
 import org.onosproject.net.mcast.McastEvent;
+import org.onosproject.net.mcast.McastRoute;
 import org.onosproject.net.mcast.McastRouteInfo;
 import org.onosproject.net.topology.TopologyService;
 import org.onosproject.segmentrouting.config.SegmentRoutingAppConfig;
@@ -62,6 +63,7 @@ import org.onosproject.store.service.Versioned;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.time.Instant;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
@@ -69,9 +71,15 @@ import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.stream.Collectors;
 
 import static com.google.common.base.Preconditions.checkState;
+import static java.util.concurrent.Executors.newScheduledThreadPool;
+import static org.onlab.util.Tools.groupedThreads;
 import static org.onosproject.segmentrouting.SegmentRoutingManager.INTERNAL_VLAN;
 
 /**
@@ -87,6 +95,49 @@ public class McastHandler {
     private final KryoNamespace.Builder mcastKryo;
     private final ConsistentMap<McastStoreKey, McastRole> mcastRoleStore;
 
+    // Mcast lock to serialize local operations
+    private final Lock mcastLock = new ReentrantLock();
+
+    /**
+     * Acquires the lock used when making mcast changes.
+     */
+    private void mcastLock() {
+        mcastLock.lock();
+    }
+
+    /**
+     * Releases the lock used when making mcast changes.
+     */
+    private void mcastUnlock() {
+        mcastLock.unlock();
+    }
+
+    // Stability threshold for Mcast. Seconds
+    private static final long MCAST_STABLITY_THRESHOLD = 5;
+    // Last change done
+    private Instant lastMcastChange = Instant.now();
+
+    /**
+     * Determines if mcast in the network has been stable in the last
+     * MCAST_STABLITY_THRESHOLD seconds, by comparing the current time
+     * to the last mcast change timestamp.
+     *
+     * @return true if stable
+     */
+    private boolean isMcastStable() {
+        long last = (long) (lastMcastChange.toEpochMilli() / 1000.0);
+        long now = (long) (Instant.now().toEpochMilli() / 1000.0);
+        log.debug("Mcast stable since {}s", now - last);
+        return (now - last) > MCAST_STABLITY_THRESHOLD;
+    }
+
+    // Verify interval for Mcast
+    private static final long MCAST_VERIFY_INTERVAL = 30;
+
+    // Executor for mcast bucket corrector
+    private ScheduledExecutorService executorService
+            = newScheduledThreadPool(1, groupedThreads("mcastBktCorrector", "mcastbktC-%d", log));
+
     /**
      * Role in the multicast tree.
      */
@@ -129,6 +180,10 @@ public class McastHandler {
                 .withName("onos-mcast-role-store")
                 .withSerializer(Serializer.using(mcastKryo.build("McastHandler-Role")))
                 .build();
+        // Init the executor service and the buckets corrector
+        executorService.scheduleWithFixedDelay(new McastBucketCorrector(), 10,
+                                               MCAST_VERIFY_INTERVAL,
+                                               TimeUnit.SECONDS);
     }
 
     /**
@@ -144,6 +199,13 @@ public class McastHandler {
         });
     }
 
+    /**
+     * Clean up when deactivating the application.
+     */
+    protected void terminate() {
+        executorService.shutdown();
+    }
+
     /**
      * Processes the SOURCE_ADDED event.
      *
@@ -160,9 +222,7 @@ public class McastHandler {
         Set<ConnectPoint> sinks = mcastRouteInfo.sinks();
         IpAddress mcastIp = mcastRouteInfo.route().group();
 
-        sinks.forEach(sink -> {
+        sinks.forEach(sink -> processSinkAddedInternal(source, sink, mcastIp));
-            processSinkAddedInternal(source, sink, mcastIp);
-        });
     }
 
     /**
@@ -200,9 +260,24 @@ public class McastHandler {
         ConnectPoint sink = mcastRouteInfo.sink().orElse(null);
         IpAddress mcastIp = mcastRouteInfo.route().group();
 
+        processSinkRemovedInternal(source, sink, mcastIp);
+    }
+
+    /**
+     * Removes a path from source to sink for given multicast group.
+     *
+     * @param source connect point of the multicast source
+     * @param sink connection point of the multicast sink
+     * @param mcastIp multicast group IP address
+     */
+    private void processSinkRemovedInternal(ConnectPoint source, ConnectPoint sink,
+                                          IpAddress mcastIp) {
+        lastMcastChange = Instant.now();
+        mcastLock();
+        try {
             // Continue only when this instance is the master of source device
             if (!srManager.mastershipService.isLocalMaster(source.deviceId())) {
-            log.info("Skip {} due to lack of mastership of the source device {}",
+                log.debug("Skip {} due to lack of mastership of the source device {}",
                          mcastIp, source.deviceId());
                 return;
             }
@@ -211,7 +286,8 @@ public class McastHandler {
             if (source.deviceId().equals(sink.deviceId())) {
                 // Source and sink are on even the same port. There must be something wrong.
                 if (source.port().equals(sink.port())) {
-                log.warn("Sink is on the same port of source. Abort");
+                    log.warn("Skip {} since sink {} is on the same port of source {}. Abort",
+                             mcastIp, sink, source);
                     return;
                 }
                 removePortFromDevice(sink.deviceId(), sink.port(), mcastIp, assignedVlan(source));
@@ -231,13 +307,19 @@ public class McastHandler {
                 Collections.reverse(links);
                 for (Link link : links) {
                     if (isLast) {
-                    isLast = removePortFromDevice(link.src().deviceId(), link.src().port(),
+                        isLast = removePortFromDevice(
+                                link.src().deviceId(),
+                                link.src().port(),
                                 mcastIp,
-                            assignedVlan(link.src().deviceId().equals(source.deviceId()) ? source : null));
+                                assignedVlan(link.src().deviceId().equals(source.deviceId()) ? source : null)
+                        );
                         mcastRoleStore.remove(new McastStoreKey(mcastIp, link.src().deviceId()));
                     }
                 }
             }
+        } finally {
+            mcastUnlock();
+        }
     }
 
     /**
@@ -249,10 +331,13 @@ public class McastHandler {
      */
     private void processSinkAddedInternal(ConnectPoint source, ConnectPoint sink,
             IpAddress mcastIp) {
+        lastMcastChange = Instant.now();
+        mcastLock();
+        try {
             // Continue only when this instance is the master of source device
             if (!srManager.mastershipService.isLocalMaster(source.deviceId())) {
-            log.info("Skip {} due to lack of mastership of the source device {}",
+                log.debug("Skip {} due to lack of mastership of the source device {}",
-                    source.deviceId());
+                          mcastIp, source.deviceId());
                 return;
             }
 
@@ -263,7 +348,8 @@ public class McastHandler {
             if (source.deviceId().equals(sink.deviceId())) {
                 // Source and sink are on even the same port. There must be something wrong.
                 if (source.port().equals(sink.port())) {
-                log.warn("Sink is on the same port of source. Abort");
+                    log.warn("Skip {} since sink {} is on the same port of source {}. Abort",
+                             mcastIp, sink, source);
                     return;
                 }
                 addPortToDevice(sink.deviceId(), sink.port(), mcastIp, assignedVlan(source));
@@ -298,6 +384,9 @@ public class McastHandler {
                 log.warn("Unable to find a path from {} to {}. Abort sinkAdded",
                          source.deviceId(), sink.deviceId());
             }
+        } finally {
+            mcastUnlock();
+        }
     }
 
     /**
@@ -306,6 +395,10 @@ public class McastHandler {
      * @param affectedLink Link that is going down
      */
     protected void processLinkDown(Link affectedLink) {
+        lastMcastChange = Instant.now();
+        mcastLock();
+        try {
+            // Get groups affected by the link down event
             getAffectedGroups(affectedLink).forEach(mcastIp -> {
                 // TODO Optimize when the group editing is in place
                 log.debug("Processing link down {} for group {}",
@@ -329,7 +422,7 @@ public class McastHandler {
 
                 // Continue only when this instance is the master of source device
                 if (!srManager.mastershipService.isLocalMaster(source.deviceId())) {
-                log.info("Skip {} due to lack of mastership of the source device {}",
+                    log.debug("Skip {} due to lack of mastership of the source device {}",
                              source.deviceId());
                     return;
                 }
@@ -356,6 +449,9 @@ public class McastHandler {
                     }
                 });
             });
+        } finally {
+            mcastUnlock();
+        }
     }
 
     /**
@@ -364,6 +460,9 @@ public class McastHandler {
      * @param deviceDown device going down
      */
     protected void processDeviceDown(DeviceId deviceDown) {
+        lastMcastChange = Instant.now();
+        mcastLock();
+        try {
             // Get the mcast groups affected by the device going down
             getAffectedGroups(deviceDown).forEach(mcastIp -> {
                 // TODO Optimize when the group editing is in place
@@ -391,7 +490,7 @@ public class McastHandler {
                 if (!srManager.mastershipService.isLocalMaster(source.deviceId())) {
                     // When the source is available we just check the mastership
                     if (srManager.deviceService.isAvailable(source.deviceId())) {
-                    log.info("Skip {} due to lack of mastership of the source device {}",
+                        log.debug("Skip {} due to lack of mastership of the source device {}",
                                  mcastIp, source.deviceId());
                         return;
                     }
@@ -401,7 +500,7 @@ public class McastHandler {
                             source.deviceId().toString()).leaderNodeId();
                     // Verify if this node is the leader
                     if (!srManager.clusterService.getLocalNode().id().equals(leader)) {
-                    log.info("Skip {} due to lack of leadership on the topic {}",
+                        log.debug("Skip {} due to lack of leadership on the topic {}",
                                  mcastIp, source.deviceId());
                         return;
                     }
@@ -461,6 +560,9 @@ public class McastHandler {
                     });
                 }
             });
+        } finally {
+            mcastUnlock();
+        }
     }
 
     /**
@@ -1056,7 +1158,9 @@ public class McastHandler {
     }
 
     /**
-     * Adds or removes filtering objective for given device and port.
+     * Updates filtering objective for given device and port.
+     * It is called in general when the mcast config has been
+     * changed.
      *
      * @param deviceId device ID
      * @param portNum ingress port number
@@ -1065,6 +1169,11 @@ public class McastHandler {
      */
     protected void updateFilterToDevice(DeviceId deviceId, PortNumber portNum,
                                         VlanId vlanId, boolean install) {
+        lastMcastChange = Instant.now();
+        mcastLock();
+        try {
+            // Iterates over the route and updates properly the filtering objective
+            // on the source device.
             srManager.multicastRouteService.getRoutes().forEach(mcastRoute -> {
                 ConnectPoint source = srManager.multicastRouteService.fetchSource(mcastRoute);
                 if (source.deviceId().equals(deviceId) && source.port().equals(portNum)) {
@@ -1075,5 +1184,103 @@ public class McastHandler {
                     }
                 }
             });
+        } finally {
+            mcastUnlock();
+        }
+    }
+
+    /**
+     * Performs bucket verification operation for all mcast groups in the devices.
+     * Firstly, it verifies that mcast is stable before trying verification operation.
+     * Verification consists in creating new nexts with VERIFY operation. Actually,
+     * the operation is totally delegated to the driver.
+     */
+     private final class McastBucketCorrector implements Runnable {
+
+        @Override
+        public void run() {
+            // Verify if the Mcast has been stable for MCAST_STABLITY_THRESHOLD
+            if (!isMcastStable()) {
+                return;
+            }
+            // Acquires lock
+            mcastLock();
+            try {
+                // Iterates over the routes and verify the related next objectives
+                srManager.multicastRouteService.getRoutes()
+                    .stream()
+                    .map(McastRoute::group)
+                    .forEach(mcastIp -> {
+                        log.trace("Running mcast buckets corrector for mcast group: {}",
+                                  mcastIp);
+
+                        // For each group we get current information in the store
+                        // and issue a check of the next objectives in place
+                        DeviceId ingressDevice = getDevice(mcastIp, McastRole.INGRESS)
+                                .stream().findAny().orElse(null);
+                        DeviceId transitDevice = getDevice(mcastIp, McastRole.TRANSIT)
+                                .stream().findAny().orElse(null);
+                        Set<DeviceId> egressDevices = getDevice(mcastIp, McastRole.EGRESS);
+                        ConnectPoint source = getSource(mcastIp);
+
+                        // Do not proceed if ingress device or source of this group are missing
+                        if (ingressDevice == null || source == null) {
+                            log.warn("Unable to run buckets corrector. " +
+                                             "Missing ingress {} or source {} for group {}",
+                                     ingressDevice, source, mcastIp);
+                            return;
+                        }
+
+                        // Continue only when this instance is the master of source device
+                        if (!srManager.mastershipService.isLocalMaster(source.deviceId())) {
+                            log.trace("Unable to run buckets corrector. " +
+                                             "Skip {} due to lack of mastership " +
+                                             "of the source device {}",
+                                     mcastIp, source.deviceId());
+                            return;
+                        }
+
+                        // Create the set of the devices to be processed
+                        ImmutableSet.Builder<DeviceId> devicesBuilder = ImmutableSet.builder();
+                        devicesBuilder.add(ingressDevice);
+                        if (transitDevice != null) {
+                            devicesBuilder.add(transitDevice);
+                        }
+                        if (!egressDevices.isEmpty()) {
+                            devicesBuilder.addAll(egressDevices);
+                        }
+                        Set<DeviceId> devicesToProcess = devicesBuilder.build();
+
+                        // Iterate over the devices
+                        devicesToProcess.forEach(deviceId -> {
+                            McastStoreKey currentKey = new McastStoreKey(mcastIp, deviceId);
+                            // If next exists in our store verify related next objective
+                            if (mcastNextObjStore.containsKey(currentKey)) {
+                                NextObjective currentNext = mcastNextObjStore.get(currentKey).value();
+                                // Get current ports
+                                Set<PortNumber> currentPorts = getPorts(currentNext.next());
+                                // Rebuild the next objective
+                                currentNext = nextObjBuilder(
+                                        mcastIp,
+                                        assignedVlan(deviceId.equals(source.deviceId()) ? source : null),
+                                        currentPorts,
+                                        currentNext.id()
+                                ).verify();
+                                // Send to the flowobjective service
+                                srManager.flowObjectiveService.next(deviceId, currentNext);
+                            } else {
+                                log.warn("Unable to run buckets corrector." +
+                                                 "Missing next for {} and group {}",
+                                         deviceId, mcastIp);
+                            }
+                        });
+
+                    });
+            } finally {
+                // Finally, it releases the lock
+                mcastUnlock();
+            }
+
+        }
     }
 }

apps/segmentrouting/src/main/java/org/onosproject/segmentrouting/SegmentRoutingManager.java

@@ -490,6 +490,8 @@ public class SegmentRoutingManager implements SegmentRoutingService {
         portNextObjStore.destroy();
         tunnelStore.destroy();
         policyStore.destroy();
+
+        mcastHandler.terminate();
         log.info("Stopped");
     }
 

drivers/default/src/main/java/org/onosproject/driver/pipeline/ofdpa/Ofdpa2GroupHandler.java

@@ -1588,8 +1588,8 @@ public class Ofdpa2GroupHandler {
      *             modified to match the given next objective
      */
     protected void verifyGroup(NextObjective nextObjective, NextGroup next) {
-        if (nextObjective.type() != NextObjective.Type.HASHED) {
+        if (nextObjective.type() == NextObjective.Type.SIMPLE) {
-            log.warn("verification not supported for {} group", nextObjective.type());
+            log.warn("verification not supported for indirect group");
             fail(nextObjective, ObjectiveError.UNSUPPORTED);
             return;
         }
@@ -1640,17 +1640,25 @@ public class Ofdpa2GroupHandler {
             indicesToRemove.addAll(otherIndices);
         }
 
+        log.debug("Buckets to create {}", bucketsToCreate);
+        log.debug("Indices to remove {}", indicesToRemove);
+
         if (!bucketsToCreate.isEmpty()) {
             log.info("creating {} buckets as part of nextId: {} verification",
                      bucketsToCreate.size(), nextObjective.id());
             //create a nextObjective only with these buckets
             NextObjective.Builder nextObjBuilder = DefaultNextObjective.builder()
                     .withId(nextObjective.id())
-                    .withType(NextObjective.Type.HASHED)
+                    .withType(nextObjective.type())
                     .withMeta(nextObjective.meta())
                     .fromApp(nextObjective.appId());
-            bucketsToCreate.forEach(bucket -> nextObjBuilder.addTreatment(bucket));
+            bucketsToCreate.forEach(nextObjBuilder::addTreatment);
+            // According to the next type we call the proper add function
+            if (nextObjective.type() == NextObjective.Type.HASHED) {
                 addBucketToHashGroup(nextObjBuilder.addToExisting(), allActiveKeys);
+            } else {
+                addBucketToBroadcastGroup(nextObjBuilder.addToExisting(), allActiveKeys);
+            }
         }
 
         if (!indicesToRemove.isEmpty()) {
@@ -1667,9 +1675,9 @@ public class Ofdpa2GroupHandler {
             // Nevertheless groupStore may not be in sync due to bug in the store
             // - see CORD-1844. XXX When this bug is fixed, the rest of this verify
             // method will not be required.
-            GroupKey hashGroupKey = allActiveKeys.get(0).peekFirst();
+            GroupKey topGroupKey = allActiveKeys.get(0).peekFirst();
-            Group hashGroup = groupService.getGroup(deviceId, hashGroupKey);
+            Group topGroup = groupService.getGroup(deviceId, topGroupKey);
-            int actualGroupSize = hashGroup.buckets().buckets().size();
+            int actualGroupSize = topGroup.buckets().buckets().size();
             int objGroupSize = nextObjective.next().size();
             if (actualGroupSize != objGroupSize) {
                 log.warn("Mismatch detected in device:{}, nextId:{}, nextObjective-size"
@@ -1677,9 +1685,10 @@ public class Ofdpa2GroupHandler {
                         objGroupSize, actualGroupSize);
             }
             if (actualGroupSize > objGroupSize) {
+                // Group in the device has more chains
                 List<GroupBucket> bucketsToRemove = Lists.newArrayList();
                 //check every bucket in the actual group
-                for (GroupBucket bucket : hashGroup.buckets().buckets()) {
+                for (GroupBucket bucket : topGroup.buckets().buckets()) {
                     GroupInstruction g = (GroupInstruction) bucket.treatment()
                                             .allInstructions().iterator().next();
                     GroupId gidToCheck = g.groupId(); // the group pointed to
@@ -1707,11 +1716,12 @@ public class Ofdpa2GroupHandler {
                             + "buckets to remove");
                 } else {
                     GroupBuckets removeBuckets = new GroupBuckets(bucketsToRemove);
-                    groupService.removeBucketsFromGroup(deviceId, hashGroupKey,
+                    groupService.removeBucketsFromGroup(deviceId, topGroupKey,
-                                                        removeBuckets, hashGroupKey,
+                                                        removeBuckets, topGroupKey,
                                                         nextObjective.appId());
                 }
             } else if (actualGroupSize < objGroupSize) {
+                // Group in the device has less chains
                 // should also add buckets not in group-store but in obj-store
                 List<GroupBucket> bucketsToAdd = Lists.newArrayList();
                 //check every bucket in the obj
@@ -1727,7 +1737,7 @@ public class Ofdpa2GroupHandler {
                         continue;
                     }
                     boolean matches = false;
-                    for (GroupBucket bucket : hashGroup.buckets().buckets()) {
+                    for (GroupBucket bucket : topGroup.buckets().buckets()) {
                         GroupInstruction g = (GroupInstruction) bucket.treatment()
                                                 .allInstructions().iterator().next();
                         GroupId gidToCheck = g.groupId(); // the group pointed to
@@ -1741,7 +1751,12 @@ public class Ofdpa2GroupHandler {
                         TrafficTreatment t = DefaultTrafficTreatment.builder()
                                                 .group(pointedGroup.id())
                                                 .build();
+                        // Create the proper bucket according to the next type
+                        if (nextObjective.type() == NextObjective.Type.HASHED) {
                             bucketsToAdd.add(DefaultGroupBucket.createSelectGroupBucket(t));
+                        } else {
+                            bucketsToAdd.add(DefaultGroupBucket.createAllGroupBucket(t));
+                        }
                     }
                 }
                 if (bucketsToAdd.isEmpty()) {
@@ -1749,8 +1764,8 @@ public class Ofdpa2GroupHandler {
                             + "buckets to add");
                 } else {
                     GroupBuckets addBuckets = new GroupBuckets(bucketsToAdd);
-                    groupService.addBucketsToGroup(deviceId, hashGroupKey,
+                    groupService.addBucketsToGroup(deviceId, topGroupKey,
-                                                   addBuckets, hashGroupKey,
+                                                   addBuckets, topGroupKey,
                                                    nextObjective.appId());
                 }
             }