LogDog: Add signed GS URL fetching.

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.
@@ skipping 33 matching lines @@
         // 16MB is larger than the maximum log entry size
         maxStreamRecordSize = 2 * types.MaxLogEntryDataSize
 )
 
 // Options is the set of configuration options for this Storage instance.
 //
 // Unlike other Storage instances, this is bound to a single archived stream.
 // Project and Path parameters in requests will be ignored in favor of the
 // Google Storage URLs.
 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
+        // Index is the Google Storage URL for the stream's index.
+        Index gs.Path
+        // Stream is the Google Storage URL for the stream's entries.
+        Stream gs.Path
 
         // 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
-
         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),
         }
 
-        if !s.streamPath.IsFullPath() {
-                return nil, fmt.Errorf("invalid stream URL: %q", s.streamPath)
+        if !s.Stream.IsFullPath() {
+                return nil, fmt.Errorf("invalid stream URL: %q", s.Stream)
         }
-        if s.indexPath != "" && !s.indexPath.IsFullPath() {
-                return nil, fmt.Errorf("invalid index URL: %v", s.indexPath)
+        if s.Index != "" && !s.Index.IsFullPath() {
+                return nil, fmt.Errorf("invalid index URL: %v", s.Index)
         }
 
         return &s, nil
 }
 
 func (s *storageImpl) Close() {}
 
 func (s *storageImpl) Config(storage.Config) error  { return storage.ErrReadOnly }
 func (s *storageImpl) Put(storage.PutRequest) error { return storage.ErrReadOnly }
 
@@ skipping 29 matching lines @@
         // apply that limit too.
         // Get an archive reader.
         var (
                 offset = st.startOffset
                 length = st.length()
         )
 
         log.Fields{
                 "offset": offset,
                 "length": length,
-                "path":   s.streamPath,
+                "path":   s.Stream,
         }.Debugf(s, "Creating stream reader for range.")
-        storageReader, err := s.Client.NewReader(s.streamPath, int64(offset), length)
+        storageReader, err := s.Client.NewReader(s.Stream, int64(offset), length)
         if err != nil {
                 log.WithError(err).Errorf(s, "Failed to create stream Reader.")
                 return errors.Annotate(err).Reason("failed to create stream Reader").Err()
         }
         defer func() {
                 if tmpErr := storageReader.Close(); tmpErr != nil {
                         // (Non-fatal)
                         log.WithError(tmpErr).Warningf(s, "Error closing stream Reader.")
                 }
         }()
@@ skipping 130 matching lines @@
         }
 }
 
 // getIndex returns the cached log stream index, fetching it if necessary.
 func (s *storageImpl) getIndex() (*logpb.LogIndex, error) {
         idx := s.index.Load()
         if idx != nil {
                 return idx.(*logpb.LogIndex), nil
         }
 
-        index, err := loadIndex(s, s.Client, s.indexPath, s.Cache)
+        index, err := loadIndex(s, s.Client, s.Index, 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{}
 
         default:
@@ skipping 168 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
 }