public class SorcererException extends Exception {

	public SorcererException() {
		super();
	}

	public SorcererException(String message) {
		super(message);
	}

	public SorcererException(Throwable t) {
		super(t);
	}

	public SorcererException(String message, Throwable t) {
		super(message, t);
	}
} public enum ExecutionStatus {
	DISABLED,
	SUCCESS,
	COMPLETED,
	DEPENDENCY_FAILURE,
	RUNNING,
	ERROR,
	ABORTED
} public class MetricUnit {

	private static final String MODULE_NAME = "Audience_Reports";
	private static final String SERVICE_PREFIX = "Workflow";

	public static final String METRIC_NAME_KEY = "Metric_Name";
	public static final String TASK_NAME_KEY = "Task_Name";
	public static final String PIPELINE_NAME_KEY = "Pipeline_Name";

	private static final String GRAPHITE_KEY = "Graphite_Key";
	private static final String SUCCESSFUL_FINISH_TIME_KEY = "Successful_Finish_Time";

	// TODO: remove once graphite is retired
	private boolean graphiteOnly = false;

	private Map<String, String> keyUnits;

	public MetricUnit(boolean graphiteOnly) {
		this.graphiteOnly = graphiteOnly;
		keyUnits = Maps.newHashMap();
	}

	public void setMetricNameForTask(String metricName, String taskName) {
		if (graphiteOnly) {
			keyUnits.put(GRAPHITE_KEY, metricName);
		} else {
			keyUnits.put(METRIC_NAME_KEY, metricName);
			keyUnits.put(TASK_NAME_KEY, taskName);
		}
	}

	public void setMetricNameForPipeline(String metricName, String pipelineName) {
		if (graphiteOnly) {
			keyUnits.put(GRAPHITE_KEY, metricName);
		} else {
			keyUnits.put(METRIC_NAME_KEY, metricName);
			keyUnits.put(PIPELINE_NAME_KEY, pipelineName);
		}
	}

	public void addTags(String tagKey, String tagValue) {
		keyUnits.put(tagKey, tagValue);
	}
	/**
	 * Utility class to generate keys for different monitoring dashboard
	 *
	 */
	public String getGraphiteKey() {
		if (graphiteOnly) {
			return keyUnits.get(GRAPHITE_KEY);
		}

		String tagName = "";
		if (keyUnits.get(TASK_NAME_KEY) != null) {
			tagName = keyUnits.get(TASK_NAME_KEY);
		} else if (keyUnits.get(PIPELINE_NAME_KEY) != null) {
			tagName = keyUnits.get(PIPELINE_NAME_KEY);
		}
		Joiner joiner = Joiner.on('.');
		return joiner.join(SERVICE_PREFIX, tagName, keyUnits.get(METRIC_NAME_KEY));
	}

	public String getModuleName() {
		return MODULE_NAME;
	}

	public String getTSDBKey() {
		return Joiner.on('.').join(SERVICE_PREFIX, keyUnits.get(METRIC_NAME_KEY));
	}

	public String getTSDBTags() {
		String tags = "";
		if (keyUnits.get(TASK_NAME_KEY) != null) {
			tags = Joiner.on('=').join(TASK_NAME_KEY, keyUnits.get(TASK_NAME_KEY));
		} else if (keyUnits.get(PIPELINE_NAME_KEY) != null) {
			tags = Joiner.on('=').join(PIPELINE_NAME_KEY, keyUnits.get(PIPELINE_NAME_KEY));
		}
		return tags;
	}

	public boolean isGraphiteOnly() {
		return graphiteOnly;
	}

	public String getTaskName() {
		return keyUnits.get(TASK_NAME_KEY);
	}

	public static MetricUnit getMetricUnit(boolean graphiteOnly, String taskName, String metricName) {
		MetricUnit unit = new MetricUnit(graphiteOnly);
		unit.setMetricNameForTask(metricName, taskName);
		return unit;
	}

	@Override
	public int hashCode() {
		return keyUnits.hashCode();
	}

	@Override
	public boolean equals(Object obj) {
		if (obj instanceof MetricUnit) {
			MetricUnit unit = (MetricUnit) obj;
			for (String key: this.keyUnits.keySet()) {
				if (!unit.keyUnits.containsKey(key)) {
					return false;
				} else if (unit.keyUnits.get(key).equals(this.keyUnits.get(key))){
					return false;
				}
			}
			return true;
		}
		return false;
	}
} public class MetricsMonitor {
	private static final Logger logger =
			LoggerFactory.getLogger(MetricsMonitor.class);

	private final Map<MetricUnit, Long> METRICS = Maps.newConcurrentMap();

	/** Singleton */
	private static final MetricsMonitor instance = new MetricsMonitor();

	private MetricsMonitor() {

	}

	public static MetricsMonitor getInstance() {
		return instance;
	}

	public Map<String, Object> getGraphiteMetrics() {
		Map<String, Object> metrics = Maps.newTreeMap();
		for (Map.Entry<MetricUnit, Long> entry: METRICS.entrySet()) {
			metrics.put(entry.getKey().getGraphiteKey(), entry.getValue());
		}
		return metrics;
	}

	public void addGenericMetric(MetricUnit unit, Long value) {
		instance.METRICS.put(unit, value);
		instance.pushOutMetrics();
	}

	public void addSuccessMetric(MetricUnit unit, int todayDateKey) {
		instance.updateMetricsFromTaskStatus(unit, todayDateKey);
		instance.pushOutMetrics();
	}

	private void updateMetricsFromTaskStatus(MetricUnit unit, int todayDateKey) {
		String taskName = unit.getTaskName();
		if (taskName == null) {
			logger.error("Metrics don't have a task name!");
			return;
		}

		// Check status from status file
		Long statusSuccessTime;
		statusSuccessTime = StatusManager.get().getTaskLastSuccessTime(taskName, todayDateKey).toInstant().getMillis();


		// If no status file found then return
		if (statusSuccessTime == -1) {
			return;
		}

		// Get last success time
		Long serviceSuccessTime = -1L;
		if (METRICS.containsKey(unit)) {
			serviceSuccessTime = METRICS.get(unit);
		}

		// Update metrics with latest success time
		if (statusSuccessTime > serviceSuccessTime) {
			logger.debug("{} status file modified time greater than existing last success time. " +
					"Replacing metric.", taskName);
			METRICS.put(unit, statusSuccessTime);
		}
	}

	public void pushOutMetrics() {
		for (Map.Entry<MetricUnit, Long> entry: METRICS.entrySet()) {
			MetricUnit key = entry.getKey();

			// some metrics are graphite only
			if (!key.isGraphiteOnly()) {
				pushOutTSDBMetrics(key, entry.getValue());
			}
		}
	}

	private void pushOutTSDBMetrics(MetricUnit unit, Long value) {

	}
} public enum Status {

	/**
	 * Represents a task that is a part of a scheduled pipeline but is not yet
	 * eligible for execution, or more specifically the previous tasks in the
	 * workflow have not completed or one of
	 * {@link com.turn.sorcerer.task.Task#getDependencies(int)} is returning
	 * false
	 */
	PENDING("PENDING"),

	/**
	 * Represents either:
	 *
	 * <li>A task that is in-progress, or more specifically
	 * {@link com.turn.sorcerer.task.Task#exec} method is running
	 * </li>
	 *
	 * <li>
	 * A pipeline whose initial task has been launched but the pipeline is not
	 * complete, or more specifically all of its memeber tasks have not
	 * completed successfully.
	 * </li>
	 */
	IN_PROGRESS("RUNNING"),

	/**
	 * Represents either:
	 *
	 * <li>
	 * A successfully completed task, or more specifically
	 * {@link com.turn.sorcerer.task.Task#exec} has completed with no
	 * exceptions
	 * </li>
	 *
	 * <li>
	 * A successfully completed pipeline, or more specifically a pipeline
	 * with all member tasks successfully completed.
	 * </li>
	 */
	SUCCESS("SUCCESS"),

	/**
	 * Represents a task with a previous iteration run that resulted in an
	 * error state, or more specifically
	 * {@link com.turn.sorcerer.task.Task#exec} was executed with an exception
	 * thrown
	 */
	ERROR("ERROR"),
	;

	// String representation of status
	private String string;

	private Status(String string) {
		this.string = string;
	}

	/**
	 * Returns string representation of state.
	 */
	public String getString() {
		return string;
	}
} public class StatusManager {

	private static final Logger logger =
			LoggerFactory.getLogger(StatusManager.class);

	private static final StatusManager INSTANCE= new StatusManager();

	public static StatusManager get() {
		return INSTANCE;
	}

	StatusStorage taskStorage = SorcererInjector.get().getStorageInstance().setType("tasks");
	StatusStorage pipelineStorage = SorcererInjector.get().getStorageInstance().setType("pipelines");

	private StatusManager() {

	}

	public boolean initialized() {
		try {
			taskStorage.init();
			pipelineStorage.init();
		} catch (IOException e) {
			logger.debug("Storage layer could not initialize", e);
			return false;
		}

		return true;
	}

	public void commitTaskStatus(TaskType type, int seq, Status status) {
		commitTaskStatus(type.getName(), seq, status, DateTime.now(), false);
	}

	public void commitTaskStatus(TaskType type, int seq, Status status, boolean overwrite) {
		commitTaskStatus(type.getName(), seq, status, DateTime.now(), overwrite);
	}

	private void commitTaskStatus(
			String taskName, int seq, Status status, DateTime time, boolean overwrite) {

		if (Status.PENDING.equals(status)) {
			try {
				taskStorage.clearAllStatuses(taskName, seq);
			} catch (IOException e) {
				logger.error("Storage layer unreachable!", e);
			}
			return;
		}

		try {
			taskStorage.commitStatus(taskName, seq, status, time, overwrite);
		} catch (IOException e) {
			logger.error(ExceptionUtils.getStackTrace(e));
		}
	}

	public Status checkTaskStatus(String taskName, int seq) {
		try {
			return taskStorage.checkStatus(taskName, seq);
		} catch (IOException e) {
			logger.error("Storage layer unreachable!", e);
		}
		return Status.PENDING;
	}

	public Status checkPipelineStatus(String pipelineName, int seq) {
		try {
			return pipelineStorage.checkStatus(pipelineName, seq);
		} catch (IOException e) {
			logger.error("Storage layer unreachable!", e);
		}
		return Status.PENDING;
	}

	public void removeInProgressTaskStatus(TaskType type, int seq) {
		removeInProgressTaskStatus(type.getName(), seq);
	}

	private void removeInProgressTaskStatus(String taskName, int seq) {
		try {
			taskStorage.removeStatus(taskName, seq, Status.IN_PROGRESS);
		} catch (IOException e) {
			logger.error("Storage layer unreachable!", e);
		}
	}

	public boolean isTaskComplete(TaskType type, int seq) {
		try {
			return Status.SUCCESS.equals(taskStorage.checkStatus(type.getName(), seq));
		} catch (IOException e) {
			logger.error("Storage layer unreachable!", e);
			return false;
		}
	}

	public boolean isTaskRunning(TaskType type, int seq) {
		try {
			return Status.IN_PROGRESS.equals(taskStorage.checkStatus(type.getName(), seq));
		} catch (IOException e) {
			logger.error("Storage layer unreachable!", e);
			return false;
		}
	}

	public boolean isTaskInError(TaskType type, int seq) {
		try {
			return Status.ERROR.equals(taskStorage.checkStatus(type.getName(), seq));
		} catch (IOException e) {
			logger.error("Storage layer unreachable!", e);
			return false;
		}
	}

	public void clearTaskStatus(TaskType type, int seq) {
		try {
			taskStorage.clearAllStatuses(type.getName(), seq);
		} catch (IOException e) {
			logger.error("Storage layer unreachable!", e);
		}
	}

	public DateTime getTaskLastSuccessTime(TaskType type, int seq) {
		return getTaskLastUpdateTimeForStatus(type, seq, Status.SUCCESS);
	}

	public DateTime getTaskLastSuccessTime(String taskName, int seq) {
		try {
			return taskStorage.getStatusUpdateTime(taskName, seq, Status.SUCCESS);
		} catch (IOException e) {
			logger.error("Storage layer unreachable!", e);
			return null;
		}
	}

	private DateTime getTaskLastUpdateTimeForStatus(TaskType type, int seq, Status status) {
		try {
			return taskStorage.getStatusUpdateTime(type.getName(), seq, status);
		} catch (IOException e) {
			logger.error("Storage layer unreachable!", e);
			return null;
		}
	}

	public int getCurrentIterationNumberForPipeline(PipelineType type) {
		return getCurrentIterationNumberForPipeline(type.getName());
	}

	public int getCurrentIterationNumberForPipeline(String pipelineName) {
		try {
			return pipelineStorage.getCurrentIterNo(pipelineName);
		} catch (IOException e) {
			logger.error("Storage layer unreachable!", e);
		}
		return 0;
	}

	public void commitPipelineStatus(PipelineType type, int seq, Status status) {
		commitPipelineStatus(type.getName(), seq, status, DateTime.now(), false);
	}

	public void commitPipelineStatus(PipelineType type, int seq, Status status, boolean overwrite) {
		commitPipelineStatus(type.getName(), seq, status, DateTime.now(), overwrite);
	}

	private void commitPipelineStatus(String taskName, int seq, Status status, DateTime time, boolean overwrite) {
		try {
			pipelineStorage.commitStatus(taskName, seq, status, time, overwrite);
		} catch (IOException e) {
			logger.error("Storage layer unreachable!", e);
		}
	}

	public boolean isPipelineComplete(PipelineType type, int seq) {
		try {
			return Status.SUCCESS.equals(pipelineStorage.checkStatus(type.getName(), seq));
		} catch (IOException e) {
			logger.error("Storage layer unreachable!", e);
			return false;
		}
	}

	public void clearPipelineStatus(PipelineType type, int seq) {
		try {
			pipelineStorage.clearAllStatuses(type.getName(), seq);
		} catch (IOException e) {
			logger.error("Storage layer unreachable!", e);
		}
	}
} public class Context {

	private int iterationNumber;
	private final TypedDictionary properties;
	private final Map<String, Long> metrics = Maps.newHashMap();

	public Context(int iterationNumber) {
		this.iterationNumber = iterationNumber;
		this.properties = new TypedDictionary();
	}

	public Context(int iterationNumber, TypedDictionary parameters) {
		properties = new TypedDictionary();
		this.iterationNumber = iterationNumber;
		properties.putAll(parameters);
	}


	public int getIterationNumber() {
		return this.iterationNumber;
	}

	public void addMetric(String key, Long value) {
		metrics.put(key, value);
	}

	public void putProperty(String key, Object value) {
		properties.put(key, value);
	}

	public TypedDictionary getProperties() {
		return this.properties;
	}
} public class ExecutableTask {

	private final int sequenceNumber;
	private final TaskType type;
	private final Task task;

	protected ExecutableTask(TaskType type, Task task, int seq) {
		this.type = type;
		this.task = task;
		this.sequenceNumber = seq;
	}

	public void initialize(Context context) {
		task.init(context);
	}

	public boolean checkDependencies() {
		// If the task doesn't have dependencies, return true
		if (task.getDependencies(sequenceNumber) == null ||
				task.getDependencies(sequenceNumber).size() == 0) {
			return true;
		}
		ImmutableList<Dependency> dependencies =
				ImmutableList.copyOf(task.getDependencies(sequenceNumber));

		// If any dependencies not met, return false
		for (Dependency dependency : dependencies) {
			if (dependency.check(sequenceNumber) == false) {
				return false;
			}
		}

		return true;
	}

	public void execute(Context context) throws SorcererException {
		try {
			task.exec(context);
		} catch (Exception e) {
			throw new SorcererException(e);
		}

	}

	public void abort() {
		if (this.task != null) {
			this.task.abort();
		}
	}

	public boolean isCompleted() {
		return StatusManager.get().isTaskComplete(this.type, this.sequenceNumber);
	}

	public boolean isRunning() {
		return StatusManager.get().isTaskRunning(this.type, this.sequenceNumber);
	}

	public boolean hasError() {
		return StatusManager.get().isTaskInError(this.type, this.sequenceNumber);
	}

	public String name() {
		return this.type.getName();
	}

	@Override
	public String toString() {
		return name() + ":" + sequenceNumber;
	}
} public class TaskFactory {

	private static TaskFactory INSTANCE = new TaskFactory();

	public static TaskFactory get() {
		return INSTANCE;
	}

	private TaskFactory() {

	}

	private Task getTask(TaskType type) {
		if (TaskType.FORK_EXEC.equals(type.getExec())) {
			return new ForkTask(type);
		}

		if (TaskType.JOIN_EXEC.equals(type.getExec())) {
			return new JoinTask(type.getName());
		}

		return SorcererInjector.get().getInstance(type);

	}

	public ExecutableTask getExecutableTask(TaskType type, int seq) {

		return new ExecutableTask(type, getTask(type), seq);
	}

} public class TaskType {

	private boolean enabled = true;

	private String name;

	private List<String> next;

	private String exec = "class";

	private CRITICALITY criticality = CRITICALITY.HIGH;

	private String sla;

	private Integer _sla_seconds;

	private boolean external = false;

	public TaskType() {}

	// For mocking and testing
	// SHOULD NEVER BE CALLED in production
	public TaskType(String name, List<String> next, String exec) {
		this.name = name;
		this.next = next;
		this.exec = exec == null ? this.exec : exec;
	}

	public boolean isEnabled() {
		return this.enabled;
	}

	public String getName() {
		return this.name;
	}

	public String getExec() {
		return this.exec.toLowerCase();
	}

	public List<String> getNextTaskNames() {
		return this.next;
	}

	public CRITICALITY getCriticality() {
		return this.criticality;
	}

	public boolean isExternalJob() {
		return this.external;
	}

	public int getSLA() {
		if (_sla_seconds == null) {
			try {
				initSLA();
			} catch (SorcererException e) {
				_sla_seconds = -1;
			}
		}

		return _sla_seconds;
	}

	public void initSLA() throws SorcererException {
		try {
			_sla_seconds = parseSLA(sla);
		} catch (Exception e) {
			throw new SorcererException("Incorrectly formatted SLA field for task " + this.name, e);
		}
	}

	public static int parseSLA(String slaString) throws Exception {
		if (slaString == null || slaString.trim().length() == 0) {
			return -1;
		}

		String trimmed = slaString.trim().toLowerCase();

		char unit = trimmed.charAt(trimmed.length() - 1);

		int num = Integer.parseInt(trimmed.substring(0, trimmed.length() - 2));

		switch (unit) {
			case 'd':
				return num * SEC_IN_DAY;
			case 'h':
				return num * SEC_IN_HOUR;
			case 'm':
				return num * SEC_IN_MINUTE;
			case 's':
				return num;
			default:
				throw new Exception("Invalid sla unit provided: " + unit);
		}
	}

	public String toString() {
		return MoreObjects.toStringHelper(this)
				.add("name", this.name)
				.add("next", this.next)
				.add("exec", this.exec)
				.add("criticality", this.criticality.name())
				.add("sla", this.sla)
				.toString();
	}

	/**
	 * The name field will be unique among the tasks so we use it for hashing
	 */
	@Override
	public int hashCode() {
		return name.hashCode();
	}

	@Override
	public boolean equals(Object o) {
		return o instanceof TaskType
				&& this.getName().equals(((TaskType) o).getName());

	}


	/**
	 * Supported task execution types
	 */
	public static final String FORK_EXEC = "fork";
	public static final String JOIN_EXEC = "join";

	public static final ImmutableMap<String, Class<? extends Task>> SUPPORTED_EXEC =
			ImmutableMap.<String, Class<? extends Task>>builder()
					.put(FORK_EXEC, ForkTask.class)
					.put(JOIN_EXEC, JoinTask.class)
					.build();

	/**
	 * Task criticality
	 */
	public enum CRITICALITY {
		HIGH,
		LOW
	}

	/**
	 *
	 */
	private static final int SEC_IN_DAY = 86400;
	private static final int SEC_IN_HOUR = 3600;
	private static final int SEC_IN_MINUTE = 60;
} public class Constants {

	public static final String REPORT_DATE_KEY = "-datekey";

	public static final String REPORT_DATE_STRING = "-date";

	public static final String ADHOC = "-adhoc";

	public static final String REPORT_ITERATION_NUMBER = "-iteration";

	public static final Marker SORCERER = MarkerFactory.getMarker("sorcerer");
} public class TypedDictionary {
	private Map<String, Object> map;

	public TypedDictionary() {
		map = Maps.newHashMap();
	}

	public <E> E get(String key, Class<E> type) {
		if (!map.containsKey(key)) return null;
		Object o = map.get(key);
		if (o == null) return null;
		return type.cast(o);
	}

	public boolean getBool(String key) {
		return get(key, Boolean.class);
	}

	public long getLong(String key) {
		return get(key, Long.class);
	}

	public String getString(String key) {
		return get(key, String.class);
	}

	public double getDouble(String key) {
		return get(key, Double.class);
	}

	public boolean getBool(String key, boolean def) {
		Boolean i = get(key, Boolean.class);
		return i == null? def: i;
	}

	public long getLong(String key, long def) {
		Long i = get(key, Long.class);
		return i == null? def: i;
	}

	public String getString(String key, String def) {
		String i = get(key, String.class);
		return i == null? def: i;
	}

	public double getDouble(String key, double def) {
		Double i = get(key, Double.class);
		return i == null? def: i;
	}

	public void put(String key, Object o) {
		map.put(key, o);
	}

	public void putAll(Map<String, ? extends Object> props) {
		if (props == null || props.size() == 0) {
			return;
		}

		this.map.putAll(props);
	}

	public void putAll(TypedDictionary other) {
		if (other == null || other.size() == 0) {
			return;
		}

		this.map.putAll(other.map);
	}

	public int size() {
		return map.size();
	}
}
import com.turn.sorcerer.exception.SorcererException;
import com.turn.sorcerer.executor.TaskExecutionResult.ExecutionStatus;
import com.turn.sorcerer.metrics.MetricUnit;
import com.turn.sorcerer.metrics.MetricsMonitor;
import com.turn.sorcerer.status.Status;
import com.turn.sorcerer.status.StatusManager;
import com.turn.sorcerer.task.Context;
import com.turn.sorcerer.task.executable.ExecutableTask;
import com.turn.sorcerer.task.executable.TaskFactory;
import com.turn.sorcerer.task.type.TaskType;
import com.turn.sorcerer.util.Constants;
import com.turn.sorcerer.util.TypedDictionary;
import java.text.SimpleDateFormat;
import java.util.Map;
import java.util.concurrent.Callable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/*
 * Copyright (c) 2015, Turn Inc. All Rights Reserved.
 * Use of this source code is governed by a BSD-style license that can be found
 * in the LICENSE file.
 */

package com.turn.sorcerer.executor;




/**
 * Class Description Here
 *
 * @author tshiou
 */
public class TaskExecutor implements Callable<TaskExecutionResult>, Abortable {
	private static final Logger LOGGER =
			LoggerFactory.getLogger(TaskExecutor.class);

	private final TaskType taskType;
	private boolean adhoc = false;

	// dashboard monitor keys for every task
	public static final String TASK_START_TIME_METRIC = "Task_Start_Time";
	public static final String COMPLETION_TIME_METRIC = "Task_Completion_Time";
	public static final String SUCCESSFUL_FINISH_TIME_METRIC = "Successful_Finish_Time";

	protected long taskStartTime = 0;
	protected long taskFinishTime = 0;

	private ExecutableTask task = null;
	private int jobId;
	private boolean abortSignalReceived = false;
	protected final SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy_MM_dd");

	private TypedDictionary taskArgs;

	public TaskExecutor(TaskType t, int jobId, Map<String, String> taskArgs, boolean adhoc) {
		this.taskType = t;

		this.taskArgs = new TypedDictionary();
		this.taskArgs.putAll(taskArgs);
		this.taskArgs.put(Constants.ADHOC, adhoc);
		this.jobId = jobId;
		this.adhoc = adhoc;
	}

	@Override
	public TaskExecutionResult call() throws Exception {

		LOGGER.debug("Attempting to start task {}", taskType.getName());

		TaskExecutionResult status = new TaskExecutionResult();
		status.setTask(taskType);
		status.setStatus(TaskExecutionResult.ExecutionStatus.SUCCESS);

		// Get instance of task
		this.task = TaskFactory.get().getExecutableTask(this.taskType, this.jobId);

		Context context = new Context(jobId, taskArgs);