<commit_before>/*
 * Copyright (c) 2012 - 2017 Splice Machine, Inc.
 *
 * This file is part of Splice Machine.
 * Splice Machine is free software: you can redistribute it and/or modify it under the terms of the
 * GNU Affero General Public License as published by the Free Software Foundation, either
 * version 3, or (at your option) any later version.
 * Splice Machine is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
 * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
 * See the GNU Affero General Public License for more details.
 * You should have received a copy of the GNU Affero General Public License along with Splice Machine.
 * If not, see <http://www.gnu.org/licenses/>.
 *
 */

package com.splicemachine.olap;

import com.splicemachine.EngineDriver;
import com.splicemachine.access.HConfiguration;
import com.splicemachine.access.api.SConfiguration;
import com.splicemachine.access.configuration.HBaseConfiguration;
import com.splicemachine.hbase.ZkUtils;
import com.splicemachine.si.impl.driver.SIDriver;
import org.apache.commons.lang3.SystemUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.zookeeper.RecoverableZooKeeper;
import org.apache.hadoop.mapreduce.MRJobConfig;
import org.apache.hadoop.util.StringUtils;
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.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.util.ConverterUtils;
import org.apache.hadoop.yarn.util.Records;
import org.apache.log4j.Logger;
import org.apache.spark.util.ShutdownHookManager;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import scala.runtime.AbstractFunction0;
import scala.runtime.BoxedUnit;

import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;

import static org.apache.zookeeper.KeeperException.Code.NODEEXISTS;
import static org.apache.zookeeper.KeeperException.Code.NONODE;

/**
 * Created by dgomezferro on 29/08/2017.
 */
public class OlapServerSubmitter implements Runnable {

    // Staging directory for any temporary jars or files
    private static final String SPLICE_STAGING = ".spliceStaging";

    private static final String KEYTAB_KEY = "splice.spark.yarn.keytab";

    // Staging directory is private! -> rwx--------
    private static final FsPermission STAGING_DIR_PERMISSION = FsPermission.createImmutable(Short.parseShort("700", 8));

    private static final Logger LOG = Logger.getLogger(OlapServerSubmitter.class);
    private final ServerName serverName;
    private volatile boolean stop = false;
    private CountDownLatch stopLatch = new CountDownLatch(1);
    private Path appStagingBaseDir;
    private String amKeytabFileName = null;

    private Configuration conf;

    public OlapServerSubmitter(ServerName serverName) {
        this.serverName = serverName;
    }

    public void run() {

        try {
            // Create yarnClient
            conf = HConfiguration.unwrapDelegate();

            YarnClient yarnClient = YarnClient.createYarnClient();
            yarnClient.init(conf);
            yarnClient.start();

            SConfiguration sconf = HConfiguration.getConfiguration();

            int maxAttempts = sconf.getOlapServerSubmitAttempts();
            int memory = sconf.getOlapServerMemory();
            int memoryOverhead = sconf.getOlapServerMemoryOverhead();
            int cpuCores = sconf.getOlapVirtualCores();
            int olapPort = sconf.getOlapServerBindPort();
            String stagingDir = sconf.getOlapServerStagingDirectory();
            if (stagingDir != null) {
                this.appStagingBaseDir = new Path(stagingDir);
            } else {
                this.appStagingBaseDir = FileSystem.get(conf).getHomeDirectory();
            }

            for (int i = 0; i<maxAttempts; ++i) {
                // Clear ZooKeeper path
                clearZookeeper(ZkUtils.getRecoverableZooKeeper(), serverName);

                // Create application via yarnClient
                YarnClientApplication app = yarnClient.createApplication();

                GetNewApplicationResponse newAppResponse = app.getNewApplicationResponse();
                ApplicationId appId = newAppResponse.getApplicationId();

                Path appStagingDirPath = new Path(appStagingBaseDir, getAppStagingDir(appId));
                // Create staging dir
                FileSystem fs = appStagingDirPath.getFileSystem(conf);
                FileSystem.mkdirs(fs, appStagingDirPath, new FsPermission(STAGING_DIR_PERMISSION));

                String keytab = System.getProperty(KEYTAB_KEY);
                Map<String, LocalResource> localResources = new HashMap<>();
                if (keytab != null) {
                    LOG.info(KEYTAB_KEY + " is set, adding it to local resources");
                    String trimmedPath = keytab.trim();
                    URI localURI = resolveURI(trimmedPath);
                    Path srcPath = getQualifiedLocalPath(localURI, conf);

                    String name = srcPath.getName();
                    FileSystem srcFs = srcPath.getFileSystem(conf);
                    FileUtil.copy(srcFs, srcPath, fs, appStagingDirPath, false, conf);
                    amKeytabFileName = name;

                    Path keytabPath = new Path(appStagingDirPath, name);
                    FileStatus destStatus = fs.getFileStatus(keytabPath);
                    LocalResource amJarRsrc = Records.newRecord(LocalResource.class);
                    amJarRsrc.setType(LocalResourceType.FILE);
                    amJarRsrc.setVisibility(LocalResourceVisibility.PRIVATE);
                    amJarRsrc.setResource(ConverterUtils.getYarnUrlFromPath(keytabPath));
                    amJarRsrc.setTimestamp(destStatus.getModificationTime());
                    amJarRsrc.setSize(destStatus.getLen());
                    
                    localResources.put(name, amJarRsrc);
                }

                // Set up the container launch context for the application master
                ContainerLaunchContext amContainer =
                        Records.newRecord(ContainerLaunchContext.class);
                amContainer.setCommands(
                        Collections.singletonList(prepareCommands(
                                "$JAVA_HOME/bin/java",
                                " -Xmx" + memory + "M " + outOfMemoryErrorArgument() + " " +
                                        OlapServerMaster.class.getCanonicalName() +
                                        " " + serverName.toString() + " " + olapPort +
                                        " 1>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/stdout" +
                                        " 2>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/stderr"
                        ))
                );

                // Setup CLASSPATH for ApplicationMaster
                Map<String, String> appMasterEnv = new HashMap<String, String>();
                setupAppMasterEnv(appMasterEnv, conf);
                amContainer.setEnvironment(appMasterEnv);
                amContainer.setLocalResources(localResources);


                // Set up resource type requirements for ApplicationMaster
                Resource capability = Records.newRecord(Resource.class);
                capability.setMemory(memory + memoryOverhead);
                capability.setVirtualCores(cpuCores);

                // Finally, set-up ApplicationSubmissionContext for the application
                ApplicationSubmissionContext appContext =
                        app.getApplicationSubmissionContext();
                appContext.setApplicationName("OlapServer"); // application name
                appContext.setAMContainerSpec(amContainer);
                appContext.setResource(capability);
                appContext.setQueue("default"); // queue
                appContext.setMaxAppAttempts(1);<commit_msg>SPLICE-2067: Fix for Client cannot authenticate via:[TOKEN, KERBEROS] (#1516)<commit_after>/*
 * Copyright (c) 2012 - 2017 Splice Machine, Inc.
 *
 * This file is part of Splice Machine.
 * Splice Machine is free software: you can redistribute it and/or modify it under the terms of the
 * GNU Affero General Public License as published by the Free Software Foundation, either
 * version 3, or (at your option) any later version.
 * Splice Machine is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
 * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
 * See the GNU Affero General Public License for more details.
 * You should have received a copy of the GNU Affero General Public License along with Splice Machine.
 * If not, see <http://www.gnu.org/licenses/>.
 *
 */

package com.splicemachine.olap;

import com.splicemachine.EngineDriver;
import com.splicemachine.access.HConfiguration;
import com.splicemachine.access.api.SConfiguration;
import com.splicemachine.access.configuration.HBaseConfiguration;
import com.splicemachine.hbase.ZkUtils;
import com.splicemachine.si.impl.driver.SIDriver;
import org.apache.commons.lang3.SystemUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.zookeeper.RecoverableZooKeeper;
import org.apache.hadoop.io.DataOutputBuffer;
import org.apache.hadoop.mapreduce.MRJobConfig;
import org.apache.hadoop.security.Credentials;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.util.StringUtils;
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.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.util.ConverterUtils;
import org.apache.hadoop.yarn.util.Records;
import org.apache.log4j.Logger;
import org.apache.spark.util.ShutdownHookManager;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import scala.runtime.AbstractFunction0;
import scala.runtime.BoxedUnit;

import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;

import static org.apache.zookeeper.KeeperException.Code.NODEEXISTS;
import static org.apache.zookeeper.KeeperException.Code.NONODE;

/**
 * Created by dgomezferro on 29/08/2017.
 */
public class OlapServerSubmitter implements Runnable {

    // Staging directory for any temporary jars or files
    private static final String SPLICE_STAGING = ".spliceStaging";

    private static final String KEYTAB_KEY = "splice.spark.yarn.keytab";

    // Staging directory is private! -> rwx--------
    private static final FsPermission STAGING_DIR_PERMISSION = FsPermission.createImmutable(Short.parseShort("700", 8));

    private static final Logger LOG = Logger.getLogger(OlapServerSubmitter.class);
    private final ServerName serverName;
    private volatile boolean stop = false;
    private CountDownLatch stopLatch = new CountDownLatch(1);
    private Path appStagingBaseDir;
    private String amKeytabFileName = null;

    private Configuration conf;

    public OlapServerSubmitter(ServerName serverName) {
        this.serverName = serverName;
    }

    public void run() {

        try {
            // Create yarnClient
            conf = HConfiguration.unwrapDelegate();

            YarnClient yarnClient = YarnClient.createYarnClient();
            yarnClient.init(conf);
            yarnClient.start();

            SConfiguration sconf = HConfiguration.getConfiguration();

            int maxAttempts = sconf.getOlapServerSubmitAttempts();
            int memory = sconf.getOlapServerMemory();
            int memoryOverhead = sconf.getOlapServerMemoryOverhead();
            int cpuCores = sconf.getOlapVirtualCores();
            int olapPort = sconf.getOlapServerBindPort();
            String stagingDir = sconf.getOlapServerStagingDirectory();
            if (stagingDir != null) {
                this.appStagingBaseDir = new Path(stagingDir);
            } else {
                this.appStagingBaseDir = FileSystem.get(conf).getHomeDirectory();
            }

            for (int i = 0; i<maxAttempts; ++i) {
                // Clear ZooKeeper path
                clearZookeeper(ZkUtils.getRecoverableZooKeeper(), serverName);

                // Create application via yarnClient
                YarnClientApplication app = yarnClient.createApplication();

                GetNewApplicationResponse newAppResponse = app.getNewApplicationResponse();
                ApplicationId appId = newAppResponse.getApplicationId();

                Path appStagingDirPath = new Path(appStagingBaseDir, getAppStagingDir(appId));
                // Create staging dir
                FileSystem fs = appStagingDirPath.getFileSystem(conf);
                FileSystem.mkdirs(fs, appStagingDirPath, new FsPermission(STAGING_DIR_PERMISSION));

                String keytab = System.getProperty(KEYTAB_KEY);
                Map<String, LocalResource> localResources = new HashMap<>();
                if (keytab != null) {
                    LOG.info(KEYTAB_KEY + " is set, adding it to local resources");
                    String trimmedPath = keytab.trim();
                    URI localURI = resolveURI(trimmedPath);
                    Path srcPath = getQualifiedLocalPath(localURI, conf);

                    String name = srcPath.getName();
                    FileSystem srcFs = srcPath.getFileSystem(conf);
                    FileUtil.copy(srcFs, srcPath, fs, appStagingDirPath, false, conf);
                    amKeytabFileName = name;

                    Path keytabPath = new Path(appStagingDirPath, name);
                    FileStatus destStatus = fs.getFileStatus(keytabPath);
                    LocalResource amJarRsrc = Records.newRecord(LocalResource.class);
                    amJarRsrc.setType(LocalResourceType.FILE);
                    amJarRsrc.setVisibility(LocalResourceVisibility.PRIVATE);
                    amJarRsrc.setResource(ConverterUtils.getYarnUrlFromPath(keytabPath));
                    amJarRsrc.setTimestamp(destStatus.getModificationTime());
                    amJarRsrc.setSize(destStatus.getLen());
                    
                    localResources.put(name, amJarRsrc);
                }

                // Set up the container launch context for the application master
                ContainerLaunchContext amContainer =
                        Records.newRecord(ContainerLaunchContext.class);
                amContainer.setCommands(
                        Collections.singletonList(prepareCommands(
                                "$JAVA_HOME/bin/java",
                                " -Xmx" + memory + "M " + outOfMemoryErrorArgument() + " " +
                                        OlapServerMaster.class.getCanonicalName() +
                                        " " + serverName.toString() + " " + olapPort +
                                        " 1>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/stdout" +
                                        " 2>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/stderr"
                        ))
                );

                // Setup CLASSPATH for ApplicationMaster
                Map<String, String> appMasterEnv = new HashMap<String, String>();
                setupAppMasterEnv(appMasterEnv, conf);
                amContainer.setEnvironment(appMasterEnv);
                amContainer.setLocalResources(localResources);

                // Setup security tokens
                if (UserGroupInformation.isSecurityEnabled()) {
                    Credentials credentials = new Credentials();
                    String tokenRenewer = conf.get(YarnConfiguration.RM_PRINCIPAL);
                    if (tokenRenewer == null || tokenRenewer.length() == 0) {
                        throw new IOException(
                                "Can't get Master Kerberos principal for the RM to use as renewer");
                    }

                    final Token<?> tokens[] = fs.addDelegationTokens(tokenRenewer, credentials);
                    if (tokens != null) {
                        for (Token<?> token : tokens) {
                            LOG.info("Got dt for " + fs.getUri() + "; " + token);
                        }
                    }
                    DataOutputBuffer dob = new DataOutputBuffer();
                    credentials.writeTokenStorageToStream(dob);
                    ByteBuffer fsTokens = ByteBuffer.wrap(dob.getData(), 0, dob.getLength());
                    amContainer.setTokens(fsTokens);
                }

                // Set up resource type requirements for ApplicationMaster
                Resource capability = Records.newRecord(Resource.class);
                capability.setMemory(memory + memoryOverhead);
                capability.setVirtualCores(cpuCores);

                // Finally, set-up ApplicationSubmissionContext for the application
                ApplicationSubmissionContext appContext =
                        app.getApplicationSubmissionContext();
                appContext.setApplicationName("OlapServer"); // application name
                appContext.setAMContainerSpec(amContainer);
                appContext.setResource(capability);
                appContext.setQueue("default"); // queue
                appContext.setMaxAppAttempts(1);
{'fraction_non_alphanumeric': 0.08587593453222873, 'fraction_numerical': 0.0034350373812891493, 'mean_word_length': 2.5332857397822597, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4265548', 'n_tokens_mistral': 5332, 'n_tokens_neox': 5052, 'n_words': 1100}