Repository: aurora Updated Branches: refs/heads/master 920263b0e -> 7fd920e0a
Exposing task reconciliation stats. Bugs closed: AURORA-1047 Reviewed at https://reviews.apache.org/r/34373/ Project: http://git-wip-us.apache.org/repos/asf/aurora/repo Commit: http://git-wip-us.apache.org/repos/asf/aurora/commit/7fd920e0 Tree: http://git-wip-us.apache.org/repos/asf/aurora/tree/7fd920e0 Diff: http://git-wip-us.apache.org/repos/asf/aurora/diff/7fd920e0 Branch: refs/heads/master Commit: 7fd920e0a5eed738e9ea019a18290c58b6051e59 Parents: 920263b Author: Maxim Khutornenko <[email protected]> Authored: Wed May 20 10:48:35 2015 -0700 Committer: Maxim Khutornenko <[email protected]> Committed: Wed May 20 10:48:35 2015 -0700 ---------------------------------------------------------------------- .../aurora/scheduler/UserTaskLauncher.java | 20 ++++++++++++++++++-- .../aurora/scheduler/UserTaskLauncherTest.java | 15 ++++++++++++--- 2 files changed, 30 insertions(+), 5 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/aurora/blob/7fd920e0/src/main/java/org/apache/aurora/scheduler/UserTaskLauncher.java ---------------------------------------------------------------------- diff --git a/src/main/java/org/apache/aurora/scheduler/UserTaskLauncher.java b/src/main/java/org/apache/aurora/scheduler/UserTaskLauncher.java index f1e5dd2..5af691d 100644 --- a/src/main/java/org/apache/aurora/scheduler/UserTaskLauncher.java +++ b/src/main/java/org/apache/aurora/scheduler/UserTaskLauncher.java @@ -36,7 +36,9 @@ import org.apache.aurora.gen.ScheduleStatus; import org.apache.aurora.scheduler.async.OfferManager; import org.apache.aurora.scheduler.base.Conversions; import org.apache.aurora.scheduler.mesos.Driver; +import org.apache.aurora.scheduler.state.StateChangeResult; import org.apache.aurora.scheduler.state.StateManager; +import org.apache.aurora.scheduler.stats.CachedCounters; import org.apache.aurora.scheduler.storage.Storage; import org.apache.mesos.Protos.OfferID; import org.apache.mesos.Protos.TaskStatus; @@ -61,12 +63,15 @@ public class UserTaskLauncher extends AbstractExecutionThreadService implements @VisibleForTesting static final String MEMORY_LIMIT_DISPLAY = "Task used more memory than requested."; + private static final String STATUS_STAT_FORMAT = "status_update_%s_%s"; + private final Storage storage; private final OfferManager offerManager; private final StateManager stateManager; private final Driver driver; private final BlockingQueue<TaskStatus> pendingUpdates; private final int maxBatchSize; + private final CachedCounters counters; private final AtomicReference<Thread> threadReference = new AtomicReference<>(); @@ -93,7 +98,8 @@ public class UserTaskLauncher extends AbstractExecutionThreadService implements StateManager stateManager, final Driver driver, @StatusUpdateQueue BlockingQueue<TaskStatus> pendingUpdates, - @MaxBatchSize Integer maxBatchSize) { + @MaxBatchSize Integer maxBatchSize, + CachedCounters counters) { this.storage = requireNonNull(storage); this.offerManager = requireNonNull(offerManager); @@ -101,6 +107,7 @@ public class UserTaskLauncher extends AbstractExecutionThreadService implements this.driver = requireNonNull(driver); this.pendingUpdates = requireNonNull(pendingUpdates); this.maxBatchSize = requireNonNull(maxBatchSize); + this.counters = requireNonNull(counters); Stats.exportSize("status_updates_queue_size", this.pendingUpdates); @@ -181,12 +188,16 @@ public class UserTaskLauncher extends AbstractExecutionThreadService implements message = Optional.of(MEMORY_LIMIT_DISPLAY); } - stateManager.changeState( + StateChangeResult result = stateManager.changeState( storeProvider, status.getTaskId().getValue(), Optional.<ScheduleStatus>absent(), translatedState, message); + + if (status.hasReason()) { + counters.get(statName(status, result)).incrementAndGet(); + } } } }); @@ -199,4 +210,9 @@ public class UserTaskLauncher extends AbstractExecutionThreadService implements } } } + + @VisibleForTesting + static String statName(TaskStatus status, StateChangeResult result) { + return String.format(STATUS_STAT_FORMAT, status.getReason(), result); + } } http://git-wip-us.apache.org/repos/asf/aurora/blob/7fd920e0/src/test/java/org/apache/aurora/scheduler/UserTaskLauncherTest.java ---------------------------------------------------------------------- diff --git a/src/test/java/org/apache/aurora/scheduler/UserTaskLauncherTest.java b/src/test/java/org/apache/aurora/scheduler/UserTaskLauncherTest.java index f4631c1..fe2fc95 100644 --- a/src/test/java/org/apache/aurora/scheduler/UserTaskLauncherTest.java +++ b/src/test/java/org/apache/aurora/scheduler/UserTaskLauncherTest.java @@ -29,9 +29,11 @@ import org.apache.aurora.scheduler.mesos.Driver; import org.apache.aurora.scheduler.mesos.Offers; import org.apache.aurora.scheduler.state.StateChangeResult; import org.apache.aurora.scheduler.state.StateManager; +import org.apache.aurora.scheduler.stats.CachedCounters; import org.apache.aurora.scheduler.storage.Storage.StorageException; import org.apache.aurora.scheduler.storage.entities.IHostAttributes; import org.apache.aurora.scheduler.storage.testing.StorageTestUtil; +import org.apache.aurora.scheduler.testing.FakeStatsProvider; import org.apache.mesos.Protos.OfferID; import org.apache.mesos.Protos.TaskID; import org.apache.mesos.Protos.TaskState; @@ -43,8 +45,10 @@ import org.junit.Test; import static org.apache.aurora.gen.ScheduleStatus.FAILED; import static org.apache.aurora.gen.ScheduleStatus.RUNNING; +import static org.apache.aurora.scheduler.UserTaskLauncher.statName; import static org.easymock.EasyMock.expect; import static org.easymock.EasyMock.expectLastCall; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; public class UserTaskLauncherTest extends EasyMockTest { @@ -61,6 +65,7 @@ public class UserTaskLauncherTest extends EasyMockTest { private StorageTestUtil storageUtil; private Driver driver; private BlockingQueue<TaskStatus> queue; + private FakeStatsProvider stats; private UserTaskLauncher launcher; @@ -71,6 +76,7 @@ public class UserTaskLauncherTest extends EasyMockTest { storageUtil = new StorageTestUtil(this); driver = createMock(Driver.class); queue = new LinkedBlockingQueue<>(); + stats = new FakeStatsProvider(); launcher = new UserTaskLauncher( storageUtil.storage, @@ -78,7 +84,8 @@ public class UserTaskLauncherTest extends EasyMockTest { stateManager, driver, queue, - 1000); + 1000, + new CachedCounters(stats)); launcher.startAsync(); } @@ -101,6 +108,7 @@ public class UserTaskLauncherTest extends EasyMockTest { public void testForwardsStatusUpdates() throws Exception { TaskStatus status = TaskStatus.newBuilder() .setState(TaskState.TASK_RUNNING) + .setReason(TaskStatus.Reason.REASON_RECONCILIATION) .setTaskId(TaskID.newBuilder().setValue(TASK_ID_A)) .setMessage("fake message") .build(); @@ -126,8 +134,8 @@ public class UserTaskLauncherTest extends EasyMockTest { control.replay(); assertTrue(launcher.statusUpdate(status)); - assertTrue(latch.await(5L, TimeUnit.SECONDS)); + assertEquals(1L, stats.getValue(statName(status, StateChangeResult.SUCCESS))); } @Test @@ -242,7 +250,8 @@ public class UserTaskLauncherTest extends EasyMockTest { stateManager, driver, queue, - 1000); + 1000, + new CachedCounters(stats)); expect(queue.add(EasyMock.<TaskStatus>anyObject())) .andReturn(true);
