D-Net

package eu.dnetlib.data.hadoop.oozie;

import java.io.IOException;

import java.util.*;

import com.google.common.collect.Maps;

import com.google.common.collect.Sets;

import org.apache.commons.io.IOUtils;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.oozie.client.OozieClient;

import org.apache.oozie.client.OozieClientException;

import org.apache.oozie.client.WorkflowAction;

import org.apache.oozie.client.WorkflowJob;

import org.apache.oozie.client.WorkflowJob.Status;

import eu.dnetlib.data.hadoop.action.JobCompletion;

import eu.dnetlib.data.hadoop.action.JobMonitor;

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

import static java.lang.String.format;

public class OozieJobMonitor extends JobMonitor {

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

	private final OozieClient oozieClient;

	private final String jobId;

	public static final String ACTION_TYPE_SUBWORKFLOW = "sub-workflow";

	private Set<String> workflowActions = Sets.newHashSet();

	@Deprecated

	public OozieJobMonitor(final OozieClient oozieClient, String jobId, final JobCompletion callback) {

		super(callback);

		this.oozieClient = oozieClient;

		this.jobId = jobId;

	}

	public OozieJobMonitor(final OozieClient oozieClient, String jobId, final JobCompletion callback, final Set<String> workflowActions) {

		super(callback);

		this.oozieClient = oozieClient;

		this.jobId = jobId;

		this.workflowActions = workflowActions;

	}

	@Override

	public void run() {

		try {

			log.info("waiting for oozie job completion: " + getHadoopId());

			Status status = Status.PREP;

			while (status.equals(Status.PREP) || status.equals(Status.RUNNING)) {

				Thread.sleep(monitorSleepTimeSecs * 1000);

				try {

					final Status currentStatus = doGetStatus();

					if (!currentStatus.equals(status)) {

						status = currentStatus;

						lastActivity = new Date();

					}

				} catch (Throwable e) {

					log.warn(format("error polling status for job %s", jobId), e);

				}

			}

			log.debug(format("job %s finihsed with status: %s", jobId, status));

			if (Status.SUCCEEDED.equals(status)) {

				// TODO set some content to return to the blackboard msg.

				log.info(format("looking for oozie job(%s) output values: %s", getHadoopId(), workflowActions));

				final Properties report = getReport(getOozieClient(), getHadoopId(), workflowActions);

				if (report != null) {

					final Map<String, String> map = Maps.newHashMap();

					report.forEach((k, v) -> map.put(k.toString(), v.toString()));

					log.info("found oozie job report, size: " + map.size());

					getCallback().done(map);

				} else {

					log.warn("cannot find oozie job report!");

					getCallback().done(new HashMap<>());

				}

            } else {

				// TODO retrieve some failure information from the oozie client.

				String msg = format("hadoop job: %s failed with status: %s, oozie log:\n %s\n", getHadoopId(), getStatus(), getOozieClient().getJobLog(getHadoopId()));

				getCallback().failed(msg, new HadoopServiceException(msg));

            }

		} catch (Throwable e) {

			getCallback().failed(getHadoopId(), e);

		}

	}

	/**

	 * Provides report entries when found for given oozie job identifier. Returns null when report not found.

	 */

	private static Properties getReport(final OozieClient oozieClient, final String oozieJobId, final Set<String> workflowActions) throws OozieClientException, IOException {

		WorkflowJob oozieJob = oozieClient.getJobInfo(oozieJobId);

		for (WorkflowAction currentAction : oozieJob.getActions()) {

			log.info(String.format("looking for workflow actions to report, current: '%s'", currentAction.getName()));

			if (workflowActions.contains(currentAction.getName())) {

				log.info(String.format("found workflow action %s", currentAction.getName()));

				if (ACTION_TYPE_SUBWORKFLOW.equals(currentAction.getType())) {

					log.info(String.format("looking for sub-workflow actions external id: %s", currentAction.getExternalId()));

					Properties subworkflowProperties = getReport(oozieClient, currentAction.getExternalId(), workflowActions);

					if (subworkflowProperties != null) {

						return subworkflowProperties;

					}

				} else if (StringUtils.isNotBlank(currentAction.getData())) {

					Properties properties = new Properties();

					properties.load(IOUtils.toInputStream(currentAction.getData()));

					log.info(String.format("found workflow action(%s) properties size %s", currentAction.getName(), properties.values().size()));

					return properties;

				}

			} else {

				log.info(String.format("cannot find workflow action(%s) properties", currentAction.getName()));

			}

		}

		return null;

	}

	@Override

	public String getHadoopId() {

		return jobId;

	}

	public OozieClient getOozieClient() {

		return oozieClient;

	}

	@Override

	public String getStatus() {

		try {

			return doGetStatus().toString();

		} catch (OozieClientException e) {

			log.error("error accessing job status", e);

			return "UNKNOWN";

		}

	}

	private Status doGetStatus() throws OozieClientException {

		return getOozieClient().getJobInfo(getHadoopId()).getStatus();

	}

	@Override

	public Date getLastActivity() {

		return lastActivity;

	}

	@Override

	public Date getStartTime() throws HadoopServiceException {

		try {

			return getOozieClient().getJobInfo(getHadoopId()).getStartTime();

		} catch (OozieClientException e) {

			throw new HadoopServiceException("unable to read job start time", e);

		}

	}

	@Override

	public String getTrackerUrl() {

		return getOozieClient().getOozieUrl();

	}

	@Override

	public void kill() {

		try {

			getOozieClient().kill(getHadoopId());

		} catch (OozieClientException e) {

			log.error("unable to kill job: " + getHadoopId(), e);

		}

	}

}