vercel · chelkyl · Mar 7, 2022 · Mar 7, 2022 · Mar 7, 2022 · Mar 10, 2022
@@ -79,6 +79,9 @@ func main() {
 			return &run.RunCommand{Config: cf, Ui: ui, Processes: processes},
 				nil
 		},
+		"logs": func() (cli.Command, error) {
+			return &run.LogsCommand{Config: cf, UI: ui}, nil
+		},
 		"prune": func() (cli.Command, error) {
 			return &prune.PruneCommand{Config: cf, Ui: ui}, nil
 		},

@@ -5,6 +5,7 @@ package cache
 
 import (
 	"sync"
+	"time"
 
 	"github.com/vercel/turborepo/cli/internal/config"
 )
@@ -24,6 +25,7 @@ type asyncCache struct {
 type cacheRequest struct {
 	target   string
 	key      string
+	start    time.Time
 	duration int
 	files    []string
 }
@@ -40,17 +42,18 @@ func newAsyncCache(realCache Cache, config *config.Config) Cache {
 	return c
 }
 
-func (c *asyncCache) Put(target string, key string, duration int, files []string) error {
+func (c *asyncCache) Put(target string, key string, start time.Time, duration int, files []string) error {
 	c.requests <- cacheRequest{
 		target:   target,
 		key:      key,
 		files:    files,
 		duration: duration,
+		start:    start,
 	}
 	return nil
 }
 
-func (c *asyncCache) Fetch(target string, key string, files []string) (bool, []string, int, error) {
+func (c *asyncCache) Fetch(target string, key string, files []string) (bool, []string, time.Time, int, error) {
 	return c.realCache.Fetch(target, key, files)
 }
 
@@ -72,7 +75,7 @@ func (c *asyncCache) Shutdown() {
 // run implements the actual async logic.
 func (c *asyncCache) run() {
 	for r := range c.requests {
-		c.realCache.Put(r.target, r.key, r.duration, r.files)
+		c.realCache.Put(r.target, r.key, r.start, r.duration, r.files) //nolint:golint,errcheck // nothing to do
 	}
 	c.wg.Done()
 }
@@ -7,6 +7,7 @@ package cache
 
 import (
 	"fmt"
+	"time"
 
 	"github.com/vercel/turborepo/cli/internal/analytics"
 	"github.com/vercel/turborepo/cli/internal/config"
@@ -18,9 +19,9 @@ import (
 type Cache interface {
 	// Fetch returns true if there is a cache it. It is expected to move files
 	// into their correct position as a side effect
-	Fetch(target string, hash string, files []string) (bool, []string, int, error)
+	Fetch(target string, hash string, files []string) (bool, []string, time.Time, int, error)
 	// Put caches files for a given hash
-	Put(target string, hash string, duration int, files []string) error
+	Put(target string, hash string, start time.Time, duration int, files []string) error
 	Clean(target string)
 	CleanAll()
 	Shutdown()
@@ -29,11 +30,15 @@ type Cache interface {
 const cacheEventHit = "HIT"
 const cacheEventMiss = "MISS"
 
+// notime is the time used to represent invalid or undefined time
+var notime = time.Date(0, time.January, 0, 0, 0, 0, 0, time.UTC)
+
 type CacheEvent struct {
-	Source   string `mapstructure:"source"`
-	Event    string `mapstructure:"event"`
-	Hash     string `mapstructure:"hash"`
-	Duration int    `mapstructure:"duration"`
+	Source   string    `mapstructure:"source"`
+	Event    string    `mapstructure:"event"`
+	Hash     string    `mapstructure:"hash"`
+	Start    time.Time `mapstructure:"start"`
+	Duration int       `mapstructure:"duration"`
 }
 
 // New creates a new cache
@@ -68,16 +73,16 @@ type cacheMultiplexer struct {
 	caches []Cache
 }
 
-func (mplex cacheMultiplexer) Put(target string, key string, duration int, files []string) error {
-	return mplex.storeUntil(target, key, duration, files, len(mplex.caches))
+func (mplex cacheMultiplexer) Put(target string, key string, start time.Time, duration int, files []string) error {
+	return mplex.storeUntil(target, key, start, duration, files, len(mplex.caches))
 }
 
 // storeUntil stores artifacts into higher priority caches than the given one.
 // Used after artifact retrieval to ensure we have them in eg. the directory cache after
 // downloading from the RPC cache.
 // This is a little inefficient since we could write the file to plz-out then copy it to the dir cache,
 // but it's hard to fix that without breaking the cache abstraction.
-func (mplex cacheMultiplexer) storeUntil(target string, key string, duration int, outputGlobs []string, stopAt int) error {
+func (mplex cacheMultiplexer) storeUntil(target string, key string, start time.Time, duration int, outputGlobs []string, stopAt int) error {
 	// Attempt to store on all caches simultaneously.
 	g := &errgroup.Group{}
 	for i, cache := range mplex.caches {
@@ -86,7 +91,7 @@ func (mplex cacheMultiplexer) storeUntil(target string, key string, duration int
 		}
 		c := cache
 		g.Go(func() error {
-			return c.Put(target, key, duration, outputGlobs)
+			return c.Put(target, key, start, duration, outputGlobs)
 		})
 	}
 
@@ -97,19 +102,19 @@ func (mplex cacheMultiplexer) storeUntil(target string, key string, duration int
 	return nil
 }
 
-func (mplex cacheMultiplexer) Fetch(target string, key string, files []string) (bool, []string, int, error) {
+func (mplex cacheMultiplexer) Fetch(target string, key string, files []string) (bool, []string, time.Time, int, error) {
 	// Retrieve from caches sequentially; if we did them simultaneously we could
 	// easily write the same file from two goroutines at once.
 	for i, cache := range mplex.caches {
-		if ok, actualFiles, duration, err := cache.Fetch(target, key, files); ok {
+		if ok, actualFiles, start, duration, err := cache.Fetch(target, key, files); ok {
 			// Store this into other caches. We can ignore errors here because we know
 			// we have previously successfully stored in a higher-priority cache, and so the overall
 			// result is a success at fetching. Storing in lower-priority caches is an optimization.
-			mplex.storeUntil(target, key, duration, actualFiles, i)
-			return ok, actualFiles, duration, err
+			mplex.storeUntil(target, key, start, duration, actualFiles, i) //nolint:golint,errcheck // nothing to do
+			return ok, actualFiles, start, duration, err
 		}
 	}
-	return false, files, 0, nil
+	return false, files, notime, 0, nil
 }
 
 func (mplex cacheMultiplexer) Clean(target string) {

@@ -4,12 +4,14 @@
 package cache
 
 import (
+	"bufio"
 	"encoding/json"
 	"fmt"
 	"io/ioutil"
 	"os"
 	"path/filepath"
 	"runtime"
+	"time"
 
 	"github.com/vercel/turborepo/cli/internal/analytics"
 	"github.com/vercel/turborepo/cli/internal/config"
@@ -29,31 +31,31 @@ func newFsCache(config *config.Config, recorder analytics.Recorder) Cache {
 }
 
 // Fetch returns true if items are cached. It moves them into position as a side effect.
-func (f *fsCache) Fetch(target, hash string, _unusedOutputGlobs []string) (bool, []string, int, error) {
+func (f *fsCache) Fetch(target, hash string, _unusedOutputGlobs []string) (bool, []string, time.Time, int, error) {
 	cachedFolder := filepath.Join(f.cacheDirectory, hash)
 
 	// If it's not in the cache bail now
 	if !fs.PathExists(cachedFolder) {
-		f.logFetch(false, hash, 0)
-		return false, nil, 0, nil
+		f.logFetch(false, hash, notime, 0)
+		return false, nil, notime, 0, nil
 	}
 
 	// Otherwise, copy it into position
 	err := fs.RecursiveCopyOrLinkFile(cachedFolder, target, fs.DirPermissions, true, true)
 	if err != nil {
 		// TODO: what event to log here?
-		return false, nil, 0, fmt.Errorf("error moving artifact from cache into %v: %w", target, err)
+		return false, nil, notime, 0, fmt.Errorf("error moving artifact from cache into %v: %w", target, err)
 	}
 
 	meta, err := ReadCacheMetaFile(filepath.Join(f.cacheDirectory, hash+"-meta.json"))
 	if err != nil {
-		return false, nil, 0, fmt.Errorf("error reading cache metadata: %w", err)
+		return false, nil, notime, 0, fmt.Errorf("error reading cache metadata: %w", err)
 	}
-	f.logFetch(true, hash, meta.Duration)
-	return true, nil, meta.Duration, nil
+	f.logFetch(true, hash, meta.Start, meta.Duration)
+	return true, nil, meta.Start, meta.Duration, nil
 }
 
-func (f *fsCache) logFetch(hit bool, hash string, duration int) {
+func (f *fsCache) logFetch(hit bool, hash string, start time.Time, duration int) {
 	var event string
 	if hit {
 		event = cacheEventHit
@@ -65,11 +67,12 @@ func (f *fsCache) logFetch(hit bool, hash string, duration int) {
 		Event:    event,
 		Hash:     hash,
 		Duration: duration,
+		Start:    start,
 	}
 	f.recorder.LogEvent(payload)
 }
 
-func (f *fsCache) Put(target, hash string, duration int, files []string) error {
+func (f *fsCache) Put(target, hash string, start time.Time, duration int, files []string) error {
 	g := new(errgroup.Group)
 
 	numDigesters := runtime.NumCPU()
@@ -108,6 +111,7 @@ func (f *fsCache) Put(target, hash string, duration int, files []string) error {
 	WriteCacheMetaFile(filepath.Join(f.cacheDirectory, hash+"-meta.json"), &CacheMetadata{
 		Duration: duration,
 		Hash:     hash,
+		Start:    start,
 	})
 
 	return nil
@@ -126,8 +130,9 @@ func (cache *fsCache) Shutdown() {}
 // CacheMetadata stores duration and hash information for a cache entry so that aggregate Time Saved calculations
 // can be made from artifacts from various caches
 type CacheMetadata struct {
-	Hash     string `json:"hash"`
-	Duration int    `json:"duration"`
+	Hash     string    `json:"hash"`
+	Duration int       `json:"duration"`
+	Start    time.Time `json:"start"`
 }
 
 // WriteCacheMetaFile writes cache metadata file at a path
@@ -156,3 +161,37 @@ func ReadCacheMetaFile(path string) (*CacheMetadata, error) {
 	}
 	return &config, nil
 }
+
+// AppendHashesFile adds a hash to a file at path
+// Note: naively assuming locks are not needed
+func AppendHashesFile(path string, hash string) error {
+	file, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
+	if err != nil {
+		return err
+	}
+
+	defer file.Close() //nolint:golint,errcheck // nothing to do
+
+	if _, err = file.WriteString(hash + "\n"); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// ReadHashesFile reads hashes stored line by line from a file at path
+func ReadHashesFile(path string) ([]string, error) {
+	file, err := os.Open(path)
+	if err != nil {
+		return nil, err
+	}
+
+	defer file.Close() //nolint:golint,errcheck // nothing to do
+
+	var hashes []string
+	scanner := bufio.NewScanner(file)
+	for scanner.Scan() {
+		hashes = append(hashes, scanner.Text())
+	}
+	return hashes, scanner.Err()
+}
@@ -53,7 +53,8 @@ var mtime = time.Date(2000, time.January, 1, 0, 0, 0, 0, time.UTC)
 // nobody is the usual uid / gid of the 'nobody' user.
 const nobody = 65534
 
-func (cache *httpCache) Put(target, hash string, duration int, files []string) error {
+// start is ignored
+func (cache *httpCache) Put(target, hash string, start time.Time, duration int, files []string) error {
 	// if cache.writable {
 	cache.requestLimiter.acquire()
 	defer cache.requestLimiter.release()
@@ -135,19 +136,19 @@ func (cache *httpCache) storeFile(tw *tar.Writer, repoRelativePath string) error
 	return err
 }
 
-func (cache *httpCache) Fetch(target, key string, _unusedOutputGlobs []string) (bool, []string, int, error) {
+func (cache *httpCache) Fetch(target, key string, _unusedOutputGlobs []string) (bool, []string, time.Time, int, error) {
 	cache.requestLimiter.acquire()
 	defer cache.requestLimiter.release()
 	hit, files, duration, err := cache.retrieve(key)
 	if err != nil {
 		// TODO: analytics event?
-		return false, files, duration, fmt.Errorf("failed to retrieve files from HTTP cache: %w", err)
+		return false, files, notime, duration, fmt.Errorf("failed to retrieve files from HTTP cache: %w", err)
 	}
-	cache.logFetch(hit, key, duration)
-	return hit, files, duration, err
+	cache.logFetch(hit, key, notime, duration)
+	return hit, files, notime, duration, err
 }
 
-func (cache *httpCache) logFetch(hit bool, hash string, duration int) {
+func (cache *httpCache) logFetch(hit bool, hash string, start time.Time, duration int) {
 	var event string
 	if hit {
 		event = cacheEventHit
@@ -159,6 +160,7 @@ func (cache *httpCache) logFetch(hit bool, hash string, duration int) {
 		Event:    event,
 		Hash:     hash,
 		Duration: duration,
+		Start:    start,
 	}
 	cache.recorder.LogEvent(payload)
 }