<commit_before>package com.ebay.myriad.scheduler;

import com.ebay.myriad.configuration.MyriadConfiguration;
import com.ebay.myriad.executor.NMTaskConfig;
import com.ebay.myriad.state.NodeTask;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import com.google.gson.Gson;
import com.google.protobuf.ByteString;
import org.apache.mesos.Protos;
import org.apache.mesos.Protos.*;
import org.apache.mesos.Protos.CommandInfo.URI;
import org.apache.mesos.Protos.Value.Scalar;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.inject.Inject;
import java.util.Objects;
import java.util.UUID;

public interface TaskFactory {
    TaskInfo createTask(Offer offer, TaskID taskId, NodeTask nodeTask);

    class NMTaskFactoryImpl implements TaskFactory {
        public static final String EXECUTOR_NAME = "myriad_task";
        public static final String EXECUTOR_PREFIX = "myriad_executor";
        private static final Logger LOGGER = LoggerFactory
                .getLogger(NMTaskFactoryImpl.class);
        private MyriadConfiguration cfg;
        private TaskUtils taskUtils;

        @Inject
        public NMTaskFactoryImpl(MyriadConfiguration cfg, TaskUtils taskUtils) {
            this.cfg = cfg;
            this.taskUtils = taskUtils;
        }

        private static String getFileName(String uri) {
            int lastSlash = uri.lastIndexOf('/');
            if (lastSlash == -1) {
                return uri;
            } else {
                String fileName = uri.substring(lastSlash + 1);
                Preconditions.checkArgument(!Strings.isNullOrEmpty(fileName),
                        "URI should not have a slash at the end");
                return fileName;
            }
        }

        @Override
        public TaskInfo createTask(Offer offer, TaskID taskId, NodeTask nodeTask) {
            Objects.requireNonNull(offer, "Offer should be non-null");
            Objects.requireNonNull(nodeTask, "NodeTask should be non-null");

            NMProfile profile = nodeTask.getProfile();
            NMTaskConfig nmTaskConfig = new NMTaskConfig();
            nmTaskConfig.setAdvertisableCpus(taskUtils.getTaskCpus(profile));
            nmTaskConfig.setAdvertisableMem(taskUtils.getTaskMemory(profile));
            nmTaskConfig.setUser(this.cfg.getNodeManagerConfiguration()
                    .getUser().orNull());
            nmTaskConfig.setJvmOpts(this.cfg.getNodeManagerConfiguration()
                    .getJvmOpts().orNull());

            String taskConfigJSON = new Gson().toJson(nmTaskConfig);

            Scalar taskMemory = Value.Scalar.newBuilder()
                    .setValue(taskUtils.getTaskMemory(profile)).build();
            Scalar taskCpus = Value.Scalar.newBuilder()
                    .setValue(taskUtils.getTaskCpus(profile)).build();
            Scalar executorMemory = Value.Scalar.newBuilder()
                    .setValue(taskUtils.getExecutorMemory()).build();
            Scalar executorCpus = Value.Scalar.newBuilder()
                    .setValue(taskUtils.getExecutorCpus()).build();

            String executorPath = cfg.getMyriadExecutorConfiguration()
                    .getPath();
            URI executorURI = URI.newBuilder().setValue(executorPath)
                    .setExecutable(true).build();
            CommandInfo commandInfo = CommandInfo.newBuilder()
                    .addUris(executorURI).setUser(nmTaskConfig.getUser())
                    .setValue("java -jar " + getFileName(executorPath)).build();

            ExecutorID executorId = Protos.ExecutorID.newBuilder()
                    .setValue(EXECUTOR_PREFIX + offer.getSlaveId().getValue())
                    .build();
            ExecutorInfo executorInfo = ExecutorInfo
                    .newBuilder()
                    .setCommand(commandInfo)
                    .setName(EXECUTOR_NAME)
                    .addResources(
                            Resource.newBuilder().setName("cpus")
                                    .setType(Value.Type.SCALAR)
                                    .setScalar(executorCpus).build())
                    .addResources(
                            Resource.newBuilder().setName("mem")<commit_msg>Now configuring executor's environment as per user specified env vars.<commit_after>package com.ebay.myriad.scheduler;

import com.ebay.myriad.configuration.MyriadConfiguration;
import com.ebay.myriad.executor.NMTaskConfig;
import com.ebay.myriad.state.NodeTask;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import com.google.gson.Gson;
import com.google.protobuf.ByteString;
import org.apache.mesos.Protos;
import org.apache.mesos.Protos.*;
import org.apache.mesos.Protos.CommandInfo.URI;
import org.apache.mesos.Protos.Value.Scalar;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.inject.Inject;
import java.util.Arrays;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;

public interface TaskFactory {
    TaskInfo createTask(Offer offer, TaskID taskId, NodeTask nodeTask);

    class NMTaskFactoryImpl implements TaskFactory {
        public static final String EXECUTOR_NAME = "myriad_task";
        public static final String EXECUTOR_PREFIX = "myriad_executor";
        private static final Logger LOGGER = LoggerFactory
                .getLogger(NMTaskFactoryImpl.class);
        private MyriadConfiguration cfg;
        private TaskUtils taskUtils;

        @Inject
        public NMTaskFactoryImpl(MyriadConfiguration cfg, TaskUtils taskUtils) {
            this.cfg = cfg;
            this.taskUtils = taskUtils;
        }

        private static String getFileName(String uri) {
            int lastSlash = uri.lastIndexOf('/');
            if (lastSlash == -1) {
                return uri;
            } else {
                String fileName = uri.substring(lastSlash + 1);
                Preconditions.checkArgument(!Strings.isNullOrEmpty(fileName),
                        "URI should not have a slash at the end");
                return fileName;
            }
        }

        @Override
        public TaskInfo createTask(Offer offer, TaskID taskId, NodeTask nodeTask) {
            Objects.requireNonNull(offer, "Offer should be non-null");
            Objects.requireNonNull(nodeTask, "NodeTask should be non-null");

            NMProfile profile = nodeTask.getProfile();
            NMTaskConfig nmTaskConfig = new NMTaskConfig();
            nmTaskConfig.setAdvertisableCpus(taskUtils.getTaskCpus(profile));
            nmTaskConfig.setAdvertisableMem(taskUtils.getTaskMemory(profile));
            nmTaskConfig.setUser(this.cfg.getNodeManagerConfiguration()
                    .getUser().orNull());
            nmTaskConfig.setJvmOpts(this.cfg.getNodeManagerConfiguration()
                    .getJvmOpts().orNull());

            String taskConfigJSON = new Gson().toJson(nmTaskConfig);

            Scalar taskMemory = Value.Scalar.newBuilder()
                    .setValue(taskUtils.getTaskMemory(profile)).build();
            Scalar taskCpus = Value.Scalar.newBuilder()
                    .setValue(taskUtils.getTaskCpus(profile)).build();
            Scalar executorMemory = Value.Scalar.newBuilder()
                    .setValue(taskUtils.getExecutorMemory()).build();
            Scalar executorCpus = Value.Scalar.newBuilder()
                    .setValue(taskUtils.getExecutorCpus()).build();

            String executorPath = cfg.getMyriadExecutorConfiguration()
                    .getPath();
            URI executorURI = URI.newBuilder().setValue(executorPath)
                    .setExecutable(true).build();

            Environment.Builder environment = Environment.newBuilder();
            Map<String, String> yarnEnvironment = cfg.getYarnEnvironment();
            for (String key: yarnEnvironment.keySet()) {
                String value = yarnEnvironment.get(key);
                environment.addVariables(Environment.Variable.newBuilder()
                        .setName(key)
                        .setValue(value).build());
            }

            CommandInfo commandInfo = CommandInfo.newBuilder()
                    .addUris(executorURI).setUser(nmTaskConfig.getUser())
                    .setValue("export CAPSULE_CACHE_DIR=`pwd`;echo $CAPSULE_CACHE_DIR; java -Dcapsule.log=verbose -jar " + getFileName(executorPath)).setUser("hduser")
                    .setEnvironment(environment).build();

            ExecutorID executorId = Protos.ExecutorID.newBuilder()
                    .setValue(EXECUTOR_PREFIX + offer.getSlaveId().getValue())
                    .build();
            ExecutorInfo executorInfo = ExecutorInfo
                    .newBuilder()
                    .setCommand(commandInfo)
                    .setName(EXECUTOR_NAME)
                    .addResources(
                            Resource.newBuilder().setName("cpus")
                                    .setType(Value.Type.SCALAR)
                                    .setScalar(executorCpus).build())
                    .addResources(
                            Resource.newBuilder().setName("mem")
{'fraction_non_alphanumeric': 0.09482852476864453, 'fraction_numerical': 0.0008709853021230267, 'mean_word_length': 2.162134251290878, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6199314', 'n_tokens_mistral': 2441, 'n_tokens_neox': 2361, 'n_words': 430}