3 files changed, 112 insertions, 3 deletions

diff --git a/weed/filesys/meta_cache/id_mapper.go b/weed/filesys/meta_cache/id_mapper.go
new file mode 100644
index 000000000..4799669e4
--- /dev/null
+++ b/weed/filesys/meta_cache/id_mapper.go
@@ -0,0 +1,101 @@
+package meta_cache
+
+import (
+	"fmt"
+	"strconv"
+	"strings"
+)
+
+type UidGidMapper struct {
+	uidMapper *IdMapper
+	gidMapper *IdMapper
+}
+
+type IdMapper struct {
+	localToFiler map[uint32]uint32
+	filerToLocal map[uint32]uint32
+}
+
+// UidGidMapper translates local uid/gid to filer uid/gid
+// The local storage always persists the same as the filer.
+// The local->filer translation happens when updating the filer first and later saving to meta_cache.
+// And filer->local happens when reading from the meta_cache.
+func NewUidGidMapper(uidPairsStr, gidPairStr string) (*UidGidMapper, error) {
+	uidMapper, err := newIdMapper(uidPairsStr)
+	if err != nil {
+		return nil, err
+	}
+	gidMapper, err := newIdMapper(gidPairStr)
+	if err != nil {
+		return nil, err
+	}
+
+	return &UidGidMapper{
+		uidMapper: uidMapper,
+		gidMapper: gidMapper,
+	}, nil
+}
+
+func (m *UidGidMapper) LocalToFiler(uid, gid uint32) (uint32,uint32) {
+	return m.uidMapper.LocalToFiler(uid), m.gidMapper.LocalToFiler(gid)
+}
+func (m *UidGidMapper) FilerToLocal(uid, gid uint32) (uint32,uint32) {
+	return m.uidMapper.FilerToLocal(uid), m.gidMapper.FilerToLocal(gid)
+}
+
+func (m *IdMapper) LocalToFiler(id uint32) (uint32) {
+	value, found := m.localToFiler[id]
+	if found {
+		return value
+	}
+	return id
+}
+func (m *IdMapper) FilerToLocal(id uint32) (uint32) {
+	value, found := m.filerToLocal[id]
+	if found {
+		return value
+	}
+	return id
+}
+
+func newIdMapper(pairsStr string) (*IdMapper, error) {
+
+	localToFiler, filerToLocal, err := parseUint32Pairs(pairsStr)
+	if err != nil {
+		return nil, err
+	}
+
+	return &IdMapper{
+		localToFiler: localToFiler,
+		filerToLocal: filerToLocal,
+	}, nil
+
+}
+
+func parseUint32Pairs(pairsStr string) (localToFiler, filerToLocal map[uint32]uint32, err error) {
+
+	if pairsStr == "" {
+		return
+	}
+
+	localToFiler = make(map[uint32]uint32)
+	filerToLocal = make(map[uint32]uint32)
+	for _, pairStr := range strings.Split(pairsStr, ",") {
+		pair := strings.Split(pairStr, ":")
+		localUidStr, filerUidStr := pair[0], pair[1]
+		localUid, localUidErr := strconv.Atoi(localUidStr)
+		if localUidErr != nil {
+			err = fmt.Errorf("failed to parse local %d: %v", localUidStr, localUidErr)
+			return
+		}
+		filerUid, filerUidErr := strconv.Atoi(filerUidStr)
+		if filerUidErr != nil {
+			err = fmt.Errorf("failed to parse remote %s: %v", filerUidStr, filerUidErr)
+			return
+		}
+		localToFiler[uint32(localUid)] = uint32(filerUid)
+		filerToLocal[uint32(filerUid)] = uint32(localUid)
+	}
+
+	return
+}
diff --git a/weed/filesys/meta_cache/meta_cache.go b/weed/filesys/meta_cache/meta_cache.go
index f714fde09..ac193a493 100644
--- a/weed/filesys/meta_cache/meta_cache.go
+++ b/weed/filesys/meta_cache/meta_cache.go
@@ -20,12 +20,14 @@ type MetaCache struct {
 	actualStore filer.FilerStore
 	sync.RWMutex
 	visitedBoundary *bounded_tree.BoundedTree
+	uidGidMapper    *UidGidMapper
 }
 
-func NewMetaCache(dbFolder string) *MetaCache {
+func NewMetaCache(dbFolder string, uidGidMapper *UidGidMapper) *MetaCache {
 	return &MetaCache{
 		actualStore:     openMetaStore(dbFolder),
 		visitedBoundary: bounded_tree.NewBoundedTree(),
+		uidGidMapper:    uidGidMapper,
 	}
 }
 
@@ -58,7 +60,7 @@ func (mc *MetaCache) doInsertEntry(ctx context.Context, entry *filer.Entry) erro
 	return mc.actualStore.InsertEntry(ctx, entry)
 }
 
-func (mc *MetaCache) AtomicUpdateEntry(ctx context.Context, oldPath util.FullPath, newEntry *filer.Entry) error {
+func (mc *MetaCache) AtomicUpdateEntryFromFiler(ctx context.Context, oldPath util.FullPath, newEntry *filer.Entry) error {
 	mc.Lock()
 	defer mc.Unlock()
 
@@ -103,6 +105,7 @@ func (mc *MetaCache) FindEntry(ctx context.Context, fp util.FullPath) (entry *fi
 	if err != nil {
 		return nil, err
 	}
+	mc.mapIdFromFilerToLocal(entry)
 	filer_pb.AfterEntryDeserialization(entry.Chunks)
 	return
 }
@@ -122,6 +125,7 @@ func (mc *MetaCache) ListDirectoryEntries(ctx context.Context, dirPath util.Full
 		return nil, err
 	}
 	for _, entry := range entries {
+		mc.mapIdFromFilerToLocal(entry)
 		filer_pb.AfterEntryDeserialization(entry.Chunks)
 	}
 	return entries, err
@@ -132,3 +136,7 @@ func (mc *MetaCache) Shutdown() {
 	defer mc.Unlock()
 	mc.actualStore.Shutdown()
 }
+
+func (mc *MetaCache) mapIdFromFilerToLocal(entry *filer.Entry) {
+	entry.Attr.Uid, entry.Attr.Gid = mc.uidGidMapper.FilerToLocal(entry.Attr.Uid, entry.Attr.Gid)
+}
diff --git a/weed/filesys/meta_cache/meta_cache_subscribe.go b/weed/filesys/meta_cache/meta_cache_subscribe.go
index bd98666ed..c20edb9b7 100644
--- a/weed/filesys/meta_cache/meta_cache_subscribe.go
+++ b/weed/filesys/meta_cache/meta_cache_subscribe.go
@@ -39,7 +39,7 @@ func SubscribeMetaEvents(mc *MetaCache, selfSignature int32, client filer_pb.Fil
 			glog.V(4).Infof("creating %v", key)
 			newEntry = filer.FromPbEntry(dir, message.NewEntry)
 		}
-		return mc.AtomicUpdateEntry(context.Background(), oldPath, newEntry)
+		return mc.AtomicUpdateEntryFromFiler(context.Background(), oldPath, newEntry)
 	}
 
 	for {