Enable clear/purge for failed digests

golden/go/filediffstore/filediffstore.go

 package filediffstore
 
 import (
         "bytes"
         "crypto/md5"
         "encoding/base64"
         "encoding/json"
         "fmt"
         "image"
         "image/png"
         "io"
         "io/ioutil"
         "net/http"
         "os"
         "path/filepath"
+        "strings"
         "sync"
         "time"
 
         "github.com/boltdb/bolt"
         "github.com/hashicorp/golang-lru"
         metrics "github.com/rcrowley/go-metrics"
         "github.com/skia-dev/glog"
         "go.skia.org/infra/go/fileutil"
         "go.skia.org/infra/go/gs"
         "go.skia.org/infra/go/util"
@@ skipping 184 matching lines @@
 
         return f.failureDB.Update(func(tx *bolt.Tx) error {
                 bucket, err := tx.CreateBucketIfNotExists([]byte(FAILURE_BUCKET))
                 if err != nil {
                         return err
                 }
                 return bucket.Put([]byte(failure.Digest), jsonData)
         })
 }
 
+func (f *FileDiffStore) purgeDigestFailures(digests []string) error {
+        updated := false
+        err := f.failureDB.Update(func(tx *bolt.Tx) error {
+                bucket := tx.Bucket([]byte(FAILURE_BUCKET))
+                if bucket == nil {
+                        return nil
+                }
+
+                for _, d := range digests {
+                        if bucket.Get([]byte(d)) != nil {
+                                updated = true
+                                if err := bucket.Delete([]byte(d)); err != nil {
+                                        return err
+                                }
+                        }
+                }
+
+                return nil
+        })
+
+        if (err == nil) && updated {
+                return f.loadDigestFailures()
+        }
+        return err
+}
+
 // loadDigestFailures loads all digest failures to
 func (f *FileDiffStore) loadDigestFailures() error {
         newFailures := make(map[string]*diff.DigestFailure, len(f.unavailableDigests))
         err := f.failureDB.View(func(tx *bolt.Tx) error {
                 bucket := tx.Bucket([]byte(FAILURE_BUCKET))
                 if bucket == nil {
                         return nil
                 }
 
                 cursor := bucket.Cursor()
                 for k, v := cursor.First(); k != nil; k, v = cursor.Next() {
                         dFailure := &diff.DigestFailure{}
                         if err := json.Unmarshal(v, dFailure); err != nil {
                                 return err
                         }
                         newFailures[string(k)] = dFailure
                 }
                 return nil
         })
         if err == nil {
                 f.unavailableMutex.Lock()
                 f.unavailableDigests = newFailures
                 f.unavailableMutex.Unlock()
         }
         return err
 }
 
-func (f *FileDiffStore) PurgeDigests(digests []string, purgeGS bool) {
-        // TODO (stephana): To be implemented in next CL.
+func (f *FileDiffStore) PurgeDigests(digests []string, purgeGS bool) error {
+        // Remove from GS if requested.
+        if purgeGS {
+                for _, d := range digests {
+                        if err := f.removeImageFromGS(d); err != nil {
+                                return err
+                        }
+                }
+        }
+
+        for _, d := range digests {
+                if err := f.removeImageFromCache(d); err != nil {
+                        return err
+                }
+        }
+
+        // Remove from image cache.
+        for _, d := range digests {
+                f.imageCache.Remove(d)
+        }
+
+        // Remove all metrics from disk cache.
+        if err := f.removeDiffMetricsFromFileCache(digests); err != nil {
+                return err
+        }
+
+        // Remove all diff metrics from LRU cache.
+        for _, ki := range f.diffCache.Keys() {
+                k := ki.(string)
+                for _, d := range digests {
+                        if strings.Contains(k, d) {
+                                f.diffCache.Remove(ki)
+                        }
+                }
+        }
+        return f.purgeDigestFailures(digests)
 }
 
 type WorkerReq struct {
         id     interface{}
         respCh chan<- *WorkerResp
 }
 
 type WorkerResp struct {
         id  interface{}
         val interface{}
@@ skipping 259 matching lines @@
         d, err := json.MarshalIndent(diffMetrics, "", "    ")
         if err != nil {
                 return fmt.Errorf("Failed to encode to JSON: %s", err)
         }
         if _, err := f.Write(d); err != nil {
                 return fmt.Errorf("Failed to write to file: %v", err)
         }
         return nil
 }
 
+func (fs *FileDiffStore) removeDiffMetricsFromFileCache(digests []string) error {
+        fs.diffDirLock.Lock()
+        defer fs.diffDirLock.Unlock()
+
+        // Walk the entire cache and remove all files are contained in the list of digests.
+        return filepath.Walk(fs.localDiffMetricsDir, func(path string, info os.FileInfo, err error) error {
+                if !info.IsDir() {
+                        for _, d := range digests {
+                                if (len(d) > 0) && strings.Contains(info.Name(), d) {
+                                        if err := os.Remove(path); err != nil {
+                                                return err
+                                        }
+                                }
+                        }
+                }
+                return nil
+        })
+}
+
 // Returns the file basename to use for the specified digests.
 // Eg: Returns 111-222 since 111 < 222 when 111 and 222 are specified as inputs
 // regardless of the order.
 func getDiffBasename(d1, d2 string) string {
         if d1 < d2 {
                 return fmt.Sprintf("%s-%s", d1, d2)
         }
         return fmt.Sprintf("%s-%s", d2, d1)
 }
 
@@ skipping 151 matching lines @@
                 glog.Errorf("Error fetching file for digest %s: %s", d, err)
         }
 
         if err != nil {
                 glog.Errorf("Failed fetching file after %d attempts", MAX_URI_GET_TRIES)
                 downloadFailureCount.Inc(1)
         }
         return err
 }
 
+func (fs *FileDiffStore) removeImageFromGS(d string) error {
+        storage, err := storage.New(fs.client)
+        if err != nil {
+                return fmt.Errorf("Failed to create interface to Google Storage: %s\n", err)
+        }
+
+        objLocation := filepath.Join(fs.storageBaseDir, fmt.Sprintf("%s.%s", d, IMG_EXTENSION))
+        if err := storage.Objects.Delete(fs.gsBucketName, objLocation).Do(); err != nil {
+                return fmt.Errorf("Unable to delete %s/%s:  %s", fs.gsBucketName, objLocation, err)
+        }
+        return nil
+}
+
+func (fs *FileDiffStore) removeImageFromCache(d string) error {
+        fs.digestDirLock.Lock()
+        defer fs.digestDirLock.Unlock()
+        path := fs.getDigestImagePath(d)
+        if _, err := os.Stat(path); os.IsNotExist(err) {
+                return nil
+        }
+        return os.Remove(path)
+}
+
 // Returns the response body of the specified GS object. Tries MAX_URI_GET_TRIES
 // times if download is unsuccessful. Client must close the response body when
 // finished with it.
 func (fs *FileDiffStore) getRespBody(res *storage.Object) (io.ReadCloser, error) {
         request, err := gs.RequestForStorageURL(res.MediaLink)
         if err != nil {
                 return nil, fmt.Errorf("Unable to create Storage MediaURI request: %s\n", err)
         }
 
         resp, err := fs.client.Do(request)
@@ skipping 109 matching lines @@
 
 func (fs *FileDiffStore) createRadixPath(baseDir, fileName string) (string, error) {
         targetPath := fileutil.TwoLevelRadixPath(baseDir, fileName)
         radixDir, _ := filepath.Split(targetPath)
         if err := os.MkdirAll(radixDir, 0700); err != nil {
                 return "", err
         }
 
         return targetPath, nil
 }