diff --git a/weed/command/fix.go b/weed/command/fix.go index f0499a818..b5016d961 100644 --- a/weed/command/fix.go +++ b/weed/command/fix.go @@ -31,6 +31,7 @@ var cmdFix = &Command{ var ( fixVolumeCollection = cmdFix.Flag.String("collection", "", "an optional volume collection name, if specified only it will be processed") fixVolumeId = cmdFix.Flag.Int64("volumeId", 0, "an optional volume id, if not 0 (default) only it will be processed") + fixIgnoreError = cmdFix.Flag.Bool("ignoreError", false, "an optional, if true will be processed despite errors") ) type VolumeFileScanner4Fix struct { @@ -126,11 +127,21 @@ func doFixOneVolume(basepath string, baseFileName string, collection string, vol } if err := storage.ScanVolumeFile(basepath, collection, vid, storage.NeedleMapInMemory, scanner); err != nil { - glog.Fatalf("scan .dat File: %v", err) + err := fmt.Errorf("scan .dat File: %v", err) + if *fixIgnoreError { + glog.Error(err) + } else { + glog.Fatal(err) + } } if err := nm.SaveToIdx(indexFileName); err != nil { os.Remove(indexFileName) - glog.Fatalf("save to .idx File: %v", err) + err := fmt.Errorf("save to .idx File: %v", err) + if *fixIgnoreError { + glog.Error(err) + } else { + glog.Fatal(err) + } } } diff --git a/weed/shell/command_volume_check_disk.go b/weed/shell/command_volume_check_disk.go index 7195318fb..41c28b810 100644 --- a/weed/shell/command_volume_check_disk.go +++ b/weed/shell/command_volume_check_disk.go @@ -13,6 +13,7 @@ import ( "google.golang.org/grpc" "io" "math" + "time" ) func init() { @@ -121,6 +122,7 @@ func (c *commandVolumeCheckDisk) checkBoth(a *VolumeReplica, b *VolumeReplica, a }() // read index db + readIndexDbCutoffFrom := uint64(time.Now().UnixNano()) if err = c.readIndexDatabase(aDB, a.info.Collection, a.info.Id, pb.NewServerAddressFromDataNode(a.location.dataNode), verbose, writer); err != nil { return true, true, fmt.Errorf("readIndexDatabase %s volume %d: %v", a.location.dataNode, a.info.Id, err) } @@ -129,25 +131,37 @@ func (c *commandVolumeCheckDisk) checkBoth(a *VolumeReplica, b *VolumeReplica, a } // find and make up the differences - if aHasChanges, err = c.doVolumeCheckDisk(bDB, aDB, b, a, verbose, writer, applyChanges, nonRepairThreshold); err != nil { + if aHasChanges, err = c.doVolumeCheckDisk(bDB, aDB, b, a, verbose, writer, applyChanges, nonRepairThreshold, readIndexDbCutoffFrom); err != nil { return true, true, fmt.Errorf("doVolumeCheckDisk source:%s target:%s volume %d: %v", b.location.dataNode.Id, a.location.dataNode.Id, b.info.Id, err) } - if bHasChanges, err = c.doVolumeCheckDisk(aDB, bDB, a, b, verbose, writer, applyChanges, nonRepairThreshold); err != nil { + if bHasChanges, err = c.doVolumeCheckDisk(aDB, bDB, a, b, verbose, writer, applyChanges, nonRepairThreshold, readIndexDbCutoffFrom); err != nil { return true, true, fmt.Errorf("doVolumeCheckDisk source:%s target:%s volume %d: %v", a.location.dataNode.Id, b.location.dataNode.Id, a.info.Id, err) } return } -func (c *commandVolumeCheckDisk) doVolumeCheckDisk(minuend, subtrahend *needle_map.MemDb, source, target *VolumeReplica, verbose bool, writer io.Writer, applyChanges bool, nonRepairThreshold float64) (hasChanges bool, err error) { +func (c *commandVolumeCheckDisk) doVolumeCheckDisk(minuend, subtrahend *needle_map.MemDb, source, target *VolumeReplica, verbose bool, writer io.Writer, applyChanges bool, nonRepairThreshold float64, cutoffFromAtNs uint64) (hasChanges bool, err error) { // find missing keys // hash join, can be more efficient var missingNeedles []needle_map.NeedleValue var counter int - minuend.AscendingVisit(func(value needle_map.NeedleValue) error { + doCutoffOfLastNeedle := true + minuend.DescendingVisit(func(value needle_map.NeedleValue) error { counter++ - if _, found := subtrahend.Get(value.Key); !found && value.Size.IsValid() { + if _, found := subtrahend.Get(value.Key); !found { + if doCutoffOfLastNeedle { + if needleMeta, err := readNeedleMeta(c.env.option.GrpcDialOption, pb.NewServerAddressFromDataNode(source.location.dataNode), source.info.Id, value); err == nil { + // needles older than the cutoff time are not missing yet + if needleMeta.AppendAtNs > cutoffFromAtNs { + return nil + } + doCutoffOfLastNeedle = false + } + } missingNeedles = append(missingNeedles, value) + } else if doCutoffOfLastNeedle { + doCutoffOfLastNeedle = false } return nil }) @@ -166,7 +180,6 @@ func (c *commandVolumeCheckDisk) doVolumeCheckDisk(minuend, subtrahend *needle_m } for _, needleValue := range missingNeedles { - needleBlob, err := readSourceNeedleBlob(c.env.option.GrpcDialOption, pb.NewServerAddressFromDataNode(source.location.dataNode), source.info.Id, needleValue) if err != nil { return hasChanges, err diff --git a/weed/shell/commands.go b/weed/shell/commands.go index 87aeed2bb..66fdcb6bd 100644 --- a/weed/shell/commands.go +++ b/weed/shell/commands.go @@ -1,7 +1,11 @@ package shell import ( + "context" "fmt" + "github.com/seaweedfs/seaweedfs/weed/operation" + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/storage/needle_map" "io" "net/url" "strconv" @@ -148,3 +152,20 @@ func findInputDirectory(args []string) (input string) { } return input } + +func readNeedleMeta(grpcDialOption grpc.DialOption, volumeServer pb.ServerAddress, volumeId uint32, needleValue needle_map.NeedleValue) (resp *volume_server_pb.ReadNeedleMetaResponse, err error) { + err = operation.WithVolumeServerClient(false, volumeServer, grpcDialOption, + func(client volume_server_pb.VolumeServerClient) error { + if resp, err = client.ReadNeedleMeta(context.Background(), &volume_server_pb.ReadNeedleMetaRequest{ + VolumeId: volumeId, + NeedleId: uint64(needleValue.Key), + Offset: needleValue.Offset.ToActualOffset(), + Size: int32(needleValue.Size), + }); err != nil { + return err + } + return nil + }, + ) + return +} diff --git a/weed/storage/needle_map/memdb.go b/weed/storage/needle_map/memdb.go index 463245cd1..f2b161792 100644 --- a/weed/storage/needle_map/memdb.go +++ b/weed/storage/needle_map/memdb.go @@ -6,6 +6,7 @@ import ( "os" "github.com/syndtr/goleveldb/leveldb" + "github.com/syndtr/goleveldb/leveldb/iterator" "github.com/syndtr/goleveldb/leveldb/opt" "github.com/syndtr/goleveldb/leveldb/storage" @@ -61,17 +62,47 @@ func (cm *MemDb) Get(key NeedleId) (*NeedleValue, bool) { } // Visit visits all entries or stop if any error when visiting +func doVisit(iter iterator.Iterator, visit func(NeedleValue) error) (ret error) { + key := BytesToNeedleId(iter.Key()) + data := iter.Value() + offset := BytesToOffset(data[0:OffsetSize]) + size := BytesToSize(data[OffsetSize : OffsetSize+SizeSize]) + + needle := NeedleValue{Key: key, Offset: offset, Size: size} + ret = visit(needle) + if ret != nil { + return + } + return nil +} + func (cm *MemDb) AscendingVisit(visit func(NeedleValue) error) (ret error) { iter := cm.db.NewIterator(nil, nil) + if iter.First() { + if ret = doVisit(iter, visit); ret != nil { + return + } + } for iter.Next() { - key := BytesToNeedleId(iter.Key()) - data := iter.Value() - offset := BytesToOffset(data[0:OffsetSize]) - size := BytesToSize(data[OffsetSize : OffsetSize+SizeSize]) + if ret = doVisit(iter, visit); ret != nil { + return + } + } + iter.Release() + ret = iter.Error() - needle := NeedleValue{Key: key, Offset: offset, Size: size} - ret = visit(needle) - if ret != nil { + return +} + +func (cm *MemDb) DescendingVisit(visit func(NeedleValue) error) (ret error) { + iter := cm.db.NewIterator(nil, nil) + if iter.Last() { + if ret = doVisit(iter, visit); ret != nil { + return + } + } + for iter.Prev() { + if ret = doVisit(iter, visit); ret != nil { return } }