package com.cloudera.flume.agent.diskfailover;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import java.io.File;
import java.io.IOException;
import java.util.Collection;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import com.cloudera.flume.agent.DirectMasterRPC;
import com.cloudera.flume.agent.FlumeNode;
import com.cloudera.flume.agent.LogicalNode;
import com.cloudera.flume.conf.FlumeSpecException;
import com.cloudera.flume.core.EventSink;
import com.cloudera.flume.core.EventSource;
import com.cloudera.flume.core.EventUtil;
import com.cloudera.flume.handlers.debug.NoNlASCIISynthSource;
import com.cloudera.flume.handlers.rolling.RollSink;
import com.cloudera.flume.handlers.rolling.TimeTrigger;
import com.cloudera.flume.master.FlumeMaster;
import com.cloudera.flume.reporter.ReportEvent;
import com.cloudera.flume.reporter.ReportManager;
import com.cloudera.flume.reporter.Reportable;
import com.cloudera.flume.reporter.aggregator.AccumulatorSink;
import com.cloudera.flume.reporter.aggregator.CounterSink;
import com.cloudera.util.BenchmarkHarness;
import com.cloudera.util.FileUtil;
public static Logger LOG = Logger.getLogger(TestConcurrentDFOMan.class);
@Before
Logger.getLogger(CounterSink.class).setLevel(Level.DEBUG);
Logger.getLogger(AccumulatorSink.class).setLevel(Level.DEBUG);
Logger.getLogger(RollSink.class).setLevel(Level.DEBUG);
}
@Test
public void test1thread()
throws IOException, InterruptedException {
doTestConcurrentDFOMans(1, 10000, 60000);
}
@Test
public void test10thread()
throws IOException, InterruptedException {
doTestConcurrentDFOMans(10, 10000, 60000);
}
@Test
public void test100thread()
throws IOException, InterruptedException {
doTestConcurrentDFOMans(100, 1000, 60000);
}
@Test
@Ignore("Test takes too long")
public void test1000thread()
throws IOException, InterruptedException {
doTestConcurrentDFOMans(1000, 100, 120000);
}
@Test
FlumeSpecException {
doTestLogicalNodesConcurrentDFOMans(5, 10000, 180000);
}
@Test
InterruptedException, FlumeSpecException {
doTestLogicalNodesConcurrentDFOMans(10, 10000, 180000);
}
@Test
InterruptedException, FlumeSpecException {
doTestLogicalNodesConcurrentDFOMans(10, 1000, 60000);
}
@Test
FlumeSpecException {
doTestLogicalNodesConcurrentDFOMans(10, 10000, 120000);
}
@Test
FlumeSpecException {
doTestLogicalNodesConcurrentDFOMans(100, 1000, 60000);
}
@Test
@Ignore("takes too long")
InterruptedException, FlumeSpecException {
doTestLogicalNodesConcurrentDFOMans(100, 10000, 1800000);
}
@Test
@Ignore("takes too long")
FlumeSpecException {
doTestLogicalNodesConcurrentDFOMans(1000, 100, 60000);
}
int timeout) throws IOException, InterruptedException {
final CountDownLatch started = new CountDownLatch(threads);
final CountDownLatch done = new CountDownLatch(threads);
final DiskFailoverManager[] dfos = new DiskFailoverManager[threads];
for (int i = 0; i < threads; i++) {
final int idx = i;
new Thread("Concurrent-" + i) {
@Override
try {
File f1 = FileUtil.mktempdir();
AccumulatorSink cnt1 = new AccumulatorSink("count." + idx);
DiskFailoverManager dfoMan = new NaiveFileFailoverManager(f1);
dfos[idx] = dfoMan;
EventSink snk = new DiskFailoverDeco<EventSink>(cnt1, dfoMan,
new TimeTrigger(100), 50);
ReportManager.get().add(cnt1);
EventSource src = new NoNlASCIISynthSource(events + idx, 100);
src.open();
snk.open();
started.countDown();
EventUtil.dumpAll(src, snk);
src.close();
snk.close();
FileUtil.rmr(f1);
} catch (Exception e) {
LOG.error(e, e);
} finally {
done.countDown();
}
}
}.start();
}
started.await();
boolean ok = done.await(timeout, TimeUnit.MILLISECONDS);
assertTrue("Test timed out", ok);
for (int i = 0; i < threads; i++) {
AccumulatorSink cnt = (AccumulatorSink) ReportManager.get()
.getReportable("count." + i);
int exp = events + i;
LOG
.info("count." + i + " expected " + exp + " and got "
+ cnt.getCount());
assertEquals(exp, (int) cnt.getCount());
ReportEvent rpt = dfos[i].getReport();
LOG.info(rpt);
long failovered = rpt.getLongMetric(DiskFailoverManager.A_MSG_WRITING);
assertEquals(events + i, failovered);
}
}
final int events, int timeout) throws IOException, InterruptedException,
FlumeSpecException {
BenchmarkHarness.setupLocalWriteDir();
FlumeMaster master = new FlumeMaster();
FlumeNode node = new FlumeNode(new DirectMasterRPC(master), false, false);
final Reportable[] dfos = new Reportable[threads];
for (int i = 0; i < threads; i++) {
String name = "test." + i;
String report = "report." + i;
int count = events + i;
String src = "asciisynth(" + count + ",100)";
String snk = "{ diskFailover => counter(\"" + report + "\") } ";
node.getLogicalNodeManager().testingSpawn(name, src, snk);
dfos[i] = node.getLogicalNodeManager().get(name);
}
Thread.sleep(500);
waitForEmptyDFOs(node, timeout);
boolean success = true;
for (int i = 0; i < threads; i++) {
LOG.info(dfos[i].getReport());
}
for (int i = 0; i < threads; i++) {
CounterSink cnt = (CounterSink) ReportManager.get().getReportable(
"report." + i);
LOG.info(i + " expected " + (events + i) + " and got " + cnt.getCount());
success &= ((events + i) == cnt.getCount());
assertEquals(events + i, cnt.getCount());
}
assertTrue("Counts did not line up", success);
BenchmarkHarness.cleanupLocalWriteDir();
}
throws InterruptedException {
boolean done = false;
long start = System.currentTimeMillis();
while (!done) {
if (System.currentTimeMillis() - start > timeout) {
fail("Test took too long");
}
Collection<LogicalNode> lns = node.getLogicalNodeManager().getNodes();
done = areDFOsReconfigured(lns) && areDFOsEmpty(lns);
if (!done) {
Thread.sleep(250);
}
}
}
for (LogicalNode n : lns) {
long val = n.getReport().getLongMetric(LogicalNode.A_RECONFIGURES);
if (val == 0) {
return false;
}
}
return true;
}
for (LogicalNode n : lns) {
DiskFailoverManager dfo = FlumeNode.getInstance().getDFOManager(
n.getName());
if (!dfo.isEmpty())
return false;
}
return true;
}
}