/*
 * 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.slider.client.ipc;

import com.google.common.annotations.VisibleForTesting;
import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.slider.api.ClusterDescription;
import org.apache.slider.api.ClusterNode;
import org.apache.slider.api.SliderClusterProtocol;
import org.apache.slider.api.StateValues;
import org.apache.slider.api.proto.Messages;

import static org.apache.slider.api.proto.RestTypeMarshalling.*;
import org.apache.slider.api.types.ApplicationLivenessInformation;
import org.apache.slider.api.types.ComponentInformation;
import org.apache.slider.api.types.ContainerInformation;
import org.apache.slider.api.types.PingInformation;
import org.apache.slider.common.tools.Duration;
import org.apache.slider.core.conf.AggregateConf;
import org.apache.slider.core.conf.ConfTree;
import org.apache.slider.core.conf.ConfTreeOperations;
import org.apache.slider.core.exceptions.NoSuchNodeException;
import org.apache.slider.core.exceptions.SliderException;
import org.apache.slider.core.exceptions.WaitTimeoutException;
import org.apache.slider.core.persist.ConfTreeSerDeser;
import org.apache.slider.server.services.security.SecurityStore;
import org.apache.slider.server.services.security.SignCertResponse;
import org.codehaus.jackson.JsonParseException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Cluster operations at a slightly higher level than the RPC code
 */
public class SliderClusterOperations {
  protected static final Logger
    log = LoggerFactory.getLogger(SliderClusterOperations.class);
  
  private final SliderClusterProtocol appMaster;
  private static final Messages.EmptyPayloadProto EMPTY;
  static {
    EMPTY = Messages.EmptyPayloadProto.newBuilder().build(); 
  }

  public SliderClusterOperations(SliderClusterProtocol appMaster) {
    this.appMaster = appMaster;
  }

  @Override
  public String toString() {
    final StringBuilder sb =
        new StringBuilder("SliderClusterOperations{");
    sb.append("IPC binding=").append(appMaster);
    sb.append('}');
    return sb.toString();
  }

  /**
   * Get a node from the AM
   * @param uuid uuid of node
   * @return deserialized node
   * @throws IOException IO problems
   * @throws NoSuchNodeException if the node isn't found
   */
  public ClusterNode getNode(String uuid)
    throws IOException, NoSuchNodeException, YarnException {
    Messages.GetNodeRequestProto req =
      Messages.GetNodeRequestProto.newBuilder().setUuid(uuid).build();
    Messages.GetNodeResponseProto node = appMaster.getNode(req);
    return ClusterNode.fromProtobuf(node.getClusterNode());
  }

  /**
   * Unmarshall a list of nodes from a protobud response
   * @param nodes node list
   * @return possibly empty list of cluster nodes
   * @throws IOException
   */
  public List<ClusterNode> convertNodeWireToClusterNodes(List<Messages.RoleInstanceState> nodes)
    throws IOException {
    List<ClusterNode> nodeList = new ArrayList<ClusterNode>(nodes.size());
    for (Messages.RoleInstanceState node : nodes) {
      nodeList.add(ClusterNode.fromProtobuf(node));
    }
    return nodeList;
  }

  /**
   * Echo text (debug action)
   * @param text text
   * @return the text, echoed back
   * @throws YarnException
   * @throws IOException
   */
  public String echo(String text) throws
                                          YarnException,
                                          IOException {
    Messages.EchoRequestProto.Builder builder =
      Messages.EchoRequestProto.newBuilder();
    builder.setText(text);
    Messages.EchoRequestProto req =
      builder.build();
    Messages.EchoResponseProto response =
      appMaster.echo(req);
    return response.getText();
  }


  /**
   * Connect to a live cluster and get its current state
   * @return its description
   */
  public ClusterDescription getClusterDescription()
    throws YarnException, IOException {
    
    Messages.GetJSONClusterStatusRequestProto req =
      Messages.GetJSONClusterStatusRequestProto.newBuilder().build();
    Messages.GetJSONClusterStatusResponseProto resp =
      appMaster.getJSONClusterStatus(req);
    String statusJson = resp.getClusterSpec();
    try {
      return ClusterDescription.fromJson(statusJson);
    } catch (JsonParseException e) {
      log.error(
        "Exception " + e + " parsing:\n" + statusJson,
        e);
      throw e;
    }
  }

  /**
   * Get the AM instance definition.
   * <p>
   *   See {@link SliderClusterProtocol#getInstanceDefinition(Messages.GetInstanceDefinitionRequestProto)}
   * @return current slider AM aggregate definition
   * @throws YarnException
   * @throws IOException
   */
  public AggregateConf getInstanceDefinition()
    throws YarnException, IOException {
    Messages.GetInstanceDefinitionRequestProto.Builder builder =
      Messages.GetInstanceDefinitionRequestProto.newBuilder();

    Messages.GetInstanceDefinitionRequestProto request = builder.build();
    Messages.GetInstanceDefinitionResponseProto response =
      appMaster.getInstanceDefinition(request);

    ConfTreeSerDeser confTreeSerDeser = new ConfTreeSerDeser();

    ConfTree internal = confTreeSerDeser.fromJson(response.getInternal());
    ConfTree resources = confTreeSerDeser.fromJson(response.getResources());
    ConfTree app = confTreeSerDeser.fromJson(response.getApplication());
    AggregateConf instanceDefinition =
      new AggregateConf(resources, app, internal);
    return instanceDefinition;
  }
  /**
   * Kill a container
   * @param id container ID
   * @return a success flag
   * @throws YarnException
   * @throws IOException
   */
  public boolean killContainer(String id) throws
                                          YarnException,
                                          IOException {
    Messages.KillContainerRequestProto.Builder builder =
      Messages.KillContainerRequestProto.newBuilder();
    builder.setId(id);
    Messages.KillContainerRequestProto req =
      builder.build();
    Messages.KillContainerResponseProto response =
      appMaster.killContainer(req);
    return response.getSuccess();
  }

  /**
   * List all node UUIDs in a role
   * @param role role name or "" for all
   * @return an array of UUID strings
   * @throws IOException
   * @throws YarnException
   */
  public String[] listNodeUUIDsByRole(String role) throws
                                                   IOException,
                                                   YarnException {
    Collection<String> uuidList = innerListNodeUUIDSByRole(role);
    String[] uuids = new String[uuidList.size()];
    return uuidList.toArray(uuids);
  }

  public List<String> innerListNodeUUIDSByRole(String role) throws
                                                             IOException,
                                                             YarnException {
    Messages.ListNodeUUIDsByRoleRequestProto req =
      Messages.ListNodeUUIDsByRoleRequestProto
              .newBuilder()
              .setRole(role)
              .build();
    Messages.ListNodeUUIDsByRoleResponseProto resp =
      appMaster.listNodeUUIDsByRole(req);
    return resp.getUuidList();
  }

  /**
   * List all nodes in a role. This is a double round trip: once to list
   * the nodes in a role, another to get their details
   * @param role
   * @return an array of ContainerNode instances
   * @throws IOException
   * @throws YarnException
   */
  public List<ClusterNode> listClusterNodesInRole(String role) throws
                                                               IOException,
                                                               YarnException {

    Collection<String> uuidList = innerListNodeUUIDSByRole(role);
    Messages.GetClusterNodesRequestProto req =
      Messages.GetClusterNodesRequestProto
              .newBuilder()
              .addAllUuid(uuidList)
              .build();
    Messages.GetClusterNodesResponseProto resp = appMaster.getClusterNodes(req);
    return convertNodeWireToClusterNodes(resp.getClusterNodeList());
  }

  /**
   * Get the details on a list of uuids
   * @param uuids
   * @return a possibly empty list of node details
   * @throws IOException
   * @throws YarnException
   */
  @VisibleForTesting
  public List<ClusterNode> listClusterNodes(String[] uuids) throws
                                                            IOException,
                                                            YarnException {

    Messages.GetClusterNodesRequestProto req =
      Messages.GetClusterNodesRequestProto
              .newBuilder()
              .addAllUuid(Arrays.asList(uuids))
              .build();
    Messages.GetClusterNodesResponseProto resp = appMaster.getClusterNodes(req);
    return convertNodeWireToClusterNodes(resp.getClusterNodeList());
  }

  /**
   * Wait for an instance of a named role to be live (or past it in the lifecycle)
   * @param role role to look for
   * @param timeout time to wait
   * @return the state. If still in CREATED, the cluster didn't come up
   * in the time period. If LIVE, all is well. If >LIVE, it has shut for a reason
   * @throws IOException IO
   * @throws SliderException Slider
   * @throws WaitTimeoutException if the wait timed out
   */
  @VisibleForTesting
  public int waitForRoleInstanceLive(String role, long timeout)
    throws WaitTimeoutException, IOException, YarnException {
    Duration duration = new Duration(timeout);
    duration.start();
    boolean live = false;
    int state = StateValues.STATE_CREATED;

    log.info("Waiting {} millis for a live node in role {}", timeout, role);
    try {
      while (!live) {
        // see if there is a node in that role yet
        List<String> uuids = innerListNodeUUIDSByRole(role);
        String[] containers = uuids.toArray(new String[uuids.size()]);
        int roleCount = containers.length;
        ClusterNode roleInstance = null;
        if (roleCount != 0) {
  
          // if there is, get the node
          roleInstance = getNode(containers[0]);
          if (roleInstance != null) {
            state = roleInstance.state;
            live = state >= StateValues.STATE_LIVE;
          }
        }
        if (!live) {
          if (duration.getLimitExceeded()) {
            throw new WaitTimeoutException(
              String.format("Timeout after %d millis" +
                            " waiting for a live instance of type %s; " +
                            "instances found %d %s",
                            timeout, role, roleCount,
                            (roleInstance != null
                             ? (" instance -\n" + roleInstance.toString())
                             : "")
                           ));
          } else {
            try {
              Thread.sleep(1000);
            } catch (InterruptedException ignored) {
              // ignored
            }
          }
        }
      }
    } finally {
      duration.close();
    }
    return state;
  }

  /**
   * Flex operation
   * @param resources new resources
   * @return the response
   * @throws IOException
   */
  public boolean flex(ConfTree resources) throws IOException {
    Messages.FlexClusterRequestProto request =
      Messages.FlexClusterRequestProto.newBuilder()
              .setClusterSpec(resources.toJson())
              .build();
    Messages.FlexClusterResponseProto response =
      appMaster.flexCluster(request);
    return response.getResponse();
  }


  /**
   * Commit (possibly delayed) AM suicide
   *
   * @param signal exit code
   * @param text text text to log
   * @param delay delay in millis
   * @throws YarnException
   * @throws IOException
   */
  public void amSuicide(String text, int signal, int delay)
      throws IOException {
    Messages.AMSuicideRequestProto.Builder builder =
      Messages.AMSuicideRequestProto.newBuilder();
    if (text != null) {
      builder.setText(text);
    }
    builder.setSignal(signal);
    builder.setDelay(delay);
    Messages.AMSuicideRequestProto req =
      builder.build();
    Messages.AMSuicideResponseProto response =
      appMaster.amSuicide(req);
  }

  /**
   * Get the application liveness
   * @return current liveness information
   * @throws IOException
   */
  public ApplicationLivenessInformation getLivenessInformation() throws IOException {
    Messages.GetApplicationLivenessRequestProto.Builder builder =
        Messages.GetApplicationLivenessRequestProto.newBuilder();
    Messages.ApplicationLivenessInformationProto wire =
        appMaster.getLivenessInformation(builder.build());
    return unmarshall(wire);

  }

   
  public AggregateConf getModelDesired() throws IOException {
    return unmarshallToAggregateConf(appMaster.getModelDesired(EMPTY));
  }

  
  public ConfTreeOperations getModelDesiredAppconf() throws IOException {
    return unmarshallToCTO(
        appMaster.getModelDesiredAppconf(EMPTY));
  }

  
  public ConfTreeOperations getModelDesiredResources() throws IOException {
    return unmarshallToCTO(
        appMaster.getModelDesiredResources(EMPTY));
  }

  
  public AggregateConf getModelResolved() throws IOException {
    return unmarshallToAggregateConf(
        appMaster.getModelResolved(EMPTY));
  }

  
  public ConfTreeOperations getModelResolvedAppconf() throws IOException {
    return unmarshallToCTO(
        appMaster.getModelResolvedAppconf(EMPTY));
  }

  
  public ConfTreeOperations getModelResolvedResources() throws IOException {
    return unmarshallToCTO(
        appMaster.getModelDesiredResources(EMPTY));
  }

  
  public ConfTreeOperations getLiveResources() throws IOException {
    return unmarshallToCTO(
        appMaster.getLiveResources(EMPTY));
  }

  
  public Map<String, ContainerInformation> enumContainers() throws IOException {
    Messages.GetLiveContainersResponseProto response =
        appMaster.getLiveContainers(
            Messages.GetLiveContainersRequestProto.newBuilder().build());

    int namesCount = response.getNamesCount();
    int records = response.getContainersCount();
    if (namesCount != records) {
      throw new IOException("Number of names returned (" + namesCount
                      + ") does not match the number of records returned: " 
                      + records);
    }
    Map<String, ContainerInformation> map =
        new HashMap<String, ContainerInformation>(namesCount);
    for (int i = 0; i < namesCount; i++) {
      map.put(response.getNames(i), unmarshall(response.getContainers(i)));
    }
    return map;
  }

  
  public ContainerInformation getContainer(String containerId) throws
      IOException {
    Messages.ContainerInformationProto response =
        appMaster.getLiveContainer(
            Messages.GetLiveContainerRequestProto.newBuilder()
                                                 .setContainerId(containerId)
                                                 .build());
    return unmarshall(response);
  }

  public List<ContainerInformation> getContainers() throws IOException {
    Messages.GetLiveContainersResponseProto response = appMaster
        .getLiveContainers(Messages.GetLiveContainersRequestProto.newBuilder()
                                                                 .build());
    return unmarshall(response);
  }

  public Map<String, ComponentInformation> enumComponents() throws IOException {
    Messages.GetLiveComponentsResponseProto response =
        appMaster.getLiveComponents(
            Messages.GetLiveComponentsRequestProto.newBuilder().build());

    int namesCount = response.getNamesCount();
    int records = response.getComponentsCount();
    if (namesCount != records) {
      throw new IOException("Number of names returned (" + namesCount
                            +
                            ") does not match the number of records returned: "
                            + records);
    }
    Map<String, ComponentInformation> map =
        new HashMap<String, ComponentInformation>(namesCount);
    for (int i = 0; i < namesCount; i++) {
      map.put(response.getNames(i), unmarshall(response.getComponents(i)));
    }
    return map;
  }

  
  public ComponentInformation getComponent(String componentName)
      throws IOException {
    Messages.GetLiveComponentRequestProto.Builder builder =
        Messages.GetLiveComponentRequestProto.newBuilder();
    builder.setName(componentName);
    Messages.ComponentInformationProto proto =
        appMaster.getLiveComponent(builder.build());
    
    return unmarshall(proto);
  }

  
  public PingInformation ping(String text) throws IOException {
    return null;
  }

  
  public void stop(String text) throws IOException {
    amSuicide(text, 3, 0);
  }

  
  public ApplicationLivenessInformation getApplicationLiveness() throws
      IOException {
    Messages.ApplicationLivenessInformationProto proto =
        appMaster.getLivenessInformation(
            Messages.GetApplicationLivenessRequestProto.newBuilder().build()
        );
    return unmarshall(proto);
  }

  public byte[] getClientCertificateStore(String hostname, String clientId,
      String password, String type) throws IOException {
    Messages.GetCertificateStoreRequestProto.Builder
        builder = Messages.GetCertificateStoreRequestProto.newBuilder();
    if (hostname != null) {
      builder.setHostname(hostname);
    }
    Messages.GetCertificateStoreRequestProto requestProto =
        builder.setRequesterId(clientId)
               .setPassword(password)
               .setType(type)
               .build();
    Messages.GetCertificateStoreResponseProto response =
        appMaster.getClientCertificateStore(requestProto);

    return unmarshall(response);
  }

}