package com.cloudera.flume.reporter;
import java.io.IOException;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.cloudera.flume.agent.FlumeNode;
import com.cloudera.flume.agent.LogicalNode;
import com.cloudera.flume.agent.MasterRPC;
import com.cloudera.flume.conf.FlumeConfiguration;
import com.cloudera.util.Clock;
static final Logger LOG = LoggerFactory.getLogger(MasterReportPusher.class);
final FlumeConfiguration cfg;
final ReportManager rptMan;
final MasterRPC masterRPC;
final PusherThread pusherThread = new PusherThread();
volatile boolean shutdown = false;
MasterRPC rpcMan) {
this.cfg = cfg;
this.rptMan = rptMan;
this.masterRPC = rpcMan;
}
shutdown = true;
}
pusherThread.start();
}
doShutdown();
}
Map<String, Reportable> reportables = rptMan.getReportables();
for (Entry<String, Reportable> e : reportables.entrySet()) {
reports.put(e.getKey(), e.getValue().getReport());
}
}
Collection<LogicalNode> lnodes =
FlumeNode.getInstance().getLogicalNodeManager().getNodes();
for (LogicalNode n : lnodes) {
n.getReports(reports);
}
}
try {
while (!shutdown) {
Clock.sleep(cfg.getReporterPollPeriod());
Map<String, ReportEvent> reports = new HashMap<String, ReportEvent>();
queryReportMan(reports);
querySrcSinkReports(reports);
masterRPC.putReports(reports);
}
} catch (InterruptedException e) {
LOG.warn("MasterReportPusher.PusherThread was interrupted", e);
} catch (IOException e) {
LOG.error("IOException in MasterReportPusher.PusherThread", e);
}
}
}
}