package com.cloudera.flume.agent.durability;
import java.io.File;
import java.io.IOException;
import java.util.Collection;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.cloudera.flume.conf.Context;
import com.cloudera.flume.core.Attributes;
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.flume.core.EventSource;
import com.cloudera.flume.handlers.endtoend.AckChecksumInjector;
import com.cloudera.flume.handlers.endtoend.AckListener;
import com.cloudera.flume.handlers.hdfs.SeqfileEventSink;
import com.cloudera.flume.handlers.hdfs.SeqfileEventSource;
import com.cloudera.flume.handlers.rolling.RollSink;
import com.cloudera.flume.handlers.rolling.RollTrigger;
import com.cloudera.flume.handlers.rolling.Tagger;
import com.cloudera.flume.reporter.ReportEvent;
import com.cloudera.util.FileUtil;
import com.google.common.base.Preconditions;
static final Logger LOG = LoggerFactory.getLogger(NaiveFileWALManager.class);
final private ConcurrentHashMap<String, WALData> table = new ConcurrentHashMap<String, WALData>();
final private LinkedBlockingQueue<String> writingQ = new LinkedBlockingQueue<String>();
final private LinkedBlockingQueue<String> loggedQ = new LinkedBlockingQueue<String>();
final private LinkedBlockingQueue<String> sendingQ = new LinkedBlockingQueue<String>();
final private LinkedBlockingQueue<String> sentQ = new LinkedBlockingQueue<String>();
public static final String IMPORTDIR = "import";
public static final String WRITINGDIR = "writing";
public static final String LOGGEDDIR = "logged";
public static final String SENDINGDIR = "sending";
public static final String SENTDIR = "sent";
public static final String ERRORDIR = "error";
public static final String DONEDIR = "done";
private AtomicLong importedCount = new AtomicLong(0);
private AtomicLong writingCount = new AtomicLong(0);
private AtomicLong loggedCount = new AtomicLong(0);
private AtomicLong sendingCount = new AtomicLong(0);
private AtomicLong sentCount = new AtomicLong(0);
private AtomicLong ackedCount = new AtomicLong(0);
private AtomicLong retryCount = new AtomicLong(0);
private AtomicLong recoverCount = new AtomicLong(0);
private AtomicLong errCount = new AtomicLong(0);
private volatile boolean closed = false;
State s;
String tag;
this.s = State.WRITING;
this.tag = tag;
}
WALData data = new WALData(tag);
data.s = State.LOGGED;
return data;
}
};
File importDir, writingDir, loggedDir, sendingDir, sentDir, doneDir,
errorDir;
File baseDir;
File writingDir = new File(baseDir, WRITINGDIR);
File loggedDir = new File(baseDir, LOGGEDDIR);
File xmitableDir = new File(baseDir, SENDINGDIR);
File ackedDir = new File(baseDir, SENTDIR);
File doneDir = new File(baseDir, DONEDIR);
File errDir = new File(baseDir, ERRORDIR);
Preconditions.checkNotNull(writingDir);
Preconditions.checkNotNull(loggedDir);
Preconditions.checkNotNull(xmitableDir);
Preconditions.checkNotNull(ackedDir);
Preconditions.checkNotNull(errDir);
Preconditions.checkNotNull(doneDir);
this.importDir = new File(baseDir, IMPORTDIR);
this.writingDir = writingDir;
this.loggedDir = loggedDir;
this.sendingDir = xmitableDir;
this.sentDir = ackedDir;
this.doneDir = doneDir;
this.errorDir = errDir;
this.baseDir = baseDir;
}
synchronized public void open()
throws IOException {
if (!FileUtil.makeDirs(importDir)) {
throw new IOException("Unable to create import dir: " + importDir);
}
if (!FileUtil.makeDirs(writingDir)) {
throw new IOException("Unable to create writing dir: " + writingDir);
}
if (!FileUtil.makeDirs(loggedDir)) {
throw new IOException("Unable to create logged dir: " + loggedDir);
}
if (!FileUtil.makeDirs(sendingDir)) {
throw new IOException("Unable to create sending dir: " + sendingDir);
}
if (!FileUtil.makeDirs(sentDir)) {
throw new IOException("Unable to create import dir: " + sentDir);
}
if (!FileUtil.makeDirs(doneDir)) {
throw new IOException("Unable to create writing dir: " + doneDir);
}
if (!FileUtil.makeDirs(errorDir)) {
throw new IOException("Unable to create logged dir: " + errorDir);
}
closed = false;
}
return Collections.unmodifiableCollection(writingQ);
}
return Collections.unmodifiableCollection(loggedQ);
}
return Collections.unmodifiableCollection(sendingQ);
}
return Collections.unmodifiableCollection(sentQ);
}
synchronized public void stopDrains()
throws IOException {
closed = true;
}
synchronized public void recover()
throws IOException {
for (String f : writingDir.list()) {
File old = new File(writingDir, f);
if (!old.isFile() || !old.renameTo(new File(loggedDir, f))) {
throw new IOException("Unable to recover - couldn't rename " + old
+ " to " + loggedDir + f);
}
LOG.debug("Recover moved " + f + " from WRITING to LOGGED");
}
for (String f : sendingDir.list()) {
File old = new File(sendingDir, f);
if (!old.isFile() || !old.renameTo(new File(loggedDir, f))) {
throw new IOException("Unable to recover - couldn't rename " + old
+ " to " + loggedDir + f);
}
LOG.debug("Recover moved " + f + " from SENDING to LOGGED");
}
for (String f : sentDir.list()) {
File old = new File(sentDir, f);
if (!old.isFile() || !old.renameTo(new File(loggedDir, f))) {
throw new IOException("Unable to recover - couldn't rename " + old
+ " to " + loggedDir + f);
}
LOG.debug("Recover moved " + f + " from SENT to LOGGED");
}
for (String f : loggedDir.list()) {
WALData data = WALData.recovered(f);
table.put(f, data);
loggedQ.add(f);
recoverCount.incrementAndGet();
LOG.debug("Recover loaded " + f);
}
}
AckListener al) throws IOException {
File dir = getDir(State.WRITING);
final String tag = tagger.newTag();
EventSink bareSink = new SeqfileEventSink(new File(dir, tag)
.getAbsoluteFile());
EventSink curSink = new AckChecksumInjector<EventSink>(bareSink, tag
.getBytes(), al);
writingQ.add(tag);
WALData data = new WALData(tag);
table.put(tag, data);
writingCount.incrementAndGet();
return new EventSinkDecorator<EventSink>(curSink) {
@Override
public void append(Event e)
throws IOException {
getSink().append(e);
}
@Override
public void close()
throws IOException {
super.close();
synchronized (NaiveFileWALManager.this) {
if (!writingQ.contains(tag)) {
LOG.warn("Already changed tag " + tag + " out of WRITING state");
return;
}
LOG.info("File lives in " + getFile(tag));
changeState(tag, State.WRITING, State.LOGGED);
loggedCount.incrementAndGet();
}
}
};
}
throws IOException {
File dir = getDir(State.WRITING);
final String tag = tagger.newTag();
EventSink curSink = new SeqfileEventSink(new File(dir, tag)
.getAbsoluteFile());
writingQ.add(tag);
WALData data = new WALData(tag);
table.put(tag, data);
return new EventSinkDecorator<EventSink>(curSink) {
@Override
public void append(Event e)
throws IOException {
LOG.debug("Appending event: {}", e);
getSink().append(e);
}
@Override
public void close()
throws IOException {
super.close();
changeState(tag, State.WRITING, State.LOGGED);
}
};
}
@Override
final AckListener ackQueue, long checkMs) throws IOException {
return new RollSink(ctx, "ackingWal", t, checkMs) {
@Override
public EventSink
newSink(Context ctx)
throws IOException {
return newAckWritingSink(t.getTagger(), ackQueue);
}
};
}
private LinkedBlockingQueue<String>
getQueue(State state) {
Preconditions.checkNotNull(state);
switch (state) {
case WRITING:
return writingQ;
case LOGGED:
return loggedQ;
case SENDING:
return sendingQ;
case SENT:
return sentQ;
case E2EACKED:
case IMPORT:
default:
return null;
}
}
private File
getDir(State state) {
Preconditions.checkNotNull(state);
switch (state) {
case IMPORT:
return importDir;
case WRITING:
return writingDir;
case LOGGED:
return loggedDir;
case SENDING:
return sendingDir;
case SENT:
return sentDir;
case ERROR:
return errorDir;
case E2EACKED:
return doneDir;
default:
return null;
}
}
Preconditions.checkNotNull(tag, "Attempted to get file for empty tag");
WALData data = table.get(tag);
Preconditions.checkNotNull(data, "Data for tag " + tag + " was empty.");
File dir = getDir(data.s);
return new File(dir, tag);
}
synchronized void changeState(String tag, State oldState, State newState)
throws IOException {
WALData data = table.get(tag);
Preconditions.checkArgument(data != null, "Tag " + tag + " has no data");
Preconditions.checkArgument(tag.equals(data.tag),
"Data associated with tag didn't match tag " + tag);
if (LOG.isDebugEnabled()) {
LOG.debug("Change " + data.s + "/" + oldState + " to " + newState + " : "
+ tag);
}
if (oldState == null) {
oldState = data.s;
}
Preconditions.checkState(data.s == oldState, "Expected state to be "
+ oldState + " but was " + data.s);
if (oldState == State.ERROR) {
throw new IllegalStateException("Cannot move from error state");
}
File orig = getFile(tag);
File newf = new File(getDir(newState), tag);
boolean success = orig.renameTo(newf);
if (!success) {
throw new IOException("Move " + orig + " -> " + newf + "failed!");
}
if (newState == State.E2EACKED) {
LOG.debug("Deleting WAL file: " + newf.getAbsoluteFile());
boolean res = newf.delete();
if (!res) {
LOG.warn("Failed to delete complete WAL file: "
+ newf.getAbsoluteFile());
}
}
LOG.debug("old state is " + oldState);
getQueue(oldState).remove(tag);
BlockingQueue<String> q = getQueue(newState);
if (q != null) {
q.add(tag);
}
data.s = newState;
}
final String tag;
final EventSource src;
Preconditions.checkNotNull(src, "StateChangeDeco called with null src");
this.src = src;
this.tag = tag;
}
@Override
public void open()
throws IOException {
try {
src.open();
} catch (IOException ioe) {
changeState(tag, State.SENDING, State.ERROR);
errCount.incrementAndGet();
throw ioe;
}
}
@Override
public void close()
throws IOException {
try {
src.close();
changeState(tag, State.SENDING, State.SENT);
sentCount.incrementAndGet();
} catch (IOException ioe) {
LOG.warn("close had a problem " + src, ioe);
changeState(tag, null, State.ERROR);
throw ioe;
}
}
@Override
public Event
next()
throws IOException {
try {
Event e1 = src.next();
if (e1 == null)
return null;
Event e2 = EventImpl.unselect(e1, RollSink.DEFAULT_ROLL_TAG);
updateEventProcessingStats(e2);
return e2;
} catch (IOException ioe) {
LOG.warn("next had a problem " + src, ioe);
changeState(tag, null, State.ERROR);
errCount.incrementAndGet();
throw ioe;
}
}
@Override
public void getReports(String namePrefix, Map<String, ReportEvent> reports) {
super.getReports(namePrefix, reports);
src.getReports(namePrefix + getName() + ".", reports);
}
@Override
return "NaiveFileWALManager (dir=" + baseDir.getAbsolutePath() + " )";
}
}
String sendingTag = null;
try {
while (sendingTag == null) {
sendingTag = loggedQ.poll(200, TimeUnit.MILLISECONDS);
if (sendingTag == null) {
synchronized (this) {
if (closed && loggedQ.isEmpty() && sendingQ.isEmpty())
return null;
}
}
}
} catch (InterruptedException e) {
LOG.error("interrupted", e);
throw new IOException(e);
}
LOG.info("opening log file " + sendingTag);
changeState(sendingTag, State.LOGGED, State.SENDING);
sendingCount.incrementAndGet();
File curFile = getFile(sendingTag);
EventSource curSource = new SeqfileEventSource(curFile.getAbsolutePath());
return new StateChangeDeco(curSource, sendingTag);
}
synchronized public void toAcked(String tag)
throws IOException {
changeState(tag, State.SENT, State.E2EACKED);
ackedCount.incrementAndGet();
}
synchronized public void retry(String tag)
throws IOException {
WALData data = table.get(tag);
if (data != null) {
if (data.s == State.SENDING || data.s == State.LOGGED) {
LOG.warn("There was a race that happend with SENT vs SENDING states");
return;
}
}
changeState(tag, State.SENT, State.LOGGED);
retryCount.incrementAndGet();
}
@Override
return new WALSource(this);
}
for (String fn : importDir.list()) {
WALData data = WALData.recovered(fn);
synchronized (this) {
table.put(fn, data);
loggedQ.add(fn);
importedCount.incrementAndGet();
}
}
}
@Override
return "naiveWal";
}
@Override
synchronized public ReportEvent
getReport() {
ReportEvent rpt = new ReportEvent(getName());
Attributes.setLong(rpt, A_IMPORTED, importedCount.get());
Attributes.setLong(rpt, A_WRITING, writingCount.get());
Attributes.setLong(rpt, A_LOGGED, loggedCount.get());
Attributes.setLong(rpt, A_SENDING, sendingCount.get());
Attributes.setLong(rpt, A_SENT, sentCount.get());
Attributes.setLong(rpt, A_ACKED, ackedCount.get());
Attributes.setLong(rpt, A_RETRY, retryCount.get());
Attributes.setLong(rpt, A_ERROR, errCount.get());
Attributes.setLong(rpt, A_RECOVERED, recoverCount.get());
Attributes.setLong(rpt, A_IN_LOGGED, loggedQ.size());
Attributes.setLong(rpt, A_IN_SENT, sentQ.size());
return rpt;
}
@Override
synchronized public boolean isEmpty() {
return writingQ.isEmpty() && loggedQ.isEmpty() && sendingQ.isEmpty()
&& sentQ.isEmpty();
}
}