public class CommonConfigurationKeys {

  /** Location of zookeeper servers */
  public static final String ZOOKEEPER_ADDRESS_KEY = "hms.zookeeper.address";
  /** Default location of zookeeper servers */
  public static final String ZOOKEEPER_ADDRESS_DEFAULT = "localhost:2181";
  
  /** Path to zookeeper cluster root */
  public static final String ZOOKEEPER_CLUSTER_ROOT_KEY = "hms.zookeeper.cluster.path";
  /** Default location of zookeeper cluster root */
  public static final String ZOOKEEPER_CLUSTER_ROOT_DEFAULT = "/clusters";
  
  /** Path to zookeeper command queue */
  public static final String ZOOKEEPER_COMMAND_QUEUE_PATH_KEY = "hms.zookeeper.command.queue.path";
  /** Default location of zookeeper command queue */
  public static final String ZOOKEEPER_COMMAND_QUEUE_PATH_DEFAULT = "/cmdqueue";
  
  /** Path to zookeeper live controller queue */
  public static final String ZOOKEEPER_LIVE_CONTROLLER_PATH_KEY = "hms.zookeeper.live.controller.path";
  /** Default location of zookeeper live controller queue */
  public static final String ZOOKEEPER_LIVE_CONTROLLER_PATH_DEFAULT = "/livecontrollers";
  
  /** Path to zookeeper lock queue */
  public static final String ZOOKEEPER_LOCK_QUEUE_PATH_KEY = "hms.zookeeper.lock.queue.path";
  /** Default location of zookeeper lock queue */
  public static final String ZOOKEEPER_LOCK_QUEUE_PATH_DEFAULT = "/locks";
 
  /** Reference key for path to nodes manifest */
  public static final String ZOOKEEPER_NODES_MANIFEST_KEY = "hms.nodes.manifest.path";
  /** Default location of nodes manifest */
  public static final String ZOOKEEPER_NODES_MANIFEST_PATH_DEFAULT = "/nodes-manifest";
  
  /** Zeroconf zookeeper type */
  public static final String ZEROCONF_ZOOKEEPER_TYPE = "_zookeeper._tcp.local.";
  
  /** Path to zookeeper status qeueue */
  public static final String ZOOKEEPER_STATUS_QUEUE_PATH_KEY = "hms.zookeeper.status.queue.path";
  /** Default location of zookeeper status queue */
  public static final String ZOOKEEPER_STATUS_QUEUE_PATH_DEFAULT = "/status";
  
  /** Path to zookeeper software manifest */
  public static final String ZOOKEEPER_SOFTWARE_MANIFEST_KEY = "hms.software.manifest.path";
  
  /** Default location of software manifest */
  public static final String ZOOKEEPER_SOFTWARE_MANIFEST_PATH_DEFAULT = "/software-manifest";
  
  /** Path to zookeeper config blueprint */
  public static final String ZOOKEEPER_CONFIG_BLUEPRINT_PATH_DEFAULT = "/config-blueprint";
} @XmlRootElement
@XmlAccessorType(XmlAccessType.FIELD)
@XmlType(name = "", propOrder = {})
public enum Status {
  UNQUEUED, QUEUED, STARTED, SUCCEEDED, FAILED, INSTALLED, STOPPED;

  public static class StatusAdapter extends XmlAdapter<String, Status> {

    @Override
    public String marshal(Status obj) throws Exception {
      return obj.toString();
    }

    @Override
    public Status unmarshal(String str) throws Exception {
      for (Status j : Status.class.getEnumConstants()) {
        if (j.toString().equals(str)) {
          return j;
        }
      }
      throw new Exception("Can't convert " + str + " to "
          + Status.class.getName());
    }

  }
} @JsonTypeInfo(use=JsonTypeInfo.Id.CLASS, include=JsonTypeInfo.As.PROPERTY, property="@action")
@XmlSeeAlso({ ScriptAction.class, DaemonAction.class, PackageAction.class })
@XmlRootElement
public abstract class Action extends RestSource {
  @XmlElement
  protected int actionId;
  
  /**
   * Reference to original command which generated this action.
   */
  @XmlElement
  protected String cmdPath;
  
  /**
   * Unique identifier of the action type.
   */
  @XmlElement
  protected String actionType;
  
  /**
   * A list of states, this action depends on.
   */
  @XmlElement
  protected List<ActionDependency> dependencies;
  
  /**
   * When the action is successfully executed, expectedResults stores the state
   * entry for the action.
   */
  @XmlElement
  protected List<StateEntry> expectedResults;
  
  /**
   * Role is a reference to a list of nodes that should execute this action.
   */
  @XmlElement
  protected String role;
  
  public int getActionId() {
    return actionId;
  }
  
  public String getCmdPath() {
    return cmdPath;
  }
  
  public String getActionType() {
    return actionType;
  }
  
  public List<ActionDependency> getDependencies() {
    return dependencies;
  }
  
  public List<StateEntry> getExpectedResults() {
    return expectedResults;
  }
  
  public String getRole() {
    return role;
  }
  
  public void setActionId(int actionId) {
    this.actionId = actionId;
  }
  
  public void setCmdPath(String cmdPath) {
    this.cmdPath = cmdPath;
  }
  
  public void setActionType(String actionType) {
    this.actionType = actionType;
  }
  
  public void setDependencies(List<ActionDependency> dependencies) {
    this.dependencies = dependencies;
  }
  
  public void setExpectedResults(List<StateEntry> expectedResults) {
    this.expectedResults = expectedResults;
  }
  
  public void setRole(String role) {
    this.role = role;
  }
  
  public String toString() {
    StringBuilder sb = new StringBuilder();
    sb.append("actionId=");
    sb.append(actionId);
    sb.append(", cmdPath=");
    sb.append(cmdPath);
    sb.append(", actionType=");
    sb.append(actionType);
    if (role != null) {
      sb.append(", role=");
      sb.append(role);
    }
    sb.append(", dependencies=[");
    if (dependencies != null) {
      for(ActionDependency a : dependencies) {
        sb.append(a);
        sb.append(", ");
      }
    }
    sb.append("]");
    sb.append(", expectedResults=[");
    if (expectedResults != null) {
      for(StateEntry a : expectedResults) {
        sb.append(a);
        sb.append(", ");
      }
    }
    sb.append("]");
    return sb.toString();
  }
} @XmlRootElement
@XmlAccessorType(XmlAccessType.FIELD) 
@XmlType(name="", propOrder = {})
public class ActionDependency extends RestSource {
    @XmlElement
    protected List<String> hosts;
    @XmlElement
    protected List<StateEntry> states;
    @XmlElement
    protected Set<String> roles;
    
    public ActionDependency(){
    }
    
    public ActionDependency(Set<String> roles, List<StateEntry> states) {
      this.roles = roles;
      this.states = states;
    }
    
    public ActionDependency(List<String> hosts, List<StateEntry> states) {
      this.hosts = hosts;
      this.states = states;
    }
    
    public List<String> getHosts() {
      return hosts;
    }
    
    public List<StateEntry> getStates() {
      return states;
    }
    
    public Set<String> getRoles() {
      return roles;
    }
    
    public void setHosts(List<String> hosts) {
      this.hosts = hosts;
    }
    
    public void setStates(List<StateEntry> states) {
      this.states = states;
    }
    
    public void setRoles(Set<String> roles) {
      this.roles = roles;
    }
    
    public String toString() {
      StringBuilder sb = new StringBuilder();
      sb.append("[hosts={");
      if (hosts != null) {
        for(String a : hosts) {
          sb.append(a);
          sb.append(", ");
        }
      }
      sb.append("}, states={");
      if (states != null) {
        for(StateEntry a : states) {
          sb.append(a);
          sb.append(", ");
        }
      }
      sb.append("}]");
      return sb.toString();
    }
} @XmlRootElement
@XmlType(propOrder = { "daemonName" })
public class DaemonAction extends Action {

  @XmlElement(name="daemon")
  private String daemonName;
  
  public String getDaemonName() {
    return daemonName;
  }
  
  public void setDaemonName(String daemonName) {
    this.daemonName = daemonName;
  }
  
  public String toString() {
    StringBuilder sb = new StringBuilder();
    sb.append(super.toString());
    sb.append(", daemon=");
    sb.append(daemonName);
    return sb.toString();
  }
} @XmlRootElement
@XmlType(propOrder = { "packages", "dryRun" })
public class PackageAction extends Action {
  @XmlElement
  private PackageInfo[] packages;
  @XmlElement(name="dry-run")
  private boolean dryRun = false;
  
  public PackageInfo[] getPackages() {
    return packages;
  }
  
  public boolean getDryRun() {
    return dryRun;
  }

  public void setPackages(PackageInfo[] packages) {
    this.packages = packages;
  }
  
  public void setDryRun(boolean dryRun) {
    this.dryRun = dryRun;
  }
  
  public String toString() {
    StringBuilder sb = new StringBuilder();
    sb.append(super.toString());
    sb.append(", dry-run=");
    sb.append(dryRun);
    sb.append(", packages=");
    if (packages != null) {
      for(PackageInfo p : packages) {
        sb.append(p);
        sb.append(" ");
      }
    }
    return sb.toString();
  }
} @XmlRootElement
@XmlType(propOrder = { "script", "parameters" })
public class ScriptAction extends Action {
  @XmlElement
  private String script;
  @XmlElement(name="parameters")
  private String[] parameters;
  
  public String getScript() {
    return script;
  }
  
  public String[] getParameters() {
    return parameters;
  }
  
  public void setScript(String script) {
    this.script =script;
  }
  
  public void setParameters(String[] parameters) {
    this.parameters = parameters;
  }
  
  public String toString() {
    StringBuilder sb = new StringBuilder();
    sb.append(super.toString());
    sb.append(", script=");
    sb.append(script);
    sb.append(", parameters=");
    if (parameters != null) {
      for (String p : parameters) {
        sb.append(p);
        sb.append(" ");
      }
    }
    if (role != null) {
      sb.append(", role=");
      sb.append(role);
    }
    return sb.toString();
  }
} @XmlRootElement
@XmlAccessorType(XmlAccessType.FIELD)
public static class StateEntry {
  private static final int PRIME = 16777619;
    
  @XmlElement
  @XmlJavaTypeAdapter(StateTypeAdapter.class)
  protected StateType type;
  @XmlElement
  protected String name;
  @XmlElement
  @XmlJavaTypeAdapter(StatusAdapter.class)
  protected Status status;
    
  public StateEntry(){
  }
    
  public StateEntry(StateType type, String name, Status status) {
    this.type = type;
    this.name = name;
    this.status = status;
  }
    
  public StateType getType() {
    return type;
  }
    
  public String getName() {
    return name;
  }
    
  public Status getStatus() {
    return status;
  }
    
  public void setType(StateType type) {
    this.type = type;
  }
    
  public void setName(String name) {
    this.name = name;
  }
    
  public void setStatus(Status status) {
    this.status = status;
  }
    
  static boolean isEqual(Object a, Object b) {
    return a == null ? b == null : a.equals(b);
  }

  @Override
  public boolean equals(Object obj) {
    if (obj == this) {
      return true;
    }
    if (obj instanceof StateEntry) {
      StateEntry that = (StateEntry) obj;
      return this.type == that.type
          && isEqual(this.name, that.name);
    }
    return false;
  }
    
  @Override
  public int hashCode() {
    int result = 1;
    result = PRIME * result + ((type == null) ? 0 : type.hashCode());
    result = PRIME * result + ((name == null) ? 0 : name.hashCode());
    return result;
  }
    
  public String toString() {
    StringBuilder sb = new StringBuilder();
    sb.append("(");
    sb.append(type);
    sb.append(":");
    sb.append(name);
    sb.append(":");
    sb.append(status);
    sb.append(")");
    return sb.toString();
  }
} @XmlRootElement
public enum StateType {
  PACKAGE, DAEMON;
} @XmlAccessorType(XmlAccessType.FIELD) 
@XmlType(name="", propOrder = {})
@XmlRootElement
public class ConfigManifest  extends Manifest {
  private static Log LOG = LogFactory.getLog(ConfigManifest.class);

  @XmlElement
  private List<Action> actions;
  
  public List<Action> getActions() {
    return actions;
  }
  
  public void setActions(List<Action> actions) {
    this.actions = actions;
  }

  public void expand(NodesManifest nodes) {
    List<Role> roles = nodes.getRoles();
    Pattern p = Pattern.compile("\\$\\{(.*?)\\}");
    int index = 0;
    for(Action action: actions) {
      if(action instanceof ScriptAction) {
        String[] params = ((ScriptAction) action).getParameters();
        String[] expandedParams = new String[params.length];
        int i = 0;
        for(String param : params) {
          expandedParams[i]=param;          
          Matcher m = p.matcher(param);
          while(m.find()) {
            String token = m.group(1);
            for(Role role : roles) {
              if(role.name.equals(token)) {
                String[] hosts = role.getHosts();
                String replacement = StringUtils.join(hosts, ",");
                expandedParams[i] = param.replace(m.group(0), replacement);
              }
            }
          }
          i++;
        }
        ((ScriptAction) action).setParameters(expandedParams);
        actions.set(index, action);
      }
      index++;
    }    
  }
} public class ExceptionUtil {
  public static String getStackTrace(Throwable t) {
    StringWriter sw = new StringWriter();
    PrintWriter pw = new PrintWriter(sw);
    t.printStackTrace(pw);
    pw.flush();
    return sw.toString();
  }
} public class JAXBUtil {

  private static ObjectMapper mapper = new ObjectMapper();
  private static AnnotationIntrospector introspector = new JaxbAnnotationIntrospector();
  
  public JAXBUtil() {
    mapper.getDeserializationConfig().setAnnotationIntrospector(introspector);
    mapper.getSerializationConfig().setAnnotationIntrospector(introspector);    
  }
  
  public static byte[] write(RestSource x) throws IOException {
    try {
      return mapper.writeValueAsBytes(x);
    } catch (Throwable e) {
      throw new IOException(e);
    }
  }
  
  public static <T> T read(byte[] buffer, java.lang.Class<T> c) throws IOException {
    return (T) mapper.readValue(buffer, 0, buffer.length, c);
  }

  public static String print(RestSource x) throws IOException {
    try {
      JsonFactory jf = new JsonFactory();
      StringWriter sw = new StringWriter();
      JsonGenerator jp = jf.createJsonGenerator(sw);
      jp.useDefaultPrettyPrinter();
      mapper.writeValue(jp, x);
      jp.close();
      String buffer = sw.toString(); 
      return buffer;
    } catch (Throwable e) {
      throw new IOException(e);
    }    
  }
} public class ZookeeperUtil {
  public static String COMMAND_STATUS = "/status";
  private static final Pattern BASENAME = Pattern.compile(".*?([^/]*)$");
  
  public static String getClusterPath(String clusterName) {
    StringBuilder clusterNode = new StringBuilder();
    clusterNode.append(CommonConfigurationKeys.ZOOKEEPER_CLUSTER_ROOT_DEFAULT);
    clusterNode.append("/");
    clusterNode.append(clusterName);
    return clusterNode.toString();
  }
  
  public static String getCommandStatusPath(String cmdPath) {
    StringBuilder cmdStatusPath = new StringBuilder();
    cmdStatusPath.append(cmdPath);
    cmdStatusPath.append(COMMAND_STATUS);
    return cmdStatusPath.toString();
  }
  
  public static String getBaseURL(String url) {
      Matcher matcher = BASENAME.matcher(url);
      if (matcher.matches()) {
        return matcher.group(1);
      } else {
        throw new IllegalArgumentException("Can't parse " + url);
      }
  }
  
  public static String getNodesManifestPath(String id) {
    StringBuilder nodesPath = new StringBuilder();
    nodesPath.append(CommonConfigurationKeys.ZOOKEEPER_NODES_MANIFEST_PATH_DEFAULT);
    nodesPath.append("/");
    nodesPath.append(id);
    return nodesPath.toString();
  }

  public static String getSoftwareManifestPath(String id) {
    StringBuilder nodesPath = new StringBuilder();
    nodesPath.append(CommonConfigurationKeys.ZOOKEEPER_SOFTWARE_MANIFEST_PATH_DEFAULT);
    nodesPath.append("/");
    nodesPath.append(id);
    return nodesPath.toString();
  }
  
  public static String getConfigManifestPath(String id) {
    StringBuilder configPath = new StringBuilder();
    configPath.append(CommonConfigurationKeys.ZOOKEEPER_CONFIG_BLUEPRINT_PATH_DEFAULT);
    configPath.append("/");
    configPath.append(id);
    return configPath.toString();
  }
} public class Controller implements Watcher {
  private static Log LOG = LogFactory.getLog(Controller.class);
  public static String CONTROLLER_PREFIX = "v1";
  public static int CONTROLLER_PORT = 4080;
  private static Controller instance = new Controller();
  private Server server = null;
  private String credential = null;
  
  private ZooKeeper zk;
  private ClientHandler clientHandler;
  private CommandHandler commandHandler;
  public volatile boolean running = true; // true while controller runs
  private String zookeeperAddress = CommonConfigurationKeys.ZOOKEEPER_ADDRESS_DEFAULT;

  public static Controller getInstance() {
    return instance;
  }

  public ZooKeeper getZKInstance() {
    return this.zk;
  }
  
  public ClientHandler getClientHandler() {
    return clientHandler;
  }
  
  public CommandHandler getCommandHandler() {
    return commandHandler;
  }
  
  public void process(WatchedEvent event) {
    if (event.getType() == Event.EventType.None) {
        // We are are being told that the state of the
        // connection has changed
        switch (event.getState()) {
        case SyncConnected:
            // In this particular example we don't need to do anything
            // here - watches are automatically re-registered with 
            // server and any watches triggered while the client was 
            // disconnected will be delivered (in order of course)
            break;
        case Expired:
            // It's all over
            running = false;
            commandHandler.stop();
            break;
        }
    }
  }
  
  public void parseConfig() {
    StringBuilder confPath = new StringBuilder();
    String confDir = System.getProperty("HMS_CONF_DIR");
    if(confDir==null) {
      confDir = "/etc/hms";
    }
    confPath.append(confDir);
    confPath.append("/hms.ini");
    try {
      HierarchicalINIConfiguration ini = new HierarchicalINIConfiguration(confPath.toString());
      zookeeperAddress = ini.getSection("zookeeper").getString("quorum", null);
      String user = ini.getSection("zookeeper").getString("user", null);
      String password = ini.getSection("zookeeper").getString("password", null);
      if(user!=null && password!=null) {
        credential = new StringBuilder().append(user).append(":").append(password).toString();
      }
    } catch (Exception e) {
      LOG.warn("Invalid HMS configuration file: " + confPath);
      zookeeperAddress = null;
    }
    LOG.info("ZooKeeper Quorum in "+confPath.toString()+": "+zookeeperAddress);
  }
  
  // Resolve the list of zookeeper hosts from HMS beacons
  public void initmDNS() {
    try {
      ServiceDiscoveryUtil sdu = new ServiceDiscoveryUtil(CommonConfigurationKeys.ZEROCONF_ZOOKEEPER_TYPE);
      sdu.start();
      Thread.sleep(5000);
      Collection<String> list = sdu.resolve();
      if(list.size()>0) {
        StringBuffer buf = new StringBuffer();
        String delimiter = "";
        for(String addr : list) {
          buf.append(delimiter);
          buf.append(addr);
          delimiter = ",";
        }
        zookeeperAddress = buf.toString();
      }
      sdu.close();
      if(zookeeperAddress.equals("")) {
        throw new RuntimeException("Unknown ZooKeeper location.");
      }
      LOG.info("Discovered zookeeper location: "+zookeeperAddress);
    } catch(Exception e) {
      zookeeperAddress = CommonConfigurationKeys.ZOOKEEPER_ADDRESS_DEFAULT;
      LOG.info("Use default zookeeper location: "+zookeeperAddress);
    }
  }
  
  public void start() {
    try {
      //System.out.close();
      //System.err.close();
      parseConfig();
      if(zookeeperAddress == null) {
        initmDNS();
      }
      run();
    } catch(Exception e) {
      LOG.error(ExceptionUtil.getStackTrace(e));
      System.exit(-1);
    }
  }

  private void createDirectory(String path) throws KeeperException, InterruptedException {
    try {
      zk.create(path, new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
      if(credential!=null) {
        zk.setACL(path, Ids.CREATOR_ALL_ACL, -1);
      }
      LOG.info("Created HMS cluster root at " + CommonConfigurationKeys.ZOOKEEPER_CLUSTER_ROOT_DEFAULT);        
    } catch (KeeperException.NodeExistsException e) {
    } catch (KeeperException.AuthFailedException e) {
      LOG.warn("Failed to authenticate for "+path);
    }
  }
  
  private void initializeZooKeeper() throws KeeperException, InterruptedException, IOException {
    zk = new ZooKeeper(zookeeperAddress, 600000, this);
    if(credential!=null) {
      zk.addAuthInfo("digest", credential.getBytes());
    }
    String[] list = {
        CommonConfigurationKeys.ZOOKEEPER_CLUSTER_ROOT_DEFAULT,
        CommonConfigurationKeys.ZOOKEEPER_COMMAND_QUEUE_PATH_DEFAULT,
        CommonConfigurationKeys.ZOOKEEPER_LOCK_QUEUE_PATH_DEFAULT,
        CommonConfigurationKeys.ZOOKEEPER_LIVE_CONTROLLER_PATH_DEFAULT,
        CommonConfigurationKeys.ZOOKEEPER_NODES_MANIFEST_PATH_DEFAULT,
        CommonConfigurationKeys.ZOOKEEPER_STATUS_QUEUE_PATH_DEFAULT,
        CommonConfigurationKeys.ZOOKEEPER_SOFTWARE_MANIFEST_PATH_DEFAULT,
        CommonConfigurationKeys.ZOOKEEPER_CONFIG_BLUEPRINT_PATH_DEFAULT
    };
    for(String path : list) {
      createDirectory(path);
    }
    
    // Populate default configuration
    zk.create(CommonConfigurationKeys.ZOOKEEPER_CONFIG_BLUEPRINT_PATH_DEFAULT+"/delete-hadoop-1.0-cluster", JAXBUtil.write(Examples.deleteHadoopCluster), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    zk.create(CommonConfigurationKeys.ZOOKEEPER_CONFIG_BLUEPRINT_PATH_DEFAULT+"/create-hadoop-1.0-cluster", JAXBUtil.write(Examples.createHadoopCluster), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
  }
  
  public void run() {
    try {
      initializeZooKeeper();
      LOG.info("Connected to ZooKeeper");
      clientHandler = new ClientHandler(zk);
      commandHandler = new CommandHandler(zk, 200);
      commandHandler.start();
    } catch (Exception e) {
      LOG.error(ExceptionUtil.getStackTrace(e));
    }
    server = new Server(CONTROLLER_PORT);

    try {
      Context root = new Context(server, "/", Context.SESSIONS);
      String HMS_HOME = System.getenv("HMS_HOME");
      root.setBaseResource(new ResourceCollection(new Resource[]
        {
          Resource.newResource(HMS_HOME+"/webapps/")
        }));
      ServletHolder rootServlet = root.addServlet(DefaultServlet.class, "/");
      rootServlet.setInitOrder(1);
      
      ServletHolder sh = new ServletHolder(ServletContainer.class);
      sh.setInitParameter("com.sun.jersey.config.property.resourceConfigClass", "com.sun.jersey.api.core.PackagesResourceConfig");
      sh.setInitParameter("com.sun.jersey.config.property.packages", "org.apache.hms.controller.rest");      
      root.addServlet(sh, "/"+CONTROLLER_PREFIX+"/*");
      sh.setInitOrder(2);
      server.setStopAtShutdown(true);
      server.start();
    } catch (Exception e) {
      LOG.error(ExceptionUtil.getStackTrace(e));
    }
  }
  
  public void stop() throws Exception {
    try {
      commandHandler.stop();
      server.stop();
    } catch (Exception e) {
      LOG.error(ExceptionUtil.getStackTrace(e));
    }
  }
  
  /**
   * Wait for service to finish.
   * (Normally, it runs forever.)
   */
  public void join() {
    try {
      this.commandHandler.join();
    } catch (InterruptedException ie) {
    }
  }

  public static void main(String[] args) {
    DaemonWatcher.createInstance(System.getProperty("PID"), 9100);
    try {
      Controller controller = Controller.getInstance();
      if (controller != null) {
        controller.start();
        controller.join();
      }
    } catch(Throwable t) {
      DaemonWatcher.bailout(1);
    }
  }

}
import java.util.ArrayList;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import javax.ws.rs.DELETE;
import javax.ws.rs.GET;
import javax.ws.rs.POST;
import javax.ws.rs.PUT;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
import javax.ws.rs.WebApplicationException;
import javax.ws.rs.core.Context;
import javax.ws.rs.core.Response;
import javax.ws.rs.core.UriInfo;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hms.common.conf.CommonConfigurationKeys;
import org.apache.hms.common.entity.Status;
import org.apache.hms.common.entity.action.Action;
import org.apache.hms.common.entity.action.ActionDependency;
import org.apache.hms.common.entity.action.DaemonAction;
import org.apache.hms.common.entity.action.PackageAction;
import org.apache.hms.common.entity.action.ScriptAction;
import org.apache.hms.common.entity.cluster.MachineState.StateEntry;
import org.apache.hms.common.entity.cluster.MachineState.StateType;
import org.apache.hms.common.entity.manifest.ConfigManifest;
import org.apache.hms.common.util.ExceptionUtil;
import org.apache.hms.common.util.JAXBUtil;
import org.apache.hms.common.util.ZookeeperUtil;
import org.apache.hms.controller.Controller;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.data.Stat;
/*
 * 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.hms.controller.rest;





@Path("config")
public class ConfigManager {
  private static Log LOG = LogFactory.getLog(ConfigManager.class);
  
  @GET
  @Path("manifest/create-hadoop-cluster")
  public ConfigManifest getSample(@Context UriInfo uri) {
    return Examples.createHadoopCluster;
  }
  

  @GET
  @Path("manifest/delete-cluster")
  public ConfigManifest getDestroyCluster() {    
    return Examples.deleteHadoopCluster;
  }

  @GET
  @Path("manifest/delete-hadoop-cluster")
  public ConfigManifest getDeleteCluster() {
    List<StateEntry> states = new LinkedList<StateEntry>();
    Set<String> roles = new HashSet<String>();
    List<Action> actions = new ArrayList<Action>();
    // Stop Tasktrackers
    DaemonAction taskTrackerAction = new DaemonAction();
    taskTrackerAction.setDaemonName("hadoop-tasktracker");
    taskTrackerAction.setActionType("stop");
    taskTrackerAction.setRole("tasktracker");
    // Setup expected result
    List<StateEntry> expectedTasktrackerResults = new LinkedList<StateEntry>();
    expectedTasktrackerResults.add(new StateEntry(StateType.DAEMON, "hadoop-tasktracker", Status.STOPPED));
    taskTrackerAction.setExpectedResults(expectedTasktrackerResults);
    actions.add(taskTrackerAction);

    // Stop Jobtracker
    DaemonAction jobTrackerAction = new DaemonAction();
    jobTrackerAction.setDaemonName("hadoop-jobtracker");
    jobTrackerAction.setActionType("stop");
    jobTrackerAction.setRole("jobtracker");
    // Setup tasktracker stop dependencies
    List<ActionDependency> dep = new LinkedList<ActionDependency>();
    roles = new HashSet<String>();
    states = new LinkedList<StateEntry>();
    states.add(new StateEntry(StateType.DAEMON, "hadoop-jobtracker", Status.STARTED));
    roles.add("jobtracker");
    dep.add(new ActionDependency(roles, states));
    jobTrackerAction.setDependencies(dep);
    // Setup expected result
    List<StateEntry> expectedJobtrackerResults = new LinkedList<StateEntry>();
    expectedJobtrackerResults.add(new StateEntry(StateType.DAEMON, "hadoop-jobtracker", Status.STOPPED));
    jobTrackerAction.setExpectedResults(expectedJobtrackerResults);
    actions.add(jobTrackerAction);

    // Stop Datanodes
    DaemonAction datanodeAction = new DaemonAction();
    datanodeAction.setDaemonName("hadoop-datanode");
    datanodeAction.setActionType("stop");
    datanodeAction.setRole("datanode");
    // Setup datanode stop dependencies
    dep = new LinkedList<ActionDependency>();
    roles = new HashSet<String>();
    states = new LinkedList<StateEntry>();
    states.add(new StateEntry(StateType.DAEMON, "hadoop-datanode", Status.STARTED));
    roles.add("datanode");
    dep.add(new ActionDependency(roles, states));
    datanodeAction.setDependencies(dep);
    // Setup expected result
    List<StateEntry> expectedDatanodeResults = new LinkedList<StateEntry>();
    expectedDatanodeResults.add(new StateEntry(StateType.DAEMON, "hadoop-datanode", Status.STOPPED));
    datanodeAction.setExpectedResults(expectedDatanodeResults);
    actions.add(datanodeAction);

    // Stop Namenode
    DaemonAction namenodeAction = new DaemonAction();
    namenodeAction.setDaemonName("hadoop-namenode");
    namenodeAction.setActionType("stop");
    namenodeAction.setRole("namenode");
    // Setup namenode stop dependencies
    dep = new LinkedList<ActionDependency>();
    roles = new HashSet<String>();
    states = new LinkedList<StateEntry>();
    states.add(new StateEntry(StateType.DAEMON, "hadoop-namenode", Status.STARTED));
    roles.add("namenode");
    dep.add(new ActionDependency(roles, states));
    namenodeAction.setDependencies(dep);
    // Setup expected result
    List<StateEntry> expectedNamenodeResults = new LinkedList<StateEntry>();
    expectedNamenodeResults.add(new StateEntry(StateType.DAEMON, "hadoop-namenode", Status.STOPPED));
    namenodeAction.setExpectedResults(expectedNamenodeResults);
    actions.add(namenodeAction);
    ConfigManifest cm = new ConfigManifest();
    cm.setActions(actions);
    return cm;
  }
  
  @POST
  @Path("blueprint/{name}")
  public Response createBlueprint(@Context UriInfo uri, 
      @PathParam("name") String name, ConfigManifest blueprint) {
    Response res;
    try {
      ZooKeeper zk = Controller.getInstance().getZKInstance();
      blueprint.setUrl(uri.getAbsolutePath().toURL());
      byte[] data = JAXBUtil.write(blueprint);
      Stat stat = zk.exists(CommonConfigurationKeys.ZOOKEEPER_CONFIG_BLUEPRINT_PATH_DEFAULT+'/'+name, false);
      if(stat==null) {
        zk.create(CommonConfigurationKeys.ZOOKEEPER_CONFIG_BLUEPRINT_PATH_DEFAULT+'/'+name, data, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
      } else {
        throw new WebApplicationException(409);
      }
      res = Response.created(uri.getAbsolutePath()).build();
      return res;
    } catch(WebApplicationException e) {
      throw e;
    } catch(Exception e) {
      LOG.error(ExceptionUtil.getStackTrace(e));
      throw new WebApplicationException(500);
    }
  }
  
  @PUT
  @Path("blueprint/{name}")
  public Response updateBlueprint(@Context UriInfo uri, 
      @PathParam("name") String oldName, ConfigManifest blueprint) {
    Response res;
    try {
      ZooKeeper zk = Controller.getInstance().getZKInstance();
      String newName = oldName;
      if(blueprint.getUrl()!=null) {
        newName = ZookeeperUtil.getBaseURL(blueprint.getUrl().toString());