diff --git a/hbase-it/src/test/java/org/apache/hadoop/hbase/IntegrationTestDataIngestSlowDeterministic.java b/hbase-it/src/test/java/org/apache/hadoop/hbase/IntegrationTestDataIngestSlowDeterministic.java new file mode 100644 index 00000000000..cd52310b81f --- /dev/null +++ b/hbase-it/src/test/java/org/apache/hadoop/hbase/IntegrationTestDataIngestSlowDeterministic.java @@ -0,0 +1,76 @@ +/** + * 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.hadoop.hbase; + +import org.apache.hadoop.hbase.util.ChaosMonkey; +import org.apache.hadoop.hbase.util.ChaosMonkey.BatchRestartRs; +import org.apache.hadoop.hbase.util.ChaosMonkey.RestartActiveMaster; +import org.apache.hadoop.hbase.util.ChaosMonkey.RestartRandomRs; +import org.apache.hadoop.hbase.util.ChaosMonkey.RestartRsHoldingMeta; +import org.apache.hadoop.hbase.util.ChaosMonkey.RestartRsHoldingRoot; +import org.apache.hadoop.hbase.util.ChaosMonkey.RollingBatchRestartRs; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.experimental.categories.Category; + +/** + * A system test which does large data ingestion and verify using {@link LoadTestTool}. + * It performs a set of actions deterministically using ChaosMonkey, then starts killing + * things randomly. You can configure how long should the load test run by using + * "hbase.IntegrationTestDataIngestSlowDeterministic.runtime" configuration parameter. + */ +@Category(IntegrationTests.class) +public class IntegrationTestDataIngestSlowDeterministic extends IngestIntegrationTestBase { + private static final int SERVER_COUNT = 4; // number of slaves for the smallest cluster + private static final long DEFAULT_RUN_TIME = 30 * 60 * 1000; + private static final long CHAOS_EVERY_MS = 150 * 1000; // Chaos every 2.5 minutes. + + private ChaosMonkey monkey; + + @Before + public void setUp() throws Exception { + super.setUp(SERVER_COUNT); + ChaosMonkey.Action[] actions = new ChaosMonkey.Action[] { + new RestartRandomRs(60000), + new BatchRestartRs(5000, 0.5f), + new RestartActiveMaster(5000), + new RollingBatchRestartRs(5000, 1.0f), + new RestartRsHoldingMeta(35000), + new RestartRsHoldingRoot(35000) + }; + monkey = new ChaosMonkey(util, new ChaosMonkey.CompositeSequenialPolicy( + new ChaosMonkey.DoActionsOncePolicy(CHAOS_EVERY_MS, actions), + new ChaosMonkey.PeriodicRandomActionPolicy(CHAOS_EVERY_MS, actions))); + monkey.start(); + } + + @After + public void tearDown() throws Exception { + if (monkey != null) { + monkey.stop("tearDown"); + monkey.waitForStop(); + } + super.tearDown(); + } + + @Test + public void testDataIngest() throws Exception { + runIngestTest(DEFAULT_RUN_TIME, 2500, 10, 1024, 10); + } +} diff --git a/hbase-it/src/test/java/org/apache/hadoop/hbase/IntegrationTestDataIngestWithChaosMonkey.java b/hbase-it/src/test/java/org/apache/hadoop/hbase/IntegrationTestDataIngestWithChaosMonkey.java index 842ecf96bc1..72891b1849c 100644 --- a/hbase-it/src/test/java/org/apache/hadoop/hbase/IntegrationTestDataIngestWithChaosMonkey.java +++ b/hbase-it/src/test/java/org/apache/hadoop/hbase/IntegrationTestDataIngestWithChaosMonkey.java @@ -18,11 +18,6 @@ package org.apache.hadoop.hbase; -import java.io.IOException; - -import junit.framework.Assert; - -import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.hbase.util.ChaosMonkey; import org.junit.After; import org.junit.Before; diff --git a/hbase-it/src/test/java/org/apache/hadoop/hbase/util/ChaosMonkey.java b/hbase-it/src/test/java/org/apache/hadoop/hbase/util/ChaosMonkey.java index 3f694094e29..93d09f62af2 100644 --- a/hbase-it/src/test/java/org/apache/hadoop/hbase/util/ChaosMonkey.java +++ b/hbase-it/src/test/java/org/apache/hadoop/hbase/util/ChaosMonkey.java @@ -453,16 +453,122 @@ public class ChaosMonkey extends AbstractHBaseTool implements Stoppable { } } + /** A policy that runs multiple other policies one after the other */ + public static class CompositeSequenialPolicy extends Policy { + private List policies; + public CompositeSequenialPolicy(Policy... policies) { + this.policies = Arrays.asList(policies); + } + + @Override + public void stop(String why) { + super.stop(why); + for (Policy p : policies) { + p.stop(why); + } + } + + @Override + public void run() { + for (Policy p : policies) { + p.run(); + } + } + + @Override + public void init(PolicyContext context) throws Exception { + super.init(context); + for (Policy p : policies) { + p.init(context); + } + } + } + + /** A policy which does stuff every time interval. */ + public static abstract class PeriodicPolicy extends Policy { + private long periodMs; + + public PeriodicPolicy(long periodMs) { + this.periodMs = periodMs; + } + + @Override + public void run() { + // Add some jitter. + int jitter = new Random().nextInt((int)periodMs); + LOG.info("Sleeping for " + jitter + " to add jitter"); + Threads.sleep(jitter); + + while (!isStopped()) { + long start = System.currentTimeMillis(); + runOneIteration(); + + if (isStopped()) return; + long sleepTime = periodMs - (System.currentTimeMillis() - start); + if (sleepTime > 0) { + LOG.info("Sleeping for: " + sleepTime); + Threads.sleep(sleepTime); + } + } + } + + protected abstract void runOneIteration(); + + @Override + public void init(PolicyContext context) throws Exception { + super.init(context); + LOG.info("Using ChaosMonkey Policy: " + this.getClass() + ", period: " + periodMs); + } + } + + + /** A policy which performs a sequence of actions deterministically. */ + public static class DoActionsOncePolicy extends PeriodicPolicy { + private List actions; + + public DoActionsOncePolicy(long periodMs, List actions) { + super(periodMs); + this.actions = new ArrayList(actions); + } + + public DoActionsOncePolicy(long periodMs, Action... actions) { + this(periodMs, Arrays.asList(actions)); + } + + @Override + protected void runOneIteration() { + if (actions.isEmpty()) { + this.stop("done"); + return; + } + Action action = actions.remove(0); + + try { + action.perform(); + } catch (Exception ex) { + LOG.warn("Exception occured during performing action: " + + StringUtils.stringifyException(ex)); + } + } + + @Override + public void init(PolicyContext context) throws Exception { + super.init(context); + for (Action action : actions) { + action.init(this.context); + } + } + } + /** * A policy, which picks a random action according to the given weights, * and performs it every configurable period. */ - public static class PeriodicRandomActionPolicy extends Policy { - private long periodMs; + public static class PeriodicRandomActionPolicy extends PeriodicPolicy { private List> actions; public PeriodicRandomActionPolicy(long periodMs, List> actions) { - this.periodMs = periodMs; + super(periodMs); this.actions = actions; } @@ -472,7 +578,7 @@ public class ChaosMonkey extends AbstractHBaseTool implements Stoppable { } public PeriodicRandomActionPolicy(long periodMs, Action... actions) { - this.periodMs = periodMs; + super(periodMs); this.actions = new ArrayList>(actions.length); for (Action action : actions) { this.actions.add(new Pair(action, 1)); @@ -480,35 +586,19 @@ public class ChaosMonkey extends AbstractHBaseTool implements Stoppable { } @Override - public void run() { - //add some jitter - int jitter = new Random().nextInt((int)periodMs); - LOG.info("Sleeping for " + jitter + " to add jitter"); - Threads.sleep(jitter); - - while (!isStopped()) { - long start = System.currentTimeMillis(); - Action action = selectWeightedRandomItem(actions); - - try { - action.perform(); - } catch (Exception ex) { - LOG.warn("Exception occured during performing action: " - + StringUtils.stringifyException(ex)); - } - - long sleepTime = periodMs - (System.currentTimeMillis() - start); - if (sleepTime > 0) { - LOG.info("Sleeping for:" + sleepTime); - Threads.sleep(sleepTime); - } + protected void runOneIteration() { + Action action = selectWeightedRandomItem(actions); + try { + action.perform(); + } catch (Exception ex) { + LOG.warn("Exception occured during performing action: " + + StringUtils.stringifyException(ex)); } } @Override public void init(PolicyContext context) throws Exception { super.init(context); - LOG.info("Using ChaosMonkey Policy: " + this.getClass() + ", period:" + periodMs); for (Pair action : actions) { action.getFirst().init(this.context); } @@ -661,4 +751,4 @@ public class ChaosMonkey extends AbstractHBaseTool implements Stoppable { System.exit(ret); } -} \ No newline at end of file +}