1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
|
package main
import (
"bytes"
"context"
"errors"
"flag"
"fmt"
"io"
"math"
"os"
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/operation"
"github.com/seaweedfs/seaweedfs/weed/pb"
"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
"github.com/seaweedfs/seaweedfs/weed/security"
"github.com/seaweedfs/seaweedfs/weed/storage/idx"
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
"github.com/seaweedfs/seaweedfs/weed/storage/types"
"github.com/seaweedfs/seaweedfs/weed/util"
"google.golang.org/grpc"
)
var (
serversStr = flag.String("volumeServers", "", "comma-delimited list of volume servers to diff the volume against")
volumeId = flag.Int("volumeId", -1, "a volume id to diff from servers")
volumeCollection = flag.String("collection", "", "the volume collection name")
grpcDialOption grpc.DialOption
)
/*
Diff the volume's files across multiple volume servers.
diff_volume_servers -volumeServers 127.0.0.1:8080,127.0.0.1:8081 -volumeId 5
Example Output:
reference 127.0.0.1:8081
fileId volumeServer message
5,01617c3f61 127.0.0.1:8080 wrongSize
*/
func main() {
flag.Parse()
util.LoadConfiguration("security", false)
grpcDialOption = security.LoadClientTLS(util.GetViper(), "grpc.client")
vid := uint32(*volumeId)
servers := pb.ServerAddresses(*serversStr).ToAddresses()
if len(servers) < 2 {
glog.Fatalf("You must specify more than 1 server\n")
}
var referenceServer pb.ServerAddress
var maxOffset int64
allFiles := map[pb.ServerAddress]map[types.NeedleId]needleState{}
for _, addr := range servers {
files, offset, err := getVolumeFiles(vid, addr)
if err != nil {
glog.Fatalf("Failed to copy idx from volume server %s\n", err)
}
allFiles[addr] = files
if offset > maxOffset {
referenceServer = addr
}
}
same := true
fmt.Println("reference", referenceServer)
fmt.Println("fileId volumeServer message")
for nid, n := range allFiles[referenceServer] {
for addr, files := range allFiles {
if addr == referenceServer {
continue
}
var diffMsg string
n2, ok := files[nid]
if !ok {
if n.state == stateDeleted {
continue
}
diffMsg = "missing"
} else if n2.state != n.state {
switch n.state {
case stateDeleted:
diffMsg = "notDeleted"
case statePresent:
diffMsg = "deleted"
}
} else if n2.size != n.size {
diffMsg = "wrongSize"
} else {
continue
}
same = false
// fetch the needle details
var id string
var err error
if n.state == statePresent {
id, err = getNeedleFileId(vid, nid, referenceServer)
} else {
id, err = getNeedleFileId(vid, nid, addr)
}
if err != nil {
glog.Fatalf("Failed to get needle info %d from volume server %s\n", nid, err)
}
fmt.Println(id, addr, diffMsg)
}
}
if !same {
os.Exit(1)
}
}
const (
stateDeleted uint8 = 1
statePresent uint8 = 2
)
type needleState struct {
state uint8
size types.Size
}
func getVolumeFiles(v uint32, addr pb.ServerAddress) (map[types.NeedleId]needleState, int64, error) {
var idxFile *bytes.Reader
err := operation.WithVolumeServerClient(false, addr, grpcDialOption, func(vs volume_server_pb.VolumeServerClient) error {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
copyFileClient, err := vs.CopyFile(ctx, &volume_server_pb.CopyFileRequest{
VolumeId: v,
Ext: ".idx",
CompactionRevision: math.MaxUint32,
StopOffset: math.MaxInt64,
Collection: *volumeCollection,
})
if err != nil {
return err
}
var buf bytes.Buffer
for {
resp, err := copyFileClient.Recv()
if errors.Is(err, io.EOF) {
break
}
if err != nil {
return err
}
buf.Write(resp.FileContent)
}
idxFile = bytes.NewReader(buf.Bytes())
return nil
})
if err != nil {
return nil, 0, err
}
var maxOffset int64
files := map[types.NeedleId]needleState{}
err = idx.WalkIndexFile(idxFile, 0, func(key types.NeedleId, offset types.Offset, size types.Size) error {
if offset.IsZero() || size.IsDeleted() {
files[key] = needleState{
state: stateDeleted,
size: size,
}
} else {
files[key] = needleState{
state: statePresent,
size: size,
}
}
if actual := offset.ToActualOffset(); actual > maxOffset {
maxOffset = actual
}
return nil
})
if err != nil {
return nil, 0, err
}
return files, maxOffset, nil
}
func getNeedleFileId(v uint32, nid types.NeedleId, addr pb.ServerAddress) (string, error) {
var id string
err := operation.WithVolumeServerClient(false, addr, grpcDialOption, func(vs volume_server_pb.VolumeServerClient) error {
resp, err := vs.VolumeNeedleStatus(context.Background(), &volume_server_pb.VolumeNeedleStatusRequest{
VolumeId: v,
NeedleId: uint64(nid),
})
if err != nil {
return err
}
id = needle.NewFileId(needle.VolumeId(v), resp.NeedleId, resp.Cookie).String()
return nil
})
return id, err
}
|
