hyracks/hyracks-control/hyracks-control-cc/src/main/java/edu/uci/ics/hyracks/control/cc/job/JobRun.java - asterixdb - Git at Google

 /*
  * Copyright 2009-2010 by The Regents of the University of California
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * you may obtain a copy of the License from
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
 package edu.uci.ics.hyracks.control.cc.job;

 import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;

 import org.json.JSONArray;
 import org.json.JSONException;
 import org.json.JSONObject;

 import edu.uci.ics.hyracks.api.dataflow.ActivityId;
 import edu.uci.ics.hyracks.api.dataflow.ConnectorDescriptorId;
 import edu.uci.ics.hyracks.api.dataflow.TaskId;
 import edu.uci.ics.hyracks.api.dataflow.connectors.IConnectorPolicy;
 import edu.uci.ics.hyracks.api.deployment.DeploymentId;
 import edu.uci.ics.hyracks.api.exceptions.HyracksException;
 import edu.uci.ics.hyracks.api.job.ActivityCluster;
 import edu.uci.ics.hyracks.api.job.ActivityClusterGraph;
 import edu.uci.ics.hyracks.api.job.ActivityClusterId;
 import edu.uci.ics.hyracks.api.job.IActivityClusterGraphGenerator;
 import edu.uci.ics.hyracks.api.job.JobFlag;
 import edu.uci.ics.hyracks.api.job.JobId;
 import edu.uci.ics.hyracks.api.job.JobStatus;
 import edu.uci.ics.hyracks.api.partitions.PartitionId;
 import edu.uci.ics.hyracks.control.cc.ClusterControllerService;
 import edu.uci.ics.hyracks.control.cc.partitions.PartitionMatchMaker;
 import edu.uci.ics.hyracks.control.cc.scheduler.ActivityPartitionDetails;
 import edu.uci.ics.hyracks.control.cc.scheduler.JobScheduler;
 import edu.uci.ics.hyracks.control.common.job.profiling.om.JobProfile;

 public class JobRun implements IJobStatusConditionVariable {
     private final DeploymentId deploymentId;

     private final JobId jobId;

     private final IActivityClusterGraphGenerator acgg;

     private final ActivityClusterGraph acg;

     private final JobScheduler scheduler;

     private final EnumSet<JobFlag> jobFlags;

     private final Map<ActivityClusterId, ActivityClusterPlan> activityClusterPlanMap;

     private final PartitionMatchMaker pmm;

     private final Set<String> participatingNodeIds;

     private final Set<String> cleanupPendingNodeIds;

     private final JobProfile profile;

     private final Map<ConnectorDescriptorId, IConnectorPolicy> connectorPolicyMap;

     private long createTime;

     private long startTime;

     private long endTime;

     private JobStatus status;

     private Exception exception;

     private JobStatus pendingStatus;

     private Exception pendingException;

     public JobRun(ClusterControllerService ccs, DeploymentId deploymentId, JobId jobId,
             IActivityClusterGraphGenerator acgg, EnumSet<JobFlag> jobFlags) {
         this.deploymentId = deploymentId;
         this.jobId = jobId;
         this.acgg = acgg;
         this.acg = acgg.initialize();
         this.scheduler = new JobScheduler(ccs, this, acgg.getConstraints());
         this.jobFlags = jobFlags;
         activityClusterPlanMap = new HashMap<ActivityClusterId, ActivityClusterPlan>();
         pmm = new PartitionMatchMaker();
         participatingNodeIds = new HashSet<String>();
         cleanupPendingNodeIds = new HashSet<String>();
         profile = new JobProfile(jobId);
         connectorPolicyMap = new HashMap<ConnectorDescriptorId, IConnectorPolicy>();
     }

     public DeploymentId getDeploymentId() {
         return deploymentId;
     }

     public JobId getJobId() {
         return jobId;
     }

     public ActivityClusterGraph getActivityClusterGraph() {
         return acg;
     }

     public EnumSet<JobFlag> getFlags() {
         return jobFlags;
     }

     public Map<ActivityClusterId, ActivityClusterPlan> getActivityClusterPlanMap() {
         return activityClusterPlanMap;
     }

     public PartitionMatchMaker getPartitionMatchMaker() {
         return pmm;
     }

     public synchronized void setStatus(JobStatus status, Exception exception) {
         this.status = status;
         this.exception = exception;
         notifyAll();
     }

     public synchronized JobStatus getStatus() {
         return status;
     }

     public synchronized Exception getException() {
         return exception;
     }

     public void setPendingStatus(JobStatus status, Exception exception) {
         this.pendingStatus = status;
         this.pendingException = exception;
     }

     public JobStatus getPendingStatus() {
         return pendingStatus;
     }

     public synchronized Exception getPendingException() {
         return pendingException;
     }

     public long getCreateTime() {
         return createTime;
     }

     public void setCreateTime(long createTime) {
         this.createTime = createTime;
     }

     public long getStartTime() {
         return startTime;
     }

     public void setStartTime(long startTime) {
         this.startTime = startTime;
     }

     public long getEndTime() {
         return endTime;
     }

     public void setEndTime(long endTime) {
         this.endTime = endTime;
     }

     @Override
     public synchronized void waitForCompletion() throws Exception {
         while (status != JobStatus.TERMINATED && status != JobStatus.FAILURE) {
             wait();
         }
         if (exception != null) {
             throw new HyracksException("Job Failed", exception);
         }
     }

     public Set<String> getParticipatingNodeIds() {
         return participatingNodeIds;
     }

     public Set<String> getCleanupPendingNodeIds() {
         return cleanupPendingNodeIds;
     }

     public JobProfile getJobProfile() {
         return profile;
     }

     public JobScheduler getScheduler() {
         return scheduler;
     }

     public Map<ConnectorDescriptorId, IConnectorPolicy> getConnectorPolicyMap() {
         return connectorPolicyMap;
     }

     public JSONObject toJSON() throws JSONException {
         JSONObject result = new JSONObject();

         result.put("job-id", jobId.toString());
         result.put("status", getStatus());
         result.put("create-time", getCreateTime());
         result.put("start-time", getCreateTime());
         result.put("end-time", getCreateTime());

         JSONArray aClusters = new JSONArray();
         for (ActivityCluster ac : acg.getActivityClusterMap().values()) {
             JSONObject acJSON = new JSONObject();

             acJSON.put("activity-cluster-id", String.valueOf(ac.getId()));

             JSONArray activitiesJSON = new JSONArray();
             for (ActivityId aid : ac.getActivityMap().keySet()) {
                 activitiesJSON.put(aid);
             }
             acJSON.put("activities", activitiesJSON);

             JSONArray dependenciesJSON = new JSONArray();
             for (ActivityCluster dependency : ac.getDependencies()) {
                 dependenciesJSON.put(String.valueOf(dependency.getId()));
             }
             acJSON.put("dependencies", dependenciesJSON);

             ActivityClusterPlan acp = activityClusterPlanMap.get(ac.getId());
             if (acp == null) {
                 acJSON.put("plan", (Object) null);
             } else {
                 JSONObject planJSON = new JSONObject();

                 JSONArray acTasks = new JSONArray();
                 for (Map.Entry<ActivityId, ActivityPlan> e : acp.getActivityPlanMap().entrySet()) {
                     ActivityPlan acPlan = e.getValue();
                     JSONObject entry = new JSONObject();
                     entry.put("activity-id", e.getKey().toString());

                     ActivityPartitionDetails apd = acPlan.getActivityPartitionDetails();
                     entry.put("partition-count", apd.getPartitionCount());

                     JSONArray inPartCountsJSON = new JSONArray();
                     int[] inPartCounts = apd.getInputPartitionCounts();
                     if (inPartCounts != null) {
                         for (int i : inPartCounts) {
                             inPartCountsJSON.put(i);
                         }
                     }
                     entry.put("input-partition-counts", inPartCountsJSON);

                     JSONArray outPartCountsJSON = new JSONArray();
                     int[] outPartCounts = apd.getOutputPartitionCounts();
                     if (outPartCounts != null) {
                         for (int o : outPartCounts) {
                             outPartCountsJSON.put(o);
                         }
                     }
                     entry.put("output-partition-counts", outPartCountsJSON);

                     JSONArray tasks = new JSONArray();
                     for (Task t : acPlan.getTasks()) {
                         JSONObject task = new JSONObject();

                         task.put("task-id", t.getTaskId().toString());

                         JSONArray dependentTasksJSON = new JSONArray();
                         for (TaskId dependent : t.getDependents()) {
                             dependentTasksJSON.put(dependent.toString());
                         }
                         task.put("dependents", dependentTasksJSON);

                         JSONArray dependencyTasksJSON = new JSONArray();
                         for (TaskId dependency : t.getDependencies()) {
                             dependencyTasksJSON.put(dependency.toString());
                         }
                         task.put("dependencies", dependencyTasksJSON);

                         tasks.put(task);
                     }
                     entry.put("tasks", tasks);

                     acTasks.put(entry);
                 }
                 planJSON.put("activities", acTasks);

                 JSONArray tClusters = new JSONArray();
                 for (TaskCluster tc : acp.getTaskClusters()) {
                     JSONObject c = new JSONObject();
                     c.put("task-cluster-id", String.valueOf(tc.getTaskClusterId()));

                     JSONArray tasks = new JSONArray();
                     for (Task t : tc.getTasks()) {
                         tasks.put(t.getTaskId().toString());
                     }
                     c.put("tasks", tasks);

                     JSONArray prodParts = new JSONArray();
                     for (PartitionId p : tc.getProducedPartitions()) {
                         prodParts.put(p.toString());
                     }
                     c.put("produced-partitions", prodParts);

                     JSONArray reqdParts = new JSONArray();
                     for (PartitionId p : tc.getRequiredPartitions()) {
                         reqdParts.put(p.toString());
                     }
                     c.put("required-partitions", reqdParts);

                     JSONArray attempts = new JSONArray();
                     List<TaskClusterAttempt> tcAttempts = tc.getAttempts();
                     if (tcAttempts != null) {
                         for (TaskClusterAttempt tca : tcAttempts) {
                             JSONObject attempt = new JSONObject();
                             attempt.put("attempt", tca.getAttempt());
                             attempt.put("status", tca.getStatus());
                             attempt.put("start-time", tca.getStartTime());
                             attempt.put("end-time", tca.getEndTime());

                             JSONArray taskAttempts = new JSONArray();
                             for (TaskAttempt ta : tca.getTaskAttempts().values()) {
                                 JSONObject taskAttempt = new JSONObject();
                                 taskAttempt.put("task-id", ta.getTaskAttemptId().getTaskId());
                                 taskAttempt.put("task-attempt-id", ta.getTaskAttemptId());
                                 taskAttempt.put("status", ta.getStatus());
                                 taskAttempt.put("node-id", ta.getNodeId());
                                 taskAttempt.put("start-time", ta.getStartTime());
                                 taskAttempt.put("end-time", ta.getEndTime());
                                 String failureDetails = ta.getFailureDetails();
                                 if (failureDetails != null) {
                                     taskAttempt.put("failure-details", failureDetails);
                                 }
                                 taskAttempts.put(taskAttempt);
                             }
                             attempt.put("task-attempts", taskAttempts);

                             attempts.put(attempt);
                         }
                     }
                     c.put("attempts", attempts);

                     tClusters.put(c);
                 }
                 planJSON.put("task-clusters", tClusters);

                 acJSON.put("plan", planJSON);
             }
             aClusters.put(acJSON);
         }
         result.put("activity-clusters", aClusters);

         result.put("profile", profile.toJSON());

         return result;
     }
 }
	/*
	* Copyright 2009-2010 by The Regents of the University of California
	* Licensed under the Apache License, Version 2.0 (the "License");
	* you may not use this file except in compliance with the License.
	* you may obtain a copy of the License from
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/
	package edu.uci.ics.hyracks.control.cc.job;

	import java.util.EnumSet;
	import java.util.HashMap;
	import java.util.HashSet;
	import java.util.List;
	import java.util.Map;
	import java.util.Set;

	import org.json.JSONArray;
	import org.json.JSONException;
	import org.json.JSONObject;

	import edu.uci.ics.hyracks.api.dataflow.ActivityId;
	import edu.uci.ics.hyracks.api.dataflow.ConnectorDescriptorId;
	import edu.uci.ics.hyracks.api.dataflow.TaskId;
	import edu.uci.ics.hyracks.api.dataflow.connectors.IConnectorPolicy;
	import edu.uci.ics.hyracks.api.deployment.DeploymentId;
	import edu.uci.ics.hyracks.api.exceptions.HyracksException;
	import edu.uci.ics.hyracks.api.job.ActivityCluster;
	import edu.uci.ics.hyracks.api.job.ActivityClusterGraph;
	import edu.uci.ics.hyracks.api.job.ActivityClusterId;
	import edu.uci.ics.hyracks.api.job.IActivityClusterGraphGenerator;
	import edu.uci.ics.hyracks.api.job.JobFlag;
	import edu.uci.ics.hyracks.api.job.JobId;
	import edu.uci.ics.hyracks.api.job.JobStatus;
	import edu.uci.ics.hyracks.api.partitions.PartitionId;
	import edu.uci.ics.hyracks.control.cc.ClusterControllerService;
	import edu.uci.ics.hyracks.control.cc.partitions.PartitionMatchMaker;
	import edu.uci.ics.hyracks.control.cc.scheduler.ActivityPartitionDetails;
	import edu.uci.ics.hyracks.control.cc.scheduler.JobScheduler;
	import edu.uci.ics.hyracks.control.common.job.profiling.om.JobProfile;

	public class JobRun implements IJobStatusConditionVariable {
	private final DeploymentId deploymentId;

	private final JobId jobId;

	private final IActivityClusterGraphGenerator acgg;

	private final ActivityClusterGraph acg;

	private final JobScheduler scheduler;

	private final EnumSet<JobFlag> jobFlags;

	private final Map<ActivityClusterId, ActivityClusterPlan> activityClusterPlanMap;

	private final PartitionMatchMaker pmm;

	private final Set<String> participatingNodeIds;

	private final Set<String> cleanupPendingNodeIds;

	private final JobProfile profile;

	private final Map<ConnectorDescriptorId, IConnectorPolicy> connectorPolicyMap;

	private long createTime;

	private long startTime;

	private long endTime;

	private JobStatus status;

	private Exception exception;

	private JobStatus pendingStatus;

	private Exception pendingException;

	public JobRun(ClusterControllerService ccs, DeploymentId deploymentId, JobId jobId,
	IActivityClusterGraphGenerator acgg, EnumSet<JobFlag> jobFlags) {
	this.deploymentId = deploymentId;
	this.jobId = jobId;
	this.acgg = acgg;
	this.acg = acgg.initialize();
	this.scheduler = new JobScheduler(ccs, this, acgg.getConstraints());
	this.jobFlags = jobFlags;
	activityClusterPlanMap = new HashMap<ActivityClusterId, ActivityClusterPlan>();
	pmm = new PartitionMatchMaker();
	participatingNodeIds = new HashSet<String>();
	cleanupPendingNodeIds = new HashSet<String>();
	profile = new JobProfile(jobId);
	connectorPolicyMap = new HashMap<ConnectorDescriptorId, IConnectorPolicy>();
	}

	public DeploymentId getDeploymentId() {
	return deploymentId;
	}

	public JobId getJobId() {
	return jobId;
	}

	public ActivityClusterGraph getActivityClusterGraph() {
	return acg;
	}

	public EnumSet<JobFlag> getFlags() {
	return jobFlags;
	}

	public Map<ActivityClusterId, ActivityClusterPlan> getActivityClusterPlanMap() {
	return activityClusterPlanMap;
	}

	public PartitionMatchMaker getPartitionMatchMaker() {
	return pmm;
	}

	public synchronized void setStatus(JobStatus status, Exception exception) {
	this.status = status;
	this.exception = exception;
	notifyAll();
	}

	public synchronized JobStatus getStatus() {
	return status;
	}

	public synchronized Exception getException() {
	return exception;
	}

	public void setPendingStatus(JobStatus status, Exception exception) {
	this.pendingStatus = status;
	this.pendingException = exception;
	}

	public JobStatus getPendingStatus() {
	return pendingStatus;
	}

	public synchronized Exception getPendingException() {
	return pendingException;
	}

	public long getCreateTime() {
	return createTime;
	}

	public void setCreateTime(long createTime) {
	this.createTime = createTime;
	}

	public long getStartTime() {
	return startTime;
	}

	public void setStartTime(long startTime) {
	this.startTime = startTime;
	}

	public long getEndTime() {
	return endTime;
	}

	public void setEndTime(long endTime) {
	this.endTime = endTime;
	}

	@Override
	public synchronized void waitForCompletion() throws Exception {
	while (status != JobStatus.TERMINATED && status != JobStatus.FAILURE) {
	wait();
	}
	if (exception != null) {
	throw new HyracksException("Job Failed", exception);
	}
	}

	public Set<String> getParticipatingNodeIds() {
	return participatingNodeIds;
	}

	public Set<String> getCleanupPendingNodeIds() {
	return cleanupPendingNodeIds;
	}

	public JobProfile getJobProfile() {
	return profile;
	}

	public JobScheduler getScheduler() {
	return scheduler;
	}

	public Map<ConnectorDescriptorId, IConnectorPolicy> getConnectorPolicyMap() {
	return connectorPolicyMap;
	}

	public JSONObject toJSON() throws JSONException {
	JSONObject result = new JSONObject();

	result.put("job-id", jobId.toString());
	result.put("status", getStatus());
	result.put("create-time", getCreateTime());
	result.put("start-time", getCreateTime());
	result.put("end-time", getCreateTime());

	JSONArray aClusters = new JSONArray();
	for (ActivityCluster ac : acg.getActivityClusterMap().values()) {
	JSONObject acJSON = new JSONObject();

	acJSON.put("activity-cluster-id", String.valueOf(ac.getId()));

	JSONArray activitiesJSON = new JSONArray();
	for (ActivityId aid : ac.getActivityMap().keySet()) {
	activitiesJSON.put(aid);
	}
	acJSON.put("activities", activitiesJSON);

	JSONArray dependenciesJSON = new JSONArray();
	for (ActivityCluster dependency : ac.getDependencies()) {
	dependenciesJSON.put(String.valueOf(dependency.getId()));
	}
	acJSON.put("dependencies", dependenciesJSON);

	ActivityClusterPlan acp = activityClusterPlanMap.get(ac.getId());
	if (acp == null) {
	acJSON.put("plan", (Object) null);
	} else {
	JSONObject planJSON = new JSONObject();

	JSONArray acTasks = new JSONArray();
	for (Map.Entry<ActivityId, ActivityPlan> e : acp.getActivityPlanMap().entrySet()) {
	ActivityPlan acPlan = e.getValue();
	JSONObject entry = new JSONObject();
	entry.put("activity-id", e.getKey().toString());

	ActivityPartitionDetails apd = acPlan.getActivityPartitionDetails();
	entry.put("partition-count", apd.getPartitionCount());

	JSONArray inPartCountsJSON = new JSONArray();
	int[] inPartCounts = apd.getInputPartitionCounts();
	if (inPartCounts != null) {
	for (int i : inPartCounts) {
	inPartCountsJSON.put(i);
	}
	}
	entry.put("input-partition-counts", inPartCountsJSON);

	JSONArray outPartCountsJSON = new JSONArray();
	int[] outPartCounts = apd.getOutputPartitionCounts();
	if (outPartCounts != null) {
	for (int o : outPartCounts) {
	outPartCountsJSON.put(o);
	}
	}
	entry.put("output-partition-counts", outPartCountsJSON);

	JSONArray tasks = new JSONArray();
	for (Task t : acPlan.getTasks()) {
	JSONObject task = new JSONObject();

	task.put("task-id", t.getTaskId().toString());

	JSONArray dependentTasksJSON = new JSONArray();
	for (TaskId dependent : t.getDependents()) {
	dependentTasksJSON.put(dependent.toString());
	}
	task.put("dependents", dependentTasksJSON);

	JSONArray dependencyTasksJSON = new JSONArray();
	for (TaskId dependency : t.getDependencies()) {
	dependencyTasksJSON.put(dependency.toString());
	}
	task.put("dependencies", dependencyTasksJSON);

	tasks.put(task);
	}
	entry.put("tasks", tasks);

	acTasks.put(entry);
	}
	planJSON.put("activities", acTasks);

	JSONArray tClusters = new JSONArray();
	for (TaskCluster tc : acp.getTaskClusters()) {
	JSONObject c = new JSONObject();
	c.put("task-cluster-id", String.valueOf(tc.getTaskClusterId()));

	JSONArray tasks = new JSONArray();
	for (Task t : tc.getTasks()) {
	tasks.put(t.getTaskId().toString());
	}
	c.put("tasks", tasks);

	JSONArray prodParts = new JSONArray();
	for (PartitionId p : tc.getProducedPartitions()) {
	prodParts.put(p.toString());
	}
	c.put("produced-partitions", prodParts);

	JSONArray reqdParts = new JSONArray();
	for (PartitionId p : tc.getRequiredPartitions()) {
	reqdParts.put(p.toString());
	}
	c.put("required-partitions", reqdParts);

	JSONArray attempts = new JSONArray();
	List<TaskClusterAttempt> tcAttempts = tc.getAttempts();
	if (tcAttempts != null) {
	for (TaskClusterAttempt tca : tcAttempts) {
	JSONObject attempt = new JSONObject();
	attempt.put("attempt", tca.getAttempt());
	attempt.put("status", tca.getStatus());
	attempt.put("start-time", tca.getStartTime());
	attempt.put("end-time", tca.getEndTime());

	JSONArray taskAttempts = new JSONArray();
	for (TaskAttempt ta : tca.getTaskAttempts().values()) {
	JSONObject taskAttempt = new JSONObject();
	taskAttempt.put("task-id", ta.getTaskAttemptId().getTaskId());
	taskAttempt.put("task-attempt-id", ta.getTaskAttemptId());
	taskAttempt.put("status", ta.getStatus());
	taskAttempt.put("node-id", ta.getNodeId());
	taskAttempt.put("start-time", ta.getStartTime());
	taskAttempt.put("end-time", ta.getEndTime());
	String failureDetails = ta.getFailureDetails();
	if (failureDetails != null) {
	taskAttempt.put("failure-details", failureDetails);
	}
	taskAttempts.put(taskAttempt);
	}
	attempt.put("task-attempts", taskAttempts);

	attempts.put(attempt);
	}
	}
	c.put("attempts", attempts);

	tClusters.put(c);
	}
	planJSON.put("task-clusters", tClusters);

	acJSON.put("plan", planJSON);
	}
	aClusters.put(acJSON);
	}
	result.put("activity-clusters", aClusters);

	result.put("profile", profile.toJSON());

	return result;
	}
	}