LogDog: Update to archival V2.

server/logdog/storage/archive/storage.go

 // Copyright 2015 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license 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"
-        "errors"
+        "fmt"
         "io"
         "io/ioutil"
         "sort"
-        "strings"
         "sync"
 
         "golang.org/x/net/context"
 
         "github.com/golang/protobuf/proto"
         "github.com/luci/luci-go/common/gcloud/gs"
         "github.com/luci/luci-go/common/iotools"
         "github.com/luci/luci-go/common/logdog/types"
         log "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/common/proto/logdog/logpb"
@@ skipping 22 matching lines @@
         // Client is the HTTP client to use for authentication.
         //
         // Closing this Storage instance does not close the underlying Client.
         Client gs.Client
 }
 
 type storageImpl struct {
         *Options
         context.Context
 
-        streamBucket string
-        streamPath   string
-        indexBucket  string
-        indexPath    string
+        streamPath gs.Path

[dnj, 2016/04/11 17:20:05]

This was just updated to use "gs.Path" instead of storing separate bucket/path.

+        indexPath  gs.Path
 
         indexMu     sync.Mutex
         index       *logpb.LogIndex
         closeClient bool
 }
 
 // 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),
         }
 
-        s.indexBucket, s.indexPath = splitGSURL(o.IndexURL)
-        if s.indexBucket == "" || s.indexPath == "" {
-                return nil, errors.New("invalid index URL")
+        if !s.streamPath.IsFullPath() {
+                return nil, fmt.Errorf("invalid stream URL: %q", s.streamPath)
+        }
+        if !s.indexPath.IsFullPath() {
+                return nil, fmt.Errorf("invalid index URL: %v", s.indexPath)
         }
 
-        s.streamBucket, s.streamPath = splitGSURL(o.StreamURL)
-        if s.streamBucket == "" || s.streamPath == "" {
-                return nil, errors.New("invalid stream URL")
-        }
         return &s, nil
 }
 
 func (s *storageImpl) Close() {
         if s.closeClient {
                 _ = s.Client.Close()
         }
 }
 
 func (s *storageImpl) Config(storage.Config) error  { return storage.ErrReadOnly }
 func (s *storageImpl) Put(storage.PutRequest) error { return storage.ErrReadOnly }
 
 func (s *storageImpl) Get(req storage.GetRequest, cb storage.GetCallback) error {
         idx, err := s.getIndex()
         if err != nil {
                 return err
         }
 
         // Identify the byte offsets that we want to fetch from the entries stream.
         st := s.buildGetStrategy(&req, idx)
         if st.lastIndex == -1 || req.Index > st.lastIndex {
                 // No records to read.
                 return nil
         }
 
-        r, err := s.Client.NewReader(s.streamBucket, s.streamPath, gs.Options{
+        r, err := s.Client.NewReader(s.streamPath, gs.Options{
                 From: st.from,
                 To:   st.to,
         })
         if err != nil {
                 log.WithError(err).Errorf(s, "Failed to create stream Reader.")
                 return err
         }
         defer func() {
                 if err := r.Close(); err != nil {
                         log.WithError(err).Warningf(s, "Error closing stream Reader.")
@@ skipping 74 matching lines @@
                 return nil, 0, err
         }
 
         // Get the offset of the last record.
         if len(idx.Entries) == 0 {
                 return nil, 0, nil
         }
         lle := idx.Entries[len(idx.Entries)-1]
 
         // Get a Reader for the Tail entry.
-        r, err := s.Client.NewReader(s.streamBucket, s.streamPath, gs.Options{
+        r, err := s.Client.NewReader(s.streamPath, gs.Options{
                 From: int64(lle.Offset),
         })
         if err != nil {
                 log.Fields{
                         log.ErrorKey: err,
                         "offset":     lle.Offset,
                 }.Errorf(s, "Failed to create reader.")
                 return nil, 0, err
         }
         defer func() {
@@ skipping 11 matching lines @@
 
         return d, types.MessageIndex(lle.StreamIndex), 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()
 
         if s.index == nil {
-                r, err := s.Client.NewReader(s.indexBucket, s.indexPath, gs.Options{})
+                r, err := s.Client.NewReader(s.indexPath, gs.Options{})
                 if err != nil {
                         log.WithError(err).Errorf(s, "Failed to create index Reader.")
                         return nil, err
                 }
                 defer func() {
                         if err := r.Close(); err != nil {
                                 log.WithError(err).Warningf(s, "Error closing index Reader.")
                         }
                 }()
                 indexData, err := ioutil.ReadAll(r)
                 if err != nil {
                         log.WithError(err).Errorf(s, "Failed to read index.")
                         return nil, err
                 }
 
                 index := logpb.LogIndex{}
                 if err := proto.Unmarshal(indexData, &index); err != nil {
                         log.WithError(err).Errorf(s, "Failed to unmarshal index.")
                         return nil, err
                 }
 
                 s.index = &index
         }
         return s.index, nil
 }
 
-func splitGSURL(u string) (string, string) {
-        parts := strings.SplitN(strings.TrimPrefix(u, "gs://"), "/", 2)
-        if len(parts) == 1 {
-                return parts[0], ""
-        }
-        return parts[0], parts[1]
-}
-
 type getStrategy struct {
         // from is the beginning byte offset of the log entry stream.
         from int64
         // to is the ending byte offset of the log entry stream.
         to int64
 
         // lastIndex is the last log entry index in the stream. This will be -1 if
         // there are no entries in the stream.
         lastIndex types.MessageIndex
 }
@@ skipping 40 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
 }