<commit_before>/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.eagle.jpm.spark.history.crawl;

import org.apache.eagle.jpm.spark.entity.*;
import org.apache.eagle.jpm.spark.history.SparkHistoryJobAppConfig;
import org.apache.eagle.jpm.util.*;
import org.apache.eagle.log.base.taggedlog.TaggedLogAPIEntity;
import org.apache.eagle.service.client.EagleServiceClientException;
import org.apache.eagle.service.client.impl.EagleServiceBaseClient;
import org.apache.eagle.service.client.impl.EagleServiceClientImpl;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import org.apache.commons.lang.ArrayUtils;
import org.json.simple.JSONArray;
import org.json.simple.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.*;

public class JHFSparkEventReader {
    private static final Logger LOG = LoggerFactory.getLogger(JHFSparkEventReader.class);

    private static final int FLUSH_LIMIT = 500;
    private long firstTaskLaunchTime;
    private long lastEventTime;

    private Map<String, SparkExecutor> executors;
    private SparkApp app;
    private Map<Integer, SparkJob> jobs;
    private Map<String, SparkStage> stages;
    private Map<Integer, Set<String>> jobStageMap;
    private Map<Long, SparkTask> tasks;
    private EagleServiceClientImpl client;
    private Map<String, Map<Integer, Boolean>> stageTaskStatusMap;

    private List<TaggedLogAPIEntity> createEntities;

    private SparkHistoryJobAppConfig config;
    private Config conf;

    public JHFSparkEventReader(SparkHistoryJobAppConfig config, Map<String, String> baseTags, SparkApplicationInfo info) {
        app = new SparkApp();
        app.setTags(new HashMap<String, String>(baseTags));
        app.setYarnState(info.getState());
        app.setYarnStatus(info.getFinalStatus());
        createEntities = new ArrayList<>();
        jobs = new HashMap<Integer, SparkJob>();
        stages = new HashMap<String, SparkStage>();
        jobStageMap = new HashMap<Integer, Set<String>>();
        tasks = new HashMap<Long, SparkTask>();
        executors = new HashMap<String, SparkExecutor>();
        stageTaskStatusMap = new HashMap<>();
        conf = config.getConfig();
        this.config = config;
        this.initiateClient();
    }

    public SparkApp getApp() {
        return this.app;
    }

    public void read(JSONObject eventObj) {
        String eventType = (String) eventObj.get("Event");
        if (eventType.equalsIgnoreCase(SparkEventType.SparkListenerApplicationStart.toString())) {
            handleAppStarted(eventObj);
        } else if (eventType.equalsIgnoreCase(SparkEventType.SparkListenerEnvironmentUpdate.toString())) {
            handleEnvironmentSet(eventObj);
        } else if (eventType.equalsIgnoreCase(SparkEventType.SparkListenerExecutorAdded.toString())) {
            handleExecutorAdd(eventObj);
        } else if (eventType.equalsIgnoreCase(SparkEventType.SparkListenerBlockManagerAdded.toString())) {
            handleBlockManagerAdd(eventObj);
        } else if (eventType.equalsIgnoreCase(SparkEventType.SparkListenerJobStart.toString())) {
            handleJobStart(eventObj);
        } else if (eventType.equalsIgnoreCase(SparkEventType.SparkListenerStageSubmitted.toString())) {
            handleStageSubmit(eventObj);
        } else if (eventType.equalsIgnoreCase(SparkEventType.SparkListenerTaskStart.toString())) {
            handleTaskStart(eventObj);
        } else if (eventType.equalsIgnoreCase(SparkEventType.SparkListenerTaskEnd.toString())) {
            handleTaskEnd(eventObj);
        } else if (eventType.equalsIgnoreCase(SparkEventType.SparkListenerStageCompleted.toString())) {
            handleStageComplete(eventObj);
        } else if (eventType.equalsIgnoreCase(SparkEventType.SparkListenerJobEnd.toString())) {
            handleJobEnd(eventObj);
        } else if (eventType.equalsIgnoreCase(SparkEventType.SparkListenerExecutorRemoved.toString())) {
            handleExecutorRemoved(eventObj);
        } else if (eventType.equalsIgnoreCase(SparkEventType.SparkListenerApplicationEnd.toString())) {
            handleAppEnd(eventObj);
        } else if (eventType.equalsIgnoreCase(SparkEventType.SparkListenerBlockManagerRemoved.toString())) {
            //nothing to do now
        } else {
            LOG.info("Not registered event type:" + eventType);
        }

    }

    private void handleEnvironmentSet(JSONObject event) {
        app.setConfig(new JobConfig());
        JSONObject sparkProps = (JSONObject) event.get("Spark Properties");

        String[] props = {"spark.yarn.app.id", "spark.executor.memory", "spark.driver.host", "spark.driver.port",
            "spark.driver.memory", "spark.scheduler.pool", "spark.executor.cores", "spark.yarn.am.memory",
            "spark.yarn.am.cores", "spark.yarn.executor.memoryOverhead", "spark.yarn.driver.memoryOverhead", "spark.yarn.am.memoryOverhead", "spark.master"};

        String[] additionalJobConf = null;
        if (conf.hasPath("spark.jobConf.additional.info")) {
            additionalJobConf = conf.getString("spark.jobConf.additional.info").split(",\\s*");
        }

        String[] jobConf = (String[]) ArrayUtils.addAll(additionalJobConf, props);
        for (String prop : jobConf) {
            if (sparkProps.containsKey(prop)) {
                app.getConfig().getConfig().put(prop, (String) sparkProps.get(prop));
            }
        }
    }

    private Object getConfigVal(JobConfig config, String configName, String type) {
        if (config.getConfig().containsKey(configName)) {
            Object val = config.getConfig().get(configName);
            if (type.equalsIgnoreCase(Integer.class.getName())) {
                return Integer.parseInt((String) val);
            } else {
                return val;
            }
        } else {
            if (type.equalsIgnoreCase(Integer.class.getName())) {
                return conf.getInt("spark.defaultVal." + configName);
            } else {
                return conf.getString("spark.defaultVal." + configName);
            }
        }
    }

    private boolean isClientMode(JobConfig config) {
        return config.getConfig().get("spark.master").equalsIgnoreCase("yarn-client");
    }

    private void handleAppStarted(JSONObject event) {
        //need update all entities tag before app start
        List<TaggedLogAPIEntity> entities = new ArrayList<TaggedLogAPIEntity>();
        entities.addAll(this.executors.values());
        entities.add(this.app);

        long appStartTime = JSONUtils.getLong(event, "Timestamp", lastEventTime);
        for (TaggedLogAPIEntity entity : entities) {
            entity.getTags().put(SparkJobTagName.SPARK_APP_ID.toString(), JSONUtils.getString(event, "App ID"));
            entity.getTags().put(SparkJobTagName.SPARK_APP_NAME.toString(), JSONUtils.getString(event, "App Name"));
            // In yarn-client mode, attemptId is not available in the log, so we set attemptId = 1.
            String attemptId = isClientMode(this.app.getConfig()) ? "1" : JSONUtils.getString(event, "App Attempt ID");
            entity.getTags().put(SparkJobTagName.SPARK_APP_ATTEMPT_ID.toString(), attemptId);
            // the second argument of getNormalizeName() is changed to null because the original code contains sensitive text
            // original second argument looks like: this.app.getConfig().getConfig().get("xxx"), "xxx" is the sensitive text
            entity.getTags().put(SparkJobTagName.SPARK_APP_NORM_NAME.toString(), this.getNormalizedName(JSONUtils.getString(event, "App Name"), null));
            entity.getTags().put(SparkJobTagName.SPARK_USER.toString(), JSONUtils.getString(event, "UserPrincipal"));

            entity.setTimestamp(appStartTime);
        }

        this.app.setStartTime(appStartTime);
        this.lastEventTime = appStartTime;
    }

    private void handleExecutorAdd(JSONObject event) {
        String executorID = (String) event.get("Executor ID");
        long executorAddTime = JSONUtils.getLong(event, "Timestamp", lastEventTime);
        this.lastEventTime = executorAddTime;
        SparkExecutor executor = this.initiateExecutor(executorID, executorAddTime);

        JSONObject executorInfo = JSONUtils.getJSONObject(event, "Executor Info");<commit_msg>MINOR: change UserPrinciple into User in JHFSparkEventReader.java<commit_after>/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.eagle.jpm.spark.history.crawl;

import org.apache.eagle.jpm.spark.entity.*;
import org.apache.eagle.jpm.spark.history.SparkHistoryJobAppConfig;
import org.apache.eagle.jpm.util.*;
import org.apache.eagle.log.base.taggedlog.TaggedLogAPIEntity;
import org.apache.eagle.service.client.EagleServiceClientException;
import org.apache.eagle.service.client.impl.EagleServiceBaseClient;
import org.apache.eagle.service.client.impl.EagleServiceClientImpl;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import org.apache.commons.lang.ArrayUtils;
import org.json.simple.JSONArray;
import org.json.simple.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.*;

public class JHFSparkEventReader {
    private static final Logger LOG = LoggerFactory.getLogger(JHFSparkEventReader.class);

    private static final int FLUSH_LIMIT = 500;
    private long firstTaskLaunchTime;
    private long lastEventTime;

    private Map<String, SparkExecutor> executors;
    private SparkApp app;
    private Map<Integer, SparkJob> jobs;
    private Map<String, SparkStage> stages;
    private Map<Integer, Set<String>> jobStageMap;
    private Map<Long, SparkTask> tasks;
    private EagleServiceClientImpl client;
    private Map<String, Map<Integer, Boolean>> stageTaskStatusMap;

    private List<TaggedLogAPIEntity> createEntities;

    private SparkHistoryJobAppConfig config;
    private Config conf;

    public JHFSparkEventReader(SparkHistoryJobAppConfig config, Map<String, String> baseTags, SparkApplicationInfo info) {
        app = new SparkApp();
        app.setTags(new HashMap<String, String>(baseTags));
        app.setYarnState(info.getState());
        app.setYarnStatus(info.getFinalStatus());
        createEntities = new ArrayList<>();
        jobs = new HashMap<Integer, SparkJob>();
        stages = new HashMap<String, SparkStage>();
        jobStageMap = new HashMap<Integer, Set<String>>();
        tasks = new HashMap<Long, SparkTask>();
        executors = new HashMap<String, SparkExecutor>();
        stageTaskStatusMap = new HashMap<>();
        conf = config.getConfig();
        this.config = config;
        this.initiateClient();
    }

    public SparkApp getApp() {
        return this.app;
    }

    public void read(JSONObject eventObj) {
        String eventType = (String) eventObj.get("Event");
        if (eventType.equalsIgnoreCase(SparkEventType.SparkListenerApplicationStart.toString())) {
            handleAppStarted(eventObj);
        } else if (eventType.equalsIgnoreCase(SparkEventType.SparkListenerEnvironmentUpdate.toString())) {
            handleEnvironmentSet(eventObj);
        } else if (eventType.equalsIgnoreCase(SparkEventType.SparkListenerExecutorAdded.toString())) {
            handleExecutorAdd(eventObj);
        } else if (eventType.equalsIgnoreCase(SparkEventType.SparkListenerBlockManagerAdded.toString())) {
            handleBlockManagerAdd(eventObj);
        } else if (eventType.equalsIgnoreCase(SparkEventType.SparkListenerJobStart.toString())) {
            handleJobStart(eventObj);
        } else if (eventType.equalsIgnoreCase(SparkEventType.SparkListenerStageSubmitted.toString())) {
            handleStageSubmit(eventObj);
        } else if (eventType.equalsIgnoreCase(SparkEventType.SparkListenerTaskStart.toString())) {
            handleTaskStart(eventObj);
        } else if (eventType.equalsIgnoreCase(SparkEventType.SparkListenerTaskEnd.toString())) {
            handleTaskEnd(eventObj);
        } else if (eventType.equalsIgnoreCase(SparkEventType.SparkListenerStageCompleted.toString())) {
            handleStageComplete(eventObj);
        } else if (eventType.equalsIgnoreCase(SparkEventType.SparkListenerJobEnd.toString())) {
            handleJobEnd(eventObj);
        } else if (eventType.equalsIgnoreCase(SparkEventType.SparkListenerExecutorRemoved.toString())) {
            handleExecutorRemoved(eventObj);
        } else if (eventType.equalsIgnoreCase(SparkEventType.SparkListenerApplicationEnd.toString())) {
            handleAppEnd(eventObj);
        } else if (eventType.equalsIgnoreCase(SparkEventType.SparkListenerBlockManagerRemoved.toString())) {
            //nothing to do now
        } else {
            LOG.info("Not registered event type:" + eventType);
        }

    }

    private void handleEnvironmentSet(JSONObject event) {
        app.setConfig(new JobConfig());
        JSONObject sparkProps = (JSONObject) event.get("Spark Properties");

        String[] props = {"spark.yarn.app.id", "spark.executor.memory", "spark.driver.host", "spark.driver.port",
            "spark.driver.memory", "spark.scheduler.pool", "spark.executor.cores", "spark.yarn.am.memory",
            "spark.yarn.am.cores", "spark.yarn.executor.memoryOverhead", "spark.yarn.driver.memoryOverhead", "spark.yarn.am.memoryOverhead", "spark.master"};

        String[] additionalJobConf = null;
        if (conf.hasPath("spark.jobConf.additional.info")) {
            additionalJobConf = conf.getString("spark.jobConf.additional.info").split(",\\s*");
        }

        String[] jobConf = (String[]) ArrayUtils.addAll(additionalJobConf, props);
        for (String prop : jobConf) {
            if (sparkProps.containsKey(prop)) {
                app.getConfig().getConfig().put(prop, (String) sparkProps.get(prop));
            }
        }
    }

    private Object getConfigVal(JobConfig config, String configName, String type) {
        if (config.getConfig().containsKey(configName)) {
            Object val = config.getConfig().get(configName);
            if (type.equalsIgnoreCase(Integer.class.getName())) {
                return Integer.parseInt((String) val);
            } else {
                return val;
            }
        } else {
            if (type.equalsIgnoreCase(Integer.class.getName())) {
                return conf.getInt("spark.defaultVal." + configName);
            } else {
                return conf.getString("spark.defaultVal." + configName);
            }
        }
    }

    private boolean isClientMode(JobConfig config) {
        return config.getConfig().get("spark.master").equalsIgnoreCase("yarn-client");
    }

    private void handleAppStarted(JSONObject event) {
        //need update all entities tag before app start
        List<TaggedLogAPIEntity> entities = new ArrayList<TaggedLogAPIEntity>();
        entities.addAll(this.executors.values());
        entities.add(this.app);

        long appStartTime = JSONUtils.getLong(event, "Timestamp", lastEventTime);
        for (TaggedLogAPIEntity entity : entities) {
            entity.getTags().put(SparkJobTagName.SPARK_APP_ID.toString(), JSONUtils.getString(event, "App ID"));
            entity.getTags().put(SparkJobTagName.SPARK_APP_NAME.toString(), JSONUtils.getString(event, "App Name"));
            // In yarn-client mode, attemptId is not available in the log, so we set attemptId = 1.
            String attemptId = isClientMode(this.app.getConfig()) ? "1" : JSONUtils.getString(event, "App Attempt ID");
            entity.getTags().put(SparkJobTagName.SPARK_APP_ATTEMPT_ID.toString(), attemptId);
            // the second argument of getNormalizeName() is changed to null because the original code contains sensitive text
            // original second argument looks like: this.app.getConfig().getConfig().get("xxx"), "xxx" is the sensitive text
            entity.getTags().put(SparkJobTagName.SPARK_APP_NORM_NAME.toString(), this.getNormalizedName(JSONUtils.getString(event, "App Name"), null));
            entity.getTags().put(SparkJobTagName.SPARK_USER.toString(), JSONUtils.getString(event, "User"));

            entity.setTimestamp(appStartTime);
        }

        this.app.setStartTime(appStartTime);
        this.lastEventTime = appStartTime;
    }

    private void handleExecutorAdd(JSONObject event) {
        String executorID = (String) event.get("Executor ID");
        long executorAddTime = JSONUtils.getLong(event, "Timestamp", lastEventTime);
        this.lastEventTime = executorAddTime;
        SparkExecutor executor = this.initiateExecutor(executorID, executorAddTime);

        JSONObject executorInfo = JSONUtils.getJSONObject(event, "Executor Info");
{'fraction_non_alphanumeric': 0.1079542344117808, 'fraction_numerical': 0.0012043575847156074, 'mean_word_length': 3.421103581800581, 'pattern_counts': {'":': 0, '<': 45, '<?xml version=': 0, '>': 45, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7327512', 'n_tokens_mistral': 5052, 'n_tokens_neox': 4896, 'n_words': 1077}