text:
package org.apache.asterix.aoya;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileFilter;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Vector;
import java.util.regex.Pattern;

import javax.xml.bind.JAXBContext;
import javax.xml.bind.JAXBException;
import javax.xml.bind.Marshaller;

import org.apache.asterix.common.configuration.AsterixConfiguration;
import org.apache.asterix.common.configuration.Coredump;
import org.apache.asterix.common.configuration.Store;
import org.apache.asterix.common.configuration.TransactionLogDir;
import org.apache.asterix.event.schema.yarnCluster.Cluster;
import org.apache.asterix.event.schema.yarnCluster.Node;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.GnuParser;
import org.apache.commons.cli.HelpFormatter;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.ParseException;
import org.apache.commons.io.filefilter.WildcardFileFilter;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.service.Service.STATE;
import org.apache.hadoop.yarn.api.ApplicationConstants;
import org.apache.hadoop.yarn.api.protocolrecords.GetNewApplicationResponse;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ApplicationReport;
import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext;
import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
import org.apache.hadoop.yarn.api.records.LocalResource;
import org.apache.hadoop.yarn.api.records.LocalResourceType;
import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.YarnApplicationState;
import org.apache.hadoop.yarn.client.api.YarnClient;
import org.apache.hadoop.yarn.client.api.YarnClientApplication;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.ApplicationNotFoundException;
import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.hadoop.yarn.util.ConverterUtils;
import org.apache.hadoop.yarn.util.Records;

import com.google.common.collect.ImmutableMap;

@InterfaceAudience.Public
@InterfaceStability.Unstable
public class AsterixYARNClient {

    public static enum Mode {
        INSTALL("install"),
        START("start"),
        STOP("stop"),
        KILL("kill"),
        DESTROY("destroy"),
        ALTER("alter"),
        LIBINSTALL("libinstall"),
        DESCRIBE("describe"),
        BACKUP("backup"),
        LSBACKUP("lsbackups"),
        RMBACKUP("rmbackup"),
        RESTORE("restore"),
        NOOP("");

        public final String alias;

        Mode(String alias) {
            this.alias = alias;
        }

        public static Mode fromAlias(String a) {
            return STRING_TO_MODE.get(a.toLowerCase());
        }
    }

    public static final Map<String, AsterixYARNClient.Mode> STRING_TO_MODE = ImmutableMap.<String, AsterixYARNClient
            .Mode> builder().put(Mode.INSTALL.alias, Mode.INSTALL).put(Mode.START.alias, Mode.START)
            .put(Mode.STOP.alias, Mode.STOP).put(Mode.KILL.alias, Mode.KILL).put(Mode.DESTROY.alias, Mode.DESTROY)
            .put(Mode.ALTER.alias, Mode.ALTER).put(Mode.LIBINSTALL.alias, Mode.LIBINSTALL)
            .put(Mode.DESCRIBE.alias, Mode.DESCRIBE).put(Mode.BACKUP.alias, Mode.BACKUP)
            .put(Mode.LSBACKUP.alias, Mode.LSBACKUP).put(Mode.RMBACKUP.alias, Mode.RMBACKUP)
            .put(Mode.RESTORE.alias, Mode.RESTORE).build();
    private static final Log LOG = LogFactory.getLog(AsterixYARNClient.class);
    public static final String CONF_DIR_REL = ".asterix" + File.separator;
    private static final String instanceLock = "instance";
    public static final String CONFIG_DEFAULT_NAME = "cluster-config.xml";
    public static final String PARAMS_DEFAULT_NAME = "asterix-configuration.xml";
    private static String DEFAULT_PARAMETERS_PATH = "conf" + File.separator + "base-asterix-configuration.xml";
    private static String MERGED_PARAMETERS_PATH = "conf" + File.separator + PARAMS_DEFAULT_NAME;
    private static final String JAVA_HOME = System.getProperty("java.home");
    public static final String NC_JAVA_OPTS_KEY = "nc.java.opts";
    public static final String CC_JAVA_OPTS_KEY = "cc.java.opts";
    public static final String CC_REST_PORT_KEY = "api.port";
    private Mode mode = Mode.NOOP;

    // Hadoop Configuration
    private Configuration conf;
    private YarnClient yarnClient;
    // Application master specific info to register a new Application with
    // RM/ASM
    private String appName = "";
    // App master priority
    private int amPriority = 0;
    // Queue for App master
    private String amQueue = "";
    // Amt. of memory resource to request for to run the App Master
    private int amMemory = 1000;

    // Main class to invoke application master
    private final String appMasterMainClass = "org.apache.asterix.aoya.AsterixApplicationMaster";

    //instance name
    private String instanceName = "";
    //location of distributable AsterixDB zip
    private String asterixZip = "";
    // Location of cluster configuration
    private String asterixConf = "";
    // Location of optional external libraries
    private String extLibs = "";

    private String instanceFolder = "";

    // log4j.properties file
    // if available, add to local resources and set into classpath
    private String log4jPropFile = "";

    // Debug flag
    boolean debugFlag = false;
    private boolean refresh = false;
    private boolean force = false;

    // Command line options
    private Options opts;
    private String libDataverse;
    private String snapName = "";
    private String baseConfig = ".";
    private String ccJavaOpts = "";
    private String ncJavaOpts = "";

    //Ports
    private int ccRestPort = 19002;

    /**
     * @param args
     *            Command line arguments
     */
    public static void main(String[] args) {

        try {
            AsterixYARNClient client = new AsterixYARNClient();
            try {
                client.init(args);
                AsterixYARNClient.execute(client);
            } catch (ParseException | ApplicationNotFoundException e) {
                LOG.fatal(e);
                client.printUsage();
                System.exit(-1);
            }
        } catch (Exception e) {
            LOG.fatal("Error running client", e);
            System.exit(1);
        }
        LOG.info("Command executed successfully.");
        System.exit(0);
    }

    public static void execute(AsterixYARNClient client) throws IOException, YarnException {
        YarnClientApplication app;
        List<DFSResourceCoordinate> res;

        System.out.println("JAVA HOME: " + JAVA_HOME);
        switch (client.mode) {
            case START:
                startAction(client);
                break;
            case STOP:
                try {
                    client.stopInstance();
                } catch (ApplicationNotFoundException e) {
                    LOG.info(e);
                    System.out.println("Asterix instance by that name already exited or was never started");
                    client.deleteLockFile();
                }
                break;
            case KILL:
                if (client.isRunning() && Utils.confirmAction(
                        "Are you sure you want to kill this instance? In-progress tasks will be aborted")) {
                    try {
                        AsterixYARNClient.killApplication(client.getLockFile(), client.yarnClient);
                    } catch (ApplicationNotFoundException e) {
                        LOG.info(e);
                        System.out.println("Asterix instance by that name already exited or was never started");
                        client.deleteLockFile();
                    }
                } else if (!client.isRunning()) {
                    System.out.println("Asterix instance by that name already exited or was never started");
                    client.deleteLockFile();
                }
                break;
            case DESCRIBE:
                Utils.listInstances(client.conf, CONF_DIR_REL);
                break;
            case INSTALL:
                installAction(client);
                break;
            case LIBINSTALL:
                client.installExtLibs();
                break;
            case ALTER:
                client.writeAsterixConfig(Utils.parseYarnClusterConfig(client.asterixConf));
                client.installAsterixConfig(true);
                System.out.println("Configuration successfully modified");
                break;
            case DESTROY:
                try {
                    if (client.force || Utils.confirmAction(
                            "Are you really sure you want to obliterate this instance? This action cannot be undone!")) {
                        app = client.makeApplicationContext();
                        res = client.deployConfig();
                        res.addAll(client.distributeBinaries());
                        client.removeInstance(app, res);
                    }
                } catch (YarnException | IOException e) {
                    LOG.error("Asterix failed to deploy on to cluster");
                    throw e;
                }
                break;
            case BACKUP:
                if (client.force || Utils.confirmAction("Performing a backup will stop a running instance.")) {
                    app = client.makeApplicationContext();
                    res = client.deployConfig();
                    res.addAll(client.distributeBinaries());
                    client.backupInstance(app, res);
                }
                break;
            case LSBACKUP:
                Utils.listBackups(client.conf, CONF_DIR_REL, client.instanceName);
                break;
            case RMBACKUP:
                Utils.rmBackup(client.conf, CONF_DIR_REL, client.instanceName, Long.parseLong(client.snapName));
                break;
            case RESTORE:
                if (client.force || Utils.confirmAction("Performing a restore will stop a running instance.")) {
                    app = client.makeApplicationContext();
                    res = client.deployConfig();
                    res.addAll(client.distributeBinaries());
                    client.restoreInstance(app, res);
                }
                break;
            default:
                LOG.fatal(
                        "Unknown mode. Known client modes are: start, stop, install, describe, kill, destroy, describe, backup, restore, lsbackup, rmbackup");
                client.printUsage();
                System.exit(-1);
        }
    }

    private static void startAction(AsterixYARNClient client) throws YarnException {
        YarnClientApplication app;
        List<DFSResourceCoordinate> res;
        ApplicationId appId;
        try {
            app = client.makeApplicationContext();
            res = client.deployConfig();
            res.addAll(client.distributeBinaries());
            appId = client.deployAM(app, res, client.mode);
            LOG.info("Asterix started up with Application ID: " + appId.toString());
            if (Utils.waitForLiveness(appId, "Waiting for AsterixDB instance to resume ", client.yarnClient,
                    client.instanceName, client.conf, client.ccRestPort)) {
                System.out.println("Asterix successfully deployed and is now running.");
            } else {
                LOG.fatal("AsterixDB appears to have failed to install and start");
                throw new YarnException("AsterixDB appears to have failed to install and start");
            }
        } catch (IOException e) {
            throw new YarnException(e);
        }
    }

    private static void installAction(AsterixYARNClient client) throws YarnException {
        YarnClientApplication app;
        List<DFSResourceCoordinate> res;
        ApplicationId appId;
        try {
            app = client.makeApplicationContext();
            client.installConfig();
            client.writeAsterixConfig(Utils.parseYarnClusterConfig(client.asterixConf));
            client.installAsterixConfig(false);
            res = client.deployConfig();
            res.addAll(client.distributeBinaries());

            appId = client.deployAM(app, res, client.mode);
            LOG.info("Asterix started up with Application ID: " + appId.toString());
            if (Utils.waitForLiveness(appId, "Waiting for new AsterixDB Instance to start ", client.yarnClient,
                    client.instanceName, client.conf, client.ccRestPort)) {
                System.out.println("Asterix successfully deployed and is now running.");
            } else {
                LOG.fatal("AsterixDB appears to have failed to install and start");
                throw new YarnException("AsterixDB appears to have failed to install and start");
            }
        } catch (IOException e) {
            LOG.fatal("Asterix failed to deploy on to cluster");
            throw new YarnException(e);
        }
    }

    public AsterixYARNClient(Configuration conf) throws Exception {

        this.conf = conf;
        yarnClient = YarnClient.createYarnClient();
        //If the HDFS jars aren't on the classpath this won't be set
        if (conf.get("fs.hdfs.impl", null) == conf.get("fs.file.impl", null)) { //only would happen if both are null
            conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
            conf.set("fs.file.impl", "org.apache.hadoop.fs.LocalFileSystem");
        }
        yarnClient.init(conf);
        opts = parseConf(conf);
    }

    private static Options parseConf(Configuration conf) {
        Options opts = new Options();
        opts.addOption(new Option("appname", true, "Application Name. Default value - Asterix"));
        opts.addOption(new Option("priority", true, "Application Priority. Default 0"));
        opts.addOption(new Option("queue", true, "RM Queue in which this application is to be submitted"));
        opts.addOption(new Option("master_memory", true,
                "Amount of memory in MB to be requested to run the application master"));
        opts.addOption(new Option("log_properties", true, "log4j.properties file"));
        opts.addOption(new Option("n", "name", true, "Asterix instance name (required)"));
        opts.addOption(
                new Option("zip", "asterixZip", true, "zip file with AsterixDB inside- if in non-default location"));
        opts.addOption(new Option("bc", "baseConfig", true,
                "base Asterix parameters configuration file if not in default position"));
        opts.addOption(new Option("c", "asterixConf", true, "Asterix cluster config (required on install)"));
        opts.addOption(new Option("l", "externalLibs", true, "Libraries to deploy along with Asterix instance"));
        opts.addOption(new Option("ld", "libDataverse", true, "Dataverse to deploy external libraries to"));
        opts.addOption(new Option("r", "refresh", false,
                "If starting an existing instance, this will replace them with the local copy on startup"));
        opts.addOption(new Option("appId", true, "ApplicationID to monitor if running client in status monitor mode"));
        opts.addOption(new Option("masterLibsDir", true, "Directory that contains the JARs needed to run the AM"));
        opts.addOption(new Option("s", "snapshot", true,
                "Backup timestamp for arguments requiring a specific backup (rm, restore)"));
        opts.addOption(new Option("v", "debug", false, "Dump out debug information"));
        opts.addOption(new Option("help", false, "Print usage"));
        opts.addOption(
                new Option("f", "force", false, "Execute this command as fully as possible, disregarding any caution"));
        return opts;
    }

    /**
    */
    public AsterixYARNClient() throws Exception {
        this(new YarnConfiguration());
    }

    /**
     * Helper function to print out usage
     */
    private void printUsage() {
        new HelpFormatter().printHelp("Asterix YARN client. Usage: asterix [options] [mode]", opts);
    }

    /**
     * Initialize the client's arguments and parameters before execution.
     *
     * @param args
     *            - Standard command-line arguments.
     * @throws ParseException
     */
    public void init(String[] args) throws ParseException {

        CommandLine cliParser = new GnuParser().parse(opts, args);
        if (cliParser.hasOption("help")) {
            printUsage();
            return;
        }
        //initialize most things
        debugFlag = cliParser.hasOption("debug");
        force = cliParser.hasOption("force");
        baseConfig = cliParser.getOptionValue("baseConfig");
        extLibs = cliParser.getOptionValue("externalLibs");
        libDataverse = cliParser.getOptionValue("libDataverse");

        appName = cliParser.getOptionValue("appname", "AsterixDB");
        amPriority = Integer.parseInt(cliParser.getOptionValue("priority", "0"));
        amQueue = cliParser.getOptionValue("queue", "default");
        amMemory = Integer.parseInt(cliParser.getOptionValue("master_memory", "10"));

        instanceName = cliParser.getOptionValue("name");
        instanceFolder = instanceName + '/';
        appName = appName + ": " + instanceName;

        asterixConf = cliParser.getOptionValue("asterixConf");

        log4jPropFile = cliParser.getOptionValue("log_properties", "");

        //see if the given argument values are sane in general
        checkConfSanity(args, cliParser);

        //intialize the mode, see if it is a valid one.
        initMode(args, cliParser);

        //now check the validity of the arguments given the mode
        checkModeSanity(args, cliParser);

        //if we are going to refresh the binaries, find that out
        refresh = cliParser.hasOption("refresh");
        //same goes for snapshot restoration/removal
        snapName = cliParser.getOptionValue("snapshot");

        if (!cliParser.hasOption("asterixZip")
                && (mode == Mode.INSTALL || mode == Mode.ALTER || mode == Mode.DESTROY || mode == Mode.BACKUP)) {

            asterixZip = cliParser.getOptionValue("asterixZip", getAsterixDistributableLocation().getAbsolutePath());
        } else {
            asterixZip = cliParser.getOptionValue("asterixZip");
        }

    }

    /**
     * Cursory sanity checks for argument sanity, without considering the mode of the client
     *
     * @param args
     * @param cliParser
     *            The parsed arguments.
     * @throws ParseException
     */
    private void checkConfSanity(String[] args, CommandLine cliParser) throws ParseException {
        String message = null;

        //Sanity check for no args
        if (args.length == 0) {
            message = "No args specified for client to initialize";
        }
        //AM memory should be a sane value
        else if (amMemory < 0) {
            message = "Invalid memory specified for application master, exiting." + " Specified memory=" + amMemory;
        }
        //we're good!
        else {
            return;
        }
        //default:
        throw new ParseException(message);

    }

    /**
     * Initialize the mode of the client from the arguments.
     *
     * @param args
     * @param cliParser
     * @throws ParseException
     */
    private void initMode(String[] args, CommandLine cliParser) throws ParseException {
        @SuppressWarnings("unchecked")
        List<String> clientVerb = cliParser.getArgList();
        String message = null;
        //Now check if there is a mode
        if (clientVerb == null || clientVerb.size() < 1) {
            message = "You must specify an action.";
        }
        //But there can only be one mode...
        else if (clientVerb.size() > 1) {
            message = "Trailing arguments, or too many arguments. Only one action may be performed at a time.";
        }
        if (message != null) {
            throw new ParseException(message);
        }
        //Now we can initialize the mode and check it against parameters
        mode = Mode.fromAlias(clientVerb.get(0));
        if (mode == null) {
            mode = Mode.NOOP;
        }
    }

    /**
     * Determine if the command line arguments are sufficient for the requested client mode.
     *
     * @param args
     *            The command line arguments.
     * @param cliParser
     *            Parsed command line arguments.
     * @throws ParseException
     */

    private void checkModeSanity(String[] args, CommandLine cliParser) throws ParseException {

        String message = null;
        //The only time you can use the client without specifiying an instance, is to list all of the instances it sees.
        if (!cliParser.hasOption("name") && mode != Mode.DESCRIBE) {
            message = "You must give a name for the instance to be acted upon";
        } else if (mode == Mode.INSTALL && !cliParser.hasOption("asterixConf")) {
            message = "No Configuration XML given. Please specify a config for cluster installation";
        } else if (mode != Mode.START && cliParser.hasOption("refresh")) {
            message = "Cannot specify refresh in any mode besides start, mode is: " + mode;
        } else if (cliParser.hasOption("snapshot") && !(mode == Mode.RESTORE || mode == Mode.RMBACKUP)) {
            message = "Cannot specify a snapshot to restore in any mode besides restore or rmbackup, mode is: " + mode;
        } else if ((mode == Mode.ALTER || mode == Mode.INSTALL) && baseConfig == null
                && !(new File(DEFAULT_PARAMETERS_PATH).exists())) {
            message = "Default asterix parameters file is not in the default location, and no custom location is specified";
        }
        //nothing is wrong, so exit
        else {
            return;
        }
        //otherwise, something is bad.
        throw new ParseException(message);

    }

    /**
     * Find the distributable asterix bundle, be it in the default location or in a user-specified location.
     *
     * @return
     */
    private File getAsterixDistributableLocation() {
        //Look in the PWD for the "asterix" folder
        File tarDir = new File("asterix");
        if (!tarDir.exists()) {
            throw new IllegalArgumentException(
                    "Default directory structure not in use- please specify an asterix zip and base config file to distribute");
        }
        FileFilter tarFilter = new WildcardFileFilter("asterix-server*.zip");
        File[] tarFiles = tarDir.listFiles(tarFilter);
        if (tarFiles.length != 1) {
            throw new IllegalArgumentException(
                    "There is more than one canonically named asterix distributable in the default directory. Please leave only one there.");
        }
        return tarFiles[0];
    }

    /**
     * Initialize and register the application attempt with the YARN ResourceManager.
     *
     * @return
     * @throws IOException
     * @throws YarnException
     */
    public YarnClientApplication makeApplicationContext() throws IOException, YarnException {

        //first check to see if an instance already exists.
        FileSystem fs = FileSystem.get(conf);
        Path lock = new Path(fs.getHomeDirectory(), CONF_DIR_REL + instanceFolder + instanceLock);
        LOG.info("Running Deployment");
        yarnClient.start();
        if (fs.exists(lock)) {
            ApplicationId lockAppId = getLockFile();
            try {
                ApplicationReport previousAppReport = yarnClient.getApplicationReport(lockAppId);
                YarnApplicationState prevStatus = previousAppReport.getYarnApplicationState();
                if (!(prevStatus == YarnApplicationState.FAILED || prevStatus == YarnApplicationState.KILLED
                        || prevStatus == YarnApplicationState.FINISHED) && mode != Mode.DESTROY && mode != Mode.BACKUP
                        && mode != Mode.RESTORE) {
                    throw new IllegalStateException("Instance is already running in: " + lockAppId);
                } else if (mode != Mode.DESTROY && mode != Mode.BACKUP && mode != Mode.RESTORE) {
                    //stale lock file
                    LOG.warn("Stale lockfile detected. Instance attempt " + lockAppId + " may have exited abnormally");
                    deleteLockFile();
                }
            } catch (YarnException e) {
                LOG.warn(
                        "Stale lockfile detected, but the RM has no record of this application's last run. This is normal if the cluster was restarted.");
                deleteLockFile();
            }
        }

        // Get a new application id
        YarnClientApplication app = yarnClient.createApplication();
        GetNewApplicationResponse appResponse = app.getNewApplicationResponse();
        int maxMem = appResponse.getMaximumResourceCapability().getMemory();
        LOG.info("Max mem capabililty of resources in this cluster " + maxMem);

        // A resource ask cannot exceed the max.
        if (amMemory > maxMem) {
            LOG.info("AM memory specified above max threshold of cluster. Using max value." + ", specified=" + amMemory
                    + ", max=" + maxMem);
            amMemory = maxMem;
        }

        // set the application name
        ApplicationSubmissionContext appContext = app.getApplicationSubmissionContext();
        appContext.setApplicationName(appName);

        return app;
    }

    /**
     * Upload the Asterix cluster description on to the DFS. This will persist the state of the instance.
     *
     * @return
     * @throws YarnException
     * @throws IOException
     */
    private List<DFSResourceCoordinate> deployConfig() throws YarnException, IOException {

        FileSystem fs = FileSystem.get(conf);
        List<DFSResourceCoordinate> resources = new ArrayList<DFSResourceCoordinate>(2);

        String pathSuffix = CONF_DIR_REL + instanceFolder + CONFIG_DEFAULT_NAME;
        Path dstConf = new Path(fs.getHomeDirectory(), pathSuffix);
        FileStatus destStatus;
        try {
            destStatus = fs.getFileStatus(dstConf);
        } catch (IOException e) {
            throw new YarnException("Asterix instance by that name does not appear to exist in DFS");
        }
        LocalResource asterixConfLoc = Records.newRecord(LocalResource.class);
        asterixConfLoc.setType(LocalResourceType.FILE);
        asterixConfLoc.setVisibility(LocalResourceVisibility.PRIVATE);
        asterixConfLoc.setResource(ConverterUtils.getYarnUrlFromPath(dstConf));
        asterixConfLoc.setTimestamp(destStatus.getModificationTime());

        DFSResourceCoordinate conf = new DFSResourceCoordinate();
        conf.envs.put(dstConf.toUri().toString(), AConstants.CONFLOCATION);
        conf.envs.put(Long.toString(asterixConfLoc.getSize()), AConstants.CONFLEN);
        conf.envs.put(Long.toString(asterixConfLoc.getTimestamp()), AConstants.CONFTIMESTAMP);
        conf.name = CONFIG_DEFAULT_NAME;
        conf.res = asterixConfLoc;
        resources.add(conf);

        return resources;

    }

    /**
     * Install the current Asterix parameters to the DFS. This can be modified via alter.
     *
     * @throws YarnException
     * @throws IOException
     */
    private void installConfig() throws YarnException, IOException {
        FileSystem fs = FileSystem.get(conf);
        String pathSuffix = CONF_DIR_REL + instanceFolder + CONFIG_DEFAULT_NAME;
        Path dstConf = new Path(fs.getHomeDirectory(), pathSuffix);
        try {
            fs.getFileStatus(dstConf);
            if (mode == Mode.INSTALL) {
                throw new IllegalStateException("Instance with this name already exists.");
            }
        } catch (FileNotFoundException e) {
            if (mode == Mode.START) {
                throw new IllegalStateException("Instance does not exist for this user", e);
            }
        }
        if (mode == Mode.INSTALL) {
            Path src = new Path(asterixConf);
            fs.copyFromLocalFile(false, true, src, dstConf);
        }

    }

    /**
     * Upload External libraries and functions to HDFS for an instance to use when started
     *
     * @throws IllegalStateException
     * @throws IOException
     */

    private void installExtLibs() throws IllegalStateException, IOException {
        FileSystem fs = FileSystem.get(conf);
        if (!instanceExists()) {
            throw new IllegalStateException("No instance by name " + instanceName + " found.");
        }
        if (isRunning()) {
            throw new IllegalStateException(
                    "Instance " + instanceName + " is running. Please stop it before installing any libraries.");
        }
        String libPathSuffix = CONF_DIR_REL + instanceFolder + "library" + Path.SEPARATOR + libDataverse
                + Path.SEPARATOR;
        Path src = new Path(extLibs);
        String fullLibPath = libPathSuffix + src.getName();
        Path libFilePath = new Path(fs.getHomeDirectory(), fullLibPath);
        LOG.info("Copying Asterix external library to DFS");
        fs.copyFromLocalFile(false, true, src, libFilePath);
    }

    /**
     * Finds the minimal classes and JARs needed to start the AM only.
     *
     * @return Resources the AM needs to start on the initial container.
     * @throws IllegalStateException
     * @throws IOException
     */
    private List<DFSResourceCoordinate> installAmLibs() throws IllegalStateException, IOException {
        List<DFSResourceCoordinate> resources = new ArrayList<DFSResourceCoordinate>(2);
        FileSystem fs = FileSystem.get(conf);
        String fullLibPath = CONF_DIR_REL + instanceFolder + "am_jars" + Path.SEPARATOR;
        String[] cp = System.getProperty("java.class.path").split(System.getProperty("path.separator"));
        String asterixJarPattern = "^(asterix).*(jar)$"; //starts with asterix,ends with jar
        String commonsJarPattern = "^(commons).*(jar)$";
        String surefireJarPattern = "^(surefire).*(jar)$"; //for maven tests
        String jUnitTestPattern = "^(asterix-yarn" + File.separator + "target)$";

        LOG.info(File.separator);
        for (String j : cp) {
            String[] pathComponents = j.split(Pattern.quote(File.separator));
            LOG.info(j);
            LOG.info(pathComponents[pathComponents.length - 1]);
            if (pathComponents[pathComponents.length - 1].matches(asterixJarPattern)
                    || pathComponents[pathComponents.length - 1].matches(commonsJarPattern)
                    || pathComponents[pathComponents.length - 1].matches(surefireJarPattern)
                    || pathComponents[pathComponents.length - 1].matches(jUnitTestPattern)) {
                LOG.info("Loading JAR/classpath: " + j);
                File f = new File(j);
                Path dst = new Path(fs.getHomeDirectory(), fullLibPath + f.getName());
                if (!fs.exists(dst) || refresh) {
                    fs.copyFromLocalFile(false, true, new Path(f.getAbsolutePath()), dst);
                }
                FileStatus dstSt = fs.getFileStatus(dst);
                LocalResource amLib = Records.newRecord(LocalResource.class);
                amLib.setType(LocalResourceType.FILE);
                amLib.setVisibility(LocalResourceVisibility.PRIVATE);
                amLib.setResource(ConverterUtils.getYarnUrlFromPath(dst));
                amLib.setTimestamp(dstSt.getModificationTime());
                amLib.setSize(dstSt.getLen());
                DFSResourceCoordinate amLibCoord = new DFSResourceCoordinate();
                amLibCoord.res = amLib;
                amLibCoord.name = f.getName();
                if (f.getName().contains("asterix-yarn") || f.getName().contains("surefire")) {
                    amLibCoord.envs.put(dst.toUri().toString(), AConstants.APPLICATIONMASTERJARLOCATION);
                    amLibCoord.envs.put(Long.toString(dstSt.getLen()), AConstants.APPLICATIONMASTERJARLEN);
                    amLibCoord.envs.put(Long.toString(dstSt.getModificationTime()),
                            AConstants.APPLICATIONMASTERJARTIMESTAMP);
                }
                resources.add(amLibCoord);
            }

        }
        if (resources.size() == 0) {
            throw new IOException("Required JARs are missing. Please check your directory structure");
        }
        return resources;
    }

    /**
     * Uploads a AsterixDB cluster configuration to HDFS for the AM to use.
     *
     * @param overwrite
     *            Overwrite existing configurations by the same name.
     * @throws IllegalStateException
     * @throws IOException
     */
    private void installAsterixConfig(boolean overwrite) throws IllegalStateException, IOException {
        FileSystem fs = FileSystem.get(conf);
        File srcfile = new File(MERGED_PARAMETERS_PATH);
        Path src = new Path(srcfile.getCanonicalPath());
        String pathSuffix = CONF_DIR_REL + instanceFolder + File.separator + PARAMS_DEFAULT_NAME;
        Path dst = new Path(fs.getHomeDirectory(), pathSuffix);
        if (fs.exists(dst) && !overwrite) {

            throw new IllegalStateException(
                    "Instance exists. Please delete an existing instance before trying to overwrite");
        }
        fs.copyFromLocalFile(false, true, src, dst);
    }

    /**
     * Uploads binary resources to HDFS for use by the AM
     *
     * @return
     * @throws IOException
     * @throws YarnException
     */
    public List<DFSResourceCoordinate> distributeBinaries() throws IOException, YarnException {

        List<DFSResourceCoordinate> resources = new ArrayList<DFSResourceCoordinate>(2);
        // Copy the application master jar to the filesystem
        // Create a local resource to point to the destination jar path
        FileSystem fs = FileSystem.get(conf);
        Path src, dst;
        FileStatus destStatus;
        String pathSuffix;

        // adding info so we can add the jar to the App master container path

        // Add the asterix tarfile to HDFS for easy distribution
        // Keep it all archived for now so add it as a file...

        pathSuffix = CONF_DIR_REL + instanceFolder + "asterix-server.zip";
        dst = new Path(fs.getHomeDirectory(), pathSuffix);
        if (refresh) {
            if (fs.exists(dst)) {
                fs.delete(dst, false);
            }
        }
        if (!fs.exists(dst)) {
            src = new Path(asterixZip);
            LOG.info("Copying Asterix distributable to DFS");
            fs.copyFromLocalFile(false, true, src, dst);
        }
        destStatus = fs.getFileStatus(dst);
        LocalResource asterixTarLoc = Records.newRecord(LocalResource.class);
        asterixTarLoc.setType(LocalResourceType.ARCHIVE);
        asterixTarLoc.setVisibility(LocalResourceVisibility.PRIVATE);
        asterixTarLoc.setResource(ConverterUtils.getYarnUrlFromPath(dst));
        asterixTarLoc.setTimestamp(destStatus.getModificationTime());

        // adding info so we can add the tarball to the App master container path
        DFSResourceCoordinate tar = new DFSResourceCoordinate();
        tar.envs.put(dst.toUri().toString(), AConstants.TARLOCATION);
        tar.envs.put(Long.toString(asterixTarLoc.getSize()), AConstants.TARLEN);
        tar.envs.put(Long.toString(asterixTarLoc.getTimestamp()), AConstants.TARTIMESTAMP);
        tar.res = asterixTarLoc;
        tar.name = "asterix-server.zip";
        resources.add(tar);

        // Set the log4j properties if needed
        if (!log4jPropFile.isEmpty()) {
            Path log4jSrc = new Path(log4jPropFile);
            Path log4jDst = new Path(fs.getHomeDirectory(), "log4j.props");
            fs.copyFromLocalFile(false, true, log4jSrc, log4jDst);
            FileStatus log4jFileStatus = fs.getFileStatus(log4jDst);
            LocalResource log4jRsrc = Records.newRecord(LocalResource.class);
            log4jRsrc.setType(LocalResourceType.FILE);
            log4jRsrc.setVisibility(LocalResourceVisibility.PRIVATE);
            log4jRsrc.setResource(ConverterUtils.getYarnUrlFromURI(log4jDst.toUri()));
            log4jRsrc.setTimestamp(log4jFileStatus.getModificationTime());
            log4jRsrc.setSize(log4jFileStatus.getLen());
            DFSResourceCoordinate l4j = new DFSResourceCoordinate();
            tar.res = log4jRsrc;
            tar.name = "log4j.properties";
            resources.add(l4j);
        }

        resources.addAll(installAmLibs());
        return resources;
    }

    /**
     * Submits the request to start the AsterixApplicationMaster to the YARN ResourceManager.
     *
     * @param app
     *            The application attempt handle.
     * @param resources
     *            Resources to be distributed as part of the container launch
     * @param mode
     *            The mode of the ApplicationMaster
     * @return The application ID of the new Asterix instance.
     * @throws IOException
     * @throws YarnException
     */

    public ApplicationId deployAM(YarnClientApplication app, List<DFSResourceCoordinate> resources, Mode mode)
            throws IOException, YarnException {

        // Set up the container launch context for the application master
        ContainerLaunchContext amContainer = Records.newRecord(ContainerLaunchContext.class);

        ApplicationSubmissionContext appContext = app.getApplicationSubmissionContext();

        // Set local resource info into app master container launch context
        Map<String, LocalResource> localResources = new HashMap<String, LocalResource>();
        for (DFSResourceCoordinate res : resources) {
            localResources.put(res.name, res.res);
        }
        amContainer.setLocalResources(localResources);
        // Set the env variables to be setup in the env where the application
        // master will be run
        LOG.info("Set the environment for the application master");
        Map<String, String> env = new HashMap<String, String>();

        // using the env info, the application master will create the correct
        // local resource for the
        // eventual containers that will be launched to execute the shell
        // scripts
        for (DFSResourceCoordinate res : resources) {
            if (res.envs == null) { //some entries may not have environment variables.
                continue;
            }
            for (Map.Entry<String, String> e : res.envs.entrySet()) {
                env.put(e.getValue(), e.getKey());
            }
        }
        //this is needed for when the RM address isn't known from the environment of the AM
        env.put(AConstants.RMADDRESS, conf.get("yarn.resourcemanager.address"));
        env.put(AConstants.RMSCHEDULERADDRESS, conf.get("yarn.resourcemanager.scheduler.address"));
        ///add miscellaneous environment variables.
        env.put(AConstants.INSTANCESTORE, CONF_DIR_REL + instanceFolder);
        env.put(AConstants.DFS_BASE, FileSystem.get(conf).getHomeDirectory().toUri().toString());
        env.put(AConstants.CC_JAVA_OPTS, ccJavaOpts);
        env.put(AConstants.NC_JAVA_OPTS, ncJavaOpts);

        // Add AppMaster.jar location to classpath
        // At some point we should not be required to add
        // the hadoop specific classpaths to the env.
        // It should be provided out of the box.
        // For now setting all required classpaths including
        // the classpath to "." for the application jar
        StringBuilder classPathEnv = new StringBuilder("").append("./*");
        for (String c : conf.getStrings(YarnConfiguration.YARN_APPLICATION_CLASSPATH,
                YarnConfiguration.DEFAULT_YARN_APPLICATION_CLASSPATH)) {
            classPathEnv.append(File.pathSeparatorChar);
            classPathEnv.append(c.trim());
        }
        classPathEnv.append(File.pathSeparatorChar).append("." + File.separator + "log4j.properties");

        // add the runtime classpath needed for tests to work
        if (conf.getBoolean(YarnConfiguration.IS_MINI_YARN_CLUSTER, false)) {
            LOG.info("In YARN MiniCluster");
            classPathEnv.append(System.getProperty("path.separator"));
            classPathEnv.append(System.getProperty("java.class.path"));
            env.put("HADOOP_CONF_DIR", System.getProperty("user.dir") + File.separator + "target" + File.separator);
        }
        LOG.info("AM Classpath:" + classPathEnv.toString());
        env.put("CLASSPATH", classPathEnv.toString());

        amContainer.setEnvironment(env);

        // Set the necessary command to execute the application master
        Vector<CharSequence> vargs = new Vector<CharSequence>(30);

        // Set java executable command
        LOG.info("Setting up app master command");
        vargs.add(JAVA_HOME + File.separator + "bin" + File.separator + "java");
        // Set class name
        vargs.add(appMasterMainClass);
        //Set params for Application Master
        if (debugFlag) {
            vargs.add("-debug");
        }
        if (mode == Mode.DESTROY) {
            vargs.add("-obliterate");
        } else if (mode == Mode.BACKUP) {
            vargs.add("-backup");
        } else if (mode == Mode.RESTORE) {
            vargs.add("-restore " + snapName);
        } else if (mode == Mode.INSTALL) {
            vargs.add("-initial ");
        }
        if (refresh) {
            vargs.add("-refresh");
        }
        //vargs.add("/bin/ls -alh asterix-server.zip/repo");
        vargs.add("1>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + File.separator + "AppMaster.stdout");
        vargs.add("2>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + File.separator + "AppMaster.stderr");
        // Get final commmand
        StringBuilder command = new StringBuilder();
        for (CharSequence str : vargs) {
            command.append(str).append(" ");
        }

        LOG.info("Completed setting up app master command " + command.toString());
        List<String> commands = new ArrayList<String>();
        commands.add(command.toString());
        amContainer.setCommands(commands);

        // Set up resource type requirements
        // For now, only memory is supported so we set memory requirements
        Resource capability = Records.newRecord(Resource.class);
        capability.setMemory(amMemory);
        appContext.setResource(capability);

        // Service data is a binary blob that can be passed to the application
        // Not needed in this scenario
        // amContainer.setServiceData(serviceData);

        // The following are not required for launching an application master
        // amContainer.setContainerId(containerId);

        appContext.setAMContainerSpec(amContainer);

        // Set the priority for the application master
        Priority pri = Records.newRecord(Priority.class);
        // TODO - what is the range for priority? how to decide?
        pri.setPriority(amPriority);
        appContext.setPriority(pri);

        // Set the queue to which this application is to be submitted in the RM
        appContext.setQueue(amQueue);

        // Submit the application to the applications manager
        // SubmitApplicationResponse submitResp =
        // applicationsManager.submitApplication(appRequest);
        // Ignore the response as either a valid response object is returned on
        // success
        // or an exception thrown to denote some form of a failure
        LOG.info("Submitting application to ASM");

        yarnClient.submitApplication(appContext);

        //now write the instance lock
        if (mode == Mode.INSTALL || mode == Mode.START) {
            FileSystem fs = FileSystem.get(conf);
            Path lock = new Path(fs.getHomeDirectory(), CONF_DIR_REL + instanceFolder + instanceLock);
            if (fs.exists(lock)) {
                throw new IllegalStateException("Somehow, this instance has been launched twice. ");
            }
            BufferedWriter out = new BufferedWriter(new OutputStreamWriter(fs.create(lock, true)));
            try {
                out.write(app.getApplicationSubmissionContext().getApplicationId().toString());
                out.close();
            } finally {
                out.close();
            }
        }
        return app.getApplicationSubmissionContext().getApplicationId();

    }

    /**
     * Asks YARN to kill a given application by appId
     *
     * @param appId
     *            The application to kill.
     * @param yarnClient
     *            The YARN client object that is connected to the RM.
     * @throws YarnException
     * @throws IOException
     */

    public static void killApplication(ApplicationId appId, YarnClient yarnClient) throws YarnException, IOException {
        if (appId == null) {
            throw new YarnException("No Application given to kill");
        }
        if (yarnClient.isInState(STATE.INITED)) {
            yarnClient.start();
        }
        YarnApplicationState st;
        ApplicationReport rep = yarnClient.getApplicationReport(appId);
        st = rep.getYarnApplicationState();
        if (st == YarnApplicationState.FINISHED || st == YarnApplicationState.KILLED
                || st == YarnApplicationState.FAILED) {
            LOG.info("Application " + appId + " already exited.");
            return;
        }
        LOG.info("Killing applicaiton with ID: " + appId);
        yarnClient.killApplication(appId);

    }

    /**
     * Tries to stop a running AsterixDB instance gracefully.
     *
     * @throws IOException
     * @throws YarnException
     */
    private void stopInstanceIfRunning() throws IOException, YarnException {
        FileSystem fs = FileSystem.get(conf);
        String pathSuffix = CONF_DIR_REL + instanceFolder + CONFIG_DEFAULT_NAME;
        Path dstConf = new Path(fs.getHomeDirectory(), pathSuffix);
        //if the instance is up, fix that
        if (isRunning()) {
            try {
                this.stopInstance();
            } catch (IOException e) {
                throw new YarnException(e);
            }
        } else if (!fs.exists(dstConf)) {
            throw new YarnException("No instance configured with that name exists");
        }
    }

    /**
     * Start a YARN job to delete local AsterixDB resources of an extant instance
     *
     * @param app
     *            The Client connection
     * @param resources
     *            AM resources
     * @throws IOException
     * @throws YarnException
     */

    private void removeInstance(YarnClientApplication app, List<DFSResourceCoordinate> resources)
            throws IOException, YarnException {
        FileSystem fs = FileSystem.get(conf);
        //if the instance is up, fix that
        stopInstanceIfRunning();
        //now try deleting all of the on-disk artifacts on the cluster
        ApplicationId deleter = deployAM(app, resources, Mode.DESTROY);
        boolean delete_start = Utils.waitForApplication(deleter, yarnClient, "Waiting for deletion to start",
                ccRestPort);
        if (!delete_start) {
            if (force) {
                fs.delete(new Path(CONF_DIR_REL + instanceFolder), true);
                LOG.error("Forcing deletion of HDFS resources");
            }
            LOG.fatal(" of on-disk persistient resources on individual nodes failed.");
            throw new YarnException();
        }
        boolean deleted = waitForCompletion(deleter, "Deletion in progress");
        if (!(deleted || force)) {
            LOG.fatal("Cleanup of on-disk persistent resources failed.");
            return;
        } else {
            fs.delete(new Path(CONF_DIR_REL + instanceFolder), true);
        }
        System.out.println("Deletion of instance succeeded.");

    }

    /**
     * Start a YARN job to copy all data-containing resources of an AsterixDB instance to HDFS
     *
     * @param app
     * @param resources
     * @throws IOException
     * @throws YarnException
     */

    private void backupInstance(YarnClientApplication app, List<DFSResourceCoordinate> resources)
            throws IOException, YarnException {
        stopInstanceIfRunning();
        ApplicationId backerUpper = deployAM(app, resources, Mode.BACKUP);
        boolean backupStart;
        backupStart = Utils.waitForApplication(backerUpper, yarnClient,
                "Waiting for backup " + backerUpper.toString() + "to start", ccRestPort);
        if (!backupStart) {
            LOG.fatal("Backup failed to start");
            throw new YarnException();
        }
        boolean complete;
        complete = waitForCompletion(backerUpper, "Backup in progress");
        if (!complete) {
            LOG.fatal("Backup failed- timeout waiting for completion");
            return;
        }
        System.out.println("Backup of instance succeeded.");
    }

    /**
     * Start a YARN job to copy a set of resources from backupInstance to restore the state of an extant AsterixDB instance
     *
     * @param app
     * @param resources
     * @throws IOException
     * @throws YarnException
     */

    private void restoreInstance(YarnClientApplication app, List<DFSResourceCoordinate> resources)
            throws IOException, YarnException {
        stopInstanceIfRunning();
        ApplicationId restorer = deployAM(app, resources, Mode.RESTORE);
        boolean restoreStart = Utils.waitForApplication(restorer, yarnClient, "Waiting for restore to start",
                ccRestPort);
        if (!restoreStart) {
            LOG.fatal("Restore failed to start");
            throw new YarnException();
        }
        boolean complete = waitForCompletion(restorer, "Restore in progress");
        if (!complete) {
            LOG.fatal("Restore failed- timeout waiting for completion");
            return;
        }
        System.out.println("Restoration of instance succeeded.");
    }

    /**
     * Stops the instance and remove the lockfile to allow a restart.
     *
     * @throws IOException
     * @throws JAXBException
     * @throws YarnException
     */

    private void stopInstance() throws IOException, YarnException {
        ApplicationId appId = getLockFile();
        //get CC rest API port if it is nonstandard
        readConfigParams(locateConfig());
        if (yarnClient.isInState(STATE.INITED)) {
            yarnClient.start();
        }
        System.out.println("Stopping instance " + instanceName);
        if (!isRunning()) {
            LOG.fatal("AsterixDB instance by that name is stopped already");
            return;
        }
        try {
            String ccIp = Utils.getCCHostname(instanceName, conf);
            Utils.sendShutdownCall(ccIp, ccRestPort);
        } catch (IOException e) {
            LOG.error("Error while trying to issue safe shutdown:", e);
        }
        //now make sure it is actually gone and not "stuck"
        String message = "Waiting for AsterixDB to shut down";
        boolean completed = waitForCompletion(appId, message);
        if (!completed && force) {
            LOG.warn("Instance failed to stop gracefully, now killing it");
            try {
                AsterixYARNClient.killApplication(appId, yarnClient);
                completed = true;
            } catch (YarnException e1) {
                LOG.fatal("Could not stop nor kill instance gracefully.", e1);
                return;
            }
        }
        if (completed) {
            deleteLockFile();
        }
    }

    private void deleteLockFile() throws IOException {
        if (instanceName == null || instanceName == "") {
            return;
        }
        FileSystem fs = FileSystem.get(conf);
        Path lockPath = new Path(fs.getHomeDirectory(), CONF_DIR_REL + instanceName + '/' + instanceLock);
        if (fs.exists(lockPath)) {
            fs.delete(lockPath, false);
        }
    }

    private boolean instanceExists() throws IOException {
        FileSystem fs = FileSystem.get(conf);
        String pathSuffix = CONF_DIR_REL + instanceFolder + CONFIG_DEFAULT_NAME;
        Path dstConf = new Path(fs.getHomeDirectory(), pathSuffix);
        return fs.exists(dstConf);
    }

    private boolean isRunning() throws IOException {
        FileSystem fs = FileSystem.get(conf);
        String pathSuffix = CONF_DIR_REL + instanceFolder + CONFIG_DEFAULT_NAME;
        Path dstConf = new Path(fs.getHomeDirectory(), pathSuffix);
        if (fs.exists(dstConf)) {
            Path lock = new Path(fs.getHomeDirectory(), CONF_DIR_REL + instanceFolder + instanceLock);
            return fs.exists(lock);
        } else {
            return false;
        }
    }

    private ApplicationId getLockFile() throws IOException, YarnException {
        if (instanceFolder == "") {
            throw new IllegalStateException("Instance name not given.");
        }
        FileSystem fs = FileSystem.get(conf);
        Path lockPath = new Path(fs.getHomeDirectory(), CONF_DIR_REL + instanceFolder + instanceLock);
        if (!fs.exists(lockPath)) {
            throw new YarnException("Instance appears to not be running. If you know it is, try using kill");
        }
        BufferedReader br = new BufferedReader(new InputStreamReader(fs.open(lockPath)));
        String lockAppId = br.readLine();
        br.close();
        return ConverterUtils.toApplicationId(lockAppId);
    }

    public static ApplicationId getLockFile(String instanceName, Configuration conf) throws IOException {
        if (instanceName == "") {
            throw new IllegalStateException("Instance name not given.");
        }
        FileSystem fs = FileSystem.get(conf);
        Path lockPath = new Path(fs.getHomeDirectory(), CONF_DIR_REL + instanceName + '/' + instanceLock);
        if (!fs.exists(lockPath)) {
            return null;
        }
        BufferedReader br = new BufferedReader(new InputStreamReader(fs.open(lockPath)));
        String lockAppId = br.readLine();
        br.close();
        return ConverterUtils.toApplicationId(lockAppId);
    }

    /**
     * Locate the Asterix parameters file.
     *
     * @return
     * @throws FileNotFoundException
     * @throws IOException
     */
    private AsterixConfiguration locateConfig() throws FileNotFoundException, IOException {
        AsterixConfiguration configuration;
        String configPathBase = MERGED_PARAMETERS_PATH;
        if (baseConfig != null) {
            configuration = Utils.loadAsterixConfig(baseConfig);
            configPathBase = new File(baseConfig).getParentFile().getAbsolutePath() + File.separator
                    + PARAMS_DEFAULT_NAME;
            MERGED_PARAMETERS_PATH = configPathBase;
        } else {
            configuration = Utils.loadAsterixConfig(DEFAULT_PARAMETERS_PATH);
        }
        return configuration;
    }

    /**
     *
     */
    private void readConfigParams(AsterixConfiguration configuration) {
        //this is the "base" config that is inside the zip, we start here
        for (org.apache.asterix.common.configuration.Property property : configuration.getProperty()) {
            if (property.getName().equalsIgnoreCase(CC_JAVA_OPTS_KEY)) {
                ccJavaOpts = property.getValue();
            } else if (property.getName().equalsIgnoreCase(NC_JAVA_OPTS_KEY)) {
                ncJavaOpts = property.getValue();
            } else if (property.getName().equalsIgnoreCase(CC_REST_PORT_KEY)) {
                ccRestPort = Integer.parseInt(property.getValue());
            }

        }
    }

    /**
     * Retrieves necessary information from the cluster configuration and splices it into the Asterix configuration parameters
     *
     * @param cluster
     * @throws FileNotFoundException
     * @throws IOException
     */

    private void writeAsterixConfig(Cluster cluster) throws FileNotFoundException, IOException {
        String metadataNodeId = Utils.getMetadataNode(cluster).getId();
        String asterixInstanceName = instanceName;

        AsterixConfiguration configuration = locateConfig();

        readConfigParams(configuration);

        String version = Utils.getAsterixVersionFromClasspath();
        configuration.setVersion(version);

        configuration.setInstanceName(asterixInstanceName);
        List<Store> stores = new ArrayList<Store>();
        String storeDir = cluster.getStore().trim();
        for (Node node : cluster.getNode()) {
            String iodevices = node.getIodevices() == null ? cluster.getIodevices() : node.getIodevices();
            String[] nodeIdDevice = iodevices.split(",");
            StringBuilder nodeStores = new StringBuilder();
            for (int i = 0; i < nodeIdDevice.length; i++) {
                nodeStores.append(nodeIdDevice[i] + File.separator + storeDir + ",");
            }
            //remove last comma
            nodeStores.deleteCharAt(nodeStores.length() - 1);
            stores.add(new Store(node.getId(), nodeStores.toString()));
        }
        configuration.setStore(stores);
        List<Coredump> coredump = new ArrayList<Coredump>();
        String coredumpDir = null;
        List<TransactionLogDir> txnLogDirs = new ArrayList<TransactionLogDir>();
        String txnLogDir = null;
        for (Node node : cluster.getNode()) {
            coredumpDir = node.getLogDir() == null ? cluster.getLogDir() : node.getLogDir();
            coredump.add(new Coredump(node.getId(), coredumpDir + "coredump" + File.separator));
            txnLogDir = node.getTxnLogDir() == null ? cluster.getTxnLogDir() : node.getTxnLogDir(); //node or cluster-wide
            txnLogDirs.add(new TransactionLogDir(node.getId(), txnLogDir
                    + (txnLogDir.charAt(txnLogDir.length() - 1) == File.separatorChar ? File.separator : "")
                    + "txnLogs" //if the string doesn't have a trailing / add one
                    + File.separator));
        }
        configuration.setMetadataNode(metadataNodeId);
        configuration.setCoredump(coredump);
        configuration.setTransactionLogDir(txnLogDirs);
        FileOutputStream os = new FileOutputStream(MERGED_PARAMETERS_PATH);
        try {
            JAXBContext ctx = JAXBContext.newInstance(AsterixConfiguration.class);
            Marshaller marshaller = ctx.createMarshaller();
            marshaller.setProperty(Marshaller.JAXB_FORMATTED_OUTPUT, true);

            marshaller.marshal(configuration, os);
        } catch (JAXBException e) {
            throw new IOException(e);
        } finally {
            os.close();
        }
    }

    private boolean waitForCompletion(ApplicationId appId, String message) throws YarnException, IOException {
        return Utils.waitForApplication(appId, yarnClient, message, ccRestPort);
    }

    private class DFSResourceCoordinate {
        String name;
        LocalResource res;
        Map<String, String> envs;

        public DFSResourceCoordinate() {
            envs = new HashMap<String, String>(3);
        }
    }
}


text:
class AddCamposDevise < ActiveRecord::Migration[5.1]
  def change
    add_column :users, :first_name, :string
    add_column :users, :last_name, :string
    add_column :users, :cpf, :string
  end
end

redpajama_set_name: RedPajamaGithub


text:
package com.markjmind.markjbox.study;

/**
 * Created by markj on 2015-12-04.
 */
public class GsonItem {
    private String Product;
    private String Maker;
    private Integer Price;

    public String getProduct() {
        return Product;
    }
    public void setProduct(String product) {
        Product = product;
    }
    public String getMaker() {
        return Maker;
    }
    public void setMaker(String maker) {
        Maker = maker;
    }
    public Integer getPrice() {
        return Price;
    }
    public void setPrice(Integer price) {
        Price = price;
    }
}


text:
from fake_switches.cisco.command_processor.config_interface import \
    ConfigInterfaceCommandProcessor

class DellConfigInterfaceCommandProcessor(ConfigInterfaceCommandProcessor):
    def init(self, switch_configuration, terminal_controller, logger,
                 piping_processor, *args):
        super(DellConfigInterfaceCommandProcessor, self).init(
            switch_configuration, terminal_controller, logger, piping_processor,
            *args)
        self.description_strip_chars = "\"'"

    def get_prompt(self):
        if self.port.name.startswith("ethernet"):
            short_name = self.port.name.split(' ')[1]
        elif self.port.name.startswith("port-channel"):
            short_name = "ch{}".format(self.port.name.split(' ')[1])
        else:
            short_name = self.port.name.replace(' ', '').lower()
        return self.switch_configuration.name + "(config-if-%s)#" % short_name

    def do_description(self, *args):
        super(DellConfigInterfaceCommandProcessor, self).do_description(*args)
        self.write_line("")

    def do_no_description(self, *_):
        super(DellConfigInterfaceCommandProcessor, self).do_no_description(*_)
        self.write_line("")

    def do_no_shutdown(self, *_):
        super(DellConfigInterfaceCommandProcessor, self).do_no_shutdown(*_)
        self.write_line("")

    def do_shutdown(self, *_):
        super(DellConfigInterfaceCommandProcessor, self).do_shutdown(*_)
        self.write_line("")

    def do_spanning_tree(self, *args):
        if "disable".startswith(args[0]):
            self.port.spanning_tree = False
        if "portfast".startswith(args[0]):
            self.port.spanning_tree_portfast = True
        self.write_line("")

    def do_no_spanning_tree(self, *args):
        if "disable".startswith(args[0]):
            self.port.spanning_tree = None
        if "portfast".startswith(args[0]):
            self.port.spanning_tree_portfast = None
        self.write_line("")

    def do_lldp(self, *args):
        self.configure_lldp_port(args, target_value=True)
        self.write_line("")

    def do_no_lldp(self, *args):
        self.configure_lldp_port(args, target_value=False)
        self.write_line("")

    def configure_lldp_port(self, args, target_value):
        if "transmit".startswith(args[0]):
            self.port.lldp_transmit = target_value
        elif "receive".startswith(args[0]):
            self.port.lldp_receive = target_value
        elif "med".startswith(args[0]) and "transmit-tlv".startswith(args[1]):
            if "capabilities".startswith(args[2]):
                self.port.lldp_med_transmit_capabilities = target_value
            elif "network-policy".startswith(args[2]):
                self.port.lldp_med_transmit_network_policy = target_value

    def do_name(self, *args):
        if len(args) == 0:
            self.write_line("")
            self.write_line("Command not found / Incomplete command. Use ? to list commands.")
        elif len(args) > 1:
            self.write_line("                                     ^")
            self.write_line("% Invalid input detected at '^' marker.")
        elif len(args[0]) > 32:
            self.write_line("Name must be 32 characters or less.")
        else:
            vlan = self.switch_configuration.get_vlan(self.port.vlan_id)
            vlan.name = args[0]

        self.write_line("")

    def do_switchport(self, *args):
        if "access".startswith(args[0]) and "vlan".startswith(args[1]):
            self.set_access_vlan(int(args[2]))
        elif "mode".startswith(args[0]):
            self.set_switchport_mode(args[1])
        elif ("general".startswith(args[0]) or "trunk".startswith(args[0])) and "allowed".startswith(args[1]):
            if "general".startswith(args[0]) and self.port.mode != "general":
                self.write_line("Interface not in General Mode.")
            elif "trunk".startswith(args[0]) and self.port.mode != "trunk":
                self.write_line("Interface not in Trunk Mode.")
            elif "vlan".startswith(args[2]):
                if len(args) > 5:
                    self.write_line("                                                                 ^")
                    self.write_line("% Invalid input detected at '^' marker.")
                else:
                    operation = args[3]
                    vlan_range = args[4]
                    self.update_trunk_vlans(operation, vlan_range)
                    return
        elif "general".startswith(args[0]) and "pvid".startswith(args[1]):
            self.set_trunk_native_vlan(int(args[2]))

        self.write_line("")

    def do_no_switchport(self, *args):
        if "access".startswith(args[0]):
            if "vlan".startswith(args[1]):
                self.print_vlan_warning()
                self.port.access_vlan = None
        elif "general".startswith(args[0]):
            if "pvid".startswith(args[1]):
                self.port.trunk_native_vlan = None
        elif "mode".startswith(args[0]):
            self.port.trunk_vlans = None
            self.port.mode = None
            self.port.trunk_native_vlan = None
            self.port.trunk_vlans = None

        self.write_line("")

    def do_mtu(self, *args):
        if len(args) > 1:
            self.write_line("                                  ^")
            self.write_line("% Invalid input detected at '^' marker.")
            self.write_line("")
            return

        try:
            value = int(args[0])
        except ValueError:
            self.write_line("                            ^")
            self.write_line("Invalid input. Please specify an integer in the range 1518 to 9216.")
            return

        if not (1518 <= value <= 9216):
            self.write_line("                            ^")
            self.write_line("Value is out of range. The valid range is 1518 to 9216.")
            return

        self.port.mtu = value
        self.write_line("")

    def do_no_mtu(self, *args):
        self.port.mtu = None
        self.write_line("")

    def set_switchport_mode(self, mode):
        if mode not in ("access", "trunk", "general"):
            self.write_line("                                         ^")
            self.write_line("% Invalid input detected at '^' marker.")
        else:
            if self.port.mode != mode:
                self.port.mode = mode
                self.port.access_vlan = None
                self.port.trunk_native_vlan = None
                self.port.trunk_vlans = None

    def set_access_vlan(self, vlan_id):
        self.print_vlan_warning()
        vlan = self.switch_configuration.get_vlan(int(vlan_id))
        if vlan:
            self.port.access_vlan = vlan.number
        else:
            self.write_line("")
            self.write_line("VLAN ID not found.")

    def set_trunk_native_vlan(self, native_vlan):
        if self.port.mode != "general":
            self.write_line("")
            self.write_line("Port is not general port.")
        else:
            vlan = self.switch_configuration.get_vlan(native_vlan)
            if vlan is None:
                self.write_line("Could not configure pvid.")
            else:
                self.port.trunk_native_vlan = vlan.number

    def update_trunk_vlans(self, operation, vlan_range):
        try:
            vlans = parse_vlan_list(vlan_range)
        except ValueError:
            self.write_line("VLAN range - separate non-consecutive IDs with ',' and no spaces.  Use '-' for range.")
            self.write_line("")
            return

        self.print_vlan_warning()

        vlans_not_found = [v for v in vlans if self.switch_configuration.get_vlan(v) is None]
        if len(vlans_not_found) > 0:
            self.write_line("")
            self.write_line("          Failure Information")
            self.write_line("---------------------------------------")
            self.write_line("   VLANs failed to be configured : {}".format(len(vlans_not_found)))
            self.write_line("---------------------------------------")
            self.write_line("   VLAN             Error")
            self.write_line("---------------------------------------")
            for vlan in vlans_not_found:
                self.write_line("VLAN      {: >4} ERROR: This VLAN does not exist.".format(vlan))
            return

        if "add".startswith(operation):
            if self.port.trunk_vlans is None:
                self.port.trunk_vlans = []
            self.port.trunk_vlans = list(set(self.port.trunk_vlans + vlans))
        if "remove".startswith(operation):
            for v in vlans:
                if v in self.port.trunk_vlans:
                    self.port.trunk_vlans.remove(v)
            if len(self.port.trunk_vlans) == 0:
                self.port.trunk_vlans = None

        self.write_line("")

    def print_vlan_warning(self):
        self.write_line("Warning: The use of large numbers of VLANs or interfaces may cause significant")
        self.write_line("delays in applying the configuration.")
        self.write_line("")

def parse_vlan_list(param):
    ranges = param.split(",")
    vlans = []
    for r in ranges:
        if "-" in r:
            start, stop = r.split("-")
            if stop < start:
                raise ValueError
            vlans += [v for v in range(int(start), int(stop) + 1)]
        else:
            vlans.append(int(r))

    return vlans


text:
from fabric.api import *
from fabric.contrib.files import exists
import argparse
import json
import random
import string
import sys
from dlab.notebook_lib import *
from dlab.fab import *
from dlab.common_lib import *
import backoff
import os
import re

def enable_proxy(proxy_host, proxy_port):
    try:
        proxy_string = "http://%s:%s" % (proxy_host, proxy_port)
        sudo('sed -i "/^export http_proxy/d" /etc/profile')
        sudo('sed -i "/^export https_proxy/d" /etc/profile')
        sudo('echo export http_proxy=' + proxy_string + ' >> /etc/profile')
        sudo('echo export https_proxy=' + proxy_string + ' >> /etc/profile')
        if exists('/etc/apt/apt.conf'):
            sudo("sed -i '/^Acquire::http::Proxy/d' /etc/apt/apt.conf")
        sudo("echo 'Acquire::http::Proxy \"" + proxy_string + "\";' >> /etc/apt/apt.conf")

        print("Renewing gpg key")
        renew_gpg_key()
    except:
        sys.exit(1)

def ensure_r_local_kernel(spark_version, os_user, templates_dir, kernels_dir):
    if not exists('/home/' + os_user + '/.ensure_dir/r_local_kernel_ensured'):
        try:
            sudo('R -e "IRkernel::installspec()"')
            r_version = sudo("R --version | awk '/version / {print $3}'")
            put(templates_dir + 'r_template.json', '/tmp/r_template.json')
            sudo('sed -i "s|R_VER|' + r_version + '|g" /tmp/r_template.json')
            sudo('sed -i "s|SP_VER|' + spark_version + '|g" /tmp/r_template.json')
            sudo('\cp -f /tmp/r_template.json {}/ir/kernel.json'.format(kernels_dir))
            sudo('ln -s /opt/spark/ /usr/local/spark')
            sudo('cd /usr/local/spark/R/lib/SparkR; R -e "devtools::install(\'.\')"')
            sudo('chown -R ' + os_user + ':' + os_user + ' /home/' + os_user + '/.local')
            sudo('touch /home/' + os_user + '/.ensure_dir/r_local_kernel_ensured')
        except:
            sys.exit(1)

@backoff.on_exception(backoff.expo, SystemExit, max_tries=20)
def add_marruter_key():
    try:
        sudo('add-apt-repository -y ppa:marutter/rrutter')
    except:
        sys.exit(1)

def ensure_r(os_user, r_libs, region, r_mirror):
    if not exists('/home/' + os_user + '/.ensure_dir/r_ensured'):
        try:
            if region == 'cn-north-1':
                r_repository = r_mirror
            else:
                r_repository = 'http://cran.us.r-project.org'
            add_marruter_key()
            sudo('apt update')
            sudo('apt-get install -y libcurl4-openssl-dev libssl-dev libreadline-dev')
            sudo('apt-get install -y cmake')
            sudo('apt-get install -y r-base r-base-dev')
            sudo('R CMD javareconf')
            sudo('cd /root; git clone https://github.com/zeromq/zeromq4-x.git; cd zeromq4-x/; mkdir build; cd build; cmake ..; make install; ldconfig')
            for i in r_libs:
                sudo('R -e "install.packages(\'{}\',repos=\'{}\')"'.format(i, r_repository))
            sudo('R -e "library(\'devtools\');install.packages(repos=\'{}\',c(\'rzmq\',\'repr\',\'digest\',\'stringr\',\'RJSONIO\',\'functional\',\'plyr\'))"'.format(r_repository))
            try:
                sudo('R -e "library(\'devtools\');install_github(\'IRkernel/repr\');install_github(\'IRkernel/IRdisplay\');install_github(\'IRkernel/IRkernel\');"')
            except:
                sudo('R -e "options(download.file.method = "wget");library(\'devtools\');install_github(\'IRkernel/repr\');install_github(\'IRkernel/IRdisplay\');install_github(\'IRkernel/IRkernel\');"')
            if os.environ['application'] == 'tensor-rstudio':
                sudo('R -e "library(\'devtools\');install_github(\'rstudio/keras\');"')
            sudo('R -e "install.packages(\'RJDBC\',repos=\'{}\',dep=TRUE)"'.format(r_repository))
            sudo('touch /home/' + os_user + '/.ensure_dir/r_ensured')
        except:
            sys.exit(1)

def install_rstudio(os_user, local_spark_path, rstudio_pass, rstudio_version):
    if not exists('/home/' + os_user + '/.ensure_dir/rstudio_ensured'):
        try:
            sudo('apt-get install -y r-base')
            sudo('apt-get install -y gdebi-core')
            sudo('wget https://download2.rstudio.org/rstudio-server-{}-amd64.deb'.format(rstudio_version))
            sudo('gdebi -n rstudio-server-{}-amd64.deb'.format(rstudio_version))
            sudo('mkdir -p /mnt/var')
            sudo('chown {0}:{0} /mnt/var'.format(os_user))
            if os.environ['application'] == 'tensor-rstudio':
                sudo("sed -i '/ExecStart/s|=|=/bin/bash -c \"export LD_LIBRARY_PATH=$LD_LIBRARY_PATH:/opt/cudnn/lib64:/usr/local/cuda/lib64; |g' /etc/systemd/system/rstudio-server.service")
                sudo("sed -i '/ExecStart/s|$|\"|g' /etc/systemd/system/rstudio-server.service")
                sudo("systemctl daemon-reload")
            sudo('touch /home/{}/.Renviron'.format(os_user))
            sudo('chown {0}:{0} /home/{0}/.Renviron'.format(os_user))
            sudo('''echo 'SPARK_HOME="{0}"' >> /home/{1}/.Renviron'''.format(local_spark_path, os_user))
            sudo('touch /home/{}/.Rprofile'.format(os_user))
            sudo('chown {0}:{0} /home/{0}/.Rprofile'.format(os_user))
            sudo('''echo 'library(SparkR, lib.loc = c(file.path(Sys.getenv("SPARK_HOME"), "R", "lib")))' >> /home/{}/.Rprofile'''.format(os_user))
            http_proxy = run('echo $http_proxy')
            https_proxy = run('echo $https_proxy')
            sudo('''echo 'Sys.setenv(http_proxy = \"{}\")' >> /home/{}/.Rprofile'''.format(http_proxy, os_user))
            sudo('''echo 'Sys.setenv(https_proxy = \"{}\")' >> /home/{}/.Rprofile'''.format(https_proxy, os_user))
            sudo('rstudio-server start')
            sudo('echo "{0}:{1}" | chpasswd'.format(os_user, rstudio_pass))
            sudo("sed -i '/exit 0/d' /etc/rc.local")
            sudo('''bash -c "echo \'sed -i 's/^#SPARK_HOME/SPARK_HOME/' /home/{}/.Renviron\' >> /etc/rc.local"'''.format(os_user))
            sudo("bash -c 'echo exit 0 >> /etc/rc.local'")
            sudo('touch /home/{}/.ensure_dir/rstudio_ensured'.format(os_user))
        except:
            sys.exit(1)
    else:
        try:
            sudo('echo "{0}:{1}" | chpasswd'.format(os_user, rstudio_pass))
        except:
            sys.exit(1)

def ensure_matplot(os_user):
    if not exists('/home/' + os_user + '/.ensure_dir/matplot_ensured'):
        try:
            sudo('apt-get build-dep -y python-matplotlib')
            sudo('pip2 install matplotlib==2.0.2 --no-cache-dir')
            sudo('pip3 install matplotlib==2.0.2 --no-cache-dir')
            if os.environ['application'] in ('tensor', 'deeplearning'):
                sudo('python2.7 -m pip install -U numpy=={} --no-cache-dir'.format(os.environ['notebook_numpy_version']))
                sudo('python3.5 -m pip install -U numpy=={} --no-cache-dir'.format(os.environ['notebook_numpy_version']))
            sudo('touch /home/' + os_user + '/.ensure_dir/matplot_ensured')
        except:
            sys.exit(1)

@backoff.on_exception(backoff.expo, SystemExit, max_tries=10)
def add_sbt_key():
    sudo(
        'apt-key adv --keyserver hkp://keyserver.ubuntu.com:80 --recv 642AC823')

def ensure_sbt(os_user):
    if not exists('/home/' + os_user + '/.ensure_dir/sbt_ensured'):
        try:
            sudo('apt-get install -y apt-transport-https')
            sudo('echo "deb https://dl.bintray.com/sbt/debian /" | sudo tee -a /etc/apt/sources.list.d/sbt.list')
            add_sbt_key()
            sudo('apt-get update')
            sudo('apt-get install -y sbt')
            sudo('touch /home/' + os_user + '/.ensure_dir/sbt_ensured')
        except:
            sys.exit(1)

def ensure_scala(scala_link, scala_version, os_user):
    if not exists('/home/' + os_user + '/.ensure_dir/scala_ensured'):
        try:
            sudo('wget {}scala-{}.deb -O /tmp/scala.deb'.format(scala_link, scala_version))
            sudo('dpkg -i /tmp/scala.deb')
            sudo('touch /home/' + os_user + '/.ensure_dir/scala_ensured')
        except:
            sys.exit(1)

def ensure_jre_jdk(os_user):
    if not exists('/home/' + os_user + '/.ensure_dir/jre_jdk_ensured'):
        try:
            sudo('apt-get install -y default-jre')
            sudo('apt-get install -y default-jdk')
            sudo('touch /home/' + os_user + '/.ensure_dir/jre_jdk_ensured')
        except:
            sys.exit(1)

def ensure_additional_python_libs(os_user):
    if not exists('/home/' + os_user + '/.ensure_dir/additional_python_libs_ensured'):
        try:
            sudo('apt-get install -y libjpeg8-dev zlib1g-dev')
            if os.environ['application'] in ('jupyter', 'zeppelin'):
                sudo('pip2 install NumPy=={} SciPy pandas Sympy Pillow sklearn --no-cache-dir'.format(os.environ['notebook_numpy_version']))
                sudo('pip3 install NumPy=={} SciPy pandas Sympy Pillow sklearn --no-cache-dir'.format(os.environ['notebook_numpy_version']))
            if os.environ['application'] in ('tensor', 'deeplearning'):
                sudo('pip2 install opencv-python h5py --no-cache-dir')
                sudo('pip3 install opencv-python h5py --no-cache-dir')
            sudo('touch /home/' + os_user + '/.ensure_dir/additional_python_libs_ensured')
        except:
            sys.exit(1)

def ensure_python3_specific_version(python3_version, os_user):
    if not exists('/home/' + os_user + '/.ensure_dir/python3_specific_version_ensured'):
        try:
            if len(python3_version) < 4:
                python3_version = python3_version + ".0"
            sudo('wget https://www.python.org/ftp/python/{0}/Python-{0}.tgz'.format(python3_version))
            sudo('tar xzf Python-{0}.tgz; cd Python-{0}; ./configure --prefix=/usr/local; make altinstall'.format(python3_version))
            sudo('touch /home/' + os_user + '/.ensure_dir/python3_specific_version_ensured')
        except:
            sys.exit(1)

def ensure_python2_libraries(os_user):
    if not exists('/home/' + os_user + '/.ensure_dir/python2_libraries_ensured'):
        try:
            try:
                sudo('apt-get install -y libssl-dev python-virtualenv')
            except:
                sudo('pip2 install virtualenv --no-cache-dir')
                sudo('apt-get install -y libssl-dev')
            try:
                sudo('pip2 install tornado=={0} ipython ipykernel=={1} --no-cache-dir' \
                     .format(os.environ['notebook_tornado_version'], os.environ['notebook_ipykernel_version']))
            except:
                sudo('pip2 install tornado=={0} ipython==5.0.0 ipykernel=={1} --no-cache-dir' \
                     .format(os.environ['notebook_tornado_version'], os.environ['notebook_ipykernel_version']))
            sudo('pip2 install -U pip=={} --no-cache-dir'.format(os.environ['conf_pip_version']))
            sudo('pip2 install boto3 backoff --no-cache-dir')
            sudo('pip2 install fabvenv fabric-virtualenv future --no-cache-dir')
            sudo('touch /home/' + os_user + '/.ensure_dir/python2_libraries_ensured')
        except:
            sys.exit(1)

def ensure_python3_libraries(os_user):
    if not exists('/home/' + os_user + '/.ensure_dir/python3_libraries_ensured'):
        try:
            sudo('apt-get install python3-setuptools')
            sudo('apt install -y python3-pip')
            try:
                sudo('pip3 install tornado=={0} ipython ipykernel=={1} --no-cache-dir' \
                     .format(os.environ['notebook_tornado_version'], os.environ['notebook_ipykernel_version']))
            except:
                sudo('pip3 install tornado=={0} ipython==5.0.0 ipykernel=={1} --no-cache-dir' \
                     .format(os.environ['notebook_tornado_version'], os.environ['notebook_ipykernel_version']))
            sudo('pip3 install -U pip=={} --no-cache-dir'.format(os.environ['conf_pip_version']))
            sudo('pip3 install boto3 --no-cache-dir')
            sudo('pip3 install fabvenv fabric-virtualenv future --no-cache-dir')
            sudo('touch /home/' + os_user + '/.ensure_dir/python3_libraries_ensured')
        except:
            sys.exit(1)

def install_tensor(os_user, cuda_version, cuda_file_name,
                   cudnn_version, cudnn_file_name, tensorflow_version,
                   templates_dir, nvidia_version):
    if not exists('/home/{}/.ensure_dir/tensor_ensured'.format(os_user)):
        try:
            # install nvidia drivers
            sudo('echo "blacklist nouveau" >> /etc/modprobe.d/blacklist-nouveau.conf')
            sudo('echo "options nouveau modeset=0" >> /etc/modprobe.d/blacklist-nouveau.conf')
            sudo('update-initramfs -u')
            with settings(warn_only=True):
                reboot(wait=150)
            sudo('apt-get -y install dkms')
            kernel_version = run('uname -r | tr -d "[..0-9-]"')
            if kernel_version == 'azure':
                sudo('apt-get -y install linux-modules-extra-`uname -r`')
            else:
                #legacy support for old kernels
                sudo('if [[ $(apt-cache search linux-image-extra-`uname -r`) ]]; then apt-get -y install linux-image-extra-`uname -r`; else apt-get -y install linux-modules-extra-`uname -r`; fi;')
            sudo('wget http://us.download.nvidia.com/XFree86/Linux-x86_64/{0}/NVIDIA-Linux-x86_64-{0}.run -O /home/{1}/NVIDIA-Linux-x86_64-{0}.run'.format(nvidia_version, os_user))
            sudo('/bin/bash /home/{0}/NVIDIA-Linux-x86_64-{1}.run -s --dkms'.format(os_user, nvidia_version))
            sudo('rm -f /home/{0}/NVIDIA-Linux-x86_64-{1}.run'.format(os_user, nvidia_version))
            # install cuda
            sudo('python3.5 -m pip install --upgrade pip=={0} wheel numpy=={1} --no-cache-dir'. format(os.environ['conf_pip_version'], os.environ['notebook_numpy_version']))
            sudo('wget -P /opt https://developer.nvidia.com/compute/cuda/{0}/prod/local_installers/{1}'.format(cuda_version, cuda_file_name))
            sudo('sh /opt/{} --silent --toolkit'.format(cuda_file_name))
            sudo('mv /usr/local/cuda-{} /opt/'.format(cuda_version))
            sudo('ln -s /opt/cuda-{0} /usr/local/cuda-{0}'.format(cuda_version))
            sudo('rm -f /opt/{}'.format(cuda_file_name))
            # install cuDNN
            run('wget http://developer.download.nvidia.com/compute/redist/cudnn/v{0}/{1} -O /tmp/{1}'.format(cudnn_version, cudnn_file_name))
            run('tar xvzf /tmp/{} -C /tmp'.format(cudnn_file_name))
            sudo('mkdir -p /opt/cudnn/include')
            sudo('mkdir -p /opt/cudnn/lib64')
            sudo('mv /tmp/cuda/include/cudnn.h /opt/cudnn/include')
            sudo('mv /tmp/cuda/lib64/libcudnn* /opt/cudnn/lib64')
            sudo('chmod a+r /opt/cudnn/include/cudnn.h /opt/cudnn/lib64/libcudnn*')
            run('echo "export LD_LIBRARY_PATH=\"$LD_LIBRARY_PATH:/opt/cudnn/lib64:/usr/local/cuda/lib64\"" >> ~/.bashrc')
            # install TensorFlow and run TensorBoard
            sudo('python2.7 -m pip install --upgrade https://storage.googleapis.com/tensorflow/linux/gpu/tensorflow_gpu-{}-cp27-none-linux_x86_64.whl --no-cache-dir'.format(tensorflow_version))
            sudo('python3 -m pip install --upgrade https://storage.googleapis.com/tensorflow/linux/gpu/tensorflow_gpu-{}-cp35-cp35m-linux_x86_64.whl --no-cache-dir'.format(tensorflow_version))
            sudo('mkdir /var/log/tensorboard; chown {0}:{0} -R /var/log/tensorboard'.format(os_user))
            put('{}tensorboard.service'.format(templates_dir), '/tmp/tensorboard.service')
            sudo("sed -i 's|OS_USR|{}|' /tmp/tensorboard.service".format(os_user))
            sudo("chmod 644 /tmp/tensorboard.service")
            sudo('\cp /tmp/tensorboard.service /etc/systemd/system/')
            sudo("systemctl daemon-reload")
            sudo("systemctl enable tensorboard")
            sudo("systemctl start tensorboard")
            sudo('touch /home/{}/.ensure_dir/tensor_ensured'.format(os_user))

        except:
            sys.exit(1)

def install_maven(os_user):
    if not exists('/home/' + os_user + '/.ensure_dir/maven_ensured'):
        sudo('apt-get -y install maven')
        sudo('touch /home/' + os_user + '/.ensure_dir/maven_ensured')

def install_livy_dependencies(os_user):
    if not exists('/home/' + os_user + '/.ensure_dir/livy_dependencies_ensured'):
        sudo('apt-get -y install libkrb5-dev')
        sudo('pip2 install cloudpickle requests requests-kerberos flake8 flaky pytest --no-cache-dir')
        sudo('pip3 install cloudpickle requests requests-kerberos flake8 flaky pytest --no-cache-dir')
        sudo('touch /home/' + os_user + '/.ensure_dir/livy_dependencies_ensured')

def install_maven_emr(os_user):
    if not os.path.exists('/home/' + os_user + '/.ensure_dir/maven_ensured'):
        local('sudo apt-get -y install maven')
        local('touch /home/' + os_user + '/.ensure_dir/maven_ensured')

def install_livy_dependencies_emr(os_user):
    if not os.path.exists('/home/' + os_user + '/.ensure_dir/livy_dependencies_ensured'):
        local('sudo apt-get -y install libkrb5-dev')
        local('sudo pip2 install cloudpickle requests requests-kerberos flake8 flaky pytest --no-cache-dir')
        local('sudo pip3 install cloudpickle requests requests-kerberos flake8 flaky pytest --no-cache-dir')
        local('touch /home/' + os_user + '/.ensure_dir/livy_dependencies_ensured')

def install_nodejs(os_user):
    if not exists('/home/{}/.ensure_dir/nodejs_ensured'.format(os_user)):
        sudo('curl -sL https://deb.nodesource.com/setup_6.x | sudo -E bash -')
        sudo('apt-get install -y nodejs')
        sudo('touch /home/{}/.ensure_dir/nodejs_ensured'.format(os_user))

def install_os_pkg(requisites):
    status = list()
    error_parser = "Could not|No matching|Error:|failed|Requires:"
    try:
        print("Updating repositories and installing requested tools: {}".format(requisites))
        sudo('apt-get update')
        for os_pkg in requisites:
            sudo('DEBIAN_FRONTEND=noninteractive apt-get -y install {0} 2>&1 | if ! grep -w -E  "({1})" >  /tmp/os_install_{0}.log; then  echo "" > /tmp/os_install_{0}.log;fi'.format(os_pkg, error_parser))
            err = sudo('cat /tmp/os_install_{}.log'.format(os_pkg)).replace('"', "'")
            sudo('apt list --installed | if ! grep {0}/ > /tmp/os_install_{0}.list; then  echo "" > /tmp/os_install_{0}.list;fi'.format(os_pkg))
            res = sudo('cat /tmp/os_install_{}.list'.format(os_pkg))
            if res:
                ansi_escape = re.compile(r'\x1b[^m]*m')
                ver = ansi_escape.sub('', res).split("\r\n")
                version = [i for i in ver if os_pkg in i][0].split(' ')[1]
                status.append({"group": "os_pkg", "name": os_pkg, "version": version, "status": "installed"})
            else:
                status.append({"group": "os_pkg", "name": os_pkg, "status": "failed", "error_message": err})
        sudo('unattended-upgrades -v')
        sudo('export LC_ALL=C')
        return status
    except:
        return "Fail to install OS packages"

@backoff.on_exception(backoff.expo, SystemExit, max_tries=10)
def remove_os_pkg(pkgs):
    try:
        sudo('apt remove --purge -y {}'.format(' '.join(pkgs)))
    except:
        sys.exit(1)

def get_available_os_pkgs():
    try:
        os_pkgs = dict()
        ansi_escape = re.compile(r'\x1b[^m]*m')
        sudo('apt-get update')
        apt_raw = sudo("apt list")
        apt_list = ansi_escape.sub('', apt_raw).split("\r\n")
        for pkg in apt_list:
            if "/" in pkg:
                os_pkgs[pkg.split('/')[0]] = pkg.split(' ')[1]
        return os_pkgs
    except:
        sys.exit(1)

def install_caffe(os_user, region, caffe_version):
    if not exists('/home/{}/.ensure_dir/caffe_ensured'.format(os_user)):
        env.shell = "/bin/bash -l -c -i"
        sudo('apt-get install -y python-dev')
        sudo('apt-get install -y python3-dev')
        sudo('apt-get install -y libprotobuf-dev libleveldb-dev libsnappy-dev libopencv-dev libhdf5-serial-dev protobuf-compiler')
        sudo('apt-get install -y --no-install-recommends libboost-all-dev')
        sudo('apt-get install -y libatlas-base-dev libopenblas-dev')
        sudo('apt-get install -y libgflags-dev libgoogle-glog-dev liblmdb-dev')
        with cd('/usr/lib/x86_64-linux-gnu/'):
            sudo('ln -s libhdf5_serial_hl.so.10.0.2 libhdf5_hl.so')
            sudo('ln -s libhdf5_serial.so.10.1.0 libhdf5.so')
        sudo('git clone https://github.com/BVLC/caffe.git')
        with cd('/home/{}/caffe/'.format(os_user)):
            sudo('git checkout {}'.format(caffe_version))
            sudo('pip2 install -r python/requirements.txt --no-cache-dir')
            sudo('pip3 install -r python/requirements.txt --no-cache-dir')
            sudo('echo "CUDA_DIR := /usr/local/cuda" > Makefile.config')
            cuda_arch = sudo("/opt/cuda-8.0/extras/demo_suite/deviceQuery | grep 'CUDA Capability' | tr -d ' ' | cut -f2 -d ':'")
            sudo('echo "CUDA_ARCH := -gencode arch=compute_{0},code=sm_{0}" >> Makefile.config'.format(cuda_arch.replace('.', '')))
            sudo('echo "PYTHON_INCLUDE := /usr/include/python2.7 /usr/local/lib/python2.7/dist-packages/numpy/core/include" >> Makefile.config')
            sudo('echo "BLAS := open" >> Makefile.config')
            sudo('echo "BLAS_INCLUDE := /usr/include/openblas" >> Makefile.config')
            #sudo('echo "OPENCV_VERSION := 3" >> Makefile.config')
            sudo('echo "LIBRARIES += glog gflags protobuf boost_system boost_filesystem m hdf5_serial_hl hdf5_serial" >> Makefile.config')
            sudo('echo "PYTHON_LIB := /usr/lib" >> Makefile.config')
            sudo('echo "INCLUDE_DIRS := \\\$(PYTHON_INCLUDE) /usr/local/include /usr/include/hdf5/serial/ /usr /usr/lib /usr/include/python2.7 /usr/local/lib/python2.7/dist-packages/numpy/core/include" >> Makefile.config')
            sudo('echo "LIBRARY_DIRS := \\\$(PYTHON_LIB) /usr/local/lib /usr/lib /usr /usr/lib" >> Makefile.config')
            sudo('echo "BUILD_DIR := build" >> Makefile.config')
            sudo('echo "DISTRIBUTE_DIR := distribute" >> Makefile.config')
            sudo('echo "TEST_GPUID := 0" >> Makefile.config')
            sudo('echo "Q ?= @" >> Makefile.config')
            sudo('make all -j$(nproc)')
            sudo('make test -j$(nproc)')
            run('make runtest')
            sudo('make pycaffe')
        sudo('touch /home/' + os_user + '/.ensure_dir/caffe_ensured')

def install_caffe2(os_user, caffe2_version, cmake_version):
    if not exists('/home/{}/.ensure_dir/caffe2_ensured'.format(os_user)):
        env.shell = "/bin/bash -l -c -i"
        sudo('apt-get update')
        sudo('apt-get install -y --no-install-recommends build-essential cmake git libgoogle-glog-dev libprotobuf-dev'
             ' protobuf-compiler python-dev python-pip')
        sudo('pip2 install numpy=={} protobuf --no-cache-dir'.format(os.environ['notebook_numpy_version']))
        sudo('pip3 install numpy=={} protobuf --no-cache-dir'.format(os.environ['notebook_numpy_version']))
        sudo('apt-get install -y --no-install-recommends libgflags-dev')
        sudo('apt-get install -y --no-install-recommends libgtest-dev libiomp-dev libleveldb-dev liblmdb-dev '
             'libopencv-dev libopenmpi-dev libsnappy-dev openmpi-bin openmpi-doc python-pydot')
        sudo('pip2 install flask graphviz hypothesis jupyter matplotlib==2.0.2 pydot python-nvd3 pyyaml requests scikit-image '
             'scipy setuptools tornado --no-cache-dir')
        sudo('pip3 install flask graphviz hypothesis jupyter matplotlib==2.0.2 pydot python-nvd3 pyyaml requests scikit-image '
             'scipy setuptools tornado --no-cache-dir')
        sudo('cp -f /opt/cudnn/include/* /opt/cuda-8.0/include/')
        sudo('cp -f /opt/cudnn/lib64/* /opt/cuda-8.0/lib64/')
        sudo('wget https://cmake.org/files/v{2}/cmake-{1}.tar.gz -O /home/{0}/cmake-{1}.tar.gz'.format(
            os_user, cmake_version, cmake_version.split('.')[0] + "." + cmake_version.split('.')[1]))
        sudo('tar -zxvf cmake-{}.tar.gz'.format(cmake_version))
        with cd('/home/{}/cmake-{}/'.format(os_user, cmake_version)):
            sudo('./bootstrap --prefix=/usr/local && make && make install')
        sudo('ln -s /usr/local/bin/cmake /bin/cmake{}'.format(cmake_version))
        sudo('git clone https://github.com/pytorch/pytorch.git')
        with cd('/home/{}/pytorch/'.format(os_user)):
            sudo('git submodule update --init')
            with settings(warn_only=True):
                sudo('git checkout v{}'.format(caffe2_version))
                sudo('git submodule update --recursive')
            sudo('mkdir build && cd build && cmake{} .. && make "-j$(nproc)" install'.format(cmake_version))
        sudo('touch /home/' + os_user + '/.ensure_dir/caffe2_ensured')

def install_cntk(os_user, cntk_version):
    if not exists('/home/{}/.ensure_dir/cntk_ensured'.format(os_user)):
        sudo('pip2 install https://cntk.ai/PythonWheel/GPU/cntk-{}-cp27-cp27mu-linux_x86_64.whl --no-cache-dir'.format(cntk_version))
        sudo('pip3 install https://cntk.ai/PythonWheel/GPU/cntk-{}-cp35-cp35m-linux_x86_64.whl --no-cache-dir'.format(cntk_version))
        sudo('touch /home/{}/.ensure_dir/cntk_ensured'.format(os_user))

def install_keras(os_user, keras_version):
    if not exists('/home/{}/.ensure_dir/keras_ensured'.format(os_user)):
        sudo('pip2 install keras=={} --no-cache-dir'.format(keras_version))
        sudo('pip3 install keras=={} --no-cache-dir'.format(keras_version))
        sudo('touch /home/{}/.ensure_dir/keras_ensured'.format(os_user))

def install_theano(os_user, theano_version):
    if not exists('/home/{}/.ensure_dir/theano_ensured'.format(os_user)):
        sudo('python2.7 -m pip install Theano=={} --no-cache-dir'.format(theano_version))
        sudo('python3 -m pip install Theano=={} --no-cache-dir'.format(theano_version))
        sudo('touch /home/{}/.ensure_dir/theano_ensured'.format(os_user))

def install_mxnet(os_user, mxnet_version):
    if not exists('/home/{}/.ensure_dir/mxnet_ensured'.format(os_user)):
        sudo('pip2 install mxnet-cu80=={} opencv-python --no-cache-dir'.format(mxnet_version))
        sudo('pip3 install mxnet-cu80=={} opencv-python --no-cache-dir'.format(mxnet_version))
        sudo('touch /home/{}/.ensure_dir/mxnet_ensured'.format(os_user))

def install_torch(os_user):
    if not exists('/home/{}/.ensure_dir/torch_ensured'.format(os_user)):
        run('git clone https://github.com/torch/distro.git ~/torch --recursive')
        with cd('/home/{}/torch/'.format(os_user)):
            run('bash install-deps;')
            run('./install.sh -b')
        run('source /home/{}/.bashrc'.format(os_user))
        sudo('touch /home/{}/.ensure_dir/torch_ensured'.format(os_user))

def install_gitlab_cert(os_user, certfile):
    try:
        sudo('mv -f /home/{0}/{1} /etc/ssl/certs/{1}'.format(os_user, certfile))
    except Exception as err:
        print('Failed to install gitlab certificate. {}'.format(str(err)))


text:
  {% if form.non_field_errors %}{% for error in form.non_field_errors %}

        <button type="button" class="close" data-dismiss="alert">×</button>
      {{ error }}
{% endfor %}
  {% endif %}

   {% for field in form %}
    {% if field.is_hidden %}
        {{ field }}
    {% else %}

        {{ field.label }}

            {{ field }}
          {% if field.help_text %}{{ field.help_text|safe }}{% endif %}
          {% if field.errors %}

{% for error in field.errors %}

{{ error }}
{% endfor %}

{% endif %}

    {% endif %}
{% endfor %}