%evaluateConfigure
% Entry point for job configuration

%   Copyright 2014-2020 The MathWorks, Inc.

function out = evaluateConfigure(confJarFile, hadoopClasspath, hadoopVersion, hadoopFullVersion, jobDataDir, jobTempDir, logDir, serializedConfiguration)
import matlab.io.datastore.internal.HadoopConfiguration;
import parallel.internal.hadoop.HadoopWorker;
import com.mathworks.toolbox.parallel.mapreduce.PairSocket;
import com.mathworks.toolbox.parallel.mapreduce.SerializationUtils;
out = [];

persistent alreadyConfigured
if isempty(alreadyConfigured)
    alreadyConfigured = true;
    mlock
else
    return;
end

% Ensure MATLAB places all temporary files for the job in the temp location
% that Hadoop allows us to.
if ~isempty(jobTempDir)
    setenv('TEMP', jobTempDir);
    setenv('TMP', jobTempDir);
end

% Ensure logging is enabled early
logLevel = iSetupLogging();
if logLevel > 0
    iEnableJavaLogging(fullfile(logDir, sprintf('worker_%i.log', feature('getpid'))));
end

try
    % PairSocket underpins any streaming IPC between this process and the parent
    % Hadoop process. It requires us to initialize it explicitly before use.
    PairSocket.initializeNativeLibrary();
    
    iAddAdditionalClasspath(hadoopClasspath, confJarFile, hadoopVersion, hadoopFullVersion);
    
    guard = matlab.io.datastore.internal.ContextClassLoaderGuard(); %#ok<NASGU>
    
    % Set the global configuration that all HDFS/Sequence file data access uses.
    % We ensure the username is set to allow Hadoop classes to operate on
    % HDFS as the right user.
    hdfsLoader = com.mathworks.storage.hdfsloader.GlobalHdfsLoader.get();
    configuration = hdfsLoader.newConfiguration();
    configuration.readFields(...
        SerializationUtils.decodeFromString(serializedConfiguration));
    
    % In simple authentication mode, the workers are run as the yarn daemon
    % user. We have to be careful to ensure HDFS file access is done as the
    % submitting user. This is not needed in Kerberos authentication mode
    % because workers are run as the submitting user. If we accidentally set
    % the username while in Kerberos authentication mode, Hadoop ignores all
    % credentials and fails (g1296020).
    authenticationMode = char(configuration.get('hadoop.security.authentication', 'simple'));
    if strcmpi(authenticationMode, 'simple')
        configuration.set('mathworks.hdfs.username', configuration.get('mathworks.hdfs.username', getenv('USER')));
    end
    
    HadoopConfiguration.setGlobalConfiguration(configuration);
    
    HadoopWorker.configure(jobDataDir);
catch err
    dctSchedulerMessage(0, 'Failed when configuring the worker for the current Hadoop Job, the error was, ', err);
    rethrow(err);
end
end

% Mirror the classpath of the parent Hadoop process by adding items to the
% dynamic classpath.
function iAddAdditionalClasspath(hadoopClasspath, confJarFile, hadoopVersion, hadoopFullVersion)
iSetHadoopHomeIfNeeded(fileparts(confJarFile));

if strcmp(hadoopVersion, '1')
    additionalClasspath = {...
        fullfile(toolboxdir('shared'), 'bigdata', 'jar', 'hadoop.1.jar'); ...
        fullfile(toolboxdir('parallel'), 'bigdata', 'jar', 'hadoopjob.1.jar')};
else
    % Assume compatible with version 2.
    additionalClasspath = {...
        fullfile(toolboxdir('shared'), 'bigdata', 'jar', 'hadoop.2.jar'); ...
        fullfile(toolboxdir('parallel'), 'bigdata', 'jar', 'hadoopjob.2.jar'); ...
        };
end

additionalOptions = {};
hadoopClasspath = strtrim(char(hadoopClasspath));
if ~isempty(hadoopClasspath)
    additionalOptions = {'HadoopClasspath', strsplit(hadoopClasspath, pathsep)'};
end

matlab.io.internal.vfs.hadoop.hadoopLoader('load', ...
    'HadoopVersion', hadoopFullVersion, ...
    additionalOptions{:}, ...
    'AdditionalClasspath', additionalClasspath);
end

% Set the HADOOP_HOME environment variable if it is not already set.
%
% This should not be needed as we are inheriting the environment from a
% HADOOP process and HADOOP currently ensures HADOOP_HOME or HADOOP_PREFIX
% is set. We still do it for robustness against future versions of HADOOP.
function iSetHadoopHomeIfNeeded(confJarFile)
if ~isempty(getenv('HADOOP_HOME')) && ~isempty(getenv('HADOOP_PREFIX'))
    return;
end

% We assume here that the jar file containing org.apache.hadoop.conf.Configuration
% exists inside of the Hadoop installation folder.
prevPath = '';
currentPath = confJarFile;
while ~isequal(currentPath, prevPath)
    status = exist(fullfile(currentPath, 'bin', 'hadoop'), 'file');
    if status ~= 0
        setenv('HADOOP_HOME', currentPath);
        return;
    end
    
    prevPath = currentPath;
    currentPath = fileparts(currentPath);
end

% Reaching this point means we could not find the HADOOP installation
% folder using the given path. At this point, it is up-to hadoopLoader to
% discover the necessary information on its own.
dctSchedulerMessage(1, sprintf('Could not find the HADOOP installation folder from ''%s''.\n', confJarFile));

end

% Setup the logging framework for MATLAB side of PCT.
function logLevel = iSetupLogging()

try
    logLevel = parallel.internal.logging.getWorkerLogLevel();
    setSchedulerMessageHandler(iMessageHandler(logLevel));
catch err
    dctSchedulerMessage(0, 'Evaluating task with DistcompEvaluateFileTask failed, the error was', err);
    throw(err.getReport());
end
end

% Create a message handler for PCT log framework that forwards to java
% logs.
function handler = iMessageHandler(desiredLevel)

loglvl = com.mathworks.toolbox.parallel.pctutil.logging.DistcompLevel.getLevelFromValue(desiredLevel);
LOGGER = com.mathworks.toolbox.parallel.pctutil.logging.RootLog.LOG;
LOGGER.setLevel(loglvl);

handler = @nHandler;

    function nHandler(msg, messageLevel)    
    
    if messageLevel > desiredLevel
        return;
    end    
    
    % This handler puts messages to two places - stdout (via fprintf)
    fprintf('%s | %s\n', datestr(now, 31), msg)
    
    % And to the java logger
    level = com.mathworks.toolbox.parallel.pctutil.logging.DistcompLevel.getLevelFromValue(messageLevel);
    LOGGER.log(level, ['MATLAB: ', msg]);
    end
end

% Enable the java logger to the given file.
function iEnableJavaLogging(filename)
% Attach a FileHandler to the distcomp package logger. In particular this 
% means that log messages from the pmode infrastructure get captured.
append = true;
levelNum = 6; % Maximum amount of logging
try
    handler = com.mathworks.toolbox.parallel.pctutil.logging.DistcompFileHandler(filename, append, levelNum);
    logger = com.mathworks.toolbox.parallel.pctutil.logging.RootLog.LOG;
    logger.addHandler(handler);
catch
    % Nothing we can do here.
end
end
