separate into notification.toml, add gcp pub/sub message queue

This commit is contained in:
Chris Lu 2018-11-01 01:11:09 -07:00
parent 6219a9ad1f
commit db584ff7f8
7 changed files with 240 additions and 29 deletions

View file

@ -36,6 +36,7 @@ var cmdFilerReplicate = &Command{
func runFilerReplicate(cmd *Command, args []string) bool { func runFilerReplicate(cmd *Command, args []string) bool {
weed_server.LoadConfiguration("replication", true) weed_server.LoadConfiguration("replication", true)
weed_server.LoadConfiguration("notification", true)
config := viper.GetViper() config := viper.GetViper()
var notificationInput sub.NotificationInput var notificationInput sub.NotificationInput
@ -54,7 +55,8 @@ func runFilerReplicate(cmd *Command, args []string) bool {
} }
if notificationInput == nil { if notificationInput == nil {
println("Please follow 'weed scaffold -config=repliaction' to see example notification configurations.") println("No notification is defined in notification.toml file.")
println("Please follow 'weed scaffold -config=notification' to see example notification configurations.")
return true return true
} }
@ -85,7 +87,7 @@ func runFilerReplicate(cmd *Command, args []string) bool {
} }
if dataSink == nil { if dataSink == nil {
println("no data sink configured:") println("no data sink configured in replication.toml:")
for _, sk := range sink.Sinks { for _, sk := range sink.Sinks {
println(" " + sk.GetName()) println(" " + sk.GetName())
} }

View file

@ -19,7 +19,7 @@ var cmdScaffold = &Command{
var ( var (
outputPath = cmdScaffold.Flag.String("output", "", "if not empty, save the configuration file to this directory") outputPath = cmdScaffold.Flag.String("output", "", "if not empty, save the configuration file to this directory")
config = cmdScaffold.Flag.String("config", "filer", "[filer|replication] the configuration file to generate") config = cmdScaffold.Flag.String("config", "filer", "[filer|notification|replication] the configuration file to generate")
) )
func runScaffold(cmd *Command, args []string) bool { func runScaffold(cmd *Command, args []string) bool {
@ -28,6 +28,8 @@ func runScaffold(cmd *Command, args []string) bool {
switch *config { switch *config {
case "filer": case "filer":
content = FILER_TOML_EXAMPLE content = FILER_TOML_EXAMPLE
case "notification":
content = NOTIFICATION_TOML_EXAMPLE
case "replication": case "replication":
content = REPLICATION_TOML_EXAMPLE content = REPLICATION_TOML_EXAMPLE
} }
@ -37,7 +39,7 @@ func runScaffold(cmd *Command, args []string) bool {
} }
if *outputPath != "" { if *outputPath != "" {
ioutil.WriteFile(filepath.Join(*outputPath, *config+".toml"), []byte(content), 0x755) ioutil.WriteFile(filepath.Join(*outputPath, *config+".toml"), []byte(content), 0644)
} else { } else {
println(content) println(content)
} }
@ -131,20 +133,34 @@ addresses = [
"localhost:30006", "localhost:30006",
] ]
`
NOTIFICATION_TOML_EXAMPLE = `
# A sample TOML config file for SeaweedFS filer store
# Used by both "weed filer" or "weed server -filer" and "weed filer.replicate"
# Put this file to one of the location, with descending priority
# ./notification.toml
# $HOME/.seaweedfs/notification.toml
# /etc/seaweedfs/notification.toml
#################################################### ####################################################
# notification # notification
# sends filer updates for each file to an external message queue # send and receive filer updates for each file to an external message queue
#################################################### ####################################################
[notification.log] [notification.log]
# this is only for debugging perpose and does not work with "weed filer.replicate"
enabled = false enabled = false
[notification.kafka] [notification.kafka]
enabled = false enabled = false
hosts = [ hosts = [
"localhost:9092" "localhost:9092"
] ]
topic = "seaweedfs_filer" topic = "seaweedfs_filer"
offsetFile = "./last.offset"
offsetSaveIntervalSeconds = 10
[notification.aws_sqs] [notification.aws_sqs]
# experimental, let me know if it works # experimental, let me know if it works
@ -154,7 +170,16 @@ aws_secret_access_key = "" # if empty, loads from the shared credentials
region = "us-east-2" region = "us-east-2"
sqs_queue_name = "my_filer_queue" # an existing queue name sqs_queue_name = "my_filer_queue" # an existing queue name
[notification.google_pub_sub]
# read credentials doc at https://cloud.google.com/docs/authentication/getting-started
enabled = false
google_application_credentials = "/path/to/x.json" # path to json credential file
project_id = "" # an existing project id
topic = "seaweedfs_filer_topic" # a topic, auto created if does not exists
` `
REPLICATION_TOML_EXAMPLE = ` REPLICATION_TOML_EXAMPLE = `
# A sample TOML config file for replicating SeaweedFS filer # A sample TOML config file for replicating SeaweedFS filer
# Used with "weed filer.replicate" # Used with "weed filer.replicate"
@ -168,22 +193,6 @@ enabled = true
grpcAddress = "localhost:18888" grpcAddress = "localhost:18888"
directory = "/buckets" # all files under this directory tree are replicated directory = "/buckets" # all files under this directory tree are replicated
[notification.kafka]
enabled = false
hosts = [
"localhost:9092"
]
topic = "seaweedfs_filer1_to_filer2"
offsetFile = "./last.offset"
offsetSaveIntervalSeconds = 10
[notification.aws_sqs]
enabled = false
aws_access_key_id = "" # if empty, loads from the shared credentials file (~/.aws/credentials).
aws_secret_access_key = "" # if empty, loads from the shared credentials file (~/.aws/credentials).
region = "us-east-2"
sqs_queue_name = "my_filer_queue" # an existing queue name
[sink.filer] [sink.filer]
enabled = false enabled = false
grpcAddress = "localhost:18888" grpcAddress = "localhost:18888"

View file

@ -35,7 +35,7 @@ func LoadConfiguration(config *viper.Viper) {
store.GetName(), err) store.GetName(), err)
} }
Queue = store Queue = store
glog.V(0).Infof("Configure message queue for %s", store.GetName()) glog.V(0).Infof("Configure notification message queue for %s", store.GetName())
return return
} }
} }

View file

@ -0,0 +1,89 @@
package google_pub_sub
import (
"fmt"
"os"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/notification"
"github.com/chrislusf/seaweedfs/weed/util"
"github.com/golang/protobuf/proto"
"google.golang.org/api/option"
"context"
"cloud.google.com/go/pubsub"
)
func init() {
notification.MessageQueues = append(notification.MessageQueues, &GooglePubSub{})
}
type GooglePubSub struct {
topic *pubsub.Topic
}
func (k *GooglePubSub) GetName() string {
return "google_pub_sub"
}
func (k *GooglePubSub) Initialize(configuration util.Configuration) (err error) {
glog.V(0).Infof("notification.google_pub_sub.project_id: %v", configuration.GetString("project_id"))
glog.V(0).Infof("notification.google_pub_sub.topic: %v", configuration.GetString("topic"))
return k.initialize(
configuration.GetString("google_application_credentials"),
configuration.GetString("project_id"),
configuration.GetString("topic"),
)
}
func (k *GooglePubSub) initialize(google_application_credentials, projectId, topicName string) (err error) {
ctx := context.Background()
// Creates a client.
if google_application_credentials == "" {
var found bool
google_application_credentials, found = os.LookupEnv("GOOGLE_APPLICATION_CREDENTIALS")
if !found {
glog.Fatalf("need to specific GOOGLE_APPLICATION_CREDENTIALS env variable or google_application_credentials in filer.toml")
}
}
client, err := pubsub.NewClient(ctx, projectId, option.WithCredentialsFile(google_application_credentials))
if err != nil {
glog.Fatalf("Failed to create client: %v", err)
}
k.topic = client.Topic(topicName)
if exists, err := k.topic.Exists(ctx); err == nil {
if !exists {
k.topic, err = client.CreateTopic(ctx, topicName)
if err != nil {
glog.Fatalf("Failed to create topic %s: %v", topicName, err)
}
}
} else {
glog.Fatalf("Failed to check topic %s: %v", topicName, err)
}
return nil
}
func (k *GooglePubSub) SendMessage(key string, message proto.Message) (err error) {
bytes, err := proto.Marshal(message)
if err != nil {
return
}
ctx := context.Background()
result := k.topic.Publish(ctx, &pubsub.Message{
Data: bytes,
Attributes: map[string]string{"key": key},
})
_, err = result.Get(ctx)
if err != nil {
return fmt.Errorf("send message to google pub sub %s: %v", k.topic.String(), err)
}
return nil
}

View file

@ -3,11 +3,11 @@ package gcssink
import ( import (
"context" "context"
"fmt" "fmt"
"log"
"os" "os"
"cloud.google.com/go/storage" "cloud.google.com/go/storage"
"github.com/chrislusf/seaweedfs/weed/filer2" "github.com/chrislusf/seaweedfs/weed/filer2"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb" "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
"github.com/chrislusf/seaweedfs/weed/replication/sink" "github.com/chrislusf/seaweedfs/weed/replication/sink"
"github.com/chrislusf/seaweedfs/weed/replication/source" "github.com/chrislusf/seaweedfs/weed/replication/source"
@ -56,12 +56,12 @@ func (g *GcsSink) initialize(google_application_credentials, bucketName, dir str
var found bool var found bool
google_application_credentials, found = os.LookupEnv("GOOGLE_APPLICATION_CREDENTIALS") google_application_credentials, found = os.LookupEnv("GOOGLE_APPLICATION_CREDENTIALS")
if !found { if !found {
log.Fatalf("need to specific GOOGLE_APPLICATION_CREDENTIALS env variable or google_application_credentials in replication.toml") glog.Fatalf("need to specific GOOGLE_APPLICATION_CREDENTIALS env variable or google_application_credentials in replication.toml")
} }
} }
client, err := storage.NewClient(ctx, option.WithCredentialsFile(google_application_credentials)) client, err := storage.NewClient(ctx, option.WithCredentialsFile(google_application_credentials))
if err != nil { if err != nil {
log.Fatalf("Failed to create client: %v", err) glog.Fatalf("Failed to create client: %v", err)
} }
g.client = client g.client = client

View file

@ -0,0 +1,109 @@
package sub
import (
"context"
"fmt"
"os"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
"github.com/chrislusf/seaweedfs/weed/util"
"github.com/golang/protobuf/proto"
"cloud.google.com/go/pubsub"
"google.golang.org/api/option"
)
func init() {
NotificationInputs = append(NotificationInputs, &GooglePubSubInput{})
}
type GooglePubSubInput struct {
sub *pubsub.Subscription
topicName string
messageChan chan *pubsub.Message
}
func (k *GooglePubSubInput) GetName() string {
return "google_pub_sub"
}
func (k *GooglePubSubInput) Initialize(configuration util.Configuration) error {
glog.V(0).Infof("notification.google_pub_sub.project_id: %v", configuration.GetString("project_id"))
glog.V(0).Infof("notification.google_pub_sub.topic: %v", configuration.GetString("topic"))
return k.initialize(
configuration.GetString("google_application_credentials"),
configuration.GetString("project_id"),
configuration.GetString("topic"),
)
}
func (k *GooglePubSubInput) initialize(google_application_credentials, projectId, topicName string) (err error) {
ctx := context.Background()
// Creates a client.
if google_application_credentials == "" {
var found bool
google_application_credentials, found = os.LookupEnv("GOOGLE_APPLICATION_CREDENTIALS")
if !found {
glog.Fatalf("need to specific GOOGLE_APPLICATION_CREDENTIALS env variable or google_application_credentials in filer.toml")
}
}
client, err := pubsub.NewClient(ctx, projectId, option.WithCredentialsFile(google_application_credentials))
if err != nil {
glog.Fatalf("Failed to create client: %v", err)
}
k.topicName = topicName
topic := client.Topic(topicName)
if exists, err := topic.Exists(ctx); err == nil {
if !exists {
topic, err = client.CreateTopic(ctx, topicName)
if err != nil {
glog.Fatalf("Failed to create topic %s: %v", topicName, err)
}
}
} else {
glog.Fatalf("Failed to check topic %s: %v", topicName, err)
}
subscriptionName := "seaweedfs_sub"
k.sub = client.Subscription(subscriptionName)
if exists, err := k.sub.Exists(ctx); err == nil {
if !exists {
k.sub, err = client.CreateSubscription(ctx, subscriptionName, pubsub.SubscriptionConfig{Topic: topic})
if err != nil {
glog.Fatalf("Failed to create subscription %s: %v", subscriptionName, err)
}
}
} else {
glog.Fatalf("Failed to check subscription %s: %v", topicName, err)
}
k.messageChan = make(chan *pubsub.Message, 1)
go k.sub.Receive(ctx, func(ctx context.Context, m *pubsub.Message) {
k.messageChan <- m
m.Ack()
})
return err
}
func (k *GooglePubSubInput) ReceiveMessage() (key string, message *filer_pb.EventNotification, err error) {
m := <-k.messageChan
// process the message
key = m.Attributes["key"]
message = &filer_pb.EventNotification{}
err = proto.Unmarshal(m.Data, message)
if err != nil {
err = fmt.Errorf("unmarshal message from google pubsub %s: %v", k.topicName, err)
return
}
return
}

View file

@ -14,6 +14,7 @@ import (
"github.com/chrislusf/seaweedfs/weed/notification" "github.com/chrislusf/seaweedfs/weed/notification"
_ "github.com/chrislusf/seaweedfs/weed/notification/kafka" _ "github.com/chrislusf/seaweedfs/weed/notification/kafka"
_ "github.com/chrislusf/seaweedfs/weed/notification/aws_sqs" _ "github.com/chrislusf/seaweedfs/weed/notification/aws_sqs"
_ "github.com/chrislusf/seaweedfs/weed/notification/google_pub_sub"
_ "github.com/chrislusf/seaweedfs/weed/notification/log" _ "github.com/chrislusf/seaweedfs/weed/notification/log"
"github.com/chrislusf/seaweedfs/weed/security" "github.com/chrislusf/seaweedfs/weed/security"
"github.com/spf13/viper" "github.com/spf13/viper"
@ -52,6 +53,7 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
go fs.filer.KeepConnectedToMaster() go fs.filer.KeepConnectedToMaster()
LoadConfiguration("filer", true) LoadConfiguration("filer", true)
LoadConfiguration("notification", false)
v := viper.GetViper() v := viper.GetViper()
fs.filer.LoadConfiguration(v) fs.filer.LoadConfiguration(v)
@ -81,15 +83,15 @@ func LoadConfiguration(configFileName string, required bool) {
glog.V(0).Infof("Reading %s.toml from %s", configFileName, viper.ConfigFileUsed()) glog.V(0).Infof("Reading %s.toml from %s", configFileName, viper.ConfigFileUsed())
if err := viper.ReadInConfig(); err != nil { // Handle errors reading the config file if err := viper.MergeInConfig(); err != nil { // Handle errors reading the config file
glog.V(0).Infof("Reading %s: %v", viper.ConfigFileUsed(), err) glog.V(0).Infof("Reading %s: %v", viper.ConfigFileUsed(), err)
if required { if required {
glog.Errorf("Failed to load %s.toml file from current directory, or $HOME/.seaweedfs/, or /etc/seaweedfs/"+ glog.Fatalf("Failed to load %s.toml file from current directory, or $HOME/.seaweedfs/, or /etc/seaweedfs/"+
"\n\nPlease follow this example and add a filer.toml file to "+ "\n\nPlease follow this example and add a filer.toml file to "+
"current directory, or $HOME/.seaweedfs/, or /etc/seaweedfs/:\n"+ "current directory, or $HOME/.seaweedfs/, or /etc/seaweedfs/:\n"+
" https://github.com/chrislusf/seaweedfs/blob/master/weed/%s.toml\n"+ " https://github.com/chrislusf/seaweedfs/blob/master/weed/%s.toml\n"+
"\n\nOr use this command to generate the default toml file\n"+ "\nOr use this command to generate the default toml file\n"+
" weed scaffold -config=%s -output=.\n", " weed scaffold -config=%s -output=.\n\n\n",
configFileName, configFileName, configFileName) configFileName, configFileName, configFileName)
} }
} }