/**
 *
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.hadoop.hbase.master;

import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.RegionInfo;
import org.apache.hadoop.hbase.favored.FavoredNodeAssignmentHelper;
import org.apache.hadoop.hbase.favored.FavoredNodesPlan;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
 * Helper class that is used by {@link RegionPlacementMaintainer} to print
 * information for favored nodes
 *
 */
@InterfaceAudience.Private
public class AssignmentVerificationReport {
  private static final Logger LOG = LoggerFactory.getLogger(
      AssignmentVerificationReport.class.getName());

  private TableName tableName = null;
  private boolean enforceLocality = false;
  private boolean isFilledUp = false;

  private int totalRegions = 0;
  private int totalRegionServers = 0;
  // for unassigned regions
  private List<RegionInfo> unAssignedRegionsList = new ArrayList<>();

  // For regions without valid favored nodes
  private List<RegionInfo> regionsWithoutValidFavoredNodes = new ArrayList<>();

  // For regions not running on the favored nodes
  private List<RegionInfo> nonFavoredAssignedRegionList = new ArrayList<>();

  // For regions running on the favored nodes
  private int totalFavoredAssignments = 0;
  private int[] favoredNodes = new int[FavoredNodeAssignmentHelper.FAVORED_NODES_NUM];
  private float[] favoredNodesLocalitySummary =
      new float[FavoredNodeAssignmentHelper.FAVORED_NODES_NUM];
  private float actualLocalitySummary = 0;

  // For region balancing information
  private float avgRegionsOnRS = 0;
  private int maxRegionsOnRS = 0;
  private int minRegionsOnRS = Integer.MAX_VALUE;
  private Set<ServerName> mostLoadedRSSet = new HashSet<>();
  private Set<ServerName> leastLoadedRSSet = new HashSet<>();

  private float avgDispersionScore = 0;
  private float maxDispersionScore = 0;
  private Set<ServerName> maxDispersionScoreServerSet = new HashSet<>();
  private float minDispersionScore = Float.MAX_VALUE;
  private Set<ServerName> minDispersionScoreServerSet = new HashSet<>();

  private float avgDispersionNum = 0;
  private float maxDispersionNum = 0;
  private Set<ServerName> maxDispersionNumServerSet = new HashSet<>();
  private float minDispersionNum = Float.MAX_VALUE;
  private Set<ServerName> minDispersionNumServerSet = new HashSet<>();

  public void fillUp(TableName tableName, SnapshotOfRegionAssignmentFromMeta snapshot,
      Map<String, Map<String, Float>> regionLocalityMap) {
    // Set the table name
    this.tableName = tableName;

    // Get all the regions for this table
    List<RegionInfo> regionInfoList =
      snapshot.getTableToRegionMap().get(tableName);
    // Get the total region num for the current table
    this.totalRegions = regionInfoList.size();

    // Get the existing assignment plan
    FavoredNodesPlan favoredNodesAssignment = snapshot.getExistingAssignmentPlan();
    // Get the region to region server mapping
    Map<RegionInfo, ServerName> currentAssignment =
      snapshot.getRegionToRegionServerMap();
    // Initialize the server to its hosing region counter map
    Map<ServerName, Integer> serverToHostingRegionCounterMap = new HashMap<>();

    Map<ServerName, Integer> primaryRSToRegionCounterMap = new HashMap<>();
    Map<ServerName, Set<ServerName>> primaryToSecTerRSMap = new HashMap<>();

    // Check the favored nodes and its locality information
    // Also keep tracker of the most loaded and least loaded region servers
    for (RegionInfo region : regionInfoList) {
      try {
        ServerName currentRS = currentAssignment.get(region);
        // Handle unassigned regions
        if (currentRS == null) {
          unAssignedRegionsList.add(region);
          continue;
        }

        // Keep updating the server to is hosting region counter map
        Integer hostRegionCounter = serverToHostingRegionCounterMap.get(currentRS);
        if (hostRegionCounter == null) {
          hostRegionCounter = Integer.valueOf(0);
        }
        hostRegionCounter = hostRegionCounter.intValue() + 1;
        serverToHostingRegionCounterMap.put(currentRS, hostRegionCounter);

        // Get the favored nodes from the assignment plan and verify it.
        List<ServerName> favoredNodes = favoredNodesAssignment.getFavoredNodes(region);
        if (favoredNodes == null ||
            favoredNodes.size() != FavoredNodeAssignmentHelper.FAVORED_NODES_NUM) {
          regionsWithoutValidFavoredNodes.add(region);
          continue;
        }
        // Get the primary, secondary and tertiary region server
        ServerName primaryRS =
          favoredNodes.get(FavoredNodesPlan.Position.PRIMARY.ordinal());
        ServerName secondaryRS =
          favoredNodes.get(FavoredNodesPlan.Position.SECONDARY.ordinal());
        ServerName tertiaryRS =
          favoredNodes.get(FavoredNodesPlan.Position.TERTIARY.ordinal());

        // Update the primary rs to its region set map
        Integer regionCounter = primaryRSToRegionCounterMap.get(primaryRS);
        if (regionCounter == null) {
          regionCounter = Integer.valueOf(0);
        }
        regionCounter = regionCounter.intValue() + 1;
        primaryRSToRegionCounterMap.put(primaryRS, regionCounter);

        // Update the primary rs to secondary and tertiary rs map
        Set<ServerName> secAndTerSet = primaryToSecTerRSMap.get(primaryRS);
        if (secAndTerSet == null) {
          secAndTerSet = new HashSet<>();
        }
        secAndTerSet.add(secondaryRS);
        secAndTerSet.add(tertiaryRS);
        primaryToSecTerRSMap.put(primaryRS, secAndTerSet);

        // Get the position of the current region server in the favored nodes list
        FavoredNodesPlan.Position favoredNodePosition =
          FavoredNodesPlan.getFavoredServerPosition(favoredNodes, currentRS);

        // Handle the non favored assignment.
        if (favoredNodePosition == null) {
          nonFavoredAssignedRegionList.add(region);
          continue;
        }
        // Increase the favored nodes assignment.
        this.favoredNodes[favoredNodePosition.ordinal()]++;
        totalFavoredAssignments++;

        // Summary the locality information for each favored nodes
        if (regionLocalityMap != null) {
          // Set the enforce locality as true;
          this.enforceLocality = true;

          // Get the region degree locality map
          Map<String, Float> regionDegreeLocalityMap =
            regionLocalityMap.get(region.getEncodedName());
          if (regionDegreeLocalityMap == null) {
            continue; // ignore the region which doesn't have any store files.
          }

          // Get the locality summary for each favored nodes
          for (FavoredNodesPlan.Position p : FavoredNodesPlan.Position.values()) {
            ServerName favoredNode = favoredNodes.get(p.ordinal());
            // Get the locality for the current favored nodes
            Float locality =
              regionDegreeLocalityMap.get(favoredNode.getHostname());
            if (locality != null) {
              this.favoredNodesLocalitySummary[p.ordinal()] += locality;
            }
          }

          // Get the locality summary for the current region server
          Float actualLocality =
            regionDegreeLocalityMap.get(currentRS.getHostname());
          if (actualLocality != null) {
            this.actualLocalitySummary += actualLocality;
          }
        }
      } catch (Exception e) {
        LOG.error("Cannot verify the region assignment for region " +
            ((region == null) ? " null " : region.getRegionNameAsString()) +
            "because of " + e);
      }
    }

    float dispersionScoreSummary = 0;
    float dispersionNumSummary = 0;
    // Calculate the secondary score for each primary region server
    for (Map.Entry<ServerName, Integer> entry :
      primaryRSToRegionCounterMap.entrySet()) {
      ServerName primaryRS = entry.getKey();
      Integer regionsOnPrimary = entry.getValue();

      // Process the dispersion number and score
      float dispersionScore = 0;
      int dispersionNum = 0;
      if (primaryToSecTerRSMap.get(primaryRS) != null
          && regionsOnPrimary.intValue() != 0) {
        dispersionNum = primaryToSecTerRSMap.get(primaryRS).size();
        dispersionScore = dispersionNum /
          ((float) regionsOnPrimary.intValue() * 2);
      }
      // Update the max dispersion score
      if (dispersionScore > this.maxDispersionScore) {
        this.maxDispersionScoreServerSet.clear();
        this.maxDispersionScoreServerSet.add(primaryRS);
        this.maxDispersionScore = dispersionScore;
      } else if (dispersionScore == this.maxDispersionScore) {
        this.maxDispersionScoreServerSet.add(primaryRS);
      }

      // Update the max dispersion num
      if (dispersionNum > this.maxDispersionNum) {
        this.maxDispersionNumServerSet.clear();
        this.maxDispersionNumServerSet.add(primaryRS);
        this.maxDispersionNum = dispersionNum;
      } else if (dispersionNum == this.maxDispersionNum) {
        this.maxDispersionNumServerSet.add(primaryRS);
      }

      // Update the min dispersion score
      if (dispersionScore < this.minDispersionScore) {
        this.minDispersionScoreServerSet.clear();
        this.minDispersionScoreServerSet.add(primaryRS);
        this.minDispersionScore = dispersionScore;
      } else if (dispersionScore == this.minDispersionScore) {
        this.minDispersionScoreServerSet.add(primaryRS);
      }

      // Update the min dispersion num
      if (dispersionNum < this.minDispersionNum) {
        this.minDispersionNumServerSet.clear();
        this.minDispersionNumServerSet.add(primaryRS);
        this.minDispersionNum = dispersionNum;
      } else if (dispersionNum == this.minDispersionNum) {
        this.minDispersionNumServerSet.add(primaryRS);
      }

      dispersionScoreSummary += dispersionScore;
      dispersionNumSummary += dispersionNum;
    }

    // Update the avg dispersion score
    if (primaryRSToRegionCounterMap.keySet().size() != 0) {
      this.avgDispersionScore = dispersionScoreSummary /
         (float) primaryRSToRegionCounterMap.keySet().size();
      this.avgDispersionNum = dispersionNumSummary /
         (float) primaryRSToRegionCounterMap.keySet().size();
    }

    // Fill up the most loaded and least loaded region server information
    for (Map.Entry<ServerName, Integer> entry :
      serverToHostingRegionCounterMap.entrySet()) {
      ServerName currentRS = entry.getKey();
      int hostRegionCounter = entry.getValue().intValue();

      // Update the most loaded region server list and maxRegionsOnRS
      if (hostRegionCounter > this.maxRegionsOnRS) {
        maxRegionsOnRS = hostRegionCounter;
        this.mostLoadedRSSet.clear();
        this.mostLoadedRSSet.add(currentRS);
      } else if (hostRegionCounter == this.maxRegionsOnRS) {
        this.mostLoadedRSSet.add(currentRS);
      }

      // Update the least loaded region server list and minRegionsOnRS
      if (hostRegionCounter < this.minRegionsOnRS) {
        this.minRegionsOnRS = hostRegionCounter;
        this.leastLoadedRSSet.clear();
        this.leastLoadedRSSet.add(currentRS);
      } else if (hostRegionCounter == this.minRegionsOnRS) {
        this.leastLoadedRSSet.add(currentRS);
      }
    }

    // and total region servers
    this.totalRegionServers = serverToHostingRegionCounterMap.keySet().size();
    this.avgRegionsOnRS = (totalRegionServers == 0) ? 0 :
      (totalRegions / (float) totalRegionServers);
    // Set the isFilledUp as true
    isFilledUp = true;
  }

  /**
   * Use this to project the dispersion scores
   * @param tableName
   * @param snapshot
   * @param newPlan
   */
  public void fillUpDispersion(TableName tableName,
      SnapshotOfRegionAssignmentFromMeta snapshot, FavoredNodesPlan newPlan) {
    // Set the table name
    this.tableName = tableName;
    // Get all the regions for this table
    List<RegionInfo> regionInfoList = snapshot.getTableToRegionMap().get(
        tableName);
    // Get the total region num for the current table
    this.totalRegions = regionInfoList.size();
    FavoredNodesPlan plan = null;
    if (newPlan == null) {
      plan = snapshot.getExistingAssignmentPlan();
    } else {
      plan = newPlan;
    }
    // Get the region to region server mapping
    Map<ServerName, Integer> primaryRSToRegionCounterMap = new HashMap<>();
    Map<ServerName, Set<ServerName>> primaryToSecTerRSMap = new HashMap<>();

    // Check the favored nodes and its locality information
    // Also keep tracker of the most loaded and least loaded region servers
    for (RegionInfo region : regionInfoList) {
      try {
        // Get the favored nodes from the assignment plan and verify it.
        List<ServerName> favoredNodes = plan.getFavoredNodes(region);
        if (favoredNodes == null
            || favoredNodes.size() != FavoredNodeAssignmentHelper.FAVORED_NODES_NUM) {
          regionsWithoutValidFavoredNodes.add(region);
          continue;
        }
        // Get the primary, secondary and tertiary region server
        ServerName primaryRS = favoredNodes
            .get(FavoredNodesPlan.Position.PRIMARY.ordinal());
        ServerName secondaryRS = favoredNodes
            .get(FavoredNodesPlan.Position.SECONDARY.ordinal());
        ServerName tertiaryRS = favoredNodes
            .get(FavoredNodesPlan.Position.TERTIARY.ordinal());

        // Update the primary rs to its region set map
        Integer regionCounter = primaryRSToRegionCounterMap.get(primaryRS);
        if (regionCounter == null) {
          regionCounter = Integer.valueOf(0);
        }
        regionCounter = regionCounter.intValue() + 1;
        primaryRSToRegionCounterMap.put(primaryRS, regionCounter);

        // Update the primary rs to secondary and tertiary rs map
        Set<ServerName> secAndTerSet = primaryToSecTerRSMap.get(primaryRS);
        if (secAndTerSet == null) {
          secAndTerSet = new HashSet<>();
        }
        secAndTerSet.add(secondaryRS);
        secAndTerSet.add(tertiaryRS);
        primaryToSecTerRSMap.put(primaryRS, secAndTerSet);
      } catch (Exception e) {
        LOG.error("Cannot verify the region assignment for region "
            + ((region == null) ? " null " : region.getRegionNameAsString())
            + "because of " + e);
      }
    }
    float dispersionScoreSummary = 0;
    float dispersionNumSummary = 0;
    // Calculate the secondary score for each primary region server
    for (Map.Entry<ServerName, Integer> entry :
      primaryRSToRegionCounterMap.entrySet()) {
      ServerName primaryRS = entry.getKey();
      Integer regionsOnPrimary = entry.getValue();

      // Process the dispersion number and score
      float dispersionScore = 0;
      int dispersionNum = 0;
      if (primaryToSecTerRSMap.get(primaryRS) != null
          && regionsOnPrimary.intValue() != 0) {
        dispersionNum = primaryToSecTerRSMap.get(primaryRS).size();
        dispersionScore = dispersionNum /
          ((float) regionsOnPrimary.intValue() * 2);
      }

      // Update the max dispersion num
      if (dispersionNum > this.maxDispersionNum) {
        this.maxDispersionNumServerSet.clear();
        this.maxDispersionNumServerSet.add(primaryRS);
        this.maxDispersionNum = dispersionNum;
      } else if (dispersionNum == this.maxDispersionNum) {
        this.maxDispersionNumServerSet.add(primaryRS);
      }

      // Update the min dispersion score
      if (dispersionScore < this.minDispersionScore) {
        this.minDispersionScoreServerSet.clear();
        this.minDispersionScoreServerSet.add(primaryRS);
        this.minDispersionScore = dispersionScore;
      } else if (dispersionScore == this.minDispersionScore) {
        this.minDispersionScoreServerSet.add(primaryRS);
      }

      // Update the min dispersion num
      if (dispersionNum < this.minDispersionNum) {
        this.minDispersionNumServerSet.clear();
        this.minDispersionNumServerSet.add(primaryRS);
        this.minDispersionNum = dispersionNum;
      } else if (dispersionNum == this.minDispersionNum) {
        this.minDispersionNumServerSet.add(primaryRS);
      }

      dispersionScoreSummary += dispersionScore;
      dispersionNumSummary += dispersionNum;
    }

    // Update the avg dispersion score
    if (primaryRSToRegionCounterMap.keySet().size() != 0) {
      this.avgDispersionScore = dispersionScoreSummary /
         (float) primaryRSToRegionCounterMap.keySet().size();
      this.avgDispersionNum = dispersionNumSummary /
         (float) primaryRSToRegionCounterMap.keySet().size();
    }
  }

  /**
   * @return list which contains just 3 elements: average dispersion score, max
   * dispersion score and min dispersion score as first, second and third element
   * respectively.
   *
   */
  public List<Float> getDispersionInformation() {
    List<Float> dispersion = new ArrayList<>();
    dispersion.add(avgDispersionScore);
    dispersion.add(maxDispersionScore);
    dispersion.add(minDispersionScore);
    return dispersion;
  }

  public void print(boolean isDetailMode) {
    if (!isFilledUp) {
      System.err.println("[Error] Region assignment verification report" +
          "hasn't been filled up");
    }
    DecimalFormat df = new java.text.DecimalFormat( "#.##");

    // Print some basic information
    System.out.println("Region Assignment Verification for Table: " + tableName +
        "\n\tTotal regions : " + totalRegions);

    // Print the number of regions on each kinds of the favored nodes
    System.out.println("\tTotal regions on favored nodes " +
        totalFavoredAssignments);
    for (FavoredNodesPlan.Position p : FavoredNodesPlan.Position.values()) {
      System.out.println("\t\tTotal regions on "+ p.toString() +
          " region servers: " + favoredNodes[p.ordinal()]);
    }
    // Print the number of regions in each kinds of invalid assignment
    System.out.println("\tTotal unassigned regions: " +
        unAssignedRegionsList.size());
    if (isDetailMode) {
      for (RegionInfo region : unAssignedRegionsList) {
        System.out.println("\t\t" + region.getRegionNameAsString());
      }
    }

    System.out.println("\tTotal regions NOT on favored nodes: " +
        nonFavoredAssignedRegionList.size());
    if (isDetailMode) {
      for (RegionInfo region : nonFavoredAssignedRegionList) {
        System.out.println("\t\t" + region.getRegionNameAsString());
      }
    }

    System.out.println("\tTotal regions without favored nodes: " +
        regionsWithoutValidFavoredNodes.size());
    if (isDetailMode) {
      for (RegionInfo region : regionsWithoutValidFavoredNodes) {
        System.out.println("\t\t" + region.getRegionNameAsString());
      }
    }

    // Print the locality information if enabled
    if (this.enforceLocality && totalRegions != 0) {
      // Print the actual locality for this table
      float actualLocality = 100 *
        this.actualLocalitySummary / (float) totalRegions;
      System.out.println("\n\tThe actual avg locality is " +
          df.format(actualLocality) + " %");

      // Print the expected locality if regions are placed on the each kinds of
      // favored nodes
      for (FavoredNodesPlan.Position p : FavoredNodesPlan.Position.values()) {
        float avgLocality = 100 *
          (favoredNodesLocalitySummary[p.ordinal()] / (float) totalRegions);
        System.out.println("\t\tThe expected avg locality if all regions" +
            " on the " + p.toString() + " region servers: "
            + df.format(avgLocality) + " %");
      }
    }

    // Print the region balancing information
    System.out.println("\n\tTotal hosting region servers: " +
        totalRegionServers);
    // Print the region balance information
    if (totalRegionServers != 0) {
      System.out.println(
          "\tAvg dispersion num: " +df.format(avgDispersionNum) +
          " hosts;\tMax dispersion num: " + df.format(maxDispersionNum) +
          " hosts;\tMin dispersion num: " + df.format(minDispersionNum) +
          " hosts;");

      System.out.println("\t\tThe number of the region servers with the max" +
          " dispersion num: " + this.maxDispersionNumServerSet.size());
      if (isDetailMode) {
        printHServerAddressSet(maxDispersionNumServerSet);
      }

      System.out.println("\t\tThe number of the region servers with the min" +
          " dispersion num: " + this.minDispersionNumServerSet.size());
      if (isDetailMode) {
        printHServerAddressSet(maxDispersionNumServerSet);
      }

      System.out.println(
          "\tAvg dispersion score: " + df.format(avgDispersionScore) +
          ";\tMax dispersion score: " + df.format(maxDispersionScore) +
          ";\tMin dispersion score: " + df.format(minDispersionScore) + ";");

      System.out.println("\t\tThe number of the region servers with the max" +
          " dispersion score: " + this.maxDispersionScoreServerSet.size());
      if (isDetailMode) {
        printHServerAddressSet(maxDispersionScoreServerSet);
      }

      System.out.println("\t\tThe number of the region servers with the min" +
          " dispersion score: " + this.minDispersionScoreServerSet.size());
      if (isDetailMode) {
        printHServerAddressSet(minDispersionScoreServerSet);
      }

      System.out.println(
          "\tAvg regions/region server: " + df.format(avgRegionsOnRS) +
          ";\tMax regions/region server: " + maxRegionsOnRS +
          ";\tMin regions/region server: " + minRegionsOnRS + ";");

      // Print the details about the most loaded region servers
      System.out.println("\t\tThe number of the most loaded region servers: "
          + mostLoadedRSSet.size());
      if (isDetailMode) {
        printHServerAddressSet(mostLoadedRSSet);
      }

      // Print the details about the least loaded region servers
      System.out.println("\t\tThe number of the least loaded region servers: "
          + leastLoadedRSSet.size());
      if (isDetailMode) {
        printHServerAddressSet(leastLoadedRSSet);
      }
    }
    System.out.println("==============================");
  }

  /**
   * Return the unassigned regions
   * @return unassigned regions
   */
  List<RegionInfo> getUnassignedRegions() {
    return unAssignedRegionsList;
  }

  /**
   * Return the regions without favored nodes
   * @return regions without favored nodes
   */
  List<RegionInfo> getRegionsWithoutValidFavoredNodes() {
    return regionsWithoutValidFavoredNodes;
  }

  /**
   * Return the regions not assigned to its favored nodes
   * @return regions not assigned to its favored nodes
   */
  List<RegionInfo> getNonFavoredAssignedRegions() {
    return nonFavoredAssignedRegionList;
  }

  /**
   * Return the number of regions assigned to their favored nodes
   * @return number of regions assigned to their favored nodes
   */
  int getTotalFavoredAssignments() {
    return totalFavoredAssignments;
  }

  /**
   * Return the number of regions based on the position (primary/secondary/
   * tertiary) assigned to their favored nodes
   * @param position
   * @return the number of regions
   */
  int getNumRegionsOnFavoredNodeByPosition(FavoredNodesPlan.Position position) {
    return favoredNodes[position.ordinal()];
  }

  private void printHServerAddressSet(Set<ServerName> serverSet) {
    if (serverSet == null) {
      return ;
    }
    int i = 0;
    for (ServerName addr : serverSet){
      if ((i++) % 3 == 0) {
        System.out.print("\n\t\t\t");
      }
      System.out.print(addr.getHostAndPort() + " ; ");
    }
    System.out.println("\n");
  }
}