diff options
| -rw-r--r-- | weed/command/watch.go | 4 | ||||
| -rw-r--r-- | weed/filer2/filer_notify.go | 75 | ||||
| -rw-r--r-- | weed/server/filer_grpc_server_listen.go | 25 |
3 files changed, 92 insertions, 12 deletions
diff --git a/weed/command/watch.go b/weed/command/watch.go index 966040fbb..b46707a62 100644 --- a/weed/command/watch.go +++ b/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) diff --git a/weed/filer2/filer_notify.go b/weed/filer2/filer_notify.go index a55e1724b..e48026b52 100644 --- a/weed/filer2/filer_notify.go +++ b/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 + } + } +} diff --git a/weed/server/filer_grpc_server_listen.go b/weed/server/filer_grpc_server_listen.go index 1da257202..2a18f1950 100644 --- a/weed/server/filer_grpc_server_listen.go +++ b/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 |