[Goldeneye] ONOS-4017: Mastership service considers Region information when determining mastership.

Change-Id: I6c79239f2e071d865bf04e4d9d790ca9b2d04694

@@ -67,6 +67,14 @@
 
         <dependency>
             <groupId>org.onosproject</groupId>
+            <artifactId>onos-core-dist</artifactId>
+            <scope>test</scope>
+            <classifier>tests</classifier>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.onosproject</groupId>
             <artifactId>onos-incubator-api</artifactId>
             <scope>test</scope>
             <classifier>tests</classifier>

@@ -18,6 +18,7 @@ package org.onosproject.cluster.impl;
 import com.codahale.metrics.Timer;
 import com.codahale.metrics.Timer.Context;
 import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
 import com.google.common.util.concurrent.Futures;
 import org.apache.felix.scr.annotations.Activate;
 import org.apache.felix.scr.annotations.Component;
@@ -42,9 +43,13 @@ import org.onosproject.mastership.MastershipTerm;
 import org.onosproject.mastership.MastershipTermService;
 import org.onosproject.net.DeviceId;
 import org.onosproject.net.MastershipRole;
+import org.onosproject.net.region.Region;
+import org.onosproject.net.region.RegionService;
 import org.slf4j.Logger;
 
+import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
@@ -89,8 +94,12 @@ public class MastershipManager
     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
     protected MetricsService metricsService;
 
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected RegionService regionService;
+
     private NodeId localNodeId;
     private Timer requestRoleTimer;
+    public boolean useRegionForBalanceRoles;
 
     @Activate
     public void activate() {
@@ -212,6 +221,28 @@ public class MastershipManager
             }
         }
 
+        if (useRegionForBalanceRoles && balanceRolesUsingRegions(controllerDevices)) {
+            return;
+        }
+
+        // Now re-balance the buckets until they are roughly even.
+        List<CompletableFuture<Void>> balanceBucketsFutures = balanceControllerNodes(controllerDevices, deviceCount);
+
+        CompletableFuture<Void> balanceRolesFuture = CompletableFuture.allOf(
+                balanceBucketsFutures.toArray(new CompletableFuture[balanceBucketsFutures.size()]));
+
+        Futures.getUnchecked(balanceRolesFuture);
+    }
+
+    /**
+     * Balances the nodes specified in controllerDevices.
+     *
+     * @param controllerDevices controller nodes to devices map
+     * @param deviceCount number of devices mastered by controller nodes
+     * @return list of setRole futures for "moved" devices
+     */
+    private List<CompletableFuture<Void>> balanceControllerNodes(
+            Map<ControllerNode, Set<DeviceId>> controllerDevices, int deviceCount) {
         // Now re-balance the buckets until they are roughly even.
         List<CompletableFuture<Void>> balanceBucketsFutures = Lists.newLinkedList();
         int rounds = controllerDevices.keySet().size();
@@ -221,12 +252,16 @@ public class MastershipManager
             ControllerNode largest = findBucket(false, controllerDevices);
             balanceBucketsFutures.add(balanceBuckets(smallest, largest, controllerDevices, deviceCount));
         }
-        CompletableFuture<Void> balanceRolesFuture = CompletableFuture.allOf(
-                balanceBucketsFutures.toArray(new CompletableFuture[balanceBucketsFutures.size()]));
-
-        Futures.getUnchecked(balanceRolesFuture);
+        return balanceBucketsFutures;
     }
 
+    /**
+     * Finds node with the minimum/maximum devices from a list of nodes.
+     *
+     * @param min true: minimum, false: maximum
+     * @param controllerDevices controller nodes to devices map
+     * @return controller node with minimum/maximum devices
+     */
     private ControllerNode findBucket(boolean min,
                                       Map<ControllerNode, Set<DeviceId>>  controllerDevices) {
         int xSize = min ? Integer.MAX_VALUE : -1;
@@ -241,6 +276,15 @@ public class MastershipManager
         return xNode;
     }
 
+    /**
+     * Balance the node buckets by moving devices from largest to smallest node.
+     *
+     * @param smallest node that is master of the smallest number of devices
+     * @param largest node that is master of the largest number of devices
+     * @param controllerDevices controller nodes to devices map
+     * @param deviceCount number of devices mastered by controller nodes
+     * @return list of setRole futures for "moved" devices
+     */
     private CompletableFuture<Void> balanceBuckets(ControllerNode smallest, ControllerNode largest,
                                 Map<ControllerNode, Set<DeviceId>>  controllerDevices,
                                 int deviceCount) {
@@ -272,6 +316,149 @@ public class MastershipManager
         return CompletableFuture.allOf(setRoleFutures.toArray(new CompletableFuture[setRoleFutures.size()]));
     }
 
+    /**
+     * Balances the nodes considering Region information.
+     *
+     * @param allControllerDevices controller nodes to devices map
+     * @return true: nodes balanced; false: nodes not balanced
+     */
+    private boolean balanceRolesUsingRegions(Map<ControllerNode, Set<DeviceId>> allControllerDevices) {
+        Set<Region> regions = regionService.getRegions();
+        if (regions.isEmpty()) {
+            return false; // no balancing was done using regions.
+        }
+
+        // handle nodes belonging to regions
+        Set<ControllerNode> nodesInRegions = Sets.newHashSet();
+        for (Region region : regions) {
+            Map<ControllerNode, Set<DeviceId>> activeRegionControllers =
+                    balanceRolesInRegion(region, allControllerDevices);
+            nodesInRegions.addAll(activeRegionControllers.keySet());
+        }
+
+        // handle nodes not belonging to any region
+        Set<ControllerNode> nodesNotInRegions = Sets.difference(allControllerDevices.keySet(), nodesInRegions);
+        if (!nodesNotInRegions.isEmpty()) {
+            int deviceCount = 0;
+            Map<ControllerNode, Set<DeviceId>> controllerDevicesNotInRegions = new HashMap<>();
+            for (ControllerNode controllerNode: nodesNotInRegions) {
+                controllerDevicesNotInRegions.put(controllerNode, allControllerDevices.get(controllerNode));
+                deviceCount += allControllerDevices.get(controllerNode).size();
+            }
+            // Now re-balance the buckets until they are roughly even.
+            List<CompletableFuture<Void>> balanceBucketsFutures =
+                    balanceControllerNodes(controllerDevicesNotInRegions, deviceCount);
+
+            CompletableFuture<Void> balanceRolesFuture = CompletableFuture.allOf(
+                    balanceBucketsFutures.toArray(new CompletableFuture[balanceBucketsFutures.size()]));
+
+            Futures.getUnchecked(balanceRolesFuture);
+        }
+        return true; // balancing was done using regions.
+    }
+
+    /**
+     * Balances the nodes in specified region.
+     *
+     * @param region region in which nodes are to be balanced
+     * @param allControllerDevices controller nodes to devices map
+     * @return controller nodes that were balanced
+     */
+    private Map<ControllerNode, Set<DeviceId>> balanceRolesInRegion(Region region,
+         Map<ControllerNode, Set<DeviceId>> allControllerDevices) {
+
+        // retrieve all devices associated with specified region
+        Set<DeviceId> devicesInRegion = regionService.getRegionDevices(region.id());
+        log.info("Region {} has {} devices.", region.id(), devicesInRegion.size());
+        if (devicesInRegion.isEmpty()) {
+            return new HashMap<>(); // no devices in this region, so nothing to balance.
+        }
+
+        List<Set<NodeId>> mastersList = region.masters();
+        log.info("Region {} has {} sets of masters.", region.id(), mastersList.size());
+        if (mastersList.isEmpty()) {
+            // TODO handle devices that belong to a region, which has no masters defined
+            return new HashMap<>(); // for now just leave devices alone
+        }
+
+        // get the region's preferred set of masters
+        Set<DeviceId> devicesInMasters = Sets.newHashSet();
+        Map<ControllerNode, Set<DeviceId>> regionalControllerDevices =
+                getRegionsPreferredMasters(region, devicesInMasters, allControllerDevices);
+
+        // Now re-balance the buckets until they are roughly even.
+        List<CompletableFuture<Void>> balanceBucketsFutures =
+                balanceControllerNodes(regionalControllerDevices, devicesInMasters.size());
+
+        // handle devices that are not currently mastered by the master node set
+        Set<DeviceId> devicesNotMasteredWithControllers = Sets.difference(devicesInRegion, devicesInMasters);
+        if (!devicesNotMasteredWithControllers.isEmpty()) {
+            // active controllers in master node set are already balanced, just
+            // assign device mastership in sequence
+            List<ControllerNode> sorted = new ArrayList<>(regionalControllerDevices.keySet());
+            Collections.sort(sorted, (o1, o2) ->
+                    ((Integer) (regionalControllerDevices.get(o1)).size())
+                            .compareTo((Integer) (regionalControllerDevices.get(o2)).size()));
+            int deviceIndex = 0;
+            for (DeviceId deviceId : devicesNotMasteredWithControllers) {
+                ControllerNode cnode = sorted.get(deviceIndex % sorted.size());
+                balanceBucketsFutures.add(setRole(cnode.id(), deviceId, MASTER));
+                regionalControllerDevices.get(cnode).add(deviceId);
+                deviceIndex++;
+            }
+        }
+
+        CompletableFuture<Void> balanceRolesFuture = CompletableFuture.allOf(
+                balanceBucketsFutures.toArray(new CompletableFuture[balanceBucketsFutures.size()]));
+
+        Futures.getUnchecked(balanceRolesFuture);
+
+        // update the map before returning
+        regionalControllerDevices.forEach((controllerNode, deviceIds) -> {
+            regionalControllerDevices.put(controllerNode, new HashSet<>(getDevicesOf(controllerNode.id())));
+        });
+
+        return regionalControllerDevices;
+    }
+
+    /**
+     * Get region's preferred set of master nodes - the first master node set that has at
+     * least one active node.
+     *
+     * @param region region for which preferred set of master nodes is requested
+     * @param devicesInMasters device set to track devices in preferred set of master nodes
+     * @param allControllerDevices controller nodes to devices map
+     * @return region's preferred master nodes (and devices that use them as masters)
+     */
+    private Map<ControllerNode, Set<DeviceId>> getRegionsPreferredMasters(Region region,
+            Set<DeviceId> devicesInMasters,
+            Map<ControllerNode, Set<DeviceId>> allControllerDevices) {
+        Map<ControllerNode, Set<DeviceId>> regionalControllerDevices = new HashMap<>();
+        int listIndex = 0;
+        for (Set<NodeId> masterSet: region.masters()) {
+            log.info("Region {} masters set {} has {} nodes.",
+                     region.id(), listIndex, masterSet.size());
+            if (masterSet.isEmpty()) { // nothing on this level
+                listIndex++;
+                continue;
+            }
+            // Create buckets reflecting current ownership.
+            for (NodeId nodeId : masterSet) {
+                if (clusterService.getState(nodeId) == ACTIVE) {
+                    ControllerNode controllerNode = clusterService.getNode(nodeId);
+                    Set<DeviceId> devicesOf = new HashSet<>(allControllerDevices.get(controllerNode));
+                    regionalControllerDevices.put(controllerNode, devicesOf);
+                    devicesInMasters.addAll(devicesOf);
+                    log.info("Active Node {} has {} devices.", nodeId, devicesOf.size());
+                }
+            }
+            if (!regionalControllerDevices.isEmpty()) {
+                break; // now have a set of >0 active controllers
+            }
+            listIndex++; // keep on looking
+        }
+        return regionalControllerDevices;
+    }
 
     public class InternalDelegate implements MastershipStoreDelegate {
         @Override

@@ -15,14 +15,18 @@
  */
 package org.onosproject.cluster.impl;
 
+import java.util.List;
 import java.util.Set;
+import java.util.function.Consumer;
 
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
+import org.onlab.junit.TestUtils;
 import org.onlab.packet.IpAddress;
 import org.onosproject.cluster.ClusterService;
-import org.onosproject.cluster.ClusterServiceAdapter;
 import org.onosproject.cluster.ControllerNode;
 import org.onosproject.cluster.DefaultControllerNode;
 import org.onosproject.cluster.NodeId;
@@ -31,17 +35,24 @@ import org.onosproject.mastership.MastershipService;
 import org.onosproject.mastership.MastershipStore;
 import org.onosproject.mastership.MastershipTermService;
 import org.onosproject.net.DeviceId;
+import org.onosproject.net.region.Region;
+import org.onosproject.net.region.RegionId;
+import org.onosproject.net.region.RegionStore;
+import org.onosproject.net.region.impl.RegionManager;
+import org.onosproject.store.cluster.StaticClusterService;
+import org.onosproject.store.region.impl.DistributedRegionStore;
+import org.onosproject.store.service.TestStorageService;
 import org.onosproject.store.trivial.SimpleMastershipStore;
 
 import com.google.common.collect.Sets;
 import com.google.common.util.concurrent.Futures;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
+import static org.junit.Assert.*;
 import static org.onosproject.net.MastershipRole.MASTER;
 import static org.onosproject.net.MastershipRole.NONE;
 import static org.onosproject.net.MastershipRole.STANDBY;
 import static org.onosproject.net.NetTestTools.injectEventDispatcher;
+import static org.onosproject.net.region.Region.Type.METRO;
 
 /**
  * Test codifying the mastership service contracts.
@@ -54,16 +65,47 @@ public class MastershipManagerTest {
     private static final DeviceId DEV_MASTER = DeviceId.deviceId("of:1");
     private static final DeviceId DEV_OTHER = DeviceId.deviceId("of:2");
 
+    private static final RegionId RID1 = RegionId.regionId("r1");
+    private static final RegionId RID2 = RegionId.regionId("r2");
+    private static final DeviceId DID1 = DeviceId.deviceId("foo:d1");
+    private static final DeviceId DID2 = DeviceId.deviceId("foo:d2");
+    private static final DeviceId DID3 = DeviceId.deviceId("foo:d3");
+    private static final NodeId NID1 = NodeId.nodeId("n1");
+    private static final NodeId NID2 = NodeId.nodeId("n2");
+    private static final NodeId NID3 = NodeId.nodeId("n3");
+    private static final NodeId NID4 = NodeId.nodeId("n4");
+    private static final ControllerNode CNODE1 =
+            new DefaultControllerNode(NID1, IpAddress.valueOf("127.0.1.1"));
+    private static final ControllerNode CNODE2 =
+            new DefaultControllerNode(NID2, IpAddress.valueOf("127.0.1.2"));
+    private static final ControllerNode CNODE3 =
+            new DefaultControllerNode(NID3, IpAddress.valueOf("127.0.1.3"));
+    private static final ControllerNode CNODE4 =
+            new DefaultControllerNode(NID4, IpAddress.valueOf("127.0.1.4"));
+
+
     private MastershipManager mgr;
     protected MastershipService service;
+    private TestRegionManager regionManager;
+    private RegionStore regionStore;
+    private TestClusterService testClusterService;
 
     @Before
-    public void setUp() {
+    public void setUp() throws Exception {
         mgr = new MastershipManager();
         service = mgr;
         injectEventDispatcher(mgr, new TestEventDispatcher());
-        mgr.clusterService = new TestClusterService();
+        testClusterService = new TestClusterService();
+        mgr.clusterService = testClusterService;
         mgr.store = new TestSimpleMastershipStore(mgr.clusterService);
+        regionStore = new DistributedRegionStore();
+        TestUtils.setField(regionStore, "storageService", new TestStorageService());
+        TestUtils.callMethod(regionStore, "activate",
+                             new Class<?>[] {});
+        regionManager = new TestRegionManager();
+        TestUtils.setField(regionManager, "store", regionStore);
+        regionManager.activate();
+        mgr.regionService = regionManager;
         mgr.activate();
     }
 
@@ -72,6 +114,8 @@ public class MastershipManagerTest {
         mgr.deactivate();
         mgr.clusterService = null;
         injectEventDispatcher(mgr, null);
+        regionManager.deactivate();
+        mgr.regionService = null;
         mgr.store = null;
     }
 
@@ -154,7 +198,139 @@ public class MastershipManagerTest {
         assertEquals("inconsistent terms: ", 3, ts.getMastershipTerm(DEV_MASTER).termNumber());
     }
 
-    private final class TestClusterService extends ClusterServiceAdapter {
+    @Test
+    public void balanceWithRegion1() {
+        //set up region - 2 sets of masters with 1 node in each
+        Set<NodeId> masterSet1 = ImmutableSet.of(NID1);
+        Set<NodeId> masterSet2 = ImmutableSet.of(NID2);
+        List<Set<NodeId>> masters = ImmutableList.of(masterSet1, masterSet2);
+        Region r = regionManager.createRegion(RID1, "R1", METRO, masters);
+        regionManager.addDevices(RID1, ImmutableSet.of(DID1, DID2));
+        Set<DeviceId> deviceIds = regionManager.getRegionDevices(RID1);
+        assertEquals("incorrect device count", 2, deviceIds.size());
+
+        testClusterService.put(CNODE1, ControllerNode.State.ACTIVE);
+        testClusterService.put(CNODE2, ControllerNode.State.ACTIVE);
+
+        //set master to non region nodes
+        mgr.setRole(NID_LOCAL, DID1, MASTER);
+        mgr.setRole(NID_LOCAL, DID2, MASTER);
+        assertEquals("wrong local role:", MASTER, mgr.getLocalRole(DID1));
+        assertEquals("wrong local role:", MASTER, mgr.getLocalRole(DID2));
+        assertEquals("wrong master:", NID_LOCAL, mgr.getMasterFor(DID1));
+        assertEquals("wrong master:", NID_LOCAL, mgr.getMasterFor(DID2));
+
+        //do region balancing
+        mgr.useRegionForBalanceRoles = true;
+        mgr.balanceRoles();
+        assertEquals("wrong master:", NID1, mgr.getMasterFor(DID1));
+        assertEquals("wrong master:", NID1, mgr.getMasterFor(DID2));
+
+        // make N1 inactive
+        testClusterService.put(CNODE1, ControllerNode.State.INACTIVE);
+        mgr.balanceRoles();
+        assertEquals("wrong master:", NID2, mgr.getMasterFor(DID1));
+        assertEquals("wrong master:", NID2, mgr.getMasterFor(DID2));
+
+    }
+
+    @Test
+    public void balanceWithRegion2() {
+        //set up region - 2 sets of masters with (3 nodes, 1 node)
+        Set<NodeId> masterSet1 = ImmutableSet.of(NID1, NID3, NID4);
+        Set<NodeId> masterSet2 = ImmutableSet.of(NID2);
+        List<Set<NodeId>> masters = ImmutableList.of(masterSet1, masterSet2);
+        Region r = regionManager.createRegion(RID1, "R1", METRO, masters);
+        Set<DeviceId> deviceIdsOrig = ImmutableSet.of(DID1, DID2, DID3, DEV_OTHER);
+        regionManager.addDevices(RID1, deviceIdsOrig);
+        Set<DeviceId> deviceIds = regionManager.getRegionDevices(RID1);
+        assertEquals("incorrect device count", deviceIdsOrig.size(), deviceIds.size());
+        assertEquals("incorrect devices in region", deviceIdsOrig, deviceIds);
+
+        testClusterService.put(CNODE1, ControllerNode.State.ACTIVE);
+        testClusterService.put(CNODE2, ControllerNode.State.ACTIVE);
+        testClusterService.put(CNODE3, ControllerNode.State.ACTIVE);
+        testClusterService.put(CNODE4, ControllerNode.State.ACTIVE);
+
+        //set master to non region nodes
+        deviceIdsOrig.forEach(deviceId1 -> mgr.setRole(NID_LOCAL, deviceId1, MASTER));
+        checkDeviceMasters(deviceIds, Sets.newHashSet(NID_LOCAL), deviceId ->
+                assertEquals("wrong local role:", MASTER, mgr.getLocalRole(deviceId)));
+
+        //do region balancing
+        mgr.useRegionForBalanceRoles = true;
+        mgr.balanceRoles();
+        Set<NodeId> expectedMasters = Sets.newHashSet(NID1, NID3, NID4);
+        checkDeviceMasters(deviceIds, expectedMasters);
+
+        // make N1 inactive
+        testClusterService.put(CNODE1, ControllerNode.State.INACTIVE);
+        expectedMasters.remove(NID1);
+        mgr.balanceRoles();
+        checkDeviceMasters(deviceIds, expectedMasters);
+
+        // make N4 inactive
+        testClusterService.put(CNODE4, ControllerNode.State.INACTIVE);
+        expectedMasters.remove(NID4);
+        mgr.balanceRoles();
+        checkDeviceMasters(deviceIds, expectedMasters);
+
+        // make N3 inactive
+        testClusterService.put(CNODE3, ControllerNode.State.INACTIVE);
+        expectedMasters = Sets.newHashSet(NID2);
+        mgr.balanceRoles();
+        checkDeviceMasters(deviceIds, expectedMasters);
+
+        // make N3 active
+        testClusterService.put(CNODE3, ControllerNode.State.ACTIVE);
+        expectedMasters = Sets.newHashSet(NID3);
+        mgr.balanceRoles();
+        checkDeviceMasters(deviceIds, expectedMasters);
+
+        // make N4 active
+        testClusterService.put(CNODE4, ControllerNode.State.ACTIVE);
+        expectedMasters.add(NID4);
+        mgr.balanceRoles();
+        checkDeviceMasters(deviceIds, expectedMasters);
+
+        // make N1 active
+        testClusterService.put(CNODE1, ControllerNode.State.ACTIVE);
+        expectedMasters.add(NID1);
+        mgr.balanceRoles();
+        checkDeviceMasters(deviceIds, expectedMasters);
+    }
+
+    private void checkDeviceMasters(Set<DeviceId> deviceIds, Set<NodeId> expectedMasters) {
+        checkDeviceMasters(deviceIds, expectedMasters, null);
+    }
+
+    private void checkDeviceMasters(Set<DeviceId> deviceIds, Set<NodeId> expectedMasters,
+                                     Consumer<DeviceId> checkRole) {
+        // each device's master must be contained in the list of expectedMasters
+        deviceIds.stream().forEach(deviceId -> {
+            assertTrue("wrong master:", expectedMasters.contains(mgr.getMasterFor(deviceId)));
+            if (checkRole != null) {
+                checkRole.accept(deviceId);
+            }
+        });
+        // each node in expectedMasters must have approximately the same number of devices
+        if (expectedMasters.size() > 1) {
+            int minValue = Integer.MAX_VALUE;
+            int maxDevices = -1;
+            for (NodeId nodeId: expectedMasters) {
+                int numDevicesManagedByNode = mgr.getDevicesOf(nodeId).size();
+                if (numDevicesManagedByNode < minValue) {
+                    minValue = numDevicesManagedByNode;
+                }
+                if (numDevicesManagedByNode > maxDevices) {
+                    maxDevices = numDevicesManagedByNode;
+                }
+                assertTrue("not balanced:", maxDevices - minValue <= 1);
+            }
+        }
+    }
+
+    private final class TestClusterService extends StaticClusterService {
 
         ControllerNode local = new DefaultControllerNode(NID_LOCAL, LOCALHOST);
 
@@ -163,11 +339,10 @@ public class MastershipManagerTest {
             return local;
         }
 
-        @Override
-        public Set<ControllerNode> getNodes() {
-            return Sets.newHashSet();
+        public void put(ControllerNode cn, ControllerNode.State state) {
+            nodes.put(cn.id(), cn);
+            nodeStates.put(cn.id(), state);
         }
-
     }
 
     private final class TestSimpleMastershipStore extends SimpleMastershipStore
@@ -177,4 +352,10 @@ public class MastershipManagerTest {
             super.clusterService = clusterService;
         }
     }
+
+    private class TestRegionManager extends RegionManager {
+        TestRegionManager() {
+            eventDispatcher = new TestEventDispatcher();
+        }
+    }
 }