pass backend config from master to volume servers

weed/storage/backend/backend.go

@@ -3,10 +3,11 @@ package backend
 import (
 	"io"
 	"os"
+	"strings"
 	"time"
 
 	"github.com/chrislusf/seaweedfs/weed/glog"
-	"github.com/chrislusf/seaweedfs/weed/util"
+	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
 	"github.com/spf13/viper"
 )
 
@@ -21,14 +22,17 @@ type BackendStorageFile interface {
 }
 
 type BackendStorage interface {
-	Name() string
+	ToProperties() map[string]string
 	NewStorageFile(key string) BackendStorageFile
 }
 
+type StringProperties interface {
+	GetString(key string) string
+}
 type StorageType string
 type BackendStorageFactory interface {
 	StorageType() StorageType
-	BuildStorage(configuration util.Configuration, id string) (BackendStorage, error)
+	BuildStorage(configuration StringProperties, id string) (BackendStorage, error)
 }
 
 var (
@@ -49,6 +53,9 @@ func LoadConfiguration(config *viper.Viper) {
 		}
 		backendTypeSub := backendSub.Sub(backendTypeName)
 		for backendStorageId, _ := range backendSub.GetStringMap(backendTypeName) {
+			if !backendTypeSub.GetBool(backendStorageId + ".enabled") {
+				continue
+			}
 			backendStorage, buildErr := backendStorageFactory.BuildStorage(backendTypeSub.Sub(backendStorageId), backendStorageId)
 			if buildErr != nil {
 				glog.Fatalf("fail to create backend storage %s.%s", backendTypeName, backendStorageId)
@@ -61,3 +68,54 @@ func LoadConfiguration(config *viper.Viper) {
 	}
 
 }
+
+func LoadFromPbStorageBackends(storageBackends []*master_pb.StorageBackend) {
+
+	for _, storageBackend := range storageBackends {
+		backendStorageFactory, found := BackendStorageFactories[StorageType(storageBackend.Type)]
+		if !found {
+			glog.Warningf("storage type %s not found", storageBackend.Type)
+			continue
+		}
+		backendStorage, buildErr := backendStorageFactory.BuildStorage(newProperties(storageBackend.Properties), storageBackend.Id)
+		if buildErr != nil {
+			glog.Fatalf("fail to create backend storage %s.%s", storageBackend.Type, storageBackend.Id)
+		}
+		BackendStorages[storageBackend.Type+"."+storageBackend.Id] = backendStorage
+		if storageBackend.Id == "default" {
+			BackendStorages[storageBackend.Type] = backendStorage
+		}
+	}
+}
+
+type Properties struct {
+	m map[string]string
+}
+
+func newProperties(m map[string]string) *Properties {
+	return &Properties{m: m}
+}
+
+func (p *Properties) GetString(key string) string {
+	if v, found := p.m[key]; found {
+		return v
+	}
+	return ""
+}
+
+func ToPbStorageBackends() (backends []*master_pb.StorageBackend) {
+	for sName, s := range BackendStorages {
+		parts := strings.Split(sName, ".")
+		if len(parts) != 2 {
+			continue
+		}
+
+		sType, sId := parts[0], parts[1]
+		backends = append(backends, &master_pb.StorageBackend{
+			Type:       sType,
+			Id:         sId,
+			Properties: s.ToProperties(),
+		})
+	}
+	return
+}