Repository: puppetlabs/wash
Pull request #181 (2019-04-02 21:18:42)

Properly refresh the entry's attributes in FUSE


Patch:
diff --git a/README.md b/README.md
index bfc9efb..fda6bcc 100644
--- a/README.md
+++ b/README.md
@@ -173,9 +173,7 @@ If the `wash` daemon exits with a exit status of 255, that typically means that
 
 ## Roadmap
 
-We're actively soliciting community feedback and input on our roadmap! Don't hesitate to file issues for new features, new plugin types, new primitives, new command-line tools, or anything else that crosses your mind. You can also chat with us directly on `#wash` on [Slack](https://slack.puppet.com/).
-
-Project maintainers are not actively working on all of these things, but any of these are directions we would support others in pursuing.
+We're activilely soliciting community feedback and input on our roadmap! Don't hesitate to file issues for new features, new plugin types, new primitives, new command-line tools, or anything else that crosses your mind. You can also chat with us directly on `#wash` on [Slack](https://slack.puppet.com/).
 
 ### Primitives
 
diff --git a/api/helpers.go b/api/helpers.go
index 8afb7a4..441be4c 100644
--- a/api/helpers.go
+++ b/api/helpers.go
@@ -40,8 +40,6 @@ func findEntry(ctx context.Context, root plugin.Entry, segments []string) (plugi
 	path := strings.Join(segments, "/")
 
 	curEntry := root
-	curEntryID := "/" + plugin.CName(root)
-
 	visitedSegments := make([]string, 0, cap(segments))
 	for _, segment := range segments {
 		switch curGroup := curEntry.(type) {
@@ -57,19 +55,8 @@ func findEntry(ctx context.Context, root plugin.Entry, segments []string) (plugi
 			}
 
 			// Search for the specific entry
-			var found bool
-			for _, entry := range entries {
-				if plugin.CName(entry) == segment {
-					found = true
-
-					curEntry = entry
-					curEntryID += "/" + segment
-					visitedSegments = append(visitedSegments, segment)
-
-					break
-				}
-			}
-			if !found {
+			entry, ok := entries[segment]
+			if !ok {
 				reason := fmt.Sprintf("The %v entry does not exist", segment)
 				if len(visitedSegments) != 0 {
 					reason += fmt.Sprintf(" in the %v group", strings.Join(visitedSegments, "/"))
@@ -77,6 +64,9 @@ func findEntry(ctx context.Context, root plugin.Entry, segments []string) (plugi
 
 				return nil, entryNotFoundResponse(path, reason)
 			}
+
+			curEntry = entry
+			visitedSegments = append(visitedSegments, segment)
 		default:
 			reason := fmt.Sprintf("The entry %v is not a group", strings.Join(visitedSegments, "/"))
 			return nil, entryNotFoundResponse(path, reason)
diff --git a/api/list.go b/api/list.go
index b02a652..2b734fd 100644
--- a/api/list.go
+++ b/api/list.go
@@ -64,11 +64,11 @@ var listHandler handler = func(w http.ResponseWriter, r *http.Request, p params)
 		return result
 	}
 
-	result := make([]apitypes.ListEntry, len(entries)+1)
-	result[0] = info(group)
+	result := make([]apitypes.ListEntry, 0, len(entries)+1)
+	result = append(result, info(group))
 
-	for i, entry := range entries {
-		result[i+1] = info(entry)
+	for _, entry := range entries {
+		result = append(result, info(entry))
 	}
 	journal.Record(ctx, "API: List %v %+v", path, result)
 
diff --git a/fuse/core.go b/fuse/core.go
index 1b346b2..8fe01ae 100644
--- a/fuse/core.go
+++ b/fuse/core.go
@@ -3,6 +3,7 @@ package fuse
 
 import (
 	"context"
+	"fmt"
 	"os"
 	"os/user"
 	"strconv"
@@ -29,7 +30,7 @@ func newRoot(registry *plugin.Registry) Root {
 // Root presents the root of the filesystem.
 func (r *Root) Root() (fs.Node, error) {
 	log.Infof("Entering root of filesystem")
-	return newDir(r.registry), nil
+	return newDir(nil, r.registry), nil
 }
 
 func getIDs() (uint32, uint32) {
@@ -53,17 +54,17 @@ func getIDs() (uint32, uint32) {
 
 var uid, gid = getIDs()
 
-var attrRefreshInterval = 5 * time.Second
-
 type fuseNode struct {
 	ftype             string
+	parent            plugin.Group
 	entry             plugin.Entry
 	entryCreationTime time.Time
 }
 
-func newFuseNode(ftype string, entry plugin.Entry) *fuseNode {
+func newFuseNode(ftype string, parent plugin.Group, entry plugin.Entry) *fuseNode {
 	return &fuseNode{
 		ftype:             ftype,
+		parent:            parent,
 		entry:             entry,
 		entryCreationTime: time.Now(),
 	}
@@ -77,8 +78,8 @@ const sizeUnknown = ^uint64(0)
 
 // Applies attributes where non-default, and sets defaults otherwise.
 func (f *fuseNode) applyAttr(a *fuse.Attr, attr *plugin.EntryAttributes) {
-	// This doesn't quite work for some reason.
-	a.Valid = attrRefreshInterval
+	// Setting a.Valid to 1 second avoids frequent Attr calls.
+	a.Valid = 1 * time.Second
 
 	// TODO: tie this to actual hard links in plugins
 	a.Nlink = 1
@@ -115,17 +116,33 @@ func (f *fuseNode) applyAttr(a *fuse.Attr, attr *plugin.EntryAttributes) {
 }
 
 func (f *fuseNode) Attr(ctx context.Context, a *fuse.Attr) error {
-	// TODO: The code below's a temporary hack to show that the refreshing
-	// behavior works. It blows up after several successive ls calls b/c each
-	// child simultaneously attempts to refresh their attributes.
-	if time.Since(f.entryCreationTime) >= attrRefreshInterval {
-		err := plugin.RefreshAttributes(ctx, f.entry)
+	var attr plugin.EntryAttributes
+	if f.parent == nil {
+		attr = plugin.Attributes(f.entry)
+	} else {
+		// FUSE caches nodes for a long time, meaning there's a chance
+		// that f's attributes are outdated. CachedList returns the entry's
+		// and its sibling's updated attributes in a single request, so use
+		// it to get f's updated attributes.
+		entries, err := plugin.CachedList(ctx, f.parent)
 		if err != nil {
+			err := fmt.Errorf("could not refresh the attributes: %v", err)
+			log.Warnf("FUSE: Error[Attr,%v]: %v", f, err)
+			return err
+		}
+		updatedEntry, ok := entries[plugin.CName(f.entry)]
+		if !ok {
+			err := fmt.Errorf("entry does not exist anymore")
 			log.Warnf("FUSE: Error[Attr,%v]: %v", f, err)
 			return err
 		}
+		attr = plugin.Attributes(updatedEntry)
+		// NOTE: We could set f.entry to updatedEntry, but doing so would require
+		// a separate mutex which may hinder performance. Since updating f.entry
+		// is not strictly necessary for the other FUSE operations, we choose to
+		// leave it alone.
 	}
-	attr := plugin.Attributes(f.entry)
+
 	f.applyAttr(a, &attr)
 	log.Infof("FUSE: Attr[%v] %v %v", f.ftype, f, a)
 	return nil
diff --git a/fuse/dir.go b/fuse/dir.go
index 1f0a182..4ede297 100644
--- a/fuse/dir.go
+++ b/fuse/dir.go
@@ -20,17 +20,17 @@ var _ fs.Node = (*dir)(nil)
 var _ = fs.NodeRequestLookuper(&dir{})
 var _ = fs.HandleReadDirAller(&dir{})
 
-func newDir(e plugin.Group) *dir {
-	return &dir{newFuseNode("d", e)}
+func newDir(p plugin.Group, e plugin.Group) *dir {
+	return &dir{newFuseNode("d", p, e)}
 }
 
-func (d *dir) children(ctx context.Context) ([]plugin.Entry, error) {
+func (d *dir) children(ctx context.Context) (map[string]plugin.Entry, error) {
 	// Cache List requests. FUSE often lists the contents then immediately calls find on individual entries.
 	if plugin.ListAction.IsSupportedOn(d.entry) {
 		return plugin.CachedList(ctx, d.entry.(plugin.Group))
 	}
 
-	return []plugin.Entry{}, fuse.ENOENT
+	return map[string]plugin.Entry{}, fuse.ENOENT
 }
 
 // Lookup searches a directory for children.
@@ -44,31 +44,31 @@ func (d *dir) Lookup(ctx context.Context, req *fuse.LookupRequest, resp *fuse.Lo
 		return nil, fuse.ENOENT
 	}
 
-	for _, entry := range entries {
-		cname := plugin.CName(entry)
-		if cname == req.Name {
-			log.Infof("FUSE: Find[d,pid=%v] %v/%v", req.Pid, d, cname)
-			if plugin.ListAction.IsSupportedOn(entry) {
-				childdir := newDir(entry.(plugin.Group))
-				journal.Record(ctx, "FUSE: Found directory %v", childdir)
-				// Prefetch directory entries into the cache
-				go func() {
-					// Need to use a different context here because we still want the prefetch
-					// to happen even when the current context is cancelled.
-					jid := journal.PIDToID(int(req.Pid))
-					ctx := context.WithValue(context.Background(), journal.Key, jid)
-					_, err := childdir.children(ctx)
-					journal.Record(ctx, "FUSE: Prefetching children of %v complete: %v", childdir, err)
-				}()
-				return childdir, nil
-			}
-
-			journal.Record(ctx, "FUSE: Found file %v/%v", d, cname)
-			return newFile(entry), nil
-		}
+	cname := req.Name
+	entry, ok := entries[cname]
+	if !ok {
+		journal.Record(ctx, "FUSE: %v not found in %v", req.Name, d)
+		return nil, fuse.ENOENT
 	}
-	journal.Record(ctx, "FUSE: %v not found in %v", req.Name, d)
-	return nil, fuse.ENOENT
+
+	log.Infof("FUSE: Find[d,pid=%v] %v/%v", req.Pid, d, cname)
+	if plugin.ListAction.IsSupportedOn(entry) {
+		childdir := newDir(d.entry.(plugin.Group), entry.(plugin.Group))
+		journal.Record(ctx, "FUSE: Found directory %v", childdir)
+		// Prefetch directory entries into the cache
+		go func() {
+			// Need to use a different context here because we still want the prefetch
+			// to happen even when the current context is cancelled.
+			jid := journal.PIDToID(int(req.Pid))
+			ctx := context.WithValue(context.Background(), journal.Key, jid)
+			_, err := childdir.children(ctx)
+			journal.Record(ctx, "FUSE: Prefetching children of %v complete: %v", childdir, err)
+		}()
+		return childdir, nil
+	}
+
+	journal.Record(ctx, "FUSE: Found file %v/%v", d, cname)
+	return newFile(d.entry.(plugin.Group), entry), nil
 }
 
 // ReadDirAll lists all children of the directory.
@@ -84,14 +84,14 @@ func (d *dir) ReadDirAll(ctx context.Context) ([]fuse.Dirent, error) {
 
 	log.Infof("FUSE: List %v in %v", len(entries), d)
 
-	res := make([]fuse.Dirent, len(entries))
-	for i, entry := range entries {
+	res := make([]fuse.Dirent, 0, len(entries))
+	for cname, entry := range entries {
 		var de fuse.Dirent
-		de.Name = plugin.CName(entry)
-		if plugin.ListAction.IsSupportedOn(d.entry) {
+		de.Name = cname
+		if plugin.ListAction.IsSupportedOn(entry) {
 			de.Type = fuse.DT_Dir
 		}
-		res[i] = de
+		res = append(res, de)
 	}
 	journal.Record(ctx, "FUSE: Listed in %v: %+v", d, res)
 	return res, nil
diff --git a/fuse/file.go b/fuse/file.go
index 7070c9f..20b7ed5 100644
--- a/fuse/file.go
+++ b/fuse/file.go
@@ -22,8 +22,8 @@ var _ = fs.NodeOpener(&file{})
 var _ = fs.NodeGetxattrer(&file{})
 var _ = fs.NodeListxattrer(&file{})
 
-func newFile(e plugin.Entry) *file {
-	return &file{newFuseNode("f", e)}
+func newFile(p plugin.Group, e plugin.Entry) *file {
+	return &file{newFuseNode("f", p, e)}
 }
 
 // Open a file for reading.
diff --git a/plugin/cache.go b/plugin/cache.go
index 1d87dbc..f86b99d 100644
--- a/plugin/cache.go
+++ b/plugin/cache.go
@@ -136,7 +136,10 @@ func (c DuplicateCNameErr) Error() string {
 
 // CachedList caches a Group object's List method. It also sets the
 // children's IDs to <parent_id> + "/" + <child_cname>.
-func CachedList(ctx context.Context, g Group) ([]Entry, error) {
+//
+// CachedList returns a map of <entry_cname> => <entry_object> to optimize
+// querying a specific entry.
+func CachedList(ctx context.Context, g Group) (map[string]Entry, error) {
 	cachedEntries, err := cachedDefaultOp(ListOp, g, func() (interface{}, error) {
 		entries, err := g.List(ctx)
 		if err != nil {
@@ -163,14 +166,14 @@ func CachedList(ctx context.Context, g Group) ([]Entry, error) {
 			entry.setID(id)
 		}
 
-		return entries, nil
+		return searchedEntries, nil
 	})
 
 	if err != nil {
 		return nil, err
 	}
 
-	return cachedEntries.([]Entry), nil
+	return cachedEntries.(map[string]Entry), nil
 }
 
 // CachedOpen caches a Readable object's Open method.


Test patch:
diff --git a/api/cache_test.go b/api/cache_test.go
index 4bc7886..695801e 100644
--- a/api/cache_test.go
+++ b/api/cache_test.go
@@ -85,8 +85,9 @@ func (suite *CacheHandlerTestSuite) TestClearCache() {
 	group.SetTestID("/dir")
 	group.On("List", mock.Anything).Return([]plugin.Entry{}, nil)
 
+	expectedChildren := make(map[string]plugin.Entry)
 	if children, err := plugin.CachedList(context.Background(), group); suite.Nil(err) {
-		suite.Equal([]plugin.Entry{}, children)
+		suite.Equal(expectedChildren, children)
 	}
 
 	// Test clearing a different cache
@@ -97,7 +98,7 @@ func (suite *CacheHandlerTestSuite) TestClearCache() {
 	suite.Equal("[]\n", w.Body.String())
 
 	if children, err := plugin.CachedList(context.Background(), group); suite.Nil(err) {
-		suite.Equal([]plugin.Entry{}, children)
+		suite.Equal(expectedChildren, children)
 	}
 
 	// Test clearing the cache
@@ -108,7 +109,7 @@ func (suite *CacheHandlerTestSuite) TestClearCache() {
 	suite.Equal(`["List::/dir"]`, strings.TrimSpace(w.Body.String()))
 
 	if children, err := plugin.CachedList(context.Background(), group); suite.Nil(err) {
-		suite.Equal([]plugin.Entry{}, children)
+		suite.Equal(expectedChildren, children)
 	}
 
 	group.AssertNumberOfCalls(suite.T(), "List", 2)
diff --git a/plugin/cache_test.go b/plugin/cache_test.go
index f5e1f4b..4cedffa 100644
--- a/plugin/cache_test.go
+++ b/plugin/cache_test.go
@@ -200,7 +200,13 @@ func (suite *CacheTestSuite) TestDuplicateCNameErr() {
 	suite.Regexp("my_plugin plugin", err.Error())
 }
 
-func (suite *CacheTestSuite) testCachedDefaultOp(op defaultOpCode, opName string, mockValue interface{}, cachedDefaultOp cachedDefaultOpFunc) {
+func (suite *CacheTestSuite) testCachedDefaultOp(
+	op defaultOpCode,
+	opName string,
+	opValue interface{},
+	mungedOpValue interface{},
+	cachedDefaultOp cachedDefaultOpFunc,
+) {
 	ctx := context.Background()
 
 	// Test that cachedDefaultOp panics if the cache == nil
@@ -218,20 +224,20 @@ func (suite *CacheTestSuite) testCachedDefaultOp(op defaultOpCode, opName string
 	// entry.id() == "" (possible if the entry's just been created and
 	// it needs to query its metadata to set its initial attributes).
 	suite.Panics(panicFunc, "entry.id() returned an empty ID")
-	entry.On(opName, ctx).Return(mockValue, nil)
+	entry.On(opName, ctx).Return(opValue, nil)
 	v, err := cachedDefaultOp(ctx, entry)
 	if suite.NoError(err) {
-		suite.Equal(mockValue, v)
+		suite.Equal(mungedOpValue, v)
 	}
 
 	// Test that cachedDefaultOp does _not_ call cache#GetOrUpdate for an
 	// entry that's turned off caching
 	entry.SetTestID("id")
-	entry.On(opName, ctx).Return(mockValue, nil)
+	entry.On(opName, ctx).Return(opValue, nil)
 	entry.DisableCachingFor(op)
 	v, err = cachedDefaultOp(ctx, entry)
 	if suite.NoError(err) {
-		suite.Equal(mockValue, v)
+		suite.Equal(mungedOpValue, v)
 	}
 	suite.cache.AssertNotCalled(suite.T(), "GetOrUpdate")
 
@@ -240,20 +246,30 @@ func (suite *CacheTestSuite) testCachedDefaultOp(op defaultOpCode, opName string
 	// right arguments.
 	opTTL := 5 * time.Second
 	entry.SetTTLOf(op, opTTL)
-	entry.On(opName, ctx).Return(mockValue, nil)
+	entry.On(opName, ctx).Return(opValue, nil)
 	opKey := opName + "::" + "id"
-	generateValueMatcher := suite.makeGenerateValueMatcher(mockValue)
-	suite.cache.On("GetOrUpdate", opKey, opTTL, false, mock.MatchedBy(generateValueMatcher)).Return(mockValue, nil).Once()
+	generateValueMatcher := suite.makeGenerateValueMatcher(mungedOpValue)
+	suite.cache.On("GetOrUpdate", opKey, opTTL, false, mock.MatchedBy(generateValueMatcher)).Return(mungedOpValue, nil).Once()
 	v, err = cachedDefaultOp(ctx, entry)
 	if suite.NoError(err) {
-		suite.Equal(mockValue, v)
+		suite.Equal(mungedOpValue, v)
 	}
 	suite.cache.AssertCalled(suite.T(), "GetOrUpdate", opKey, opTTL, false, mock.MatchedBy(generateValueMatcher))
 }
 
+func toMap(children []Entry) map[string]Entry {
+	mp := make(map[string]Entry)
+	for _, child := range children {
+		mp[CName(child)] = child
+	}
+
+	return mp
+}
+
 func (suite *CacheTestSuite) TestCachedListDefaultOp() {
 	mockChildren := []Entry{newCacheTestsMockEntry("mockChild")}
-	suite.testCachedDefaultOp(ListOp, "List", mockChildren, func(ctx context.Context, e Entry) (interface{}, error) {
+	mungedOpValue := toMap(mockChildren)
+	suite.testCachedDefaultOp(ListOp, "List", mockChildren, mungedOpValue, func(ctx context.Context, e Entry) (interface{}, error) {
 		return CachedList(ctx, e.(Group))
 	})
 }
@@ -301,9 +317,9 @@ func (suite *CacheTestSuite) TestCachedListSetEntryID() {
 	entry.On("List", ctx).Return(mockChildren, nil).Once()
 	children, err := CachedList(ctx, entry)
 	if suite.NoError(err) {
-		if suite.Equal(mockChildren, children) {
-			suite.Equal("/foo#child1", children[0].id())
-			suite.Equal("/child2", children[1].id())
+		if suite.Equal(toMap(mockChildren), children) {
+			suite.Equal("/foo#child1", children["foo#child1"].id())
+			suite.Equal("/child2", children["child2"].id())
 		}
 	}
 
@@ -314,23 +330,23 @@ func (suite *CacheTestSuite) TestCachedListSetEntryID() {
 	entry.On("List", ctx).Return(mockChildren, nil).Once()
 	children, err = CachedList(ctx, entry)
 	if suite.NoError(err) {
-		if suite.Equal(mockChildren, children) {
-			suite.Equal("/parent/foo#child1", children[0].id())
-			suite.Equal("/parent/child2", children[1].id())
+		if suite.Equal(toMap(mockChildren), children) {
+			suite.Equal("/parent/foo#child1", children["foo#child1"].id())
+			suite.Equal("/parent/child2", children["child2"].id())
 		}
 	}
 }
 
 func (suite *CacheTestSuite) TestCachedOpen() {
 	mockReader := strings.NewReader("foo")
-	suite.testCachedDefaultOp(OpenOp, "Open", mockReader, func(ctx context.Context, e Entry) (interface{}, error) {
+	suite.testCachedDefaultOp(OpenOp, "Open", mockReader, mockReader, func(ctx context.Context, e Entry) (interface{}, error) {
 		return CachedOpen(ctx, e.(Readable))
 	})
 }
 
 func (suite *CacheTestSuite) TestCachedMetadata() {
 	mockEntryMetadata := EntryMetadata{"foo": "bar"}
-	suite.testCachedDefaultOp(MetadataOp, "Metadata", mockEntryMetadata, func(ctx context.Context, e Entry) (interface{}, error) {
+	suite.testCachedDefaultOp(MetadataOp, "Metadata", mockEntryMetadata, mockEntryMetadata, func(ctx context.Context, e Entry) (interface{}, error) {
 		return CachedMetadata(ctx, e)
 	})
 }