package com.cloudera.flume.handlers.debug;
import java.io.IOException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.cloudera.flume.conf.Context;
import com.cloudera.flume.conf.SinkFactory.SinkDecoBuilder;
import com.cloudera.flume.core.Event;
import com.cloudera.flume.core.EventImpl;
import com.cloudera.flume.core.EventSink;
import com.cloudera.flume.core.EventSinkDecorator;
import com.cloudera.util.CharEncUtils;
import com.cloudera.util.Clock;
import com.cloudera.util.NetUtils;
import com.google.common.base.Preconditions;
EventSinkDecorator<S> {
static final Logger LOG = LoggerFactory.getLogger(BenchmarkInjectDecorator.class);
final static AtomicInteger count = new AtomicInteger();
public final static String ATTR_BENCHMARK = "BenchmarkInject";
public final static String ATTR_BENCHMARK_TAG = "BenchmarkInjectTag";
public final static byte[] BENCH_START = "start".getBytes(CharEncUtils.RAW);
public final static byte[] BENCH_FIRST = "first".getBytes(CharEncUtils.RAW);
public final static byte[] BENCH_STOP = "stop".getBytes(CharEncUtils.RAW);
public final static byte[] BENCH_ERROR = "error".getBytes(CharEncUtils.RAW);
final byte[] tag;
AtomicBoolean first = new AtomicBoolean(true);
super(s);
Preconditions.checkArgument(tag != null);
this.tag = tag.getBytes();
}
super(s);
String t = NetUtils.localhost()
+ String.format("-%06d-", count.getAndIncrement()) + Clock.timeStamp();
tag = t.getBytes();
}
e.set(ATTR_BENCHMARK_TAG, tag);
e.set(ATTR_BENCHMARK, evt);
return e;
}
@Override
public void open()
throws IOException {
super.open();
super.append(tagBench(new EventImpl(new byte[0]), BENCH_START));
}
@Override
public void close()
throws IOException {
super.append(tagBench(getReport(), BENCH_STOP));
super.close();
}
@Override
public void append(Event e)
throws IOException {
if (first.getAndSet(false)) {
super.append(tagBench(new EventImpl(new byte[0]), BENCH_FIRST));
}
super.append(e);
}
public static SinkDecoBuilder
builder() {
return new SinkDecoBuilder() {
@Override
public EventSinkDecorator<EventSink>
build(Context context,
String... argv) {
Preconditions.checkArgument(argv.length == 0 || argv.length == 1,
"usage: benchinject[(tag)]");
String tag = (argv.length == 0) ? NetUtils.localhost()
+ String.format("-%06d-", count.getAndIncrement())
+ Clock.timeStamp() : argv[0];
return new BenchmarkInjectDecorator<EventSink>(null, tag);
}
};
}
}