support env variables to overwrite toml file

This commit is contained in:
Chris Lu 2020-01-29 09:09:55 -08:00
parent 27b94cb65b
commit d335f04de6
60 changed files with 268 additions and 247 deletions

2
go.mod
View file

@ -69,7 +69,7 @@ require (
github.com/spf13/jwalterweatherman v1.1.0 // indirect github.com/spf13/jwalterweatherman v1.1.0 // indirect
github.com/spf13/viper v1.4.0 github.com/spf13/viper v1.4.0
github.com/streadway/amqp v0.0.0-20190827072141-edfb9018d271 // indirect github.com/streadway/amqp v0.0.0-20190827072141-edfb9018d271 // indirect
github.com/stretchr/testify v1.4.0 // indirect github.com/stretchr/testify v1.4.0
github.com/syndtr/goleveldb v1.0.0 github.com/syndtr/goleveldb v1.0.0
github.com/tidwall/gjson v1.3.2 github.com/tidwall/gjson v1.3.2
github.com/tidwall/match v1.0.1 github.com/tidwall/match v1.0.1

View file

@ -7,10 +7,8 @@ import (
"log" "log"
"math/rand" "math/rand"
"github.com/chrislusf/seaweedfs/weed/security"
"github.com/spf13/viper"
"github.com/chrislusf/seaweedfs/weed/operation" "github.com/chrislusf/seaweedfs/weed/operation"
"github.com/chrislusf/seaweedfs/weed/security"
"github.com/chrislusf/seaweedfs/weed/util" "github.com/chrislusf/seaweedfs/weed/util"
) )
@ -23,7 +21,7 @@ func main() {
flag.Parse() flag.Parse()
util.LoadConfiguration("security", false) util.LoadConfiguration("security", false)
grpcDialOption := security.LoadClientTLS(viper.Sub("grpc"), "client") grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.client")
for i := 0; i < *repeat; i++ { for i := 0; i < *repeat; i++ {
assignResult, err := operation.Assign(*master, grpcDialOption, &operation.VolumeAssignRequest{Count: 1}) assignResult, err := operation.Assign(*master, grpcDialOption, &operation.VolumeAssignRequest{Count: 1})

View file

@ -25,7 +25,7 @@ func main() {
flag.Parse() flag.Parse()
util2.LoadConfiguration("security", false) util2.LoadConfiguration("security", false)
grpcDialOption := security.LoadClientTLS(viper.Sub("grpc"), "client") grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.client")
vid := needle.VolumeId(*volumeId) vid := needle.VolumeId(*volumeId)

View file

@ -3,8 +3,6 @@ package command
import ( import (
"fmt" "fmt"
"github.com/spf13/viper"
"github.com/chrislusf/seaweedfs/weed/security" "github.com/chrislusf/seaweedfs/weed/security"
"github.com/chrislusf/seaweedfs/weed/storage/needle" "github.com/chrislusf/seaweedfs/weed/storage/needle"
"github.com/chrislusf/seaweedfs/weed/storage/super_block" "github.com/chrislusf/seaweedfs/weed/storage/super_block"
@ -66,7 +64,7 @@ var cmdBackup = &Command{
func runBackup(cmd *Command, args []string) bool { func runBackup(cmd *Command, args []string) bool {
util.LoadConfiguration("security", false) util.LoadConfiguration("security", false)
grpcDialOption := security.LoadClientTLS(viper.Sub("grpc"), "client") grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.client")
if *s.volumeId == -1 { if *s.volumeId == -1 {
return false return false

View file

@ -15,7 +15,6 @@ import (
"sync" "sync"
"time" "time"
"github.com/spf13/viper"
"google.golang.org/grpc" "google.golang.org/grpc"
"github.com/chrislusf/seaweedfs/weed/glog" "github.com/chrislusf/seaweedfs/weed/glog"
@ -109,7 +108,7 @@ var (
func runBenchmark(cmd *Command, args []string) bool { func runBenchmark(cmd *Command, args []string) bool {
util.LoadConfiguration("security", false) util.LoadConfiguration("security", false)
b.grpcDialOption = security.LoadClientTLS(viper.Sub("grpc"), "client") b.grpcDialOption = security.LoadClientTLS(util.GetViper(), "grpc.client")
fmt.Printf("This is SeaweedFS version %s %s %s\n", util.VERSION, runtime.GOOS, runtime.GOARCH) fmt.Printf("This is SeaweedFS version %s %s %s\n", util.VERSION, runtime.GOOS, runtime.GOARCH)
if *b.maxCpu < 1 { if *b.maxCpu < 1 {

View file

@ -6,14 +6,13 @@ import (
"strings" "strings"
"time" "time"
"github.com/chrislusf/seaweedfs/weed/security" "google.golang.org/grpc/reflection"
"github.com/spf13/viper"
"github.com/chrislusf/seaweedfs/weed/glog" "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/security"
"github.com/chrislusf/seaweedfs/weed/server" "github.com/chrislusf/seaweedfs/weed/server"
"github.com/chrislusf/seaweedfs/weed/util" "github.com/chrislusf/seaweedfs/weed/util"
"google.golang.org/grpc/reflection"
) )
var ( var (
@ -145,7 +144,7 @@ func (fo *FilerOptions) startFiler() {
if err != nil { if err != nil {
glog.Fatalf("failed to listen on grpc port %d: %v", grpcPort, err) glog.Fatalf("failed to listen on grpc port %d: %v", grpcPort, err)
} }
grpcS := util.NewGrpcServer(security.LoadServerTLS(viper.Sub("grpc"), "filer")) grpcS := util.NewGrpcServer(security.LoadServerTLS(util.GetViper(), "grpc.filer"))
filer_pb.RegisterSeaweedFilerServer(grpcS, fs) filer_pb.RegisterSeaweedFilerServer(grpcS, fs)
reflection.Register(grpcS) reflection.Register(grpcS)
go grpcS.Serve(grpcL) go grpcS.Serve(grpcL)

View file

@ -14,13 +14,13 @@ import (
"sync" "sync"
"time" "time"
"google.golang.org/grpc"
"github.com/chrislusf/seaweedfs/weed/operation" "github.com/chrislusf/seaweedfs/weed/operation"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb" "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
"github.com/chrislusf/seaweedfs/weed/security" "github.com/chrislusf/seaweedfs/weed/security"
"github.com/chrislusf/seaweedfs/weed/util" "github.com/chrislusf/seaweedfs/weed/util"
"github.com/chrislusf/seaweedfs/weed/wdclient" "github.com/chrislusf/seaweedfs/weed/wdclient"
"github.com/spf13/viper"
"google.golang.org/grpc"
) )
var ( var (
@ -105,7 +105,7 @@ func runCopy(cmd *Command, args []string) bool {
filerGrpcPort := filerPort + 10000 filerGrpcPort := filerPort + 10000
filerGrpcAddress := fmt.Sprintf("%s:%d", filerUrl.Hostname(), filerGrpcPort) filerGrpcAddress := fmt.Sprintf("%s:%d", filerUrl.Hostname(), filerGrpcPort)
copy.grpcDialOption = security.LoadClientTLS(viper.Sub("grpc"), "client") copy.grpcDialOption = security.LoadClientTLS(util.GetViper(), "grpc.client")
ctx := context.Background() ctx := context.Background()

View file

@ -39,7 +39,7 @@ func runFilerReplicate(cmd *Command, args []string) bool {
util.LoadConfiguration("security", false) util.LoadConfiguration("security", false)
util.LoadConfiguration("replication", true) util.LoadConfiguration("replication", true)
util.LoadConfiguration("notification", true) util.LoadConfiguration("notification", true)
config := viper.GetViper() config := util.GetViper()
var notificationInput sub.NotificationInput var notificationInput sub.NotificationInput
@ -47,8 +47,7 @@ func runFilerReplicate(cmd *Command, args []string) bool {
for _, input := range sub.NotificationInputs { for _, input := range sub.NotificationInputs {
if config.GetBool("notification." + input.GetName() + ".enabled") { if config.GetBool("notification." + input.GetName() + ".enabled") {
viperSub := config.Sub("notification." + input.GetName()) if err := input.Initialize(config, "notification."+input.GetName()+"."); err != nil {
if err := input.Initialize(viperSub); err != nil {
glog.Fatalf("Failed to initialize notification input for %s: %+v", glog.Fatalf("Failed to initialize notification input for %s: %+v",
input.GetName(), err) input.GetName(), err)
} }
@ -66,10 +65,9 @@ func runFilerReplicate(cmd *Command, args []string) bool {
// avoid recursive replication // avoid recursive replication
if config.GetBool("notification.source.filer.enabled") && config.GetBool("notification.sink.filer.enabled") { if config.GetBool("notification.source.filer.enabled") && config.GetBool("notification.sink.filer.enabled") {
sourceConfig, sinkConfig := config.Sub("source.filer"), config.Sub("sink.filer") if config.GetString("source.filer.grpcAddress") == config.GetString("sink.filer.grpcAddress") {
if sourceConfig.GetString("grpcAddress") == sinkConfig.GetString("grpcAddress") { fromDir := config.GetString("source.filer.directory")
fromDir := sourceConfig.GetString("directory") toDir := config.GetString("sink.filer.directory")
toDir := sinkConfig.GetString("directory")
if strings.HasPrefix(toDir, fromDir) { if strings.HasPrefix(toDir, fromDir) {
glog.Fatalf("recursive replication! source directory %s includes the sink directory %s", fromDir, toDir) glog.Fatalf("recursive replication! source directory %s includes the sink directory %s", fromDir, toDir)
} }
@ -79,8 +77,7 @@ func runFilerReplicate(cmd *Command, args []string) bool {
var dataSink sink.ReplicationSink var dataSink sink.ReplicationSink
for _, sk := range sink.Sinks { for _, sk := range sink.Sinks {
if config.GetBool("sink." + sk.GetName() + ".enabled") { if config.GetBool("sink." + sk.GetName() + ".enabled") {
viperSub := config.Sub("sink." + sk.GetName()) if err := sk.Initialize(config, "sink."+sk.GetName()+"."); err != nil {
if err := sk.Initialize(viperSub); err != nil {
glog.Fatalf("Failed to initialize sink for %s: %+v", glog.Fatalf("Failed to initialize sink for %s: %+v",
sk.GetName(), err) sk.GetName(), err)
} }
@ -98,7 +95,7 @@ func runFilerReplicate(cmd *Command, args []string) bool {
return true return true
} }
replicator := replication.NewReplicator(config.Sub("source.filer"), dataSink) replicator := replication.NewReplicator(config, "source.filer.", dataSink)
for { for {
key, m, err := notificationInput.ReceiveMessage() key, m, err := notificationInput.ReceiveMessage()

View file

@ -8,15 +8,15 @@ import (
"strings" "strings"
"github.com/chrislusf/raft/protobuf" "github.com/chrislusf/raft/protobuf"
"github.com/gorilla/mux"
"google.golang.org/grpc/reflection"
"github.com/chrislusf/seaweedfs/weed/glog" "github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/pb/master_pb" "github.com/chrislusf/seaweedfs/weed/pb/master_pb"
"github.com/chrislusf/seaweedfs/weed/security" "github.com/chrislusf/seaweedfs/weed/security"
"github.com/chrislusf/seaweedfs/weed/server" "github.com/chrislusf/seaweedfs/weed/server"
"github.com/chrislusf/seaweedfs/weed/storage/backend" "github.com/chrislusf/seaweedfs/weed/storage/backend"
"github.com/chrislusf/seaweedfs/weed/util" "github.com/chrislusf/seaweedfs/weed/util"
"github.com/gorilla/mux"
"github.com/spf13/viper"
"google.golang.org/grpc/reflection"
) )
var ( var (
@ -102,7 +102,7 @@ func runMaster(cmd *Command, args []string) bool {
func startMaster(masterOption MasterOptions, masterWhiteList []string) { func startMaster(masterOption MasterOptions, masterWhiteList []string) {
backend.LoadConfiguration(viper.GetViper()) backend.LoadConfiguration(util.GetViper())
myMasterAddress, peers := checkPeers(*masterOption.ip, *masterOption.port, *masterOption.peers) myMasterAddress, peers := checkPeers(*masterOption.ip, *masterOption.port, *masterOption.peers)
@ -115,7 +115,7 @@ func startMaster(masterOption MasterOptions, masterWhiteList []string) {
glog.Fatalf("Master startup error: %v", e) glog.Fatalf("Master startup error: %v", e)
} }
// start raftServer // start raftServer
raftServer := weed_server.NewRaftServer(security.LoadClientTLS(viper.Sub("grpc"), "master"), raftServer := weed_server.NewRaftServer(security.LoadClientTLS(util.GetViper(), "grpc.master"),
peers, myMasterAddress, *masterOption.metaFolder, ms.Topo, *masterOption.pulseSeconds) peers, myMasterAddress, *masterOption.metaFolder, ms.Topo, *masterOption.pulseSeconds)
if raftServer == nil { if raftServer == nil {
glog.Fatalf("please verify %s is writable, see https://github.com/chrislusf/seaweedfs/issues/717", *masterOption.metaFolder) glog.Fatalf("please verify %s is writable, see https://github.com/chrislusf/seaweedfs/issues/717", *masterOption.metaFolder)
@ -129,7 +129,7 @@ func startMaster(masterOption MasterOptions, masterWhiteList []string) {
glog.Fatalf("master failed to listen on grpc port %d: %v", grpcPort, err) glog.Fatalf("master failed to listen on grpc port %d: %v", grpcPort, err)
} }
// Create your protocol servers. // Create your protocol servers.
grpcS := util.NewGrpcServer(security.LoadServerTLS(viper.Sub("grpc"), "master")) grpcS := util.NewGrpcServer(security.LoadServerTLS(util.GetViper(), "grpc.master"))
master_pb.RegisterSeaweedServer(grpcS, ms) master_pb.RegisterSeaweedServer(grpcS, ms)
protobuf.RegisterRaftServer(grpcS, raftServer) protobuf.RegisterRaftServer(grpcS, raftServer)
reflection.Register(grpcS) reflection.Register(grpcS)

View file

@ -13,7 +13,6 @@ import (
"time" "time"
"github.com/jacobsa/daemonize" "github.com/jacobsa/daemonize"
"github.com/spf13/viper"
"github.com/chrislusf/seaweedfs/weed/filesys" "github.com/chrislusf/seaweedfs/weed/filesys"
"github.com/chrislusf/seaweedfs/weed/glog" "github.com/chrislusf/seaweedfs/weed/glog"
@ -148,7 +147,7 @@ func RunMount(filer, filerMountRootPath, dir, collection, replication, dataCente
err = fs.Serve(c, filesys.NewSeaweedFileSystem(&filesys.Option{ err = fs.Serve(c, filesys.NewSeaweedFileSystem(&filesys.Option{
FilerGrpcAddress: filerGrpcAddress, FilerGrpcAddress: filerGrpcAddress,
GrpcDialOption: security.LoadClientTLS(viper.Sub("grpc"), "client"), GrpcDialOption: security.LoadClientTLS(util.GetViper(), "grpc.client"),
FilerMountRootPath: mountRoot, FilerMountRootPath: mountRoot,
Collection: collection, Collection: collection,
Replication: replication, Replication: replication,

View file

@ -1,18 +1,17 @@
package command package command
import ( import (
"fmt"
"net/http" "net/http"
"time" "time"
"github.com/chrislusf/seaweedfs/weed/security" "github.com/chrislusf/seaweedfs/weed/security"
"github.com/spf13/viper"
"fmt" "github.com/gorilla/mux"
"github.com/chrislusf/seaweedfs/weed/glog" "github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/s3api" "github.com/chrislusf/seaweedfs/weed/s3api"
"github.com/chrislusf/seaweedfs/weed/util" "github.com/chrislusf/seaweedfs/weed/util"
"github.com/gorilla/mux"
) )
var ( var (
@ -69,7 +68,7 @@ func (s3opt *S3Options) startS3Server() bool {
FilerGrpcAddress: filerGrpcAddress, FilerGrpcAddress: filerGrpcAddress,
DomainName: *s3opt.domainName, DomainName: *s3opt.domainName,
BucketsPath: *s3opt.filerBucketsPath, BucketsPath: *s3opt.filerBucketsPath,
GrpcDialOption: security.LoadClientTLS(viper.Sub("grpc"), "client"), GrpcDialOption: security.LoadClientTLS(util.GetViper(), "grpc.client"),
}) })
if s3ApiServer_err != nil { if s3ApiServer_err != nil {
glog.Fatalf("S3 API Server startup error: %v", s3ApiServer_err) glog.Fatalf("S3 API Server startup error: %v", s3ApiServer_err)

View file

@ -14,6 +14,14 @@ var cmdScaffold = &Command{
Short: "generate basic configuration files", Short: "generate basic configuration files",
Long: `Generate filer.toml with all possible configurations for you to customize. Long: `Generate filer.toml with all possible configurations for you to customize.
The options can also be overwritten by environment variables.
For example, the filer.toml mysql password can be overwritten by environment variable
export weed.mysql.password=some_password
Environment variable rules:
* Prefix fix with "WEED_"
* Upppercase the reset of variable name.
* Replace '.' with '_'
`, `,
} }

View file

@ -6,7 +6,6 @@ import (
"github.com/chrislusf/seaweedfs/weed/security" "github.com/chrislusf/seaweedfs/weed/security"
"github.com/chrislusf/seaweedfs/weed/shell" "github.com/chrislusf/seaweedfs/weed/shell"
"github.com/chrislusf/seaweedfs/weed/util" "github.com/chrislusf/seaweedfs/weed/util"
"github.com/spf13/viper"
) )
var ( var (
@ -31,7 +30,7 @@ var cmdShell = &Command{
func runShell(command *Command, args []string) bool { func runShell(command *Command, args []string) bool {
util.LoadConfiguration("security", false) util.LoadConfiguration("security", false)
shellOptions.GrpcDialOption = security.LoadClientTLS(viper.Sub("grpc"), "client") shellOptions.GrpcDialOption = security.LoadClientTLS(util.GetViper(), "grpc.client")
var filerPwdErr error var filerPwdErr error
shellOptions.FilerHost, shellOptions.FilerPort, shellOptions.Directory, filerPwdErr = util.ParseFilerUrl(*shellInitialFilerUrl) shellOptions.FilerHost, shellOptions.FilerPort, shellOptions.Directory, filerPwdErr = util.ParseFilerUrl(*shellInitialFilerUrl)

View file

@ -6,11 +6,9 @@ import (
"os" "os"
"path/filepath" "path/filepath"
"github.com/chrislusf/seaweedfs/weed/operation"
"github.com/chrislusf/seaweedfs/weed/security" "github.com/chrislusf/seaweedfs/weed/security"
"github.com/chrislusf/seaweedfs/weed/util" "github.com/chrislusf/seaweedfs/weed/util"
"github.com/spf13/viper"
"github.com/chrislusf/seaweedfs/weed/operation"
) )
var ( var (
@ -63,7 +61,7 @@ var cmdUpload = &Command{
func runUpload(cmd *Command, args []string) bool { func runUpload(cmd *Command, args []string) bool {
util.LoadConfiguration("security", false) util.LoadConfiguration("security", false)
grpcDialOption := security.LoadClientTLS(viper.Sub("grpc"), "client") grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.client")
if len(args) == 0 { if len(args) == 0 {
if *upload.dir == "" { if *upload.dir == "" {

View file

@ -234,7 +234,7 @@ func (v VolumeServerOptions) startGrpcService(vs volume_server_pb.VolumeServerSe
if err != nil { if err != nil {
glog.Fatalf("failed to listen on grpc port %d: %v", grpcPort, err) glog.Fatalf("failed to listen on grpc port %d: %v", grpcPort, err)
} }
grpcS := util.NewGrpcServer(security.LoadServerTLS(viper.Sub("grpc"), "volume")) grpcS := util.NewGrpcServer(security.LoadServerTLS(util.GetViper(), "grpc.volume"))
volume_server_pb.RegisterVolumeServerServer(grpcS, vs) volume_server_pb.RegisterVolumeServerServer(grpcS, vs)
reflection.Register(grpcS) reflection.Register(grpcS)
go func() { go func() {

View file

@ -11,7 +11,6 @@ import (
"github.com/chrislusf/seaweedfs/weed/security" "github.com/chrislusf/seaweedfs/weed/security"
"github.com/chrislusf/seaweedfs/weed/server" "github.com/chrislusf/seaweedfs/weed/server"
"github.com/chrislusf/seaweedfs/weed/util" "github.com/chrislusf/seaweedfs/weed/util"
"github.com/spf13/viper"
) )
var ( var (
@ -75,7 +74,7 @@ func (wo *WebDavOption) startWebDav() bool {
ws, webdavServer_err := weed_server.NewWebDavServer(&weed_server.WebDavOption{ ws, webdavServer_err := weed_server.NewWebDavServer(&weed_server.WebDavOption{
Filer: *wo.filer, Filer: *wo.filer,
FilerGrpcAddress: filerGrpcAddress, FilerGrpcAddress: filerGrpcAddress,
GrpcDialOption: security.LoadClientTLS(viper.Sub("grpc"), "client"), GrpcDialOption: security.LoadClientTLS(util.GetViper(), "grpc.client"),
Collection: *wo.collection, Collection: *wo.collection,
Uid: uid, Uid: uid,
Gid: gid, Gid: gid,

View file

@ -22,10 +22,10 @@ func (store *CassandraStore) GetName() string {
return "cassandra" return "cassandra"
} }
func (store *CassandraStore) Initialize(configuration util.Configuration) (err error) { func (store *CassandraStore) Initialize(configuration util.Configuration, prefix string) (err error) {
return store.initialize( return store.initialize(
configuration.GetString("keyspace"), configuration.GetString(prefix+"keyspace"),
configuration.GetStringSlice("hosts"), configuration.GetStringSlice(prefix+"hosts"),
) )
} }

View file

@ -17,8 +17,7 @@ func (f *Filer) LoadConfiguration(config *viper.Viper) {
for _, store := range Stores { for _, store := range Stores {
if config.GetBool(store.GetName() + ".enabled") { if config.GetBool(store.GetName() + ".enabled") {
viperSub := config.Sub(store.GetName()) if err := store.Initialize(config, store.GetName()+"."); err != nil {
if err := store.Initialize(viperSub); err != nil {
glog.Fatalf("Failed to initialize store for %s: %+v", glog.Fatalf("Failed to initialize store for %s: %+v",
store.GetName(), err) store.GetName(), err)
} }

View file

@ -28,13 +28,13 @@ func (store *EtcdStore) GetName() string {
return "etcd" return "etcd"
} }
func (store *EtcdStore) Initialize(configuration weed_util.Configuration) (err error) { func (store *EtcdStore) Initialize(configuration weed_util.Configuration, prefix string) (err error) {
servers := configuration.GetString("servers") servers := configuration.GetString(prefix + "servers")
if servers == "" { if servers == "" {
servers = "localhost:2379" servers = "localhost:2379"
} }
timeout := configuration.GetString("timeout") timeout := configuration.GetString(prefix + "timeout")
if timeout == "" { if timeout == "" {
timeout = "3s" timeout = "3s"
} }

View file

@ -350,21 +350,21 @@ func TestChunksReading(t *testing.T) {
{ {
Chunks: []*filer_pb.FileChunk{ Chunks: []*filer_pb.FileChunk{
{Offset: 0, Size: 43175947, FileId: "2,111fc2cbfac1", Mtime: 1}, {Offset: 0, Size: 43175947, FileId: "2,111fc2cbfac1", Mtime: 1},
{Offset: 43175936, Size: 52981771-43175936, FileId: "2,112a36ea7f85", Mtime: 2}, {Offset: 43175936, Size: 52981771 - 43175936, FileId: "2,112a36ea7f85", Mtime: 2},
{Offset: 52981760, Size: 72564747-52981760, FileId: "4,112d5f31c5e7", Mtime: 3}, {Offset: 52981760, Size: 72564747 - 52981760, FileId: "4,112d5f31c5e7", Mtime: 3},
{Offset: 72564736, Size: 133255179-72564736, FileId: "1,113245f0cdb6", Mtime: 4}, {Offset: 72564736, Size: 133255179 - 72564736, FileId: "1,113245f0cdb6", Mtime: 4},
{Offset: 133255168, Size: 137269259-133255168, FileId: "3,1141a70733b5", Mtime: 5}, {Offset: 133255168, Size: 137269259 - 133255168, FileId: "3,1141a70733b5", Mtime: 5},
{Offset: 137269248, Size: 153578836-137269248, FileId: "1,114201d5bbdb", Mtime: 6}, {Offset: 137269248, Size: 153578836 - 137269248, FileId: "1,114201d5bbdb", Mtime: 6},
}, },
Offset: 0, Offset: 0,
Size: 153578836, Size: 153578836,
Expected: []*ChunkView{ Expected: []*ChunkView{
{Offset: 0, Size: 43175936, FileId: "2,111fc2cbfac1", LogicOffset: 0}, {Offset: 0, Size: 43175936, FileId: "2,111fc2cbfac1", LogicOffset: 0},
{Offset: 0, Size: 52981760-43175936, FileId: "2,112a36ea7f85", LogicOffset: 43175936}, {Offset: 0, Size: 52981760 - 43175936, FileId: "2,112a36ea7f85", LogicOffset: 43175936},
{Offset: 0, Size: 72564736-52981760, FileId: "4,112d5f31c5e7", LogicOffset: 52981760}, {Offset: 0, Size: 72564736 - 52981760, FileId: "4,112d5f31c5e7", LogicOffset: 52981760},
{Offset: 0, Size: 133255168-72564736, FileId: "1,113245f0cdb6", LogicOffset: 72564736}, {Offset: 0, Size: 133255168 - 72564736, FileId: "1,113245f0cdb6", LogicOffset: 72564736},
{Offset: 0, Size: 137269248-133255168, FileId: "3,1141a70733b5", LogicOffset: 133255168}, {Offset: 0, Size: 137269248 - 133255168, FileId: "3,1141a70733b5", LogicOffset: 133255168},
{Offset: 0, Size: 153578836-137269248, FileId: "1,114201d5bbdb", LogicOffset: 137269248}, {Offset: 0, Size: 153578836 - 137269248, FileId: "1,114201d5bbdb", LogicOffset: 137269248},
}, },
}, },
} }

View file

@ -14,7 +14,7 @@ type FilerStore interface {
// GetName gets the name to locate the configuration in filer.toml file // GetName gets the name to locate the configuration in filer.toml file
GetName() string GetName() string
// Initialize initializes the file store // Initialize initializes the file store
Initialize(configuration util.Configuration) error Initialize(configuration util.Configuration, prefix string) error
InsertEntry(context.Context, *Entry) error InsertEntry(context.Context, *Entry) error
UpdateEntry(context.Context, *Entry) (err error) UpdateEntry(context.Context, *Entry) (err error)
// err == filer2.ErrNotFound if not found // err == filer2.ErrNotFound if not found
@ -47,8 +47,8 @@ func (fsw *FilerStoreWrapper) GetName() string {
return fsw.actualStore.GetName() return fsw.actualStore.GetName()
} }
func (fsw *FilerStoreWrapper) Initialize(configuration util.Configuration) error { func (fsw *FilerStoreWrapper) Initialize(configuration util.Configuration, prefix string) error {
return fsw.actualStore.Initialize(configuration) return fsw.actualStore.Initialize(configuration, prefix)
} }
func (fsw *FilerStoreWrapper) InsertEntry(ctx context.Context, entry *Entry) error { func (fsw *FilerStoreWrapper) InsertEntry(ctx context.Context, entry *Entry) error {

View file

@ -30,8 +30,8 @@ func (store *LevelDBStore) GetName() string {
return "leveldb" return "leveldb"
} }
func (store *LevelDBStore) Initialize(configuration weed_util.Configuration) (err error) { func (store *LevelDBStore) Initialize(configuration weed_util.Configuration, prefix string) (err error) {
dir := configuration.GetString("dir") dir := configuration.GetString(prefix + "dir")
return store.initialize(dir) return store.initialize(dir)
} }

View file

@ -30,8 +30,8 @@ func (store *LevelDB2Store) GetName() string {
return "leveldb2" return "leveldb2"
} }
func (store *LevelDB2Store) Initialize(configuration weed_util.Configuration) (err error) { func (store *LevelDB2Store) Initialize(configuration weed_util.Configuration, prefix string) (err error) {
dir := configuration.GetString("dir") dir := configuration.GetString(prefix + "dir")
return store.initialize(dir, 8) return store.initialize(dir, 8)
} }

View file

@ -26,16 +26,16 @@ func (store *MysqlStore) GetName() string {
return "mysql" return "mysql"
} }
func (store *MysqlStore) Initialize(configuration util.Configuration) (err error) { func (store *MysqlStore) Initialize(configuration util.Configuration, prefix string) (err error) {
return store.initialize( return store.initialize(
configuration.GetString("username"), configuration.GetString(prefix+"username"),
configuration.GetString("password"), configuration.GetString(prefix+"password"),
configuration.GetString("hostname"), configuration.GetString(prefix+"hostname"),
configuration.GetInt("port"), configuration.GetInt(prefix+"port"),
configuration.GetString("database"), configuration.GetString(prefix+"database"),
configuration.GetInt("connection_max_idle"), configuration.GetInt(prefix+"connection_max_idle"),
configuration.GetInt("connection_max_open"), configuration.GetInt(prefix+"connection_max_open"),
configuration.GetBool("interpolateParams"), configuration.GetBool(prefix+"interpolateParams"),
) )
} }

View file

@ -26,16 +26,16 @@ func (store *PostgresStore) GetName() string {
return "postgres" return "postgres"
} }
func (store *PostgresStore) Initialize(configuration util.Configuration) (err error) { func (store *PostgresStore) Initialize(configuration util.Configuration, prefix string) (err error) {
return store.initialize( return store.initialize(
configuration.GetString("username"), configuration.GetString(prefix+"username"),
configuration.GetString("password"), configuration.GetString(prefix+"password"),
configuration.GetString("hostname"), configuration.GetString(prefix+"hostname"),
configuration.GetInt("port"), configuration.GetInt(prefix+"port"),
configuration.GetString("database"), configuration.GetString(prefix+"database"),
configuration.GetString("sslmode"), configuration.GetString(prefix+"sslmode"),
configuration.GetInt("connection_max_idle"), configuration.GetInt(prefix+"connection_max_idle"),
configuration.GetInt("connection_max_open"), configuration.GetInt(prefix+"connection_max_open"),
) )
} }

View file

@ -18,16 +18,16 @@ func (store *RedisClusterStore) GetName() string {
return "redis_cluster" return "redis_cluster"
} }
func (store *RedisClusterStore) Initialize(configuration util.Configuration) (err error) { func (store *RedisClusterStore) Initialize(configuration util.Configuration, prefix string) (err error) {
configuration.SetDefault("useReadOnly", true) configuration.SetDefault(prefix+"useReadOnly", true)
configuration.SetDefault("routeByLatency", true) configuration.SetDefault(prefix+"routeByLatency", true)
return store.initialize( return store.initialize(
configuration.GetStringSlice("addresses"), configuration.GetStringSlice(prefix+"addresses"),
configuration.GetString("password"), configuration.GetString(prefix+"password"),
configuration.GetBool("useReadOnly"), configuration.GetBool(prefix+"useReadOnly"),
configuration.GetBool("routeByLatency"), configuration.GetBool(prefix+"routeByLatency"),
) )
} }

View file

@ -18,11 +18,11 @@ func (store *RedisStore) GetName() string {
return "redis" return "redis"
} }
func (store *RedisStore) Initialize(configuration util.Configuration) (err error) { func (store *RedisStore) Initialize(configuration util.Configuration, prefix string) (err error) {
return store.initialize( return store.initialize(
configuration.GetString("address"), configuration.GetString(prefix+"address"),
configuration.GetString("password"), configuration.GetString(prefix+"password"),
configuration.GetInt("database"), configuration.GetInt(prefix+"database"),
) )
} }

View file

@ -30,8 +30,8 @@ func (store *TikvStore) GetName() string {
return "tikv" return "tikv"
} }
func (store *TikvStore) Initialize(configuration weed_util.Configuration) (err error) { func (store *TikvStore) Initialize(configuration weed_util.Configuration, prefix string) (err error) {
pdAddr := configuration.GetString("pdAddress") pdAddr := configuration.GetString(prefix + "pdAddress")
return store.initialize(pdAddr) return store.initialize(pdAddr)
} }

View file

@ -74,7 +74,7 @@ func subList(list *IntervalLinkedList, start, stop int64) *IntervalLinkedList {
nodes = append(nodes, &IntervalNode{ nodes = append(nodes, &IntervalNode{
Data: t.Data[nodeStart-t.Offset : nodeStop-t.Offset], Data: t.Data[nodeStart-t.Offset : nodeStop-t.Offset],
Offset: nodeStart, Offset: nodeStart,
Size: nodeStop-nodeStart, Size: nodeStop - nodeStart,
Next: nil, Next: nil,
}) })
} }

View file

@ -27,14 +27,14 @@ func (k *AwsSqsPub) GetName() string {
return "aws_sqs" return "aws_sqs"
} }
func (k *AwsSqsPub) Initialize(configuration util.Configuration) (err error) { func (k *AwsSqsPub) Initialize(configuration util.Configuration, prefix string) (err error) {
glog.V(0).Infof("filer.notification.aws_sqs.region: %v", configuration.GetString("region")) glog.V(0).Infof("filer.notification.aws_sqs.region: %v", configuration.GetString(prefix+"region"))
glog.V(0).Infof("filer.notification.aws_sqs.sqs_queue_name: %v", configuration.GetString("sqs_queue_name")) glog.V(0).Infof("filer.notification.aws_sqs.sqs_queue_name: %v", configuration.GetString(prefix+"sqs_queue_name"))
return k.initialize( return k.initialize(
configuration.GetString("aws_access_key_id"), configuration.GetString(prefix+"aws_access_key_id"),
configuration.GetString("aws_secret_access_key"), configuration.GetString(prefix+"aws_secret_access_key"),
configuration.GetString("region"), configuration.GetString(prefix+"region"),
configuration.GetString("sqs_queue_name"), configuration.GetString(prefix+"sqs_queue_name"),
) )
} }

View file

@ -11,7 +11,7 @@ type MessageQueue interface {
// GetName gets the name to locate the configuration in filer.toml file // GetName gets the name to locate the configuration in filer.toml file
GetName() string GetName() string
// Initialize initializes the file store // Initialize initializes the file store
Initialize(configuration util.Configuration) error Initialize(configuration util.Configuration, prefix string) error
SendMessage(key string, message proto.Message) error SendMessage(key string, message proto.Message) error
} }
@ -21,7 +21,7 @@ var (
Queue MessageQueue Queue MessageQueue
) )
func LoadConfiguration(config *viper.Viper) { func LoadConfiguration(config *viper.Viper, prefix string) {
if config == nil { if config == nil {
return return
@ -30,9 +30,8 @@ func LoadConfiguration(config *viper.Viper) {
validateOneEnabledQueue(config) validateOneEnabledQueue(config)
for _, queue := range MessageQueues { for _, queue := range MessageQueues {
if config.GetBool(queue.GetName() + ".enabled") { if config.GetBool(prefix + queue.GetName() + ".enabled") {
viperSub := config.Sub(queue.GetName()) if err := queue.Initialize(config, prefix+queue.GetName()+"."); err != nil {
if err := queue.Initialize(viperSub); err != nil {
glog.Fatalf("Failed to initialize notification for %s: %+v", glog.Fatalf("Failed to initialize notification for %s: %+v",
queue.GetName(), err) queue.GetName(), err)
} }

View file

@ -18,12 +18,13 @@ import (
"context" "context"
"fmt" "fmt"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/notification"
"github.com/chrislusf/seaweedfs/weed/util"
"github.com/golang/protobuf/proto" "github.com/golang/protobuf/proto"
"gocloud.dev/pubsub" "gocloud.dev/pubsub"
_ "gocloud.dev/pubsub/awssnssqs" _ "gocloud.dev/pubsub/awssnssqs"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/notification"
"github.com/chrislusf/seaweedfs/weed/util"
// _ "gocloud.dev/pubsub/azuresb" // _ "gocloud.dev/pubsub/azuresb"
_ "gocloud.dev/pubsub/gcppubsub" _ "gocloud.dev/pubsub/gcppubsub"
_ "gocloud.dev/pubsub/natspubsub" _ "gocloud.dev/pubsub/natspubsub"
@ -43,8 +44,8 @@ func (k *GoCDKPubSub) GetName() string {
return "gocdk_pub_sub" return "gocdk_pub_sub"
} }
func (k *GoCDKPubSub) Initialize(config util.Configuration) error { func (k *GoCDKPubSub) Initialize(configuration util.Configuration, prefix string) error {
k.topicURL = config.GetString("topic_url") k.topicURL = configuration.GetString(prefix + "topic_url")
glog.V(0).Infof("notification.gocdk_pub_sub.topic_url: %v", k.topicURL) glog.V(0).Infof("notification.gocdk_pub_sub.topic_url: %v", k.topicURL)
topic, err := pubsub.OpenTopic(context.Background(), k.topicURL) topic, err := pubsub.OpenTopic(context.Background(), k.topicURL)
if err != nil { if err != nil {

View file

@ -25,13 +25,13 @@ func (k *GooglePubSub) GetName() string {
return "google_pub_sub" return "google_pub_sub"
} }
func (k *GooglePubSub) Initialize(configuration util.Configuration) (err error) { func (k *GooglePubSub) Initialize(configuration util.Configuration, prefix string) (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.project_id: %v", configuration.GetString(prefix+"project_id"))
glog.V(0).Infof("notification.google_pub_sub.topic: %v", configuration.GetString("topic")) glog.V(0).Infof("notification.google_pub_sub.topic: %v", configuration.GetString(prefix+"topic"))
return k.initialize( return k.initialize(
configuration.GetString("google_application_credentials"), configuration.GetString(prefix+"google_application_credentials"),
configuration.GetString("project_id"), configuration.GetString(prefix+"project_id"),
configuration.GetString("topic"), configuration.GetString(prefix+"topic"),
) )
} }

View file

@ -21,12 +21,12 @@ func (k *KafkaQueue) GetName() string {
return "kafka" return "kafka"
} }
func (k *KafkaQueue) Initialize(configuration util.Configuration) (err error) { func (k *KafkaQueue) Initialize(configuration util.Configuration, prefix string) (err error) {
glog.V(0).Infof("filer.notification.kafka.hosts: %v\n", configuration.GetStringSlice("hosts")) glog.V(0).Infof("filer.notification.kafka.hosts: %v\n", configuration.GetStringSlice(prefix+"hosts"))
glog.V(0).Infof("filer.notification.kafka.topic: %v\n", configuration.GetString("topic")) glog.V(0).Infof("filer.notification.kafka.topic: %v\n", configuration.GetString(prefix+"topic"))
return k.initialize( return k.initialize(
configuration.GetStringSlice("hosts"), configuration.GetStringSlice(prefix+"hosts"),
configuration.GetString("topic"), configuration.GetString(prefix+"topic"),
) )
} }

View file

@ -18,7 +18,7 @@ func (k *LogQueue) GetName() string {
return "log" return "log"
} }
func (k *LogQueue) Initialize(configuration util.Configuration) (err error) { func (k *LogQueue) Initialize(configuration util.Configuration, prefix string) (err error) {
return nil return nil
} }

View file

@ -18,10 +18,10 @@ type Replicator struct {
source *source.FilerSource source *source.FilerSource
} }
func NewReplicator(sourceConfig util.Configuration, dataSink sink.ReplicationSink) *Replicator { func NewReplicator(sourceConfig util.Configuration, configPrefix string, dataSink sink.ReplicationSink) *Replicator {
source := &source.FilerSource{} source := &source.FilerSource{}
source.Initialize(sourceConfig) source.Initialize(sourceConfig, configPrefix)
dataSink.SetSourceFiler(source) dataSink.SetSourceFiler(source)

View file

@ -35,12 +35,12 @@ func (g *AzureSink) GetSinkToDirectory() string {
return g.dir return g.dir
} }
func (g *AzureSink) Initialize(configuration util.Configuration) error { func (g *AzureSink) Initialize(configuration util.Configuration, prefix string) error {
return g.initialize( return g.initialize(
configuration.GetString("account_name"), configuration.GetString(prefix+"account_name"),
configuration.GetString("account_key"), configuration.GetString(prefix+"account_key"),
configuration.GetString("container"), configuration.GetString(prefix+"container"),
configuration.GetString("directory"), configuration.GetString(prefix+"directory"),
) )
} }

View file

@ -31,12 +31,12 @@ func (g *B2Sink) GetSinkToDirectory() string {
return g.dir return g.dir
} }
func (g *B2Sink) Initialize(configuration util.Configuration) error { func (g *B2Sink) Initialize(configuration util.Configuration, prefix string) error {
return g.initialize( return g.initialize(
configuration.GetString("b2_account_id"), configuration.GetString(prefix+"b2_account_id"),
configuration.GetString("b2_master_application_key"), configuration.GetString(prefix+"b2_master_application_key"),
configuration.GetString("bucket"), configuration.GetString(prefix+"bucket"),
configuration.GetString("directory"), configuration.GetString(prefix+"directory"),
) )
} }

View file

@ -3,10 +3,11 @@ package filersink
import ( import (
"context" "context"
"fmt" "fmt"
"github.com/chrislusf/seaweedfs/weed/security"
"github.com/spf13/viper"
"google.golang.org/grpc" "google.golang.org/grpc"
"github.com/chrislusf/seaweedfs/weed/security"
"github.com/chrislusf/seaweedfs/weed/filer2" "github.com/chrislusf/seaweedfs/weed/filer2"
"github.com/chrislusf/seaweedfs/weed/glog" "github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb" "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
@ -38,13 +39,13 @@ func (fs *FilerSink) GetSinkToDirectory() string {
return fs.dir return fs.dir
} }
func (fs *FilerSink) Initialize(configuration util.Configuration) error { func (fs *FilerSink) Initialize(configuration util.Configuration, prefix string) error {
return fs.initialize( return fs.initialize(
configuration.GetString("grpcAddress"), configuration.GetString(prefix+"grpcAddress"),
configuration.GetString("directory"), configuration.GetString(prefix+"directory"),
configuration.GetString("replication"), configuration.GetString(prefix+"replication"),
configuration.GetString("collection"), configuration.GetString(prefix+"collection"),
configuration.GetInt("ttlSec"), configuration.GetInt(prefix+"ttlSec"),
) )
} }
@ -59,7 +60,7 @@ func (fs *FilerSink) initialize(grpcAddress string, dir string,
fs.replication = replication fs.replication = replication
fs.collection = collection fs.collection = collection
fs.ttlSec = int32(ttlSec) fs.ttlSec = int32(ttlSec)
fs.grpcDialOption = security.LoadClientTLS(viper.Sub("grpc"), "client") fs.grpcDialOption = security.LoadClientTLS(util.GetViper(), "grpc.client")
return nil return nil
} }

View file

@ -34,11 +34,11 @@ func (g *GcsSink) GetSinkToDirectory() string {
return g.dir return g.dir
} }
func (g *GcsSink) Initialize(configuration util.Configuration) error { func (g *GcsSink) Initialize(configuration util.Configuration, prefix string) error {
return g.initialize( return g.initialize(
configuration.GetString("google_application_credentials"), configuration.GetString(prefix+"google_application_credentials"),
configuration.GetString("bucket"), configuration.GetString(prefix+"bucket"),
configuration.GetString("directory"), configuration.GetString(prefix+"directory"),
) )
} }

View file

@ -9,7 +9,7 @@ import (
type ReplicationSink interface { type ReplicationSink interface {
GetName() string GetName() string
Initialize(configuration util.Configuration) error Initialize(configuration util.Configuration, prefix string) error
DeleteEntry(ctx context.Context, key string, isDirectory, deleteIncludeChunks bool) error DeleteEntry(ctx context.Context, key string, isDirectory, deleteIncludeChunks bool) error
CreateEntry(ctx context.Context, key string, entry *filer_pb.Entry) error CreateEntry(ctx context.Context, key string, entry *filer_pb.Entry) error
UpdateEntry(ctx context.Context, key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool) (foundExistingEntry bool, err error) UpdateEntry(ctx context.Context, key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool) (foundExistingEntry bool, err error)

View file

@ -39,16 +39,16 @@ func (s3sink *S3Sink) GetSinkToDirectory() string {
return s3sink.dir return s3sink.dir
} }
func (s3sink *S3Sink) Initialize(configuration util.Configuration) error { func (s3sink *S3Sink) Initialize(configuration util.Configuration, prefix string) error {
glog.V(0).Infof("sink.s3.region: %v", configuration.GetString("region")) glog.V(0).Infof("sink.s3.region: %v", configuration.GetString(prefix+"region"))
glog.V(0).Infof("sink.s3.bucket: %v", configuration.GetString("bucket")) glog.V(0).Infof("sink.s3.bucket: %v", configuration.GetString(prefix+"bucket"))
glog.V(0).Infof("sink.s3.directory: %v", configuration.GetString("directory")) glog.V(0).Infof("sink.s3.directory: %v", configuration.GetString(prefix+"directory"))
return s3sink.initialize( return s3sink.initialize(
configuration.GetString("aws_access_key_id"), configuration.GetString(prefix+"aws_access_key_id"),
configuration.GetString("aws_secret_access_key"), configuration.GetString(prefix+"aws_secret_access_key"),
configuration.GetString("region"), configuration.GetString(prefix+"region"),
configuration.GetString("bucket"), configuration.GetString(prefix+"bucket"),
configuration.GetString("directory"), configuration.GetString(prefix+"directory"),
) )
} }

View file

@ -3,13 +3,14 @@ package source
import ( import (
"context" "context"
"fmt" "fmt"
"github.com/chrislusf/seaweedfs/weed/security"
"github.com/spf13/viper"
"google.golang.org/grpc"
"io" "io"
"net/http" "net/http"
"strings" "strings"
"google.golang.org/grpc"
"github.com/chrislusf/seaweedfs/weed/security"
"github.com/chrislusf/seaweedfs/weed/glog" "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/util" "github.com/chrislusf/seaweedfs/weed/util"
@ -25,17 +26,17 @@ type FilerSource struct {
Dir string Dir string
} }
func (fs *FilerSource) Initialize(configuration util.Configuration) error { func (fs *FilerSource) Initialize(configuration util.Configuration, prefix string) error {
return fs.initialize( return fs.initialize(
configuration.GetString("grpcAddress"), configuration.GetString(prefix+"grpcAddress"),
configuration.GetString("directory"), configuration.GetString(prefix+"directory"),
) )
} }
func (fs *FilerSource) initialize(grpcAddress string, dir string) (err error) { func (fs *FilerSource) initialize(grpcAddress string, dir string) (err error) {
fs.grpcAddress = grpcAddress fs.grpcAddress = grpcAddress
fs.Dir = dir fs.Dir = dir
fs.grpcDialOption = security.LoadClientTLS(viper.Sub("grpc"), "client") fs.grpcDialOption = security.LoadClientTLS(util.GetViper(), "grpc.client")
return nil return nil
} }

View file

@ -27,14 +27,14 @@ func (k *AwsSqsInput) GetName() string {
return "aws_sqs" return "aws_sqs"
} }
func (k *AwsSqsInput) Initialize(configuration util.Configuration) error { func (k *AwsSqsInput) Initialize(configuration util.Configuration, prefix string) error {
glog.V(0).Infof("replication.notification.aws_sqs.region: %v", configuration.GetString("region")) glog.V(0).Infof("replication.notification.aws_sqs.region: %v", configuration.GetString(prefix+"region"))
glog.V(0).Infof("replication.notification.aws_sqs.sqs_queue_name: %v", configuration.GetString("sqs_queue_name")) glog.V(0).Infof("replication.notification.aws_sqs.sqs_queue_name: %v", configuration.GetString(prefix+"sqs_queue_name"))
return k.initialize( return k.initialize(
configuration.GetString("aws_access_key_id"), configuration.GetString(prefix+"aws_access_key_id"),
configuration.GetString("aws_secret_access_key"), configuration.GetString(prefix+"aws_secret_access_key"),
configuration.GetString("region"), configuration.GetString(prefix+"region"),
configuration.GetString("sqs_queue_name"), configuration.GetString(prefix+"sqs_queue_name"),
) )
} }

View file

@ -27,8 +27,8 @@ func (k *GoCDKPubSubInput) GetName() string {
return "gocdk_pub_sub" return "gocdk_pub_sub"
} }
func (k *GoCDKPubSubInput) Initialize(config util.Configuration) error { func (k *GoCDKPubSubInput) Initialize(configuration util.Configuration, prefix string) error {
subURL := config.GetString("sub_url") subURL := configuration.GetString(prefix + "sub_url")
glog.V(0).Infof("notification.gocdk_pub_sub.sub_url: %v", subURL) glog.V(0).Infof("notification.gocdk_pub_sub.sub_url: %v", subURL)
sub, err := pubsub.OpenSubscription(context.Background(), subURL) sub, err := pubsub.OpenSubscription(context.Background(), subURL)
if err != nil { if err != nil {

View file

@ -27,13 +27,13 @@ func (k *GooglePubSubInput) GetName() string {
return "google_pub_sub" return "google_pub_sub"
} }
func (k *GooglePubSubInput) Initialize(configuration util.Configuration) error { func (k *GooglePubSubInput) Initialize(configuration util.Configuration, prefix string) error {
glog.V(0).Infof("notification.google_pub_sub.project_id: %v", configuration.GetString("project_id")) glog.V(0).Infof("notification.google_pub_sub.project_id: %v", configuration.GetString(prefix+"project_id"))
glog.V(0).Infof("notification.google_pub_sub.topic: %v", configuration.GetString("topic")) glog.V(0).Infof("notification.google_pub_sub.topic: %v", configuration.GetString(prefix+"topic"))
return k.initialize( return k.initialize(
configuration.GetString("google_application_credentials"), configuration.GetString(prefix+"google_application_credentials"),
configuration.GetString("project_id"), configuration.GetString(prefix+"project_id"),
configuration.GetString("topic"), configuration.GetString(prefix+"topic"),
) )
} }

View file

@ -28,14 +28,14 @@ func (k *KafkaInput) GetName() string {
return "kafka" return "kafka"
} }
func (k *KafkaInput) Initialize(configuration util.Configuration) error { func (k *KafkaInput) Initialize(configuration util.Configuration, prefix string) error {
glog.V(0).Infof("replication.notification.kafka.hosts: %v\n", configuration.GetStringSlice("hosts")) glog.V(0).Infof("replication.notification.kafka.hosts: %v\n", configuration.GetStringSlice(prefix+"hosts"))
glog.V(0).Infof("replication.notification.kafka.topic: %v\n", configuration.GetString("topic")) glog.V(0).Infof("replication.notification.kafka.topic: %v\n", configuration.GetString(prefix+"topic"))
return k.initialize( return k.initialize(
configuration.GetStringSlice("hosts"), configuration.GetStringSlice(prefix+"hosts"),
configuration.GetString("topic"), configuration.GetString(prefix+"topic"),
configuration.GetString("offsetFile"), configuration.GetString(prefix+"offsetFile"),
configuration.GetInt("offsetSaveIntervalSeconds"), configuration.GetInt(prefix+"offsetSaveIntervalSeconds"),
) )
} }

View file

@ -9,7 +9,7 @@ type NotificationInput interface {
// GetName gets the name to locate the configuration in sync.toml file // GetName gets the name to locate the configuration in sync.toml file
GetName() string GetName() string
// Initialize initializes the file store // Initialize initializes the file store
Initialize(configuration util.Configuration) error Initialize(configuration util.Configuration, prefix string) error
ReceiveMessage() (key string, message *filer_pb.EventNotification, err error) ReceiveMessage() (key string, message *filer_pb.EventNotification, err error)
} }

View file

@ -22,7 +22,7 @@ func LoadServerTLS(config *viper.Viper, component string) grpc.ServerOption {
glog.Errorf("load cert/key error: %v", err) glog.Errorf("load cert/key error: %v", err)
return nil return nil
} }
caCert, err := ioutil.ReadFile(config.GetString("ca")) caCert, err := ioutil.ReadFile(config.GetString(component + ".ca"))
if err != nil { if err != nil {
glog.Errorf("read ca cert file error: %v", err) glog.Errorf("read ca cert file error: %v", err)
return nil return nil
@ -49,7 +49,7 @@ func LoadClientTLS(config *viper.Viper, component string) grpc.DialOption {
glog.Errorf("load cert/key error: %v", err) glog.Errorf("load cert/key error: %v", err)
return grpc.WithInsecure() return grpc.WithInsecure()
} }
caCert, err := ioutil.ReadFile(config.GetString("ca")) caCert, err := ioutil.ReadFile(config.GetString(component + ".ca"))
if err != nil { if err != nil {
glog.Errorf("read ca cert file error: %v", err) glog.Errorf("read ca cert file error: %v", err)
return grpc.WithInsecure() return grpc.WithInsecure()

View file

@ -14,8 +14,6 @@ import (
"github.com/chrislusf/seaweedfs/weed/stats" "github.com/chrislusf/seaweedfs/weed/stats"
"github.com/chrislusf/seaweedfs/weed/util" "github.com/chrislusf/seaweedfs/weed/util"
"github.com/spf13/viper"
"github.com/chrislusf/seaweedfs/weed/filer2" "github.com/chrislusf/seaweedfs/weed/filer2"
_ "github.com/chrislusf/seaweedfs/weed/filer2/cassandra" _ "github.com/chrislusf/seaweedfs/weed/filer2/cassandra"
_ "github.com/chrislusf/seaweedfs/weed/filer2/etcd" _ "github.com/chrislusf/seaweedfs/weed/filer2/etcd"
@ -61,7 +59,7 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
fs = &FilerServer{ fs = &FilerServer{
option: option, option: option,
grpcDialOption: security.LoadClientTLS(viper.Sub("grpc"), "filer"), grpcDialOption: security.LoadClientTLS(util.GetViper(), "grpc.filer"),
} }
if len(option.Masters) == 0 { if len(option.Masters) == 0 {
@ -72,7 +70,7 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
go fs.filer.KeepConnectedToMaster() go fs.filer.KeepConnectedToMaster()
v := viper.GetViper() v := util.GetViper()
if !util.LoadConfiguration("filer", false) { if !util.LoadConfiguration("filer", false) {
v.Set("leveldb2.enabled", true) v.Set("leveldb2.enabled", true)
v.Set("leveldb2.dir", option.DefaultLevelDbDir) v.Set("leveldb2.dir", option.DefaultLevelDbDir)
@ -86,7 +84,7 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
fs.option.recursiveDelete = v.GetBool("filer.options.recursive_delete") fs.option.recursiveDelete = v.GetBool("filer.options.recursive_delete")
fs.filer.LoadConfiguration(v) fs.filer.LoadConfiguration(v)
notification.LoadConfiguration(v.Sub("notification")) notification.LoadConfiguration(v, "notification.")
handleStaticResources(defaultMux) handleStaticResources(defaultMux)
if !option.DisableHttp { if !option.DisableHttp {

View file

@ -14,6 +14,9 @@ import (
"time" "time"
"github.com/chrislusf/raft" "github.com/chrislusf/raft"
"github.com/gorilla/mux"
"google.golang.org/grpc"
"github.com/chrislusf/seaweedfs/weed/glog" "github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/pb/master_pb" "github.com/chrislusf/seaweedfs/weed/pb/master_pb"
"github.com/chrislusf/seaweedfs/weed/security" "github.com/chrislusf/seaweedfs/weed/security"
@ -22,9 +25,6 @@ import (
"github.com/chrislusf/seaweedfs/weed/topology" "github.com/chrislusf/seaweedfs/weed/topology"
"github.com/chrislusf/seaweedfs/weed/util" "github.com/chrislusf/seaweedfs/weed/util"
"github.com/chrislusf/seaweedfs/weed/wdclient" "github.com/chrislusf/seaweedfs/weed/wdclient"
"github.com/gorilla/mux"
"github.com/spf13/viper"
"google.golang.org/grpc"
) )
const ( const (
@ -69,7 +69,7 @@ type MasterServer struct {
func NewMasterServer(r *mux.Router, option *MasterOption, peers []string) *MasterServer { func NewMasterServer(r *mux.Router, option *MasterOption, peers []string) *MasterServer {
v := viper.GetViper() v := util.GetViper()
signingKey := v.GetString("jwt.signing.key") signingKey := v.GetString("jwt.signing.key")
v.SetDefault("jwt.signing.expires_after_seconds", 10) v.SetDefault("jwt.signing.expires_after_seconds", 10)
expiresAfterSec := v.GetInt("jwt.signing.expires_after_seconds") expiresAfterSec := v.GetInt("jwt.signing.expires_after_seconds")
@ -83,7 +83,7 @@ func NewMasterServer(r *mux.Router, option *MasterOption, peers []string) *Maste
preallocateSize = int64(option.VolumeSizeLimitMB) * (1 << 20) preallocateSize = int64(option.VolumeSizeLimitMB) * (1 << 20)
} }
grpcDialOption := security.LoadClientTLS(v.Sub("grpc"), "master") grpcDialOption := security.LoadClientTLS(v, "grpc.master")
ms := &MasterServer{ ms := &MasterServer{
option: option, option: option,
preallocateSize: preallocateSize, preallocateSize: preallocateSize,
@ -183,7 +183,7 @@ func (ms *MasterServer) proxyToLeader(f func(w http.ResponseWriter, r *http.Requ
func (ms *MasterServer) startAdminScripts() { func (ms *MasterServer) startAdminScripts() {
var err error var err error
v := viper.GetViper() v := util.GetViper()
adminScripts := v.GetString("master.maintenance.scripts") adminScripts := v.GetString("master.maintenance.scripts")
glog.V(0).Infof("adminScripts:\n%v", adminScripts) glog.V(0).Infof("adminScripts:\n%v", adminScripts)
if adminScripts == "" { if adminScripts == "" {
@ -201,7 +201,7 @@ func (ms *MasterServer) startAdminScripts() {
masterAddress := "localhost:" + strconv.Itoa(ms.option.Port) masterAddress := "localhost:" + strconv.Itoa(ms.option.Port)
var shellOptions shell.ShellOptions var shellOptions shell.ShellOptions
shellOptions.GrpcDialOption = security.LoadClientTLS(viper.Sub("grpc"), "master") shellOptions.GrpcDialOption = security.LoadClientTLS(v, "grpc.master")
shellOptions.Masters = &masterAddress shellOptions.Masters = &masterAddress
shellOptions.FilerHost, shellOptions.FilerPort, shellOptions.Directory, err = util.ParseFilerUrl(filerURL) shellOptions.FilerHost, shellOptions.FilerPort, shellOptions.Directory, err = util.ParseFilerUrl(filerURL)
@ -250,7 +250,7 @@ func (ms *MasterServer) startAdminScripts() {
func (ms *MasterServer) createSequencer(option *MasterOption) sequence.Sequencer { func (ms *MasterServer) createSequencer(option *MasterOption) sequence.Sequencer {
var seq sequence.Sequencer var seq sequence.Sequencer
v := viper.GetViper() v := util.GetViper()
seqType := strings.ToLower(v.GetString(SequencerType)) seqType := strings.ToLower(v.GetString(SequencerType))
glog.V(1).Infof("[%s] : [%s]", SequencerType, seqType) glog.V(1).Infof("[%s] : [%s]", SequencerType, seqType)
switch strings.ToLower(seqType) { switch strings.ToLower(seqType) {

View file

@ -5,16 +5,17 @@ import (
"net" "net"
"time" "time"
"google.golang.org/grpc"
"github.com/chrislusf/seaweedfs/weed/security" "github.com/chrislusf/seaweedfs/weed/security"
"github.com/chrislusf/seaweedfs/weed/storage/backend" "github.com/chrislusf/seaweedfs/weed/storage/backend"
"github.com/chrislusf/seaweedfs/weed/storage/erasure_coding" "github.com/chrislusf/seaweedfs/weed/storage/erasure_coding"
"github.com/spf13/viper"
"google.golang.org/grpc" "golang.org/x/net/context"
"github.com/chrislusf/seaweedfs/weed/glog" "github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/pb/master_pb" "github.com/chrislusf/seaweedfs/weed/pb/master_pb"
"github.com/chrislusf/seaweedfs/weed/util" "github.com/chrislusf/seaweedfs/weed/util"
"golang.org/x/net/context"
) )
func (vs *VolumeServer) GetMaster() string { func (vs *VolumeServer) GetMaster() string {
@ -26,7 +27,7 @@ func (vs *VolumeServer) heartbeat() {
vs.store.SetDataCenter(vs.dataCenter) vs.store.SetDataCenter(vs.dataCenter)
vs.store.SetRack(vs.rack) vs.store.SetRack(vs.rack)
grpcDialOption := security.LoadClientTLS(viper.Sub("grpc"), "volume") grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.volume")
var err error var err error
var newLeader string var newLeader string

View file

@ -7,8 +7,7 @@ import (
"google.golang.org/grpc" "google.golang.org/grpc"
"github.com/chrislusf/seaweedfs/weed/stats" "github.com/chrislusf/seaweedfs/weed/stats"
"github.com/chrislusf/seaweedfs/weed/util"
"github.com/spf13/viper"
"github.com/chrislusf/seaweedfs/weed/glog" "github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/security" "github.com/chrislusf/seaweedfs/weed/security"
@ -47,7 +46,7 @@ func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string,
fileSizeLimitMB int, fileSizeLimitMB int,
) *VolumeServer { ) *VolumeServer {
v := viper.GetViper() v := util.GetViper()
signingKey := v.GetString("jwt.signing.key") signingKey := v.GetString("jwt.signing.key")
v.SetDefault("jwt.signing.expires_after_seconds", 10) v.SetDefault("jwt.signing.expires_after_seconds", 10)
expiresAfterSec := v.GetInt("jwt.signing.expires_after_seconds") expiresAfterSec := v.GetInt("jwt.signing.expires_after_seconds")
@ -64,7 +63,7 @@ func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string,
needleMapKind: needleMapKind, needleMapKind: needleMapKind,
FixJpgOrientation: fixJpgOrientation, FixJpgOrientation: fixJpgOrientation,
ReadRedirect: readRedirect, ReadRedirect: readRedirect,
grpcDialOption: security.LoadClientTLS(viper.Sub("grpc"), "volume"), grpcDialOption: security.LoadClientTLS(util.GetViper(), "grpc.volume"),
compactionBytePerSecond: int64(compactionMBPerSecond) * 1024 * 1024, compactionBytePerSecond: int64(compactionMBPerSecond) * 1024 * 1024,
fileSizeLimitBytes: int64(fileSizeLimitMB) * 1024 * 1024, fileSizeLimitBytes: int64(fileSizeLimitMB) * 1024 * 1024,
} }

View file

@ -17,8 +17,6 @@ import (
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb" "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
"github.com/chrislusf/seaweedfs/weed/util" "github.com/chrislusf/seaweedfs/weed/util"
"github.com/spf13/viper"
"github.com/chrislusf/seaweedfs/weed/filer2" "github.com/chrislusf/seaweedfs/weed/filer2"
"github.com/chrislusf/seaweedfs/weed/glog" "github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/security" "github.com/chrislusf/seaweedfs/weed/security"
@ -49,7 +47,7 @@ func NewWebDavServer(option *WebDavOption) (ws *WebDavServer, err error) {
ws = &WebDavServer{ ws = &WebDavServer{
option: option, option: option,
grpcDialOption: security.LoadClientTLS(viper.Sub("grpc"), "filer"), grpcDialOption: security.LoadClientTLS(util.GetViper(), "grpc.filer"),
Handler: &webdav.Handler{ Handler: &webdav.Handler{
FileSystem: fs, FileSystem: fs,
LockSystem: webdav.NewMemLS(), LockSystem: webdav.NewMemLS(),

View file

@ -5,8 +5,6 @@ import (
"fmt" "fmt"
"io" "io"
"github.com/spf13/viper"
"github.com/chrislusf/seaweedfs/weed/filer2" "github.com/chrislusf/seaweedfs/weed/filer2"
"github.com/chrislusf/seaweedfs/weed/notification" "github.com/chrislusf/seaweedfs/weed/notification"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb" "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
@ -42,8 +40,8 @@ func (c *commandFsMetaNotify) Do(args []string, commandEnv *CommandEnv, writer i
} }
util.LoadConfiguration("notification", true) util.LoadConfiguration("notification", true)
v := viper.GetViper() v := util.GetViper()
notification.LoadConfiguration(v.Sub("notification")) notification.LoadConfiguration(v, "notification.")
ctx := context.Background() ctx := context.Background()

View file

@ -35,7 +35,7 @@ type StringProperties interface {
type StorageType string type StorageType string
type BackendStorageFactory interface { type BackendStorageFactory interface {
StorageType() StorageType StorageType() StorageType
BuildStorage(configuration StringProperties, id string) (BackendStorage, error) BuildStorage(configuration StringProperties, configPrefix string, id string) (BackendStorage, error)
} }
var ( var (
@ -48,19 +48,17 @@ func LoadConfiguration(config *viper.Viper) {
StorageBackendPrefix := "storage.backend" StorageBackendPrefix := "storage.backend"
backendSub := config.Sub(StorageBackendPrefix)
for backendTypeName := range config.GetStringMap(StorageBackendPrefix) { for backendTypeName := range config.GetStringMap(StorageBackendPrefix) {
backendStorageFactory, found := BackendStorageFactories[StorageType(backendTypeName)] backendStorageFactory, found := BackendStorageFactories[StorageType(backendTypeName)]
if !found { if !found {
glog.Fatalf("backend storage type %s not found", backendTypeName) glog.Fatalf("backend storage type %s not found", backendTypeName)
} }
backendTypeSub := backendSub.Sub(backendTypeName) for backendStorageId := range config.GetStringMap(StorageBackendPrefix + "." + backendTypeName) {
for backendStorageId := range backendSub.GetStringMap(backendTypeName) { if !config.GetBool(StorageBackendPrefix + "." + backendTypeName + "." + backendStorageId + ".enabled") {
if !backendTypeSub.GetBool(backendStorageId + ".enabled") {
continue continue
} }
backendStorage, buildErr := backendStorageFactory.BuildStorage(backendTypeSub.Sub(backendStorageId), backendStorageId) backendStorage, buildErr := backendStorageFactory.BuildStorage(config,
StorageBackendPrefix+"."+backendTypeName+"."+backendStorageId+".", backendStorageId)
if buildErr != nil { if buildErr != nil {
glog.Fatalf("fail to create backend storage %s.%s", backendTypeName, backendStorageId) glog.Fatalf("fail to create backend storage %s.%s", backendTypeName, backendStorageId)
} }
@ -82,7 +80,7 @@ func LoadFromPbStorageBackends(storageBackends []*master_pb.StorageBackend) {
glog.Warningf("storage type %s not found", storageBackend.Type) glog.Warningf("storage type %s not found", storageBackend.Type)
continue continue
} }
backendStorage, buildErr := backendStorageFactory.BuildStorage(newProperties(storageBackend.Properties), storageBackend.Id) backendStorage, buildErr := backendStorageFactory.BuildStorage(newProperties(storageBackend.Properties), "", storageBackend.Id)
if buildErr != nil { if buildErr != nil {
glog.Fatalf("fail to create backend storage %s.%s", storageBackend.Type, storageBackend.Id) glog.Fatalf("fail to create backend storage %s.%s", storageBackend.Type, storageBackend.Id)
} }

View file

@ -26,8 +26,8 @@ type S3BackendFactory struct {
func (factory *S3BackendFactory) StorageType() backend.StorageType { func (factory *S3BackendFactory) StorageType() backend.StorageType {
return backend.StorageType("s3") return backend.StorageType("s3")
} }
func (factory *S3BackendFactory) BuildStorage(configuration backend.StringProperties, id string) (backend.BackendStorage, error) { func (factory *S3BackendFactory) BuildStorage(configuration backend.StringProperties, configPrefix string, id string) (backend.BackendStorage, error) {
return newS3BackendStorage(configuration, id) return newS3BackendStorage(configuration, configPrefix, id)
} }
type S3BackendStorage struct { type S3BackendStorage struct {
@ -39,13 +39,13 @@ type S3BackendStorage struct {
conn s3iface.S3API conn s3iface.S3API
} }
func newS3BackendStorage(configuration backend.StringProperties, id string) (s *S3BackendStorage, err error) { func newS3BackendStorage(configuration backend.StringProperties, configPrefix string, id string) (s *S3BackendStorage, err error) {
s = &S3BackendStorage{} s = &S3BackendStorage{}
s.id = id s.id = id
s.aws_access_key_id = configuration.GetString("aws_access_key_id") s.aws_access_key_id = configuration.GetString(configPrefix + "aws_access_key_id")
s.aws_secret_access_key = configuration.GetString("aws_secret_access_key") s.aws_secret_access_key = configuration.GetString(configPrefix + "aws_secret_access_key")
s.region = configuration.GetString("region") s.region = configuration.GetString(configPrefix + "region")
s.bucket = configuration.GetString("bucket") s.bucket = configuration.GetString(configPrefix + "bucket")
s.conn, err = createSession(s.aws_access_key_id, s.aws_secret_access_key, s.region) s.conn, err = createSession(s.aws_access_key_id, s.aws_secret_access_key, s.region)
glog.V(0).Infof("created backend storage s3.%s for region %s bucket %s", s.id, s.region, s.bucket) glog.V(0).Infof("created backend storage s3.%s for region %s bucket %s", s.id, s.region, s.bucket)

View file

@ -1,8 +1,11 @@
package util package util
import ( import (
"github.com/chrislusf/seaweedfs/weed/glog" "strings"
"github.com/spf13/viper" "github.com/spf13/viper"
"github.com/chrislusf/seaweedfs/weed/glog"
) )
type Configuration interface { type Configuration interface {
@ -37,3 +40,11 @@ func LoadConfiguration(configFileName string, required bool) (loaded bool) {
return true return true
} }
func GetViper() *viper.Viper {
v := viper.GetViper()
v.AutomaticEnv()
v.SetEnvPrefix("weed")
v.SetEnvKeyReplacer(strings.NewReplacer(".", "_"))
return v
}

24
weed/util/config_test.go Normal file
View file

@ -0,0 +1,24 @@
package util
import (
"os"
"testing"
"github.com/stretchr/testify/assert"
)
func TestAllKeysWithEnv(t *testing.T) {
v := GetViper()
v.BindEnv("id")
v.BindEnv("foo", "foo")
// bind and define environment variables (including a nested one)
os.Setenv("WEED_ID", "13")
os.Setenv("WEED_FOO_BAR", "baz")
sub := v.Sub("foo")
assert.Equal(t, "13", v.GetString("id"))
assert.Equal(t, "baz", sub.GetString("bar"))
}