Merge branch 'master' of https://github.com/seaweedfs/seaweedfs-csi-driver

3 files changed, 64 insertions, 4 deletions

diff --git a/pkg/driver/driver.go b/pkg/driver/driver.go
index 67a6db0..5a375e1 100644
--- a/pkg/driver/driver.go
+++ b/pkg/driver/driver.go
@@ -5,6 +5,7 @@ import (
 	"os"
 
 	"github.com/container-storage-interface/spec/lib/go/csi"
+	"github.com/seaweedfs/seaweedfs-csi-driver/pkg/datalocality"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
@@ -44,6 +45,8 @@ type SeaweedFsDriver struct {
 	UidMap            string
 	GidMap            string
 	signature         int32
+	DataCenter        string
+	DataLocality      datalocality.DataLocality
 }
 
 func NewSeaweedFsDriver(filer, nodeID, endpoint string) *SeaweedFsDriver {
diff --git a/pkg/driver/mounter_seaweedfs.go b/pkg/driver/mounter_seaweedfs.go
index 51adfdb..12142af 100644
--- a/pkg/driver/mounter_seaweedfs.go
+++ b/pkg/driver/mounter_seaweedfs.go
@@ -7,6 +7,7 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/seaweedfs/seaweedfs-csi-driver/pkg/datalocality"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/util"
 )
@@ -74,7 +75,8 @@ func (seaweedFs *seaweedFsMounter) Mount(target string) (Unmounter, error) {
 		args = append(args, fmt.Sprintf("-collectionQuotaMB=%d", capacityMB))
 	}
 
-	// Initial values for override-able args
+	// Values for override-able args
+	//  Whitelist for merging with volContext
 	argsMap := map[string]string {
 		"collection": 			seaweedFs.collection,
 		"filer": 				strings.Join(filers, ","),
@@ -83,6 +85,34 @@ func (seaweedFs *seaweedFsMounter) Mount(target string) (Unmounter, error) {
 		"concurrentWriters":	fmt.Sprint(seaweedFs.driver.ConcurrentWriters),
 		"map.uid":				seaweedFs.driver.UidMap,
 		"map.gid":				seaweedFs.driver.GidMap,
+		"disk":					"",
+		"dataCenter":			"",
+		"replication":			"",
+		"ttl":					"",
+		"chunkSizeLimitMB":		"",
+		"volumeServerAccess":	"",
+		"readRetryTime":		"",
+	}
+
+	// Handle DataLocality
+	dataLocality := seaweedFs.driver.DataLocality;
+	// Try to override when set in context
+	if dataLocalityStr, ok := seaweedFs.volContext["dataLocality"]; ok{
+		// Convert to enum
+		dataLocalityRes, ok := datalocality.FromString(dataLocalityStr)
+		if(!ok){
+			glog.Warning("volumeContext 'dataLocality' invalid");
+		}else{
+			dataLocality = dataLocalityRes
+		}
+	}
+	if err := CheckDataLocality(&dataLocality, &seaweedFs.driver.DataCenter); err != nil {
+		return nil, err
+	}
+	// Settings based on type
+	switch(dataLocality){
+	case datalocality.Write_preferLocalDc:
+		argsMap["dataCenter"] = seaweedFs.driver.DataCenter;
 	}
 
 	// volContext-parameter -> mount-arg
@@ -94,11 +124,15 @@ func (seaweedFs *seaweedFsMounter) Mount(target string) (Unmounter, error) {
 		"diskType":		"disk",
 	}
 
+	// Explicitly ignored volContext args e.g. handled somewhere else
+	ignoreArgs := []string{
+		"volumeCapacity",
+		"dataLocality",
+	}
+
 	//	Merge volContext into argsMap with key-mapping
 	for arg, value := range seaweedFs.volContext {
-		if(arg == "volumeCapacity"){	// Ignore volumeCapacity, not the nicest solution like this :/
-			continue
-		}
+		if(in_arr(ignoreArgs, arg)){continue}
 
 		// Check if key-mapping exists
 		newArg, ok := parameterArgMap[arg]
@@ -106,6 +140,12 @@ func (seaweedFs *seaweedFsMounter) Mount(target string) (Unmounter, error) {
 			arg = newArg
 		}
 
+		// Check if arg can be applied
+		if _, ok := argsMap[arg]; !ok {
+			glog.Warningf("VolumeContext '%s' ignored", arg)
+			continue
+		}
+
 		// Write to args-map
 		argsMap[arg] = value
 	}
@@ -153,3 +193,12 @@ func parseVolumeCapacity(volumeCapacity string) int64 {
 	capacityMB := capacity / 1024 / 1024
 	return capacityMB
 }
+
+func in_arr(arr []string, val string) bool {
+	for _, v := range arr {
+		if(val == v) {
+			return true
+		}
+	}
+	return false
+}
diff --git a/pkg/driver/utils.go b/pkg/driver/utils.go
index b629543..e103079 100644
--- a/pkg/driver/utils.go
+++ b/pkg/driver/utils.go
@@ -8,6 +8,7 @@ import (
 	"sync"
 
 	"github.com/container-storage-interface/spec/lib/go/csi"
+	"github.com/seaweedfs/seaweedfs-csi-driver/pkg/datalocality"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"golang.org/x/net/context"
 	"google.golang.org/grpc"
@@ -121,3 +122,10 @@ func (km *KeyMutex) GetMutex(key string) *sync.Mutex {
 func (km *KeyMutex) RemoveMutex(key string) {
 	km.mutexes.Delete(key)
 }
+
+func CheckDataLocality(dataLocality *datalocality.DataLocality, dataCenter *string) error {
+	if(*dataLocality != datalocality.None && *dataCenter == ""){
+		return fmt.Errorf("dataLocality set, but not all locality-definitions were set")
+	}
+	return nil
+}