Github-Mirror/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-09-08 15:41:15 +02:00
Code Issues Packages Projects Releases Wiki Activity

expose whether a volume replica is backed by remote storage

Browse Source 
Volume locations returned by lookups do not indicate whether a replica
has been tiered to remote storage. Readers cannot distinguish a local
replica from a remote-backed one, so they may hit a remote-backed
replica first even when a local replica is available.

Add DataInRemote to the lookup location message, populate it from the
master's volume info, and carry it through the wdclient vid map so
clients can prefer local replicas when resolving chunk locations.
This commit is contained in:
Bruce Zou
2026-09-02 17:51:52 +08:00
committed by bruce-zzz
parent 68a175ef6f
commit 554aa13c5b
18 changed files with 238 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
other/java/client/src/main/proto/filer.proto
+2
View File
@@ -576,6 +576,7 @@ message Location {
string public_url = 2;
uint32 grpc_port = 3;
string data_center = 4;
bool data_in_remote = 5;
}
message LookupVolumeResponse {
map<string, Locations> locations_map = 1;
@@ -662,6 +663,7 @@ message SubscribeMetadataResponse {
int64 ts_ns = 3;
repeated SubscribeMetadataResponse events = 4; // batch of additional events (backlog catch-up)
repeated LogFileChunkRef log_file_refs = 5; // log file chunk refs for client direct-read
int64 flushed_ts_ns = 6; // local log-buffer flush watermark: everything at or below it is on disk
}
message ListMetadataSubscribersRequest {
repeated string client_types = 1; // optional filter by client type, e.g. "mount"; empty = all
seaweed-volume/proto/master.proto
+2
View File
@@ -222,6 +222,7 @@ message VolumeLocation {
uint32 grpc_port = 7;
repeated uint32 new_ec_vids = 8;
repeated uint32 deleted_ec_vids = 9;
repeated uint32 remote_vids = 10;
}
message ClusterNodeUpdate {
@@ -267,6 +268,7 @@ message Location {
string public_url = 2;
uint32 grpc_port = 3;
string data_center = 4;
bool data_in_remote = 5;
}
message AssignRequest {
weed/filer/reader_at.go
+10
View File
@@ -117,9 +117,15 @@ func LookupFn(filerClient filer_pb.FilerClient) wdclient.LookupFileIdFunctionTyp
fcDataCenter := filerClient.GetDataCenter()
var sameDcTargetUrls, otherTargetUrls []string
localUrls := make(map[string]bool)
for _, loc := range locations.Locations {
volumeServerAddress := filerClient.AdjustedUrl(loc)
targetUrl := fmt.Sprintf("http://%s/%s", volumeServerAddress, fileId)
glog.V(4).Infof("lookup %s => %s, data in remote storage tier: %v", fileId, targetUrl, loc.DataInRemote)
if !loc.DataInRemote {
localUrls[targetUrl] = true
}
if fcDataCenter == "" || fcDataCenter != loc.DataCenter {
otherTargetUrls = append(otherTargetUrls, targetUrl)
} else {
@@ -132,6 +138,10 @@ func LookupFn(filerClient filer_pb.FilerClient) wdclient.LookupFileIdFunctionTyp
rand.Shuffle(len(otherTargetUrls), func(i, j int) {
otherTargetUrls[i], otherTargetUrls[j] = otherTargetUrls[j], otherTargetUrls[i]
})
if len(localUrls) > 0 {
sameDcTargetUrls = util.ReorderToFront(localUrls, sameDcTargetUrls)
otherTargetUrls = util.ReorderToFront(localUrls, otherTargetUrls)
}
// Prefer same data center
targetUrls = append(sameDcTargetUrls, otherTargetUrls...)
return
weed/operation/lookup.go
+26 -9
View File
@@ -15,10 +15,11 @@ import (
)
type Location struct {
Url string `json:"url,omitempty"`
PublicUrl string `json:"publicUrl,omitempty"`
DataCenter string `json:"dataCenter,omitempty"`
GrpcPort int `json:"grpcPort,omitempty"`
Url string `json:"url,omitempty"`
PublicUrl string `json:"publicUrl,omitempty"`
DataCenter string `json:"dataCenter,omitempty"`
GrpcPort int `json:"grpcPort,omitempty"`
DataInRemote bool `json:"dataInRemote,omitempty"`
}
func (l *Location) ServerAddress() pb.ServerAddress {
@@ -42,6 +43,8 @@ var (
)
func LookupFileId(masterFn GetMasterFn, grpcDialOption grpc.DialOption, fileId string) (fullUrl string, jwt string, err error) {
var location string
parts := strings.Split(fileId, ",")
if len(parts) != 2 {
return "", jwt, errors.New("Invalid fileId " + fileId)
@@ -53,7 +56,20 @@ func LookupFileId(masterFn GetMasterFn, grpcDialOption grpc.DialOption, fileId s
if len(lookup.Locations) == 0 {
return "", jwt, errors.New("File Not Found")
}
return "http://" + lookup.Locations[rand.IntN(len(lookup.Locations))].Url + "/" + fileId, lookup.Jwt, nil
localUrls := make([]string, 0, len(lookup.Locations))
for _, loc := range lookup.Locations {
if !loc.DataInRemote {
localUrls = append(localUrls, loc.Url)
}
}
if len(localUrls) > 0 {
location = "http://" + localUrls[rand.IntN(len(localUrls))] + "/" + fileId
} else {
location = "http://" + lookup.Locations[rand.IntN(len(lookup.Locations))].Url + "/" + fileId
}
return location, lookup.Jwt, nil
}
func LookupVolumeId(masterFn GetMasterFn, grpcDialOption grpc.DialOption, vid string) (*LookupResult, error) {
@@ -102,10 +118,11 @@ func LookupVolumeIds(masterFn GetMasterFn, grpcDialOption grpc.DialOption, vids
var locations []Location
for _, loc := range vidLocations.Locations {
locations = append(locations, Location{
Url: loc.Url,
PublicUrl: loc.PublicUrl,
DataCenter: loc.DataCenter,
GrpcPort: int(loc.GrpcPort),
Url: loc.Url,
PublicUrl: loc.PublicUrl,
DataCenter: loc.DataCenter,
GrpcPort: int(loc.GrpcPort),
DataInRemote: loc.DataInRemote,
})
}
if vidLocations.Error == "" {
weed/pb/filer.proto
+1
View File
@@ -579,6 +579,7 @@ message Location {
string public_url = 2;
uint32 grpc_port = 3;
string data_center = 4;
bool data_in_remote = 5;
}
message LookupVolumeResponse {
map<string, Locations> locations_map = 1;
weed/pb/filer_pb/filer.pb.go
+11 -2
View File
@@ -3433,6 +3433,7 @@ type Location struct {
PublicUrl string `protobuf:"bytes,2,opt,name=public_url,json=publicUrl,proto3" json:"public_url,omitempty"`
GrpcPort uint32 `protobuf:"varint,3,opt,name=grpc_port,json=grpcPort,proto3" json:"grpc_port,omitempty"`
DataCenter string `protobuf:"bytes,4,opt,name=data_center,json=dataCenter,proto3" json:"data_center,omitempty"`
DataInRemote bool `protobuf:"varint,5,opt,name=data_in_remote,json=dataInRemote,proto3" json:"data_in_remote,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
@@ -3495,6 +3496,13 @@ func (x *Location) GetDataCenter() string {
return ""
}
func (x *Location) GetDataInRemote() bool {
if x != nil {
return x.DataInRemote
}
return false
}
type LookupVolumeResponse struct {
state protoimpl.MessageState `protogen:"open.v1"`
LocationsMap map[string]*Locations `protobuf:"bytes,1,rep,name=locations_map,json=locationsMap,proto3" json:"locations_map,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"`
@@ -7287,14 +7295,15 @@ const file_filer_proto_rawDesc = "" +
"\n" +
"volume_ids\x18\x01 \x03(\tR\tvolumeIds\"=\n" +
"\tLocations\x120\n" +
"\tlocations\x18\x01 \x03(\v2\x12.filer_pb.LocationR\tlocations\"y\n" +
"\tlocations\x18\x01 \x03(\v2\x12.filer_pb.LocationR\tlocations\"\x9f\x01\n" +
"\bLocation\x12\x10\n" +
"\x03url\x18\x01 \x01(\tR\x03url\x12\x1d\n" +
"\n" +
"public_url\x18\x02 \x01(\tR\tpublicUrl\x12\x1b\n" +
"\tgrpc_port\x18\x03 \x01(\rR\bgrpcPort\x12\x1f\n" +
"\vdata_center\x18\x04 \x01(\tR\n" +
"dataCenter\"\xc3\x01\n" +
"dataCenter\x12$\n" +
"\x0edata_in_remote\x18\x05 \x01(\bR\fdataInRemote\"\xc3\x01\n" +
"\x14LookupVolumeResponse\x12U\n" +
"\rlocations_map\x18\x01 \x03(\v20.filer_pb.LookupVolumeResponse.LocationsMapEntryR\flocationsMap\x1aT\n" +
"\x11LocationsMapEntry\x12\x10\n" +
weed/pb/filer_pb/filer_vtproto.pb.go
+33
View File
@@ -3136,6 +3136,16 @@ func (m *Location) MarshalToSizedBufferVT(dAtA []byte) (int, error) {
i -= len(m.unknownFields)
copy(dAtA[i:], m.unknownFields)
}
if m.DataInRemote {
i--
if m.DataInRemote {
dAtA[i] = 1
} else {
dAtA[i] = 0
}
i--
dAtA[i] = 0x28
}
if len(m.DataCenter) > 0 {
i -= len(m.DataCenter)
copy(dAtA[i:], m.DataCenter)
@@ -7494,6 +7504,9 @@ func (m *Location) SizeVT() (n int) {
if l > 0 {
n += 1 + l + protohelpers.SizeOfVarint(uint64(l))
}
if m.DataInRemote {
n += 2
}
n += len(m.unknownFields)
return n
}
@@ -17370,6 +17383,26 @@ func (m *Location) UnmarshalVT(dAtA []byte) error {
}
m.DataCenter = string(dAtA[iNdEx:postIndex])
iNdEx = postIndex
case 5:
if wireType != 0 {
return fmt.Errorf("proto: wrong wireType = %d for field DataInRemote", wireType)
}
var v int
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return protohelpers.ErrIntOverflow
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
v |= int(b&0x7F) << shift
if b < 0x80 {
break
}
}
m.DataInRemote = bool(v != 0)
default:
iNdEx = preIndex
skippy, err := protohelpers.Skip(dAtA[iNdEx:])
weed/pb/master.proto
+2
View File
@@ -222,6 +222,7 @@ message VolumeLocation {
uint32 grpc_port = 7;
repeated uint32 new_ec_vids = 8;
repeated uint32 deleted_ec_vids = 9;
repeated uint32 remote_vids = 10;
}
message ClusterNodeUpdate {
@@ -267,6 +268,7 @@ message Location {
string public_url = 2;
uint32 grpc_port = 3;
string data_center = 4;
bool data_in_remote = 5;
}
message AssignRequest {
weed/pb/master_pb/master.pb.go
+24 -4
View File
@@ -1061,6 +1061,7 @@ type VolumeLocation struct {
GrpcPort uint32 `protobuf:"varint,7,opt,name=grpc_port,json=grpcPort,proto3" json:"grpc_port,omitempty"`
NewEcVids []uint32 `protobuf:"varint,8,rep,packed,name=new_ec_vids,json=newEcVids,proto3" json:"new_ec_vids,omitempty"`
DeletedEcVids []uint32 `protobuf:"varint,9,rep,packed,name=deleted_ec_vids,json=deletedEcVids,proto3" json:"deleted_ec_vids,omitempty"`
RemoteVids []uint32 `protobuf:"varint,10,rep,packed,name=remote_vids,json=remoteVids,proto3" json:"remote_vids,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
@@ -1158,6 +1159,13 @@ func (x *VolumeLocation) GetDeletedEcVids() []uint32 {
return nil
}
func (x *VolumeLocation) GetRemoteVids() []uint32 {
if x != nil {
return x.RemoteVids
}
return nil
}
type ClusterNodeUpdate struct {
state protoimpl.MessageState `protogen:"open.v1"`
NodeType string `protobuf:"bytes,1,opt,name=node_type,json=nodeType,proto3" json:"node_type,omitempty"`
@@ -1460,6 +1468,7 @@ type Location struct {
PublicUrl string `protobuf:"bytes,2,opt,name=public_url,json=publicUrl,proto3" json:"public_url,omitempty"`
GrpcPort uint32 `protobuf:"varint,3,opt,name=grpc_port,json=grpcPort,proto3" json:"grpc_port,omitempty"`
DataCenter string `protobuf:"bytes,4,opt,name=data_center,json=dataCenter,proto3" json:"data_center,omitempty"`
DataInRemote bool `protobuf:"varint,5,opt,name=data_in_remote,json=dataInRemote,proto3" json:"data_in_remote,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
@@ -1522,6 +1531,13 @@ func (x *Location) GetDataCenter() string {
return ""
}
func (x *Location) GetDataInRemote() bool {
if x != nil {
return x.DataInRemote
}
return false
}
type AssignRequest struct {
state protoimpl.MessageState `protogen:"open.v1"`
Count uint64 `protobuf:"varint,1,opt,name=count,proto3" json:"count,omitempty"`
@@ -5072,7 +5088,7 @@ const file_master_proto_rawDesc = "" +
"filerGroup\x12\x1f\n" +
"\vdata_center\x18\x06 \x01(\tR\n" +
"dataCenter\x12\x12\n" +
"\x04rack\x18\a \x01(\tR\x04rack\"\x9d\x02\n" +
"\x04rack\x18\a \x01(\tR\x04rack\"\xbe\x02\n" +
"\x0eVolumeLocation\x12\x10\n" +
"\x03url\x18\x01 \x01(\tR\x03url\x12\x1d\n" +
"\n" +
@@ -5084,7 +5100,10 @@ const file_master_proto_rawDesc = "" +
"dataCenter\x12\x1b\n" +
"\tgrpc_port\x18\a \x01(\rR\bgrpcPort\x12\x1e\n" +
"\vnew_ec_vids\x18\b \x03(\rR\tnewEcVids\x12&\n" +
"\x0fdeleted_ec_vids\x18\t \x03(\rR\rdeletedEcVids\"\xa6\x01\n" +
"\x0fdeleted_ec_vids\x18\t \x03(\rR\rdeletedEcVids\x12\x1f\n" +
"\vremote_vids\x18\n" +
" \x03(\rR\n" +
"remoteVids\"\xa6\x01\n" +
"\x11ClusterNodeUpdate\x12\x1b\n" +
"\tnode_type\x18\x01 \x01(\tR\bnodeType\x12\x18\n" +
"\aaddress\x18\x02 \x01(\tR\aaddress\x12\x15\n" +
@@ -5112,14 +5131,15 @@ const file_master_proto_rawDesc = "" +
"\x11volume_or_file_id\x18\x01 \x01(\tR\x0evolumeOrFileId\x121\n" +
"\tlocations\x18\x02 \x03(\v2\x13.master_pb.LocationR\tlocations\x12\x14\n" +
"\x05error\x18\x03 \x01(\tR\x05error\x12\x12\n" +
"\x04auth\x18\x04 \x01(\tR\x04auth\"y\n" +
"\x04auth\x18\x04 \x01(\tR\x04auth\"\x9f\x01\n" +
"\bLocation\x12\x10\n" +
"\x03url\x18\x01 \x01(\tR\x03url\x12\x1d\n" +
"\n" +
"public_url\x18\x02 \x01(\tR\tpublicUrl\x12\x1b\n" +
"\tgrpc_port\x18\x03 \x01(\rR\bgrpcPort\x12\x1f\n" +
"\vdata_center\x18\x04 \x01(\tR\n" +
"dataCenter\"\xfe\x02\n" +
"dataCenter\x12$\n" +
"\x0edata_in_remote\x18\x05 \x01(\bR\fdataInRemote\"\xfe\x02\n" +
"\rAssignRequest\x12\x14\n" +
"\x05count\x18\x01 \x01(\x04R\x05count\x12 \n" +
"\vreplication\x18\x02 \x01(\tR\vreplication\x12\x1e\n" +
weed/server/filer_grpc_server.go
+5 -4
View File
@@ -162,10 +162,11 @@ func wdclientLocationsToPb(locations []wdclient.Location) []*filer_pb.Location {
locs := make([]*filer_pb.Location, 0, len(locations))
for _, loc := range locations {
locs = append(locs, &filer_pb.Location{
Url: loc.Url,
PublicUrl: loc.PublicUrl,
GrpcPort: uint32(loc.GrpcPort),
DataCenter: loc.DataCenter,
Url: loc.Url,
PublicUrl: loc.PublicUrl,
GrpcPort: uint32(loc.GrpcPort),
DataCenter: loc.DataCenter,
DataInRemote: loc.DataInRemote,
})
}
return locs
weed/server/master_grpc_server.go
+7 -2
View File
@@ -12,6 +12,7 @@ import (
"github.com/seaweedfs/seaweedfs/weed/cluster"
"github.com/seaweedfs/seaweedfs/weed/cluster/maintenance"
"github.com/seaweedfs/seaweedfs/weed/pb"
"github.com/seaweedfs/seaweedfs/weed/stats"
"github.com/seaweedfs/seaweedfs/weed/storage/backend"
@@ -262,7 +263,11 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
for _, v := range newVolumes {
glog.V(1).Infof("master see new volume %d from %s", uint32(v.Id), dn.Url())
message.NewVids = append(message.NewVids, uint32(v.Id))
if v.IsRemote() {
message.RemoteVids = append(message.RemoteVids, uint32(v.Id))
} else {
message.NewVids = append(message.NewVids, uint32(v.Id))
}
}
for _, v := range deletedVolumes {
glog.V(1).Infof("master see deleted volume %d from %s", uint32(v.Id), dn.Url())
@@ -307,7 +312,7 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
}
}
if len(message.NewVids) > 0 || len(message.DeletedVids) > 0 || len(message.NewEcVids) > 0 || len(message.DeletedEcVids) > 0 {
if len(message.NewVids) > 0 || len(message.DeletedVids) > 0 || len(message.NewEcVids) > 0 || len(message.DeletedEcVids) > 0 || len(message.RemoteVids) > 0 {
ms.broadcastToClients(&master_pb.KeepConnectedResponse{VolumeLocation: message})
}
weed/server/master_grpc_server_volume.go
+5 -4
View File
@@ -167,10 +167,11 @@ func (ms *MasterServer) LookupVolume(ctx context.Context, req *master_pb.LookupV
var locations []*master_pb.Location
for _, loc := range result.Locations {
locations = append(locations, &master_pb.Location{
Url: loc.Url,
PublicUrl: loc.PublicUrl,
DataCenter: loc.DataCenter,
GrpcPort: uint32(loc.GrpcPort),
Url: loc.Url,
PublicUrl: loc.PublicUrl,
DataCenter: loc.DataCenter,
GrpcPort: uint32(loc.GrpcPort),
DataInRemote: loc.DataInRemote,
})
}
var auth string
weed/server/master_server_handlers.go
+16 -9
View File
@@ -81,15 +81,21 @@ func (ms *MasterServer) findVolumeLocation(collection, vid string) operation.Loo
if ms.Topo.IsLeader() {
volumeId, newVolumeIdErr := needle.NewVolumeId(vid)
if newVolumeIdErr != nil {
err = fmt.Errorf("Unknown volume id %s", vid)
err = fmt.Errorf("unknown volume id %s", vid)
} else {
machines := ms.Topo.Lookup(collection, volumeId)
for _, loc := range machines {
volInfo, err := loc.GetVolumesById(volumeId)
if err != nil {
glog.V(0).Infof("failed to get volume info from %s: %v", loc.Url(), err)
continue
}
locations = append(locations, operation.Location{
Url: loc.Url(),
PublicUrl: loc.PublicUrl,
DataCenter: loc.GetDataCenterId(),
GrpcPort: loc.GrpcPort,
Url: loc.Url(),
PublicUrl: loc.PublicUrl,
DataCenter: loc.GetDataCenterId(),
GrpcPort: loc.GrpcPort,
DataInRemote: volInfo.IsRemote(),
})
}
}
@@ -97,10 +103,11 @@ func (ms *MasterServer) findVolumeLocation(collection, vid string) operation.Loo
machines, getVidLocationsErr := ms.MasterClient.GetVidLocations(vid)
for _, loc := range machines {
locations = append(locations, operation.Location{
Url: loc.Url,
PublicUrl: loc.PublicUrl,
DataCenter: loc.DataCenter,
GrpcPort: loc.GrpcPort,
Url: loc.Url,
PublicUrl: loc.PublicUrl,
DataCenter: loc.DataCenter,
GrpcPort: loc.GrpcPort,
DataInRemote: loc.DataInRemote,
})
}
err = getVidLocationsErr
weed/topology/topology_info.go
+9 -1
View File
@@ -100,7 +100,15 @@ func (t *Topology) ToVolumeLocations() (volumeLocations []*master_pb.VolumeLocat
DataCenter: dn.GetDataCenterId(),
GrpcPort: uint32(dn.GrpcPort),
}
volumeLocation.NewVids = dn.AppendVolumeIds(nil)
for _, v := range dn.GetVolumes() {
if v.IsRemote() {
volumeLocation.RemoteVids = append(volumeLocation.RemoteVids, uint32(v.Id))
} else {
volumeLocation.NewVids = append(volumeLocation.NewVids, uint32(v.Id))
}
}
// A single EC volume's shards can live on multiple disks of
// one DataNode, so GetEcShards returns per-(vid,disk) entries.
// Dedupe so the snapshot carries each vid once.
weed/util/slice.go
+15
View File
@@ -14,3 +14,18 @@ func DrainChannel[T any](ch chan T, first T) []T {
}
}
}
func ReorderToFront[T comparable](frontMap map[T]bool, inputSlice []T) []T {
var prioritized []T
var remaining []T
for _, item := range inputSlice {
if frontMap[item] {
prioritized = append(prioritized, item)
} else {
remaining = append(remaining, item)
}
}
return append(prioritized, remaining...)
}
weed/util/slice_test.go
+40
View File
@@ -0,0 +1,40 @@
package util
import (
"reflect"
"testing"
)
func TestReorderToFront_StringSlice(t *testing.T) {
localUrls := map[string]bool{
"http://local1": true,
"http://local2": true,
}
sameDcTargetUrls := []string{
"http://remote1",
"http://local2",
"http://remote2",
"http://local1",
}
expected1 := []string{
"http://local1",
"http://local2",
"http://remote1",
"http://remote2",
}
expected2 := []string{
"http://local2",
"http://local1",
"http://remote1",
"http://remote2",
}
result := ReorderToFront(localUrls, sameDcTargetUrls)
if !reflect.DeepEqual(result, expected1) && !reflect.DeepEqual(result, expected2) {
t.Errorf("ReorderToFront failed for strings. Got: %v, Expected1: %v, Expected2: %v", result, expected1, expected2)
}
}
weed/wdclient/masterclient.go
+11 -3
View File
@@ -109,6 +109,7 @@ func (p *masterVolumeProvider) LookupVolumeIds(ctx context.Context, volumeIds []
PublicUrl: masterLoc.PublicUrl,
GrpcPort: int(masterLoc.GrpcPort),
DataCenter: masterLoc.DataCenter,
DataInRemote: masterLoc.DataInRemote,
}
// Update cache with the location
p.masterClient.addLocation(uint32(vid), loc)
@@ -384,6 +385,12 @@ func (mc *MasterClient) updateVidMap(resp *master_pb.KeepConnectedResponse) {
glog.V(2).Infof("%s.%s: %s masterClient adds volume %d", mc.FilerGroup, mc.clientType, loc.Url, newVid)
mc.addLocation(newVid, loc)
}
for _, remoteVid := range resp.VolumeLocation.RemoteVids {
remoteLoc := loc
remoteLoc.DataInRemote = true
glog.V(2).Infof("%s.%s: %s masterClient adds remote volume %d", mc.FilerGroup, mc.clientType, remoteLoc.Url, remoteVid)
mc.addLocation(remoteVid, remoteLoc)
}
for _, deletedVid := range resp.VolumeLocation.DeletedVids {
if _, moved := stillOnServer[deletedVid]; moved {
continue
@@ -403,10 +410,11 @@ func (mc *MasterClient) updateVidMap(resp *master_pb.KeepConnectedResponse) {
glog.V(2).Infof("%s.%s: %s masterClient removes ec volume %d", mc.FilerGroup, mc.clientType, loc.Url, deletedEcVid)
mc.deleteEcLocation(deletedEcVid, loc)
}
glog.V(1).Infof("updateVidMap(%s) %s.%s: %s volume add: %d, del: %d, add ec: %d del ec: %d",
glog.V(1).Infof("updateVidMap(%s) %s.%s: %s volume add local: %d, remote: %d, del: %d, add ec: %d del ec: %d",
resp.VolumeLocation.DataCenter, mc.FilerGroup, mc.clientType, loc.Url,
len(resp.VolumeLocation.NewVids), len(resp.VolumeLocation.DeletedVids),
len(resp.VolumeLocation.NewEcVids), len(resp.VolumeLocation.DeletedEcVids))
len(resp.VolumeLocation.NewVids), len(resp.VolumeLocation.RemoteVids),
len(resp.VolumeLocation.DeletedVids), len(resp.VolumeLocation.NewEcVids),
len(resp.VolumeLocation.DeletedEcVids))
}
func (mc *MasterClient) WithClient(ctx context.Context, streamingMode bool, fn func(client master_pb.SeaweedClient) error) error {
weed/wdclient/vid_map.go
+19 -6
View File
@@ -9,9 +9,9 @@ import (
"strings"
"sync"
"github.com/seaweedfs/seaweedfs/weed/pb"
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/pb"
"github.com/seaweedfs/seaweedfs/weed/util"
)
type HasLookupFileIdFunction interface {
@@ -21,10 +21,11 @@ type HasLookupFileIdFunction interface {
type LookupFileIdFunctionType func(ctx context.Context, fileId string) (targetUrls []string, err error)
type Location struct {
Url string `json:"url,omitempty"`
PublicUrl string `json:"publicUrl,omitempty"`
DataCenter string `json:"dataCenter,omitempty"`
GrpcPort int `json:"grpcPort,omitempty"`
Url string `json:"url,omitempty"`
PublicUrl string `json:"publicUrl,omitempty"`
DataCenter string `json:"dataCenter,omitempty"`
GrpcPort int `json:"grpcPort,omitempty"`
DataInRemote bool `json:"dataInRemote,omitempty"`
}
func (l Location) ServerAddress() pb.ServerAddress {
@@ -99,7 +100,15 @@ func (vc *vidMap) LookupVolumeServerUrl(vid string) (serverUrls []string, err er
return nil, fmt.Errorf("volume %d not found", id)
}
var sameDcServers, otherDcServers []string
localUrls := make(map[string]bool)
for _, loc := range locations {
glog.V(4).Infof("lookup %s => %s, data in remote storage tier: %v", vid, loc.Url, loc.DataInRemote)
if !loc.DataInRemote {
localUrls[loc.Url] = true
}
if vc.isSameDataCenter(&loc) {
sameDcServers = append(sameDcServers, loc.Url)
} else {
@@ -112,6 +121,10 @@ func (vc *vidMap) LookupVolumeServerUrl(vid string) (serverUrls []string, err er
rand.Shuffle(len(otherDcServers), func(i, j int) {
otherDcServers[i], otherDcServers[j] = otherDcServers[j], otherDcServers[i]
})
if len(localUrls) > 0 {
sameDcServers = util.ReorderToFront(localUrls, sameDcServers)
otherDcServers = util.ReorderToFront(localUrls, otherDcServers)
}
// Prefer same data center
serverUrls = append(sameDcServers, otherDcServers...)
return