/*
 *
 * 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.rest.model;

import com.fasterxml.jackson.annotation.JsonProperty;

import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;

import javax.xml.bind.annotation.XmlAttribute;
import javax.xml.bind.annotation.XmlElement;
import javax.xml.bind.annotation.XmlElementWrapper;
import javax.xml.bind.annotation.XmlRootElement;

import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
import org.apache.hadoop.hbase.rest.ProtobufMessageHandler;
import org.apache.hadoop.hbase.rest.protobuf.generated.StorageClusterStatusMessage.StorageClusterStatus;
import org.apache.hadoop.hbase.util.ByteStringer;
import org.apache.hadoop.hbase.util.Bytes;

import org.apache.yetus.audience.InterfaceAudience;

/**
 * Representation of the status of a storage cluster:
 * <p>
 * <ul>
 * <li>regions: the total number of regions served by the cluster</li>
 * <li>requests: the total number of requests per second handled by the
 * cluster in the last reporting interval</li>
 * <li>averageLoad: the average load of the region servers in the cluster</li>
 * <li>liveNodes: detailed status of the live region servers</li>
 * <li>deadNodes: the names of region servers declared dead</li>
 * </ul>
 *
 * <pre>
 * &lt;complexType name="StorageClusterStatus"&gt;
 *   &lt;sequence&gt;
 *     &lt;element name="liveNode" type="tns:Node"
 *       maxOccurs="unbounded" minOccurs="0"&gt;
 *     &lt;/element&gt;
 *     &lt;element name="deadNode" type="string" maxOccurs="unbounded"
 *       minOccurs="0"&gt;
 *     &lt;/element&gt;
 *   &lt;/sequence&gt;
 *   &lt;attribute name="regions" type="int"&gt;&lt;/attribute&gt;
 *   &lt;attribute name="requests" type="int"&gt;&lt;/attribute&gt;
 *   &lt;attribute name="averageLoad" type="float"&gt;&lt;/attribute&gt;
 * &lt;/complexType&gt;
 *
 * &lt;complexType name="Node"&gt;
 *   &lt;sequence&gt;
 *     &lt;element name="region" type="tns:Region"
 *       maxOccurs="unbounded" minOccurs="0"&gt;&lt;/element&gt;
 *   &lt;/sequence&gt;
 *   &lt;attribute name="name" type="string"&gt;&lt;/attribute&gt;
 *   &lt;attribute name="startCode" type="int"&gt;&lt;/attribute&gt;
 *   &lt;attribute name="requests" type="int"&gt;&lt;/attribute&gt;
 *   &lt;attribute name="heapSizeMB" type="int"&gt;&lt;/attribute&gt;
 *   &lt;attribute name="maxHeapSizeMB" type="int"&gt;&lt;/attribute&gt;
 * &lt;/complexType&gt;
 *
 * &lt;complexType name="Region"&gt;
 *   &lt;attribute name="name" type="base64Binary"&gt;&lt;/attribute&gt;
 *   &lt;attribute name="stores" type="int"&gt;&lt;/attribute&gt;
 *   &lt;attribute name="storefiles" type="int"&gt;&lt;/attribute&gt;
 *   &lt;attribute name="storefileSizeMB" type="int"&gt;&lt;/attribute&gt;
 *   &lt;attribute name="memstoreSizeMB" type="int"&gt;&lt;/attribute&gt;
 *   &lt;attribute name="storefileIndexSizeMB" type="int"&gt;&lt;/attribute&gt;
 *   &lt;attribute name="readRequestsCount" type="int"&gt;&lt;/attribute&gt;
 *   &lt;attribute name="writeRequestsCount" type="int"&gt;&lt;/attribute&gt;
 *   &lt;attribute name="rootIndexSizeKB" type="int"&gt;&lt;/attribute&gt;
 *   &lt;attribute name="totalStaticIndexSizeKB" type="int"&gt;&lt;/attribute&gt;
 *   &lt;attribute name="totalStaticBloomSizeKB" type="int"&gt;&lt;/attribute&gt;
 *   &lt;attribute name="totalCompactingKVs" type="int"&gt;&lt;/attribute&gt;
 *   &lt;attribute name="currentCompactedKVs" type="int"&gt;&lt;/attribute&gt;
 * &lt;/complexType&gt;
 * </pre>
 */
@XmlRootElement(name="ClusterStatus")
@InterfaceAudience.Private
public class StorageClusterStatusModel implements Serializable, ProtobufMessageHandler {
  private static final long serialVersionUID = 1L;

  /**
   * Represents a region server.
   */
  public static class Node implements Serializable {
    private static final long serialVersionUID = 1L;

    /**
     * Represents a region hosted on a region server.
     */
    public static class Region implements Serializable {
      private static final long serialVersionUID = -1326683840086398193L;

      private byte[] name;
      private int stores;
      private int storefiles;
      private int storefileSizeMB;
      private int memstoreSizeMB;
      private long storefileIndexSizeKB;
      private long readRequestsCount;
      private long writeRequestsCount;
      private int rootIndexSizeKB;
      private int totalStaticIndexSizeKB;
      private int totalStaticBloomSizeKB;
      private long totalCompactingKVs;
      private long currentCompactedKVs;

      /**
       * Default constructor
       */
      public Region() {
      }

      /**
       * Constructor
       * @param name the region name
       */
      public Region(byte[] name) {
        this.name = name;
      }

      /**
       * Constructor
       * @param name the region name
       * @param stores the number of stores
       * @param storefiles the number of store files
       * @param storefileSizeMB total size of store files, in MB
       * @param memstoreSizeMB total size of memstore, in MB
       * @param storefileIndexSizeKB total size of store file indexes, in KB
       */
      public Region(byte[] name, int stores, int storefiles,
          int storefileSizeMB, int memstoreSizeMB, long storefileIndexSizeKB,
          long readRequestsCount, long writeRequestsCount, int rootIndexSizeKB,
          int totalStaticIndexSizeKB, int totalStaticBloomSizeKB,
          long totalCompactingKVs, long currentCompactedKVs) {
        this.name = name;
        this.stores = stores;
        this.storefiles = storefiles;
        this.storefileSizeMB = storefileSizeMB;
        this.memstoreSizeMB = memstoreSizeMB;
        this.storefileIndexSizeKB = storefileIndexSizeKB;
        this.readRequestsCount = readRequestsCount;
        this.writeRequestsCount = writeRequestsCount;
        this.rootIndexSizeKB = rootIndexSizeKB;
        this.totalStaticIndexSizeKB = totalStaticIndexSizeKB;
        this.totalStaticBloomSizeKB = totalStaticBloomSizeKB;
        this.totalCompactingKVs = totalCompactingKVs;
        this.currentCompactedKVs = currentCompactedKVs;
      }

      /**
       * @return the region name
       */
      @XmlAttribute
      public byte[] getName() {
        return name;
      }

      /**
       * @return the number of stores
       */
      @XmlAttribute
      public int getStores() {
        return stores;
      }

      /**
       * @return the number of store files
       */
      @XmlAttribute
      public int getStorefiles() {
        return storefiles;
      }

      /**
       * @return the total size of store files, in MB
       */
      @XmlAttribute
      public int getStorefileSizeMB() {
        return storefileSizeMB;
      }

      /**
       * @return memstore size, in MB
       */
      @XmlAttribute
      public int getMemStoreSizeMB() {
        return memstoreSizeMB;
      }

      /**
       * @return the total size of store file indexes, in KB
       */
      @XmlAttribute
      public long getStorefileIndexSizeKB() {
        return storefileIndexSizeKB;
      }

      /**
       * @return the current total read requests made to region
       */
      @XmlAttribute
      public long getReadRequestsCount() {
        return readRequestsCount;
      }

      /**
       * @return the current total write requests made to region
       */
      @XmlAttribute
      public long getWriteRequestsCount() {
        return writeRequestsCount;
      }

      /**
       * @return The current total size of root-level indexes for the region, in KB.
       */
      @XmlAttribute
      public int getRootIndexSizeKB() {
        return rootIndexSizeKB;
      }

      /**
       * @return The total size of static index, in KB
       */
      @XmlAttribute
      public int getTotalStaticIndexSizeKB() {
        return totalStaticIndexSizeKB;
      }

      /**
       * @return The total size of static bloom, in KB
       */
      @XmlAttribute
      public int getTotalStaticBloomSizeKB() {
        return totalStaticBloomSizeKB;
      }

      /**
       * @return The total number of compacting key-values
       */
      @XmlAttribute
      public long getTotalCompactingKVs() {
        return totalCompactingKVs;
      }

      /**
       * @return The number of current compacted key-values
       */
      @XmlAttribute
      public long getCurrentCompactedKVs() {
        return currentCompactedKVs;
      }

      /**
       * @param readRequestsCount The current total read requests made to region
       */
      public void setReadRequestsCount(long readRequestsCount) {
        this.readRequestsCount = readRequestsCount;
      }

      /**
       * @param rootIndexSizeKB The current total size of root-level indexes
       *                        for the region, in KB
       */
      public void setRootIndexSizeKB(int rootIndexSizeKB) {
        this.rootIndexSizeKB = rootIndexSizeKB;
      }

      /**
       * @param writeRequestsCount The current total write requests made to region
       */
      public void setWriteRequestsCount(long writeRequestsCount) {
        this.writeRequestsCount = writeRequestsCount;
      }

      /**
       * @param currentCompactedKVs The completed count of key values
       *                            in currently running compaction
       */
      public void setCurrentCompactedKVs(long currentCompactedKVs) {
        this.currentCompactedKVs = currentCompactedKVs;
      }

      /**
       * @param totalCompactingKVs The total compacting key values
       *                           in currently running compaction
       */
      public void setTotalCompactingKVs(long totalCompactingKVs) {
        this.totalCompactingKVs = totalCompactingKVs;
      }

      /**
       * @param totalStaticBloomSizeKB The total size of all Bloom filter blocks,
       *                               not just loaded into the block cache, in KB.
       */
      public void setTotalStaticBloomSizeKB(int totalStaticBloomSizeKB) {
        this.totalStaticBloomSizeKB = totalStaticBloomSizeKB;
      }

      /**
       * @param totalStaticIndexSizeKB The total size of all index blocks,
       *                               not just the root level, in KB.
       */
      public void setTotalStaticIndexSizeKB(int totalStaticIndexSizeKB) {
        this.totalStaticIndexSizeKB = totalStaticIndexSizeKB;
      }

      /**
       * @param name the region name
       */
      public void setName(byte[] name) {
        this.name = name;
      }

      /**
       * @param stores the number of stores
       */
      public void setStores(int stores) {
        this.stores = stores;
      }

      /**
       * @param storefiles the number of store files
       */
      public void setStorefiles(int storefiles) {
        this.storefiles = storefiles;
      }

      /**
       * @param storefileSizeMB total size of store files, in MB
       */
      public void setStorefileSizeMB(int storefileSizeMB) {
        this.storefileSizeMB = storefileSizeMB;
      }

      /**
       * @param memstoreSizeMB memstore size, in MB
       */
      public void setMemStoreSizeMB(int memstoreSizeMB) {
        this.memstoreSizeMB = memstoreSizeMB;
      }

      /**
       * @param storefileIndexSizeKB total size of store file indexes, in KB
       */
      public void setStorefileIndexSizeKB(long storefileIndexSizeKB) {
        this.storefileIndexSizeKB = storefileIndexSizeKB;
      }
    }

    private String name;
    private long startCode;
    private long requests;
    private int heapSizeMB;
    private int maxHeapSizeMB;
    private List<Region> regions = new ArrayList<>();

    /**
     * Add a region name to the list
     * @param name the region name
     */
    public void addRegion(byte[] name, int stores, int storefiles,
        int storefileSizeMB, int memstoreSizeMB, long storefileIndexSizeKB,
        long readRequestsCount, long writeRequestsCount, int rootIndexSizeKB,
        int totalStaticIndexSizeKB, int totalStaticBloomSizeKB,
        long totalCompactingKVs, long currentCompactedKVs) {
      regions.add(new Region(name, stores, storefiles, storefileSizeMB,
        memstoreSizeMB, storefileIndexSizeKB, readRequestsCount,
        writeRequestsCount, rootIndexSizeKB, totalStaticIndexSizeKB,
        totalStaticBloomSizeKB, totalCompactingKVs, currentCompactedKVs));
    }

    /**
     * @param index the index
     * @return the region name
     */
    public Region getRegion(int index) {
      return regions.get(index);
    }

    /**
     * Default constructor
     */
    public Node() {}

    /**
     * Constructor
     * @param name the region server name
     * @param startCode the region server's start code
     */
    public Node(String name, long startCode) {
      this.name = name;
      this.startCode = startCode;
    }

    /**
     * @return the region server's name
     */
    @XmlAttribute
    public String getName() {
      return name;
    }

    /**
     * @return the region server's start code
     */
    @XmlAttribute
    public long getStartCode() {
      return startCode;
    }

    /**
     * @return the current heap size, in MB
     */
    @XmlAttribute
    public int getHeapSizeMB() {
      return heapSizeMB;
    }

    /**
     * @return the maximum heap size, in MB
     */
    @XmlAttribute
    public int getMaxHeapSizeMB() {
      return maxHeapSizeMB;
    }

    /**
     * @return the list of regions served by the region server
     */
    @XmlElement(name="Region")
    public List<Region> getRegions() {
      return regions;
    }

    /**
     * @return the number of requests per second processed by the region server
     */
    @XmlAttribute
    public long getRequests() {
      return requests;
    }

    /**
     * @param name the region server's hostname
     */
    public void setName(String name) {
      this.name = name;
    }

    /**
     * @param startCode the region server's start code
     */
    public void setStartCode(long startCode) {
      this.startCode = startCode;
    }

    /**
     * @param heapSizeMB the current heap size, in MB
     */
    public void setHeapSizeMB(int heapSizeMB) {
      this.heapSizeMB = heapSizeMB;
    }

    /**
     * @param maxHeapSizeMB the maximum heap size, in MB
     */
    public void setMaxHeapSizeMB(int maxHeapSizeMB) {
      this.maxHeapSizeMB = maxHeapSizeMB;
    }

    /**
     * @param regions a list of regions served by the region server
     */
    public void setRegions(List<Region> regions) {
      this.regions = regions;
    }

    /**
     * @param requests the number of requests per second processed by the region server
     */
    public void setRequests(long requests) {
      this.requests = requests;
    }
  }

  private List<Node> liveNodes = new ArrayList<>();
  private List<String> deadNodes = new ArrayList<>();
  private int regions;
  private long requests;
  private double averageLoad;

  /**
   * Add a live node to the cluster representation.
   * @param name the region server name
   * @param startCode the region server's start code
   * @param heapSizeMB the current heap size, in MB
   * @param maxHeapSizeMB the maximum heap size, in MB
   */
  public Node addLiveNode(String name, long startCode, int heapSizeMB, int maxHeapSizeMB) {
    Node node = new Node(name, startCode);
    node.setHeapSizeMB(heapSizeMB);
    node.setMaxHeapSizeMB(maxHeapSizeMB);
    liveNodes.add(node);
    return node;
  }

  /**
   * @param index the index
   * @return the region server model
   */
  public Node getLiveNode(int index) {
    return liveNodes.get(index);
  }

  /**
   * Add a dead node to the cluster representation.
   * @param node the dead region server's name
   */
  public void addDeadNode(String node) {
    deadNodes.add(node);
  }

  /**
   * @param index the index
   * @return the dead region server's name
   */
  public String getDeadNode(int index) {
    return deadNodes.get(index);
  }

  /**
   * Default constructor
   */
  public StorageClusterStatusModel() {
  }

  /**
   * @return the list of live nodes
   */
  @XmlElement(name = "Node")
  @XmlElementWrapper(name = "LiveNodes")
  // workaround https://github.com/FasterXML/jackson-dataformat-xml/issues/192
  @JsonProperty("LiveNodes")
  public List<Node> getLiveNodes() {
    return liveNodes;
  }

  /**
   * @return the list of dead nodes
   */
  @XmlElement(name = "Node")
  @XmlElementWrapper(name = "DeadNodes")
  // workaround https://github.com/FasterXML/jackson-dataformat-xml/issues/192
  @JsonProperty("DeadNodes")
  public List<String> getDeadNodes() {
    return deadNodes;
  }

  /**
   * @return the total number of regions served by the cluster
   */
  @XmlAttribute
  public int getRegions() {
    return regions;
  }

  /**
   * @return the total number of requests per second handled by the cluster in the last reporting
   *    interval
   */
  @XmlAttribute
  public long getRequests() {
    return requests;
  }

  /**
   * @return the average load of the region servers in the cluster
   */
  @XmlAttribute
  public double getAverageLoad() {
    return averageLoad;
  }

  /**
   * @param nodes the list of live node models
   */
  public void setLiveNodes(List<Node> nodes) {
    this.liveNodes = nodes;
  }

  /**
   * @param nodes the list of dead node names
   */
  public void setDeadNodes(List<String> nodes) {
    this.deadNodes = nodes;
  }

  /**
   * @param regions the total number of regions served by the cluster
   */
  public void setRegions(int regions) {
    this.regions = regions;
  }

  /**
   * @param requests the total number of requests per second handled by the cluster
   */
  public void setRequests(long requests) {
    this.requests = requests;
  }

  /**
   * @param averageLoad the average load of region servers in the cluster
   */
  public void setAverageLoad(double averageLoad) {
    this.averageLoad = averageLoad;
  }

  @Override
  public String toString() {
    StringBuilder sb = new StringBuilder();
    sb.append(String.format("%d live servers, %d dead servers, " +
      "%.4f average load%n%n", liveNodes.size(), deadNodes.size(),
      averageLoad));
    if (!liveNodes.isEmpty()) {
      sb.append(liveNodes.size());
      sb.append(" live servers\n");
      for (Node node : liveNodes) {
        sb.append("    ");
        sb.append(node.name);
        sb.append(' ');
        sb.append(node.startCode);
        sb.append("\n        requests=");
        sb.append(node.requests);
        sb.append(", regions=");
        sb.append(node.regions.size());
        sb.append("\n        heapSizeMB=");
        sb.append(node.heapSizeMB);
        sb.append("\n        maxHeapSizeMB=");
        sb.append(node.maxHeapSizeMB);
        sb.append("\n\n");
        for (Node.Region region : node.regions) {
          sb.append("        ");
          sb.append(Bytes.toString(region.name));
          sb.append("\n            stores=");
          sb.append(region.stores);
          sb.append("\n            storefiless=");
          sb.append(region.storefiles);
          sb.append("\n            storefileSizeMB=");
          sb.append(region.storefileSizeMB);
          sb.append("\n            memstoreSizeMB=");
          sb.append(region.memstoreSizeMB);
          sb.append("\n            storefileIndexSizeKB=");
          sb.append(region.storefileIndexSizeKB);
          sb.append("\n            readRequestsCount=");
          sb.append(region.readRequestsCount);
          sb.append("\n            writeRequestsCount=");
          sb.append(region.writeRequestsCount);
          sb.append("\n            rootIndexSizeKB=");
          sb.append(region.rootIndexSizeKB);
          sb.append("\n            totalStaticIndexSizeKB=");
          sb.append(region.totalStaticIndexSizeKB);
          sb.append("\n            totalStaticBloomSizeKB=");
          sb.append(region.totalStaticBloomSizeKB);
          sb.append("\n            totalCompactingKVs=");
          sb.append(region.totalCompactingKVs);
          sb.append("\n            currentCompactedKVs=");
          sb.append(region.currentCompactedKVs);
          sb.append('\n');
        }
        sb.append('\n');
      }
    }
    if (!deadNodes.isEmpty()) {
      sb.append('\n');
      sb.append(deadNodes.size());
      sb.append(" dead servers\n");
      for (String node : deadNodes) {
        sb.append("    ");
        sb.append(node);
        sb.append('\n');
      }
    }
    return sb.toString();
  }

  @Override
  public byte[] createProtobufOutput() {
    StorageClusterStatus.Builder builder = StorageClusterStatus.newBuilder();
    builder.setRegions(regions);
    builder.setRequests(requests);
    builder.setAverageLoad(averageLoad);
    for (Node node : liveNodes) {
      StorageClusterStatus.Node.Builder nodeBuilder =
        StorageClusterStatus.Node.newBuilder();
      nodeBuilder.setName(node.name);
      nodeBuilder.setStartCode(node.startCode);
      nodeBuilder.setRequests(node.requests);
      nodeBuilder.setHeapSizeMB(node.heapSizeMB);
      nodeBuilder.setMaxHeapSizeMB(node.maxHeapSizeMB);
      for (Node.Region region : node.regions) {
        StorageClusterStatus.Region.Builder regionBuilder =
          StorageClusterStatus.Region.newBuilder();
        regionBuilder.setName(ByteStringer.wrap(region.name));
        regionBuilder.setStores(region.stores);
        regionBuilder.setStorefiles(region.storefiles);
        regionBuilder.setStorefileSizeMB(region.storefileSizeMB);
        regionBuilder.setMemStoreSizeMB(region.memstoreSizeMB);
        regionBuilder.setStorefileIndexSizeKB(region.storefileIndexSizeKB);
        regionBuilder.setReadRequestsCount(region.readRequestsCount);
        regionBuilder.setWriteRequestsCount(region.writeRequestsCount);
        regionBuilder.setRootIndexSizeKB(region.rootIndexSizeKB);
        regionBuilder.setTotalStaticIndexSizeKB(region.totalStaticIndexSizeKB);
        regionBuilder.setTotalStaticBloomSizeKB(region.totalStaticBloomSizeKB);
        regionBuilder.setTotalCompactingKVs(region.totalCompactingKVs);
        regionBuilder.setCurrentCompactedKVs(region.currentCompactedKVs);
        nodeBuilder.addRegions(regionBuilder);
      }
      builder.addLiveNodes(nodeBuilder);
    }
    for (String node : deadNodes) {
      builder.addDeadNodes(node);
    }
    return builder.build().toByteArray();
  }

  @Override
  public ProtobufMessageHandler getObjectFromMessage(byte[] message) throws IOException {
    StorageClusterStatus.Builder builder = StorageClusterStatus.newBuilder();
    ProtobufUtil.mergeFrom(builder, message);
    if (builder.hasRegions()) {
      regions = builder.getRegions();
    }
    if (builder.hasRequests()) {
      requests = builder.getRequests();
    }
    if (builder.hasAverageLoad()) {
      averageLoad = builder.getAverageLoad();
    }
    for (StorageClusterStatus.Node node : builder.getLiveNodesList()) {
      long startCode = node.hasStartCode() ? node.getStartCode() : -1;
      StorageClusterStatusModel.Node nodeModel =
        addLiveNode(node.getName(), startCode, node.getHeapSizeMB(),
          node.getMaxHeapSizeMB());
      long requests = node.hasRequests() ? node.getRequests() : 0;
      nodeModel.setRequests(requests);
      for (StorageClusterStatus.Region region : node.getRegionsList()) {
        nodeModel.addRegion(
          region.getName().toByteArray(),
          region.getStores(),
          region.getStorefiles(),
          region.getStorefileSizeMB(),
          region.getMemStoreSizeMB(),
          region.getStorefileIndexSizeKB(),
          region.getReadRequestsCount(),
          region.getWriteRequestsCount(),
          region.getRootIndexSizeKB(),
          region.getTotalStaticIndexSizeKB(),
          region.getTotalStaticBloomSizeKB(),
          region.getTotalCompactingKVs(),
          region.getCurrentCompactedKVs());
      }
    }
    for (String node : builder.getDeadNodesList()) {
      addDeadNode(node);
    }
    return this;
  }
}