Repository: puppetlabs/wash
Pull request #184 (2019-04-03 22:12:38)

Improve debugging using activity journal
Move to nanosecond granularity logging to debug slow operations.

Consistently print the journal ID for logged actions, and write all FUSE
activity to the journal.

Signed-off-by: Michael Smith <michael.smith@puppet.com>


Patch:
diff --git a/cmd/server.go b/cmd/server.go
index 204dbe3..b3b04c9 100644
--- a/cmd/server.go
+++ b/cmd/server.go
@@ -16,6 +16,7 @@ import (
 	cmdutil "github.com/puppetlabs/wash/cmd/util"
 	"github.com/puppetlabs/wash/config"
 	"github.com/puppetlabs/wash/fuse"
+	"github.com/puppetlabs/wash/journal"
 	"github.com/puppetlabs/wash/plugin"
 	"github.com/puppetlabs/wash/plugin/aws"
 	"github.com/puppetlabs/wash/plugin/docker"
@@ -68,6 +69,11 @@ func serverMain(cmd *cobra.Command, args []string) exitCode {
 		}()
 	}
 
+	// Close any open journals on shutdown to ensure remaining entries are flushed to disk.
+	defer func() {
+		journal.CloseAll()
+	}()
+
 	registry := plugin.NewRegistry()
 	loadInternalPlugins(registry)
 	if externalPluginsPath != "" {
@@ -153,7 +159,8 @@ func loadLogger(levelStr string, logfile string) (*os.File, error) {
 
 	log.SetLevel(level)
 	log.SetFormatter(&log.TextFormatter{
-		FullTimestamp: true,
+		FullTimestamp:   true,
+		TimestampFormat: time.RFC3339Nano,
 	})
 
 	var logFH *os.File
diff --git a/fuse/core.go b/fuse/core.go
index 8fe01ae..0a275d5 100644
--- a/fuse/core.go
+++ b/fuse/core.go
@@ -116,6 +116,8 @@ func (f *fuseNode) applyAttr(a *fuse.Attr, attr *plugin.EntryAttributes) {
 }
 
 func (f *fuseNode) Attr(ctx context.Context, a *fuse.Attr) error {
+	journal.Record(ctx, "FUSE: Attr %v", f)
+
 	var attr plugin.EntryAttributes
 	if f.parent == nil {
 		attr = plugin.Attributes(f.entry)
@@ -127,13 +129,15 @@ func (f *fuseNode) Attr(ctx context.Context, a *fuse.Attr) error {
 		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)
+			journal.Record(ctx, "FUSE: Attr errored %v, %v", f, err)
+			log.Warnf("FUSE: Error[Attr,%v,jid=%v]: %v", f, journal.GetID(ctx), 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)
+			journal.Record(ctx, "FUSE: Attr errored %v, %v", f, err)
+			log.Warnf("FUSE: Error[Attr,%v,jid=%v]: %v", f, journal.GetID(ctx), err)
 			return err
 		}
 		attr = plugin.Attributes(updatedEntry)
@@ -144,18 +148,19 @@ func (f *fuseNode) Attr(ctx context.Context, a *fuse.Attr) error {
 	}
 
 	f.applyAttr(a, &attr)
-	log.Infof("FUSE: Attr[%v] %v %v", f.ftype, f, a)
+	journal.Record(ctx, "FUSE: Attr finished %v", f)
+	log.Infof("FUSE: Attr[%v,jid=%v] %v %v", f.ftype, journal.GetID(ctx), f, a)
 	return nil
 }
 
 func (f *fuseNode) Listxattr(ctx context.Context, req *fuse.ListxattrRequest, resp *fuse.ListxattrResponse) error {
-	log.Infof("FUSE: Listxattr[%v,pid=%v] %v", f.ftype, req.Pid, f)
+	log.Infof("FUSE: Listxattr[%v,jid=%v] %v", f.ftype, journal.GetID(ctx), f)
 	resp.Append("wash.id")
 	return nil
 }
 
 func (f *fuseNode) Getxattr(ctx context.Context, req *fuse.GetxattrRequest, resp *fuse.GetxattrResponse) error {
-	log.Infof("FUSE: Getxattr[%v,pid=%v] %v", f.ftype, req.Pid, f)
+	log.Infof("FUSE: Getxattr[%v,jid=%v] %v", f.ftype, journal.GetID(ctx), f)
 	switch req.Name {
 	case "wash.id":
 		resp.Xattr = []byte(f.String())
diff --git a/fuse/dir.go b/fuse/dir.go
index 4ede297..763d1eb 100644
--- a/fuse/dir.go
+++ b/fuse/dir.go
@@ -39,7 +39,7 @@ func (d *dir) Lookup(ctx context.Context, req *fuse.LookupRequest, resp *fuse.Lo
 
 	entries, err := d.children(ctx)
 	if err != nil {
-		log.Warnf("FUSE: Error[Find,%v,%v]: %v", d, req.Name, err)
+		log.Warnf("FUSE: Error[Find,%v,%v,jid=%v]: %v", d, req.Name, journal.GetID(ctx), err)
 		journal.Record(ctx, "FUSE: Find %v in %v errored: %v", req.Name, d, err)
 		return nil, fuse.ENOENT
 	}
@@ -51,7 +51,7 @@ func (d *dir) Lookup(ctx context.Context, req *fuse.LookupRequest, resp *fuse.Lo
 		return nil, fuse.ENOENT
 	}
 
-	log.Infof("FUSE: Find[d,pid=%v] %v/%v", req.Pid, d, cname)
+	log.Infof("FUSE: Find[d,jid=%v] %v/%v", journal.GetID(ctx), d, cname)
 	if plugin.ListAction.IsSupportedOn(entry) {
 		childdir := newDir(d.entry.(plugin.Group), entry.(plugin.Group))
 		journal.Record(ctx, "FUSE: Found directory %v", childdir)
@@ -77,12 +77,12 @@ func (d *dir) ReadDirAll(ctx context.Context) ([]fuse.Dirent, error) {
 
 	entries, err := d.children(ctx)
 	if err != nil {
-		log.Warnf("FUSE: Error[List,%v]: %v", d, err)
+		log.Warnf("FUSE: Error[List,%v,jid=%v]: %v", d, journal.GetID(ctx), err)
 		journal.Record(ctx, "FUSE: List %v errored: %v", d, err)
 		return nil, err
 	}
 
-	log.Infof("FUSE: List %v in %v", len(entries), d)
+	log.Infof("FUSE: List[jid=%v] %v in %v", journal.GetID(ctx), len(entries), d)
 
 	res := make([]fuse.Dirent, 0, len(entries))
 	for cname, entry := range entries {
diff --git a/fuse/file.go b/fuse/file.go
index 20b7ed5..c465724 100644
--- a/fuse/file.go
+++ b/fuse/file.go
@@ -31,7 +31,7 @@ func (f *file) Open(ctx context.Context, req *fuse.OpenRequest, resp *fuse.OpenR
 	journal.Record(ctx, "FUSE: Open %v", f)
 
 	// Initiate content request and return a channel providing the results.
-	log.Infof("FUSE: Opening[pid=%v] %v", req.Pid, f)
+	log.Infof("FUSE: Opening[jid=%v] %v", journal.GetID(ctx), f)
 	if plugin.ReadAction.IsSupportedOn(f.entry) {
 		content, err := plugin.CachedOpen(ctx, f.entry.(plugin.Readable))
 		if err != nil {
@@ -40,11 +40,11 @@ func (f *file) Open(ctx context.Context, req *fuse.OpenRequest, resp *fuse.OpenR
 			return nil, err
 		}
 
-		log.Infof("FUSE: Opened[pid=%v] %v", req.Pid, f)
+		log.Infof("FUSE: Opened[jid=%v] %v", journal.GetID(ctx), f)
 		journal.Record(ctx, "FUSE: Opened %v", f)
 		return &fileHandle{r: content, id: f.String()}, nil
 	}
-	log.Warnf("FUSE: Error[Open,%v]: cannot open this entry", f)
+	log.Warnf("FUSE: Error[Open,%v,jid=%v]: cannot open this entry", f, journal.GetID(ctx))
 	journal.Record(ctx, "FUSE: Open unsupported on %v", f)
 	return nil, fuse.ENOTSUP
 }
@@ -60,7 +60,7 @@ var _ = fs.HandleReader(fileHandle{})
 
 // Release closes the open file.
 func (fh fileHandle) Release(ctx context.Context, req *fuse.ReleaseRequest) error {
-	log.Infof("FUSE: Release[pid=%v] %v", req.Pid, fh.id)
+	log.Infof("FUSE: Release[jid=%v] %v", journal.GetID(ctx), fh.id)
 	journal.Record(ctx, "FUSE: Release %v", fh.id)
 	if closer, ok := fh.r.(io.Closer); ok {
 		return closer.Close()
@@ -75,7 +75,7 @@ func (fh fileHandle) Read(ctx context.Context, req *fuse.ReadRequest, resp *fuse
 	if err == io.EOF {
 		err = nil
 	}
-	log.Infof("FUSE: Read[pid=%v] %v, %v/%v bytes starting at %v: %v", fh.id, req.Pid, n, req.Size, req.Offset, err)
+	log.Infof("FUSE: Read[jid=%v] %v, %v/%v bytes starting at %v: %v", journal.GetID(ctx), fh.id, n, req.Size, req.Offset, err)
 	journal.Record(ctx, "FUSE: Read %v/%v bytes starting at %v from %v: %v", n, req.Size, req.Offset, fh.id, err)
 	resp.Data = buf[:n]
 	return err
diff --git a/journal/core.go b/journal/core.go
index f66e230..997ea62 100644
--- a/journal/core.go
+++ b/journal/core.go
@@ -36,6 +36,12 @@ var journalDir = func() string {
 }()
 var expires = 30 * time.Second
 
+// CloseAll ensures open journals are flushed to disk and closed.
+// Use when the application is shutting down.
+func CloseAll() {
+	journalCache.Flush()
+}
+
 // Dir gets the directory where journal entries are written.
 func Dir() string {
 	return journalDir
@@ -46,6 +52,11 @@ func SetDir(dir string) {
 	journalDir = dir
 }
 
+// GetID returns the Journal ID stored in the context.
+func GetID(ctx context.Context) string {
+	return ctx.Value(Key).(string)
+}
+
 // Record writes a new entry to the journal identified by the ID at `journal.Key` in
 // the provided context. It also writes to the server logs at the debug level. If no ID
 // is registered, the entry is written to the server logs at the warning level. If the
@@ -80,11 +91,11 @@ func Record(ctx context.Context, msg string, a ...interface{}) {
 			return nil, err
 		}
 
-		// Use a syncedFile to ensure each write is committed to the disk. The file is only guaranteed to
-		// be closed when it's evicted from the cache, which may not happen before shutdown.
-		l := log.New()
-		l.Out = &syncedFile{id: id, file: f}
-		l.Level = log.TraceLevel
+		l := &log.Logger{
+			Out:       f,
+			Level:     log.TraceLevel,
+			Formatter: &log.JSONFormatter{TimestampFormat: time.RFC3339Nano},
+		}
 		return l, nil
 	})
 	if err != nil {
@@ -94,24 +105,6 @@ func Record(ctx context.Context, msg string, a ...interface{}) {
 	obj.(*log.Logger).Printf(msg, a...)
 }
 
-type syncedFile struct {
-	id   string
-	file *os.File
-}
-
-// Write syncs the data immediately after every write operation.
-func (sw *syncedFile) Write(b []byte) (n int, err error) {
-	n, err = sw.file.Write(b)
-	if err := sw.file.Sync(); err != nil {
-		log.Warnf("Error syncing journal %v to disk: %v", sw.id, err)
-	}
-	return n, err
-}
-
-func (sw *syncedFile) Close() error {
-	return sw.file.Close()
-}
-
 func closeJournal(id string, obj interface{}) {
 	logger, ok := obj.(*log.Logger)
 	if !ok {
diff --git a/plugin/aws/ec2Instance.go b/plugin/aws/ec2Instance.go
index a2dc0ec..36d3503 100644
--- a/plugin/aws/ec2Instance.go
+++ b/plugin/aws/ec2Instance.go
@@ -106,7 +106,7 @@ func (d describeInstanceResult) toMeta() plugin.EntryMetadata {
 }
 
 func (inst *ec2Instance) cachedDescribeInstance(ctx context.Context) (describeInstanceResult, error) {
-	info, err := plugin.CachedOp("DescribeInstance", inst, 15*time.Second, func() (interface{}, error) {
+	info, err := plugin.CachedOp(ctx, "DescribeInstance", inst, 15*time.Second, func() (interface{}, error) {
 		request := &ec2Client.DescribeInstancesInput{
 			InstanceIds: []*string{
 				awsSDK.String(inst.Name()),
diff --git a/plugin/aws/ec2InstanceConsoleOutput.go b/plugin/aws/ec2InstanceConsoleOutput.go
index fd63d52..444bc86 100644
--- a/plugin/aws/ec2InstanceConsoleOutput.go
+++ b/plugin/aws/ec2InstanceConsoleOutput.go
@@ -62,7 +62,7 @@ func (o consoleOutput) toMeta() plugin.EntryMetadata {
 }
 
 func (cl *ec2InstanceConsoleOutput) cachedConsoleOutput(ctx context.Context) (consoleOutput, error) {
-	output, err := plugin.CachedOp("ConsoleOutput", cl, 30*time.Second, func() (interface{}, error) {
+	output, err := plugin.CachedOp(ctx, "ConsoleOutput", cl, 30*time.Second, func() (interface{}, error) {
 		request := &ec2Client.GetConsoleOutputInput{
 			InstanceId: awsSDK.String(cl.inst.Name()),
 		}
diff --git a/plugin/aws/s3Bucket.go b/plugin/aws/s3Bucket.go
index a4da323..1f21e18 100644
--- a/plugin/aws/s3Bucket.go
+++ b/plugin/aws/s3Bucket.go
@@ -217,7 +217,7 @@ func (b *s3Bucket) getRegion(ctx context.Context) (string, error) {
 	// Note that the callback to CachedOp also creates a new client for that region.
 	// We use CachedOp with a long expiration to ensure region is fetched infrequently.
 	// You can force a retry by deleting the cache entry if there was an error.
-	resp, err := plugin.CachedOp("Region", b, 24*time.Hour, func() (interface{}, error) {
+	resp, err := plugin.CachedOp(ctx, "Region", b, 24*time.Hour, func() (interface{}, error) {
 		locRequest := &s3Client.GetBucketLocationInput{Bucket: awsSDK.String(b.Name())}
 		resp, err := b.client.GetBucketLocationWithContext(ctx, locRequest)
 		if err != nil {
diff --git a/plugin/aws/s3Object.go b/plugin/aws/s3Object.go
index 0563544..f7ad85b 100644
--- a/plugin/aws/s3Object.go
+++ b/plugin/aws/s3Object.go
@@ -51,7 +51,7 @@ func newS3Object(o *s3Client.Object, name string, bucket string, key string, cli
 }
 
 func (o *s3Object) cachedHeadObject(ctx context.Context) (*s3Client.HeadObjectOutput, error) {
-	resp, err := plugin.CachedOp("HeadObject", o, 15*time.Second, func() (interface{}, error) {
+	resp, err := plugin.CachedOp(ctx, "HeadObject", o, 15*time.Second, func() (interface{}, error) {
 		request := &s3Client.HeadObjectInput{
 			Bucket: awsSDK.String(o.bucket),
 			Key:    awsSDK.String(o.key),
diff --git a/plugin/cache.go b/plugin/cache.go
index f86b99d..96affd3 100644
--- a/plugin/cache.go
+++ b/plugin/cache.go
@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/puppetlabs/wash/datastore"
+	"github.com/puppetlabs/wash/journal"
 )
 
 var cache datastore.Cache
@@ -88,7 +89,9 @@ type opFunc func() (interface{}, error)
 // the existing CachedList, CachedOpen, and CachedMetadata methods provide.
 // For example, CachedOp could be useful to cache an API request whose response
 // lets you implement Attr() and Metadata() for the given entry.
-func CachedOp(opName string, entry Entry, ttl time.Duration, op opFunc) (interface{}, error) {
+//
+// CachedOp uses the supplied context to determine which journal to log to.
+func CachedOp(ctx context.Context, opName string, entry Entry, ttl time.Duration, op opFunc) (interface{}, error) {
 	if !opNameRegex.MatchString(opName) {
 		panic(fmt.Sprintf("The opName %v does not match %v", opName, opNameRegex.String()))
 	}
@@ -103,7 +106,7 @@ func CachedOp(opName string, entry Entry, ttl time.Duration, op opFunc) (interfa
 		panic("plugin.CachedOp: received a negative TTL")
 	}
 
-	return cachedOp(opName, entry, ttl, op)
+	return cachedOp(ctx, opName, entry, ttl, op)
 }
 
 // DuplicateCNameErr represents a duplicate cname error, which
@@ -140,7 +143,7 @@ func (c DuplicateCNameErr) Error() string {
 // 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) {
+	cachedEntries, err := cachedDefaultOp(ctx, ListOp, g, func() (interface{}, error) {
 		entries, err := g.List(ctx)
 		if err != nil {
 			return nil, err
@@ -181,7 +184,7 @@ func CachedList(ctx context.Context, g Group) (map[string]Entry, error) {
 // such as ReadAt or wrap it in a SectionReader. Using Read operations on the cached
 // reader will change it and make subsequent uses of the cached reader invalid.
 func CachedOpen(ctx context.Context, r Readable) (SizedReader, error) {
-	cachedContent, err := cachedDefaultOp(OpenOp, r, func() (interface{}, error) {
+	cachedContent, err := cachedDefaultOp(ctx, OpenOp, r, func() (interface{}, error) {
 		return r.Open(ctx)
 	})
 
@@ -194,7 +197,7 @@ func CachedOpen(ctx context.Context, r Readable) (SizedReader, error) {
 
 // CachedMetadata caches an entry's Metadata method
 func CachedMetadata(ctx context.Context, e Entry) (EntryMetadata, error) {
-	cachedMetadata, err := cachedDefaultOp(MetadataOp, e, func() (interface{}, error) {
+	cachedMetadata, err := cachedDefaultOp(ctx, MetadataOp, e, func() (interface{}, error) {
 		return e.Metadata(ctx)
 	})
 
@@ -206,15 +209,15 @@ func CachedMetadata(ctx context.Context, e Entry) (EntryMetadata, error) {
 }
 
 // Common helper for CachedList, CachedOpen and CachedMetadata
-func cachedDefaultOp(opCode defaultOpCode, entry Entry, op opFunc) (interface{}, error) {
+func cachedDefaultOp(ctx context.Context, opCode defaultOpCode, entry Entry, op opFunc) (interface{}, error) {
 	opName := defaultOpCodeToNameMap[opCode]
 	ttl := entry.getTTLOf(opCode)
 
-	return cachedOp(opName, entry, ttl, op)
+	return cachedOp(ctx, opName, entry, ttl, op)
 }
 
 // Common helper for CachedOp and cachedDefaultOp.
-func cachedOp(opName string, entry Entry, ttl time.Duration, op opFunc) (interface{}, error) {
+func cachedOp(ctx context.Context, opName string, entry Entry, ttl time.Duration, op opFunc) (interface{}, error) {
 	if cache == nil {
 		if notRunningTests() {
 			panic("The cache was not initialized. You can initialize the cache by invoking plugin.InitCache()")
@@ -224,6 +227,7 @@ func cachedOp(opName string, entry Entry, ttl time.Duration, op opFunc) (interfa
 	}
 
 	if entry.id() == "" {
+		journal.Record(ctx, "Cached op %v on %v had no cache ID", opName, entry.name())
 		// This can happen if the entry's just been created, and it calls one of the Cached*
 		// methods to query its Metadata so that it can set its initial attributes. In that case,
 		// just invoke op. Ideally we should panic here, but panicking makes it significantly more
diff --git a/plugin/kubernetes/pod.go b/plugin/kubernetes/pod.go
index cf9505d..08866aa 100644
--- a/plugin/kubernetes/pod.go
+++ b/plugin/kubernetes/pod.go
@@ -81,7 +81,7 @@ func (p *pod) fetchLogContent(ctx context.Context) ([]byte, error) {
 }
 
 func (p *pod) cachedPodInfo(ctx context.Context) (podInfoResult, error) {
-	cachedPdInfo, err := plugin.CachedOp("PodInfo", p, 15*time.Second, func() (interface{}, error) {
+	cachedPdInfo, err := plugin.CachedOp(ctx, "PodInfo", p, 15*time.Second, func() (interface{}, error) {
 		result := podInfoResult{}
 		pd, err := p.client.CoreV1().Pods(p.ns).Get(p.Name(), metav1.GetOptions{})
 		if err != nil {


Test patch:
diff --git a/journal/core_test.go b/journal/core_test.go
index 2bf0350..7b7f518 100644
--- a/journal/core_test.go
+++ b/journal/core_test.go
@@ -14,7 +14,7 @@ import (
 
 func TestRecord(t *testing.T) {
 	// Ensure the cache is cleaned up afterward.
-	defer journalCache.Flush()
+	defer CloseAll()
 
 	// Log to a journal
 	Record(context.WithValue(context.Background(), Key, "1"), "hello there")
@@ -27,7 +27,7 @@ func TestRecord(t *testing.T) {
 
 func TestLogExpired(t *testing.T) {
 	// Ensure the cache is cleaned up afterward.
-	defer journalCache.Flush()
+	defer CloseAll()
 
 	// Ensure entries use a very short
 	expires = 1 * time.Millisecond
@@ -46,7 +46,7 @@ func TestLogExpired(t *testing.T) {
 
 func TestLogReused(t *testing.T) {
 	// Ensure the cache is cleaned up afterward.
-	defer journalCache.Flush()
+	defer CloseAll()
 	ctx := context.WithValue(context.Background(), Key, "3")
 
 	// Log twice
diff --git a/plugin/cache_test.go b/plugin/cache_test.go
index 4cedffa..3fb0757 100644
--- a/plugin/cache_test.go
+++ b/plugin/cache_test.go
@@ -148,7 +148,7 @@ func (suite *CacheTestSuite) TestCachedOp() {
 	makePanicFunc := func(opName string, ttl time.Duration) func() {
 		return func() {
 			entry := newCacheTestsMockEntry("mock")
-			_, _ = CachedOp("List", entry, ttl, func() (interface{}, error) { return nil, nil })
+			_, _ = CachedOp(context.Background(), "List", entry, ttl, func() (interface{}, error) { return nil, nil })
 		}
 	}
 
@@ -178,7 +178,7 @@ func (suite *CacheTestSuite) TestCachedOp() {
 	opKey := "Op::id"
 	generateValueMatcher := suite.makeGenerateValueMatcher("result")
 	suite.cache.On("GetOrUpdate", opKey, opTTL, false, mock.MatchedBy(generateValueMatcher)).Return("result", nil).Once()
-	v, err := CachedOp(opName, entry, opTTL, op)
+	v, err := CachedOp(context.Background(), opName, entry, opTTL, op)
 	if suite.NoError(err) {
 		suite.Equal("result", v)
 	}