lsm/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2025-10-21 15:40:03 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

filer master start up with default ip address instead of just localhost

Browse Source
This commit is contained in:
Chris Lu
2020-04-18 15:17:27 -07:00
parent 5d346d44bd
commit 076c8bd3bc
16 changed files with 229 additions and 153 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
weed/messaging/broker/broker_grpc_server_publish.go
View File

@@ -27,6 +27,19 @@ func (broker *MessageBroker) Publish(stream messaging_pb.SeaweedMessaging_Publis
topicConfig := &messaging_pb.TopicConfiguration{
}
// send init response
initResponse := &messaging_pb.PublishResponse{
Config: nil,
Redirect: nil,
}
err = stream.Send(initResponse)
if err != nil {
return err
}
if initResponse.Redirect != nil {
return nil
}
// get lock
tp := TopicPartition{
@@ -87,6 +100,8 @@ func (broker *MessageBroker) Publish(stream messaging_pb.SeaweedMessaging_Publis
Headers: in.Data.Headers,
}
println("received message:", string(in.Data.Value))
data, err := proto.Marshal(m)
if err != nil {
glog.Errorf("marshall error: %v\n", err)

85
weed/messaging/broker/broker_server.go
View File

@@ -2,11 +2,11 @@ package broker
import (
"context"
"fmt"
"time"
"google.golang.org/grpc"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/pb"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
@@ -34,7 +34,9 @@ func NewMessageBroker(option *MessageBrokerOption, grpcDialOption grpc.DialOptio
topicLocks: NewTopicLocks(),
}
go messageBroker.loopForEver()
messageBroker.checkPeers()
// go messageBroker.loopForEver()
return messageBroker, nil
}
@@ -52,58 +54,55 @@ func (broker *MessageBroker) checkPeers() {
// contact a filer about masters
var masters []string
for _, filer := range broker.option.Filers {
err := broker.withFilerClient(filer, func(client filer_pb.SeaweedFilerClient) error {
resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{})
if err != nil {
return err
found := false
for !found {
for _, filer := range broker.option.Filers {
err := broker.withFilerClient(filer, func(client filer_pb.SeaweedFilerClient) error {
resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{})
if err != nil {
return err
}
masters = append(masters, resp.Masters...)
return nil
})
if err == nil {
found = true
break
}
masters = append(masters, resp.Masters...)
return nil
})
if err != nil {
fmt.Printf("failed to read masters from %+v: %v\n", broker.option.Filers, err)
return
glog.V(0).Infof("failed to read masters from %+v: %v", broker.option.Filers, err)
time.Sleep(time.Second)
}
}
glog.V(0).Infof("received master list: %s", masters)
// contact each masters for filers
var filers []string
for _, master := range masters {
err := broker.withMasterClient(master, func(client master_pb.SeaweedClient) error {
resp, err := client.ListMasterClients(context.Background(), &master_pb.ListMasterClientsRequest{
ClientType: "filer",
found = false
for !found {
for _, master := range masters {
err := broker.withMasterClient(master, func(client master_pb.SeaweedClient) error {
resp, err := client.ListMasterClients(context.Background(), &master_pb.ListMasterClientsRequest{
ClientType: "filer",
})
if err != nil {
return err
}
filers = append(filers, resp.GrpcAddresses...)
return nil
})
if err != nil {
return err
if err == nil {
found = true
break
}
fmt.Printf("filers: %+v\n", resp.GrpcAddresses)
filers = append(filers, resp.GrpcAddresses...)
return nil
})
if err != nil {
fmt.Printf("failed to list filers: %v\n", err)
return
glog.V(0).Infof("failed to list filers: %v", err)
time.Sleep(time.Second)
}
}
glog.V(0).Infof("received filer list: %s", filers)
// contact each filer about brokers
for _, filer := range filers {
err := broker.withFilerClient(filer, func(client filer_pb.SeaweedFilerClient) error {
resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{})
if err != nil {
return err
}
masters = append(masters, resp.Masters...)
return nil
})
if err != nil {
fmt.Printf("failed to read masters from %+v: %v\n", broker.option.Filers, err)
return
}
}
broker.option.Filers = filers
}