gsh-digital-services/seaweedFS
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

diff

Browse Source
This commit is contained in:
a
2022-03-28 16:50:28 +00:00
parent beb406bbbb 93615b2a49
commit bc603e534f
55 changed files with 1057 additions and 483 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
weed/command/benchmark.go
View File

@@ -129,7 +129,7 @@ func runBenchmark(cmd *Command, args []string) bool {
defer pprof.StopCPUProfile()
}
b.masterClient = wdclient.NewMasterClient(b.grpcDialOption, "client", "", "", pb.ServerAddresses(*b.masters).ToAddresses())
b.masterClient = wdclient.NewMasterClient(b.grpcDialOption, "client", "", "", pb.ServerAddresses(*b.masters).ToAddressMap())
go b.masterClient.KeepConnectedToMaster()
b.masterClient.WaitUntilConnected()

7
weed/command/filer.go
View File

@@ -13,7 +13,7 @@ import (
"github.com/chrislusf/seaweedfs/weed/pb"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
"github.com/chrislusf/seaweedfs/weed/security"
"github.com/chrislusf/seaweedfs/weed/server"
weed_server "github.com/chrislusf/seaweedfs/weed/server"
stats_collect "github.com/chrislusf/seaweedfs/weed/stats"
"github.com/chrislusf/seaweedfs/weed/util"
)
@@ -29,7 +29,7 @@ var (
)
type FilerOptions struct {
masters []pb.ServerAddress
masters map[string]pb.ServerAddress
mastersString *string
ip *string
bindIp *string
@@ -103,6 +103,7 @@ func init() {
// start iam on filer
filerStartIam = cmdFiler.Flag.Bool("iam", false, "whether to start IAM service")
filerIamOptions.ip = cmdFiler.Flag.String("iam.ip", *f.ip, "iam server http listen ip address")
filerIamOptions.port = cmdFiler.Flag.Int("iam.port", 8111, "iam server http listen port")
}
@@ -170,7 +171,7 @@ func runFiler(cmd *Command, args []string) bool {
}()
}
f.masters = pb.ServerAddresses(*f.mastersString).ToAddresses()
f.masters = pb.ServerAddresses(*f.mastersString).ToAddressMap()
f.startFiler()

2
weed/command/iam.go
View File

@@ -67,7 +67,7 @@ func (iamopt *IamOptions) startIamServer() bool {
}
}
masters := pb.ServerAddresses(*iamopt.masters).ToAddresses()
masters := pb.ServerAddresses(*iamopt.masters).ToAddressMap()
router := mux.NewRouter().SkipClean(true)
_, iamApiServer_err := iamapi.NewIamApiServer(router, &iamapi.IamServerOption{
Masters: masters,

11
weed/command/master.go
View File

@@ -127,13 +127,18 @@ func startMaster(masterOption MasterOptions, masterWhiteList []string) {
*masterOption.portGrpc = 10000 + *masterOption.port
}
if *masterOption.ipBind == "" {
*masterOption.ipBind = "localhost"
*masterOption.ipBind = *masterOption.ip
}
myMasterAddress, peers := checkPeers(*masterOption.ip, *masterOption.port, *masterOption.portGrpc, *masterOption.peers)
masterPeers := make(map[string]pb.ServerAddress)
for _, peer := range peers {
masterPeers[peer.String()] = peer
}
r := mux.NewRouter()
ms := weed_server.NewMasterServer(r, masterOption.toMasterOption(masterWhiteList), peers)
ms := weed_server.NewMasterServer(r, masterOption.toMasterOption(masterWhiteList), masterPeers)
listeningAddress := util.JoinHostPort(*masterOption.ipBind, *masterOption.port)
glog.V(0).Infof("Start Seaweed Master %s at %s", util.Version(), listeningAddress)
masterListener, masterLocalListner, e := util.NewIpAndLocalListeners(*masterOption.ipBind, *masterOption.port, 0)
@@ -144,7 +149,7 @@ func startMaster(masterOption MasterOptions, masterWhiteList []string) {
// start raftServer
raftServerOption := &weed_server.RaftServerOption{
GrpcDialOption: security.LoadClientTLS(util.GetViper(), "grpc.master"),
Peers: peers,
Peers: masterPeers,
ServerAddr: myMasterAddress,
DataDir: util.ResolvePath(*masterOption.metaFolder),
Topo: ms.Topo,

15
weed/command/master_follower.go
View File

@@ -3,17 +3,18 @@ package command
import (
"context"
"fmt"
"net/http"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/pb"
"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
"github.com/chrislusf/seaweedfs/weed/security"
"github.com/chrislusf/seaweedfs/weed/server"
weed_server "github.com/chrislusf/seaweedfs/weed/server"
"github.com/chrislusf/seaweedfs/weed/util"
"github.com/gorilla/mux"
"google.golang.org/grpc/reflection"
"net/http"
"time"
)
var (
@@ -45,13 +46,13 @@ var cmdMasterFollower = &Command{
Short: "start a master follower",
Long: `start a master follower to provide volume=>location mapping service
The master follower does not participate in master election.
The master follower does not participate in master election.
It just follow the existing masters, and listen for any volume location changes.
In most cases, the master follower is not needed. In big data centers with thousands of volume
servers. In theory, the master may have trouble to keep up with the write requests and read requests.
The master follower can relieve the master from from read requests, which only needs to
The master follower can relieve the master from from read requests, which only needs to
lookup a fileId or volumeId.
The master follower currently can handle fileId lookup requests:
@@ -82,7 +83,7 @@ func runMasterFollower(cmd *Command, args []string) bool {
func startMasterFollower(masterOptions MasterOptions) {
// collect settings from main masters
masters := pb.ServerAddresses(*mf.peers).ToAddresses()
masters := pb.ServerAddresses(*mf.peers).ToAddressMap()
var err error
grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.master")
@@ -112,7 +113,7 @@ func startMasterFollower(masterOptions MasterOptions) {
option.IsFollower = true
if *masterOptions.ipBind == "" {
*masterOptions.ipBind = "localhost"
*masterOptions.ipBind = *masterOptions.ip
}
r := mux.NewRouter()

2
weed/command/server.go
View File

@@ -191,7 +191,7 @@ func runServer(cmd *Command, args []string) bool {
// ip address
masterOptions.ip = serverIp
masterOptions.ipBind = serverBindIp
filerOptions.masters = pb.ServerAddresses(*masterOptions.peers).ToAddresses()
filerOptions.masters = pb.ServerAddresses(*masterOptions.peers).ToAddressMap()
filerOptions.ip = serverIp
filerOptions.bindIp = serverBindIp
s3Options.bindIp = serverBindIp

8
weed/filer/filechunks.go
View File

@@ -23,7 +23,13 @@ func TotalSize(chunks []*filer_pb.FileChunk) (size uint64) {
}
func FileSize(entry *filer_pb.Entry) (size uint64) {
return maxUint64(TotalSize(entry.Chunks), entry.Attributes.FileSize)
fileSize := entry.Attributes.FileSize
if entry.RemoteEntry != nil {
if entry.RemoteEntry.RemoteMtime > entry.Attributes.Mtime {
fileSize = maxUint64(fileSize, uint64(entry.RemoteEntry.RemoteSize))
}
}
return maxUint64(TotalSize(entry.Chunks), fileSize)
}
func ETag(entry *filer_pb.Entry) (etag string) {

2
weed/filer/filer.go
View File

@@ -49,7 +49,7 @@ type Filer struct {
UniqueFileId uint32
}
func NewFiler(masters []pb.ServerAddress, grpcDialOption grpc.DialOption,
func NewFiler(masters map[string]pb.ServerAddress, grpcDialOption grpc.DialOption,
filerHost pb.ServerAddress, collection string, replication string, dataCenter string, notifyFn func()) *Filer {
f := &Filer{
MasterClient: wdclient.NewMasterClient(grpcDialOption, cluster.FilerType, filerHost, dataCenter, masters),

3
weed/filer/meta_aggregator.go
View File

@@ -76,9 +76,6 @@ func (ma *MetaAggregator) setActive(address pb.ServerAddress, isActive bool) (no
}
} else {
if _, found := ma.peerStatues[address]; found {
ma.peerStatues[address] -= 1
}
if ma.peerStatues[address] <= 0 {
delete(ma.peerStatues, address)
}
}

1
weed/filer/read_write.go
View File

@@ -62,6 +62,7 @@ func SaveInsideFiler(client filer_pb.SeaweedFilerClient, dir, name string, conte
},
Content: content,
},
SkipCheckParentDirectory: true,
})
} else if err == nil {
entry := resp.Entry

2
weed/iamapi/iamapi_server.go
View File

@@ -33,7 +33,7 @@ type IamS3ApiConfigure struct {
}
type IamServerOption struct {
Masters []pb.ServerAddress
Masters map[string]pb.ServerAddress
Filer pb.ServerAddress
Port int
GrpcDialOption grpc.DialOption

7
weed/mount/weedfs_file_mkrm.go
View File

@@ -77,9 +77,10 @@ func (wfs *WFS) Mknod(cancel <-chan struct{}, in *fuse.MknodIn, name string, out
defer wfs.mapPbIdFromFilerToLocal(newEntry)
request := &filer_pb.CreateEntryRequest{
Directory: string(dirFullPath),
Entry: newEntry,
Signatures: []int32{wfs.signature},
Directory: string(dirFullPath),
Entry: newEntry,
Signatures: []int32{wfs.signature},
SkipCheckParentDirectory: true,
}
glog.V(1).Infof("mknod: %v", request)

7
weed/mount/weedfs_file_sync.go
View File

@@ -140,9 +140,10 @@ func (wfs *WFS) doFlush(fh *FileHandle, uid, gid uint32) fuse.Status {
}
request := &filer_pb.CreateEntryRequest{
Directory: string(dir),
Entry: entry,
Signatures: []int32{wfs.signature},
Directory: string(dir),
Entry: entry,
Signatures: []int32{wfs.signature},
SkipCheckParentDirectory: true,
}
glog.V(4).Infof("%s set chunks: %v", fileFullPath, len(entry.Chunks))

2
weed/pb/grpc_client_server.go
View File

@@ -206,7 +206,7 @@ func WithMasterClient(streamingMode bool, master ServerAddress, grpcDialOption g
}
func WithOneOfGrpcMasterClients(streamingMode bool, masterGrpcAddresses []ServerAddress, grpcDialOption grpc.DialOption, fn func(client master_pb.SeaweedClient) error) (err error) {
func WithOneOfGrpcMasterClients(streamingMode bool, masterGrpcAddresses map[string]ServerAddress, grpcDialOption grpc.DialOption, fn func(client master_pb.SeaweedClient) error) (err error) {
for _, masterGrpcAddress := range masterGrpcAddresses {
err = WithGrpcClient(streamingMode, func(grpcConnection *grpc.ClientConn) error {

4
weed/pb/remote.proto
View File

@@ -62,6 +62,10 @@ message RemoteConf {
string storj_secret_key = 66;
string storj_endpoint = 67;
string contabo_access_key = 68;
string contabo_secret_key = 69;
string contabo_endpoint = 70;
string contabo_region = 71;
}
message RemoteStorageMapping {

105
weed/pb/remote_pb/remote.pb.go
View File

@@ -1,7 +1,7 @@
// Code generated by protoc-gen-go. DO NOT EDIT.
// versions:
// protoc-gen-go v1.26.0
// protoc v3.17.3
// protoc v3.6.1
// source: remote.proto
package remote_pb
@@ -69,6 +69,10 @@ type RemoteConf struct {
StorjAccessKey string `protobuf:"bytes,65,opt,name=storj_access_key,json=storjAccessKey,proto3" json:"storj_access_key,omitempty"`
StorjSecretKey string `protobuf:"bytes,66,opt,name=storj_secret_key,json=storjSecretKey,proto3" json:"storj_secret_key,omitempty"`
StorjEndpoint string `protobuf:"bytes,67,opt,name=storj_endpoint,json=storjEndpoint,proto3" json:"storj_endpoint,omitempty"`
ContaboAccessKey string `protobuf:"bytes,68,opt,name=contabo_access_key,json=contaboAccessKey,proto3" json:"contabo_access_key,omitempty"`
ContaboSecretKey string `protobuf:"bytes,69,opt,name=contabo_secret_key,json=contaboSecretKey,proto3" json:"contabo_secret_key,omitempty"`
ContaboEndpoint string `protobuf:"bytes,70,opt,name=contabo_endpoint,json=contaboEndpoint,proto3" json:"contabo_endpoint,omitempty"`
ContaboRegion string `protobuf:"bytes,71,opt,name=contabo_region,json=contaboRegion,proto3" json:"contabo_region,omitempty"`
}
func (x *RemoteConf) Reset() {
@@ -390,6 +394,34 @@ func (x *RemoteConf) GetStorjEndpoint() string {
return ""
}
func (x *RemoteConf) GetContaboAccessKey() string {
if x != nil {
return x.ContaboAccessKey
}
return ""
}
func (x *RemoteConf) GetContaboSecretKey() string {
if x != nil {
return x.ContaboSecretKey
}
return ""
}
func (x *RemoteConf) GetContaboEndpoint() string {
if x != nil {
return x.ContaboEndpoint
}
return ""
}
func (x *RemoteConf) GetContaboRegion() string {
if x != nil {
return x.ContaboRegion
}
return ""
}
type RemoteStorageMapping struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
@@ -512,7 +544,7 @@ var File_remote_proto protoreflect.FileDescriptor
var file_remote_proto_rawDesc = []byte{
0x0a, 0x0c, 0x72, 0x65, 0x6d, 0x6f, 0x74, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x09,
0x72, 0x65, 0x6d, 0x6f, 0x74, 0x65, 0x5f, 0x70, 0x62, 0x22, 0xe2, 0x0d, 0x0a, 0x0a, 0x52, 0x65,
0x72, 0x65, 0x6d, 0x6f, 0x74, 0x65, 0x5f, 0x70, 0x62, 0x22, 0x90, 0x0f, 0x0a, 0x0a, 0x52, 0x65,
0x6d, 0x6f, 0x74, 0x65, 0x43, 0x6f, 0x6e, 0x66, 0x12, 0x12, 0x0a, 0x04, 0x74, 0x79, 0x70, 0x65,
0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x74, 0x79, 0x70, 0x65, 0x12, 0x12, 0x0a, 0x04,
0x6e, 0x61, 0x6d, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65,
@@ -622,35 +654,46 @@ var file_remote_proto_rawDesc = []byte{
0x79, 0x18, 0x42, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0e, 0x73, 0x74, 0x6f, 0x72, 0x6a, 0x53, 0x65,
0x63, 0x72, 0x65, 0x74, 0x4b, 0x65, 0x79, 0x12, 0x25, 0x0a, 0x0e, 0x73, 0x74, 0x6f, 0x72, 0x6a,
0x5f, 0x65, 0x6e, 0x64, 0x70, 0x6f, 0x69, 0x6e, 0x74, 0x18, 0x43, 0x20, 0x01, 0x28, 0x09, 0x52,
0x0d, 0x73, 0x74, 0x6f, 0x72, 0x6a, 0x45, 0x6e, 0x64, 0x70, 0x6f, 0x69, 0x6e, 0x74, 0x22, 0xff,
0x01, 0x0a, 0x14, 0x52, 0x65, 0x6d, 0x6f, 0x74, 0x65, 0x53, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65,
0x4d, 0x61, 0x70, 0x70, 0x69, 0x6e, 0x67, 0x12, 0x49, 0x0a, 0x08, 0x6d, 0x61, 0x70, 0x70, 0x69,
0x6e, 0x67, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x2d, 0x2e, 0x72, 0x65, 0x6d, 0x6f,
0x74, 0x65, 0x5f, 0x70, 0x62, 0x2e, 0x52, 0x65, 0x6d, 0x6f, 0x74, 0x65, 0x53, 0x74, 0x6f, 0x72,
0x61, 0x67, 0x65, 0x4d, 0x61, 0x70, 0x70, 0x69, 0x6e, 0x67, 0x2e, 0x4d, 0x61, 0x70, 0x70, 0x69,
0x6e, 0x67, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x08, 0x6d, 0x61, 0x70, 0x70, 0x69, 0x6e,
0x67, 0x73, 0x12, 0x3d, 0x0a, 0x1b, 0x70, 0x72, 0x69, 0x6d, 0x61, 0x72, 0x79, 0x5f, 0x62, 0x75,
0x63, 0x6b, 0x65, 0x74, 0x5f, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x5f, 0x6e, 0x61, 0x6d,
0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x18, 0x70, 0x72, 0x69, 0x6d, 0x61, 0x72, 0x79,
0x42, 0x75, 0x63, 0x6b, 0x65, 0x74, 0x53, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x4e, 0x61, 0x6d,
0x65, 0x1a, 0x5d, 0x0a, 0x0d, 0x4d, 0x61, 0x70, 0x70, 0x69, 0x6e, 0x67, 0x73, 0x45, 0x6e, 0x74,
0x72, 0x79, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52,
0x03, 0x6b, 0x65, 0x79, 0x12, 0x36, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20,
0x01, 0x28, 0x0b, 0x32, 0x20, 0x2e, 0x72, 0x65, 0x6d, 0x6f, 0x74, 0x65, 0x5f, 0x70, 0x62, 0x2e,
0x52, 0x65, 0x6d, 0x6f, 0x74, 0x65, 0x53, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x4c, 0x6f, 0x63,
0x61, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x3a, 0x02, 0x38, 0x01,
0x22, 0x57, 0x0a, 0x15, 0x52, 0x65, 0x6d, 0x6f, 0x74, 0x65, 0x53, 0x74, 0x6f, 0x72, 0x61, 0x67,
0x65, 0x4c, 0x6f, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d,
0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x12, 0x16, 0x0a,
0x06, 0x62, 0x75, 0x63, 0x6b, 0x65, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x62,
0x75, 0x63, 0x6b, 0x65, 0x74, 0x12, 0x12, 0x0a, 0x04, 0x70, 0x61, 0x74, 0x68, 0x18, 0x03, 0x20,
0x01, 0x28, 0x09, 0x52, 0x04, 0x70, 0x61, 0x74, 0x68, 0x42, 0x50, 0x0a, 0x10, 0x73, 0x65, 0x61,
0x77, 0x65, 0x65, 0x64, 0x66, 0x73, 0x2e, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x42, 0x0a, 0x46,
0x69, 0x6c, 0x65, 0x72, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x5a, 0x30, 0x67, 0x69, 0x74, 0x68, 0x75,
0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x63, 0x68, 0x72, 0x69, 0x73, 0x6c, 0x75, 0x73, 0x66, 0x2f,
0x73, 0x65, 0x61, 0x77, 0x65, 0x65, 0x64, 0x66, 0x73, 0x2f, 0x77, 0x65, 0x65, 0x64, 0x2f, 0x70,
0x62, 0x2f, 0x72, 0x65, 0x6d, 0x6f, 0x74, 0x65, 0x5f, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f,
0x74, 0x6f, 0x33,
0x0d, 0x73, 0x74, 0x6f, 0x72, 0x6a, 0x45, 0x6e, 0x64, 0x70, 0x6f, 0x69, 0x6e, 0x74, 0x12, 0x2c,
0x0a, 0x12, 0x63, 0x6f, 0x6e, 0x74, 0x61, 0x62, 0x6f, 0x5f, 0x61, 0x63, 0x63, 0x65, 0x73, 0x73,
0x5f, 0x6b, 0x65, 0x79, 0x18, 0x44, 0x20, 0x01, 0x28, 0x09, 0x52, 0x10, 0x63, 0x6f, 0x6e, 0x74,
0x61, 0x62, 0x6f, 0x41, 0x63, 0x63, 0x65, 0x73, 0x73, 0x4b, 0x65, 0x79, 0x12, 0x2c, 0x0a, 0x12,
0x63, 0x6f, 0x6e, 0x74, 0x61, 0x62, 0x6f, 0x5f, 0x73, 0x65, 0x63, 0x72, 0x65, 0x74, 0x5f, 0x6b,
0x65, 0x79, 0x18, 0x45, 0x20, 0x01, 0x28, 0x09, 0x52, 0x10, 0x63, 0x6f, 0x6e, 0x74, 0x61, 0x62,
0x6f, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x4b, 0x65, 0x79, 0x12, 0x29, 0x0a, 0x10, 0x63, 0x6f,
0x6e, 0x74, 0x61, 0x62, 0x6f, 0x5f, 0x65, 0x6e, 0x64, 0x70, 0x6f, 0x69, 0x6e, 0x74, 0x18, 0x46,
0x20, 0x01, 0x28, 0x09, 0x52, 0x0f, 0x63, 0x6f, 0x6e, 0x74, 0x61, 0x62, 0x6f, 0x45, 0x6e, 0x64,
0x70, 0x6f, 0x69, 0x6e, 0x74, 0x12, 0x25, 0x0a, 0x0e, 0x63, 0x6f, 0x6e, 0x74, 0x61, 0x62, 0x6f,
0x5f, 0x72, 0x65, 0x67, 0x69, 0x6f, 0x6e, 0x18, 0x47, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0d, 0x63,
0x6f, 0x6e, 0x74, 0x61, 0x62, 0x6f, 0x52, 0x65, 0x67, 0x69, 0x6f, 0x6e, 0x22, 0xff, 0x01, 0x0a,
0x14, 0x52, 0x65, 0x6d, 0x6f, 0x74, 0x65, 0x53, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x4d, 0x61,
0x70, 0x70, 0x69, 0x6e, 0x67, 0x12, 0x49, 0x0a, 0x08, 0x6d, 0x61, 0x70, 0x70, 0x69, 0x6e, 0x67,
0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x2d, 0x2e, 0x72, 0x65, 0x6d, 0x6f, 0x74, 0x65,
0x5f, 0x70, 0x62, 0x2e, 0x52, 0x65, 0x6d, 0x6f, 0x74, 0x65, 0x53, 0x74, 0x6f, 0x72, 0x61, 0x67,
0x65, 0x4d, 0x61, 0x70, 0x70, 0x69, 0x6e, 0x67, 0x2e, 0x4d, 0x61, 0x70, 0x70, 0x69, 0x6e, 0x67,
0x73, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x08, 0x6d, 0x61, 0x70, 0x70, 0x69, 0x6e, 0x67, 0x73,
0x12, 0x3d, 0x0a, 0x1b, 0x70, 0x72, 0x69, 0x6d, 0x61, 0x72, 0x79, 0x5f, 0x62, 0x75, 0x63, 0x6b,
0x65, 0x74, 0x5f, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x5f, 0x6e, 0x61, 0x6d, 0x65, 0x18,
0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x18, 0x70, 0x72, 0x69, 0x6d, 0x61, 0x72, 0x79, 0x42, 0x75,
0x63, 0x6b, 0x65, 0x74, 0x53, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x4e, 0x61, 0x6d, 0x65, 0x1a,
0x5d, 0x0a, 0x0d, 0x4d, 0x61, 0x70, 0x70, 0x69, 0x6e, 0x67, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79,
0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b,
0x65, 0x79, 0x12, 0x36, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28,
0x0b, 0x32, 0x20, 0x2e, 0x72, 0x65, 0x6d, 0x6f, 0x74, 0x65, 0x5f, 0x70, 0x62, 0x2e, 0x52, 0x65,
0x6d, 0x6f, 0x74, 0x65, 0x53, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x4c, 0x6f, 0x63, 0x61, 0x74,
0x69, 0x6f, 0x6e, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x3a, 0x02, 0x38, 0x01, 0x22, 0x57,
0x0a, 0x15, 0x52, 0x65, 0x6d, 0x6f, 0x74, 0x65, 0x53, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x4c,
0x6f, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18,
0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x12, 0x16, 0x0a, 0x06, 0x62,
0x75, 0x63, 0x6b, 0x65, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x62, 0x75, 0x63,
0x6b, 0x65, 0x74, 0x12, 0x12, 0x0a, 0x04, 0x70, 0x61, 0x74, 0x68, 0x18, 0x03, 0x20, 0x01, 0x28,
0x09, 0x52, 0x04, 0x70, 0x61, 0x74, 0x68, 0x42, 0x50, 0x0a, 0x10, 0x73, 0x65, 0x61, 0x77, 0x65,
0x65, 0x64, 0x66, 0x73, 0x2e, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x42, 0x0a, 0x46, 0x69, 0x6c,
0x65, 0x72, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x5a, 0x30, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e,
0x63, 0x6f, 0x6d, 0x2f, 0x63, 0x68, 0x72, 0x69, 0x73, 0x6c, 0x75, 0x73, 0x66, 0x2f, 0x73, 0x65,
0x61, 0x77, 0x65, 0x65, 0x64, 0x66, 0x73, 0x2f, 0x77, 0x65, 0x65, 0x64, 0x2f, 0x70, 0x62, 0x2f,
0x72, 0x65, 0x6d, 0x6f, 0x74, 0x65, 0x5f, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f,
0x33,
}
var (

15
weed/pb/server_address.go
View File

@@ -86,6 +86,14 @@ func (sa ServerAddresses) ToAddresses() (addresses []ServerAddress) {
return
}
func (sa ServerAddresses) ToAddressMap() (addresses map[string]ServerAddress) {
addresses = make(map[string]ServerAddress)
for _, address := range sa.ToAddresses() {
addresses[address.String()] = address
}
return
}
func (sa ServerAddresses) ToAddressStrings() (addresses []string) {
parts := strings.Split(string(sa), ",")
for _, address := range parts {
@@ -101,6 +109,13 @@ func ToAddressStrings(addresses []ServerAddress) []string {
}
return strings
}
func ToAddressStringsFromMap(addresses map[string]ServerAddress) []string {
var strings []string
for _, addr := range addresses {
strings = append(strings, string(addr))
}
return strings
}
func FromAddressStrings(strings []string) []ServerAddress {
var addresses []ServerAddress
for _, addr := range strings {

50
weed/remote_storage/s3/contabo.go Normal file
View File

@@ -0,0 +1,50 @@
package s3
import (
"fmt"
"os"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/aws/credentials"
"github.com/aws/aws-sdk-go/aws/session"
"github.com/aws/aws-sdk-go/service/s3"
"github.com/chrislusf/seaweedfs/weed/pb/remote_pb"
"github.com/chrislusf/seaweedfs/weed/remote_storage"
"github.com/chrislusf/seaweedfs/weed/util"
)
func init() {
remote_storage.RemoteStorageClientMakers["contabo"] = new(ContaboRemoteStorageMaker)
}
type ContaboRemoteStorageMaker struct{}
func (s ContaboRemoteStorageMaker) HasBucket() bool {
return true
}
func (s ContaboRemoteStorageMaker) Make(conf *remote_pb.RemoteConf) (remote_storage.RemoteStorageClient, error) {
client := &s3RemoteStorageClient{
conf: conf,
}
accessKey := util.Nvl(conf.ContaboAccessKey, os.Getenv("ACCESS_KEY"))
secretKey := util.Nvl(conf.ContaboSecretKey, os.Getenv("SECRET_KEY"))
config := &aws.Config{
Endpoint: aws.String(conf.ContaboEndpoint),
Region: aws.String(conf.ContaboRegion),
S3ForcePathStyle: aws.Bool(true),
S3DisableContentMD5Validation: aws.Bool(true),
}
if accessKey != "" && secretKey != "" {
config.Credentials = credentials.NewStaticCredentials(accessKey, secretKey, "")
}
sess, err := session.NewSession(config)
if err != nil {
return nil, fmt.Errorf("create contabo session: %v", err)
}
sess.Handlers.Build.PushFront(skipSha256PayloadSigning)
client.conn = s3.New(sess)
return client, nil
}

30
weed/s3api/filer_multipart.go
View File

@@ -5,6 +5,7 @@ import (
"fmt"
"github.com/chrislusf/seaweedfs/weed/s3api/s3err"
"path/filepath"
"sort"
"strconv"
"strings"
"time"
@@ -62,10 +63,15 @@ type CompleteMultipartUploadResult struct {
s3.CompleteMultipartUploadOutput
}
func (s3a *S3ApiServer) completeMultipartUpload(input *s3.CompleteMultipartUploadInput) (output *CompleteMultipartUploadResult, code s3err.ErrorCode) {
func (s3a *S3ApiServer) completeMultipartUpload(input *s3.CompleteMultipartUploadInput, parts *CompleteMultipartUpload) (output *CompleteMultipartUploadResult, code s3err.ErrorCode) {
glog.V(2).Infof("completeMultipartUpload input %v", input)
completedParts := parts.Parts
sort.Slice(completedParts, func(i, j int) bool {
return completedParts[i].PartNumber < completedParts[j].PartNumber
})
uploadDirectory := s3a.genUploadsFolder(*input.Bucket) + "/" + *input.UploadId
entries, _, err := s3a.list(uploadDirectory, "", "", false, maxPartsList)
@@ -80,14 +86,16 @@ func (s3a *S3ApiServer) completeMultipartUpload(input *s3.CompleteMultipartUploa
return nil, s3err.ErrNoSuchUpload
}
mime := pentry.Attributes.Mime
var finalParts []*filer_pb.FileChunk
var offset int64
var mime string
for _, entry := range entries {
if strings.HasSuffix(entry.Name, ".part") && !entry.IsDirectory {
if entry.Name == "0001.part" && entry.Attributes.Mime != "" {
mime = entry.Attributes.Mime
_, found := findByPartNumber(entry.Name, completedParts)
if !found {
continue
}
for _, chunk := range entry.Chunks {
p := &filer_pb.FileChunk{
@@ -156,6 +164,20 @@ func (s3a *S3ApiServer) completeMultipartUpload(input *s3.CompleteMultipartUploa
return
}
func findByPartNumber(fileName string, parts []CompletedPart) (etag string, found bool) {
partNumber, formatErr := strconv.Atoi(fileName[:4])
if formatErr != nil {
return
}
x := sort.Search(len(parts), func(i int) bool {
return parts[i].PartNumber >= partNumber
})
if parts[x].PartNumber != partNumber {
return
}
return parts[x].ETag, true
}
func (s3a *S3ApiServer) abortMultipartUpload(input *s3.AbortMultipartUploadInput) (output *s3.AbortMultipartUploadOutput, code s3err.ErrorCode) {
glog.V(2).Infof("abortMultipartUpload input %v", input)

83
weed/s3api/filer_multipart_test.go
View File

@@ -4,6 +4,7 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/s3"
"github.com/chrislusf/seaweedfs/weed/s3api/s3err"
"github.com/stretchr/testify/assert"
"testing"
"time"
)
@@ -48,3 +49,85 @@ func TestListPartsResult(t *testing.T) {
}
}
func Test_findByPartNumber(t *testing.T) {
type args struct {
fileName string
parts []CompletedPart
}
parts := []CompletedPart{
CompletedPart{
ETag: "xxx",
PartNumber: 1,
},
CompletedPart{
ETag: "yyy",
PartNumber: 3,
},
CompletedPart{
ETag: "zzz",
PartNumber: 5,
},
}
tests := []struct {
name string
args args
wantEtag string
wantFound bool
}{
{
"first",
args{
"0001.part",
parts,
},
"xxx",
true,
},
{
"second",
args{
"0002.part",
parts,
},
"",
false,
},
{
"third",
args{
"0003.part",
parts,
},
"yyy",
true,
},
{
"fourth",
args{
"0004.part",
parts,
},
"",
false,
},
{
"fifth",
args{
"0005.part",
parts,
},
"zzz",
true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
gotEtag, gotFound := findByPartNumber(tt.args.fileName, tt.args.parts)
assert.Equalf(t, tt.wantEtag, gotEtag, "findByPartNumber(%v, %v)", tt.args.fileName, tt.args.parts)
assert.Equalf(t, tt.wantFound, gotFound, "findByPartNumber(%v, %v)", tt.args.fileName, tt.args.parts)
})
}
}

9
weed/s3api/filer_util.go
View File

@@ -55,10 +55,11 @@ func (s3a *S3ApiServer) rm(parentDirectoryPath, entryName string, isDeleteData,
func doDeleteEntry(client filer_pb.SeaweedFilerClient, parentDirectoryPath string, entryName string, isDeleteData bool, isRecursive bool) error {
request := &filer_pb.DeleteEntryRequest{
Directory: parentDirectoryPath,
Name: entryName,
IsDeleteData: isDeleteData,
IsRecursive: isRecursive,
Directory: parentDirectoryPath,
Name: entryName,
IsDeleteData: isDeleteData,
IsRecursive: isRecursive,
IgnoreRecursiveError: true,
}
glog.V(1).Infof("delete entry %v/%v: %v", parentDirectoryPath, entryName, request)

36
weed/s3api/s3api_object_multipart_handlers.go
View File

@@ -1,11 +1,13 @@
package s3api
import (
"encoding/xml"
"fmt"
"github.com/chrislusf/seaweedfs/weed/glog"
xhttp "github.com/chrislusf/seaweedfs/weed/s3api/http"
"github.com/chrislusf/seaweedfs/weed/s3api/s3err"
weed_server "github.com/chrislusf/seaweedfs/weed/server"
"io"
"net/http"
"net/url"
"strconv"
@@ -56,8 +58,16 @@ func (s3a *S3ApiServer) NewMultipartUploadHandler(w http.ResponseWriter, r *http
// CompleteMultipartUploadHandler - Completes multipart upload.
func (s3a *S3ApiServer) CompleteMultipartUploadHandler(w http.ResponseWriter, r *http.Request) {
// https://docs.aws.amazon.com/AmazonS3/latest/API/API_CompleteMultipartUpload.html
bucket, object := xhttp.GetBucketAndObject(r)
parts := &CompleteMultipartUpload{}
if err := xmlDecoder(r.Body, parts, r.ContentLength); err != nil {
s3err.WriteErrorResponse(w, r, s3err.ErrMalformedXML)
return
}
// Get upload id.
uploadID, _, _, _ := getObjectResources(r.URL.Query())
@@ -65,7 +75,7 @@ func (s3a *S3ApiServer) CompleteMultipartUploadHandler(w http.ResponseWriter, r
Bucket: aws.String(bucket),
Key: objectKey(aws.String(object)),
UploadId: aws.String(uploadID),
})
}, parts)
glog.V(2).Info("CompleteMultipartUploadHandler", string(s3err.EncodeXMLResponse(response)), errCode)
@@ -268,8 +278,24 @@ func getObjectResources(values url.Values) (uploadID string, partNumberMarker, m
return
}
type byCompletedPartNumber []*s3.CompletedPart
func xmlDecoder(body io.Reader, v interface{}, size int64) error {
var lbody io.Reader
if size > 0 {
lbody = io.LimitReader(body, size)
} else {
lbody = body
}
d := xml.NewDecoder(lbody)
d.CharsetReader = func(label string, input io.Reader) (io.Reader, error) {
return input, nil
}
return d.Decode(v)
}
func (a byCompletedPartNumber) Len() int { return len(a) }
func (a byCompletedPartNumber) Swap(i, j int) { a[i], a[j] = a[j], a[i] }
func (a byCompletedPartNumber) Less(i, j int) bool { return *a[i].PartNumber < *a[j].PartNumber }
type CompleteMultipartUpload struct {
Parts []CompletedPart `xml:"Part"`
}
type CompletedPart struct {
ETag string
PartNumber int
}

2
weed/server/filer_grpc_server.go
View File

@@ -393,7 +393,7 @@ func (fs *FilerServer) GetFilerConfiguration(ctx context.Context, req *filer_pb.
clusterId, _ := fs.filer.Store.KvGet(context.Background(), []byte("clusterId"))
t := &filer_pb.GetFilerConfigurationResponse{
Masters: pb.ToAddressStrings(fs.option.Masters),
Masters: pb.ToAddressStringsFromMap(fs.option.Masters),
Collection: fs.option.Collection,
Replication: fs.option.DefaultReplication,
MaxMb: uint32(fs.option.MaxMB),

2
weed/server/filer_server.go
View File

@@ -49,7 +49,7 @@ import (
)
type FilerOption struct {
Masters []pb.ServerAddress
Masters map[string]pb.ServerAddress
Collection string
DefaultReplication string
DisableDirListing bool

4
weed/server/filer_server_handlers_tagging.go
View File

@@ -82,7 +82,9 @@ func (fs *FilerServer) DeleteTaggingHandler(w http.ResponseWriter, r *http.Reque
toDelete := strings.Split(r.URL.Query().Get("tagging"), ",")
deletions := make(map[string]struct{})
for _, deletion := range toDelete {
deletions[deletion] = struct{}{}
if deletion != "" {
deletions[deletion] = struct{}{}
}
}
// delete all tags or specific tags

2
weed/server/filer_server_handlers_write_autochunk.go
View File

@@ -131,7 +131,7 @@ func isAppend(r *http.Request) bool {
}
func skipCheckParentDirEntry(r *http.Request) bool {
return r.URL.Query().Get("skipCheckParentDir") != "true"
return r.URL.Query().Get("skipCheckParentDir") == "true"
}
func (fs *FilerServer) saveMetaData(ctx context.Context, r *http.Request, fileName string, contentType string, so *operation.StorageOption, md5bytes []byte, fileChunks []*filer_pb.FileChunk, chunkOffset int64, content []byte) (filerResult *FilerPostResult, replyerr error) {

2
weed/server/master_server.go
View File

@@ -75,7 +75,7 @@ type MasterServer struct {
Cluster *cluster.Cluster
}
func NewMasterServer(r *mux.Router, option *MasterOption, peers []pb.ServerAddress) *MasterServer {
func NewMasterServer(r *mux.Router, option *MasterOption, peers map[string]pb.ServerAddress) *MasterServer {
v := util.GetViper()
signingKey := v.GetString("jwt.signing.key")

40
weed/server/raft_server.go
View File

@@ -5,8 +5,6 @@ import (
"math/rand"
"os"
"path"
"sort"
"strings"
"time"
"google.golang.org/grpc"
@@ -21,7 +19,7 @@ import (
type RaftServerOption struct {
GrpcDialOption grpc.DialOption
Peers []pb.ServerAddress
Peers map[string]pb.ServerAddress
ServerAddr pb.ServerAddress
DataDir string
Topo *topology.Topology
@@ -31,7 +29,7 @@ type RaftServerOption struct {
}
type RaftServer struct {
peers []pb.ServerAddress // initial peers to join with
peers map[string]pb.ServerAddress // initial peers to join with
raftServer raft.Server
dataDir string
serverAddr pb.ServerAddress
@@ -81,10 +79,11 @@ func NewRaftServer(option *RaftServerOption) (*RaftServer, error) {
transporter := raft.NewGrpcTransporter(option.GrpcDialOption)
glog.V(0).Infof("Starting RaftServer with %v", option.ServerAddr)
// always clear previous log to avoid server is promotable
os.RemoveAll(path.Join(s.dataDir, "log"))
if !option.RaftResumeState {
// always clear previous metadata
os.RemoveAll(path.Join(s.dataDir, "conf"))
os.RemoveAll(path.Join(s.dataDir, "log"))
os.RemoveAll(path.Join(s.dataDir, "snapshot"))
}
if err := os.MkdirAll(path.Join(s.dataDir, "snapshot"), 0600); err != nil {
@@ -107,39 +106,26 @@ func NewRaftServer(option *RaftServerOption) (*RaftServer, error) {
return nil, err
}
for _, peer := range s.peers {
if err := s.raftServer.AddPeer(string(peer), peer.ToGrpcAddress()); err != nil {
for name, peer := range s.peers {
if err := s.raftServer.AddPeer(name, peer.ToGrpcAddress()); err != nil {
return nil, err
}
}
// Remove deleted peers
for existsPeerName := range s.raftServer.Peers() {
exists := false
var existingPeer pb.ServerAddress
for _, peer := range s.peers {
if peer.ToGrpcAddress() == existsPeerName {
exists, existingPeer = true, peer
break
}
}
if exists {
if existingPeer, found := s.peers[existsPeerName]; !found {
if err := s.raftServer.RemovePeer(existsPeerName); err != nil {
glog.V(0).Infoln(err)
return nil, err
} else {
glog.V(0).Infof("removing old peer %s", existingPeer)
glog.V(0).Infof("removing old peer: %s", existingPeer)
}
}
}
s.GrpcServer = raft.NewGrpcServer(s.raftServer)
if s.raftServer.IsLogEmpty() && isTheFirstOne(option.ServerAddr, s.peers) {
// Initialize the server by joining itself.
// s.DoJoinCommand()
}
glog.V(0).Infof("current cluster leader: %v", s.raftServer.Leader())
return s, nil
@@ -155,16 +141,6 @@ func (s *RaftServer) Peers() (members []string) {
return
}
func isTheFirstOne(self pb.ServerAddress, peers []pb.ServerAddress) bool {
sort.Slice(peers, func(i, j int) bool {
return strings.Compare(string(peers[i]), string(peers[j])) < 0
})
if len(peers) <= 0 {
return true
}
return self == peers[0]
}
func (s *RaftServer) DoJoinCommand() {
glog.V(0).Infoln("Initializing new cluster")

2
weed/server/volume_server.go
View File

@@ -23,7 +23,6 @@ type VolumeServer struct {
inFlightDownloadDataSize int64
concurrentUploadLimit int64
concurrentDownloadLimit int64
inFlightUploadDataLimitCond *sync.Cond
inFlightDownloadDataLimitCond *sync.Cond
SeedMasterNodes []pb.ServerAddress
@@ -84,7 +83,6 @@ func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string,
fileSizeLimitBytes: int64(fileSizeLimitMB) * 1024 * 1024,
isHeartbeating: true,
stopChan: make(chan bool),
inFlightUploadDataLimitCond: sync.NewCond(new(sync.Mutex)),
inFlightDownloadDataLimitCond: sync.NewCond(new(sync.Mutex)),
concurrentUploadLimit: concurrentUploadLimit,
concurrentDownloadLimit: concurrentDownloadLimit,

25
weed/server/volume_server_handlers.go
View File

@@ -1,6 +1,7 @@
package weed_server
import (
"fmt"
"net/http"
"strconv"
"strings"
@@ -39,8 +40,14 @@ func (vs *VolumeServer) privateStoreHandler(w http.ResponseWriter, r *http.Reque
stats.ReadRequest()
vs.inFlightDownloadDataLimitCond.L.Lock()
for vs.concurrentDownloadLimit != 0 && atomic.LoadInt64(&vs.inFlightDownloadDataSize) > vs.concurrentDownloadLimit {
glog.V(4).Infof("wait because inflight download data %d > %d", vs.inFlightDownloadDataSize, vs.concurrentDownloadLimit)
vs.inFlightDownloadDataLimitCond.Wait()
select {
case <-r.Context().Done():
glog.V(4).Infof("request cancelled from %s: %v", r.RemoteAddr, r.Context().Err())
return
default:
glog.V(4).Infof("wait because inflight download data %d > %d", vs.inFlightDownloadDataSize, vs.concurrentDownloadLimit)
vs.inFlightDownloadDataLimitCond.Wait()
}
}
vs.inFlightDownloadDataLimitCond.L.Unlock()
vs.GetOrHeadHandler(w, r)
@@ -51,16 +58,18 @@ func (vs *VolumeServer) privateStoreHandler(w http.ResponseWriter, r *http.Reque
// wait until in flight data is less than the limit
contentLength := getContentLength(r)
vs.inFlightUploadDataLimitCond.L.Lock()
for vs.concurrentUploadLimit != 0 && atomic.LoadInt64(&vs.inFlightUploadDataSize) > vs.concurrentUploadLimit {
glog.V(4).Infof("wait because inflight upload data %d > %d", vs.inFlightUploadDataSize, vs.concurrentUploadLimit)
vs.inFlightUploadDataLimitCond.Wait()
// exclude the replication from the concurrentUploadLimitMB
if vs.concurrentUploadLimit != 0 && r.URL.Query().Get("type") != "replicate" &&
atomic.LoadInt64(&vs.inFlightUploadDataSize) > vs.concurrentUploadLimit {
err := fmt.Errorf("reject because inflight upload data %d > %d", vs.inFlightUploadDataSize, vs.concurrentUploadLimit)
glog.V(1).Infof("too many requests: %v", err)
writeJsonError(w, r, http.StatusTooManyRequests, err)
return
}
vs.inFlightUploadDataLimitCond.L.Unlock()
atomic.AddInt64(&vs.inFlightUploadDataSize, contentLength)
defer func() {
atomic.AddInt64(&vs.inFlightUploadDataSize, -contentLength)
vs.inFlightUploadDataLimitCond.Signal()
}()
// processs uploads

2
weed/shell/commands.go
View File

@@ -46,7 +46,7 @@ var (
func NewCommandEnv(options *ShellOptions) *CommandEnv {
ce := &CommandEnv{
env: make(map[string]string),
MasterClient: wdclient.NewMasterClient(options.GrpcDialOption, pb.AdminShellClient, "", "", pb.ServerAddresses(*options.Masters).ToAddresses()),
MasterClient: wdclient.NewMasterClient(options.GrpcDialOption, pb.AdminShellClient, "", "", pb.ServerAddresses(*options.Masters).ToAddressMap()),
option: options,
}
ce.locker = exclusive_locks.NewExclusiveLocker(ce.MasterClient, "admin")

2
weed/storage/erasure_coding/ec_encoder.go
View File

@@ -220,7 +220,7 @@ func encodeDatFile(remainingSize int64, err error, baseFileName string, bufferSi
processedSize += largeBlockSize * DataShardsCount
}
for remainingSize > 0 {
encodeData(file, enc, processedSize, smallBlockSize, buffers, outputs)
err = encodeData(file, enc, processedSize, smallBlockSize, buffers, outputs)
if err != nil {
return fmt.Errorf("failed to encode small chunk data: %v", err)
}

13
weed/topology/volume_layout.go
View File

@@ -140,9 +140,7 @@ func NewVolumeLayout(rp *super_block.ReplicaPlacement, ttl *needle.TTL, diskType
}
func (vl *VolumeLayout) String() string {
vl.accessLock.RLock()
defer vl.accessLock.RUnlock()
return fmt.Sprintf("rp:%v, ttl:%v, vid2location:%v, writables:%v, volumeSizeLimit:%v", vl.rp, vl.ttl, vl.vid2location, vl.writables, vl.volumeSizeLimit)
return fmt.Sprintf("rp:%v, ttl:%v, writables:%v, volumeSizeLimit:%v", vl.rp, vl.ttl, vl.writables, vl.volumeSizeLimit)
}
func (vl *VolumeLayout) RegisterVolume(v *storage.VolumeInfo, dn *DataNode) {
@@ -220,6 +218,13 @@ func (vl *VolumeLayout) ensureCorrectWritables(vid needle.VolumeId) {
vl.setVolumeWritable(vid)
}
} else {
if !vl.enoughCopies(vid) {
glog.V(0).Infof("volume %d does not have enough copies", vid)
}
if !vl.isAllWritable(vid) {
glog.V(0).Infof("volume %d are not all writable", vid)
}
glog.V(0).Infof("volume %d remove from writable", vid)
vl.removeFromWritable(vid)
}
}
@@ -435,7 +440,7 @@ func (vl *VolumeLayout) SetVolumeCapacityFull(vid needle.VolumeId) bool {
vl.accessLock.Lock()
defer vl.accessLock.Unlock()
// glog.V(0).Infoln("Volume", vid, "reaches full capacity.")
glog.V(0).Infof("Volume %d reaches full capacity.", vid)
return vl.removeFromWritable(vid)
}

25
weed/util/config.go
View File

@@ -9,6 +9,20 @@ import (
"github.com/chrislusf/seaweedfs/weed/glog"
)
var (
ConfigurationFileDirectory DirectoryValueType
)
type DirectoryValueType string
func (s *DirectoryValueType) Set(value string) error {
*s = DirectoryValueType(value)
return nil
}
func (s *DirectoryValueType) String() string {
return string(*s)
}
type Configuration interface {
GetString(key string) string
GetBool(key string) bool
@@ -20,11 +34,12 @@ type Configuration interface {
func LoadConfiguration(configFileName string, required bool) (loaded bool) {
// find a filer store
viper.SetConfigName(configFileName) // name of config file (without extension)
viper.AddConfigPath(".") // optionally look for config in the working directory
viper.AddConfigPath("$HOME/.seaweedfs") // call multiple times to add many search paths
viper.AddConfigPath("/usr/local/etc/seaweedfs/") // search path for bsd-style config directory in
viper.AddConfigPath("/etc/seaweedfs/") // path to look for the config file in
viper.SetConfigName(configFileName) // name of config file (without extension)
viper.AddConfigPath(ResolvePath(ConfigurationFileDirectory.String())) // path to look for the config file in
viper.AddConfigPath(".") // optionally look for config in the working directory
viper.AddConfigPath("$HOME/.seaweedfs") // call multiple times to add many search paths
viper.AddConfigPath("/usr/local/etc/seaweedfs/") // search path for bsd-style config directory in
viper.AddConfigPath("/etc/seaweedfs/") // path to look for the config file in
if err := viper.MergeInConfig(); err != nil { // Handle errors reading the config file
if strings.Contains(err.Error(), "Not Found") {

2
weed/util/constants.go
View File

@@ -5,7 +5,7 @@ import (
)
var (
VERSION_NUMBER = fmt.Sprintf("%.02f", 2.94)
VERSION_NUMBER = fmt.Sprintf("%.02f", 2.96)
VERSION = sizeLimit + " " + VERSION_NUMBER
COMMIT = ""
)

2
weed/util/net_timeout.go
View File

@@ -109,7 +109,7 @@ func NewIpAndLocalListeners(host string, port int, timeout time.Duration) (ipLis
WriteTimeout: timeout,
}
if host != "localhost" && host != "" {
if host != "localhost" && host != "" && host != "0.0.0.0" && host != "127.0.0.1" {
listner, err = net.Listen("tcp", JoinHostPort("localhost", port))
if err != nil {
return

4
weed/wdclient/masterclient.go
View File

@@ -18,7 +18,7 @@ type MasterClient struct {
clientType string
clientHost pb.ServerAddress
currentMaster pb.ServerAddress
masters []pb.ServerAddress
masters map[string]pb.ServerAddress
grpcDialOption grpc.DialOption
vidMap
@@ -26,7 +26,7 @@ type MasterClient struct {
OnPeerUpdate func(update *master_pb.ClusterNodeUpdate)
}
func NewMasterClient(grpcDialOption grpc.DialOption, clientType string, clientHost pb.ServerAddress, clientDataCenter string, masters []pb.ServerAddress) *MasterClient {
func NewMasterClient(grpcDialOption grpc.DialOption, clientType string, clientHost pb.ServerAddress, clientDataCenter string, masters map[string]pb.ServerAddress) *MasterClient {
return &MasterClient{
clientType: clientType,
clientHost: clientHost,

3
weed/weed.go
View File

@@ -4,6 +4,7 @@ import (
"embed"
"fmt"
weed_server "github.com/chrislusf/seaweedfs/weed/server"
"github.com/chrislusf/seaweedfs/weed/util"
flag "github.com/chrislusf/seaweedfs/weed/util/fla9"
"io"
"io/fs"
@@ -40,6 +41,8 @@ var static embed.FS
func init() {
weed_server.StaticFS, _ = fs.Sub(static, "static")
flag.Var(&util.ConfigurationFileDirectory, "config_dir", "directory with toml configuration files")
}
func main() {