LogDog: Add signed GS URL fetching.

logdog/appengine/coordinator/service.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 coordinator
 
 import (
         "net/http"
         "sync"
         "sync/atomic"
+        "time"
 
+        "github.com/luci/luci-go/appengine/gaeauth/server/gaesigner"
         "github.com/luci/luci-go/appengine/gaemiddleware"
+        "github.com/luci/luci-go/common/clock"
         luciConfig "github.com/luci/luci-go/common/config"
         "github.com/luci/luci-go/common/errors"
         "github.com/luci/luci-go/common/gcloud/gs"
         "github.com/luci/luci-go/common/gcloud/pubsub"
         log "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/logdog/api/config/svcconfig"
         "github.com/luci/luci-go/logdog/appengine/coordinator/config"
         "github.com/luci/luci-go/logdog/common/storage"
+        "github.com/luci/luci-go/logdog/common/storage/archive"
         "github.com/luci/luci-go/logdog/common/storage/bigtable"
         "github.com/luci/luci-go/logdog/common/storage/caching"
         "github.com/luci/luci-go/server/auth"
         "github.com/luci/luci-go/server/router"
 
         gcps "cloud.google.com/go/pubsub"
+        gcst "cloud.google.com/go/storage"
         "golang.org/x/net/context"
         "google.golang.org/api/option"
         "google.golang.org/grpc"
         "google.golang.org/grpc/metadata"
 )
 
+// maxSignedURLLifetime is the maximum allowed signed URL lifetime.
+const maxSignedURLLifetime = 1 * time.Hour
+
 // Services is a set of support services used by Coordinator.
 //
 // Each Services instance is valid for a singel request, but can be re-used
 // throughout that request. This is advised, as the Services instance may
 // optionally cache values.
 //
 // Services methods are goroutine-safe.
 //
 // By default, a production set of services will be used. However, this can be
 // overridden for testing to mock the service layer.
 type Services interface {
         // Config returns the current instance and application configuration
         // instances.
         //
         // The production instance will cache the results for the duration of the
         // request.
         Config(context.Context) (*config.Config, error)
 
         // ProjectConfig returns the project configuration for the named project.
         //
         // The production instance will cache the results for the duration of the
         // request.
         //
         // Returns the same error codes as config.ProjectConfig.
         ProjectConfig(context.Context, luciConfig.ProjectName) (*svcconfig.ProjectConfig, error)
 
-        // Storage returns an intermediate storage instance for use by this service.
+        // Storage returns a Storage instance for the supplied log stream.
         //
         // The caller must close the returned instance if successful.
-        IntermediateStorage(context.Context) (storage.Storage, error)
-
-        // GSClient instantiates a Google Storage client.
-        GSClient(context.Context) (gs.Client, error)
+        StorageForStream(context.Context, *LogStreamState) (Storage, error)
 
         // ArchivalPublisher returns an ArchivalPublisher instance.
         ArchivalPublisher(context.Context) (ArchivalPublisher, error)
-
-        // StorageCache returns the storage cache instance to use, or nil for no
-        // caching.
-        StorageCache() caching.Cache
 }
 
 // ProdServices is middleware chain used by Coordinator services.
 //
 // It sets up basic GAE functionality as well as installs a production Services
 // instance.
 func ProdServices() router.MiddlewareChain {
         return gaemiddleware.BaseProd().Extend(func(c *router.Context, next router.Handler) {
                 c.Context = WithServices(c.Context, &prodServicesInst{})
                 next(c)
         })
 }
 
 // prodServicesInst is a Service exposing production faciliites. A unique
 // instance is bound to each each request.
 type prodServicesInst struct {
         sync.Mutex
 
         // gcfg is the cached global configuration.
         gcfg           *config.Config
         projectConfigs map[luciConfig.ProjectName]*cachedProjectConfig
 
         // archivalIndex is the atomically-manipulated archival index for the
         // ArchivalPublisher. This is shared between all ArchivalPublisher instances
         // from this service.
         archivalIndex int32
+
+        // signer is the signer instance to use.
+        signer gaesigner.Signer
 }
 
 func (s *prodServicesInst) Config(c context.Context) (*config.Config, error) {
         s.Lock()
         defer s.Unlock()
 
         // Load/cache the global config.
         if s.gcfg == nil {
                 var err error
                 s.gcfg, err = config.Load(c)
@@ skipping 42 matching lines @@
                 }
                 s.projectConfigs[project] = cp
         }
         return cp
 }
 
 func (s *prodServicesInst) ProjectConfig(c context.Context, project luciConfig.ProjectName) (*svcconfig.ProjectConfig, error) {
         return s.getOrCreateCachedProjectConfig(project).resolve(c)
 }
 
-func (s *prodServicesInst) IntermediateStorage(c context.Context) (storage.Storage, error) {
+func (s *prodServicesInst) StorageForStream(c context.Context, lst *LogStreamState) (Storage, error) {
+        if !lst.ArchivalState().Archived() {
+                log.Debugf(c, "Log is not archived. Fetching from intermediate storage.")
+                return s.newBigTableStorage(c)
+        }
+
+        log.Fields{
+                "indexURL":    lst.ArchiveIndexURL,
+                "streamURL":   lst.ArchiveStreamURL,
+                "archiveTime": lst.ArchivedTime,
+        }.Debugf(c, "Log is archived. Fetching from archive storage.")
+        return s.newGoogleStorage(c, gs.Path(lst.ArchiveIndexURL), gs.Path(lst.ArchiveStreamURL))
+}
+
+func (s *prodServicesInst) newBigTableStorage(c context.Context) (Storage, error) {
         cfg, err := s.Config(c)
         if err != nil {
                 return nil, err
         }
 
         // Is BigTable configured?
         if cfg.Storage == nil {
                 return nil, errors.New("no storage configuration")
         }
 
@@ skipping 34 matching lines @@
         // calls.
         c = metadata.NewContext(c, nil)
 
         st, err := bigtable.New(c, bigtable.Options{
                 Project:  bt.Project,
                 Instance: bt.Instance,
                 LogTable: bt.LogTableName,
                 ClientOptions: []option.ClientOption{
                         option.WithGRPCDialOption(grpc.WithPerRPCCredentials(creds)),
                 },
-                Cache: s.StorageCache(),
+                Cache: s.getStorageCache(),
         })
         if err != nil {
                 log.WithError(err).Errorf(c, "Failed to create BigTable instance.")
                 return nil, err
         }
-        return st, nil
+
+        return &bigTableStorage{
+                Storage: st,
+        }, nil
 }
 
-func (s *prodServicesInst) GSClient(c context.Context) (gs.Client, error) {
+func (s *prodServicesInst) newGoogleStorage(c context.Context, index, stream gs.Path) (Storage, error) {
+        gs, err := s.newGSClient(c)
+        if err != nil {
+                log.WithError(err).Errorf(c, "Failed to create Google Storage client.")
+                return nil, err
+        }
+        defer func() {
+                if gs != nil {
+                        if err := gs.Close(); err != nil {
+                                log.WithError(err).Warningf(c, "Failed to close Google Storage client.")
+                        }
+                }
+        }()
+
+        st, err := archive.New(c, archive.Options{
+                Index:  index,
+                Stream: stream,
+                Client: gs,
+                Cache:  s.getStorageCache(),
+        })
+        if err != nil {
+                log.WithError(err).Errorf(c, "Failed to create Google Storage storage instance.")
+                return nil, err
+        }
+
+        gs = nil // Don't close in defer.
+        return &googleStorage{
+                Storage: st,
+                svc:     s,
+                gs:      gs,
+                stream:  stream,
+                index:   index,
+        }, nil
+}
+
+func (s *prodServicesInst) newGSClient(c context.Context) (gs.Client, error) {
         // Get an Authenticator bound to the token scopes that we need for
         // authenticated Cloud Storage access.
         transport, err := auth.GetRPCTransport(c, auth.AsSelf, auth.WithScopes(gs.ReadOnlyScopes...))
         if err != nil {
                 log.WithError(err).Errorf(c, "Failed to create Cloud Storage transport.")
                 return nil, errors.New("failed to create Cloud Storage transport")
         }
         return gs.NewProdClient(c, transport)
 }
 
@@ skipping 41 matching lines @@
         // more than MaxInt32 archival tasks.
         v := atomic.AddInt32(&s.archivalIndex, 1) - 1
         if v < 0 {
                 panic("archival index has wrapped")
         }
         return uint64(v)
 }
 
 var storageCacheSingleton StorageCache
 
-func (s *prodServicesInst) StorageCache() caching.Cache { return &storageCacheSingleton }
+func (s *prodServicesInst) getStorageCache() caching.Cache { return &storageCacheSingleton }
+
+// Storage is an interface to storage used by the Coordinator.
+type Storage interface {
+        // Storage is the base Storage instance.
+        storage.Storage
+
+        // GetSignedURLs attempts to sign the storage's stream's RecordIO archive
+        // stream storage URL.
+        //
+        // If signing is not supported by this Storage instance, this will return
+        // a nil signing response and no error.
+        GetSignedURLs(context.Context, *URLSigningRequest) (*URLSigningResponse, error)
+}
+
+// URLSigningRequest is the set of URL signing parameters passed to a
+// Storage.GetSignedURLs call.
+type URLSigningRequest struct {
+        // Expriation is the signed URL expiration time.
+        Lifetime time.Duration
+
+        // Stream, if true, requests a signed log stream URL.
+        Stream bool
+        // Index, if true, requests a signed log stream index URL.
+        Index bool
+}
+
+// HasWork returns true if this signing request actually has work that is
+// requested.
+func (r *URLSigningRequest) HasWork() bool {
+        return (r.Stream || r.Index) && (r.Lifetime > 0)
+}
+
+// URLSigningResponse is the resulting signed URLs from a Storage.GetSignedURLs
+// call.
+type URLSigningResponse struct {
+        // Expriation is the signed URL expiration time.
+        Expiration time.Time
+
+        // Stream is the signed URL for the log stream, if requested.
+        Stream string
+        // Index is the signed URL for the log stream index, if requested.
+        Index string
+}
+
+// intermediateStorage is a Storage instance bound to BigTable.
+type bigTableStorage struct {
+        // Storage is the base storage.Storage instance.
+        storage.Storage
+}
+
+func (*bigTableStorage) GetSignedURLs(context.Context, *URLSigningRequest) (*URLSigningResponse, error) {
+        return nil, nil
+}
+
+type googleStorage struct {
+        // Storage is the base storage.Storage instance.
+        storage.Storage
+        // svc is the services instance that created this.
+        svc *prodServicesInst
+
+        // ctx is the Context that was bound at the time of of creation.
+        ctx context.Context
+        // gs is the backing Google Storage client.
+        gs gs.Client
+
+        // stream is the stream's Google Storage URL.
+        stream gs.Path
+        // index is the index's Google Storage URL.
+        index gs.Path
+
+        gsSigningOpts func(context.Context) (*gcst.SignedURLOptions, error)
+}
+
+func (si *googleStorage) Close() {
+        if err := si.gs.Close(); err != nil {
+                log.WithError(err).Warningf(si.ctx, "Failed to close Google Storage client.")
+        }
+        si.Storage.Close()
+}
+
+func (si *googleStorage) GetSignedURLs(c context.Context, req *URLSigningRequest) (*URLSigningResponse, error) {
+        info, err := si.svc.signer.ServiceInfo(c)
+        if err != nil {
+                return nil, errors.Annotate(err).InternalReason("failed to get service info").Err()
+        }
+
+        lifetime := req.Lifetime
+        switch {
+        case lifetime < 0:
+                return nil, errors.Reason("invalid signed URL lifetime: %(lifetime)s").D("lifetime", lifetime).Err()
+
+        case lifetime > maxSignedURLLifetime:
+                lifetime = maxSignedURLLifetime
+        }
+
+        // Get our signing options.
+        resp := URLSigningResponse{
+                Expiration: clock.Now(c).Add(lifetime),
+        }
+        opts := gcst.SignedURLOptions{
+                GoogleAccessID: info.ServiceAccountName,
+                SignBytes: func(b []byte) ([]byte, error) {
+                        _, signedBytes, err := si.svc.signer.SignBytes(c, b)
+                        return signedBytes, err
+                },
+                Method:  "GET",
+                Expires: resp.Expiration,
+        }
+
+        doSign := func(path gs.Path) (string, error) {
+                url, err := gcst.SignedURL(path.Bucket(), path.Filename(), &opts)
+                if err != nil {
+                        return "", errors.Annotate(err).InternalReason("failed to sign URL").
+                                D("bucket", path.Bucket()).D("filename", path.Filename).Err()
+                }
+                return url, nil
+        }
+
+        // Sign stream URL.
+        if req.Stream {
+                if resp.Stream, err = doSign(si.stream); err != nil {
+                        return nil, errors.Annotate(err).InternalReason("failed to sign stream URL").Err()
+                }
+        }
+
+        // Sign index URL.
+        if req.Index {
+                if resp.Index, err = doSign(si.index); err != nil {
+                        return nil, errors.Annotate(err).InternalReason("failed to sign index URL").Err()
+                }
+        }
+
+        return &resp, nil
+}