<commit_before>/*
 *  Copyright (c) 2015, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 *  WSO2 Inc. licenses this file to you under the Apache License,
 *  Version 2.0 (the "License"); you may not use this file except
 *  in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *  http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing,
 *  software distributed under the License is distributed on an
 *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 *  KIND, either express or implied.  See the License for the
 *  specific language governing permissions and limitations
 *  under the License.
 *
 */
package org.wso2.carbon.analytics.spark.core.internal;

import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.IMap;
import com.hazelcast.core.Member;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.deploy.master.LeaderElectable;
import org.apache.spark.deploy.master.Master;
import org.apache.spark.deploy.worker.Worker;
import org.apache.spark.serializer.KryoSerializer;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.jdbc.carbon.AnalyticsJDBCRelationProvider;
import org.apache.spark.sql.jdbc.carbon.DialectRegister;
import org.apache.spark.util.Utils;
import org.wso2.carbon.analytics.dataservice.core.AnalyticsServiceHolder;
import org.wso2.carbon.analytics.dataservice.core.clustering.AnalyticsClusterException;
import org.wso2.carbon.analytics.dataservice.core.clustering.AnalyticsClusterManager;
import org.wso2.carbon.analytics.dataservice.core.clustering.GroupEventListener;
import org.wso2.carbon.analytics.datasource.commons.exception.AnalyticsException;
import org.wso2.carbon.analytics.datasource.core.util.GenericUtils;
import org.wso2.carbon.analytics.spark.core.AnalyticsExecutionCall;
import org.wso2.carbon.analytics.spark.core.deploy.AnalyticsRecoveryModeFactory;
import org.wso2.carbon.analytics.spark.core.deploy.CheckElectedLeaderExecutionCall;
import org.wso2.carbon.analytics.spark.core.deploy.ElectLeaderExecutionCall;
import org.wso2.carbon.analytics.spark.core.deploy.InitClientExecutionCall;
import org.wso2.carbon.analytics.spark.core.deploy.StartWorkerExecutionCall;
import org.wso2.carbon.analytics.spark.core.exception.AnalyticsExecutionException;
import org.wso2.carbon.analytics.spark.core.exception.AnalyticsUDFException;
import org.wso2.carbon.analytics.spark.core.sources.AnalyticsRelationProvider;
import org.wso2.carbon.analytics.spark.core.udf.AnalyticsUDFsRegister;
import org.wso2.carbon.analytics.spark.core.udf.config.UDFConfiguration;
import org.wso2.carbon.analytics.spark.core.util.AnalyticsCommonUtils;
import org.wso2.carbon.analytics.spark.core.util.AnalyticsConstants;
import org.wso2.carbon.analytics.spark.core.util.AnalyticsQueryResult;
import org.wso2.carbon.analytics.spark.core.util.SparkTableNamesHolder;
import org.wso2.carbon.analytics.spark.utils.ComputeClasspath;
import org.wso2.carbon.utils.CarbonUtils;
import scala.None$;
import scala.Option;
import scala.Tuple2;

import javax.xml.bind.JAXBContext;
import javax.xml.bind.JAXBException;
import javax.xml.bind.Unmarshaller;
import java.io.File;
import java.io.IOException;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * This class represents the analytics query execution context.
 */
public class SparkAnalyticsExecutor implements GroupEventListener {

    private static final String CLUSTER_GROUP_NAME = "CARBON_ANALYTICS_EXECUTION";

    private static final String DEFAULT_SPARK_APP_NAME = "DefaultCarbonAnalyticsApp";

    private static final Log log = LogFactory.getLog(SparkAnalyticsExecutor.class);

    private String sparkMaster;

    private SparkConf sparkConf;

    private SQLContext sqlCtx;

    private String myHost;

    private int portOffset = 0;

    private int workerCount = 1;

    private SparkTableNamesHolder sparkTableNamesHolder;

    private UDFConfiguration udfConfiguration;

    private int redundantMasterCount = 1;

    private Set<LeaderElectable> leaderElectable = new HashSet<>();

    private AnalyticsClusterManager acm;

    private boolean masterActive = false;

    private boolean workerActive = false;

    private boolean clientActive = false;

    private boolean electedLeader = false;

    private Map<String, String> shorthandStringsMap = new HashMap<>();


    public SparkAnalyticsExecutor(String myHost, int portOffset) throws AnalyticsException {
        this.myHost = myHost;
        this.portOffset = portOffset;
        this.udfConfiguration = this.loadUDFConfiguration();

        this.acm = AnalyticsServiceHolder.getAnalyticsClusterManager();

        // sends this host name and base master port to initialize the spark conf
        String propsFile = GenericUtils.getAnalyticsConfDirectory() + File.separator +
                           "analytics" + File.separator +
                           AnalyticsConstants.SPARK_CONF_DIR + File.separator +
                           AnalyticsConstants.SPARK_DEFAULTS_FILE;
        if (!new File(propsFile).exists()) {
            throw new AnalyticsExecutionException("spark-defaults.conf file does not exists in path "
                                                  + propsFile);
        }
        this.sparkConf = initializeSparkConf(this.portOffset, propsFile);

        this.sparkMaster = getStringFromSparkConf(AnalyticsConstants.CARBON_SPARK_MASTER, "local");
        this.redundantMasterCount = this.sparkConf.getInt(AnalyticsConstants.CARBON_SPARK_MASTER_COUNT, 1);

        this.registerShorthandStrings();
        this.registerSqlDialects();
    }

    private void registerSqlDialects() {
        DialectRegister.register();
    }

    /**
     * @throws AnalyticsClusterException
     */
    public void initializeSparkServer() throws AnalyticsException {
        if (isLocalMode()) {
            if (isClusterMode()) {
                log.info("Using Carbon clustering for Spark");

                //initialize the spark table names holder in the clustered mode
                if (this.sparkTableNamesHolder == null) {
                    this.sparkTableNamesHolder = new SparkTableNamesHolder(true);
                }

                if (acm.isClusteringEnabled()) {
                    runClusteredSetupLogic();
                } else {
                    throw new AnalyticsClusterException("Spark started in the cluster mode without " +
                                                        "enabling Carbon Clustering");
                }
            } else {
                log.info("Starting SPARK in the LOCAL mode...");

                //initialize the spark table names holder in the in-memory mode
                if (this.sparkTableNamesHolder == null) {
                    this.sparkTableNamesHolder = new SparkTableNamesHolder(false);
                }

                this.initializeClient(true);
            }
        } else if (isClientMode()) {
            log.info("Client mode enabled for Spark");
            log.info("Starting SPARK CLIENT pointing to an external Spark Cluster");

            //initialize the spark table names holder in the in-memory mode
            if (this.sparkTableNamesHolder == null) {
                this.sparkTableNamesHolder = new SparkTableNamesHolder(false);
            }

            this.initializeClient(false);
        } else {
            throw new AnalyticsClusterException("Unknown mode for Spark Server start up: "
                                                + this.sparkMaster);
        }
    }

    private void cleanupMasterMap(Set<Member> members, Map<String, Object> masterMap) {
        Iterator<Map.Entry<String, Object>> itr = masterMap.entrySet().iterator();
        while (itr.hasNext()) {
            if (!members.contains(itr.next().getValue())) {
                itr.remove();
            }
        }
    }

    private void runClusteredSetupLogic() throws AnalyticsClusterException {

        //port offsetted master name
        String thisMasterUrl = "spark://" + this.myHost + ":" + this.sparkConf.
                get(AnalyticsConstants.SPARK_MASTER_PORT);

        HazelcastInstance hz = AnalyticsServiceHolder.getHazelcastInstance();
        Map<String, Object> masterMap = hz.getMap(AnalyticsConstants.SPARK_MASTER_MAP);

        this.cleanupMasterMap(hz.getCluster().getMembers(), masterMap);

        acm.joinGroup(CLUSTER_GROUP_NAME, this);
        log.info("Member joined the cluster");

        Set<String> masterUrls = masterMap.keySet();

        //start master logic
        if (masterUrls.contains(thisMasterUrl) || masterMap.size() < this.redundantMasterCount) {
            log.info("Masters available are less than the redundant master count or " +<commit_msg>fixing DAS-262 Spark Connection refused exception can be seen after restart the sever in cluster node - w/o tests 2<commit_after>/*
 *  Copyright (c) 2015, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 *  WSO2 Inc. licenses this file to you under the Apache License,
 *  Version 2.0 (the "License"); you may not use this file except
 *  in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *  http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing,
 *  software distributed under the License is distributed on an
 *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 *  KIND, either express or implied.  See the License for the
 *  specific language governing permissions and limitations
 *  under the License.
 *
 */
package org.wso2.carbon.analytics.spark.core.internal;

import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.IMap;
import com.hazelcast.core.Member;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.deploy.master.LeaderElectable;
import org.apache.spark.deploy.master.Master;
import org.apache.spark.deploy.worker.Worker;
import org.apache.spark.serializer.KryoSerializer;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.jdbc.carbon.AnalyticsJDBCRelationProvider;
import org.apache.spark.sql.jdbc.carbon.DialectRegister;
import org.apache.spark.util.Utils;
import org.wso2.carbon.analytics.dataservice.core.AnalyticsServiceHolder;
import org.wso2.carbon.analytics.dataservice.core.clustering.AnalyticsClusterException;
import org.wso2.carbon.analytics.dataservice.core.clustering.AnalyticsClusterManager;
import org.wso2.carbon.analytics.dataservice.core.clustering.GroupEventListener;
import org.wso2.carbon.analytics.datasource.commons.exception.AnalyticsException;
import org.wso2.carbon.analytics.datasource.core.util.GenericUtils;
import org.wso2.carbon.analytics.spark.core.AnalyticsExecutionCall;
import org.wso2.carbon.analytics.spark.core.deploy.AnalyticsRecoveryModeFactory;
import org.wso2.carbon.analytics.spark.core.deploy.CheckElectedLeaderExecutionCall;
import org.wso2.carbon.analytics.spark.core.deploy.ElectLeaderExecutionCall;
import org.wso2.carbon.analytics.spark.core.deploy.InitClientExecutionCall;
import org.wso2.carbon.analytics.spark.core.deploy.StartWorkerExecutionCall;
import org.wso2.carbon.analytics.spark.core.exception.AnalyticsExecutionException;
import org.wso2.carbon.analytics.spark.core.exception.AnalyticsUDFException;
import org.wso2.carbon.analytics.spark.core.sources.AnalyticsRelationProvider;
import org.wso2.carbon.analytics.spark.core.udf.AnalyticsUDFsRegister;
import org.wso2.carbon.analytics.spark.core.udf.config.UDFConfiguration;
import org.wso2.carbon.analytics.spark.core.util.AnalyticsCommonUtils;
import org.wso2.carbon.analytics.spark.core.util.AnalyticsConstants;
import org.wso2.carbon.analytics.spark.core.util.AnalyticsQueryResult;
import org.wso2.carbon.analytics.spark.core.util.SparkTableNamesHolder;
import org.wso2.carbon.analytics.spark.utils.ComputeClasspath;
import org.wso2.carbon.utils.CarbonUtils;
import scala.None$;
import scala.Option;
import scala.Tuple2;

import javax.xml.bind.JAXBContext;
import javax.xml.bind.JAXBException;
import javax.xml.bind.Unmarshaller;
import java.io.File;
import java.io.IOException;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * This class represents the analytics query execution context.
 */
public class SparkAnalyticsExecutor implements GroupEventListener {

    private static final String CLUSTER_GROUP_NAME = "CARBON_ANALYTICS_EXECUTION";

    private static final String DEFAULT_SPARK_APP_NAME = "DefaultCarbonAnalyticsApp";

    private static final Log log = LogFactory.getLog(SparkAnalyticsExecutor.class);

    private String sparkMaster;

    private SparkConf sparkConf;

    private SQLContext sqlCtx;

    private String myHost;

    private int portOffset = 0;

    private int workerCount = 1;

    private SparkTableNamesHolder sparkTableNamesHolder;

    private UDFConfiguration udfConfiguration;

    private int redundantMasterCount = 1;

    private Set<LeaderElectable> leaderElectable = new HashSet<>();

    private AnalyticsClusterManager acm;

    private boolean masterActive = false;

    private boolean workerActive = false;

    private boolean clientActive = false;

    private boolean electedLeader = false;

    private Map<String, String> shorthandStringsMap = new HashMap<>();


    public SparkAnalyticsExecutor(String myHost, int portOffset) throws AnalyticsException {
        this.myHost = myHost;
        this.portOffset = portOffset;
        this.udfConfiguration = this.loadUDFConfiguration();

        this.acm = AnalyticsServiceHolder.getAnalyticsClusterManager();

        // sends this host name and base master port to initialize the spark conf
        String propsFile = GenericUtils.getAnalyticsConfDirectory() + File.separator +
                           "analytics" + File.separator +
                           AnalyticsConstants.SPARK_CONF_DIR + File.separator +
                           AnalyticsConstants.SPARK_DEFAULTS_FILE;
        if (!new File(propsFile).exists()) {
            throw new AnalyticsExecutionException("spark-defaults.conf file does not exists in path "
                                                  + propsFile);
        }
        this.sparkConf = initializeSparkConf(this.portOffset, propsFile);

        this.sparkMaster = getStringFromSparkConf(AnalyticsConstants.CARBON_SPARK_MASTER, "local");
        this.redundantMasterCount = this.sparkConf.getInt(AnalyticsConstants.CARBON_SPARK_MASTER_COUNT, 1);

        this.registerShorthandStrings();
        this.registerSqlDialects();
    }

    private void registerSqlDialects() {
        DialectRegister.register();
    }

    /**
     * @throws AnalyticsClusterException
     */
    public void initializeSparkServer() throws AnalyticsException {
        if (isLocalMode()) {
            if (isClusterMode()) {
                log.info("Using Carbon clustering for Spark");

                //initialize the spark table names holder in the clustered mode
                if (this.sparkTableNamesHolder == null) {
                    this.sparkTableNamesHolder = new SparkTableNamesHolder(true);
                }

                if (acm.isClusteringEnabled()) {
                    runClusteredSetupLogic();
                } else {
                    throw new AnalyticsClusterException("Spark started in the cluster mode without " +
                                                        "enabling Carbon Clustering");
                }
            } else {
                log.info("Starting SPARK in the LOCAL mode...");

                //initialize the spark table names holder in the in-memory mode
                if (this.sparkTableNamesHolder == null) {
                    this.sparkTableNamesHolder = new SparkTableNamesHolder(false);
                }

                this.initializeClient(true);
            }
        } else if (isClientMode()) {
            log.info("Client mode enabled for Spark");
            log.info("Starting SPARK CLIENT pointing to an external Spark Cluster");

            //initialize the spark table names holder in the in-memory mode
            if (this.sparkTableNamesHolder == null) {
                this.sparkTableNamesHolder = new SparkTableNamesHolder(false);
            }

            this.initializeClient(false);
        } else {
            throw new AnalyticsClusterException("Unknown mode for Spark Server start up: "
                                                + this.sparkMaster);
        }
    }

    private void cleanupMasterMap(Member myself, Set<Member> members, Map<String, Object> masterMap) {
        Iterator<Map.Entry<String, Object>> itr = masterMap.entrySet().iterator();
        Map.Entry<String, Object> currentEntry;
        List<String> removeIds = new ArrayList<>();
        while (itr.hasNext()) {
            currentEntry = itr.next();
            if (!members.contains(currentEntry.getValue()) || (currentEntry.getValue().equals(myself))) {
                removeIds.add(currentEntry.getKey());
            }
        }
        for (String key : removeIds) {
            masterMap.remove(key);
        }
    }

    private void runClusteredSetupLogic() throws AnalyticsClusterException {

        //port offsetted master name
        String thisMasterUrl = "spark://" + this.myHost + ":" + this.sparkConf.
                get(AnalyticsConstants.SPARK_MASTER_PORT);

        HazelcastInstance hz = AnalyticsServiceHolder.getHazelcastInstance();
        Map<String, Object> masterMap = hz.getMap(AnalyticsConstants.SPARK_MASTER_MAP);

        this.cleanupMasterMap(hz.getCluster().getLocalMember(), hz.getCluster().getMembers(), masterMap);

        acm.joinGroup(CLUSTER_GROUP_NAME, this);
        log.info("Member joined the cluster");

        Set<String> masterUrls = masterMap.keySet();

        //start master logic
        if (masterUrls.contains(thisMasterUrl) || masterMap.size() < this.redundantMasterCount) {
            log.info("Masters available are less than the redundant master count or " +
{'fraction_non_alphanumeric': 0.0908039936941671, 'fraction_numerical': 0.00441408302679979, 'mean_word_length': 3.160690861390468, 'pattern_counts': {'":': 2, '<': 28, '<?xml version=': 0, '>': 26, 'https://': 0, 'lorem ipsum': 0, 'www.': 4, 'xml': 6}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5534191', 'n_tokens_mistral': 5531, 'n_tokens_neox': 5130, 'n_words': 1150}