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;

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();

	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 = doGetStatus();

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

				Thread.sleep(monitorSleepTimeSecs * 1000);

				Status currentStatus = doGetStatus();

				if (!status.equals(currentStatus)) {

					status = currentStatus;

					lastActivity = new Date(System.currentTimeMillis());

				}

			}

			log.debug("job " + jobId + " finihsed with status: " + status);

			if (status == Status.SUCCEEDED) {

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

				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()));

					getCallback().done(map);

				} else {

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

				}

            } else {

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

				String msg = "hadoop job: " + getHadoopId() + " failed with status: " + getStatus() + ", oozie log:\n "

						+ getOozieClient().getJobLog(getHadoopId()) + "\n\n";

				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.debug(String.format("looking for workflo actions to report, current: '%s'", currentAction.getName()));

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

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

					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()));

					return properties;

				}

			}

		}

		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);

		}

	}

}