filer confi: support hierachical configuration

weed/filer/filer_conf.go

@@ -103,21 +103,27 @@ func (fc *FilerConf) DeleteLocationConf(locationPrefix string) {
 	return
 }
 
-var (
-	EmptyFilerConfPathConf = &filer_pb.FilerConf_PathConf{}
-)
-
 func (fc *FilerConf) MatchStorageRule(path string) (pathConf *filer_pb.FilerConf_PathConf) {
+	pathConf = &filer_pb.FilerConf_PathConf{}
 	fc.rules.MatchPrefix([]byte(path), func(key []byte, value interface{}) bool {
-		pathConf = value.(*filer_pb.FilerConf_PathConf)
+		t := value.(*filer_pb.FilerConf_PathConf)
+		mergePathConf(pathConf, t)
 		return true
 	})
-	if pathConf == nil {
-		return EmptyFilerConfPathConf
-	}
 	return pathConf
 }
 
+// merge if values in b is not empty, merge them into a
+func mergePathConf(a, b *filer_pb.FilerConf_PathConf) {
+	a.Collection = util.Nvl(b.Collection, a.Collection)
+	a.Replication = util.Nvl(b.Replication, a.Replication)
+	a.Ttl = util.Nvl(b.Ttl, a.Ttl)
+	if b.DiskType != filer_pb.FilerConf_PathConf_NONE {
+		a.DiskType = b.DiskType
+	}
+	a.Fsync = b.Fsync || a.Fsync
+}
+
 func (fc *FilerConf) ToProto() *filer_pb.FilerConf {
 	m := &filer_pb.FilerConf{}
 	fc.rules.Walk(func(key []byte, value interface{}) bool {

weed/filer/filer_conf_test.go

@@ -20,10 +20,15 @@ func TestFilerConf(t *testing.T) {
 			LocationPrefix: "/buckets/abcd",
 			Collection:     "abcd",
 		},
+		{
+			LocationPrefix: "/buckets/",
+			Replication: "001",
+		},
 	}}
 	fc.doLoadConf(conf)
 
 	assert.Equal(t, "abc", fc.MatchStorageRule("/buckets/abc/jasdf").Collection)
 	assert.Equal(t, "abcd", fc.MatchStorageRule("/buckets/abcd/jasdf").Collection)
+	assert.Equal(t, "001", fc.MatchStorageRule("/buckets/abc/jasdf").Replication)
 
 }