You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by rs...@apache.org on 2017/08/25 19:30:12 UTC
[3/3] kafka git commit: KAFKA-5776;
Add the Trogdor fault injection daemon
KAFKA-5776; Add the Trogdor fault injection daemon
Author: Colin P. Mccabe <cm...@confluent.io>
Reviewers: Ismael Juma <is...@juma.me.uk>, Rajini Sivaram <ra...@googlemail.com>
Closes #3699 from cmccabe/trogdor-review
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/0772fde5
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/0772fde5
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/0772fde5
Branch: refs/heads/trunk
Commit: 0772fde562ed361beec6ad745895a0ee0cf229e8
Parents: 607c3c2
Author: Colin P. Mccabe <cm...@confluent.io>
Authored: Fri Aug 25 12:29:40 2017 -0700
Committer: Rajini Sivaram <ra...@googlemail.com>
Committed: Fri Aug 25 12:29:40 2017 -0700
----------------------------------------------------------------------
bin/trogdor.sh | 50 +++
build.gradle | 7 +
checkstyle/import-control.xml | 12 +
checkstyle/suppressions.xml | 14 +-
.../org/apache/kafka/common/utils/Utils.java | 38 +++
.../org/apache/kafka/trogdor/agent/Agent.java | 339 ++++++++++++++++++
.../apache/kafka/trogdor/agent/AgentClient.java | 158 +++++++++
.../kafka/trogdor/agent/AgentRestResource.java | 80 +++++
.../apache/kafka/trogdor/basic/BasicNode.java | 111 ++++++
.../kafka/trogdor/basic/BasicPlatform.java | 95 ++++++
.../kafka/trogdor/basic/BasicTopology.java | 58 ++++
.../apache/kafka/trogdor/common/JsonUtil.java | 52 +++
.../org/apache/kafka/trogdor/common/Node.java | 65 ++++
.../apache/kafka/trogdor/common/Platform.java | 80 +++++
.../apache/kafka/trogdor/common/Topology.java | 35 ++
.../kafka/trogdor/coordinator/Coordinator.java | 341 +++++++++++++++++++
.../trogdor/coordinator/CoordinatorClient.java | 154 +++++++++
.../coordinator/CoordinatorRestResource.java | 80 +++++
.../kafka/trogdor/coordinator/NodeManager.java | 262 ++++++++++++++
.../kafka/trogdor/fault/AbstractFaultSpec.java | 53 +++
.../org/apache/kafka/trogdor/fault/Fault.java | 50 +++
.../apache/kafka/trogdor/fault/FaultSet.java | 146 ++++++++
.../apache/kafka/trogdor/fault/FaultSpec.java | 59 ++++
.../apache/kafka/trogdor/fault/FaultState.java | 27 ++
.../trogdor/fault/NetworkPartitionFault.java | 130 +++++++
.../fault/NetworkPartitionFaultSpec.java | 65 ++++
.../apache/kafka/trogdor/fault/NoOpFault.java | 92 +++++
.../kafka/trogdor/fault/NoOpFaultSpec.java | 50 +++
.../kafka/trogdor/rest/AgentFaultsResponse.java | 52 +++
.../kafka/trogdor/rest/AgentStatusResponse.java | 59 ++++
.../trogdor/rest/CoordinatorFaultsResponse.java | 52 +++
.../trogdor/rest/CoordinatorStatusResponse.java | 59 ++++
.../trogdor/rest/CreateAgentFaultRequest.java | 69 ++++
.../rest/CreateCoordinatorFaultRequest.java | 69 ++++
.../org/apache/kafka/trogdor/rest/Empty.java | 49 +++
.../kafka/trogdor/rest/ErrorResponse.java | 68 ++++
.../apache/kafka/trogdor/rest/FaultDataMap.java | 98 ++++++
.../kafka/trogdor/rest/JsonRestServer.java | 220 ++++++++++++
.../kafka/trogdor/rest/RestExceptionMapper.java | 71 ++++
.../apache/kafka/trogdor/agent/AgentTest.java | 167 +++++++++
.../kafka/trogdor/basic/BasicPlatformTest.java | 68 ++++
.../trogdor/common/CapturingCommandRunner.java | 60 ++++
.../kafka/trogdor/common/ExpectedFaults.java | 193 +++++++++++
.../trogdor/common/MiniTrogdorCluster.java | 238 +++++++++++++
.../trogdor/coordinator/CoordinatorTest.java | 212 ++++++++++++
.../kafka/trogdor/fault/FaultSetTest.java | 126 +++++++
tools/src/test/resources/log4j.properties | 22 ++
47 files changed, 4648 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/0772fde5/bin/trogdor.sh
----------------------------------------------------------------------
diff --git a/bin/trogdor.sh b/bin/trogdor.sh
new file mode 100755
index 0000000..b211209
--- /dev/null
+++ b/bin/trogdor.sh
@@ -0,0 +1,50 @@
+#!/usr/bin/env bash
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+
+usage() {
+ cat <<EOF
+The Trogdor fault injector.
+
+Usage:
+ $0 [action] [options]
+
+Actions:
+ agent: Run the trogdor agent.
+ coordinator: Run the trogdor coordinator.
+ client: Run the client which communicates with the trogdor coordinator.
+ agent-client: Run the client which communicates with the trogdor agent.
+ help: This help message.
+EOF
+}
+
+if [[ $# -lt 1 ]]; then
+ usage
+ exit 0
+fi
+action="${1}"
+shift
+CLASS=""
+case ${action} in
+ agent) CLASS="org.apache.kafka.trogdor.agent.Agent";;
+ coordinator) CLASS="org.apache.kafka.trogdor.coordinator.Coordinator";;
+ client) CLASS="org.apache.kafka.trogdor.coordinator.CoordinatorClient";;
+ agent-client) CLASS="org.apache.kafka.trogdor.agent.AgentClient";;
+ help) usage; exit 0;;
+ *) echo "Unknown action '${action}'. Type '$0 help' for help."; exit 1;;
+esac
+
+export INCLUDE_TEST_JARS=1
+exec $(dirname $0)/kafka-run-class.sh "${CLASS}" "$@"
http://git-wip-us.apache.org/repos/asf/kafka/blob/0772fde5/build.gradle
----------------------------------------------------------------------
diff --git a/build.gradle b/build.gradle
index f9862c6..c56fe82 100644
--- a/build.gradle
+++ b/build.gradle
@@ -837,8 +837,15 @@ project(':tools') {
compile libs.jacksonDatabind
compile libs.slf4jlog4j
+ compile libs.jacksonJaxrsJsonProvider
+ compile libs.jerseyContainerServlet
+ compile libs.jettyServer
+ compile libs.jettyServlet
+ compile libs.jettyServlets
+
testCompile project(':clients')
testCompile libs.junit
+ testCompile project(':clients').sourceSets.test.output
}
javadoc {
http://git-wip-us.apache.org/repos/asf/kafka/blob/0772fde5/checkstyle/import-control.xml
----------------------------------------------------------------------
diff --git a/checkstyle/import-control.xml b/checkstyle/import-control.xml
index 8c3e3ae..489c004 100644
--- a/checkstyle/import-control.xml
+++ b/checkstyle/import-control.xml
@@ -177,6 +177,18 @@
<allow pkg="org.apache.log4j" />
</subpackage>
+ <subpackage name="trogdor">
+ <allow pkg="com.fasterxml.jackson" />
+ <allow pkg="javax.servlet" />
+ <allow pkg="javax.ws.rs" />
+ <allow pkg="net.sourceforge.argparse4j" />
+ <allow pkg="org.apache.kafka.test"/>
+ <allow pkg="org.apache.kafka.trogdor" />
+ <allow pkg="org.apache.log4j" />
+ <allow pkg="org.eclipse.jetty" />
+ <allow pkg="org.glassfish.jersey" />
+ </subpackage>
+
<subpackage name="streams">
<allow pkg="org.apache.kafka.common"/>
<allow pkg="org.apache.kafka.test"/>
http://git-wip-us.apache.org/repos/asf/kafka/blob/0772fde5/checkstyle/suppressions.xml
----------------------------------------------------------------------
diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml
index e4c244f..990e366 100644
--- a/checkstyle/suppressions.xml
+++ b/checkstyle/suppressions.xml
@@ -56,7 +56,7 @@
files="AbstractRequest.java"/>
<suppress checks="NPathComplexity"
- files="(BufferPool|MetricName|Node|ConfigDef|SslTransportLayer|MetadataResponse|KerberosLogin|SslTransportLayer|Sender|Serdes).java"/>
+ files="(BufferPool|MetricName|Node|ConfigDef|SslTransportLayer|MetadataResponse|KerberosLogin|SslTransportLayer|Sender|Serdes|Agent).java"/>
<!-- clients tests -->
<suppress checks="ClassDataAbstractionCoupling"
@@ -186,14 +186,14 @@
files="VerifiableConsumer.java"/>
<suppress checks="CyclomaticComplexity"
- files="StreamsResetter.java"/>
- <suppress checks="CyclomaticComplexity"
- files="ProducerPerformance.java"/>
+ files="(StreamsResetter|ProducerPerformance|Agent).java"/>
<suppress checks="NPathComplexity"
- files="StreamsResetter.java"/>
- <suppress checks="NPathComplexity"
- files="ProducerPerformance.java"/>
+ files="(ProducerPerformance|StreamsResetter|Agent).java"/>
+ <suppress checks="ImportControl"
+ files="SignalLogger.java"/>
+ <suppress checks="IllegalImport"
+ files="SignalLogger.java"/>
<!-- Log4J-Appender -->
<suppress checks="CyclomaticComplexity"
http://git-wip-us.apache.org/repos/asf/kafka/blob/0772fde5/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/utils/Utils.java b/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
index ee82f9a..2025107 100755
--- a/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
+++ b/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
@@ -31,6 +31,8 @@ import java.io.IOException;
import java.io.InputStream;
import java.io.PrintWriter;
import java.io.StringWriter;
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
@@ -307,6 +309,42 @@ public class Utils {
}
/**
+ * Construct a new object using a class name and parameters.
+ *
+ * @param className The full name of the class to construct.
+ * @param params A sequence of (type, object) elements.
+ * @param <T> The type of object to construct.
+ * @return The new object.
+ * @throws ClassNotFoundException If there was a problem constructing the object.
+ */
+ public static <T> T newParameterizedInstance(String className, Object... params)
+ throws ClassNotFoundException {
+ Class<?>[] argTypes = new Class<?>[params.length / 2];
+ Object[] args = new Object[params.length / 2];
+ try {
+ Class c = Class.forName(className, true, Utils.getContextOrKafkaClassLoader());
+ for (int i = 0; i < params.length / 2; i++) {
+ argTypes[i] = (Class<?>) params[2 * i];
+ args[i] = params[(2 * i) + 1];
+ }
+ Constructor<T> constructor = c.getConstructor(argTypes);
+ return constructor.newInstance(args);
+ } catch (NoSuchMethodException e) {
+ throw new ClassNotFoundException(String.format("Failed to find " +
+ "constructor with %s for %s", Utils.join(argTypes, ", "), className), e);
+ } catch (InstantiationException e) {
+ throw new ClassNotFoundException(String.format("Failed to instantiate " +
+ "%s", className), e);
+ } catch (IllegalAccessException e) {
+ throw new ClassNotFoundException(String.format("Unable to access " +
+ "constructor of %s", className), e);
+ } catch (InvocationTargetException e) {
+ throw new ClassNotFoundException(String.format("Unable to invoke " +
+ "constructor of %s", className), e);
+ }
+ }
+
+ /**
* Generates 32 bit murmur2 hash from byte array
* @param data byte array to hash
* @return 32 bit hash of the given array
http://git-wip-us.apache.org/repos/asf/kafka/blob/0772fde5/tools/src/main/java/org/apache/kafka/trogdor/agent/Agent.java
----------------------------------------------------------------------
diff --git a/tools/src/main/java/org/apache/kafka/trogdor/agent/Agent.java b/tools/src/main/java/org/apache/kafka/trogdor/agent/Agent.java
new file mode 100644
index 0000000..0cb01ad
--- /dev/null
+++ b/tools/src/main/java/org/apache/kafka/trogdor/agent/Agent.java
@@ -0,0 +1,339 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.kafka.trogdor.agent;
+
+import net.sourceforge.argparse4j.ArgumentParsers;
+import net.sourceforge.argparse4j.inf.ArgumentParser;
+import net.sourceforge.argparse4j.inf.ArgumentParserException;
+import net.sourceforge.argparse4j.inf.Namespace;
+import org.apache.kafka.common.utils.Exit;
+import org.apache.kafka.common.utils.KafkaThread;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.trogdor.common.Node;
+import org.apache.kafka.trogdor.common.Platform;
+import org.apache.kafka.trogdor.fault.Fault;
+import org.apache.kafka.trogdor.fault.FaultSet;
+import org.apache.kafka.trogdor.fault.FaultSpec;
+import org.apache.kafka.trogdor.fault.FaultState;
+import org.apache.kafka.trogdor.rest.AgentFaultsResponse;
+import org.apache.kafka.trogdor.rest.CreateAgentFaultRequest;
+import org.apache.kafka.trogdor.rest.JsonRestServer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
+
+import static net.sourceforge.argparse4j.impl.Arguments.store;
+
+/**
+ * The Trogdor agent.
+ *
+ * The agent process implements faults directly.
+ */
+public final class Agent {
+ private static final Logger log = LoggerFactory.getLogger(Agent.class);
+
+ /**
+ * The clock to use for this agent.
+ */
+ private final Time time;
+
+ /**
+ * The time at which this server was started.
+ */
+ private final long startTimeMs;
+
+ /**
+ * The platform.
+ */
+ private final Platform platform;
+
+ /**
+ * The lock protecting shutdown and faultSet.
+ */
+ private final ReentrantLock lock = new ReentrantLock();
+
+ /**
+ * The condition variable which the agent thread waits on.
+ */
+ private final Condition cond = lock.newCondition();
+
+ /**
+ * The agent runnable.
+ */
+ private final AgentRunnable runnable;
+
+ /**
+ * The REST server.
+ */
+ private final JsonRestServer restServer;
+
+ /**
+ * The agent thread.
+ */
+ private final KafkaThread thread;
+
+ /**
+ * The set of pending faults.
+ */
+ private final FaultSet pendingFaults = new FaultSet();
+
+ /**
+ * The set of faults which are running.
+ */
+ private final FaultSet runningFaults = new FaultSet();
+
+ /**
+ * The set of faults which are done.
+ */
+ private final FaultSet doneFaults = new FaultSet();
+
+ /**
+ * True if the server is shutting down.
+ */
+ private boolean shutdown = false;
+
+ class AgentRunnable implements Runnable {
+ @Override
+ public void run() {
+ log.info("Starting main service thread.");
+ try {
+ while (true) {
+ List<Fault> toStart = new ArrayList<>();
+ List<Fault> started = new ArrayList<>();
+ List<Fault> toEnd = new ArrayList<>();
+ List<Fault> ended = new ArrayList<>();
+ long now = time.milliseconds();
+ long nextWakeMs = now + (60L * 60L * 1000L);
+ lock.lock();
+ try {
+ Iterator<Fault> pending = pendingFaults.iterateByStart();
+ while (pending.hasNext()) {
+ Fault fault = pending.next();
+ toStart.add(fault);
+ long endMs = fault.spec().startMs() + fault.spec().durationMs();
+ nextWakeMs = Math.min(nextWakeMs, endMs);
+ pending.remove();
+ }
+ Iterator<Fault> running = runningFaults.iterateByEnd();
+ while (running.hasNext()) {
+ Fault fault = running.next();
+ long endMs = fault.spec().startMs() + fault.spec().durationMs();
+ if (now < endMs) {
+ nextWakeMs = Math.min(nextWakeMs, endMs);
+ break;
+ }
+ toEnd.add(fault);
+ running.remove();
+ }
+ } finally {
+ lock.unlock();
+ }
+ for (Fault fault: toStart) {
+ try {
+ log.debug("Activating fault " + fault);
+ fault.activate(platform);
+ started.add(fault);
+ } catch (Throwable e) {
+ log.error("Error activating fault " + fault.id(), e);
+ ended.add(fault);
+ }
+ }
+ for (Fault fault: toEnd) {
+ try {
+ log.debug("Deactivating fault " + fault);
+ fault.deactivate(platform);
+ } catch (Throwable e) {
+ log.error("Error deactivating fault " + fault.id(), e);
+ } finally {
+ ended.add(fault);
+ }
+ }
+ lock.lock();
+ try {
+ for (Fault fault : started) {
+ runningFaults.add(fault);
+ }
+ for (Fault fault : ended) {
+ doneFaults.add(fault);
+ }
+ if (shutdown) {
+ return;
+ }
+ if (nextWakeMs > now) {
+ log.trace("Sleeping for {} ms", nextWakeMs - now);
+ if (cond.await(nextWakeMs - now, TimeUnit.MILLISECONDS)) {
+ log.trace("AgentRunnable woke up early");
+ }
+ }
+ if (shutdown) {
+ return;
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+ } catch (Throwable t) {
+ log.error("Unhandled exception in AgentRunnable", t);
+ } finally {
+ log.info("AgentRunnable shutting down.");
+ restServer.stop();
+ }
+ }
+ }
+
+ /**
+ * Create a new Agent.
+ *
+ * @param platform The platform object to use.
+ * @param time The timekeeper to use for this Agent.
+ * @param restServer The REST server to use.
+ * @param resource The AgentRestResoure to use.
+ */
+ public Agent(Platform platform, Time time, JsonRestServer restServer,
+ AgentRestResource resource) {
+ this.platform = platform;
+ this.time = time;
+ this.restServer = restServer;
+ this.startTimeMs = time.milliseconds();
+ this.runnable = new AgentRunnable();
+ this.thread = new KafkaThread("TrogdorAgentThread", runnable, false);
+ this.thread.start();
+ resource.setAgent(this);
+ }
+
+ public int port() {
+ return this.restServer.port();
+ }
+
+ public void beginShutdown() {
+ lock.lock();
+ try {
+ if (shutdown)
+ return;
+ this.shutdown = true;
+ cond.signalAll();
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ public void waitForShutdown() {
+ try {
+ this.thread.join();
+ } catch (InterruptedException e) {
+ log.error("Interrupted while waiting for thread shutdown", e);
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ public long startTimeMs() {
+ return startTimeMs;
+ }
+
+ public AgentFaultsResponse faults() {
+ Map<String, AgentFaultsResponse.FaultData> faultData = new TreeMap<>();
+ lock.lock();
+ try {
+ updateFaultsResponse(faultData, pendingFaults, FaultState.PENDING);
+ updateFaultsResponse(faultData, runningFaults, FaultState.RUNNING);
+ updateFaultsResponse(faultData, doneFaults, FaultState.DONE);
+ } finally {
+ lock.unlock();
+ }
+ return new AgentFaultsResponse(faultData);
+ }
+
+ private void updateFaultsResponse(Map<String, AgentFaultsResponse.FaultData> faultData,
+ FaultSet faultSet, FaultState state) {
+ for (Iterator<Fault> iter = faultSet.iterateByStart();
+ iter.hasNext(); ) {
+ Fault fault = iter.next();
+ AgentFaultsResponse.FaultData data =
+ new AgentFaultsResponse.FaultData(fault.spec(), state);
+ faultData.put(fault.id(), data);
+ }
+ }
+
+ public void createFault(CreateAgentFaultRequest request) throws ClassNotFoundException {
+ lock.lock();
+ try {
+ Fault fault = FaultSpec.Util.createFault(request.id(), request.spec());
+ pendingFaults.add(fault);
+ cond.signalAll();
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ public static void main(String[] args) throws Exception {
+ ArgumentParser parser = ArgumentParsers
+ .newArgumentParser("trogdor-agent")
+ .defaultHelp(true)
+ .description("The Trogdor fault injection agent");
+ parser.addArgument("--agent.config")
+ .action(store())
+ .required(true)
+ .type(String.class)
+ .dest("config")
+ .metavar("CONFIG")
+ .help("The configuration file to use.");
+ parser.addArgument("--node-name")
+ .action(store())
+ .required(true)
+ .type(String.class)
+ .dest("node_name")
+ .metavar("NODE_NAME")
+ .help("The name of this node.");
+ Namespace res = null;
+ try {
+ res = parser.parseArgs(args);
+ } catch (ArgumentParserException e) {
+ if (args.length == 0) {
+ parser.printHelp();
+ Exit.exit(0);
+ } else {
+ parser.handleError(e);
+ Exit.exit(1);
+ }
+ }
+ String configPath = res.getString("config");
+ String nodeName = res.getString("node_name");
+
+ Platform platform = Platform.Config.parse(nodeName, configPath);
+ JsonRestServer restServer =
+ new JsonRestServer(Node.Util.getTrogdorAgentPort(platform.curNode()));
+ AgentRestResource resource = new AgentRestResource();
+ Agent agent = new Agent(platform, Time.SYSTEM, restServer, resource);
+ restServer.start(resource);
+ Runtime.getRuntime().addShutdownHook(new Thread() {
+ @Override
+ public void run() {
+ log.error("Agent shutting down...");
+ }
+ });
+ agent.waitForShutdown();
+ }
+};
http://git-wip-us.apache.org/repos/asf/kafka/blob/0772fde5/tools/src/main/java/org/apache/kafka/trogdor/agent/AgentClient.java
----------------------------------------------------------------------
diff --git a/tools/src/main/java/org/apache/kafka/trogdor/agent/AgentClient.java b/tools/src/main/java/org/apache/kafka/trogdor/agent/AgentClient.java
new file mode 100644
index 0000000..928820b
--- /dev/null
+++ b/tools/src/main/java/org/apache/kafka/trogdor/agent/AgentClient.java
@@ -0,0 +1,158 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.kafka.trogdor.agent;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import net.sourceforge.argparse4j.ArgumentParsers;
+import net.sourceforge.argparse4j.inf.ArgumentParser;
+import net.sourceforge.argparse4j.inf.ArgumentParserException;
+import net.sourceforge.argparse4j.inf.MutuallyExclusiveGroup;
+import net.sourceforge.argparse4j.inf.Namespace;
+import org.apache.kafka.common.utils.Exit;
+import org.apache.kafka.trogdor.common.JsonUtil;
+import org.apache.kafka.trogdor.rest.AgentFaultsResponse;
+import org.apache.kafka.trogdor.rest.AgentStatusResponse;
+import org.apache.kafka.trogdor.rest.CreateAgentFaultRequest;
+import org.apache.kafka.trogdor.rest.Empty;
+import org.apache.kafka.trogdor.rest.JsonRestServer;
+import org.apache.kafka.trogdor.rest.JsonRestServer.HttpResponse;
+
+import static net.sourceforge.argparse4j.impl.Arguments.store;
+import static net.sourceforge.argparse4j.impl.Arguments.storeTrue;
+
+/**
+ * A client for the Trogdor agent.
+ */
+public class AgentClient {
+ /**
+ * The URL target.
+ */
+ private final String target;
+
+ public AgentClient(String host, int port) {
+ this(String.format("%s:%d", host, port));
+ }
+
+ public AgentClient(String target) {
+ this.target = target;
+ }
+
+ public String target() {
+ return target;
+ }
+
+ private String url(String suffix) {
+ return String.format("http://%s%s", target, suffix);
+ }
+
+ public AgentStatusResponse getStatus() throws Exception {
+ HttpResponse<AgentStatusResponse> resp =
+ JsonRestServer.<AgentStatusResponse>httpRequest(url("/agent/status"), "GET",
+ null, new TypeReference<AgentStatusResponse>() { });
+ return resp.body();
+ }
+
+ public AgentFaultsResponse getFaults() throws Exception {
+ HttpResponse<AgentFaultsResponse> resp =
+ JsonRestServer.<AgentFaultsResponse>httpRequest(url("/agent/faults"), "GET",
+ null, new TypeReference<AgentFaultsResponse>() { });
+ return resp.body();
+ }
+
+ public void putFault(CreateAgentFaultRequest request) throws Exception {
+ HttpResponse<AgentFaultsResponse> resp =
+ JsonRestServer.<AgentFaultsResponse>httpRequest(url("/agent/fault"), "PUT",
+ request, new TypeReference<AgentFaultsResponse>() { });
+ resp.body();
+ }
+
+ public void invokeShutdown() throws Exception {
+ HttpResponse<Empty> resp =
+ JsonRestServer.<Empty>httpRequest(url("/agent/shutdown"), "PUT",
+ null, new TypeReference<Empty>() { });
+ resp.body();
+ }
+
+ public static void main(String[] args) throws Exception {
+ ArgumentParser parser = ArgumentParsers
+ .newArgumentParser("trogdor-agent-client")
+ .defaultHelp(true)
+ .description("The Trogdor fault injection agent client.");
+ parser.addArgument("target")
+ .action(store())
+ .required(true)
+ .type(String.class)
+ .dest("target")
+ .metavar("TARGET")
+ .help("A colon-separated host and port pair. For example, example.com:8888");
+ MutuallyExclusiveGroup actions = parser.addMutuallyExclusiveGroup();
+ actions.addArgument("--status")
+ .action(storeTrue())
+ .type(Boolean.class)
+ .dest("status")
+ .help("Get agent status.");
+ actions.addArgument("--get-faults")
+ .action(storeTrue())
+ .type(Boolean.class)
+ .dest("get_faults")
+ .help("Get agent faults.");
+ actions.addArgument("--create-fault")
+ .action(store())
+ .type(String.class)
+ .dest("create_fault")
+ .metavar("FAULT_JSON")
+ .help("Create a new fault.");
+ actions.addArgument("--shutdown")
+ .action(storeTrue())
+ .type(Boolean.class)
+ .dest("shutdown")
+ .help("Trigger agent shutdown");
+
+ Namespace res = null;
+ try {
+ res = parser.parseArgs(args);
+ } catch (ArgumentParserException e) {
+ if (args.length == 0) {
+ parser.printHelp();
+ Exit.exit(0);
+ } else {
+ parser.handleError(e);
+ Exit.exit(1);
+ }
+ }
+ String target = res.getString("target");
+ AgentClient client = new AgentClient(target);
+ if (res.getBoolean("status")) {
+ System.out.println("Got agent status: " +
+ JsonUtil.toPrettyJsonString(client.getStatus()));
+ } else if (res.getBoolean("get_faults")) {
+ System.out.println("Got agent faults: " +
+ JsonUtil.toPrettyJsonString(client.getFaults()));
+ } else if (res.getString("create_fault") != null) {
+ client.putFault(JsonUtil.JSON_SERDE.readValue(res.getString("create_fault"),
+ CreateAgentFaultRequest.class));
+ System.out.println("Created fault.");
+ } else if (res.getBoolean("shutdown")) {
+ client.invokeShutdown();
+ System.out.println("Sent shutdown request.");
+ } else {
+ System.out.println("You must choose an action. Type --help for help.");
+ Exit.exit(1);
+ }
+ }
+};
http://git-wip-us.apache.org/repos/asf/kafka/blob/0772fde5/tools/src/main/java/org/apache/kafka/trogdor/agent/AgentRestResource.java
----------------------------------------------------------------------
diff --git a/tools/src/main/java/org/apache/kafka/trogdor/agent/AgentRestResource.java b/tools/src/main/java/org/apache/kafka/trogdor/agent/AgentRestResource.java
new file mode 100644
index 0000000..690fa68
--- /dev/null
+++ b/tools/src/main/java/org/apache/kafka/trogdor/agent/AgentRestResource.java
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.kafka.trogdor.agent;
+
+import org.apache.kafka.trogdor.rest.AgentFaultsResponse;
+import org.apache.kafka.trogdor.rest.AgentStatusResponse;
+import org.apache.kafka.trogdor.rest.CreateAgentFaultRequest;
+import org.apache.kafka.trogdor.rest.Empty;
+
+import javax.servlet.ServletContext;
+import javax.ws.rs.Consumes;
+import javax.ws.rs.GET;
+import javax.ws.rs.PUT;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.MediaType;
+import java.util.concurrent.atomic.AtomicReference;
+
+
+@Path("/agent")
+@Produces(MediaType.APPLICATION_JSON)
+@Consumes(MediaType.APPLICATION_JSON)
+public class AgentRestResource {
+ private final AtomicReference<Agent> agent = new AtomicReference<>(null);
+
+ @javax.ws.rs.core.Context
+ private ServletContext context;
+
+ public void setAgent(Agent myAgent) {
+ agent.set(myAgent);
+ }
+
+ @GET
+ @Path("/status")
+ public AgentStatusResponse getStatus() throws Throwable {
+ return new AgentStatusResponse(agent().startTimeMs());
+ }
+
+ @GET
+ @Path("/faults")
+ public AgentFaultsResponse getAgentFaults() throws Throwable {
+ return agent().faults();
+ }
+
+ @PUT
+ @Path("/fault")
+ public Empty putAgentFault(CreateAgentFaultRequest request) throws Throwable {
+ agent().createFault(request);
+ return Empty.INSTANCE;
+ }
+
+ @PUT
+ @Path("/shutdown")
+ public Empty shutdown() throws Throwable {
+ agent().beginShutdown();
+ return Empty.INSTANCE;
+ }
+
+ private Agent agent() {
+ Agent myAgent = agent.get();
+ if (myAgent == null) {
+ throw new RuntimeException("AgentRestResource has not been initialized yet.");
+ }
+ return myAgent;
+ }
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/0772fde5/tools/src/main/java/org/apache/kafka/trogdor/basic/BasicNode.java
----------------------------------------------------------------------
diff --git a/tools/src/main/java/org/apache/kafka/trogdor/basic/BasicNode.java b/tools/src/main/java/org/apache/kafka/trogdor/basic/BasicNode.java
new file mode 100644
index 0000000..232a64d
--- /dev/null
+++ b/tools/src/main/java/org/apache/kafka/trogdor/basic/BasicNode.java
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.kafka.trogdor.basic;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import org.apache.kafka.trogdor.common.Node;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+public class BasicNode implements Node {
+ private final String name;
+ private final String hostname;
+ private final Map<String, String> config;
+ private final Set<String> tags;
+
+ public BasicNode(String name, String hostname, Map<String, String> config,
+ Set<String> tags) {
+ this.name = name;
+ this.hostname = hostname;
+ this.config = config;
+ this.tags = tags;
+ }
+
+ public BasicNode(String name, JsonNode root) {
+ this.name = name;
+ String hostname = "localhost";
+ Set<String> tags = Collections.emptySet();
+ Map<String, String> config = new HashMap<>();
+ for (Iterator<Map.Entry<String, JsonNode>> iter = root.fields();
+ iter.hasNext(); ) {
+ Map.Entry<String, JsonNode> entry = iter.next();
+ String key = entry.getKey();
+ JsonNode node = entry.getValue();
+ if (key.equals("hostname")) {
+ hostname = node.asText();
+ } else if (key.equals("tags")) {
+ if (!node.isArray()) {
+ throw new RuntimeException("Expected the 'tags' field to be an " +
+ "array of strings.");
+ }
+ tags = new HashSet<>();
+ for (Iterator<JsonNode> tagIter = node.elements(); tagIter.hasNext(); ) {
+ JsonNode tag = tagIter.next();
+ tags.add(tag.asText());
+ }
+ } else {
+ config.put(key, node.asText());
+ }
+ }
+ this.hostname = hostname;
+ this.tags = tags;
+ this.config = config;
+ }
+
+ @Override
+ public String name() {
+ return name;
+ }
+
+ @Override
+ public String hostname() {
+ return hostname;
+ }
+
+ @Override
+ public String getConfig(String key) {
+ return config.get(key);
+ }
+
+ @Override
+ public Set<String> tags() {
+ return tags;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(name, hostname, config, tags);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ BasicNode that = (BasicNode) o;
+ return Objects.equals(name, that.name) &&
+ Objects.equals(hostname, that.hostname) &&
+ Objects.equals(config, that.config) &&
+ Objects.equals(tags, that.tags);
+ }
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/0772fde5/tools/src/main/java/org/apache/kafka/trogdor/basic/BasicPlatform.java
----------------------------------------------------------------------
diff --git a/tools/src/main/java/org/apache/kafka/trogdor/basic/BasicPlatform.java b/tools/src/main/java/org/apache/kafka/trogdor/basic/BasicPlatform.java
new file mode 100644
index 0000000..9bd2cb9
--- /dev/null
+++ b/tools/src/main/java/org/apache/kafka/trogdor/basic/BasicPlatform.java
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.kafka.trogdor.basic;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import org.apache.kafka.common.utils.Shell;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.trogdor.common.Node;
+import org.apache.kafka.trogdor.common.Platform;
+import org.apache.kafka.trogdor.common.Topology;
+
+import java.io.IOException;
+
+/**
+ * Defines a cluster topology
+ */
+public class BasicPlatform implements Platform {
+ private final Node curNode;
+ private final BasicTopology topology;
+ private final CommandRunner commandRunner;
+
+ public interface CommandRunner {
+ String run(Node curNode, String[] command) throws IOException;
+ }
+
+ public static class ShellCommandRunner implements CommandRunner {
+ @Override
+ public String run(Node curNode, String[] command) throws IOException {
+ return Shell.execCommand(command);
+ }
+ }
+
+ public BasicPlatform(String curNodeName, BasicTopology topology,
+ CommandRunner commandRunner) {
+ this.curNode = topology.node(curNodeName);
+ if (this.curNode == null) {
+ throw new RuntimeException(String.format("No node named %s found " +
+ "in the cluster! Cluster nodes are: %s", curNodeName,
+ Utils.join(topology.nodes().keySet(), ",")));
+ }
+ this.topology = topology;
+ this.commandRunner = commandRunner;
+ }
+
+ public BasicPlatform(String curNodeName, JsonNode configRoot) {
+ JsonNode nodes = configRoot.get("nodes");
+ if (nodes == null) {
+ throw new RuntimeException("Expected to find a 'nodes' field " +
+ "in the root JSON configuration object");
+ }
+ this.topology = new BasicTopology(nodes);
+ this.curNode = topology.node(curNodeName);
+ if (this.curNode == null) {
+ throw new RuntimeException(String.format("No node named %s found " +
+ "in the cluster! Cluster nodes are: %s", curNodeName,
+ Utils.join(topology.nodes().keySet(), ",")));
+ }
+ this.commandRunner = new ShellCommandRunner();
+ }
+
+ @Override
+ public String name() {
+ return "BasicPlatform";
+ }
+
+ @Override
+ public Node curNode() {
+ return curNode;
+ }
+
+ @Override
+ public Topology topology() {
+ return topology;
+ }
+
+ @Override
+ public String runCommand(String[] command) throws IOException {
+ return commandRunner.run(curNode, command);
+ }
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/0772fde5/tools/src/main/java/org/apache/kafka/trogdor/basic/BasicTopology.java
----------------------------------------------------------------------
diff --git a/tools/src/main/java/org/apache/kafka/trogdor/basic/BasicTopology.java b/tools/src/main/java/org/apache/kafka/trogdor/basic/BasicTopology.java
new file mode 100644
index 0000000..1f29150
--- /dev/null
+++ b/tools/src/main/java/org/apache/kafka/trogdor/basic/BasicTopology.java
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.kafka.trogdor.basic;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import org.apache.kafka.trogdor.common.Node;
+import org.apache.kafka.trogdor.common.Topology;
+
+import java.util.Iterator;
+import java.util.NavigableMap;
+import java.util.TreeMap;
+
+public class BasicTopology implements Topology {
+ private final NavigableMap<String, Node> nodes;
+
+ public BasicTopology(NavigableMap<String, Node> nodes) {
+ this.nodes = nodes;
+ }
+
+ public BasicTopology(JsonNode configRoot) {
+ if (!configRoot.isObject()) {
+ throw new RuntimeException("Expected the 'nodes' element to be " +
+ "a JSON object.");
+ }
+ nodes = new TreeMap<>();
+ for (Iterator<String> iter = configRoot.fieldNames(); iter.hasNext(); ) {
+ String nodeName = iter.next();
+ JsonNode nodeConfig = configRoot.get(nodeName);
+ BasicNode node = new BasicNode(nodeName, nodeConfig);
+ nodes.put(nodeName, node);
+ }
+ }
+
+ @Override
+ public Node node(String id) {
+ return nodes.get(id);
+ }
+
+ @Override
+ public NavigableMap<String, Node> nodes() {
+ return nodes;
+ }
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/0772fde5/tools/src/main/java/org/apache/kafka/trogdor/common/JsonUtil.java
----------------------------------------------------------------------
diff --git a/tools/src/main/java/org/apache/kafka/trogdor/common/JsonUtil.java b/tools/src/main/java/org/apache/kafka/trogdor/common/JsonUtil.java
new file mode 100644
index 0000000..9ec737e
--- /dev/null
+++ b/tools/src/main/java/org/apache/kafka/trogdor/common/JsonUtil.java
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.kafka.trogdor.common;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+
+/**
+ * Utilities for working with JSON.
+ */
+public class JsonUtil {
+ public static final ObjectMapper JSON_SERDE;
+
+ static {
+ JSON_SERDE = new ObjectMapper();
+ JSON_SERDE.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false);
+ JSON_SERDE.configure(DeserializationFeature.ACCEPT_SINGLE_VALUE_AS_ARRAY, true);
+ }
+
+ public static String toJsonString(Object object) {
+ try {
+ return JSON_SERDE.writeValueAsString(object);
+ } catch (JsonProcessingException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public static String toPrettyJsonString(Object object) {
+ try {
+ return JSON_SERDE.writerWithDefaultPrettyPrinter().writeValueAsString(object);
+ } catch (JsonProcessingException e) {
+ throw new RuntimeException(e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/0772fde5/tools/src/main/java/org/apache/kafka/trogdor/common/Node.java
----------------------------------------------------------------------
diff --git a/tools/src/main/java/org/apache/kafka/trogdor/common/Node.java b/tools/src/main/java/org/apache/kafka/trogdor/common/Node.java
new file mode 100644
index 0000000..24c8488
--- /dev/null
+++ b/tools/src/main/java/org/apache/kafka/trogdor/common/Node.java
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.kafka.trogdor.common;
+
+import java.util.Set;
+
+/**
+ * Defines a node in a cluster topology
+ */
+public interface Node {
+ public static class Util {
+ public static int getIntConfig(Node node, String key, int defaultVal) {
+ String val = node.getConfig(key);
+ if (val == null) {
+ return defaultVal;
+ } else {
+ return Integer.parseInt(val);
+ }
+ }
+
+ public static int getTrogdorAgentPort(Node node) {
+ return getIntConfig(node, Platform.Config.TROGDOR_AGENT_PORT, 0);
+ }
+
+ public static int getTrogdorCoordinatorPort(Node node) {
+ return getIntConfig(node, Platform.Config.TROGDOR_COORDINATOR_PORT, 0);
+ }
+ }
+
+ /**
+ * Get name for this node.
+ */
+ String name();
+
+ /**
+ * Get hostname for this node.
+ */
+ String hostname();
+
+ /**
+ * Get the configuration value associated with the key, or null if there
+ * is none.
+ */
+ String getConfig(String key);
+
+ /**
+ * Get the tags for this node.
+ */
+ Set<String> tags();
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/0772fde5/tools/src/main/java/org/apache/kafka/trogdor/common/Platform.java
----------------------------------------------------------------------
diff --git a/tools/src/main/java/org/apache/kafka/trogdor/common/Platform.java b/tools/src/main/java/org/apache/kafka/trogdor/common/Platform.java
new file mode 100644
index 0000000..1177ace
--- /dev/null
+++ b/tools/src/main/java/org/apache/kafka/trogdor/common/Platform.java
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.kafka.trogdor.common;
+
+import com.fasterxml.jackson.databind.JsonNode;
+
+import java.io.File;
+import java.io.IOException;
+import org.apache.kafka.common.utils.Utils;
+
+/**
+ * Defines a cluster topology
+ */
+public interface Platform {
+ class Config {
+ public static final String TROGDOR_AGENT_PORT = "trogdor.agent.port";
+
+ public static final String TROGDOR_COORDINATOR_PORT = "trogdor.coordinator.port";
+
+ public static final String TROGDOR_COORDINATOR_HEARTBEAT_MS =
+ "trogdor.coordinator.heartbeat.ms";
+
+ public static final int TROGDOR_COORDINATOR_HEARTBEAT_MS_DEFAULT = 60000;
+
+ public static Platform parse(String curNodeName, String path) throws Exception {
+ JsonNode root = JsonUtil.JSON_SERDE.readTree(new File(path));
+ JsonNode platformNode = root.get("platform");
+ if (platformNode == null) {
+ throw new RuntimeException("Expected to find a 'platform' field " +
+ "in the root JSON configuration object");
+ }
+ String platformName = platformNode.textValue();
+ return Utils.newParameterizedInstance(platformName,
+ String.class, curNodeName,
+ JsonNode.class, root);
+ }
+ }
+
+ /**
+ * Get name for this platform.
+ */
+ String name();
+
+ /**
+ * Get the current node.
+ */
+ Node curNode();
+
+ /**
+ * Get the cluster topology.
+ */
+ Topology topology();
+
+ /**
+ * Run a command on this local node.
+ *
+ * Throws an exception if the command could not be run, or if the
+ * command returned a non-zero error status.
+ *
+ * @param command The command
+ *
+ * @return The command output.
+ */
+ String runCommand(String[] command) throws IOException;
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/0772fde5/tools/src/main/java/org/apache/kafka/trogdor/common/Topology.java
----------------------------------------------------------------------
diff --git a/tools/src/main/java/org/apache/kafka/trogdor/common/Topology.java b/tools/src/main/java/org/apache/kafka/trogdor/common/Topology.java
new file mode 100644
index 0000000..94a5474
--- /dev/null
+++ b/tools/src/main/java/org/apache/kafka/trogdor/common/Topology.java
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.kafka.trogdor.common;
+
+import java.util.NavigableMap;
+
+/**
+ * Defines a cluster topology
+ */
+public interface Topology {
+ /**
+ * Get the node with the given name.
+ */
+ Node node(String id);
+
+ /**
+ * Get a sorted map of node names to nodes.
+ */
+ NavigableMap<String, Node> nodes();
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/0772fde5/tools/src/main/java/org/apache/kafka/trogdor/coordinator/Coordinator.java
----------------------------------------------------------------------
diff --git a/tools/src/main/java/org/apache/kafka/trogdor/coordinator/Coordinator.java b/tools/src/main/java/org/apache/kafka/trogdor/coordinator/Coordinator.java
new file mode 100644
index 0000000..1f71fc5
--- /dev/null
+++ b/tools/src/main/java/org/apache/kafka/trogdor/coordinator/Coordinator.java
@@ -0,0 +1,341 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.kafka.trogdor.coordinator;
+
+import net.sourceforge.argparse4j.ArgumentParsers;
+import net.sourceforge.argparse4j.inf.ArgumentParser;
+import net.sourceforge.argparse4j.inf.ArgumentParserException;
+import net.sourceforge.argparse4j.inf.Namespace;
+import org.apache.kafka.common.utils.Exit;
+import org.apache.kafka.common.utils.KafkaThread;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.trogdor.common.Node;
+import org.apache.kafka.trogdor.common.Platform;
+import org.apache.kafka.trogdor.fault.Fault;
+import org.apache.kafka.trogdor.fault.FaultSet;
+import org.apache.kafka.trogdor.fault.FaultSpec;
+import org.apache.kafka.trogdor.fault.FaultState;
+import org.apache.kafka.trogdor.rest.CoordinatorFaultsResponse;
+import org.apache.kafka.trogdor.rest.CreateCoordinatorFaultRequest;
+import org.apache.kafka.trogdor.rest.JsonRestServer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
+
+import static net.sourceforge.argparse4j.impl.Arguments.store;
+
+/**
+ * The Trogdor coordinator.
+ *
+ * The coordinator manages the agent processes in the cluster.
+ */
+public final class Coordinator {
+ private static final Logger log = LoggerFactory.getLogger(Coordinator.class);
+
+ /**
+ * The clock to use for this coordinator.
+ */
+ private final Time time;
+
+ /**
+ * The start time in milliseconds.
+ */
+ private final long startTimeMs;
+
+ /**
+ * The platform.
+ */
+ private final Platform platform;
+
+ /**
+ * NodeManager objects for each node in the cluster.
+ */
+ private final Map<String, NodeManager> nodeManagers;
+
+ /**
+ * The lock protecting shutdown and faultQueue.
+ */
+ private final ReentrantLock lock = new ReentrantLock();
+
+ /**
+ * The condition variable which the coordinator thread waits on.
+ */
+ private final Condition cond = lock.newCondition();
+
+ /**
+ * The coordinator runnable.
+ */
+ private final CoordinatorRunnable runnable;
+
+ /**
+ * The REST server.
+ */
+ private final JsonRestServer restServer;
+
+ /**
+ * The coordinator thread.
+ */
+ private final KafkaThread thread;
+
+ /**
+ * True if the server is shutting down.
+ */
+ private boolean shutdown = false;
+
+ /**
+ * The set of faults which have been scheduled.
+ */
+ private final FaultSet pendingFaults = new FaultSet();
+
+ /**
+ * The set of faults which have been sent to the agents.
+ */
+ private final FaultSet doneFaults = new FaultSet();
+
+ class CoordinatorRunnable implements Runnable {
+ @Override
+ public void run() {
+ log.info("Starting main service thread.");
+ try {
+ long nextWakeMs = 0;
+ while (true) {
+ long now = time.milliseconds();
+ List<Fault> toStart = new ArrayList<>();
+ lock.lock();
+ try {
+ if (nextWakeMs > now) {
+ if (cond.await(nextWakeMs - now, TimeUnit.MILLISECONDS)) {
+ log.trace("CoordinatorRunnable woke up early.");
+ }
+ }
+ nextWakeMs = now + (60L * 60L * 1000L);
+ if (shutdown) {
+ log.info("CoordinatorRunnable shutting down.");
+ return;
+ }
+ Iterator<Fault> iter = pendingFaults.iterateByStart();
+ while (iter.hasNext()) {
+ Fault fault = iter.next();
+ if (now < fault.spec().startMs()) {
+ nextWakeMs = Math.min(nextWakeMs, fault.spec().startMs());
+ break;
+ }
+ toStart.add(fault);
+ iter.remove();
+ doneFaults.add(fault);
+ }
+ } finally {
+ lock.unlock();
+ }
+ for (Fault fault: toStart) {
+ startFault(fault);
+ }
+ }
+ } catch (Throwable t) {
+ log.error("CoordinatorRunnable shutting down with exception", t);
+ } finally {
+ log.info("CoordinatorRunnable shutting down.");
+ restServer.stop();
+ for (NodeManager nodeManager : nodeManagers.values()) {
+ nodeManager.beginShutdown();
+ }
+ for (NodeManager nodeManager : nodeManagers.values()) {
+ nodeManager.waitForShutdown();
+ }
+ }
+ }
+ }
+
+ /**
+ * Create a new Coordinator.
+ *
+ * @param platform The platform object to use.
+ * @param time The timekeeper to use for this Coordinator.
+ * @param restServer The REST server to use.
+ * @param resource The AgentRestResoure to use.
+ */
+ public Coordinator(Platform platform, Time time, JsonRestServer restServer,
+ CoordinatorRestResource resource) {
+ this.platform = platform;
+ this.time = time;
+ this.startTimeMs = time.milliseconds();
+ this.runnable = new CoordinatorRunnable();
+ this.restServer = restServer;
+ this.nodeManagers = new HashMap<>();
+ for (Node node : platform.topology().nodes().values()) {
+ if (Node.Util.getTrogdorAgentPort(node) > 0) {
+ this.nodeManagers.put(node.name(), new NodeManager(time, node));
+ }
+ }
+ if (this.nodeManagers.isEmpty()) {
+ log.warn("No agent nodes configured.");
+ }
+ this.thread = new KafkaThread("TrogdorCoordinatorThread", runnable, false);
+ this.thread.start();
+ resource.setCoordinator(this);
+ }
+
+ public int port() {
+ return this.restServer.port();
+ }
+
+ private void startFault(Fault fault) {
+ Set<String> affectedNodes = fault.targetNodes(platform.topology());
+ Set<NodeManager> affectedManagers = new HashSet<>();
+ Set<String> nonexistentNodes = new HashSet<>();
+ Set<String> nodeNames = new HashSet<>();
+ for (String affectedNode : affectedNodes) {
+ NodeManager nodeManager = nodeManagers.get(affectedNode);
+ if (nodeManager == null) {
+ nonexistentNodes.add(affectedNode);
+ } else {
+ affectedManagers.add(nodeManager);
+ nodeNames.add(affectedNode);
+ }
+ }
+ if (!nonexistentNodes.isEmpty()) {
+ log.warn("Fault {} refers to {} non-existent node(s): {}", fault.id(),
+ nonexistentNodes.size(), Utils.join(nonexistentNodes, ", "));
+ }
+ log.info("Applying fault {} on {} node(s): {}", fault.id(),
+ nodeNames.size(), Utils.join(nodeNames, ", "));
+ for (NodeManager nodeManager : affectedManagers) {
+ nodeManager.enqueueFault(fault);
+ }
+ }
+
+ public void beginShutdown() {
+ lock.lock();
+ try {
+ this.shutdown = true;
+ cond.signalAll();
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ public void waitForShutdown() {
+ try {
+ this.thread.join();
+ } catch (InterruptedException e) {
+ log.error("Interrupted while waiting for thread shutdown", e);
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ public long startTimeMs() {
+ return startTimeMs;
+ }
+
+ public CoordinatorFaultsResponse getFaults() {
+ Map<String, CoordinatorFaultsResponse.FaultData> faultData = new TreeMap<>();
+ lock.lock();
+ try {
+ getFaultsImpl(faultData, pendingFaults, FaultState.PENDING);
+ getFaultsImpl(faultData, doneFaults, FaultState.DONE);
+ } finally {
+ lock.unlock();
+ }
+ return new CoordinatorFaultsResponse(faultData);
+ }
+
+ private void getFaultsImpl(Map<String, CoordinatorFaultsResponse.FaultData> faultData,
+ FaultSet faultSet, FaultState state) {
+ for (Iterator<Fault> iter = faultSet.iterateByStart();
+ iter.hasNext(); ) {
+ Fault fault = iter.next();
+ CoordinatorFaultsResponse.FaultData data =
+ new CoordinatorFaultsResponse.FaultData(fault.spec(), state);
+ faultData.put(fault.id(), data);
+ }
+ }
+
+ public void createFault(CreateCoordinatorFaultRequest request) throws ClassNotFoundException {
+ lock.lock();
+ try {
+ Fault fault = FaultSpec.Util.createFault(request.id(), request.spec());
+ pendingFaults.add(fault);
+ cond.signalAll();
+ } finally {
+ lock.unlock();
+ }
+ }
+
+
+ public static void main(String[] args) throws Exception {
+ ArgumentParser parser = ArgumentParsers
+ .newArgumentParser("trogdor-coordinator")
+ .defaultHelp(true)
+ .description("The Trogdor fault injection coordinator");
+ parser.addArgument("--coordinator.config")
+ .action(store())
+ .required(true)
+ .type(String.class)
+ .dest("config")
+ .metavar("CONFIG")
+ .help("The configuration file to use.");
+ parser.addArgument("--node-name")
+ .action(store())
+ .required(true)
+ .type(String.class)
+ .dest("node_name")
+ .metavar("NODE_NAME")
+ .help("The name of this node.");
+ Namespace res = null;
+ try {
+ res = parser.parseArgs(args);
+ } catch (ArgumentParserException e) {
+ if (args.length == 0) {
+ parser.printHelp();
+ Exit.exit(0);
+ } else {
+ parser.handleError(e);
+ Exit.exit(1);
+ }
+ }
+ String configPath = res.getString("config");
+ String nodeName = res.getString("node_name");
+
+ Platform platform = Platform.Config.parse(nodeName, configPath);
+ JsonRestServer restServer = new JsonRestServer(
+ Node.Util.getTrogdorCoordinatorPort(platform.curNode()));
+ CoordinatorRestResource resource = new CoordinatorRestResource();
+ Coordinator coordinator = new Coordinator(platform, Time.SYSTEM,
+ restServer, resource);
+ restServer.start(resource);
+ Runtime.getRuntime().addShutdownHook(new Thread() {
+ @Override
+ public void run() {
+ log.error("Coordinator shutting down...");
+ }
+ });
+ coordinator.waitForShutdown();
+ }
+};
http://git-wip-us.apache.org/repos/asf/kafka/blob/0772fde5/tools/src/main/java/org/apache/kafka/trogdor/coordinator/CoordinatorClient.java
----------------------------------------------------------------------
diff --git a/tools/src/main/java/org/apache/kafka/trogdor/coordinator/CoordinatorClient.java b/tools/src/main/java/org/apache/kafka/trogdor/coordinator/CoordinatorClient.java
new file mode 100644
index 0000000..1137d08
--- /dev/null
+++ b/tools/src/main/java/org/apache/kafka/trogdor/coordinator/CoordinatorClient.java
@@ -0,0 +1,154 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.kafka.trogdor.coordinator;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import net.sourceforge.argparse4j.ArgumentParsers;
+import net.sourceforge.argparse4j.inf.ArgumentParser;
+import net.sourceforge.argparse4j.inf.ArgumentParserException;
+import net.sourceforge.argparse4j.inf.MutuallyExclusiveGroup;
+import net.sourceforge.argparse4j.inf.Namespace;
+import org.apache.kafka.common.utils.Exit;
+import org.apache.kafka.trogdor.common.JsonUtil;
+import org.apache.kafka.trogdor.rest.CoordinatorFaultsResponse;
+import org.apache.kafka.trogdor.rest.CoordinatorStatusResponse;
+import org.apache.kafka.trogdor.rest.CreateCoordinatorFaultRequest;
+import org.apache.kafka.trogdor.rest.Empty;
+import org.apache.kafka.trogdor.rest.JsonRestServer;
+import org.apache.kafka.trogdor.rest.JsonRestServer.HttpResponse;
+
+import static net.sourceforge.argparse4j.impl.Arguments.store;
+import static net.sourceforge.argparse4j.impl.Arguments.storeTrue;
+
+/**
+ * A client for the Trogdor coordinator.
+ */
+public class CoordinatorClient {
+ /**
+ * The URL target.
+ */
+ private final String target;
+
+ public CoordinatorClient(String host, int port) {
+ this(String.format("%s:%d", host, port));
+ }
+
+ public CoordinatorClient(String target) {
+ this.target = target;
+ }
+
+ private String url(String suffix) {
+ return String.format("http://%s%s", target, suffix);
+ }
+
+ public CoordinatorStatusResponse getStatus() throws Exception {
+ HttpResponse<CoordinatorStatusResponse> resp =
+ JsonRestServer.<CoordinatorStatusResponse>httpRequest(url("/coordinator/status"), "GET",
+ null, new TypeReference<CoordinatorStatusResponse>() { });
+ return resp.body();
+ }
+
+ public CoordinatorFaultsResponse getFaults() throws Exception {
+ HttpResponse<CoordinatorFaultsResponse> resp =
+ JsonRestServer.<CoordinatorFaultsResponse>httpRequest(url("/coordinator/faults"), "GET",
+ null, new TypeReference<CoordinatorFaultsResponse>() { });
+ return resp.body();
+ }
+
+ public void putFault(CreateCoordinatorFaultRequest request) throws Exception {
+ HttpResponse<CreateCoordinatorFaultRequest> resp =
+ JsonRestServer.<CreateCoordinatorFaultRequest>httpRequest(url("/coordinator/fault"), "PUT",
+ request, new TypeReference<CreateCoordinatorFaultRequest>() { });
+ resp.body();
+ }
+
+ public void shutdown() throws Exception {
+ HttpResponse<Empty> resp =
+ JsonRestServer.<Empty>httpRequest(url("/coordinator/shutdown"), "PUT",
+ null, new TypeReference<Empty>() { });
+ resp.body();
+ }
+
+ public static void main(String[] args) throws Exception {
+ ArgumentParser parser = ArgumentParsers
+ .newArgumentParser("trogdor-coordinator-client")
+ .defaultHelp(true)
+ .description("The Trogdor fault injection coordinator client.");
+ parser.addArgument("target")
+ .action(store())
+ .required(true)
+ .type(String.class)
+ .dest("target")
+ .metavar("TARGET")
+ .help("A colon-separated host and port pair. For example, example.com:8889");
+ MutuallyExclusiveGroup actions = parser.addMutuallyExclusiveGroup();
+ actions.addArgument("--status")
+ .action(storeTrue())
+ .type(Boolean.class)
+ .dest("status")
+ .help("Get coordinator status.");
+ actions.addArgument("--get-faults")
+ .action(storeTrue())
+ .type(Boolean.class)
+ .dest("get_faults")
+ .help("Get coordinator faults.");
+ actions.addArgument("--create-fault")
+ .action(store())
+ .type(String.class)
+ .dest("create_fault")
+ .metavar("FAULT_JSON")
+ .help("Create a new fault.");
+ actions.addArgument("--shutdown")
+ .action(storeTrue())
+ .type(Boolean.class)
+ .dest("shutdown")
+ .help("Trigger coordinator shutdown");
+
+ Namespace res = null;
+ try {
+ res = parser.parseArgs(args);
+ } catch (ArgumentParserException e) {
+ if (args.length == 0) {
+ parser.printHelp();
+ Exit.exit(0);
+ } else {
+ parser.handleError(e);
+ Exit.exit(1);
+ }
+ }
+ String target = res.getString("target");
+ CoordinatorClient client = new CoordinatorClient(target);
+ if (res.getBoolean("status")) {
+ System.out.println("Got coordinator status: " +
+ JsonUtil.toPrettyJsonString(client.getStatus()));
+ } else if (res.getBoolean("get_faults")) {
+ System.out.println("Got coordinator faults: " +
+ JsonUtil.toPrettyJsonString(client.getFaults()));
+ } else if (res.getString("create_fault") != null) {
+ client.putFault(JsonUtil.JSON_SERDE.readValue(res.getString("create_fault"),
+ CreateCoordinatorFaultRequest.class));
+ System.out.println("Created fault.");
+ } else if (res.getBoolean("shutdown")) {
+ client.shutdown();
+ System.out.println("Sent shutdown request.");
+ } else {
+ System.out.println("You must choose an action. Type --help for help.");
+ Exit.exit(1);
+ }
+ }
+};
http://git-wip-us.apache.org/repos/asf/kafka/blob/0772fde5/tools/src/main/java/org/apache/kafka/trogdor/coordinator/CoordinatorRestResource.java
----------------------------------------------------------------------
diff --git a/tools/src/main/java/org/apache/kafka/trogdor/coordinator/CoordinatorRestResource.java b/tools/src/main/java/org/apache/kafka/trogdor/coordinator/CoordinatorRestResource.java
new file mode 100644
index 0000000..60357b8
--- /dev/null
+++ b/tools/src/main/java/org/apache/kafka/trogdor/coordinator/CoordinatorRestResource.java
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.kafka.trogdor.coordinator;
+
+import org.apache.kafka.trogdor.rest.CoordinatorFaultsResponse;
+import org.apache.kafka.trogdor.rest.CoordinatorStatusResponse;
+import org.apache.kafka.trogdor.rest.CreateCoordinatorFaultRequest;
+import org.apache.kafka.trogdor.rest.Empty;
+
+import javax.servlet.ServletContext;
+import javax.ws.rs.Consumes;
+import javax.ws.rs.GET;
+import javax.ws.rs.PUT;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.MediaType;
+import java.util.concurrent.atomic.AtomicReference;
+
+
+@Path("/coordinator")
+@Produces(MediaType.APPLICATION_JSON)
+@Consumes(MediaType.APPLICATION_JSON)
+public class CoordinatorRestResource {
+ private final AtomicReference<Coordinator> coordinator = new AtomicReference<Coordinator>();
+
+ @javax.ws.rs.core.Context
+ private ServletContext context;
+
+ public void setCoordinator(Coordinator myCoordinator) {
+ coordinator.set(myCoordinator);
+ }
+
+ @GET
+ @Path("/status")
+ public CoordinatorStatusResponse getStatus() throws Throwable {
+ return new CoordinatorStatusResponse(coordinator().startTimeMs());
+ }
+
+ @GET
+ @Path("/faults")
+ public CoordinatorFaultsResponse getCoordinatorFaults() throws Throwable {
+ return coordinator().getFaults();
+ }
+
+ @PUT
+ @Path("/fault")
+ public Empty putCoordinatorFault(CreateCoordinatorFaultRequest request) throws Throwable {
+ coordinator().createFault(request);
+ return Empty.INSTANCE;
+ }
+
+ @PUT
+ @Path("/shutdown")
+ public Empty shutdown() throws Throwable {
+ coordinator().beginShutdown();
+ return Empty.INSTANCE;
+ }
+
+ private Coordinator coordinator() {
+ Coordinator myCoordinator = coordinator.get();
+ if (myCoordinator == null) {
+ throw new RuntimeException("CoordinatorRestResource has not been initialized yet.");
+ }
+ return myCoordinator;
+ }
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/0772fde5/tools/src/main/java/org/apache/kafka/trogdor/coordinator/NodeManager.java
----------------------------------------------------------------------
diff --git a/tools/src/main/java/org/apache/kafka/trogdor/coordinator/NodeManager.java b/tools/src/main/java/org/apache/kafka/trogdor/coordinator/NodeManager.java
new file mode 100644
index 0000000..04d714c
--- /dev/null
+++ b/tools/src/main/java/org/apache/kafka/trogdor/coordinator/NodeManager.java
@@ -0,0 +1,262 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.kafka.trogdor.coordinator;
+
+import org.apache.kafka.common.utils.KafkaThread;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.trogdor.agent.AgentClient;
+import org.apache.kafka.trogdor.common.Node;
+import org.apache.kafka.trogdor.common.Platform;
+import org.apache.kafka.trogdor.fault.Fault;
+import org.apache.kafka.trogdor.rest.AgentStatusResponse;
+import org.apache.kafka.trogdor.rest.CreateAgentFaultRequest;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+class NodeManager {
+ private static final Logger log = LoggerFactory.getLogger(NodeManager.class);
+
+ /**
+ * The Time object used to fetch the current time.
+ */
+ private final Time time;
+
+ /**
+ * The node which is being managed.
+ */
+ private final Node node;
+
+ /**
+ * The client for the node being managed.
+ */
+ private final AgentClient client;
+
+ /**
+ * The maximum amount of time to go without contacting the node.
+ */
+ private final long heartbeatMs;
+
+ /**
+ * True if the NodeManager is shutting down. Protected by the queueLock.
+ */
+ private boolean shutdown = false;
+
+ /**
+ * The Node Manager runnable.
+ */
+ private final NodeManagerRunnable runnable;
+
+ /**
+ * The Node Manager thread.
+ */
+ private final KafkaThread thread;
+
+ /**
+ * The lock protecting the NodeManager fields.
+ */
+ private final Lock lock = new ReentrantLock();
+
+ /**
+ * The condition variable used to wake the thread when it is waiting for a
+ * queue or shutdown change.
+ */
+ private final Condition cond = lock.newCondition();
+
+ /**
+ * A queue of faults which should be sent to this node. Protected by the lock.
+ */
+ private final List<Fault> faultQueue = new ArrayList<>();
+
+ /**
+ * The last time we successfully contacted the node. Protected by the lock.
+ */
+ private long lastContactMs = 0;
+
+ /**
+ * The current status of this node.
+ */
+ public static class NodeStatus {
+ private final String nodeName;
+ private final long lastContactMs;
+
+ NodeStatus(String nodeName, long lastContactMs) {
+ this.nodeName = nodeName;
+ this.lastContactMs = lastContactMs;
+ }
+
+ public String nodeName() {
+ return nodeName;
+ }
+
+ public long lastContactMs() {
+ return lastContactMs;
+ }
+ }
+
+ class NodeManagerRunnable implements Runnable {
+ @Override
+ public void run() {
+ try {
+ Fault fault = null;
+ long lastCommAttemptMs = 0;
+ while (true) {
+ long now = time.milliseconds();
+ if (fault != null) {
+ lastCommAttemptMs = now;
+ if (sendFault(now, fault)) {
+ fault = null;
+ }
+ }
+ long nextCommAttemptMs = lastCommAttemptMs + heartbeatMs;
+ if (now < nextCommAttemptMs) {
+ lastCommAttemptMs = now;
+ sendHeartbeat(now);
+ }
+ long waitMs = Math.max(0L, nextCommAttemptMs - now);
+ lock.lock();
+ try {
+ if (shutdown) {
+ return;
+ }
+ try {
+ cond.await(waitMs, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ log.info("{}: NodeManagerRunnable got InterruptedException", node.name());
+ Thread.currentThread().interrupt();
+ }
+ if (fault == null) {
+ if (!faultQueue.isEmpty()) {
+ fault = faultQueue.remove(0);
+ }
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+ } catch (Throwable e) {
+ log.warn("{}: exiting NodeManagerRunnable with exception", node.name(), e);
+ } finally {
+ }
+ }
+ }
+
+ NodeManager(Time time, Node node) {
+ this.time = time;
+ this.node = node;
+ this.client = new AgentClient(node.hostname(), Node.Util.getTrogdorAgentPort(node));
+ this.heartbeatMs = Node.Util.getIntConfig(node,
+ Platform.Config.TROGDOR_COORDINATOR_HEARTBEAT_MS,
+ Platform.Config.TROGDOR_COORDINATOR_HEARTBEAT_MS_DEFAULT);
+ this.runnable = new NodeManagerRunnable();
+ this.thread = new KafkaThread("NodeManagerThread(" + node.name() + ")", runnable, false);
+ this.thread.start();
+ }
+
+ private boolean sendFault(long now, Fault fault) {
+ try {
+ client.putFault(new CreateAgentFaultRequest(fault.id(), fault.spec()));
+ } catch (Exception e) {
+ log.warn("{}: error sending fault to {}.", node.name(), client.target(), e);
+ return false;
+ }
+ lock.lock();
+ try {
+ lastContactMs = now;
+ } finally {
+ lock.unlock();
+ }
+ return true;
+ }
+
+ private void sendHeartbeat(long now) {
+ AgentStatusResponse status = null;
+ try {
+ status = client.getStatus();
+ } catch (Exception e) {
+ log.warn("{}: error sending heartbeat to {}.", node.name(), client.target(), e);
+ return;
+ }
+ lock.lock();
+ try {
+ lastContactMs = now;
+ } finally {
+ lock.unlock();
+ }
+ log.debug("{}: got heartbeat status {}.", node.name(), status);
+ }
+
+ public void beginShutdown() {
+ lock.lock();
+ try {
+ if (shutdown)
+ return;
+ log.trace("{}: beginning shutdown.", node.name());
+ shutdown = true;
+ cond.signalAll();
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ public void waitForShutdown() {
+ log.trace("waiting for NodeManager({}) shutdown.", node.name());
+ try {
+ thread.join();
+ } catch (InterruptedException e) {
+ log.error("{}: Interrupted while waiting for thread shutdown", node.name(), e);
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ /**
+ * Get the current status of this node.
+ *
+ * @return The node status.
+ */
+ public NodeStatus status() {
+ lock.lock();
+ try {
+ return new NodeStatus(node.name(), lastContactMs);
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Enqueue a new fault.
+ *
+ * @param fault The fault to enqueue.
+ */
+ public void enqueueFault(Fault fault) {
+ lock.lock();
+ try {
+ log.trace("{}: added {} to fault queue.", node.name(), fault);
+ faultQueue.add(fault);
+ cond.signalAll();
+ } finally {
+ lock.unlock();
+ }
+ }
+};
http://git-wip-us.apache.org/repos/asf/kafka/blob/0772fde5/tools/src/main/java/org/apache/kafka/trogdor/fault/AbstractFaultSpec.java
----------------------------------------------------------------------
diff --git a/tools/src/main/java/org/apache/kafka/trogdor/fault/AbstractFaultSpec.java b/tools/src/main/java/org/apache/kafka/trogdor/fault/AbstractFaultSpec.java
new file mode 100644
index 0000000..5f551b5
--- /dev/null
+++ b/tools/src/main/java/org/apache/kafka/trogdor/fault/AbstractFaultSpec.java
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.kafka.trogdor.fault;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.kafka.trogdor.common.JsonUtil;
+
+
+/**
+ * A base class that can be used for FaultSpecs.
+ */
+public abstract class AbstractFaultSpec implements FaultSpec {
+ private final long startMs;
+ private final long durationMs;
+
+ protected AbstractFaultSpec(@JsonProperty("startMs") long startMs,
+ @JsonProperty("durationMs") long durationMs) {
+ this.startMs = startMs;
+ this.durationMs = durationMs;
+ }
+
+ @JsonProperty
+ @Override
+ public long startMs() {
+ return startMs;
+ }
+
+ @JsonProperty
+ @Override
+ public long durationMs() {
+ return durationMs;
+ }
+
+ @Override
+ public String toString() {
+ return JsonUtil.toJsonString(this);
+ }
+}