subscribe metadata between a range

1 files changed, 25 insertions, 5 deletions

diff --git a/weed/filer/filer_notify.go b/weed/filer/filer_notify.go
index 7857d79af..9f39d26d5 100644
--- a/weed/filer/filer_notify.go
+++ b/weed/filer/filer_notify.go
@@ -108,20 +108,31 @@ func (f *Filer) logFlushFunc(startTime, stopTime time.Time, buf []byte) {
 	}
 }
 
-func (f *Filer) ReadPersistedLogBuffer(startTime time.Time, eachLogEntryFn func(logEntry *filer_pb.LogEntry) error) (lastTsNs int64, err error) {
+func (f *Filer) ReadPersistedLogBuffer(startTime time.Time, stopTsNs int64, eachLogEntryFn func(logEntry *filer_pb.LogEntry) error) (lastTsNs int64, err error) {
 
 	startTime = startTime.UTC()
 	startDate := fmt.Sprintf("%04d-%02d-%02d", startTime.Year(), startTime.Month(), startTime.Day())
 	startHourMinute := fmt.Sprintf("%02d-%02d", startTime.Hour(), startTime.Minute())
+	var stopDate, stopHourMinute string
+	if stopTsNs != 0 {
+		stopTime := time.Unix(0, stopTsNs+24*60*60*int64(time.Nanosecond)).UTC()
+		stopDate = fmt.Sprintf("%04d-%02d-%02d", stopTime.Year(), stopTime.Month(), stopTime.Day())
+		stopHourMinute = fmt.Sprintf("%02d-%02d", stopTime.Hour(), stopTime.Minute())
+	}
 
 	sizeBuf := make([]byte, 4)
 	startTsNs := startTime.UnixNano()
 
-	dayEntries, _, listDayErr := f.ListDirectoryEntries(context.Background(), SystemLogDir, startDate, true, 366, "", "", "")
+	dayEntries, _, listDayErr := f.ListDirectoryEntries(context.Background(), SystemLogDir, startDate, true, math.MaxInt32, "", "", "")
 	if listDayErr != nil {
 		return lastTsNs, fmt.Errorf("fail to list log by day: %v", listDayErr)
 	}
 	for _, dayEntry := range dayEntries {
+		if stopDate != "" {
+			if strings.Compare(dayEntry.Name(), stopDate) > 0 {
+				break
+			}
+		}
 		// println("checking day", dayEntry.FullPath)
 		hourMinuteEntries, _, listHourMinuteErr := f.ListDirectoryEntries(context.Background(), util.NewFullPath(SystemLogDir, dayEntry.Name()), "", false, math.MaxInt32, "", "", "")
 		if listHourMinuteErr != nil {
@@ -135,9 +146,15 @@ func (f *Filer) ReadPersistedLogBuffer(startTime time.Time, eachLogEntryFn func(
 					continue
 				}
 			}
+			if dayEntry.Name() == stopDate {
+				hourMinute := util.FileNameBase(hourMinuteEntry.Name())
+				if strings.Compare(hourMinute, stopHourMinute) > 0 {
+					break
+				}
+			}
 			// println("processing", hourMinuteEntry.FullPath)
 			chunkedFileReader := NewChunkStreamReaderFromFiler(f.MasterClient, hourMinuteEntry.Chunks)
-			if lastTsNs, err = ReadEachLogEntry(chunkedFileReader, sizeBuf, startTsNs, eachLogEntryFn); err != nil {
+			if lastTsNs, err = ReadEachLogEntry(chunkedFileReader, sizeBuf, startTsNs, stopTsNs, eachLogEntryFn); err != nil {
 				chunkedFileReader.Close()
 				if err == io.EOF {
 					continue
@@ -151,7 +168,7 @@ func (f *Filer) ReadPersistedLogBuffer(startTime time.Time, eachLogEntryFn func(
 	return lastTsNs, nil
 }
 
-func ReadEachLogEntry(r io.Reader, sizeBuf []byte, ns int64, eachLogEntryFn func(logEntry *filer_pb.LogEntry) error) (lastTsNs int64, err error) {
+func ReadEachLogEntry(r io.Reader, sizeBuf []byte, startTsNs, stopTsNs int64, eachLogEntryFn func(logEntry *filer_pb.LogEntry) error) (lastTsNs int64, err error) {
 	for {
 		n, err := r.Read(sizeBuf)
 		if err != nil {
@@ -174,9 +191,12 @@ func ReadEachLogEntry(r io.Reader, sizeBuf []byte, ns int64, eachLogEntryFn func
 		if err = proto.Unmarshal(entryData, logEntry); err != nil {
 			return lastTsNs, err
 		}
-		if logEntry.TsNs <= ns {
+		if logEntry.TsNs <= startTsNs {
 			continue
 		}
+		if stopTsNs != 0 && logEntry.TsNs > stopTsNs {
+			return lastTsNs, err
+		}
 		// println("each log: ", logEntry.TsNs)
 		if err := eachLogEntryFn(logEntry); err != nil {
 			return lastTsNs, err