LogDog: Add Storage-layer data caching.

logdog/common/storage/archive/storage.go

 // Copyright 2015 The LUCI Authors. All rights reserved.
 // Use of this source code is governed under the Apache License, Version 2.0
 // that can be found in the LICENSE file.
 
 // Package archive implements a storage.Storage instance that retrieves logs
 // from a Google Storage archive.
 //
 // This is a special implementation of storage.Storage, and does not fully
 // conform to the API expecations. Namely:
 //      - It is read-only. Mutation methods will return storage.ErrReadOnly.
 //      - Storage methods ignore the supplied Path argument, instead opting for
 //        the archive configured in its Options.
 package archive
 
 import (
         "bytes"
         "fmt"
         "io"
         "io/ioutil"
         "sort"
-        "sync"
+        "sync/atomic"
 
         "github.com/luci/luci-go/common/config"
         "github.com/luci/luci-go/common/data/recordio"
         "github.com/luci/luci-go/common/errors"
         "github.com/luci/luci-go/common/gcloud/gs"
         "github.com/luci/luci-go/common/iotools"
         log "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/logdog/api/logpb"
         "github.com/luci/luci-go/logdog/common/storage"
+        "github.com/luci/luci-go/logdog/common/storage/caching"
         "github.com/luci/luci-go/logdog/common/types"
 
         cloudStorage "cloud.google.com/go/storage"
         "github.com/golang/protobuf/proto"
         "golang.org/x/net/context"
 )
 
 const (
         // maxStreamRecordSize is the maximum record size we're willing to read from
         // our archived log stream. This will help prevent out-of-memory errors if the
@@ skipping 11 matching lines @@
 type Options struct {
         // IndexURL is the Google Storage URL for the stream's index.
         IndexURL string
         // StreamURL is the Google Storage URL for the stream's entries.
         StreamURL string
 
         // Client is the HTTP client to use for authentication.
         //
         // Closing this Storage instance does not close the underlying Client.
         Client gs.Client
+
+        // Cache, if not nil, will be used to cache data.
+        Cache caching.Cache
 }
 
 type storageImpl struct {
         *Options
         context.Context
 
         streamPath gs.Path
         indexPath  gs.Path
 
-        indexMu sync.Mutex
-        index   *logpb.LogIndex
+        index atomic.Value
 }
 
 // New instantiates a new Storage instance, bound to the supplied Options.
 func New(ctx context.Context, o Options) (storage.Storage, error) {
         s := storageImpl{
                 Options: &o,
                 Context: ctx,
 
                 streamPath: gs.Path(o.StreamURL),
                 indexPath:  gs.Path(o.IndexURL),
@@ skipping 194 matching lines @@
         case lastEntry == nil:
                 return nil, storage.ErrDoesNotExist
 
         default:
                 return lastEntry, nil
         }
 }
 
 // getIndex returns the cached log stream index, fetching it if necessary.
 func (s *storageImpl) getIndex() (*logpb.LogIndex, error) {
-        s.indexMu.Lock()
-        defer s.indexMu.Unlock()
+        idx := s.index.Load()
+        if idx != nil {
+                return idx.(*logpb.LogIndex), nil
+        }
 
-        if s.index == nil {
-                index, err := loadIndex(s, s.Client, s.indexPath)
-                switch errors.Unwrap(err) {
-                case nil:
-                        break
+        index, err := loadIndex(s, s.Client, s.indexPath, s.Cache)
+        switch errors.Unwrap(err) {
+        case nil:
+                break
 
-                case cloudStorage.ErrBucketNotExist, cloudStorage.ErrObjectNotExist:
-                        // Treat a missing index the same as an empty index.
-                        log.WithError(err).Warningf(s, "Index is invalid, using empty index.")
-                        index = &logpb.LogIndex{}
+        case cloudStorage.ErrBucketNotExist, cloudStorage.ErrObjectNotExist:
+                // Treat a missing index the same as an empty index.
+                log.WithError(err).Warningf(s, "Index is invalid, using empty index.")
+                index = &logpb.LogIndex{}
 
-                default:
-                        return nil, err
-                }
+        default:
+                return nil, err
+        }
 
-                s.index = index
-        }
-        return s.index, nil
+        s.index.Store(index)
+        return index, nil
 }
 
-func loadIndex(c context.Context, client gs.Client, path gs.Path) (*logpb.LogIndex, error) {
+func loadIndex(c context.Context, client gs.Client, path gs.Path, cache caching.Cache) (*logpb.LogIndex, error) {
         // If there is no path, then return an empty index.
         if path == "" {
                 log.Infof(c, "No index path, using empty index.")
                 return &logpb.LogIndex{}, nil
         }
 
-        r, err := client.NewReader(path, 0, -1)
-        if err != nil {
-                log.WithError(err).Errorf(c, "Failed to create index Reader.")
-                return nil, errors.Annotate(err).Reason("failed to create index Reader").Err()
+        // If we have a cache, see if the index is cached.
+        var (
+                indexData []byte
+                cached    bool
+        )
+        if cache != nil {
+                indexData = getCachedLogIndexData(c, cache, path)
+                if indexData != nil {
+                        cached = true
+                }
         }
-        defer func() {
-                if err := r.Close(); err != nil {
-                        log.WithError(err).Warningf(c, "Error closing index Reader.")
+
+        if indexData == nil {
+                // No cache, or no cached entry. Load from storage.
+                r, err := client.NewReader(path, 0, -1)
+                if err != nil {
+                        log.WithError(err).Errorf(c, "Failed to create index Reader.")
+                        return nil, errors.Annotate(err).Reason("failed to create index Reader").Err()
                 }
-        }()
-        indexData, err := ioutil.ReadAll(r)
-        if err != nil {
-                log.WithError(err).Errorf(c, "Failed to read index.")
-                return nil, errors.Annotate(err).Reason("failed to read index").Err()
+                defer func() {
+                        if err := r.Close(); err != nil {
+                                log.WithError(err).Warningf(c, "Error closing index Reader.")
+                        }
+                }()
+
+                if indexData, err = ioutil.ReadAll(r); err != nil {
+                        log.WithError(err).Errorf(c, "Failed to read index.")
+                        return nil, errors.Annotate(err).Reason("failed to read index").Err()
+                }
         }
 
         index := logpb.LogIndex{}
         if err := proto.Unmarshal(indexData, &index); err != nil {
                 log.WithError(err).Errorf(c, "Failed to unmarshal index.")
                 return nil, errors.Annotate(err).Reason("failed to unmarshal index").Err()
         }
 
+        // If the index is valid, but wasn't cached previously, then cache it.
+        if cache != nil && !cached {
+                putCachedLogIndexData(c, cache, path, indexData)
+        }
+
         return &index, nil
 }
 
 type getStrategy struct {
         // startIndex is desired initial log entry index.
         startIndex types.MessageIndex
 
         // startOffset is the beginning byte offset of the log entry stream. This may
         // be lower than the offset of the starting record if the index is sparse.
         startOffset uint64
@@ skipping 102 matching lines @@
         ui := uint64(i)
         s := sort.Search(len(entries), func(i int) bool {
                 return entries[i].StreamIndex > ui
         })
 
         // The returned index is the one immediately after the index that we want. If
         // our search returned 0, the first index entry is > our search entry, and we
         // will return nil.
         return s - 1
 }