metadata log: read from any timestamp

weed/command/watch.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"time"
 
 	"github.com/chrislusf/seaweedfs/weed/pb"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
@@ -26,6 +27,7 @@ var cmdWatch = &Command{
 var (
 	watchFiler  = cmdWatch.Flag.String("filer", "localhost:8888", "filer hostname:port")
 	watchTarget = cmdWatch.Flag.String("pathPrefix", "/", "path to a folder or file, or common prefix for the folders or files on filer")
+	watchStart  = cmdWatch.Flag.Duration("timeAgo", 0, "start time before now. \"300ms\", \"1.5h\" or \"2h45m\". Valid time units are \"ns\", \"us\" (or \"µs\"), \"ms\", \"s\", \"m\", \"h\"")
 )
 
 func runWatch(cmd *Command, args []string) bool {
@@ -37,7 +39,7 @@ func runWatch(cmd *Command, args []string) bool {
 		stream, err := client.SubscribeMetadata(context.Background(), &filer_pb.SubscribeMetadataRequest{
 			ClientName: "watch",
 			PathPrefix: *watchTarget,
-			SinceNs:    0,
+			SinceNs:    time.Now().Add(-*watchStart).UnixNano(),
 		})
 		if err != nil {
 			return fmt.Errorf("listen: %v", err)

weed/filer2/filer_notify.go

@@ -1,7 +1,9 @@
 package filer2
 
 import (
+	"context"
 	"fmt"
+	"io"
 	"strings"
 	"time"
 
@@ -79,3 +81,76 @@ func (f *Filer) logFlushFunc(startTime, stopTime time.Time, buf []byte) {
 		glog.V(0).Infof("log write failed %s: %v", targetFile, err)
 	}
 }
+
+func (f *Filer) ReadPersistedLogBuffer(startTime time.Time, eachLogEntryFn func(logEntry *filer_pb.LogEntry) error) error {
+
+	startDate := fmt.Sprintf("%04d-%02d-%02d", startTime.Year(), startTime.Month(), startTime.Day())
+	startHourMinute := fmt.Sprintf("%02d-%02d.segment", startTime.Hour(), startTime.Minute())
+
+	sizeBuf := make([]byte, 4)
+	startTsNs := startTime.UnixNano()
+
+	dayEntries, listDayErr := f.ListDirectoryEntries(context.Background(), SystemLogDir, startDate, true, 366)
+	if listDayErr != nil {
+		return fmt.Errorf("fail to list log by day: %v", listDayErr)
+	}
+	for _, dayEntry := range dayEntries {
+		// println("checking day", dayEntry.FullPath)
+		hourMinuteEntries, listHourMinuteErr := f.ListDirectoryEntries(context.Background(), util.NewFullPath(SystemLogDir, dayEntry.Name()), "", false, 24*60)
+		if listHourMinuteErr != nil {
+			return fmt.Errorf("fail to list log %s by day: %v", dayEntry.Name(), listHourMinuteErr)
+		}
+		for _, hourMinuteEntry := range hourMinuteEntries {
+			// println("checking hh-mm", hourMinuteEntry.FullPath)
+			if dayEntry.Name() == startDate {
+				if strings.Compare(hourMinuteEntry.Name(), startHourMinute) < 0 {
+					continue
+				}
+			}
+			// println("processing", hourMinuteEntry.FullPath)
+			chunkedFileReader := NewChunkStreamReaderFromFiler(f.MasterClient, hourMinuteEntry.Chunks)
+			if err := readEachLogEntry(chunkedFileReader, sizeBuf, startTsNs, eachLogEntryFn); err != nil {
+				chunkedFileReader.Close()
+				if err == io.EOF {
+					break
+				}
+				return err
+			}
+			chunkedFileReader.Close()
+		}
+	}
+
+	return nil
+}
+
+func readEachLogEntry(r io.Reader, sizeBuf []byte, ns int64, eachLogEntryFn func(logEntry *filer_pb.LogEntry) error) error {
+	for {
+		n, err := r.Read(sizeBuf)
+		if err != nil {
+			return err
+		}
+		if n != 4 {
+			return fmt.Errorf("size %d bytes, expected 4 bytes", n)
+		}
+		size := util.BytesToUint32(sizeBuf)
+		entryData := make([]byte, size)
+		n, err = r.Read(entryData)
+		if err != nil {
+			return err
+		}
+		if n != int(size) {
+			return fmt.Errorf("entry data %d bytes, expected %d bytes", n, size)
+		}
+		logEntry := &filer_pb.LogEntry{}
+		if err = proto.Unmarshal(entryData, logEntry); err != nil {
+			return err
+		}
+		if logEntry.TsNs <= ns {
+			return nil
+		}
+		// println("each log: ", logEntry.TsNs)
+		if err := eachLogEntryFn(logEntry); err != nil {
+			return err
+		}
+	}
+}

weed/server/filer_grpc_server_listen.go

@@ -21,10 +21,7 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest,
 
 	defer fs.deleteClient(clientName)
 
-	lastReadTime := time.Now()
-	if req.SinceNs > 0 {
-		lastReadTime = time.Unix(0, req.SinceNs)
-	}
+	lastReadTime := time.Unix(0, req.SinceNs)
 
 	eachEventNotificationFn := func(dirPath string, eventNotification *filer_pb.EventNotification, tsNs int64) error {
 
@@ -59,12 +56,7 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest,
 		return nil
 	}
 
-	_, err := fs.filer.MetaLogBuffer.LoopProcessLogData(lastReadTime, func() bool {
-		fs.listenersLock.Lock()
-		fs.listenersCond.Wait()
-		fs.listenersLock.Unlock()
-		return true
-	}, func(logEntry *filer_pb.LogEntry) error {
+	eachLogEntryFn := func(logEntry *filer_pb.LogEntry) error {
 		event := &filer_pb.SubscribeMetadataResponse{}
 		if err := proto.Unmarshal(logEntry.Data, event); err != nil {
 			glog.Errorf("unexpected unmarshal filer_pb.SubscribeMetadataResponse: %v", err)
@@ -76,7 +68,18 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest,
 		}
 
 		return nil
-	})
+	}
+
+	if err := fs.filer.ReadPersistedLogBuffer(lastReadTime, eachLogEntryFn); err != nil {
+		return fmt.Errorf("reading from persisted logs: %v", err)
+	}
+
+	_, err := fs.filer.MetaLogBuffer.LoopProcessLogData(lastReadTime, func() bool {
+		fs.listenersLock.Lock()
+		fs.listenersCond.Wait()
+		fs.listenersLock.Unlock()
+		return true
+	}, eachLogEntryFn)
 
 	return err