package filer import ( "context" "fmt" "github.com/seaweedfs/seaweedfs/weed/util/log_buffer" "io" "math" "regexp" "strings" "time" "google.golang.org/protobuf/proto" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/notification" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/util" ) func (f *Filer) NotifyUpdateEvent(ctx context.Context, oldEntry, newEntry *Entry, deleteChunks, isFromOtherCluster bool, signatures []int32) { var fullpath string if oldEntry != nil { fullpath = string(oldEntry.FullPath) } else if newEntry != nil { fullpath = string(newEntry.FullPath) } else { return } // println("fullpath:", fullpath) if strings.HasPrefix(fullpath, SystemLogDir) { return } foundSelf := false for _, sig := range signatures { if sig == f.Signature { foundSelf = true } } if !foundSelf { signatures = append(signatures, f.Signature) } newParentPath := "" if newEntry != nil { newParentPath, _ = newEntry.FullPath.DirAndName() } eventNotification := &filer_pb.EventNotification{ OldEntry: oldEntry.ToProtoEntry(), NewEntry: newEntry.ToProtoEntry(), DeleteChunks: deleteChunks, NewParentPath: newParentPath, IsFromOtherCluster: isFromOtherCluster, Signatures: signatures, } if notification.Queue != nil { glog.V(3).Infof("notifying entry update %v", fullpath) if err := notification.Queue.SendMessage(fullpath, eventNotification); err != nil { // throw message glog.Error(err) } } f.logMetaEvent(ctx, fullpath, eventNotification) } func (f *Filer) logMetaEvent(ctx context.Context, fullpath string, eventNotification *filer_pb.EventNotification) { dir, _ := util.FullPath(fullpath).DirAndName() event := &filer_pb.SubscribeMetadataResponse{ Directory: dir, EventNotification: eventNotification, TsNs: time.Now().UnixNano(), } data, err := proto.Marshal(event) if err != nil { glog.Errorf("failed to marshal filer_pb.SubscribeMetadataResponse %+v: %v", event, err) return } f.LocalMetaLogBuffer.AddDataToBuffer([]byte(dir), data, event.TsNs) } func (f *Filer) logFlushFunc(logBuffer *log_buffer.LogBuffer, startTime, stopTime time.Time, buf []byte) { if len(buf) == 0 { return } startTime, stopTime = startTime.UTC(), stopTime.UTC() targetFile := fmt.Sprintf("%s/%04d-%02d-%02d/%02d-%02d.%08x", SystemLogDir, startTime.Year(), startTime.Month(), startTime.Day(), startTime.Hour(), startTime.Minute(), f.UniqueFilerId, // startTime.Second(), startTime.Nanosecond(), ) for { if err := f.appendToFile(targetFile, buf); err != nil { glog.V(0).Infof("metadata log write failed %s: %v", targetFile, err) time.Sleep(737 * time.Millisecond) } else { break } } } var ( VolumeNotFoundPattern = regexp.MustCompile(`volume \d+? not found`) ) func (f *Filer) ReadPersistedLogBuffer(startPosition log_buffer.MessagePosition, stopTsNs int64, eachLogEntryFn log_buffer.EachLogEntryFuncType) (lastTsNs int64, isDone bool, err error) { startDate := fmt.Sprintf("%04d-%02d-%02d", startPosition.Year(), startPosition.Month(), startPosition.Day()) startHourMinute := fmt.Sprintf("%02d-%02d", startPosition.Hour(), startPosition.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 := startPosition.UnixNano() dayEntries, _, listDayErr := f.ListDirectoryEntries(context.Background(), SystemLogDir, startDate, true, math.MaxInt32, "", "", "") if listDayErr != nil { return lastTsNs, isDone, 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 { return lastTsNs, isDone, 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 { hourMinute := util.FileNameBase(hourMinuteEntry.Name()) if strings.Compare(hourMinute, startHourMinute) < 0 { 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.GetChunks()) if lastTsNs, err = ReadEachLogEntry(chunkedFileReader, sizeBuf, startTsNs, stopTsNs, eachLogEntryFn); err != nil { chunkedFileReader.Close() if err == io.EOF { continue } if VolumeNotFoundPattern.MatchString(err.Error()) { glog.Warningf("skipping reading %s: %v", hourMinuteEntry.FullPath, err) continue } return lastTsNs, isDone, fmt.Errorf("reading %s: %v", hourMinuteEntry.FullPath, err) } chunkedFileReader.Close() } } return lastTsNs, isDone, nil } func ReadEachLogEntry(r io.Reader, sizeBuf []byte, startTsNs, stopTsNs int64, eachLogEntryFn log_buffer.EachLogEntryFuncType) (lastTsNs int64, err error) { for { n, err := r.Read(sizeBuf) if err != nil { return lastTsNs, err } if n != 4 { return lastTsNs, fmt.Errorf("size %d bytes, expected 4 bytes", n) } size := util.BytesToUint32(sizeBuf) // println("entry size", size) entryData := make([]byte, size) n, err = r.Read(entryData) if err != nil { return lastTsNs, err } if n != int(size) { return lastTsNs, fmt.Errorf("entry data %d bytes, expected %d bytes", n, size) } logEntry := &filer_pb.LogEntry{} if err = proto.Unmarshal(entryData, logEntry); err != nil { return lastTsNs, err } 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 } else { lastTsNs = logEntry.TsNs } } }