Skip to content

C
cat

梦中观雨 / cat
与 Fork 源项目一致

从无法访问的项目Fork

通知 1
Star 0
Fork 0
C
cat

体验新版 GitCode，发现更多精彩内容 >>

提交 0dfcd472 编写于 作者: F Frankie Wu
浏览文件
操作

remove TransactionReportAnalyzer, checkpoint feature

上级 f4650155
隐藏空白更改
内联 并排
Showing with 238 addition and 434 deletion
cat-consumer/src/main/java/com/dianping/cat/consumer/RealtimeConsumer.java
浏览文件 @ 0dfcd472
package com.dianping.cat.consumer;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
......@@ -29,52 +31,6 @@ import com.site.lookup.annotation.Inject;
* @since Jan 5, 2012
*/
public class RealtimeConsumer extends ContainerHolder implements MessageConsumer, Initializable, LogEnabled {
static class Period {
private long m_startTime;
private long m_endTime;
private List<MessageQueue> m_queues;
public Period(long startTime, long endTime, List<MessageQueue> queues) {
m_startTime = startTime;
m_endTime = endTime;
m_queues = queues;
}
public List<MessageQueue> getQueues() {
return m_queues;
}
public boolean isIn(long timestamp) {
return timestamp >= m_startTime && timestamp < m_endTime;
}
}
static class Task implements Runnable {
private AnalyzerFactory m_factory;
private MessageAnalyzer m_analyzer;
private MessageQueue m_queue;
public Task(AnalyzerFactory factory, MessageAnalyzer analyzer, MessageQueue queue) {
m_factory = factory;
m_analyzer = analyzer;
m_queue = queue;
}
public MessageQueue getQueue() {
return m_queue;
}
public void run() {
m_analyzer.analyze(m_queue);
m_factory.release(m_analyzer);
m_factory.release(m_queue);
}
}
private static final long HOUR = 60 * 60 * 1000L;
private static final long MINUTE = 60 * 1000L;
......@@ -99,7 +55,7 @@ public class RealtimeConsumer extends ContainerHolder implements MessageConsumer
@Inject
private long m_extraTime = FIVE_MINUTES;
@Inject
private int m_threads = 10;
......@@ -158,6 +114,19 @@ public class RealtimeConsumer extends ContainerHolder implements MessageConsumer
}
}
public void doCheckpoint() throws IOException {
SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
m_logger.info("Checkpoint starts at " + format.format(new Date()));
for (Map.Entry<String, MessageAnalyzer> e : m_currentAnalyzers.entrySet()) {
m_logger.info("Checkpoint for " + e.getKey());
e.getValue().doCheckpoint();
}
m_logger.info("Checkpoint ends at " + format.format(new Date()));
}
@Override
public void enableLogging(Logger logger) {
m_logger = logger;
......@@ -254,4 +223,50 @@ public class RealtimeConsumer extends ContainerHolder implements MessageConsumer
m_periods.add(current);
}
static class Period {
private long m_startTime;
private long m_endTime;
private List<MessageQueue> m_queues;
public Period(long startTime, long endTime, List<MessageQueue> queues) {
m_startTime = startTime;
m_endTime = endTime;
m_queues = queues;
}
public List<MessageQueue> getQueues() {
return m_queues;
}
public boolean isIn(long timestamp) {
return timestamp >= m_startTime && timestamp < m_endTime;
}
}
static class Task implements Runnable {
private AnalyzerFactory m_factory;
private MessageAnalyzer m_analyzer;
private MessageQueue m_queue;
public Task(AnalyzerFactory factory, MessageAnalyzer analyzer, MessageQueue queue) {
m_factory = factory;
m_analyzer = analyzer;
m_queue = queue;
}
public MessageQueue getQueue() {
return m_queue;
}
public void run() {
m_analyzer.analyze(m_queue);
m_factory.release(m_analyzer);
m_factory.release(m_queue);
}
}
}
\ No newline at end of file
cat-consumer/src/main/java/com/dianping/cat/consumer/build/ComponentsConfigurator.java
浏览文件 @ 0dfcd472
......@@ -13,7 +13,6 @@ import com.dianping.cat.consumer.failure.FailureReportAnalyzer.Handler;
import com.dianping.cat.consumer.failure.FailureReportAnalyzer.LongUrlHandler;
import com.dianping.cat.consumer.ip.IpAnalyzer;
import com.dianping.cat.consumer.transaction.TransactionAnalyzer;
import com.dianping.cat.consumer.transaction.TransactionReportAnalyzer;
import com.dianping.cat.message.spi.MessageConsumer;
import com.dianping.cat.message.spi.MessageManager;
import com.dianping.cat.message.spi.MessagePathBuilder;
......@@ -52,11 +51,6 @@ public class ComponentsConfigurator extends AbstractResourceConfigurator {
.req(MessageManager.class) //
.req(Handler.class, new String[] { "failure-handler", "long-url-handler" }, "m_handlers"));
all.add(C(TransactionReportAnalyzer.class).is(PER_LOOKUP) //
.req(MessageManager.class) //
.req(MessageStorage.class, "html") //
.config(E("reportPath").value("target/report/transaction/")));
all.add(C(TransactionAnalyzer.class).is(PER_LOOKUP) //
.req(BucketManager.class, MessagePathBuilder.class));
......
cat-consumer/src/main/java/com/dianping/cat/consumer/transaction/TransactionAnalyzer.java
浏览文件 @ 0dfcd472
package com.dianping.cat.consumer.transaction;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.codehaus.plexus.logging.LogEnabled;
import org.codehaus.plexus.logging.Logger;
import org.codehaus.plexus.personality.plexus.lifecycle.phase.Initializable;
import org.codehaus.plexus.personality.plexus.lifecycle.phase.InitializationException;
import com.dianping.cat.consumer.transaction.model.entity.Duration;
import com.dianping.cat.consumer.transaction.model.entity.Range;
import com.dianping.cat.consumer.transaction.model.entity.TransactionName;
import com.dianping.cat.consumer.transaction.model.entity.TransactionReport;
import com.dianping.cat.consumer.transaction.model.entity.TransactionType;
import com.dianping.cat.consumer.transaction.model.transform.DefaultXmlBuilder;
import com.dianping.cat.consumer.transaction.model.transform.DefaultXmlParser;
import com.dianping.cat.message.Message;
import com.dianping.cat.message.Transaction;
import com.dianping.cat.message.spi.AbstractMessageAnalyzer;
......@@ -26,14 +29,14 @@ import com.dianping.cat.message.spi.MessagePathBuilder;
import com.dianping.cat.message.spi.MessageTree;
import com.dianping.cat.storage.Bucket;
import com.dianping.cat.storage.BucketManager;
import com.dianping.cat.storage.internal.AbstractFileBucket;
import com.site.lookup.annotation.Inject;
/**
* @author sean.wang
* @since Jan 5, 2012
*/
public class TransactionAnalyzer extends AbstractMessageAnalyzer<TransactionReport> implements Initializable,
LogEnabled {
public class TransactionAnalyzer extends AbstractMessageAnalyzer<TransactionReport> implements LogEnabled {
private static final long MINUTE = 60 * 1000;
@Inject
......@@ -54,6 +57,11 @@ public class TransactionAnalyzer extends AbstractMessageAnalyzer<TransactionRepo
private long m_duration;
@Override
public void doCheckpoint() throws IOException {
storeReports(m_reports.values());
}
@Override
public void enableLogging(Logger logger) {
m_logger = logger;
......@@ -74,7 +82,7 @@ public class TransactionAnalyzer extends AbstractMessageAnalyzer<TransactionRepo
return reports;
}
TransactionReport generate(String domain) {
public TransactionReport generate(String domain) {
if (domain == null) {
List<String> domains = getDomains();
......@@ -111,17 +119,6 @@ public class TransactionAnalyzer extends AbstractMessageAnalyzer<TransactionRepo
return m_reports;
}
@Override
public void initialize() throws InitializationException {
String path = m_pathBuilder.getMessagePath(new Date(m_startTime));
try {
m_messageBucket = m_bucketManager.getMessageBucket(path);
} catch (Exception e) {
throw new InitializationException(String.format("Unable to create message bucket at %s.", path), e);
}
}
@Override
protected boolean isTimeout() {
long currentTime = System.currentTimeMillis();
......@@ -130,6 +127,33 @@ public class TransactionAnalyzer extends AbstractMessageAnalyzer<TransactionRepo
return currentTime > endTime;
}
void loadReports() {
String path = m_pathBuilder.getReportPath(new Date(m_startTime));
Bucket<String> bucket = null;
try {
bucket = m_bucketManager.getStringBucket(path);
if (bucket instanceof AbstractFileBucket) {
DefaultXmlParser parser = new DefaultXmlParser();
Set<String> ids = ((AbstractFileBucket<?>) bucket).getIds();
for (String id : ids) {
String xml = bucket.findById(id);
TransactionReport report = parser.parse(xml);
m_reports.put(report.getDomain(), report);
}
}
} catch (Exception e) {
m_logger.error(String.format("Error when loading transaction reports from %s!", path), e);
} finally {
if (bucket != null) {
m_bucketManager.closeBucket(bucket);
}
}
}
@Override
protected void process(MessageTree tree) {
String domain = tree.getDomain();
......@@ -249,6 +273,16 @@ public class TransactionAnalyzer extends AbstractMessageAnalyzer<TransactionRepo
m_extraTime = extraTime;
m_startTime = startTime;
m_duration = duration;
String path = m_pathBuilder.getMessagePath(new Date(m_startTime));
try {
m_messageBucket = m_bucketManager.getMessageBucket(path);
} catch (Exception e) {
throw new RuntimeException(String.format("Unable to create message bucket at %s.", path), e);
}
loadReports();
}
@Override
......@@ -258,37 +292,22 @@ public class TransactionAnalyzer extends AbstractMessageAnalyzer<TransactionRepo
}
m_bucketManager.closeBucket(m_messageBucket);
storeReports(reports);
storeLogviews();
}
void storeLogviews() {
String path = m_pathBuilder.getMessagePath(new Date(m_startTime));
Bucket<byte[]> bucket = null;
try {
bucket = m_bucketManager.getBytesBucket(path);
m_pathBuilder.getLogViewBaseDir();
} catch (Exception e) {
m_logger.error(String.format("Error when storing transaction reports to %s!", path), e);
} finally {
if (bucket != null) {
m_bucketManager.closeBucket(bucket);
}
}
}
void storeReports(List<TransactionReport> reports) {
void storeReports(Collection<TransactionReport> reports) {
String path = m_pathBuilder.getReportPath(new Date(m_startTime));
Bucket<String> bucket = null;
DefaultXmlBuilder builder = new DefaultXmlBuilder(true);
try {
bucket = m_bucketManager.getStringBucket(path);
// delete old one, not append mode
bucket.deleteAndCreate();
for (TransactionReport report : reports) {
bucket.storeById("transaction-" + report.getDomain(), report.toString());
bucket.storeById("transaction-" + report.getDomain(), builder.buildXml(report));
}
} catch (Exception e) {
m_logger.error(String.format("Error when storing transaction reports to %s!", path), e);
......
cat-consumer/src/main/java/com/dianping/cat/consumer/transaction/TransactionReportAnalyzer.java 已删除 100644 → 0
浏览文件 @ f4650155
package com.dianping.cat.consumer.transaction;
import java.io.File;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.codehaus.plexus.logging.LogEnabled;
import org.codehaus.plexus.logging.Logger;
import org.codehaus.plexus.personality.plexus.lifecycle.phase.Initializable;
import org.codehaus.plexus.personality.plexus.lifecycle.phase.InitializationException;
import com.dianping.cat.configuration.model.entity.Config;
import com.dianping.cat.configuration.model.entity.Property;
import com.dianping.cat.consumer.transaction.model.entity.TransactionName;
import com.dianping.cat.consumer.transaction.model.entity.TransactionReport;
import com.dianping.cat.consumer.transaction.model.entity.TransactionType;
import com.dianping.cat.consumer.transaction.model.transform.DefaultJsonBuilder;
import com.dianping.cat.message.Message;
import com.dianping.cat.message.Transaction;
import com.dianping.cat.message.spi.AbstractMessageAnalyzer;
import com.dianping.cat.message.spi.MessageManager;
import com.dianping.cat.message.spi.MessageStorage;
import com.dianping.cat.message.spi.MessageTree;
import com.site.helper.Files;
import com.site.lookup.annotation.Inject;
/**
* @author sean.wang
* @since Jan 5, 2012
*/
public class TransactionReportAnalyzer extends AbstractMessageAnalyzer<TransactionReport> implements Initializable,
LogEnabled {
private final static SimpleDateFormat FILE_SDF = new SimpleDateFormat("yyyyMMddHHmm");
private static final long MINUTE = 60 * 1000;
@Inject
private MessageManager messageManager;
@Inject
private MessageStorage messageStorage;
private Map<String, TransactionReport> m_reports = new HashMap<String, TransactionReport>();
private long m_extraTime;
private String m_reportPath;
private Logger m_logger;
private long m_startTime;
private long m_duration;
private TransactionReport computeMeanSquareDeviation(String domain) {
TransactionReport report = m_reports.get(domain);
if (report == null) {
return report;
}
for (TransactionType type : report.getTypes().values()) {
doOneType(type);
}
return report;
}
/**
* @param type
*/
public void doOneType(TransactionType type) {
long typeCount = 0;
long typeFailCount = 0;
double typeSum = 0;
double typeSum2 = 0;
Collection<TransactionName> names = type.getNames().values();
for (TransactionName name : names) {
long count = name.getTotalCount();
double sum = name.getSum();
double ave = sum / count;
double sum2 = name.getSum2();
double std = std(count, ave, sum2);
long failCount = name.getFailCount();
double failPercent = 100.0 * failCount / count;
name.setFailPercent(failPercent);
name.setAvg(ave);
name.setStd(std);
typeCount += count;
typeSum += sum;
typeSum2 += sum2;
typeFailCount += failCount;
if (type.getSuccessMessageUrl() == null && name.getSuccessMessageUrl() != null) {
type.setSuccessMessageUrl(name.getSuccessMessageUrl());
}
if (type.getFailMessageUrl() == null && name.getFailMessageUrl() != null) {
type.setFailMessageUrl(name.getFailMessageUrl());
}
type.setMax(Math.max(name.getMax(), type.getMax()));
type.setMin(Math.min(name.getMin(), type.getMin()));
}
type.setTotalCount(typeCount);
type.setFailCount(typeFailCount);
type.setSum(typeSum);
type.setSum2(typeSum2);
double typeAvg = typeSum / typeCount;
type.setAvg(typeAvg);
type.setFailPercent(100.0 * typeFailCount / typeCount);
type.setStd(std(typeCount, typeAvg, typeSum2));
}
@Override
public void enableLogging(Logger logger) {
m_logger = logger;
}
@Override
public List<TransactionReport> generate() {
List<TransactionReport> reports = new ArrayList<TransactionReport>();
for (String domain : m_reports.keySet()) {
reports.add(generate(domain));
}
return reports;
}
public TransactionReport generate(String domain) {
ArrayList<String> domains = new ArrayList<String>(this.m_reports.keySet());
if (domain == null) {
domain = domains.get(0);
}
TransactionReport report = computeMeanSquareDeviation(domain);
for (String ele : domains) {
report.addDomain(ele);
}
return report;
}
private String getTransactionFileName(TransactionReport report) {
StringBuffer result = new StringBuffer();
String start = FILE_SDF.format(report.getStartTime());
String end = FILE_SDF.format(report.getEndTime());
result.append(report.getDomain()).append("-").append(start).append("-").append(end);
return result.toString();
}
@Override
public void initialize() throws InitializationException {
Config config = messageManager.getClientConfig();
if (config != null) {
Property property = config.findProperty("transaction-base-dir");
if (property != null) {
m_reportPath = property.getValue();
}
}
}
@Override
protected boolean isTimeout() {
long endTime = m_startTime + m_duration + m_extraTime;
long currentTime = System.currentTimeMillis();
if (currentTime > endTime + m_extraTime) {
return true;
}
return false;
}
private void process(TransactionReport report, MessageTree tree, Message message) {
if (message instanceof Transaction) {
Transaction t = (Transaction) message;
String tType = t.getType();
String tName = t.getName();
TransactionType type = report.getTypes().get(tType);
if (type == null) {
type = new TransactionType(tType);
report.addType(type);
}
TransactionName name = type.getNames().get(tName);
if (name == null) {
name = new TransactionName(tName);
type.addName(name);
}
name.setTotalCount(name.getTotalCount() + 1);
if (!t.isSuccess()) {
name.setFailCount(name.getFailCount() + 1);
}
if (tree != null) {
if (t.isSuccess()) {
if (name.getSuccessMessageUrl() == null) {
String url = this.messageStorage.store(tree); // store first
// success
name.setSuccessMessageUrl(url);
}
} else {
String url = this.messageStorage.store(tree); // store all errors
if (name.getFailMessageUrl() == null) {
name.setFailMessageUrl(url);
}
}
}
long duration = t.getDuration();
name.setMax(Math.max(name.getMax(), duration));
name.setMin(Math.min(name.getMin(), duration));
name.setSum(name.getSum() + duration);
name.setSum2(name.getSum2() + duration * duration);
if (!t.hasChildren()) {
return;
}
List<Message> children = t.getChildren();
for (Message child : children) {
process(report, tree, child);
}
}
}
@Override
protected void process(MessageTree tree) {
String domain = tree.getDomain();
TransactionReport report = this.m_reports.get(domain);
if (report == null) {
report = new TransactionReport(domain);
report.setStartTime(new Date(m_startTime));
report.setEndTime(new Date(m_startTime + MINUTE * 59));
this.m_reports.put(domain, report);
}
Message message = tree.getMessage();
process(report, tree, message);
}
public void setAnalyzerInfo(long startTime, long duration, String domain, long extraTime) {
m_extraTime = extraTime;
m_startTime = startTime;
m_duration = duration;
}
public void setReportPath(String configPath) {
m_reportPath = configPath;
}
/**
* @param count
* @param ave
* @param sum2
* @return
*/
public double std(long count, double ave, double sum2) {
return Math.sqrt(sum2 / count - 2 * ave * ave + ave * ave);
}
@Override
protected void store(List<TransactionReport> reports) {
if (reports == null || reports.size() == 0) {
return;
}
for (TransactionReport report : reports) {
String failureFileName = getTransactionFileName(report);
String htmlPath = new StringBuilder().append(m_reportPath).append(failureFileName).append(".html").toString();
File file = new File(htmlPath);
file.getParentFile().mkdirs();
try {
Files.forIO().writeTo(file, new DefaultJsonBuilder().buildJson(report));
} catch (IOException e) {
m_logger.error(String.format("Error when writing to file(%s)!", file), e);
}
}
}
public Map<String, TransactionReport> getReports() {
return m_reports;
}
}
cat-consumer/src/main/resources/META-INF/plexus/components.xml
浏览文件 @ 0dfcd472
......@@ -73,23 +73,6 @@
</requirement>
</requirements>
</component>
<component>
<role>com.dianping.cat.consumer.transaction.TransactionReportAnalyzer</role>
<implementation>com.dianping.cat.consumer.transaction.TransactionReportAnalyzer</implementation>
<instantiation-strategy>per-lookup</instantiation-strategy>
<configuration>
<reportPath>target/report/transaction/</reportPath>
</configuration>
<requirements>
<requirement>
<role>com.dianping.cat.message.spi.MessageManager</role>
</requirement>
<requirement>
<role>com.dianping.cat.message.spi.MessageStorage</role>
<role-hint>html</role-hint>
</requirement>
</requirements>
</component>
<component>
<role>com.dianping.cat.consumer.transaction.TransactionAnalyzer</role>
<implementation>com.dianping.cat.consumer.transaction.TransactionAnalyzer</implementation>
......
cat-consumer/src/test/java/com/dianping/cat/consumer/transaction/TransactionReportMessageAnalyzerTest.java
浏览文件 @ 0dfcd472
......@@ -38,7 +38,7 @@ public class TransactionReportMessageAnalyzerTest extends ComponentTestCase{
long start = current - current % (60 * 60 * 1000) -1000L*60*60;
AnalyzerFactory factory = lookup(AnalyzerFactory.class);
TransactionReportAnalyzer analyzer = (TransactionReportAnalyzer) factory.create("transaction", start, duration, "testDomain",
TransactionAnalyzer analyzer = (TransactionAnalyzer) factory.create("transaction", start, duration, "testDomain",
extraTime);
for (int i = 1; i <= 1000; i++) {
......
cat-core/src/main/java/com/dianping/cat/message/internal/DefaultMessageManager.java
浏览文件 @ 0dfcd472
......@@ -103,7 +103,13 @@ public class DefaultMessageManager extends ContainerHolder implements MessageMan
@Override
public MessageTree getThreadLocalMessageTree() {
return getContext().m_tree;
Context ctx = m_context.get();
if (ctx != null) {
return ctx.m_tree;
} else {
return null;
}
}
@Override
......
cat-core/src/main/java/com/dianping/cat/message/spi/AbstractMessageAnalyzer.java
浏览文件 @ 0dfcd472
package com.dianping.cat.message.spi;
import java.io.IOException;
import java.util.List;
public abstract class AbstractMessageAnalyzer<R> implements MessageAnalyzer {
......@@ -28,11 +29,15 @@ public abstract class AbstractMessageAnalyzer<R> implements MessageAnalyzer {
store(result);
}
protected abstract void store(List<R> result);
public void doCheckpoint() throws IOException {
// override it
}
protected abstract List<R> generate();
protected abstract boolean isTimeout();
protected abstract void process(MessageTree tree);
protected abstract boolean isTimeout();
protected abstract void store(List<R> result);
}
cat-core/src/main/java/com/dianping/cat/message/spi/MessageAnalyzer.java
浏览文件 @ 0dfcd472
package com.dianping.cat.message.spi;
import java.io.IOException;
public interface MessageAnalyzer {
public void analyze(MessageQueue queue);
public void doCheckpoint() throws IOException;
}
cat-core/src/main/java/com/dianping/cat/message/spi/MessageManager.java
浏览文件 @ 0dfcd472
......@@ -47,7 +47,7 @@ public interface MessageManager {
/**
* Get thread local message information.
*
* @return message tree
* @return message tree, null means current thread is not setup correctly.
*/
public MessageTree getThreadLocalMessageTree();
......
cat-core/src/main/java/com/dianping/cat/message/spi/internal/DefaultMessagePathBuilder.java
浏览文件 @ 0dfcd472
......@@ -93,7 +93,7 @@ public class DefaultMessagePathBuilder implements MessagePathBuilder, Initializa
@Override
public String getMessagePath(Date timestamp) {
MessageFormat format = new MessageFormat("{0,date,yyyyMMdd}/{0,date,HH}/logview");
MessageFormat format = new MessageFormat("{0,date,yyyyMMdd}/{0,date,HH}/message");
return format.format(new Object[] { timestamp });
}
......
cat-core/src/main/java/com/dianping/cat/storage/Bucket.java
浏览文件 @ 0dfcd472
......@@ -7,6 +7,8 @@ import java.util.List;
public interface Bucket<T> extends TagThreadSupport<T> {
public void close();
public void deleteAndCreate();
public List<T> findAllByIds(List<String> ids);
public T findById(String id);
......
cat-core/src/main/java/com/dianping/cat/storage/internal/AbstractFileBucket.java
浏览文件 @ 0dfcd472
package com.dianping.cat.storage.internal;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.util.ArrayList;
......@@ -8,6 +9,7 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.locks.ReentrantLock;
import org.codehaus.plexus.logging.LogEnabled;
......@@ -58,6 +60,26 @@ public abstract class AbstractFileBucket<T> implements Bucket<T>, TagThreadSuppo
protected abstract T decode(ChannelBuffer buf) throws IOException;
@Override
public void deleteAndCreate() {
m_writeLock.lock();
m_readLock.lock();
m_idToOffsets.clear();
m_tagToIds.clear();
try {
m_file.delete();
m_writeFile = new RandomAccessFile(m_file, "rw");
m_readFile = new RandomAccessFile(m_file, "r");
} catch (FileNotFoundException e) {
m_logger.error(String.format("Error when clearing file bucket(%s)!", m_file), e);
} finally {
m_readLock.unlock();
m_writeLock.unlock();
}
}
@Override
public void enableLogging(Logger logger) {
m_logger = logger;
......@@ -147,6 +169,10 @@ public abstract class AbstractFileBucket<T> implements Bucket<T>, TagThreadSuppo
return null;
}
public Set<String> getIds() {
return m_idToOffsets.keySet();
}
@Override
public void initialize(Class<?> type, File path) throws IOException {
m_writeLock = new ReentrantLock();
......@@ -262,6 +288,11 @@ public abstract class AbstractFileBucket<T> implements Bucket<T>, TagThreadSuppo
}
}
@Override
public String toString() {
return String.format("%s[file=%s, ids=%s]", getClass().getSimpleName(), m_file, m_idToOffsets.keySet());
}
protected void updateIndex(String id, String[] tags, long offset) {
m_idToOffsets.put(id, offset);
......
cat-home/src/main/java/com/dianping/cat/report/page/home/Action.java
浏览文件 @ 0dfcd472
package com.dianping.cat.report.page.home;
public enum Action implements com.site.web.mvc.Action {
VIEW("view");
VIEW("view"),
CHECKPOINT("checkpoint");
private String m_name;
......
cat-home/src/main/java/com/dianping/cat/report/page/home/Handler.java
浏览文件 @ 0dfcd472
......@@ -4,6 +4,8 @@ import java.io.IOException;
import javax.servlet.ServletException;
import com.dianping.cat.consumer.RealtimeConsumer;
import com.dianping.cat.message.spi.MessageConsumer;
import com.dianping.cat.report.ReportPage;
import com.site.lookup.annotation.Inject;
import com.site.web.mvc.PageHandler;
......@@ -15,11 +17,18 @@ public class Handler implements PageHandler<Context> {
@Inject
private JspViewer m_jspViewer;
@Inject(type = MessageConsumer.class, value = "realtime")
private RealtimeConsumer m_realtimeConsumer;
@Override
@PayloadMeta(Payload.class)
@InboundActionMeta(name = "home")
public void handleInbound(Context ctx) throws ServletException, IOException {
// display only, no action here
Payload payload = ctx.getPayload();
if (payload.getAction() == Action.CHECKPOINT) {
m_realtimeConsumer.doCheckpoint();
}
}
@Override
......@@ -29,6 +38,7 @@ public class Handler implements PageHandler<Context> {
model.setAction(Action.VIEW);
model.setPage(ReportPage.HOME);
m_jspViewer.view(ctx, model);
}
}
cat-home/src/main/java/com/dianping/cat/report/page/home/Payload.java
浏览文件 @ 0dfcd472
......@@ -11,8 +11,8 @@ public class Payload implements ActionPayload<ReportPage, Action> {
@FieldMeta("op")
private Action m_action;
public void setAction(Action action) {
m_action = action;
public void setAction(String action) {
m_action = Action.getByName(action, Action.VIEW);
}
@Override
......@@ -32,5 +32,8 @@ public class Payload implements ActionPayload<ReportPage, Action> {
@Override
public void validate(ActionContext<?> ctx) {
if (m_action == null) {
m_action = Action.VIEW;
}
}
}
cat-home/src/main/java/com/dianping/cat/report/page/service/provider/TransactionModelProvider.java
浏览文件 @ 0dfcd472
......@@ -7,7 +7,7 @@ import java.util.List;
import java.util.Map;
import com.dianping.cat.consumer.RealtimeConsumer;
import com.dianping.cat.consumer.transaction.TransactionReportAnalyzer;
import com.dianping.cat.consumer.transaction.TransactionAnalyzer;
import com.dianping.cat.consumer.transaction.model.entity.TransactionReport;
import com.dianping.cat.consumer.transaction.model.transform.DefaultXmlBuilder;
import com.dianping.cat.message.spi.MessageConsumer;
......@@ -21,7 +21,7 @@ public class TransactionModelProvider implements ModelProvider {
@Override
public List<String> getDomains() {
TransactionReportAnalyzer analyzer = (TransactionReportAnalyzer) m_consumer.getCurrentAnalyzer("transaction");
TransactionAnalyzer analyzer = (TransactionAnalyzer) m_consumer.getCurrentAnalyzer("transaction");
List<String> domains = new ArrayList<String>(analyzer.getReports().keySet());
Collections.sort(domains, new Comparator<String>() {
@Override
......@@ -39,15 +39,15 @@ public class TransactionModelProvider implements ModelProvider {
public String getModel(Map<String, String> parameters) {
String domain = parameters.get("domain");
String index = parameters.get("index");
TransactionReportAnalyzer analyzer = null;
TransactionAnalyzer analyzer = null;
if (index == null) {
index = Constants.MEMORY_CURRENT;
}
if (index.equals(Constants.MEMORY_CURRENT)) {
analyzer = (TransactionReportAnalyzer) m_consumer.getCurrentAnalyzer("transaction");
analyzer = (TransactionAnalyzer) m_consumer.getCurrentAnalyzer("transaction");
} else if (index.equals(Constants.MEMORY_LAST)) {
analyzer = (TransactionReportAnalyzer) m_consumer.getLastAnalyzer("transaction");
analyzer = (TransactionAnalyzer) m_consumer.getLastAnalyzer("transaction");
} else {
System.err.println("historical model is not implemented yet");
}
......@@ -63,7 +63,7 @@ public class TransactionModelProvider implements ModelProvider {
@Override
public String getDefaultDomain() {
TransactionReportAnalyzer analyzer = (TransactionReportAnalyzer) m_consumer.getCurrentAnalyzer("transaction");
TransactionAnalyzer analyzer = (TransactionAnalyzer) m_consumer.getCurrentAnalyzer("transaction");
List<String> domains = new ArrayList<String>(analyzer.getReports().keySet());
Collections.sort(domains);
if (domains != null && domains.size() > 0) {
......
cat-home/src/main/java/com/dianping/cat/report/page/transaction/Handler.java
浏览文件 @ 0dfcd472
......@@ -9,6 +9,7 @@ import javax.servlet.ServletException;
import org.codehaus.plexus.personality.plexus.lifecycle.phase.Initializable;
import org.codehaus.plexus.personality.plexus.lifecycle.phase.InitializationException;
import com.dianping.cat.Cat;
import com.dianping.cat.consumer.transaction.MeanSquareDeviationComputer;
import com.dianping.cat.consumer.transaction.model.entity.Duration;
import com.dianping.cat.consumer.transaction.model.entity.Range;
......@@ -68,6 +69,8 @@ public class Handler implements PageHandler<Context>, Initializable {
return n;
}
Cat.getManager().getThreadLocalMessageTree();
return null;
}
......
cat-home/src/main/resources/META-INF/plexus/components.xml
浏览文件 @ 0dfcd472
......@@ -164,12 +164,47 @@
<requirement>
<role>com.dianping.cat.report.page.home.JspViewer</role>
</requirement>
<requirement>
<role>com.dianping.cat.message.spi.MessageConsumer</role>
<role-hint>realtime</role-hint>
<field-name>m_realtimeConsumer</field-name>
</requirement>
</requirements>
</component>
<component>
<role>com.dianping.cat.report.page.home.JspViewer</role>
<implementation>com.dianping.cat.report.page.home.JspViewer</implementation>
</component>
<component>
<role>com.dianping.cat.consumer.RealtimeConsumer</role>
<implementation>com.dianping.cat.consumer.RealtimeConsumer</implementation>
<requirements>
<requirement>
<role>org.codehaus.plexus.logging.Logger</role>
</requirement>
<requirement>
<role>java.lang.String</role>
</requirement>
<requirement>
<role>java.lang.String</role>
</requirement>
<requirement>
<role>long</role>
</requirement>
<requirement>
<role>long</role>
</requirement>
<requirement>
<role>int</role>
</requirement>
<requirement>
<role>java.util.List</role>
</requirement>
<requirement>
<role>com.dianping.cat.consumer.AnalyzerFactory</role>
</requirement>
</requirements>
</component>
<component>
<role>com.dianping.cat.report.page.transaction.Handler</role>
<implementation>com.dianping.cat.report.page.transaction.Handler</implementation>
......@@ -322,36 +357,6 @@
</requirement>
</requirements>
</component>
<component>
<role>com.dianping.cat.consumer.RealtimeConsumer</role>
<implementation>com.dianping.cat.consumer.RealtimeConsumer</implementation>
<requirements>
<requirement>
<role>org.codehaus.plexus.logging.Logger</role>
</requirement>
<requirement>
<role>java.lang.String</role>
</requirement>
<requirement>
<role>java.lang.String</role>
</requirement>
<requirement>
<role>long</role>
</requirement>
<requirement>
<role>long</role>
</requirement>
<requirement>
<role>int</role>
</requirement>
<requirement>
<role>java.util.List</role>
</requirement>
<requirement>
<role>com.dianping.cat.consumer.AnalyzerFactory</role>
</requirement>
</requirements>
</component>
<component>
<role>com.dianping.cat.report.page.problem.Handler</role>
<implementation>com.dianping.cat.report.page.problem.Handler</implementation>
......
cat-home/src/main/webapp/jsp/report/home.jsp
浏览文件 @ 0dfcd472
......@@ -9,4 +9,6 @@
Welcome to <b>Central Application Tracking (CAT)</b>.
<br><br><a href="?op=checkpoint">Do checkpoint here</a>
</a:body>
\ No newline at end of file
cat-home/src/main/webapp/jsp/report/ip.jsp
浏览文件 @ 0dfcd472
......@@ -11,7 +11,7 @@
<a:report title="Hot IP Report" timestamp="2012-02-07">
<jsp:body>
${model.b}
<table class="ip-table">
<tr><th>IP</th><th>last 1 min</th><th>last 5 mins</th><th>last 15 mins</th></tr>
<c:forEach var="m" items="${model.displayModels}" varStatus="status">
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或