Apache
/
druid
mirror of https://github.com/apache/druid.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Workers announce TaskAnnouncement rather than TaskStatus

This commit is contained in:
Gian Merlino 2013-08-20 16:14:36 -07:00
parent 9609314765
commit 455645e723
8 changed files with 175 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
indexing-service/src/main/java/com/metamx/druid/indexing/common/TaskStatus.java
View File

@ -27,8 +27,8 @@ import com.google.common.base.Preconditions;
import com.metamx.druid.indexing.common.task.TaskResource;
/**
* Represents the status of a task. The task may be ongoing ({@link #isComplete()} false) or it may be
* complete ({@link #isComplete()} true).
* Represents the status of a task from the perspective of the coordinator. The task may be ongoing
* ({@link #isComplete()} false) or it may be complete ({@link #isComplete()} true).
* <p/>
* TaskStatus objects are immutable.
*/
@ -43,36 +43,38 @@ public class TaskStatus
public static TaskStatus running(String taskId)
{
return new TaskStatus(taskId, Status.RUNNING, -1, null);
return new TaskStatus(taskId, Status.RUNNING, -1);
}
public static TaskStatus success(String taskId)
{
return new TaskStatus(taskId, Status.SUCCESS, -1, null);
return new TaskStatus(taskId, Status.SUCCESS, -1);
}
public static TaskStatus failure(String taskId)
{
return new TaskStatus(taskId, Status.FAILED, -1, null);
return new TaskStatus(taskId, Status.FAILED, -1);
}
public static TaskStatus fromCode(String taskId, Status code)
{
return new TaskStatus(taskId, code, -1);
}
private final String id;
private final Status status;
private final long duration;
private final TaskResource resource;
@JsonCreator
private TaskStatus(
@JsonProperty("id") String id,
@JsonProperty("status") Status status,
@JsonProperty("duration") long duration,
@JsonProperty("resource") TaskResource resource
@JsonProperty("duration") long duration
)
{
this.id = id;
this.status = status;
this.duration = duration;
this.resource = resource == null ? new TaskResource(id, 1) : resource;
// Check class invariants.
Preconditions.checkNotNull(id, "id");
@ -97,12 +99,6 @@ public class TaskStatus
return duration;
}
@JsonProperty("resource")
public TaskResource getResource()
{
return resource;
}
/**
* Signals that a task is not yet complete, and is still runnable on a worker. Exactly one of isRunnable,
* isSuccess, or isFailure will be true at any one time.
@ -144,7 +140,7 @@ public class TaskStatus
public TaskStatus withDuration(long _duration)
{
return new TaskStatus(id, status, _duration, resource);
return new TaskStatus(id, status, _duration);
}
@Override
@ -154,7 +150,6 @@ public class TaskStatus
.add("id", id)
.add("status", status)
.add("duration", duration)
.add("resource", resource)
.toString();
}
}

7
indexing-service/src/main/java/com/metamx/druid/indexing/coordinator/RemoteTaskRunner.java
View File

@ -41,6 +41,7 @@ import com.metamx.druid.indexing.common.task.Task;
import com.metamx.druid.indexing.common.tasklogs.TaskLogProvider;
import com.metamx.druid.indexing.coordinator.config.RemoteTaskRunnerConfig;
import com.metamx.druid.indexing.coordinator.setup.WorkerSetupData;
import com.metamx.druid.indexing.worker.TaskAnnouncement;
import com.metamx.druid.indexing.worker.Worker;
import com.metamx.emitter.EmittingLogger;
import com.metamx.http.client.HttpClient;
@ -291,9 +292,9 @@ public class RemoteTaskRunner implements TaskRunner, TaskLogProvider
runningTasks.remove(task.getId());
} else {
log.info("Task[%s] already running on %s.", task.getId(), zkWorker.getWorker().getHost());
TaskStatus status = zkWorker.getRunningTasks().get(task.getId());
if (status.isComplete()) {
taskComplete(runningTask, zkWorker, task.getId(), status);
TaskAnnouncement announcement = zkWorker.getRunningTasks().get(task.getId());
if (announcement.getTaskStatus().isComplete()) {
taskComplete(runningTask, zkWorker, task.getId(), announcement.getTaskStatus());
}
return runningTask.getResult();
}

25
indexing-service/src/main/java/com/metamx/druid/indexing/coordinator/ZkWorker.java
View File

@ -28,6 +28,7 @@ import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.metamx.druid.indexing.common.TaskStatus;
import com.metamx.druid.indexing.common.task.Task;
import com.metamx.druid.indexing.worker.TaskAnnouncement;
import com.metamx.druid.indexing.worker.Worker;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
@ -48,7 +49,7 @@ public class ZkWorker implements Closeable
{
private final Worker worker;
private final PathChildrenCache statusCache;
private final Function<ChildData, TaskStatus> cacheConverter;
private final Function<ChildData, TaskAnnouncement> cacheConverter;
private AtomicReference<DateTime> lastCompletedTaskTime = new AtomicReference<DateTime>(new DateTime());
@ -56,13 +57,13 @@ public class ZkWorker implements Closeable
{
this.worker = worker;
this.statusCache = statusCache;
this.cacheConverter = new Function<ChildData, TaskStatus>()
this.cacheConverter = new Function<ChildData, TaskAnnouncement>()
{
@Override
public TaskStatus apply(ChildData input)
public TaskAnnouncement apply(ChildData input)
{
try {
return jsonMapper.readValue(input.getData(), TaskStatus.class);
return jsonMapper.readValue(input.getData(), TaskAnnouncement.class);
}
catch (Exception e) {
throw Throwables.propagate(e);
@ -93,14 +94,14 @@ public class ZkWorker implements Closeable
return getRunningTasks().keySet();
}
public Map<String, TaskStatus> getRunningTasks()
public Map<String, TaskAnnouncement> getRunningTasks()
{
Map<String, TaskStatus> retVal = Maps.newHashMap();
for (TaskStatus taskStatus : Lists.transform(
Map<String, TaskAnnouncement> retVal = Maps.newHashMap();
for (TaskAnnouncement taskAnnouncement : Lists.transform(
statusCache.getCurrentData(),
cacheConverter
)) {
retVal.put(taskStatus.getId(), taskStatus);
retVal.put(taskAnnouncement.getTaskStatus().getId(), taskAnnouncement);
}
return retVal;
@ -110,8 +111,8 @@ public class ZkWorker implements Closeable
public int getCurrCapacityUsed()
{
int currCapacity = 0;
for (TaskStatus taskStatus : getRunningTasks().values()) {
currCapacity += taskStatus.getResource().getRequiredCapacity();
for (TaskAnnouncement taskAnnouncement : getRunningTasks().values()) {
currCapacity += taskAnnouncement.getTaskResource().getRequiredCapacity();
}
return currCapacity;
}
@ -120,8 +121,8 @@ public class ZkWorker implements Closeable
public Set<String> getAvailabilityGroups()
{
Set<String> retVal = Sets.newHashSet();
for (TaskStatus taskStatus : getRunningTasks().values()) {
retVal.add(taskStatus.getResource().getAvailabilityGroup());
for (TaskAnnouncement taskAnnouncement : getRunningTasks().values()) {
retVal.add(taskAnnouncement.getTaskResource().getAvailabilityGroup());
}
return retVal;
}

68
indexing-service/src/main/java/com/metamx/druid/indexing/worker/TaskAnnouncement.java Normal file
View File

@ -0,0 +1,68 @@
package com.metamx.druid.indexing.worker;
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.base.Preconditions;
import com.metamx.druid.indexing.common.TaskStatus;
import com.metamx.druid.indexing.common.task.Task;
import com.metamx.druid.indexing.common.task.TaskResource;
/**
* Used by workers to announce the status of tasks they are currently running. This class is immutable.
*/
public class TaskAnnouncement
{
private final TaskStatus taskStatus;
private final TaskResource taskResource;
public static TaskAnnouncement create(Task task, TaskStatus status)
{
Preconditions.checkArgument(status.getId().equals(task.getId()), "task id == status id");
return new TaskAnnouncement(null, null, status, task.getTaskResource());
}
@JsonCreator
private TaskAnnouncement(
@JsonProperty("id") String taskId,
@JsonProperty("status") TaskStatus.Status status,
@JsonProperty("taskStatus") TaskStatus taskStatus,
@JsonProperty("taskResource") TaskResource taskResource
)
{
if (taskStatus != null) {
this.taskStatus = taskStatus;
} else {
// Can be removed when backwards compat is no longer needed
this.taskStatus = TaskStatus.fromCode(taskId, status);
}
this.taskResource = taskResource == null ? new TaskResource(this.taskStatus.getId(), 1) : taskResource;
}
// Can be removed when backwards compat is no longer needed
@JsonProperty("id")
@Deprecated
public String getTaskId()
{
return taskStatus.getId();
}
// Can be removed when backwards compat is no longer needed
@JsonProperty("status")
@Deprecated
public TaskStatus.Status getStatus()
{
return taskStatus.getStatusCode();
}
@JsonProperty("taskStatus")
public TaskStatus getTaskStatus()
{
return taskStatus;
}
@JsonProperty("taskResource")
public TaskResource getTaskResource()
{
return taskResource;
}
}

18
indexing-service/src/main/java/com/metamx/druid/indexing/worker/WorkerCuratorCoordinator.java
View File

@ -180,7 +180,7 @@ public class WorkerCuratorCoordinator
}
}
public void announceStatus(TaskStatus status)
public void announceTask(TaskAnnouncement announcement)
{
synchronized (lock) {
if (!started) {
@ -188,7 +188,7 @@ public class WorkerCuratorCoordinator
}
try {
byte[] rawBytes = jsonMapper.writeValueAsBytes(status);
byte[] rawBytes = jsonMapper.writeValueAsBytes(announcement);
if (rawBytes.length > config.getMaxNumBytes()) {
throw new ISE("Length of raw bytes for task too large[%,d > %,d]", rawBytes.length, config.getMaxNumBytes());
}
@ -196,7 +196,7 @@ public class WorkerCuratorCoordinator
curatorFramework.create()
.withMode(CreateMode.EPHEMERAL)
.forPath(
getStatusPathForId(status.getId()), rawBytes
getStatusPathForId(announcement.getTaskStatus().getId()), rawBytes
);
}
catch (Exception e) {
@ -205,7 +205,7 @@ public class WorkerCuratorCoordinator
}
}
public void updateStatus(TaskStatus status)
public void updateAnnouncement(TaskAnnouncement announcement)
{
synchronized (lock) {
if (!started) {
@ -213,18 +213,18 @@ public class WorkerCuratorCoordinator
}
try {
if (curatorFramework.checkExists().forPath(getStatusPathForId(status.getId())) == null) {
announceStatus(status);
if (curatorFramework.checkExists().forPath(getStatusPathForId(announcement.getTaskStatus().getId())) == null) {
announceTask(announcement);
return;
}
byte[] rawBytes = jsonMapper.writeValueAsBytes(status);
byte[] rawBytes = jsonMapper.writeValueAsBytes(announcement);
if (rawBytes.length > config.getMaxNumBytes()) {
throw new ISE("Length of raw bytes for task too large[%,d > %,d]", rawBytes.length, config.getMaxNumBytes());
}
curatorFramework.setData()
.forPath(
getStatusPathForId(status.getId()), rawBytes
getStatusPathForId(announcement.getTaskStatus().getId()), rawBytes
);
}
catch (Exception e) {
@ -232,4 +232,4 @@ public class WorkerCuratorCoordinator
}
}
}
}
}

9
indexing-service/src/main/java/com/metamx/druid/indexing/worker/WorkerTaskMonitor.java
View File

@ -118,7 +118,12 @@ public class WorkerTaskMonitor
TaskStatus taskStatus;
try {
workerCuratorCoordinator.unannounceTask(task.getId());
workerCuratorCoordinator.announceStatus(TaskStatus.running(task.getId()));
workerCuratorCoordinator.announceTask(
TaskAnnouncement.create(
task,
TaskStatus.running(task.getId())
)
);
taskStatus = taskRunner.run(task).get();
}
catch (Exception e) {
@ -134,7 +139,7 @@ public class WorkerTaskMonitor
taskStatus = taskStatus.withDuration(System.currentTimeMillis() - startTime);
try {
workerCuratorCoordinator.updateStatus(taskStatus);
workerCuratorCoordinator.updateAnnouncement(TaskAnnouncement.create(task, taskStatus));
log.info(
"Job's finished. Completed [%s] with status [%s]",
task.getId(),

5
indexing-service/src/test/java/com/metamx/druid/indexing/coordinator/scaling/SimpleResourceManagementStrategyTest.java
View File

@ -31,6 +31,7 @@ import com.metamx.druid.indexing.common.task.Task;
import com.metamx.druid.indexing.coordinator.RemoteTaskRunnerWorkItem;
import com.metamx.druid.indexing.coordinator.ZkWorker;
import com.metamx.druid.indexing.coordinator.setup.WorkerSetupData;
import com.metamx.druid.indexing.worker.TaskAnnouncement;
import com.metamx.druid.indexing.worker.Worker;
import com.metamx.druid.jackson.DefaultObjectMapper;
import com.metamx.emitter.EmittingLogger;
@ -351,12 +352,12 @@ public class SimpleResourceManagementStrategyTest
}
@Override
public Map<String, TaskStatus> getRunningTasks()
public Map<String, TaskAnnouncement> getRunningTasks()
{
if (testTask == null) {
return Maps.newHashMap();
}
return ImmutableMap.of(testTask.getId(), TaskStatus.running(testTask.getId()));
return ImmutableMap.of(testTask.getId(), TaskAnnouncement.create(testTask, TaskStatus.running(testTask.getId())));
}
}
}

59
indexing-service/src/test/java/com/metamx/druid/indexing/worker/TaskAnnouncementTest.java Normal file
View File

@ -0,0 +1,59 @@
package com.metamx.druid.indexing.worker;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.metamx.druid.QueryGranularity;
import com.metamx.druid.aggregation.AggregatorFactory;
import com.metamx.druid.index.v1.IndexGranularity;
import com.metamx.druid.indexing.common.TaskStatus;
import com.metamx.druid.indexing.common.task.RealtimeIndexTask;
import com.metamx.druid.indexing.common.task.Task;
import com.metamx.druid.indexing.common.task.TaskResource;
import com.metamx.druid.jackson.DefaultObjectMapper;
import com.metamx.druid.realtime.Schema;
import com.metamx.druid.shard.NoneShardSpec;
import junit.framework.Assert;
import org.joda.time.Period;
import org.junit.Test;
public class TaskAnnouncementTest
{
@Test
public void testBackwardsCompatibleSerde() throws Exception
{
final Task task = new RealtimeIndexTask(
"theid",
new TaskResource("rofl", 2),
new Schema("foo", null, new AggregatorFactory[0], QueryGranularity.NONE, new NoneShardSpec()),
null,
null,
new Period("PT10M"),
IndexGranularity.HOUR,
null
);
final TaskStatus status = TaskStatus.running(task.getId());
final TaskAnnouncement announcement = TaskAnnouncement.create(task, status);
final ObjectMapper jsonMapper = new DefaultObjectMapper();
final String statusJson = jsonMapper.writeValueAsString(status);
final String announcementJson = jsonMapper.writeValueAsString(announcement);
final TaskStatus statusFromStatus = jsonMapper.readValue(statusJson, TaskStatus.class);
final TaskStatus statusFromAnnouncement = jsonMapper.readValue(announcementJson, TaskStatus.class);
final TaskAnnouncement announcementFromStatus = jsonMapper.readValue(statusJson, TaskAnnouncement.class);
final TaskAnnouncement announcementFromAnnouncement = jsonMapper.readValue(
announcementJson,
TaskAnnouncement.class
);
Assert.assertEquals("theid", statusFromStatus.getId());
Assert.assertEquals("theid", statusFromAnnouncement.getId());
Assert.assertEquals("theid", announcementFromStatus.getTaskStatus().getId());
Assert.assertEquals("theid", announcementFromAnnouncement.getTaskStatus().getId());
Assert.assertEquals("theid", announcementFromStatus.getTaskResource().getAvailabilityGroup());
Assert.assertEquals("rofl", announcementFromAnnouncement.getTaskResource().getAvailabilityGroup());
Assert.assertEquals(1, announcementFromStatus.getTaskResource().getRequiredCapacity());
Assert.assertEquals(2, announcementFromAnnouncement.getTaskResource().getRequiredCapacity());
}
}