Group event indicating failover of previously used live port

Change-Id: I32749b38d5e4fab93fa97bbf6587bd0dc91db88c

@@ -56,6 +56,13 @@ public class GroupEvent extends AbstractEvent<GroupEvent.Type, Group> {
          */
         GROUP_UPDATE_FAILED,
 
+        /**
+         * Signifies change in the first live bucket in failover group
+         * (i.e. change in which bucket is in use).
+         * Only to be used with failover Group.
+         */
+        GROUP_BUCKET_FAILOVER,
+
         // internal event between Manager <-> Store
 
         /*

@@ -41,7 +41,12 @@ public interface GroupProviderService extends ProviderService<GroupProvider> {
      * @param deviceId device identifier
      * @param groupEntries collection of group entries as seen in data plane
      */
-    void pushGroupMetrics(DeviceId deviceId,
-                       Collection<Group> groupEntries);
+    void pushGroupMetrics(DeviceId deviceId, Collection<Group> groupEntries);
 
+    /**
+     * Notifies store of group failovers.
+     *
+     * @param failoverGroups failover groups in which a failover has occurred
+     */
+    void notifyOfFailovers(Collection<Group> failoverGroups);
 }

@@ -179,4 +179,9 @@ public interface GroupStore extends Store<GroupEvent, GroupStoreDelegate> {
      * @param groupEntries the group entries as received from southbound
      */
     void pushGroupMetrics(DeviceId deviceId, Collection<Group> groupEntries);
+
+    /**
+     * Indicates failover within a failover group.
+     */
+    void notifyOfFailovers(Collection<Group> failoverGroups);
 }

@@ -646,6 +646,17 @@ public class SimpleGroupStore
         }
     }
 
+    @Override
+    public void notifyOfFailovers(Collection<Group> failoverGroups) {
+        List<GroupEvent> failoverEvents = new ArrayList<>();
+        failoverGroups.forEach(group -> {
+            if (group.type() == Group.Type.FAILOVER) {
+                failoverEvents.add(new GroupEvent(GroupEvent.Type.GROUP_BUCKET_FAILOVER, group));
+            }
+        });
+        notifyDelegate(failoverEvents);
+    }
+
     private void groupMissing(Group group) {
         switch (group.state()) {
             case PENDING_DELETE:

@@ -324,9 +324,9 @@ public class GroupManager
                 case GROUP_ADD_FAILED:
                 case GROUP_UPDATE_FAILED:
                 case GROUP_REMOVE_FAILED:
+                case GROUP_BUCKET_FAILOVER:
                     post(event);
                     break;
-
                 default:
                     break;
             }
@@ -353,6 +353,11 @@ public class GroupManager
             checkValidity();
             store.pushGroupMetrics(deviceId, groupEntries);
         }
+
+        @Override
+        public void notifyOfFailovers(Collection<Group> failoverGroups) {
+            store.notifyOfFailovers(failoverGroups);
+        }
     }
 
     private class InternalDeviceListener implements DeviceListener {
@@ -378,4 +383,5 @@ public class GroupManager
             }
         }
     }
+
 }

@@ -1312,6 +1312,17 @@ public class DistributedGroupStore
         }
     }
 
+    @Override
+    public void notifyOfFailovers(Collection<Group> failoverGroups) {
+        List<GroupEvent> failoverEvents = new ArrayList<>();
+        failoverGroups.forEach(group -> {
+            if (group.type() == Group.Type.FAILOVER) {
+                failoverEvents.add(new GroupEvent(GroupEvent.Type.GROUP_BUCKET_FAILOVER, group));
+            }
+        });
+        notifyDelegate(failoverEvents);
+    }
+
     private void garbageCollect(DeviceId deviceId,
                                 Set<Group> southboundGroupEntries,
                                 Set<StoredGroupEntry> storedGroupEntries) {

@@ -18,7 +18,10 @@ package org.onosproject.provider.of.group.impl;
 
 import static org.slf4j.LoggerFactory.getLogger;
 
+import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.atomic.AtomicLong;
@@ -31,9 +34,12 @@ import org.apache.felix.scr.annotations.ReferenceCardinality;
 import org.onosproject.core.DefaultGroupId;
 import org.onosproject.core.GroupId;
 import org.onosproject.net.DeviceId;
+import org.onosproject.net.PortNumber;
+import org.onosproject.net.device.DeviceService;
 import org.onosproject.net.driver.DriverService;
 import org.onosproject.net.group.DefaultGroup;
 import org.onosproject.net.group.Group;
+import org.onosproject.net.group.GroupBucket;
 import org.onosproject.net.group.GroupBuckets;
 import org.onosproject.net.group.GroupDescription;
 import org.onosproject.net.group.GroupOperation;
@@ -42,6 +48,7 @@ import org.onosproject.net.group.GroupOperations;
 import org.onosproject.net.group.GroupProvider;
 import org.onosproject.net.group.GroupProviderRegistry;
 import org.onosproject.net.group.GroupProviderService;
+import org.onosproject.net.group.GroupService;
 import org.onosproject.net.group.StoredGroupBucketEntry;
 import org.onosproject.net.provider.AbstractProvider;
 import org.onosproject.net.provider.ProviderId;
@@ -62,6 +69,7 @@ import org.projectfloodlight.openflow.protocol.OFGroupStatsEntry;
 import org.projectfloodlight.openflow.protocol.OFGroupStatsReply;
 import org.projectfloodlight.openflow.protocol.OFGroupType;
 import org.projectfloodlight.openflow.protocol.OFMessage;
+import org.projectfloodlight.openflow.protocol.OFPortDesc;
 import org.projectfloodlight.openflow.protocol.OFPortStatus;
 import org.projectfloodlight.openflow.protocol.OFStatsReply;
 import org.projectfloodlight.openflow.protocol.OFStatsType;
@@ -88,6 +96,12 @@ public class OpenFlowGroupProvider extends AbstractProvider implements GroupProv
     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
     protected DriverService driverService;
 
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected DeviceService deviceService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected GroupService groupService;
+
     private GroupProviderService providerService;
 
     static final int POLL_INTERVAL = 10;
@@ -389,6 +403,7 @@ public class OpenFlowGroupProvider extends AbstractProvider implements GroupProv
 
         @Override
         public void portChanged(Dpid dpid, OFPortStatus status) {
+            providerService.notifyOfFailovers(checkFailoverGroups(dpid, status));
         }
 
         @Override
@@ -396,4 +411,58 @@ public class OpenFlowGroupProvider extends AbstractProvider implements GroupProv
         }
     }
 
+    /**
+     * Builds a list of failover Groups whose primary live bucket failed over
+     * (i.e. bucket in use has changed).
+     *
+     * @param dpid    DPID of switch whose port's status changed
+     * @param status  new status of port
+     * @return        list of groups whose primary live bucket failed over
+     */
+    private List<Group> checkFailoverGroups(Dpid dpid, OFPortStatus status) {
+        List<Group> groupList = new ArrayList<>();
+        OFPortDesc desc = status.getDesc();
+        PortNumber portNumber = PortNumber.portNumber(desc.getPortNo().getPortNumber());
+        DeviceId id = DeviceId.deviceId(Dpid.uri(dpid));
+        if (desc.isEnabled()) {
+            return groupList;
+        }
+        Iterator<Group> iterator = groupService.getGroups(id).iterator();
+        while (iterator.hasNext()) {
+            Group group = iterator.next();
+            if (group.type() == GroupDescription.Type.FAILOVER &&
+                    checkFailoverGroup(group, id, portNumber)) {
+                groupList.add(group);
+            }
+        }
+    return groupList;
+    }
+
+    /**
+     * Checks whether the first live port in the failover group's bucket
+     * has failed over.
+     *
+     * @param group       failover group to be checked for failover
+     * @param id          device ID of switch whose port's status changed
+     * @param portNumber  port number of port that was disabled
+     * @return            whether the failover group experienced failover
+     */
+    private boolean checkFailoverGroup(Group group, DeviceId id,
+                                       PortNumber portNumber) {
+        boolean portReached = false;
+        boolean portEnabled = false;
+        Iterator<GroupBucket> bIterator = group.buckets().buckets().iterator();
+        GroupBucket bucket;
+        while (bIterator.hasNext() && !portReached) {
+            bucket = bIterator.next();
+            if (deviceService.getPort(id, bucket.watchPort()).isEnabled()) {
+                portEnabled = true;
+            }
+            if (bucket.watchPort().equals(portNumber)) {
+                portReached = true;
+            }
+        }
+        return portReached && !portEnabled;
+    }
+
 }

@@ -201,6 +201,10 @@ public class OpenFlowGroupProviderTest {
             this.groups = groupEntries;
         }
 
+        @Override
+        public void notifyOfFailovers(Collection<Group> groups) {
+        }
+
         public Collection<Group> getGroupEntries() {
             return groups;
         }