Merge branch 'master' of ssh://192.168.8.22:58422/cat

cat-consumer/src/main/java/com/dianping/cat/consumer/transaction/TransactionAnalyzer.java

@@ -179,7 +179,11 @@ public class TransactionAnalyzer extends AbstractMessageAnalyzer<TransactionRepo
 				String sessionTag = "s:" + tree.getSessionToken();
 				String requestTag = "r:" + messageId;
 
+				try {
 					m_messageBucket.storeById(messageId, tree, threadTag, sessionTag, requestTag);
+				} catch (IOException e) {
+					m_logger.error("", e);
+				}
 			}
 		}
 	}

cat-core/src/main/java/com/dianping/cat/message/spi/MessageStorage.java

@@ -13,7 +13,7 @@ public interface MessageStorage {
 	 *            message tree to store
 	 * @return relative path to base directory or base URL
 	 */
-	public String store(MessageTree tree);
+	public String store(MessageTree tree, String... tag);
 
 	/**
 	 * Fetch a message tree from the store.
@@ -23,6 +23,18 @@ public interface MessageStorage {
 	 */
 	public MessageTree get(String messageId);
 
+	/**
+	 * @param messageId
+	 * @return
+	 */
+	public MessageTree next(String messageId, String tag);
+
+	/**
+	 * @param messageId
+	 * @return
+	 */
+	public MessageTree previous(String messageId, String tag);
+
 	/**
 	 * 
 	 * Get relative path to base directory or base URL.

cat-core/src/main/java/com/dianping/cat/message/spi/internal/DefaultMessageStorage.java

@@ -58,7 +58,7 @@ public class DefaultMessageStorage implements MessageStorage, Initializable, Dis
 	}
 
 	@Override
-	public String store(MessageTree tree) {
+	public String store(MessageTree tree, String... tags) {
 		String path = m_builder.getLogViewPath(tree.getMessageId());
 
 		m_job.append(tree);
@@ -145,4 +145,16 @@ public class DefaultMessageStorage implements MessageStorage, Initializable, Dis
 		// TODO Auto-generated method stub
 		return null;
 	}
+
+	@Override
+	public MessageTree next(String messageId, String tag) {
+		// TODO Auto-generated method stub
+		return null;
+	}
+
+	@Override
+	public MessageTree previous(String messageId, String tag) {
+		// TODO Auto-generated method stub
+		return null;
+	}
 }

cat-core/src/main/java/com/dianping/cat/storage/Bucket.java

@@ -5,16 +5,16 @@ import java.io.IOException;
 import java.util.List;
 
 public interface Bucket<T> extends TagThreadSupport<T> {
-	public void close();
+	public void close() throws IOException;
 
-	public void deleteAndCreate();
+	public void deleteAndCreate() throws IOException;
 
-	public List<T> findAllByIds(List<String> ids);
+	public List<T> findAllByIds(List<String> ids) throws IOException;;
 
-	public T findById(String id);
+	public T findById(String id) throws IOException;;
 
-	public void initialize(Class<?> type, File path) throws IOException;
+	public void initialize(Class<?> type, File baseDir, String logicalPath) throws IOException;
 
-	public boolean storeById(String id, T data);
+	public boolean storeById(String id, T data) throws IOException;;
 
 }

cat-core/src/main/java/com/dianping/cat/storage/TagThreadSupport.java

 package com.dianping.cat.storage;
 
+import java.io.IOException;
 import java.util.List;
 
 /**
@@ -9,11 +10,11 @@ import java.util.List;
  * Sample tags: "thread:101", "session:abc", "request:xyz"
  */
 public interface TagThreadSupport<T> {
-	public boolean storeById(String id, T data, String... tags);
+	public boolean storeById(String id, T data, String... tags) throws IOException;;
 
-	public List<String> findAllIdsByTag(String tag);
+	public List<String> findAllIdsByTag(String tag) throws IOException;;
 
-	public T findNextById(String id, Direction direction, String tag);
+	public T findNextById(String id, Direction direction, String tag) throws IOException;;
 
 	public static enum Direction {
 		FORWARD,

cat-core/src/main/java/com/dianping/cat/storage/internal/AbstractFileBucket.java

@@ -174,7 +174,8 @@ public abstract class AbstractFileBucket<T> implements Bucket<T>, TagThreadSuppo
 	}
 
 	@Override
-	public void initialize(Class<?> type, File path) throws IOException {
+	public void initialize(Class<?> type, File baseDir, String logicalPath) throws IOException {
+		File path = new File(baseDir, logicalPath);
 		m_writeLock = new ReentrantLock();
 		m_readLock = new ReentrantLock();
 		m_file = path;

cat-core/src/main/java/com/dianping/cat/storage/internal/DefaultBucket.java

@@ -21,8 +21,7 @@ public class DefaultBucket<T> extends AbstractFileBucket<T> {
 
 			return (T) bytes;
 		} else {
-			throw new UnsupportedOperationException(String.format(
-			      "Only String or byte[] are supported so far, but was %s.", m_type));
+			throw new UnsupportedOperationException(String.format("Only String or byte[] are supported so far, but was %s.", m_type));
 		}
 	}
 
@@ -40,20 +39,18 @@ public class DefaultBucket<T> extends AbstractFileBucket<T> {
 			buf.writeInt(bytes.length);
 			buf.writeBytes(bytes);
 		} else {
-			throw new UnsupportedOperationException(String.format(
-			      "Only String or byte[] are supported so far, but was %s.", m_type));
+			throw new UnsupportedOperationException(String.format("Only String or byte[] are supported so far, but was %s.", m_type));
 		}
 	}
 
 	@Override
-	public void initialize(Class<?> type, File path) throws IOException {
-		super.initialize(type, path);
+	public void initialize(Class<?> type, File baseDir, String logicalPath) throws IOException {
+		super.initialize(type, baseDir, logicalPath);
 
 		m_type = type;
 
 		if (m_type != String.class && m_type != byte[].class) {
-			throw new UnsupportedOperationException(String.format(
-			      "Only String or byte[] are supported so far, but was %s.", m_type));
+			throw new UnsupportedOperationException(String.format("Only String or byte[] are supported so far, but was %s.", m_type));
 		}
 	}
 

cat-core/src/main/java/com/dianping/cat/storage/internal/DefaultBucketManager.java

@@ -22,7 +22,7 @@ public class DefaultBucketManager extends ContainerHolder implements BucketManag
 	protected Bucket<?> createBucket(String path, Class<?> type) throws IOException {
 		Bucket<?> bucket = lookup(Bucket.class, type.getName());
 
-		bucket.initialize(type, new File(m_baseDir, path));
+		bucket.initialize(type, new File(m_baseDir), path);
 		return bucket;
 	}
 
@@ -115,7 +115,11 @@ public class DefaultBucketManager extends ContainerHolder implements BucketManag
 
 	@Override
 	public void closeBucket(Bucket<?> bucket) {
+		try {
 			bucket.close();
+		} catch (IOException e) {
+			e.printStackTrace();
+		}
 		release(bucket);
 	}
 }

cat-core/src/test/java/com/dianping/cat/storage/BucketTest.java

@@ -53,7 +53,7 @@ public class BucketTest extends ComponentTestCase {
 
 		// close and reload it, check if everything is okay
 		bucket.close();
-		bucket.initialize(byte[].class, new File("target/bucket/bytes"));
+		bucket.initialize(byte[].class, new File("target/bucket/"), "bytes");
 
 		// store it and load it
 		for (int i = 0; i < 100; i++) {
@@ -99,7 +99,7 @@ public class BucketTest extends ComponentTestCase {
 
 		// close and reload it, check if everything is okay
 		bucket.close();
-		bucket.initialize(MessageTree.class, new File("target/bucket/message"));
+		bucket.initialize(MessageTree.class, new File("target/bucket/"), "message");
 
 		// check next message in the same thread
 		for (int i = 0; i < groups - 1; i++) {
@@ -135,7 +135,7 @@ public class BucketTest extends ComponentTestCase {
 
 		// close and reload it, check if everything is okay
 		bucket.close();
-		bucket.initialize(String.class, new File("target/bucket/data"));
+		bucket.initialize(String.class, new File("target/bucket/"), "data");
 
 		// store it and load it
 		for (int i = 0; i < 100; i++) {

cat-job/pom.xml

@@ -42,6 +42,11 @@
 			<artifactId>jackson-mapper-asl</artifactId>
 			<version>1.9.4</version>
 		</dependency>
+		<dependency>
+			<groupId>com.dianping.tkv</groupId>
+			<artifactId>fkv</artifactId>
+			<version>0.1.0-SNAPSHOT</version>
+		</dependency>
 		<dependency>
 			<groupId>junit</groupId>
 			<artifactId>junit</artifactId>

cat-job/src/main/java/com/dianping/cat/job/hdfs/HdfsBucket.java

+/**
+ * 
+ */
+package com.dianping.cat.job.hdfs;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import com.dianping.cat.storage.Bucket;
+import com.dianping.tkv.Meta;
+import com.dianping.tkv.hdfs.HdfsImpl;
+
+/**
+ * @author sean.wang
+ * @since Mar 9, 2012
+ */
+public class HdfsBucket implements Bucket<byte[]> {
+
+	private HdfsImpl hdfs;
+
+	private int keyLength = 32;
+
+	private int tagLength = 125;
+
+	public void setHdfs(HdfsImpl hdfs) {
+		this.hdfs = hdfs;
+	}
+
+	public void setKeyLength(int keyLength) {
+		this.keyLength = keyLength;
+	}
+
+	public void setTagLength(int tagLength) {
+		this.tagLength = tagLength;
+	}
+
+	@Override
+	public boolean storeById(String id, byte[] data, String... tags) {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public List<String> findAllIdsByTag(String tag) {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public byte[] findNextById(String id, com.dianping.cat.storage.TagThreadSupport.Direction direction, String tag) throws IOException {
+		Meta meta = hdfs.getIndex(id, tag);
+		if (meta == null) {
+			return null;
+		}
+		int nextPos = meta.getTags().get(tag).getNext();
+		return hdfs.get(nextPos);
+	}
+
+	@Override
+	public void close() throws IOException {
+		this.hdfs.close();
+	}
+
+	@Override
+	public void deleteAndCreate() {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public List<byte[]> findAllByIds(List<String> ids) throws IOException {
+		List<byte[]> values = new ArrayList<byte[]>(ids.size());
+		for (String id : ids) {
+			byte[] value = this.findById(id);
+			values.add(value);
+		}
+		return values;
+	}
+
+	@Override
+	public byte[] findById(String id) throws IOException {
+		return hdfs.get(id);
+	}
+
+	/**
+	 * @param baseDir
+	 *            e.g /data/appdata/cat/
+	 * @param logicalPath
+	 *            e.g /a/b/c
+	 */
+	@Override
+	public void initialize(Class<?> type, File baseDir, String logicalPath) throws IOException {
+		String name = new File(logicalPath).getName();
+		String indexFilename = null;
+		String dataFilename = null;
+		String hdfsDir = null;
+		hdfs = new HdfsImpl(hdfsDir, baseDir, indexFilename, dataFilename, keyLength, tagLength);
+	}
+
+	@Override
+	public boolean storeById(String id, byte[] data) {
+		throw new UnsupportedOperationException();
+	}
+
+}

cat-job/src/main/java/com/dianping/cat/job/hdfs/HdfsBucketManager.java

+/**
+ * 
+ */
+package com.dianping.cat.job.hdfs;
+
+import com.dianping.cat.storage.internal.DefaultBucketManager;
+
+/**
+ * @author sean.wang
+ * @since Mar 9, 2012
+ */
+public class HdfsBucketManager extends DefaultBucketManager {
+
+}

cat-job/src/main/java/com/dianping/cat/job/hdfs/HdfsMessageStorage.java

@@ -11,11 +11,15 @@ import org.codehaus.plexus.personality.plexus.lifecycle.phase.Disposable;
 import org.codehaus.plexus.personality.plexus.lifecycle.phase.Initializable;
 import org.codehaus.plexus.personality.plexus.lifecycle.phase.InitializationException;
 
+import com.dianping.cat.message.spi.MessagePathBuilder;
 import com.dianping.cat.message.spi.MessageStorage;
 import com.dianping.cat.message.spi.MessageTree;
 import com.site.lookup.annotation.Inject;
 
 public class HdfsMessageStorage implements MessageStorage, Initializable, Disposable, LogEnabled {
+	@Inject
+	private MessagePathBuilder m_builder;
+
 	@Inject
 	private OutputChannelManager m_manager;
 
@@ -55,12 +59,13 @@ public class HdfsMessageStorage implements MessageStorage, Initializable, Dispos
 
 	@Override
 	public String getPath(MessageTree tree) {
-		// TODO fix it
-	   return null;
+		String path = m_builder.getLogViewPath(tree.getMessageId());
+
+		return path;
 	}
 
 	@Override
-	public String store(MessageTree tree) {
+	public String store(MessageTree tree, String... tags) {
 		m_job.append(tree);
 
 		// Not available
@@ -149,4 +154,16 @@ public class HdfsMessageStorage implements MessageStorage, Initializable, Dispos
 		//TODO
 		return null;
 	}
+
+	@Override
+	public MessageTree next(String messageId, String tag) {
+		// TODO Auto-generated method stub
+		return null;
+	}
+
+	@Override
+	public MessageTree previous(String messageId, String tag) {
+		// TODO Auto-generated method stub
+		return null;
+	}
 }