Made rebalance --show-plan slightly more verbose and added yet anothe…

…r analysis for cluster balance ("zone primary").

src/java/voldemort/client/rebalance/RebalancePartitionsInfo.java
- print out hostname within plan to make it easier to read (rather than having to lookup node ID)

src/java/voldemort/utils/ClusterInstance.java
- calculate "zone primary" balance to understand which hosted partitions act as pseudo-master when zoned routing is used.

src/java/voldemort/client/rebalance/RebalancePartitionsInfo.java

@@ -6,8 +6,8 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 import java.util.Map.Entry;
+import java.util.Set;
 
 import voldemort.VoldemortException;
 import voldemort.cluster.Cluster;
@@ -165,7 +165,8 @@ public ImmutableMap<String, Object> asMap() {
                                 replicaToAddPartition.get(replicaNum));
                 } else {
                     builder.put(unbalancedStore + "replicaToAddPartitionList"
-                                + Integer.toString(replicaNum), Lists.newArrayList());
+                                        + Integer.toString(replicaNum),
+                                Lists.newArrayList());
                 }
 
                 if(replicaToDeletePartition != null
@@ -175,7 +176,8 @@ public ImmutableMap<String, Object> asMap() {
                                 replicaToDeletePartition.get(replicaNum));
                 } else {
                     builder.put(unbalancedStore + "replicaToDeletePartitionList"
-                                + Integer.toString(replicaNum), Lists.newArrayList());
+                                        + Integer.toString(replicaNum),
+                                Lists.newArrayList());
                 }
             }
         }
@@ -261,7 +263,9 @@ public List<Integer> getStealMasterPartitions() {
     @Override
     public String toString() {
         StringBuffer sb = new StringBuffer();
-        sb.append("\nRebalancePartitionsInfo(" + getStealerId() + " <--- " + getDonorId() + " ");
+        sb.append("\nRebalancePartitionsInfo(" + getStealerId() + " ["
+                  + initialCluster.getNodeById(getStealerId()).getHost() + "] <--- " + getDonorId()
+                  + " [" + initialCluster.getNodeById(getDonorId()).getHost() + "] ");
 
         for(String unbalancedStore: storeToReplicaToAddPartitionList.keySet()) {
 

src/java/voldemort/utils/ClusterInstance.java

@@ -121,19 +121,27 @@ public Pair<Double, String> analyzeBalanceVerbose() {
         HashMap<StoreDefinition, Integer> uniqueStores = KeyDistributionGenerator.getUniqueStoreDefinitionsWithCounts(storeDefs);
         List<ByteArray> keys = KeyDistributionGenerator.generateKeys(KeyDistributionGenerator.DEFAULT_NUM_KEYS);
         Set<Integer> nodeIds = cluster.getNodeIds();
+        Set<Integer> zoneIds = cluster.getZoneIds();
 
         builder.append("PARTITION DUMP\n");
         Map<Integer, Integer> primaryAggNodeIdToPartitionCount = Maps.newHashMap();
         for(Integer nodeId: nodeIds) {
             primaryAggNodeIdToPartitionCount.put(nodeId, 0);
         }
 
+        Map<Integer, Integer> aggNodeIdToZonePrimaryCount = Maps.newHashMap();
+        for(Integer nodeId: nodeIds) {
+            aggNodeIdToZonePrimaryCount.put(nodeId, 0);
+        }
+
         Map<Integer, Integer> allAggNodeIdToPartitionCount = Maps.newHashMap();
         for(Integer nodeId: nodeIds) {
             allAggNodeIdToPartitionCount.put(nodeId, 0);
         }
 
         for(StoreDefinition storeDefinition: uniqueStores.keySet()) {
+            StoreInstance storeInstance = new StoreInstance(cluster, storeDefinition);
+
             builder.append("\n");
             builder.append("Store exemplar: " + storeDefinition.getName() + "\n");
             builder.append("\tReplication factor: " + storeDefinition.getReplicationFactor() + "\n");
@@ -147,14 +155,17 @@ public Pair<Double, String> analyzeBalanceVerbose() {
                                                                                                                       storeDefinition,
                                                                                                                       true);
             Map<Integer, Integer> primaryNodeIdToPartitionCount = Maps.newHashMap();
+            Map<Integer, Integer> nodeIdToZonePrimaryCount = Maps.newHashMap();
             Map<Integer, Integer> allNodeIdToPartitionCount = Maps.newHashMap();
 
             // Print out all partitions, by replica type, per node
             builder.append("\n");
             builder.append("\tDetailed Dump:\n");
             for(Integer nodeId: nodeIds) {
-                builder.append("\tNode ID: " + nodeId + "\n");
+                builder.append("\tNode ID: " + nodeId + "in zone "
+                               + cluster.getNodeById(nodeId).getZoneId() + "\n");
                 primaryNodeIdToPartitionCount.put(nodeId, 0);
+                nodeIdToZonePrimaryCount.put(nodeId, 0);
                 allNodeIdToPartitionCount.put(nodeId, 0);
                 Set<Pair<Integer, Integer>> partitionPairs = nodeIdToAllPartitions.get(nodeId);
 
@@ -172,19 +183,47 @@ public Pair<Double, String> analyzeBalanceVerbose() {
                         partitions.add(pair.getSecond());
                     }
                     java.util.Collections.sort(partitions);
-                    builder.append("\t\t" + replicaType + " : " + partitions.size() + " : "
-                                   + partitions.toString() + "\n");
+
+                    builder.append("\t\t" + replicaType);
+                    for(int zoneId: zoneIds) {
+                        builder.append(" : z" + zoneId + " : ");
+                        List<Integer> zonePartitions = new ArrayList<Integer>();
+                        for(int partitionId: partitions) {
+                            if(cluster.getPartitionIdsInZone(zoneId).contains(partitionId)) {
+                                zonePartitions.add(partitionId);
+                            }
+                        }
+                        builder.append(zonePartitions.toString());
+
+                    }
+                    builder.append("\n");
                     if(replicaType == 0) {
                         primaryNodeIdToPartitionCount.put(nodeId,
                                                           primaryNodeIdToPartitionCount.get(nodeId)
                                                                   + partitions.size());
                     }
+
                     allNodeIdToPartitionCount.put(nodeId, allNodeIdToPartitionCount.get(nodeId)
                                                           + partitions.size());
                     replicaType++;
                 }
             }
 
+            // Go through all partition IDs and determine which node is "first"
+            // in the replicating node list for every zone. This determines the
+            // number of "zone primaries" each node hosts.
+            for(int partitionId = 0; partitionId < cluster.getNumberOfPartitions(); partitionId++) {
+                for(int zoneId: zoneIds) {
+                    for(int nodeId: storeInstance.getReplicationNodeList(partitionId)) {
+                        if(cluster.getNodeById(nodeId).getZoneId() == zoneId) {
+                            nodeIdToZonePrimaryCount.put(nodeId,
+                                                         nodeIdToZonePrimaryCount.get(nodeId) + 1);
+                            break;
+                        }
+                    }
+                }
+            }
+
             builder.append("\n");
             builder.append("\tSummary Dump:\n");
             for(Integer nodeId: nodeIds) {
@@ -193,6 +232,9 @@ public Pair<Double, String> analyzeBalanceVerbose() {
                 primaryAggNodeIdToPartitionCount.put(nodeId,
                                                      primaryAggNodeIdToPartitionCount.get(nodeId)
                                                              + (primaryNodeIdToPartitionCount.get(nodeId) * uniqueStores.get(storeDefinition)));
+                aggNodeIdToZonePrimaryCount.put(nodeId, aggNodeIdToZonePrimaryCount.get(nodeId)
+                                                        + nodeIdToZonePrimaryCount.get(nodeId)
+                                                        * uniqueStores.get(storeDefinition));
                 allAggNodeIdToPartitionCount.put(nodeId,
                                                  allAggNodeIdToPartitionCount.get(nodeId)
                                                          + (allNodeIdToPartitionCount.get(nodeId) * uniqueStores.get(storeDefinition)));
@@ -210,6 +252,10 @@ public Pair<Double, String> analyzeBalanceVerbose() {
                                                         "AGGREGATE PRIMARY-PARTITION COUNT (across all stores)");
         builder.append(summary.getSecond());
 
+        summary = summarizeBalance(aggNodeIdToZonePrimaryCount,
+                                   "AGGREGATE ZONEPRIMARY-PARTITION COUNT (across all stores)");
+        builder.append(summary.getSecond());
+
         summary = summarizeBalance(allAggNodeIdToPartitionCount,
                                    "AGGREGATE NARY-PARTITION COUNT (across all stores)");
         builder.append(summary.getSecond());