1 files changed, 47 insertions, 9 deletions
diff --git a/weed/filer/filer_conf.go b/weed/filer/filer_conf.go
index c58b26dc2..32fc647d9 100644
--- a/weed/filer/filer_conf.go
+++ b/weed/filer/filer_conf.go
@@ -3,6 +3,10 @@ package filer
 import (
 	"bytes"
 	"context"
+	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/pb"
+	"github.com/chrislusf/seaweedfs/weed/wdclient"
+	"google.golang.org/grpc"
 	"io"
 
 	"github.com/chrislusf/seaweedfs/weed/glog"
@@ -26,6 +30,29 @@ type FilerConf struct {
 	rules ptrie.Trie
 }
 
+func ReadFilerConf(filerGrpcAddress pb.ServerAddress, grpcDialOption grpc.DialOption, masterClient *wdclient.MasterClient) (*FilerConf, error) {
+	var buf bytes.Buffer
+	if err := pb.WithGrpcFilerClient(false, filerGrpcAddress, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
+		if masterClient != nil {
+			return ReadEntry(masterClient, client, DirectoryEtcSeaweedFS, FilerConfName, &buf)
+		} else {
+			content, err := ReadInsideFiler(client, DirectoryEtcSeaweedFS, FilerConfName)
+			buf = *bytes.NewBuffer(content)
+			return err
+		}
+	}); err != nil && err != filer_pb.ErrNotFound {
+		return nil, fmt.Errorf("read %s/%s: %v", DirectoryEtcSeaweedFS, FilerConfName, err)
+	}
+
+	fc := NewFilerConf()
+	if buf.Len() > 0 {
+		if err := fc.LoadFromBytes(buf.Bytes()); err != nil {
+			return nil, fmt.Errorf("parse %s/%s: %v", DirectoryEtcSeaweedFS, FilerConfName, err)
+		}
+	}
+	return fc, nil
+}
+
 func NewFilerConf() (fc *FilerConf) {
 	fc = &FilerConf{
 		rules: ptrie.New(),
@@ -48,12 +75,12 @@ func (fc *FilerConf) loadFromFiler(filer *Filer) (err error) {
 		return fc.LoadFromBytes(entry.Content)
 	}
 
-	return fc.loadFromChunks(filer, entry.Content, entry.Chunks)
+	return fc.loadFromChunks(filer, entry.Content, entry.Chunks, entry.Size())
 }
 
-func (fc *FilerConf) loadFromChunks(filer *Filer, content []byte, chunks []*filer_pb.FileChunk) (err error) {
+func (fc *FilerConf) loadFromChunks(filer *Filer, content []byte, chunks []*filer_pb.FileChunk, size uint64) (err error) {
 	if len(content) == 0 {
-		content, err = filer.readEntry(chunks)
+		content, err = filer.readEntry(chunks, size)
 		if err != nil {
 			glog.Errorf("read filer conf content: %v", err)
 			return
@@ -115,21 +142,32 @@ func (fc *FilerConf) MatchStorageRule(path string) (pathConf *filer_pb.FilerConf
 	return pathConf
 }
 
+func (fc *FilerConf) GetCollectionTtls(collection string) (ttls map[string]string) {
+	ttls = make(map[string]string)
+	fc.rules.Walk(func(key []byte, value interface{}) bool {
+		t := value.(*filer_pb.FilerConf_PathConf)
+		if t.Collection == collection {
+			ttls[t.LocationPrefix] = t.GetTtl()
+		}
+		return true
+	})
+	return ttls
+}
+
 // 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 != "" {
-		a.DiskType = b.DiskType
-	}
+	a.DiskType = util.Nvl(b.DiskType, a.DiskType)
 	a.Fsync = b.Fsync || a.Fsync
 	if b.VolumeGrowthCount > 0 {
 		a.VolumeGrowthCount = b.VolumeGrowthCount
 	}
-	if b.ReadOnly {
-		a.ReadOnly = b.ReadOnly
-	}
+	a.ReadOnly = b.ReadOnly || a.ReadOnly
+	a.DataCenter = util.Nvl(b.DataCenter, a.DataCenter)
+	a.Rack = util.Nvl(b.Rack, a.Rack)
+	a.DataNode = util.Nvl(b.DataNode, a.DataNode)
 }
 
 func (fc *FilerConf) ToProto() *filer_pb.FilerConf {