logging changes to debug

unmaintained/load_test/load_test_meta_tail/load_test_meta_tail.go

@@ -4,18 +4,20 @@ import (
 	"context"
 	"flag"
 	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"google.golang.org/grpc"
 	"io"
 	"strconv"
+	"time"
 )
 
 var (
 	dir       = flag.String("dir", "/tmp", "directory to create files")
 	n         = flag.Int("n", 100, "the number of metadata")
 	tailFiler = flag.String("filer", "localhost:8888", "the filer address")
-	isWrite = flag.Bool("write", false, "only write")
+	isWrite   = flag.Bool("write", false, "only write")
 )
 
 func main() {
@@ -33,7 +35,7 @@ func main() {
 			return nil
 		}
 		name := event.EventNotification.NewEntry.Name
-		fmt.Printf("=> %s\n", name)
+		glog.V(0).Infof("=> %s ts:%+v", name, time.Unix(0, event.TsNs))
 		id := name[4:]
 		if x, err := strconv.Atoi(id); err == nil {
 			if x != expected {
@@ -43,6 +45,7 @@ func main() {
 		} else {
 			return err
 		}
+		time.Sleep(10 * time.Millisecond)
 		return nil
 	})
 
@@ -96,6 +99,7 @@ func startSubscribeMetadata(eachEntryFunc func(event *filer_pb.SubscribeMetadata
 				return listenErr
 			}
 			if err = eachEntryFunc(resp); err != nil {
+				glog.V(0).Infof("tail last record:%+v", time.Unix(0, lastTsNs))
 				return err
 			}
 			lastTsNs = resp.TsNs

weed/filer/filer.go

@@ -53,7 +53,7 @@ func NewFiler(masters []string, grpcDialOption grpc.DialOption,
 		GrpcDialOption:      grpcDialOption,
 		FilerConf:           NewFilerConf(),
 	}
-	f.LocalMetaLogBuffer = log_buffer.NewLogBuffer(LogFlushInterval, f.logFlushFunc, notifyFn)
+	f.LocalMetaLogBuffer = log_buffer.NewLogBuffer("local", LogFlushInterval, f.logFlushFunc, notifyFn)
 	f.metaLogCollection = collection
 	f.metaLogReplication = replication
 

weed/filer/meta_aggregator.go

@@ -34,7 +34,7 @@ func NewMetaAggregator(filers []string, grpcDialOption grpc.DialOption) *MetaAgg
 		grpcDialOption: grpcDialOption,
 	}
 	t.ListenersCond = sync.NewCond(&t.ListenersLock)
-	t.MetaLogBuffer = log_buffer.NewLogBuffer(LogFlushInterval, nil, func() {
+	t.MetaLogBuffer = log_buffer.NewLogBuffer("aggr", LogFlushInterval, nil, func() {
 		t.ListenersCond.Broadcast()
 	})
 	return t
@@ -118,6 +118,7 @@ func (ma *MetaAggregator) subscribeToOneFiler(f *Filer, self string, peer string
 	}
 
 	for {
+		glog.V(4).Infof("subscribing remote %s meta change: %v", peer, time.Unix(0, lastTsNs))
 		err := pb.WithFilerClient(peer, ma.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
 			ctx, cancel := context.WithCancel(context.Background())
 			defer cancel()

weed/messaging/broker/broker_grpc_server_subscribe.go

@@ -116,7 +116,7 @@ func (broker *MessageBroker) Subscribe(stream messaging_pb.SeaweedMessaging_Subs
 			lastReadTime = time.Unix(0, processedTsNs)
 		}
 
-		lastReadTime, err = lock.logBuffer.LoopProcessLogData(lastReadTime, func() bool {
+		lastReadTime, err = lock.logBuffer.LoopProcessLogData("broker", lastReadTime, func() bool {
 			lock.Mutex.Lock()
 			lock.cond.Wait()
 			lock.Mutex.Unlock()

weed/messaging/broker/topic_manager.go

@@ -68,7 +68,7 @@ func (tm *TopicManager) buildLogBuffer(tl *TopicControl, tp TopicPartition, topi
 			glog.V(0).Infof("log write failed %s: %v", targetFile, err)
 		}
 	}
-	logBuffer := log_buffer.NewLogBuffer(time.Minute, flushFn, func() {
+	logBuffer := log_buffer.NewLogBuffer("broker", time.Minute, flushFn, func() {
 		tl.cond.Broadcast()
 	})
 

weed/server/filer_grpc_server_sub_meta.go

@@ -34,7 +34,7 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest,
 
 	for {
 
-		glog.V(0).Infof("read on disk %v aggregated subscribe %s from %+v", clientName, req.PathPrefix, lastReadTime)
+		glog.V(4).Infof("read on disk %v aggregated subscribe %s from %+v", clientName, req.PathPrefix, lastReadTime)
 
 		processedTsNs, err = fs.filer.ReadPersistedLogBuffer(lastReadTime, eachLogEntryFn)
 		if err != nil {
@@ -45,9 +45,9 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest,
 			lastReadTime = time.Unix(0, processedTsNs)
 		}
 
-		glog.V(0).Infof("read in memory %v aggregated subscribe %s from %+v", clientName, req.PathPrefix, lastReadTime)
+		glog.V(4).Infof("read in memory %v aggregated subscribe %s from %+v", clientName, req.PathPrefix, lastReadTime)
 
-		lastReadTime, err = fs.filer.MetaAggregator.MetaLogBuffer.LoopProcessLogData(lastReadTime, func() bool {
+		lastReadTime, err = fs.filer.MetaAggregator.MetaLogBuffer.LoopProcessLogData("aggMeta:"+clientName, lastReadTime, func() bool {
 			fs.filer.MetaAggregator.ListenersLock.Lock()
 			fs.filer.MetaAggregator.ListenersCond.Wait()
 			fs.filer.MetaAggregator.ListenersLock.Unlock()
@@ -101,7 +101,7 @@ func (fs *FilerServer) SubscribeLocalMetadata(req *filer_pb.SubscribeMetadataReq
 
 		glog.V(0).Infof("read in memory %v local subscribe %s from %+v", clientName, req.PathPrefix, lastReadTime)
 
-		lastReadTime, err = fs.filer.LocalMetaLogBuffer.LoopProcessLogData(lastReadTime, func() bool {
+		lastReadTime, err = fs.filer.LocalMetaLogBuffer.LoopProcessLogData("localMeta:"+clientName, lastReadTime, func() bool {
 			fs.listenersLock.Lock()
 			fs.listenersCond.Wait()
 			fs.listenersLock.Unlock()

weed/util/log_buffer/log_buffer.go

@@ -22,6 +22,7 @@ type dataToFlush struct {
 }
 
 type LogBuffer struct {
+	name          string
 	prevBuffers   *SealedBuffers
 	buf           []byte
 	idx           []int
@@ -39,8 +40,9 @@ type LogBuffer struct {
 	sync.RWMutex
 }
 
-func NewLogBuffer(flushInterval time.Duration, flushFn func(startTime, stopTime time.Time, buf []byte), notifyFn func()) *LogBuffer {
+func NewLogBuffer(name string, flushInterval time.Duration, flushFn func(startTime, stopTime time.Time, buf []byte), notifyFn func()) *LogBuffer {
 	lb := &LogBuffer{
+		name:          name,
 		prevBuffers:   newSealedBuffers(PreviousBufferCount),
 		buf:           make([]byte, BufferSize),
 		sizeBuf:       make([]byte, 4),
@@ -93,6 +95,7 @@ func (m *LogBuffer) AddToBuffer(partitionKey, data []byte, eventTsNs int64) {
 	}
 
 	if m.startTime.Add(m.flushInterval).Before(ts) || len(m.buf)-m.pos < size+4 {
+		// glog.V(4).Infof("%s copyToFlush1 start time %v, ts %v, remaining %d bytes", m.name, m.startTime, ts, len(m.buf)-m.pos)
 		m.flushChan <- m.copyToFlush()
 		m.startTime = ts
 		if len(m.buf) < size+4 {
@@ -127,9 +130,10 @@ func (m *LogBuffer) Shutdown() {
 func (m *LogBuffer) loopFlush() {
 	for d := range m.flushChan {
 		if d != nil {
-			// fmt.Printf("flush [%v, %v] size %d\n", d.startTime, d.stopTime, len(d.data.Bytes()))
+			// glog.V(4).Infof("%s flush [%v, %v] size %d", m.name, d.startTime, d.stopTime, len(d.data.Bytes()))
 			m.flushFn(d.startTime, d.stopTime, d.data.Bytes())
 			d.releaseMemory()
+			// local logbuffer is different from aggregate logbuffer here
 			m.lastFlushTime = d.stopTime
 		}
 	}
@@ -143,7 +147,6 @@ func (m *LogBuffer) loopInterval() {
 			m.Unlock()
 			return
 		}
-		// println("loop interval")
 		toFlush := m.copyToFlush()
 		m.flushChan <- toFlush
 		m.Unlock()
@@ -162,7 +165,6 @@ func (m *LogBuffer) copyToFlush() *dataToFlush {
 				data:      copiedBytes(m.buf[:m.pos]),
 			}
 		}
-		// fmt.Printf("flusing [0,%d) with %d entries\n", m.pos, len(m.idx))
 		m.buf = m.prevBuffers.SealBuffer(m.startTime, m.stopTime, m.buf, m.pos)
 		m.pos = 0
 		m.idx = m.idx[:0]
@@ -200,12 +202,9 @@ func (m *LogBuffer) ReadFromBuffer(lastReadTime time.Time) (bufferCopy *bytes.Bu
 	}
 	if lastReadTime.Before(m.startTime) {
 		// println("checking ", lastReadTime.UnixNano())
-		for i, buf := range m.prevBuffers.buffers {
+		for _, buf := range m.prevBuffers.buffers {
 			if buf.startTime.After(lastReadTime) {
-				if i == 0 {
+				// glog.V(4).Infof("%s return the %d sealed buffer %v", m.name, i, buf.startTime)
-					// println("return the earliest in memory", buf.startTime.UnixNano())
-					return copiedBytes(buf.buf[:buf.size]), nil
-				}
 				// println("return the", i, "th in memory", buf.startTime.UnixNano())
 				return copiedBytes(buf.buf[:buf.size]), nil
 			}
@@ -215,7 +214,7 @@ func (m *LogBuffer) ReadFromBuffer(lastReadTime time.Time) (bufferCopy *bytes.Bu
 				return copiedBytes(buf.buf[pos:buf.size]), nil
 			}
 		}
-		// println("return the current buf", lastReadTime.UnixNano())
+		// glog.V(4).Infof("%s return the current buf %v", m.name, lastReadTime)
 		return copiedBytes(m.buf[:m.pos]), nil
 	}
 

weed/util/log_buffer/log_buffer_test.go

@@ -10,7 +10,7 @@ import (
 )
 
 func TestNewLogBufferFirstBuffer(t *testing.T) {
-	lb := NewLogBuffer(time.Minute, func(startTime, stopTime time.Time, buf []byte) {
+	lb := NewLogBuffer("test", time.Minute, func(startTime, stopTime time.Time, buf []byte) {
 
 	}, func() {
 
@@ -27,7 +27,7 @@ func TestNewLogBufferFirstBuffer(t *testing.T) {
 	}
 
 	receivedmessageCount := 0
-	lb.LoopProcessLogData(startTime, func() bool {
+	lb.LoopProcessLogData("test", startTime, func() bool {
 		// stop if no more messages
 		return false
 	}, func(logEntry *filer_pb.LogEntry) error {

weed/util/log_buffer/log_read.go

@@ -17,10 +17,7 @@ var (
 	ResumeFromDiskError = fmt.Errorf("resumeFromDisk")
 )
 
-func (logBuffer *LogBuffer) LoopProcessLogData(
+func (logBuffer *LogBuffer) LoopProcessLogData(readerName string, startTreadTime time.Time, waitForDataFn func() bool, eachLogDataFn func(logEntry *filer_pb.LogEntry) error) (lastReadTime time.Time, err error) {
-	startTreadTime time.Time,
-	waitForDataFn func() bool,
-	eachLogDataFn func(logEntry *filer_pb.LogEntry) error) (lastReadTime time.Time, err error) {
 	// loop through all messages
 	var bytesBuf *bytes.Buffer
 	lastReadTime = startTreadTime
@@ -39,7 +36,7 @@ func (logBuffer *LogBuffer) LoopProcessLogData(
 		if err == ResumeFromDiskError {
 			return lastReadTime, ResumeFromDiskError
 		}
-		// fmt.Printf("ReadFromBuffer by %v\n", lastReadTime)
+		// glog.V(4).Infof("%s ReadFromBuffer by %v", readerName, lastReadTime)
 		if bytesBuf == nil {
 			if waitForDataFn() {
 				continue
@@ -49,7 +46,7 @@ func (logBuffer *LogBuffer) LoopProcessLogData(
 		}
 
 		buf := bytesBuf.Bytes()
-		// fmt.Printf("ReadFromBuffer by %v size %d\n", lastReadTime, len(buf))
+		// fmt.Printf("ReadFromBuffer %s by %v size %d\n", readerName, lastReadTime, len(buf))
 
 		batchSize := 0
 		var startReadTime time.Time
@@ -59,7 +56,7 @@ func (logBuffer *LogBuffer) LoopProcessLogData(
 			size := util.BytesToUint32(buf[pos : pos+4])
 			if pos+4+int(size) > len(buf) {
 				err = ResumeError
-				glog.Errorf("LoopProcessLogData: read buffer %v read %d [%d,%d) from [0,%d)", lastReadTime, batchSize, pos, pos+int(size)+4, len(buf))
+				glog.Errorf("LoopProcessLogData: %s read buffer %v read %d [%d,%d) from [0,%d)", readerName, lastReadTime, batchSize, pos, pos+int(size)+4, len(buf))
 				return
 			}
 			entryData := buf[pos+4 : pos+4+int(size)]
@@ -81,9 +78,10 @@ func (logBuffer *LogBuffer) LoopProcessLogData(
 
 			pos += 4 + int(size)
 			batchSize++
+
 		}
 
-		// fmt.Printf("sent message ts[%d,%d] size %d\n", startReadTime.UnixNano(), lastReadTime.UnixNano(), batchSize)
+		// glog.V(4).Infof("%s sent messages ts[%+v,%+v] size %d\n", readerName, startReadTime, lastReadTime, batchSize)
 	}
 
 }