You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@aurora.apache.org by wf...@apache.org on 2014/09/03 18:17:55 UTC
git commit: Add a separate main class that runs the scheduler in
local mode.
Repository: incubator-aurora
Updated Branches:
refs/heads/master d64f5a353 -> 417cb169c
Add a separate main class that runs the scheduler in local mode.
Bugs closed: AURORA-658
Reviewed at https://reviews.apache.org/r/25257/
Project: http://git-wip-us.apache.org/repos/asf/incubator-aurora/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-aurora/commit/417cb169
Tree: http://git-wip-us.apache.org/repos/asf/incubator-aurora/tree/417cb169
Diff: http://git-wip-us.apache.org/repos/asf/incubator-aurora/diff/417cb169
Branch: refs/heads/master
Commit: 417cb169cc18bda39c0057851a0f1b7efecedef8
Parents: d64f5a3
Author: Bill Farner <wf...@apache.org>
Authored: Wed Sep 3 09:14:19 2014 -0700
Committer: Bill Farner <wf...@apache.org>
Committed: Wed Sep 3 09:14:19 2014 -0700
----------------------------------------------------------------------
build.gradle | 6 +
.../aurora/scheduler/app/SchedulerMain.java | 37 ++-
.../aurora/scheduler/app/SchedulerIT.java | 13 +-
.../aurora/scheduler/app/local/FakeMaster.java | 304 +++++++++++++++++++
.../app/local/FakeNonVolatileStorage.java | 69 +++++
.../scheduler/app/local/LocalSchedulerMain.java | 108 +++++++
.../local/simulator/ClusterSimulatorModule.java | 119 ++++++++
.../scheduler/app/local/simulator/Events.java | 38 +++
.../app/local/simulator/FakeSlaves.java | 79 +++++
9 files changed, 753 insertions(+), 20 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-aurora/blob/417cb169/build.gradle
----------------------------------------------------------------------
diff --git a/build.gradle b/build.gradle
index 66c8205..9871e8d 100644
--- a/build.gradle
+++ b/build.gradle
@@ -493,3 +493,9 @@ task FlagSchemaChanges(type: Test) << {
javadoc {
classpath += sourceSets.generated.output
}
+
+run {
+ classpath += sourceSets.generated.output
+ classpath += sourceSets.test.output
+ main = 'org.apache.aurora.scheduler.app.local.LocalSchedulerMain'
+}
http://git-wip-us.apache.org/repos/asf/incubator-aurora/blob/417cb169/src/main/java/org/apache/aurora/scheduler/app/SchedulerMain.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/aurora/scheduler/app/SchedulerMain.java b/src/main/java/org/apache/aurora/scheduler/app/SchedulerMain.java
index ec31c49..40b652c 100644
--- a/src/main/java/org/apache/aurora/scheduler/app/SchedulerMain.java
+++ b/src/main/java/org/apache/aurora/scheduler/app/SchedulerMain.java
@@ -123,7 +123,7 @@ public class SchedulerMain extends AbstractApplication {
}
@VisibleForTesting
- static Iterable<? extends Module> getModules(
+ Iterable<? extends Module> getModules(
String clusterName,
String serverSetPath,
ClientConfig zkClientConfig,
@@ -134,7 +134,7 @@ public class SchedulerMain extends AbstractApplication {
.add(new StatsModule())
.add(new AppModule(clusterName, serverSetPath, zkClientConfig, statsURLPrefix))
.addAll(getExtraModules())
- .add(new LogStorageModule())
+ .add(getPersistentStorageModule())
.add(new MemStorageModule(Bindings.annotatedKeyFactory(LogStorage.WriteBehind.class)))
.add(new CronModule())
.add(new DbModule(Bindings.annotatedKeyFactory(Delegated.class)))
@@ -147,26 +147,30 @@ public class SchedulerMain extends AbstractApplication {
.build();
}
- @Override
- public Iterable<? extends Module> getModules() {
- final ClientConfig zkClientConfig = FlaggedClientConfig.create();
- // TODO(Kevin Sweeney): Push these bindings down into a "production" module.
- Module additional = new AbstractModule() {
+ protected Module getPersistentStorageModule() {
+ return new AbstractModule() {
@Override
protected void configure() {
- bind(DriverFactory.class).to(DriverFactoryImpl.class);
- bind(DriverFactoryImpl.class).in(Singleton.class);
- install(new MesosLogStreamModule(zkClientConfig));
+ install(new LogStorageModule());
}
};
+ }
- Module configModule = new AbstractModule() {
+ protected Module getMesosModules() {
+ final ClientConfig zkClientConfig = FlaggedClientConfig.create();
+ return new AbstractModule() {
@Override
protected void configure() {
- bind(ExecutorConfig.class).toInstance(new ExecutorConfig(THERMOS_EXECUTOR_PATH.get()));
+ bind(DriverFactory.class).to(DriverFactoryImpl.class);
+ bind(DriverFactoryImpl.class).in(Singleton.class);
+ install(new MesosLogStreamModule(zkClientConfig));
}
};
+ }
+ @Override
+ public Iterable<? extends Module> getModules() {
+ ClientConfig zkClientConfig = FlaggedClientConfig.create();
return ImmutableList.<Module>builder()
.add(new BackupModule(SnapshotStoreImpl.class))
.addAll(
@@ -176,8 +180,13 @@ public class SchedulerMain extends AbstractApplication {
zkClientConfig,
STATS_URL_PREFIX.get()))
.add(new ZooKeeperClientModule(zkClientConfig))
- .add(configModule)
- .add(additional)
+ .add(new AbstractModule() {
+ @Override
+ protected void configure() {
+ bind(ExecutorConfig.class).toInstance(new ExecutorConfig(THERMOS_EXECUTOR_PATH.get()));
+ }
+ })
+ .add(getMesosModules())
.build();
}
http://git-wip-us.apache.org/repos/asf/incubator-aurora/blob/417cb169/src/test/java/org/apache/aurora/scheduler/app/SchedulerIT.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/aurora/scheduler/app/SchedulerIT.java b/src/test/java/org/apache/aurora/scheduler/app/SchedulerIT.java
index 7178a92..899416f 100644
--- a/src/test/java/org/apache/aurora/scheduler/app/SchedulerIT.java
+++ b/src/test/java/org/apache/aurora/scheduler/app/SchedulerIT.java
@@ -163,7 +163,11 @@ public class SchedulerIT extends BaseZooKeeperTest {
entrySerializer = new EntrySerializer(LogStorageModule.MAX_LOG_ENTRY_SIZE.get());
zkClient = createZkClient();
+ }
+ private void startScheduler() throws Exception {
+ // TODO(wfarner): Try to accomplish all this by subclassing SchedulerMain and actually using
+ // AppLauncher.
Module testModule = new AbstractModule() {
@Override
protected void configure() {
@@ -173,28 +177,25 @@ public class SchedulerIT extends BaseZooKeeperTest {
install(new BackupModule(backupDir, SnapshotStoreImpl.class));
}
};
-
ClientConfig zkClientConfig = ClientConfig
.create(ImmutableList.of(InetSocketAddress.createUnresolved("localhost", getPort())))
.withCredentials(ZooKeeperClient.digestCredentials("mesos", "mesos"));
+ final SchedulerMain main = SchedulerMain.class.newInstance();
injector = Guice.createInjector(
ImmutableList.<Module>builder()
- .addAll(SchedulerMain.getModules(
- CLUSTER_NAME, SERVERSET_PATH, zkClientConfig, STATS_URL_PREFIX))
+ .addAll(main.getModules(CLUSTER_NAME, SERVERSET_PATH, zkClientConfig, STATS_URL_PREFIX))
.add(new LifecycleModule())
.add(new AppLauncherModule())
.add(new ZooKeeperClientModule(zkClientConfig))
.add(testModule)
.build()
);
+ injector.injectMembers(main);
lifecycle = injector.getInstance(Lifecycle.class);
- }
- private void startScheduler() throws Exception {
injector.getInstance(Key.get(ExceptionalCommand.class, StartupStage.class)).execute();
// Mimic AppLauncher running main.
- final SchedulerMain main = injector.getInstance(SchedulerMain.class);
executor.submit(new Runnable() {
@Override
public void run() {
http://git-wip-us.apache.org/repos/asf/incubator-aurora/blob/417cb169/src/test/java/org/apache/aurora/scheduler/app/local/FakeMaster.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/aurora/scheduler/app/local/FakeMaster.java b/src/test/java/org/apache/aurora/scheduler/app/local/FakeMaster.java
new file mode 100644
index 0000000..98718c4
--- /dev/null
+++ b/src/test/java/org/apache/aurora/scheduler/app/local/FakeMaster.java
@@ -0,0 +1,304 @@
+/**
+ * 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 at
+ *
+ * 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 org.apache.aurora.scheduler.app.local;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.logging.Logger;
+
+import javax.inject.Inject;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Maps;
+import com.google.common.eventbus.EventBus;
+import com.twitter.common.quantity.Amount;
+import com.twitter.common.quantity.Time;
+import com.twitter.common.util.concurrent.ExecutorServiceShutdown;
+
+import org.apache.aurora.scheduler.app.local.simulator.Events.Started;
+import org.apache.mesos.Protos.ExecutorID;
+import org.apache.mesos.Protos.Filters;
+import org.apache.mesos.Protos.FrameworkID;
+import org.apache.mesos.Protos.MasterInfo;
+import org.apache.mesos.Protos.Offer;
+import org.apache.mesos.Protos.OfferID;
+import org.apache.mesos.Protos.Request;
+import org.apache.mesos.Protos.SlaveID;
+import org.apache.mesos.Protos.Status;
+import org.apache.mesos.Protos.TaskID;
+import org.apache.mesos.Protos.TaskInfo;
+import org.apache.mesos.Protos.TaskState;
+import org.apache.mesos.Protos.TaskStatus;
+import org.apache.mesos.Scheduler;
+import org.apache.mesos.SchedulerDriver;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * A simulated master for use in scheduler testing.
+ */
+@SuppressWarnings("deprecation")
+public class FakeMaster implements SchedulerDriver {
+
+ private static final Logger LOG = Logger.getLogger(FakeMaster.class.getName());
+
+ private final Map<TaskID, Task> activeTasks =
+ Collections.synchronizedMap(Maps.<TaskID, Task>newHashMap());
+ private final Map<OfferID, Offer> idleOffers =
+ Collections.synchronizedMap(Maps.<OfferID, Offer>newHashMap());
+ private final Map<OfferID, Offer> sentOffers =
+ Collections.synchronizedMap(Maps.<OfferID, Offer>newHashMap());
+
+ private final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
+ private final CountDownLatch stopped = new CountDownLatch(1);
+
+ private final Scheduler scheduler;
+ private final EventBus eventBus;
+
+ @Inject
+ FakeMaster(Scheduler scheduler, EventBus eventBus) {
+ this.scheduler = requireNonNull(scheduler);
+ this.eventBus = requireNonNull(eventBus);
+ }
+
+ public void addResources(Iterable<Offer> offers) {
+ assertNotStopped();
+
+ synchronized (idleOffers) {
+ for (Offer offer : offers) {
+ checkState(!idleOffers.containsKey(offer.getId()), "Duplicate offer id " + offer.getId());
+ idleOffers.put(offer.getId(), offer);
+ }
+ }
+ }
+
+ public void changeState(TaskID task, TaskState state) {
+ assertNotStopped();
+
+ checkState(activeTasks.containsKey(task), "Task " + task + " does not exist.");
+ scheduler.statusUpdate(this, TaskStatus.newBuilder()
+ .setTaskId(task)
+ .setState(state)
+ .build());
+ }
+
+ @Override
+ public Status start() {
+ assertNotStopped();
+
+ scheduler.registered(this,
+ FrameworkID.newBuilder().setValue("local").build(),
+ MasterInfo.getDefaultInstance());
+
+ eventBus.post(new Started());
+
+ executor.scheduleAtFixedRate(
+ new Runnable() {
+ @Override
+ public void run() {
+ List<Offer> allOffers;
+ synchronized (sentOffers) {
+ synchronized (idleOffers) {
+ sentOffers.putAll(idleOffers);
+ allOffers = ImmutableList.copyOf(idleOffers.values());
+ idleOffers.clear();
+ }
+ }
+
+ if (allOffers.isEmpty()) {
+ LOG.info("All offers consumed, suppressing offer cycle.");
+ } else {
+ scheduler.resourceOffers(FakeMaster.this, allOffers);
+ }
+ }
+ },
+ 1,
+ 5,
+ TimeUnit.SECONDS);
+
+ return Status.DRIVER_RUNNING;
+ }
+
+ @Override
+ public Status stop(boolean failover) {
+ return stop();
+ }
+
+ @Override
+ public Status stop() {
+ stopped.countDown();
+ new ExecutorServiceShutdown(executor, Amount.of(1L, Time.SECONDS)).execute();
+ return Status.DRIVER_RUNNING;
+ }
+
+ @Override
+ public Status abort() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Status join() {
+ try {
+ stopped.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ return Status.DRIVER_RUNNING;
+ }
+
+ @Override
+ public Status run() {
+ assertNotStopped();
+ return Status.DRIVER_RUNNING;
+ }
+
+ @Override
+ public Status requestResources(Collection<Request> requests) {
+ throw new UnsupportedOperationException();
+ }
+
+ private void assertNotStopped() {
+ Preconditions.checkState(stopped.getCount() == 1, "Driver is stopped.");
+ }
+
+ private void checkState(boolean assertion, String failureMessage) {
+ if (!assertion) {
+ scheduler.error(this, failureMessage);
+ stop();
+ throw new IllegalStateException(failureMessage);
+ }
+ }
+
+ @Override
+ public Status launchTasks(
+ Collection<OfferID> offerIds,
+ Collection<TaskInfo> tasks,
+ Filters filters) {
+
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Status launchTasks(Collection<OfferID> offerIds, Collection<TaskInfo> tasks) {
+ assertNotStopped();
+
+ OfferID id = Iterables.getOnlyElement(offerIds);
+ Offer offer = sentOffers.remove(id);
+ checkState(offer != null, "Offer " + id + " is invalid.");
+
+ final TaskInfo task = Iterables.getOnlyElement(tasks);
+ synchronized (activeTasks) {
+ checkState(
+ !activeTasks.containsKey(task.getTaskId()),
+ "Task " + task.getTaskId() + " already exists.");
+ activeTasks.put(task.getTaskId(), new Task(offer, task));
+ }
+
+ executor.schedule(
+ new Runnable() {
+ @Override
+ public void run() {
+ scheduler.statusUpdate(FakeMaster.this, TaskStatus.newBuilder()
+ .setTaskId(task.getTaskId())
+ .setState(TaskState.TASK_RUNNING)
+ .build());
+ }
+ },
+ 1,
+ TimeUnit.SECONDS);
+
+ return Status.DRIVER_RUNNING;
+ }
+
+ @Override
+ public Status launchTasks(OfferID offerId, Collection<TaskInfo> tasks, Filters filters) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Status launchTasks(OfferID offerId, Collection<TaskInfo> tasks) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Status killTask(TaskID taskId) {
+ assertNotStopped();
+
+ Task task = activeTasks.remove(taskId);
+ checkState(task != null, "Task " + taskId + " not found.");
+ idleOffers.put(task.getOffer().getId(), task.getOffer());
+
+ scheduler.statusUpdate(this, TaskStatus.newBuilder()
+ .setTaskId(taskId)
+ .setState(TaskState.TASK_FINISHED)
+ .build());
+
+ return Status.DRIVER_RUNNING;
+ }
+
+ @Override
+ public Status declineOffer(OfferID offerId, Filters filters) {
+ assertNotStopped();
+
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Status declineOffer(OfferID offerId) {
+ assertNotStopped();
+ return null;
+ }
+
+ @Override
+ public Status reviveOffers() {
+ assertNotStopped();
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Status sendFrameworkMessage(ExecutorID executorId, SlaveID slaveId, byte[] data) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Status reconcileTasks(Collection<TaskStatus> statuses) {
+ throw new UnsupportedOperationException();
+ }
+
+ private static final class Task {
+ private final Offer offer;
+ private final TaskInfo task;
+
+ private Task(Offer offer, TaskInfo task) {
+ this.offer = offer;
+ this.task = task;
+ }
+
+ Offer getOffer() {
+ return offer;
+ }
+
+ TaskInfo getTask() {
+ return task;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-aurora/blob/417cb169/src/test/java/org/apache/aurora/scheduler/app/local/FakeNonVolatileStorage.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/aurora/scheduler/app/local/FakeNonVolatileStorage.java b/src/test/java/org/apache/aurora/scheduler/app/local/FakeNonVolatileStorage.java
new file mode 100644
index 0000000..5cc3b41
--- /dev/null
+++ b/src/test/java/org/apache/aurora/scheduler/app/local/FakeNonVolatileStorage.java
@@ -0,0 +1,69 @@
+/**
+ * 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 at
+ *
+ * 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 org.apache.aurora.scheduler.app.local;
+
+import javax.inject.Inject;
+
+import org.apache.aurora.scheduler.storage.Storage;
+import org.apache.aurora.scheduler.storage.Storage.MutateWork.NoResult.Quiet;
+import org.apache.aurora.scheduler.storage.Storage.NonVolatileStorage;
+
+/**
+ * A storage system that implements non-volatile storage operations, but is actually volatile.
+ */
+class FakeNonVolatileStorage implements NonVolatileStorage {
+ private final Storage delegate;
+
+ @Inject
+ FakeNonVolatileStorage(Storage delegate) {
+ this.delegate = delegate;
+ }
+
+ @Override
+ public void stop() {
+ // No-op.
+ }
+
+ @Override
+ public void snapshot() throws StorageException {
+ // No-op.
+ }
+
+ @Override
+ public void start(Quiet initializationLogic) throws StorageException {
+ // No-op.
+ }
+
+ @Override
+ public void prepare() throws StorageException {
+ delegate.prepare();
+ }
+
+ @Override
+ public <T, E extends Exception> T write(MutateWork<T, E> work) throws StorageException, E {
+ return delegate.write(work);
+ }
+
+ @Override
+ public <T, E extends Exception> T weaklyConsistentRead(Work<T, E> work)
+ throws StorageException, E {
+
+ return delegate.weaklyConsistentRead(work);
+ }
+
+ @Override
+ public <T, E extends Exception> T consistentRead(Work<T, E> work) throws StorageException, E {
+ return delegate.consistentRead(work);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-aurora/blob/417cb169/src/test/java/org/apache/aurora/scheduler/app/local/LocalSchedulerMain.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/aurora/scheduler/app/local/LocalSchedulerMain.java b/src/test/java/org/apache/aurora/scheduler/app/local/LocalSchedulerMain.java
new file mode 100644
index 0000000..43d73b1
--- /dev/null
+++ b/src/test/java/org/apache/aurora/scheduler/app/local/LocalSchedulerMain.java
@@ -0,0 +1,108 @@
+/**
+ * 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 at
+ *
+ * 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 org.apache.aurora.scheduler.app.local;
+
+import java.io.File;
+import java.util.List;
+
+import javax.annotation.Nullable;
+import javax.inject.Inject;
+import javax.inject.Singleton;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.io.Files;
+import com.google.inject.AbstractModule;
+import com.google.inject.Key;
+import com.google.inject.Module;
+import com.twitter.common.application.AppLauncher;
+
+import org.apache.aurora.codec.ThriftBinaryCodec.CodingException;
+import org.apache.aurora.gen.storage.Snapshot;
+import org.apache.aurora.scheduler.DriverFactory;
+import org.apache.aurora.scheduler.app.SchedulerMain;
+import org.apache.aurora.scheduler.app.local.simulator.ClusterSimulatorModule;
+import org.apache.aurora.scheduler.storage.DistributedSnapshotStore;
+import org.apache.aurora.scheduler.storage.Storage;
+import org.apache.aurora.scheduler.storage.Storage.NonVolatileStorage;
+import org.apache.aurora.scheduler.storage.log.LogStorage;
+import org.apache.mesos.SchedulerDriver;
+
+/**
+ * A main class that runs the scheduler in local mode, using fakes for external components.
+ */
+public class LocalSchedulerMain extends SchedulerMain {
+
+ @Override
+ protected Module getPersistentStorageModule() {
+ return new AbstractModule() {
+ @Override
+ protected void configure() {
+ bind(Storage.class).to(Key.get(Storage.class, LogStorage.WriteBehind.class));
+ bind(NonVolatileStorage.class).to(FakeNonVolatileStorage.class);
+ bind(DistributedSnapshotStore.class).toInstance(new DistributedSnapshotStore() {
+ @Override
+ public void persist(Snapshot snapshot) throws CodingException {
+ // No-op.
+ }
+ });
+ }
+ };
+ }
+
+ @Override
+ protected Module getMesosModules() {
+ return new AbstractModule() {
+ @Override
+ protected void configure() {
+ bind(DriverFactory.class).to(FakeDriverFactory.class);
+ bind(SchedulerDriver.class).to(FakeMaster.class);
+ bind(FakeMaster.class).in(Singleton.class);
+ install(new ClusterSimulatorModule());
+ }
+ };
+ }
+
+ static class FakeDriverFactory implements DriverFactory {
+ private final SchedulerDriver driver;
+
+ @Inject
+ FakeDriverFactory(SchedulerDriver driver) {
+ this.driver = driver;
+ }
+
+ @Override
+ public SchedulerDriver apply(@Nullable String input) {
+ return driver;
+ }
+ }
+
+ public static void main(String[] args) {
+ File backupDir = Files.createTempDir();
+ backupDir.deleteOnExit();
+
+ List<String> arguments = ImmutableList.<String>builder()
+ .add(args)
+ .add("-cluster_name=local")
+ .add("-serverset_path=/aurora/local/scheduler")
+ .add("-zk_endpoints=localhost:2181")
+ .add("-zk_in_proc=true")
+ .add("-backup_dir=" + backupDir.getAbsolutePath())
+ .add("-mesos_master_address=fake")
+ .add("-thermos_executor_path=fake")
+ .add("-http_port=8081")
+ .build();
+
+ AppLauncher.launch(LocalSchedulerMain.class, arguments.toArray(new String[0]));
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-aurora/blob/417cb169/src/test/java/org/apache/aurora/scheduler/app/local/simulator/ClusterSimulatorModule.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/aurora/scheduler/app/local/simulator/ClusterSimulatorModule.java b/src/test/java/org/apache/aurora/scheduler/app/local/simulator/ClusterSimulatorModule.java
new file mode 100644
index 0000000..9ee4fe2
--- /dev/null
+++ b/src/test/java/org/apache/aurora/scheduler/app/local/simulator/ClusterSimulatorModule.java
@@ -0,0 +1,119 @@
+/**
+ * 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 at
+ *
+ * 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 org.apache.aurora.scheduler.app.local.simulator;
+
+import java.util.UUID;
+
+import javax.inject.Inject;
+import javax.inject.Singleton;
+
+import com.google.common.eventbus.EventBus;
+import com.google.inject.AbstractModule;
+import com.google.inject.multibindings.Multibinder;
+import com.twitter.common.application.modules.LifecycleModule;
+import com.twitter.common.base.Command;
+
+import org.apache.aurora.scheduler.configuration.Resources;
+import org.apache.mesos.Protos;
+import org.apache.mesos.Protos.Offer;
+
+import static java.util.Objects.requireNonNull;
+
+import static org.apache.aurora.scheduler.configuration.ConfigurationManager.DEDICATED_ATTRIBUTE;
+import static org.apache.aurora.scheduler.configuration.ConfigurationManager.HOST_CONSTRAINT;
+import static org.apache.aurora.scheduler.configuration.ConfigurationManager.RACK_CONSTRAINT;
+import static org.apache.mesos.Protos.Value.Type.RANGES;
+import static org.apache.mesos.Protos.Value.Type.SCALAR;
+
+/**
+ * Module that sets up bindings to simulate fake cluster resources.
+ */
+public class ClusterSimulatorModule extends AbstractModule {
+
+ @Override
+ protected void configure() {
+ bind(FakeSlaves.class).in(Singleton.class);
+ Multibinder<Offer> offers = Multibinder.newSetBinder(binder(), Offer.class);
+ offers.addBinding()
+ .toInstance(baseOffer("slave-1", "a", 16, 16 * 1024, 100 * 1024));
+ offers.addBinding()
+ .toInstance(baseOffer("slave-2", "a", 16, 16 * 1024, 100 * 1024));
+ offers.addBinding()
+ .toInstance(baseOffer("slave-3", "b", 16, 16 * 1024, 100 * 1024));
+ offers.addBinding()
+ .toInstance(baseOffer("slave-4", "b", 16, 16 * 1024, 100 * 1024));
+ offers.addBinding()
+ .toInstance(dedicated(baseOffer("slave-5", "c", 24, 128 * 1024, 1824 * 1024), "database"));
+ offers.addBinding()
+ .toInstance(dedicated(baseOffer("slave-6", "c", 24, 128 * 1024, 1824 * 1024), "database"));
+ LifecycleModule.bindStartupAction(binder(), Register.class);
+ }
+
+ static class Register implements Command {
+ private final EventBus eventBus;
+ private final FakeSlaves slaves;
+
+ @Inject
+ Register(EventBus eventBus, FakeSlaves slaves) {
+ this.eventBus = requireNonNull(eventBus);
+ this.slaves = requireNonNull(slaves);
+ }
+
+ @Override
+ public void execute() {
+ eventBus .register(slaves);
+ }
+ }
+
+ private static Offer baseOffer(
+ String slaveId,
+ String rack,
+ double cpu,
+ double ramMb,
+ double diskMb) {
+
+ Protos.Value.Ranges portRanges = Protos.Value.Ranges.newBuilder()
+ .addRange(Protos.Value.Range.newBuilder().setBegin(40000).setEnd(41000)).build();
+
+ String host = slaveId + "-hostname";
+ return Offer.newBuilder()
+ .addResources(Protos.Resource.newBuilder().setType(SCALAR).setName(Resources.CPUS)
+ .setScalar(Protos.Value.Scalar.newBuilder().setValue(cpu)))
+ .addResources(Protos.Resource.newBuilder().setType(SCALAR).setName(Resources.RAM_MB)
+ .setScalar(Protos.Value.Scalar.newBuilder().setValue(ramMb)))
+ .addResources(Protos.Resource.newBuilder().setType(SCALAR).setName(Resources.DISK_MB)
+ .setScalar(Protos.Value.Scalar.newBuilder().setValue(diskMb)))
+ .addResources(Protos.Resource.newBuilder().setType(RANGES).setName(Resources.PORTS)
+ .setRanges(portRanges))
+ .addAttributes(Protos.Attribute.newBuilder().setType(Protos.Value.Type.TEXT)
+ .setName(HOST_CONSTRAINT)
+ .setText(Protos.Value.Text.newBuilder().setValue(host)))
+ .addAttributes(Protos.Attribute.newBuilder().setType(Protos.Value.Type.TEXT)
+ .setName(RACK_CONSTRAINT)
+ .setText(Protos.Value.Text.newBuilder().setValue(rack)))
+ .setSlaveId(Protos.SlaveID.newBuilder().setValue(slaveId))
+ .setHostname(host)
+ .setFrameworkId(Protos.FrameworkID.newBuilder().setValue("frameworkId").build())
+ .setId(Protos.OfferID.newBuilder().setValue(UUID.randomUUID().toString()))
+ .build();
+ }
+
+ private static Offer dedicated(Offer base, String dedicatedTo) {
+ return Offer.newBuilder(base)
+ .addAttributes(Protos.Attribute.newBuilder().setType(Protos.Value.Type.TEXT)
+ .setName(DEDICATED_ATTRIBUTE)
+ .setText(Protos.Value.Text.newBuilder().setValue(dedicatedTo)))
+ .build();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-aurora/blob/417cb169/src/test/java/org/apache/aurora/scheduler/app/local/simulator/Events.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/aurora/scheduler/app/local/simulator/Events.java b/src/test/java/org/apache/aurora/scheduler/app/local/simulator/Events.java
new file mode 100644
index 0000000..f36c1ff
--- /dev/null
+++ b/src/test/java/org/apache/aurora/scheduler/app/local/simulator/Events.java
@@ -0,0 +1,38 @@
+/**
+ * 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 at
+ *
+ * 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 org.apache.aurora.scheduler.app.local.simulator;
+
+import org.apache.mesos.Protos.OfferID;
+import org.apache.mesos.Protos.TaskInfo;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * Pubsub events used to signal fake cluster activity.
+ */
+public interface Events {
+
+ class Started {
+ }
+
+ class OfferAccepted {
+ public final OfferID offer;
+ public final TaskInfo task;
+
+ public OfferAccepted(OfferID offer, TaskInfo task) {
+ this.offer = requireNonNull(offer);
+ this.task = requireNonNull(task);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-aurora/blob/417cb169/src/test/java/org/apache/aurora/scheduler/app/local/simulator/FakeSlaves.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/aurora/scheduler/app/local/simulator/FakeSlaves.java b/src/test/java/org/apache/aurora/scheduler/app/local/simulator/FakeSlaves.java
new file mode 100644
index 0000000..46be83c
--- /dev/null
+++ b/src/test/java/org/apache/aurora/scheduler/app/local/simulator/FakeSlaves.java
@@ -0,0 +1,79 @@
+/**
+ * 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 at
+ *
+ * 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 org.apache.aurora.scheduler.app.local.simulator;
+
+import java.util.Set;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import javax.inject.Inject;
+
+import com.google.common.eventbus.Subscribe;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+
+import org.apache.aurora.scheduler.app.local.FakeMaster;
+import org.apache.aurora.scheduler.app.local.simulator.Events.OfferAccepted;
+import org.apache.aurora.scheduler.app.local.simulator.Events.Started;
+import org.apache.mesos.Protos.Offer;
+import org.apache.mesos.Protos.TaskState;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * A collection of fake slaves, which advertise resources to the fake master, and react to tasks
+ * being scheduled.
+ */
+class FakeSlaves {
+ private final Set<Offer> offers;
+ private final FakeMaster master;
+ private final ScheduledExecutorService executor;
+
+ @Inject
+ FakeSlaves(Set<Offer> offers, FakeMaster master) {
+ this.offers = requireNonNull(offers);
+ this.master = requireNonNull(master);
+ this.executor = Executors.newSingleThreadScheduledExecutor(
+ new ThreadFactoryBuilder().setDaemon(true).setNameFormat("FakeSlave=%d").build());
+ }
+
+ @Subscribe
+ public void start(Started event) {
+ master.addResources(offers);
+ }
+
+ @Subscribe
+ public void offerAccepted(final OfferAccepted accepted) {
+ // Move the task to starting after a delay.
+ executor.schedule(
+ new Runnable() {
+ @Override
+ public void run() {
+ master.changeState(accepted.task.getTaskId(), TaskState.TASK_STARTING);
+
+ executor.schedule(
+ new Runnable() {
+ @Override
+ public void run() {
+ master.changeState(accepted.task.getTaskId(), TaskState.TASK_RUNNING);
+ }
+ },
+ 1,
+ TimeUnit.SECONDS);
+ }
+ },
+ 1,
+ TimeUnit.SECONDS);
+ }
+}