D-Net

package eu.dnetlib.data.hadoop.action;

import java.net.URI;

import java.util.List;

import java.util.Map;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.hadoop.fs.Path;

import org.springframework.beans.factory.annotation.Autowired;

import com.google.common.collect.Iterables;

import com.google.common.collect.Sets;

import eu.dnetlib.data.hadoop.HadoopClientMap;

import eu.dnetlib.data.hadoop.JobRegistry;

import eu.dnetlib.data.hadoop.config.ClusterName;

import eu.dnetlib.data.hadoop.rmi.HadoopServiceException;

import eu.dnetlib.data.hadoop.utils.JobProfile;

import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpException;

import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService;

import eu.dnetlib.enabling.is.registry.rmi.ISRegistryException;

import eu.dnetlib.enabling.is.registry.rmi.ISRegistryService;

import eu.dnetlib.enabling.tools.blackboard.BlackboardJob;

import eu.dnetlib.enabling.tools.blackboard.BlackboardServerHandler;

import eu.dnetlib.miscutils.datetime.DateUtils;

import eu.dnetlib.miscutils.functional.hash.Hashing;

public abstract class AbstractSubmitAction extends AbstractHadoopAction {

	private static final Log log = LogFactory.getLog(AbstractSubmitAction.class); // NOPMD by marko on 11/24/08 5:02 PM

	@Autowired

	protected HadoopClientMap hadoopClientMap;

	@Autowired

	protected JobRegistry jobRegistry;

	protected abstract void submit(final JobCompletion callback, final BlackboardJob job, final String jobName, final JobProfile jobProfile)

			throws HadoopServiceException;

	@Override

	protected void executeAsync(final BlackboardServerHandler handler, final BlackboardJob job) throws HadoopServiceException {

		final String jobName = job.getParameters().get("job.name");

		final JobProfile jobProfile = loadISJobConfiguration(jobName, job.getParameters());

		validateJobParams(handler, job, jobName, jobProfile);

		if (!isSimulation(job)) {

			submit(newCompletionCallback(handler, job), job, jobName, jobProfile);

			updateJobStatus(jobName);

		} else {

			log.info(String.format("simulating job: '%s', done!", jobName));

			handler.done(job);

		}

	}

	private boolean isSimulation(final BlackboardJob job) {

		if (!job.getParameters().containsKey("simulation")) { return false; }

		final String sim = job.getParameters().get("simulation");

		return !StringUtils.isBlank(sim) && sim.equals("true");

	}

	protected void updateJobStatus(final String jobName) {

		incrementCumulativeRun(jobName);

		incrementRunningJobs(jobName);

		updateDate(jobName);

	}

	protected void incrementRunningJobs(final String jobName) {

		log.debug("increment #running jobs: " + jobName);

		updateCountElement(jobName, "RUNNING_INSTANCES", "+ 1");

	}

	protected void decrementRunningJobs(final String jobName) {

		log.debug("decrement #running jobs: " + jobName);

		updateCountElement(jobName, "RUNNING_INSTANCES", "- 1");

	}

	protected void incrementCumulativeRun(final String jobName) {

		log.debug("increment #cumulative runs for job: " + jobName);

		updateCountElement(jobName, "CUMULATIVE_RUN", "+ 1");

	}

	protected void updateDate(final String jobName) {

		log.info("increment last submission date for job: " + jobName);

		final String xquery =

				"for $x in collection('')/RESOURCE_PROFILE[" + ".//RESOURCE_TYPE/@value='HadoopJobConfigurationDSResourceType' and .//HADOOP_JOB/@name='"

						+ jobName + "'] " + " return update value $x//LAST_SUBMISSION_DATE/@value with '" + DateUtils.now_ISO8601() + "' ";

		executeXUpdate(xquery);

	}

	private void updateCountElement(final String jobName, final String element, final String delta) {

		final String xquery =

				"let $x := //RESOURCE_PROFILE[" + ".//RESOURCE_TYPE/@value='HadoopJobConfigurationDSResourceType' and .//HADOOP_JOB/@name='" + jobName

						+ "'], $tot := $x//STATUS/" + element + "/@value/number() " + delta + " return update replace $x//STATUS/" + element + " with <"

						+ element + " value='{$tot}' />";

		executeXUpdate(xquery);

	}

	protected String newJobId(final ClusterName clusterName, final Object internalId) {

		return "job-" + Hashing.md5(clusterName.toString() + internalId.toString());

	}

	protected String getDefaultLibPath(final String defaultFs) throws HadoopServiceException {

		try {

			final String libPath = queryForServiceProperty("defaultLibPath");

			final Path path = new Path(URI.create(defaultFs + libPath));

			return path.toString();

		} catch (ISLookUpException e) {

			throw new HadoopServiceException("unable to get default lib path", e);

		}

	}

	protected void validateJobParams(final BlackboardServerHandler handler, final BlackboardJob bbJob, final String jobName, final JobProfile jobProfile)

			throws HadoopServiceException {

		if (!bbJob.getParameters().keySet().containsAll(jobProfile.getRequiredParams())) {

			String msg =

					"required parameter is missing for job: " + jobName + ", required params: " + jobProfile.getRequiredParams() + "\n\nmissing params: "

							+ Sets.difference(jobProfile.getRequiredParams(), bbJob.getParameters().keySet());

			log.error(msg);

			HadoopServiceException e = new HadoopServiceException(msg);

			handler.failed(bbJob, e);

			throw e;

		}

	}

	protected JobCompletion newCompletionCallback(final BlackboardServerHandler handler, final BlackboardJob bbJob) {

		final String jobName = bbJob.getParameters().get("job.name");

		return new JobCompletion() {

			@Override

			public void done(final Map<String, String> report) {

				bbJob.getParameters().putAll(report);

				log.info(String.format("%s completed successfully, returning %s output params", jobName, report.size()));

				handler.done(bbJob);

				decrementRunningJobs(jobName);

			}

			@Override

			public void failed(final Map<String, String> report, final String msg, final Throwable e) {

				bbJob.getParameters().putAll(report);

				log.error(String.format("%s failed, returning %s output params", jobName, report.size()));

				handler.failed(bbJob, e);

				decrementRunningJobs(jobName);

			}

		};

	}

	private String queryForServiceProperty(final String key) throws ISLookUpException {

		return getServiceConfigValue("for $x in /RESOURCE_PROFILE[.//RESOURCE_TYPE/@value='HadoopServiceResourceType'] return $x//SERVICE_PROPERTIES/PROPERTY[./@ key='"

				+ key + "']/@value/string()");

	}

	private String getServiceConfigValue(final String xquery) throws ISLookUpException {

		log.debug("quering for service property: " + xquery);

		final List<String> urls = serviceLocator.getService(ISLookUpService.class).quickSearchProfile(xquery);

		if (urls == null || urls.size() != 1) { throw new IllegalStateException("unable to find unique service property, xquery: " + xquery); }

		return Iterables.getOnlyElement(urls);

	}

	private boolean executeXUpdate(final String xupdate) {

		try {

			log.debug("running xupdate: " + xupdate);

			return serviceLocator.getService(ISRegistryService.class).executeXUpdate(xupdate);

		} catch (ISRegistryException e) {

			log.error("unable to run xupdate: " + xupdate, e);

			return false;

		}

	}

}