aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--weed/command/fix.go15
-rw-r--r--weed/shell/command_volume_check_disk.go25
-rw-r--r--weed/shell/commands.go21
-rw-r--r--weed/storage/needle_map/memdb.go47
4 files changed, 92 insertions, 16 deletions
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])
-
- needle := NeedleValue{Key: key, Offset: offset, Size: size}
- ret = visit(needle)
- if ret != nil {
+ if ret = doVisit(iter, visit); ret != nil {
+ return
+ }
+ }
+ iter.Release()
+ ret = iter.Error()
+
+ 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
}
}
generated by cgit v1.2.3 (git 2.25.1) at 2025-12-21 16:48:24 +0000