working for in memory single log buffer

11 files changed, 270 insertions, 189 deletions

diff --git a/weed/filer2/filer_notify.go b/weed/filer2/filer_notify.go
index c4fcefd54..94fbae4c4 100644
--- a/weed/filer2/filer_notify.go
+++ b/weed/filer2/filer_notify.go
@@ -1,6 +1,7 @@
 package filer2
 
 import (
+	"bytes"
 	"fmt"
 	"strings"
 	"time"
@@ -81,8 +82,13 @@ func (f *Filer) logFlushFunc(startTime, stopTime time.Time, buf []byte) {
 
 func (f *Filer) ReadLogBuffer(lastReadTime time.Time, eachEventFn func(fullpath string, eventNotification *filer_pb.EventNotification) error) (newLastReadTime time.Time, err error) {
 
-	var buf []byte
-	newLastReadTime, buf = f.metaLogBuffer.ReadFromBuffer(lastReadTime)
+	var bytesBuf *bytes.Buffer
+	bytesBuf = f.metaLogBuffer.ReadFromBuffer(lastReadTime)
+	if bytesBuf == nil {
+		return
+	}
+	defer f.metaLogBuffer.ReleaseMeory(bytesBuf)
+	buf := bytesBuf.Bytes()
 	var processedTs int64
 
 	for pos := 0; pos+4 < len(buf); {
diff --git a/weed/messaging/broker/broker_append.go b/weed/messaging/broker/broker_append.go
index c1ef063fb..7194dfcfc 100644
--- a/weed/messaging/broker/broker_append.go
+++ b/weed/messaging/broker/broker_append.go
@@ -16,10 +16,6 @@ import (
 
 func (broker *MessageBroker) appendToFile(targetFile string, topicConfig *messaging_pb.TopicConfiguration, data []byte) error {
 
-	if topicConfig.IsTransient {
-		return nil
-	}
-
 	assignResult, uploadResult, err2 := broker.assignAndUpload(topicConfig, data)
 	if err2 != nil {
 		return err2
diff --git a/weed/messaging/broker/broker_grpc_server_publish.go b/weed/messaging/broker/broker_grpc_server_publish.go
index 210127be3..985f708b5 100644
--- a/weed/messaging/broker/broker_grpc_server_publish.go
+++ b/weed/messaging/broker/broker_grpc_server_publish.go
@@ -2,7 +2,6 @@ package broker
 
 import (
 	"io"
-	"time"
 
 	"github.com/golang/protobuf/proto"
 
@@ -77,16 +76,9 @@ func (broker *MessageBroker) Publish(stream messaging_pb.SeaweedMessaging_Publis
 			continue
 		}
 
-		m := &messaging_pb.Message{
-			Timestamp: time.Now().UnixNano(),
-			Key:       in.Data.Key,
-			Value:     in.Data.Value,
-			Headers:   in.Data.Headers,
-		}
-
-		// fmt.Printf("received: %d : %s\n", len(m.Value), string(m.Value))
+		// fmt.Printf("received: %d : %s\n", len(in.Data.Value), string(in.Data.Value))
 
-		data, err := proto.Marshal(m)
+		data, err := proto.Marshal(in.Data)
 		if err != nil {
 			glog.Errorf("marshall error: %v\n", err)
 			continue
diff --git a/weed/messaging/broker/broker_grpc_server_subscribe.go b/weed/messaging/broker/broker_grpc_server_subscribe.go
index acf0330c6..c358eccf6 100644
--- a/weed/messaging/broker/broker_grpc_server_subscribe.go
+++ b/weed/messaging/broker/broker_grpc_server_subscribe.go
@@ -1,6 +1,7 @@
 package broker
 
 import (
+	"fmt"
 	"io"
 	"time"
 
@@ -9,7 +10,6 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"github.com/chrislusf/seaweedfs/weed/pb/messaging_pb"
-	"github.com/chrislusf/seaweedfs/weed/util"
 )
 
 func (broker *MessageBroker) Subscribe(stream messaging_pb.SeaweedMessaging_SubscribeServer) error {
@@ -23,12 +23,22 @@ func (broker *MessageBroker) Subscribe(stream messaging_pb.SeaweedMessaging_Subs
 		return err
 	}
 
+	var messageCount int64
 	subscriberId := in.Init.SubscriberId
-	println("+ subscriber:", subscriberId)
-	defer println("- subscriber:", subscriberId)
+	fmt.Printf("+ subscriber %s\n", subscriberId)
+	defer func() {
+		fmt.Printf("- subscriber %s: %d messages\n", subscriberId, messageCount)
+	}()
 
 	// TODO look it up
 	topicConfig := &messaging_pb.TopicConfiguration{
+		IsTransient: true,
+	}
+
+	if err = stream.Send(&messaging_pb.BrokerMessage{
+		Redirect: nil,
+	}); err != nil {
+		return err
 	}
 
 	// get lock
@@ -52,7 +62,7 @@ func (broker *MessageBroker) Subscribe(stream messaging_pb.SeaweedMessaging_Subs
 	// an error returned will end the subscription
 	eachMessageFn := func(m *messaging_pb.Message) error {
 		err := stream.Send(&messaging_pb.BrokerMessage{
-			Data:     m,
+			Data: m,
 		})
 		if err != nil {
 			glog.V(0).Infof("=> subscriber %v: %+v", subscriberId, err)
@@ -60,42 +70,25 @@ func (broker *MessageBroker) Subscribe(stream messaging_pb.SeaweedMessaging_Subs
 		return err
 	}
 
-	// loop through all messages
-	for {
-
-		_, buf := lock.logBuffer.ReadFromBuffer(lastReadTime)
-
-		for pos := 0; pos+4 < len(buf); {
-
-			size := util.BytesToUint32(buf[pos : pos+4])
-			entryData := buf[pos+4 : pos+4+int(size)]
-
-			logEntry := &filer_pb.LogEntry{}
-			if err = proto.Unmarshal(entryData, logEntry); err != nil {
-				glog.Errorf("unexpected unmarshal messaging_pb.Message: %v", err)
-				pos += 4 + int(size)
-				continue
-			}
-
-			m := &messaging_pb.Message{}
-			if err = proto.Unmarshal(logEntry.Data, m); err != nil {
-				glog.Errorf("unexpected unmarshal messaging_pb.Message: %v", err)
-				pos += 4 + int(size)
-				continue
-			}
-
-			// fmt.Printf("sending : %d : %s\n", len(m.Value), string(m.Value))
-			if err = eachMessageFn(m); err != nil {
-				return err
-			}
-
-			lastReadTime = time.Unix(0, m.Timestamp)
-			pos += 4 + int(size)
-		}
-
+	messageCount, err = lock.logBuffer.LoopProcessLogData(lastReadTime, func() bool {
 		lock.Mutex.Lock()
 		lock.cond.Wait()
 		lock.Mutex.Unlock()
-	}
+		return true
+	}, func(logEntry *filer_pb.LogEntry) error {
+		m := &messaging_pb.Message{}
+		if err = proto.Unmarshal(logEntry.Data, m); err != nil {
+			glog.Errorf("unexpected unmarshal messaging_pb.Message: %v", err)
+			return err
+		}
+		// fmt.Printf("sending : %d bytes\n", len(m.Value))
+		if err = eachMessageFn(m); err != nil {
+			glog.Errorf("sending %d bytes to %s: %s", len(m.Value), subscriberId, err)
+			return err
+		}
+		return nil
+	})
+
+	return err
 
 }
diff --git a/weed/messaging/broker/topic_lock.go b/weed/messaging/broker/topic_lock.go
index 652ff0545..74861a147 100644
--- a/weed/messaging/broker/topic_lock.go
+++ b/weed/messaging/broker/topic_lock.go
@@ -41,6 +41,10 @@ func (locks *TopicLocks) buildLogBuffer(tl *TopicLock, tp TopicPartition, topicC
 
 	flushFn := func(startTime, stopTime time.Time, buf []byte) {
 
+		if topicConfig.IsTransient {
+			return
+		}
+
 		targetFile := fmt.Sprintf(
 			"%s/%s/%s/%04d-%02d-%02d/%02d-%02d.part%02d",
 			filer2.TopicsDir, tp.Namespace, tp.Topic,
diff --git a/weed/messaging/client/publisher.go b/weed/messaging/client/publisher.go
index 238b67783..3480ff55d 100644
--- a/weed/messaging/client/publisher.go
+++ b/weed/messaging/client/publisher.go
@@ -61,7 +61,7 @@ func (mc *MessagingClient) NewPublisher(namespace, topic string) (*Publisher, er
 	}, nil
 }
 
-func (p *Publisher) Publish(m *messaging_pb.RawData) error {
+func (p *Publisher) Publish(m *messaging_pb.Message) error {
 
 	return p.publishClient.Send(&messaging_pb.PublishRequest{
 		Data: m,
diff --git a/weed/pb/messaging.proto b/weed/pb/messaging.proto
index 38be82238..ff50c4ce3 100644
--- a/weed/pb/messaging.proto
+++ b/weed/pb/messaging.proto
@@ -46,14 +46,8 @@ message SubscriberMessage {
     AckMessage ack = 2;
 }
 
-message RawData {
-    bytes key = 1; // Message key
-    bytes value = 2; // Message payload
-    map<string, bytes> headers = 3; // Message headers
-}
-
 message Message {
-    int64 timestamp = 1 [jstype = JS_STRING]; // When the message was received by the broker
+    int64 event_time_ns = 1 [jstype = JS_STRING];
     bytes key = 2; // Message key
     bytes value = 3; // Message payload
     map<string, bytes> headers = 4; // Message headers
@@ -74,7 +68,7 @@ message PublishRequest {
         int32 partition = 3;
     }
     InitMessage init = 1;
-    RawData data = 2;
+    Message data = 2;
 }
 
 message PublishResponse {
diff --git a/weed/pb/messaging_pb/messaging.pb.go b/weed/pb/messaging_pb/messaging.pb.go
index 40d2223f5..bf244f87c 100644
--- a/weed/pb/messaging_pb/messaging.pb.go
+++ b/weed/pb/messaging_pb/messaging.pb.go
@@ -10,7 +10,6 @@ It is generated from these files:
 
 It has these top-level messages:
 	SubscriberMessage
-	RawData
 	Message
 	BrokerMessage
 	PublishRequest
@@ -167,53 +166,21 @@ func (m *SubscriberMessage_AckMessage) GetMessageId() int64 {
 	return 0
 }
 
-type RawData struct {
-	Key     []byte            `protobuf:"bytes,1,opt,name=key,proto3" json:"key,omitempty"`
-	Value   []byte            `protobuf:"bytes,2,opt,name=value,proto3" json:"value,omitempty"`
-	Headers map[string][]byte `protobuf:"bytes,3,rep,name=headers" json:"headers,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value,proto3"`
-}
-
-func (m *RawData) Reset()                    { *m = RawData{} }
-func (m *RawData) String() string            { return proto.CompactTextString(m) }
-func (*RawData) ProtoMessage()               {}
-func (*RawData) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{1} }
-
-func (m *RawData) GetKey() []byte {
-	if m != nil {
-		return m.Key
-	}
-	return nil
-}
-
-func (m *RawData) GetValue() []byte {
-	if m != nil {
-		return m.Value
-	}
-	return nil
-}
-
-func (m *RawData) GetHeaders() map[string][]byte {
-	if m != nil {
-		return m.Headers
-	}
-	return nil
-}
-
 type Message struct {
-	Timestamp int64             `protobuf:"varint,1,opt,name=timestamp" json:"timestamp,omitempty"`
-	Key       []byte            `protobuf:"bytes,2,opt,name=key,proto3" json:"key,omitempty"`
-	Value     []byte            `protobuf:"bytes,3,opt,name=value,proto3" json:"value,omitempty"`
-	Headers   map[string][]byte `protobuf:"bytes,4,rep,name=headers" json:"headers,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value,proto3"`
+	EventTimeNs int64             `protobuf:"varint,1,opt,name=event_time_ns,json=eventTimeNs" json:"event_time_ns,omitempty"`
+	Key         []byte            `protobuf:"bytes,2,opt,name=key,proto3" json:"key,omitempty"`
+	Value       []byte            `protobuf:"bytes,3,opt,name=value,proto3" json:"value,omitempty"`
+	Headers     map[string][]byte `protobuf:"bytes,4,rep,name=headers" json:"headers,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value,proto3"`
 }
 
 func (m *Message) Reset()                    { *m = Message{} }
 func (m *Message) String() string            { return proto.CompactTextString(m) }
 func (*Message) ProtoMessage()               {}
-func (*Message) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{2} }
+func (*Message) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{1} }
 
-func (m *Message) GetTimestamp() int64 {
+func (m *Message) GetEventTimeNs() int64 {
 	if m != nil {
-		return m.Timestamp
+		return m.EventTimeNs
 	}
 	return 0
 }
@@ -247,7 +214,7 @@ type BrokerMessage struct {
 func (m *BrokerMessage) Reset()                    { *m = BrokerMessage{} }
 func (m *BrokerMessage) String() string            { return proto.CompactTextString(m) }
 func (*BrokerMessage) ProtoMessage()               {}
-func (*BrokerMessage) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{3} }
+func (*BrokerMessage) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{2} }
 
 func (m *BrokerMessage) GetData() *Message {
 	if m != nil {
@@ -271,7 +238,7 @@ func (m *BrokerMessage_RedirectMessage) Reset()         { *m = BrokerMessage_Red
 func (m *BrokerMessage_RedirectMessage) String() string { return proto.CompactTextString(m) }
 func (*BrokerMessage_RedirectMessage) ProtoMessage()    {}
 func (*BrokerMessage_RedirectMessage) Descriptor() ([]byte, []int) {
-	return fileDescriptor0, []int{3, 0}
+	return fileDescriptor0, []int{2, 0}
 }
 
 func (m *BrokerMessage_RedirectMessage) GetNewBroker() string {
@@ -283,13 +250,13 @@ func (m *BrokerMessage_RedirectMessage) GetNewBroker() string {
 
 type PublishRequest struct {
 	Init *PublishRequest_InitMessage `protobuf:"bytes,1,opt,name=init" json:"init,omitempty"`
-	Data *RawData                    `protobuf:"bytes,2,opt,name=data" json:"data,omitempty"`
+	Data *Message                    `protobuf:"bytes,2,opt,name=data" json:"data,omitempty"`
 }
 
 func (m *PublishRequest) Reset()                    { *m = PublishRequest{} }
 func (m *PublishRequest) String() string            { return proto.CompactTextString(m) }
 func (*PublishRequest) ProtoMessage()               {}
-func (*PublishRequest) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{4} }
+func (*PublishRequest) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{3} }
 
 func (m *PublishRequest) GetInit() *PublishRequest_InitMessage {
 	if m != nil {
@@ -298,7 +265,7 @@ func (m *PublishRequest) GetInit() *PublishRequest_InitMessage {
 	return nil
 }
 
-func (m *PublishRequest) GetData() *RawData {
+func (m *PublishRequest) GetData() *Message {
 	if m != nil {
 		return m.Data
 	}
@@ -314,7 +281,7 @@ type PublishRequest_InitMessage struct {
 func (m *PublishRequest_InitMessage) Reset()                    { *m = PublishRequest_InitMessage{} }
 func (m *PublishRequest_InitMessage) String() string            { return proto.CompactTextString(m) }
 func (*PublishRequest_InitMessage) ProtoMessage()               {}
-func (*PublishRequest_InitMessage) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{4, 0} }
+func (*PublishRequest_InitMessage) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{3, 0} }
 
 func (m *PublishRequest_InitMessage) GetNamespace() string {
 	if m != nil {
@@ -345,7 +312,7 @@ type PublishResponse struct {
 func (m *PublishResponse) Reset()                    { *m = PublishResponse{} }
 func (m *PublishResponse) String() string            { return proto.CompactTextString(m) }
 func (*PublishResponse) ProtoMessage()               {}
-func (*PublishResponse) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{5} }
+func (*PublishResponse) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{4} }
 
 func (m *PublishResponse) GetConfig() *PublishResponse_ConfigMessage {
 	if m != nil {
@@ -369,7 +336,7 @@ func (m *PublishResponse_ConfigMessage) Reset()         { *m = PublishResponse_C
 func (m *PublishResponse_ConfigMessage) String() string { return proto.CompactTextString(m) }
 func (*PublishResponse_ConfigMessage) ProtoMessage()    {}
 func (*PublishResponse_ConfigMessage) Descriptor() ([]byte, []int) {
-	return fileDescriptor0, []int{5, 0}
+	return fileDescriptor0, []int{4, 0}
 }
 
 func (m *PublishResponse_ConfigMessage) GetPartitionCount() int32 {
@@ -387,7 +354,7 @@ func (m *PublishResponse_RedirectMessage) Reset()         { *m = PublishResponse
 func (m *PublishResponse_RedirectMessage) String() string { return proto.CompactTextString(m) }
 func (*PublishResponse_RedirectMessage) ProtoMessage()    {}
 func (*PublishResponse_RedirectMessage) Descriptor() ([]byte, []int) {
-	return fileDescriptor0, []int{5, 1}
+	return fileDescriptor0, []int{4, 1}
 }
 
 func (m *PublishResponse_RedirectMessage) GetNewBroker() string {
@@ -406,7 +373,7 @@ type ConfigureTopicRequest struct {
 func (m *ConfigureTopicRequest) Reset()                    { *m = ConfigureTopicRequest{} }
 func (m *ConfigureTopicRequest) String() string            { return proto.CompactTextString(m) }
 func (*ConfigureTopicRequest) ProtoMessage()               {}
-func (*ConfigureTopicRequest) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{6} }
+func (*ConfigureTopicRequest) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{5} }
 
 func (m *ConfigureTopicRequest) GetNamespace() string {
 	if m != nil {
@@ -435,7 +402,7 @@ type ConfigureTopicResponse struct {
 func (m *ConfigureTopicResponse) Reset()                    { *m = ConfigureTopicResponse{} }
 func (m *ConfigureTopicResponse) String() string            { return proto.CompactTextString(m) }
 func (*ConfigureTopicResponse) ProtoMessage()               {}
-func (*ConfigureTopicResponse) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{7} }
+func (*ConfigureTopicResponse) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{6} }
 
 type GetTopicConfigurationRequest struct {
 	Namespace string `protobuf:"bytes,1,opt,name=namespace" json:"namespace,omitempty"`
@@ -445,7 +412,7 @@ type GetTopicConfigurationRequest struct {
 func (m *GetTopicConfigurationRequest) Reset()                    { *m = GetTopicConfigurationRequest{} }
 func (m *GetTopicConfigurationRequest) String() string            { return proto.CompactTextString(m) }
 func (*GetTopicConfigurationRequest) ProtoMessage()               {}
-func (*GetTopicConfigurationRequest) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{8} }
+func (*GetTopicConfigurationRequest) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{7} }
 
 func (m *GetTopicConfigurationRequest) GetNamespace() string {
 	if m != nil {
@@ -468,7 +435,7 @@ type GetTopicConfigurationResponse struct {
 func (m *GetTopicConfigurationResponse) Reset()                    { *m = GetTopicConfigurationResponse{} }
 func (m *GetTopicConfigurationResponse) String() string            { return proto.CompactTextString(m) }
 func (*GetTopicConfigurationResponse) ProtoMessage()               {}
-func (*GetTopicConfigurationResponse) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{9} }
+func (*GetTopicConfigurationResponse) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{8} }
 
 func (m *GetTopicConfigurationResponse) GetConfiguration() *TopicConfiguration {
 	if m != nil {
@@ -487,7 +454,7 @@ type TopicConfiguration struct {
 func (m *TopicConfiguration) Reset()                    { *m = TopicConfiguration{} }
 func (m *TopicConfiguration) String() string            { return proto.CompactTextString(m) }
 func (*TopicConfiguration) ProtoMessage()               {}
-func (*TopicConfiguration) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{10} }
+func (*TopicConfiguration) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{9} }
 
 func (m *TopicConfiguration) GetPartitionCount() int32 {
 	if m != nil {
@@ -521,7 +488,6 @@ func init() {
 	proto.RegisterType((*SubscriberMessage)(nil), "messaging_pb.SubscriberMessage")
 	proto.RegisterType((*SubscriberMessage_InitMessage)(nil), "messaging_pb.SubscriberMessage.InitMessage")
 	proto.RegisterType((*SubscriberMessage_AckMessage)(nil), "messaging_pb.SubscriberMessage.AckMessage")
-	proto.RegisterType((*RawData)(nil), "messaging_pb.RawData")
 	proto.RegisterType((*Message)(nil), "messaging_pb.Message")
 	proto.RegisterType((*BrokerMessage)(nil), "messaging_pb.BrokerMessage")
 	proto.RegisterType((*BrokerMessage_RedirectMessage)(nil), "messaging_pb.BrokerMessage.RedirectMessage")
@@ -777,59 +743,57 @@ var _SeaweedMessaging_serviceDesc = grpc.ServiceDesc{
 func init() { proto.RegisterFile("messaging.proto", fileDescriptor0) }
 
 var fileDescriptor0 = []byte{
-	// 849 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x09, 0x6e, 0x88, 0x02, 0xff, 0xb4, 0x56, 0xdd, 0x6e, 0xe3, 0x44,
-	0x14, 0xde, 0xb1, 0xd3, 0x1f, 0x9f, 0xfc, 0x34, 0x8c, 0x28, 0x8a, 0x4c, 0x0b, 0xc6, 0x8b, 0x44,
-	0xa0, 0xc2, 0xaa, 0xc2, 0x4d, 0x59, 0xad, 0x84, 0xda, 0x52, 0x96, 0x48, 0x0d, 0x44, 0x93, 0xdc,
-	0xa2, 0x68, 0xe2, 0xcc, 0x66, 0x47, 0x49, 0x6c, 0xe3, 0x99, 0x10, 0xed, 0x35, 0xdc, 0x72, 0xc5,
-	0x1b, 0x70, 0xcb, 0x35, 0x0f, 0xc0, 0x03, 0xf0, 0x02, 0x3c, 0x0d, 0xf2, 0xf8, 0x27, 0x76, 0xe2,
-	0x66, 0x4b, 0xc4, 0xde, 0xd9, 0xc7, 0xdf, 0xf9, 0xce, 0xf9, 0xce, 0xcf, 0x8c, 0xe1, 0x64, 0xc1,
-	0x84, 0xa0, 0x53, 0xee, 0x4d, 0x9d, 0x20, 0xf4, 0xa5, 0x8f, 0x6b, 0x99, 0x61, 0x14, 0x8c, 0xed,
-	0x9f, 0x2b, 0xf0, 0xce, 0x60, 0x39, 0x16, 0x6e, 0xc8, 0xc7, 0x2c, 0xec, 0xa9, 0x4f, 0x0c, 0x7f,
-	0x05, 0x15, 0xee, 0x71, 0xd9, 0x42, 0x16, 0x6a, 0x57, 0x3b, 0x17, 0x4e, 0xde, 0xc5, 0xd9, 0x82,
-	0x3b, 0x5d, 0x8f, 0xcb, 0xe4, 0x99, 0x28, 0x47, 0xfc, 0x1c, 0x74, 0xea, 0xce, 0x5a, 0x9a, 0xf2,
-	0xff, 0xec, 0x4d, 0xfe, 0xd7, 0xee, 0x2c, 0x75, 0x8f, 0xdc, 0xcc, 0xbf, 0x34, 0xa8, 0xe6, 0x38,
-	0xf1, 0x19, 0x18, 0x1e, 0x5d, 0x30, 0x11, 0x50, 0x97, 0xa9, 0x9c, 0x0c, 0xb2, 0x36, 0xe0, 0x77,
-	0xe1, 0x40, 0xfa, 0x01, 0x77, 0x55, 0x34, 0x83, 0xc4, 0x2f, 0x91, 0x4f, 0x40, 0x43, 0xc9, 0x25,
-	0xf7, 0xbd, 0x96, 0x6e, 0xa1, 0xf6, 0x01, 0x59, 0x1b, 0xf0, 0x08, 0xea, 0x42, 0xd2, 0x50, 0xf6,
-	0x7d, 0x11, 0x23, 0x2a, 0x16, 0x6a, 0x37, 0x3a, 0x5f, 0xfe, 0x07, 0xa5, 0xce, 0x20, 0x4f, 0x40,
-	0x8a, 0x7c, 0xd8, 0x82, 0xaa, 0xe4, 0x0b, 0x26, 0x24, 0x5d, 0x04, 0xdf, 0x89, 0xd6, 0x81, 0x85,
-	0xda, 0x3a, 0xc9, 0x9b, 0xf0, 0x53, 0xa8, 0x8b, 0x8c, 0x7f, 0xc4, 0x27, 0xad, 0x43, 0x95, 0x7e,
-	0x6d, 0x6d, 0xec, 0x4e, 0xec, 0x2b, 0xa8, 0x17, 0xc2, 0x60, 0x80, 0xc3, 0xfb, 0xeb, 0xe1, 0xdd,
-	0x60, 0xd8, 0x7c, 0x82, 0x6b, 0x70, 0x7c, 0x77, 0x4d, 0xee, 0xbb, 0xd1, 0x1b, 0xc2, 0x75, 0x30,
-	0x86, 0xdd, 0xde, 0xdd, 0x60, 0x78, 0xdd, 0xeb, 0x37, 0x35, 0xf3, 0x02, 0x60, 0x5d, 0x56, 0x7c,
-	0x0e, 0x10, 0x2b, 0x63, 0x51, 0x24, 0xa4, 0xb2, 0x31, 0x12, 0x4b, 0x77, 0x62, 0xff, 0x81, 0xe0,
-	0x88, 0xd0, 0xd5, 0xd7, 0x54, 0x52, 0xdc, 0x04, 0x7d, 0xc6, 0x5e, 0x2b, 0x4c, 0x8d, 0x44, 0x8f,
-	0x51, 0x81, 0x7f, 0xa2, 0xf3, 0x25, 0x53, 0x05, 0xae, 0x91, 0xf8, 0x05, 0x3f, 0x87, 0xa3, 0x57,
-	0x8c, 0x4e, 0x58, 0x28, 0x5a, 0xba, 0xa5, 0xb7, 0xab, 0x1d, 0xbb, 0x58, 0xbc, 0x84, 0xcf, 0xf9,
-	0x36, 0x06, 0xdd, 0x79, 0x32, 0x7c, 0x4d, 0x52, 0x17, 0xf3, 0x19, 0xd4, 0xf2, 0x1f, 0xf2, 0x51,
-	0x8d, 0x1d, 0x51, 0x9f, 0x69, 0x57, 0xc8, 0xfe, 0x1b, 0xc1, 0x51, 0x2a, 0xcc, 0x02, 0x23, 0x2b,
-	0x6a, 0xac, 0xeb, 0x46, 0xbb, 0x44, 0x64, 0x6d, 0x4c, 0x99, 0xb5, 0x12, 0x3d, 0xfa, 0x03, 0x7a,
-	0x2a, 0x65, 0x7a, 0xd2, 0xb6, 0xff, 0xff, 0x7a, 0xfe, 0x44, 0x50, 0xbf, 0x09, 0xfd, 0xd9, 0x7a,
-	0xff, 0x3e, 0x85, 0xca, 0x84, 0x4a, 0x9a, 0xec, 0xdf, 0x69, 0x69, 0x22, 0x44, 0x41, 0xf0, 0x0b,
-	0x38, 0x0e, 0xd9, 0x84, 0x87, 0xcc, 0x95, 0xc9, 0xba, 0x6d, 0xac, 0x6b, 0x81, 0xd9, 0x21, 0x09,
-	0x36, 0x25, 0xc9, 0x9c, 0xcd, 0x4b, 0x38, 0xd9, 0xf8, 0x18, 0x4d, 0x8d, 0xc7, 0x56, 0xa3, 0xb1,
-	0x62, 0xc8, 0x16, 0x8f, 0xad, 0x62, 0x4a, 0xfb, 0x1f, 0x04, 0x8d, 0xfe, 0x72, 0x3c, 0xe7, 0xe2,
-	0x15, 0x61, 0x3f, 0x2e, 0x99, 0x88, 0xf6, 0x3e, 0x7f, 0x70, 0xb4, 0x8b, 0x99, 0x14, 0xb1, 0x25,
-	0xa7, 0x46, 0x2a, 0x5b, 0x2b, 0x93, 0x9d, 0xcc, 0x53, 0x2c, 0xdb, 0x1c, 0xbd, 0xe5, 0x13, 0xc2,
-	0xfe, 0x55, 0x83, 0x93, 0x2c, 0x61, 0x11, 0xf8, 0x9e, 0x60, 0xf8, 0x16, 0x0e, 0x5d, 0xdf, 0x7b,
-	0xc9, 0xa7, 0xe5, 0x07, 0xe3, 0x06, 0xdc, 0xb9, 0x55, 0xd8, 0x54, 0x62, 0xe2, 0x8a, 0xbb, 0x5b,
-	0x0d, 0xfb, 0x7c, 0x37, 0xcd, 0xc3, 0x2d, 0xbb, 0x82, 0x7a, 0x21, 0x06, 0xfe, 0x04, 0x4e, 0x32,
-	0x05, 0x23, 0xd7, 0x5f, 0x7a, 0x71, 0x27, 0x0e, 0x48, 0x23, 0x33, 0xdf, 0x46, 0xd6, 0x3d, 0x9a,
-	0xfd, 0x1b, 0x82, 0xd3, 0x38, 0xd8, 0x32, 0x64, 0xc3, 0xa8, 0x80, 0x69, 0xcf, 0xf7, 0xa9, 0xfd,
-	0x37, 0x50, 0x77, 0x13, 0x32, 0x9a, 0xd5, 0xbf, 0xda, 0xb1, 0x8a, 0x95, 0x50, 0x61, 0x6e, 0xf3,
-	0x38, 0x52, 0x74, 0xb3, 0x5b, 0xf0, 0xde, 0x66, 0x52, 0x71, 0xd5, 0x6c, 0x02, 0x67, 0x2f, 0x98,
-	0x2c, 0x61, 0xd8, 0x3f, 0x6b, 0x7b, 0x0a, 0xe7, 0x0f, 0x70, 0x26, 0x03, 0xb2, 0x25, 0x0b, 0xed,
-	0x27, 0xeb, 0x77, 0x04, 0x78, 0x1b, 0xf5, 0xe8, 0xf6, 0xe2, 0x0f, 0x00, 0x5c, 0x7f, 0x3e, 0x67,
-	0xae, 0x4a, 0x22, 0xd6, 0x90, 0xb3, 0x44, 0xb7, 0x53, 0xc8, 0x82, 0x39, 0x77, 0xd7, 0xc5, 0x37,
-	0x48, 0xde, 0x84, 0x3f, 0x82, 0x1a, 0x17, 0x23, 0x19, 0x52, 0x4f, 0x70, 0xe6, 0x49, 0x75, 0x3f,
-	0x1e, 0x93, 0x2a, 0x17, 0xc3, 0xd4, 0xd4, 0xf9, 0x45, 0x87, 0xe6, 0x80, 0xd1, 0x15, 0x63, 0x93,
-	0x5e, 0x2a, 0x0f, 0x7f, 0x0f, 0x46, 0x76, 0x6b, 0xe2, 0x0f, 0xdf, 0x70, 0x9d, 0x9a, 0xef, 0xef,
-	0x38, 0xaa, 0xec, 0x27, 0x6d, 0x74, 0x89, 0xf0, 0x3d, 0x1c, 0x25, 0x0b, 0x81, 0xcf, 0x76, 0x1d,
-	0x27, 0xe6, 0xf9, 0xce, 0x2d, 0x4a, 0xd8, 0x7e, 0x80, 0x46, 0x71, 0x5e, 0xf0, 0xd3, 0xa2, 0x5b,
-	0xe9, 0x88, 0x9b, 0x1f, 0xef, 0x06, 0xa5, 0x21, 0x70, 0x08, 0xa7, 0xa5, 0x03, 0x82, 0x37, 0x7e,
-	0x81, 0x76, 0x4d, 0xa6, 0x79, 0xf1, 0x28, 0x6c, 0x1a, 0xf3, 0xc6, 0x86, 0xa6, 0x88, 0xbb, 0xf0,
-	0x52, 0x38, 0xee, 0x3c, 0x6a, 0xcd, 0x4d, 0x23, 0x6b, 0x48, 0x3f, 0xfa, 0xe7, 0x1b, 0x1f, 0xaa,
-	0x5f, 0xbf, 0x2f, 0xfe, 0x0d, 0x00, 0x00, 0xff, 0xff, 0x62, 0x10, 0x0f, 0xed, 0x0d, 0x0a, 0x00,
-	0x00,
+	// 832 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x09, 0x6e, 0x88, 0x02, 0xff, 0xb4, 0x56, 0x5d, 0x8f, 0xea, 0x44,
+	0x18, 0x3e, 0x53, 0x3e, 0x76, 0x79, 0xa1, 0x80, 0x13, 0xd7, 0x90, 0xba, 0xab, 0xd8, 0x63, 0x14,
+	0xdd, 0xd8, 0x6c, 0xf0, 0x66, 0x3d, 0x39, 0x89, 0x01, 0x82, 0x47, 0x92, 0xe5, 0x48, 0x06, 0x6e,
+	0x4d, 0x53, 0xca, 0x1c, 0x76, 0x02, 0xb4, 0xb5, 0x33, 0xec, 0x66, 0xaf, 0xf5, 0xd6, 0x2b, 0xff,
+	0x81, 0xff, 0xc1, 0x1f, 0xe0, 0x6f, 0xf0, 0xce, 0x5f, 0x63, 0x3a, 0xfd, 0xa0, 0x05, 0x96, 0x5d,
+	0x49, 0xce, 0x5d, 0xfb, 0xce, 0xf3, 0x3c, 0xef, 0xf7, 0xb4, 0x50, 0x5b, 0x51, 0xce, 0xad, 0x39,
+	0x73, 0xe6, 0x86, 0xe7, 0xbb, 0xc2, 0xc5, 0x95, 0xc4, 0x60, 0x7a, 0x53, 0xfd, 0xd7, 0x3c, 0x7c,
+	0x30, 0x5e, 0x4f, 0xb9, 0xed, 0xb3, 0x29, 0xf5, 0x87, 0xf2, 0x88, 0xe2, 0xef, 0x21, 0xcf, 0x1c,
+	0x26, 0x1a, 0xa8, 0x89, 0x5a, 0xe5, 0xf6, 0xa5, 0x91, 0xa6, 0x18, 0x3b, 0x70, 0x63, 0xe0, 0x30,
+	0x11, 0x3d, 0x13, 0x49, 0xc4, 0xaf, 0x21, 0x67, 0xd9, 0x8b, 0x86, 0x22, 0xf9, 0x5f, 0x3f, 0xc5,
+	0xef, 0xd8, 0x8b, 0x98, 0x1e, 0xd0, 0xb4, 0xbf, 0x15, 0x28, 0xa7, 0x34, 0xf1, 0x39, 0x94, 0x1c,
+	0x6b, 0x45, 0xb9, 0x67, 0xd9, 0x54, 0xc6, 0x54, 0x22, 0x1b, 0x03, 0xfe, 0x10, 0x0a, 0xc2, 0xf5,
+	0x98, 0x2d, 0xbd, 0x95, 0x48, 0xf8, 0x12, 0x70, 0x3c, 0xcb, 0x17, 0x4c, 0x30, 0xd7, 0x69, 0xe4,
+	0x9a, 0xa8, 0x55, 0x20, 0x1b, 0x03, 0x36, 0x41, 0xe5, 0xc2, 0xf2, 0xc5, 0xc8, 0xe5, 0x21, 0x22,
+	0xdf, 0x44, 0xad, 0x6a, 0xfb, 0xbb, 0xff, 0x91, 0xa9, 0x31, 0x4e, 0x0b, 0x90, 0xac, 0x1e, 0x6e,
+	0x42, 0x59, 0xb0, 0x15, 0xe5, 0xc2, 0x5a, 0x79, 0x6f, 0x79, 0xa3, 0xd0, 0x44, 0xad, 0x1c, 0x49,
+	0x9b, 0xf0, 0x4b, 0x50, 0x79, 0xa2, 0x6f, 0xb2, 0x59, 0xa3, 0x28, 0xc3, 0xaf, 0x6c, 0x8c, 0x83,
+	0x99, 0x7e, 0x0d, 0x6a, 0xc6, 0x0d, 0x06, 0x28, 0xde, 0x74, 0x26, 0xfd, 0xf1, 0xa4, 0xfe, 0x02,
+	0x57, 0xe0, 0xb4, 0xdf, 0x21, 0x37, 0x83, 0xe0, 0x0d, 0x61, 0x15, 0x4a, 0x93, 0xc1, 0xb0, 0x3f,
+	0x9e, 0x74, 0x86, 0xa3, 0xba, 0xa2, 0x5d, 0x02, 0x6c, 0xca, 0x8a, 0x2f, 0x00, 0xc2, 0xcc, 0x68,
+	0xe0, 0x09, 0xc9, 0x68, 0x4a, 0x91, 0x65, 0x30, 0xd3, 0xff, 0x41, 0x70, 0x12, 0x43, 0xbf, 0x00,
+	0x95, 0xde, 0x51, 0x47, 0x98, 0x41, 0xb0, 0xa6, 0xc3, 0x43, 0x74, 0x57, 0xb9, 0x42, 0xa4, 0x2c,
+	0x0f, 0x26, 0x6c, 0x45, 0xdf, 0x72, 0x5c, 0x87, 0xdc, 0x82, 0x3e, 0xc8, 0xa2, 0x57, 0x48, 0xf0,
+	0x18, 0x34, 0xe2, 0xce, 0x5a, 0xae, 0xa9, 0x2c, 0x77, 0x85, 0x84, 0x2f, 0xf8, 0x35, 0x9c, 0xdc,
+	0x52, 0x6b, 0x46, 0x7d, 0xde, 0xc8, 0x37, 0x73, 0xad, 0x72, 0x5b, 0xcf, 0x16, 0x39, 0x2e, 0xe7,
+	0x8f, 0x21, 0xa8, 0xef, 0x08, 0xff, 0x81, 0xc4, 0x14, 0xed, 0x15, 0x54, 0xd2, 0x07, 0xb1, 0xd7,
+	0x70, 0x08, 0xb2, 0x5e, 0x95, 0x94, 0xd7, 0x57, 0xca, 0x35, 0xd2, 0xff, 0x42, 0xa0, 0x76, 0x7d,
+	0x77, 0xb1, 0x99, 0xeb, 0xaf, 0x20, 0x3f, 0xb3, 0x84, 0x15, 0xcd, 0xf5, 0xd9, 0xde, 0x40, 0x88,
+	0x84, 0xe0, 0x37, 0x70, 0xea, 0xd3, 0x19, 0xf3, 0xa9, 0x2d, 0xa2, 0x31, 0xde, 0x5a, 0x83, 0x8c,
+	0xb2, 0x41, 0x22, 0x6c, 0x2c, 0x92, 0x90, 0xb5, 0x2b, 0xa8, 0x6d, 0x1d, 0x06, 0xdd, 0x70, 0xe8,
+	0xbd, 0x39, 0x95, 0x0a, 0xc9, 0x40, 0xd3, 0xfb, 0x50, 0x52, 0xff, 0x17, 0x41, 0x75, 0xb4, 0x9e,
+	0x2e, 0x19, 0xbf, 0x25, 0xf4, 0x97, 0x35, 0xe5, 0xc1, 0x3e, 0xa5, 0x17, 0xb2, 0x95, 0x8d, 0x24,
+	0x8b, 0xdd, 0xb3, 0x8d, 0x71, 0xda, 0xca, 0x93, 0x69, 0x6b, 0xe6, 0x7b, 0xde, 0x3c, 0xfd, 0x77,
+	0x05, 0x6a, 0x49, 0xc0, 0xdc, 0x73, 0x1d, 0x4e, 0x71, 0x0f, 0x8a, 0xb6, 0xeb, 0xbc, 0x63, 0xf3,
+	0xfd, 0x17, 0xce, 0x16, 0xdc, 0xe8, 0x49, 0x6c, 0x1c, 0x77, 0x44, 0xc5, 0x83, 0x9d, 0x86, 0x7d,
+	0x73, 0x58, 0xe6, 0xf1, 0x96, 0x5d, 0x83, 0x9a, 0xf1, 0x81, 0xbf, 0x84, 0x5a, 0x92, 0x81, 0x69,
+	0xbb, 0x6b, 0x27, 0xec, 0x44, 0x81, 0x54, 0x13, 0x73, 0x2f, 0xb0, 0x1e, 0xd1, 0xec, 0x3f, 0x10,
+	0x9c, 0x85, 0xce, 0xd6, 0x3e, 0x9d, 0x04, 0x05, 0x8c, 0x7b, 0x7e, 0x4c, 0xed, 0x7f, 0x00, 0xd5,
+	0x8e, 0xc4, 0xac, 0xa4, 0xfe, 0xe5, 0x76, 0x33, 0x5b, 0x09, 0xe9, 0xa6, 0x97, 0xc6, 0x91, 0x2c,
+	0x4d, 0x6f, 0xc0, 0x47, 0xdb, 0x41, 0x85, 0x55, 0xd3, 0x09, 0x9c, 0xbf, 0xa1, 0x62, 0x8f, 0xc2,
+	0xf1, 0x51, 0xeb, 0x73, 0xb8, 0x78, 0x44, 0x33, 0x1a, 0x90, 0x9d, 0xb4, 0xd0, 0x71, 0x69, 0xfd,
+	0x89, 0x00, 0xef, 0xa2, 0x9e, 0xdd, 0x5e, 0xfc, 0x09, 0x80, 0xed, 0x2e, 0x97, 0xd4, 0x96, 0x41,
+	0x84, 0x39, 0xa4, 0x2c, 0xc1, 0xad, 0xef, 0x53, 0x6f, 0xc9, 0xec, 0x4d, 0xf1, 0x4b, 0x24, 0x6d,
+	0xc2, 0x9f, 0x41, 0x85, 0x71, 0x53, 0xf8, 0x96, 0xc3, 0x19, 0x75, 0x84, 0xfc, 0xee, 0x9c, 0x92,
+	0x32, 0xe3, 0x93, 0xd8, 0xd4, 0xfe, 0x2d, 0x07, 0xf5, 0x31, 0xb5, 0xee, 0x29, 0x9d, 0x0d, 0xe3,
+	0xf4, 0xf0, 0x4f, 0x50, 0x4a, 0xbe, 0x46, 0xf8, 0xd3, 0x27, 0x3e, 0x53, 0xda, 0xc7, 0x07, 0xae,
+	0x2a, 0xfd, 0x45, 0x0b, 0x5d, 0x21, 0x7c, 0x03, 0x27, 0xd1, 0x42, 0xe0, 0xf3, 0x43, 0xd7, 0x89,
+	0x76, 0x71, 0x70, 0x8b, 0x22, 0xb5, 0x9f, 0xa1, 0x9a, 0x9d, 0x17, 0xfc, 0x32, 0x4b, 0xdb, 0x3b,
+	0xe2, 0xda, 0xe7, 0x87, 0x41, 0xb1, 0x0b, 0xec, 0xc3, 0xd9, 0xde, 0x01, 0xc1, 0x5b, 0xbf, 0x16,
+	0x87, 0x26, 0x53, 0xbb, 0x7c, 0x16, 0x36, 0xf6, 0xd9, 0xd5, 0xa1, 0xce, 0xc3, 0x2e, 0xbc, 0xe3,
+	0x86, 0xbd, 0x0c, 0x5a, 0xd3, 0xad, 0x26, 0x0d, 0x19, 0x05, 0xff, 0x52, 0xd3, 0xa2, 0xfc, 0xa5,
+	0xfa, 0xf6, 0xbf, 0x00, 0x00, 0x00, 0xff, 0xff, 0xa8, 0x6c, 0x82, 0x62, 0x65, 0x09, 0x00, 0x00,
 }
diff --git a/weed/util/log_buffer/log_buffer.go b/weed/util/log_buffer/log_buffer.go
index c7cb90549..d875dd54b 100644
--- a/weed/util/log_buffer/log_buffer.go
+++ b/weed/util/log_buffer/log_buffer.go
@@ -1,6 +1,7 @@
 package log_buffer
 
 import (
+	"bytes"
 	"sync"
 	"time"
 
@@ -17,7 +18,7 @@ const PreviousBufferCount = 3
 type dataToFlush struct {
 	startTime time.Time
 	stopTime  time.Time
-	data      []byte
+	data      *bytes.Buffer
 }
 
 type LogBuffer struct {
@@ -108,7 +109,8 @@ func (m *LogBuffer) Shutdown() {
 func (m *LogBuffer) loopFlush() {
 	for d := range m.flushChan {
 		if d != nil {
-			m.flushFn(d.startTime, d.stopTime, d.data)
+			m.flushFn(d.startTime, d.stopTime, d.data.Bytes())
+			d.releaseMemory()
 		}
 	}
 }
@@ -140,21 +142,26 @@ func (m *LogBuffer) copyToFlush() *dataToFlush {
 	return nil
 }
 
-func (m *LogBuffer) ReadFromBuffer(lastReadTime time.Time) (ts time.Time, bufferCopy []byte) {
+func (d *dataToFlush) releaseMemory() {
+	d.data.Reset()
+	bufferPool.Put(d.data)
+}
+
+func (m *LogBuffer) ReadFromBuffer(lastReadTime time.Time) (bufferCopy *bytes.Buffer) {
 	m.RLock()
 	defer m.RUnlock()
 
-	// fmt.Printf("read from buffer: %v\n", lastReadTime)
+	// fmt.Printf("read from buffer: %v last stop time: %v\n", lastReadTime.UnixNano(), m.stopTime.UnixNano())
 
 	if lastReadTime.Equal(m.stopTime) {
-		return lastReadTime, nil
+		return nil
 	}
 	if lastReadTime.After(m.stopTime) {
 		// glog.Fatalf("unexpected last read time %v, older than latest %v", lastReadTime, m.stopTime)
-		return lastReadTime, nil
+		return nil
 	}
 	if lastReadTime.Before(m.startTime) {
-		return m.stopTime, copiedBytes(m.buf[:m.pos])
+		return copiedBytes(m.buf[:m.pos])
 	}
 
 	lastTs := lastReadTime.UnixNano()
@@ -177,7 +184,7 @@ func (m *LogBuffer) ReadFromBuffer(lastReadTime time.Time) (ts time.Time, buffer
 	for l <= h {
 		mid := (l + h) / 2
 		pos := m.idx[mid]
-		_, t := readTs(m.buf, m.idx[mid])
+		_, t := readTs(m.buf, pos)
 		if t <= lastTs {
 			l = mid + 1
 		} else if lastTs < t {
@@ -186,22 +193,32 @@ func (m *LogBuffer) ReadFromBuffer(lastReadTime time.Time) (ts time.Time, buffer
 				_, prevT = readTs(m.buf, m.idx[mid-1])
 			}
 			if prevT <= lastTs {
-				// println("found mid = ", mid)
-				return time.Unix(0, t), copiedBytes(m.buf[pos:m.pos])
+				// fmt.Printf("found l=%d, m-1=%d(ts=%d), m=%d(ts=%d), h=%d [%d, %d) \n", l, mid-1, prevT, mid, t, h, pos, m.pos)
+				return copiedBytes(m.buf[pos:m.pos])
 			}
-			h = mid - 1
+			h = mid
 		}
 		// fmt.Printf("l=%d, h=%d\n", l, h)
 	}
 
 	// FIXME: this could be that the buffer has been flushed already
-	// println("not found")
-	return lastReadTime, nil
+	return nil
 
 }
-func copiedBytes(buf []byte) (copied []byte) {
-	copied = make([]byte, len(buf))
-	copy(copied, buf)
+func (m *LogBuffer) ReleaseMeory(b *bytes.Buffer) {
+	b.Reset()
+	bufferPool.Put(b)
+}
+
+var bufferPool = sync.Pool{
+	New: func() interface{} {
+		return new(bytes.Buffer)
+	},
+}
+
+func copiedBytes(buf []byte) (copied *bytes.Buffer) {
+	copied = bufferPool.Get().(*bytes.Buffer)
+	copied.Write(buf)
 	return
 }
 
diff --git a/weed/util/log_buffer/log_buffer_test.go b/weed/util/log_buffer/log_buffer_test.go
new file mode 100644
index 000000000..dfd611bed
--- /dev/null
+++ b/weed/util/log_buffer/log_buffer_test.go
@@ -0,0 +1,41 @@
+package log_buffer
+
+import (
+	"math/rand"
+	"testing"
+	"time"
+
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+)
+
+func TestNewLogBuffer(t *testing.T) {
+	lb := NewLogBuffer(time.Second, func(startTime, stopTime time.Time, buf []byte) {
+
+	}, func() {
+
+	})
+
+	startTime := time.Now()
+
+	messageSize := 1024
+	messageCount := 994
+	var buf = make([]byte, messageSize)
+	for i := 0; i < messageCount; i++ {
+		rand.Read(buf)
+		lb.AddToBuffer(nil, buf)
+	}
+
+	receivedmessageCount := 0
+	lb.LoopProcessLogData(startTime, func() bool {
+		// stop if no more messages
+		return false
+	}, func(logEntry *filer_pb.LogEntry) error {
+		receivedmessageCount++
+		return nil
+	})
+
+	if receivedmessageCount != messageCount {
+		t.Errorf("sent %d received %d", messageCount, receivedmessageCount)
+	}
+
+}
diff --git a/weed/util/log_buffer/log_read.go b/weed/util/log_buffer/log_read.go
new file mode 100644
index 000000000..4333df1b4
--- /dev/null
+++ b/weed/util/log_buffer/log_read.go
@@ -0,0 +1,74 @@
+package log_buffer
+
+import (
+	"bytes"
+	"time"
+
+	"github.com/golang/protobuf/proto"
+
+	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	"github.com/chrislusf/seaweedfs/weed/util"
+)
+
+func (logBuffer *LogBuffer) LoopProcessLogData(
+	startTreadTime time.Time, waitForDataFn func() bool, eachLogDataFn func(logEntry *filer_pb.LogEntry) error) (processed int64, err error) {
+	// loop through all messages
+	var bytesBuf *bytes.Buffer
+	lastReadTime := startTreadTime
+	defer func() {
+		if bytesBuf != nil {
+			logBuffer.ReleaseMeory(bytesBuf)
+		}
+	}()
+
+	for {
+
+		if bytesBuf != nil {
+			logBuffer.ReleaseMeory(bytesBuf)
+		}
+		bytesBuf = logBuffer.ReadFromBuffer(lastReadTime)
+		if bytesBuf == nil {
+			if waitForDataFn() {
+				continue
+			} else {
+				return
+			}
+		}
+
+		buf := bytesBuf.Bytes()
+
+		batchSize := 0
+		var startReadTime time.Time
+
+		for pos := 0; pos+4 < len(buf); {
+
+			size := util.BytesToUint32(buf[pos : pos+4])
+			entryData := buf[pos+4 : pos+4+int(size)]
+
+			// fmt.Printf("read buffer read %d [%d,%d) from [0,%d)\n", batchSize, pos, pos+int(size)+4, len(buf))
+
+			logEntry := &filer_pb.LogEntry{}
+			if err = proto.Unmarshal(entryData, logEntry); err != nil {
+				glog.Errorf("unexpected unmarshal messaging_pb.Message: %v", err)
+				pos += 4 + int(size)
+				continue
+			}
+			lastReadTime = time.Unix(0, logEntry.TsNs)
+			if startReadTime.IsZero() {
+				startReadTime = lastReadTime
+			}
+
+			if err = eachLogDataFn(logEntry); err != nil {
+				return
+			}
+
+			pos += 4 + int(size)
+			batchSize++
+			processed++
+		}
+
+		// fmt.Printf("sent message ts[%d,%d] size %d\n", startReadTime.UnixNano(), lastReadTime.UnixNano(), batchSize)
+	}
+
+}
\ No newline at end of file