Coordinator: AppEngine context for Pub/Sub.

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/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/appengine"
         "google.golang.org/grpc"
         "google.golang.org/grpc/metadata"
 )
 
 const (
         // maxSignedURLLifetime is the maximum allowed signed URL lifetime.
         maxSignedURLLifetime = 1 * time.Hour
 
         // maxGSFetchSize is the maximum amount of data we can fetch from a single
         // Google Storage RPC call.
@@ skipping 39 matching lines @@
         StorageForStream(context.Context, *LogStreamState) (Storage, error)
 
         // ArchivalPublisher returns an ArchivalPublisher instance.
         ArchivalPublisher(context.Context) (ArchivalPublisher, error)
 }
 
 // ProdCoordinatorService is Middleware used by Coordinator services.
 //
 // It installs a production Services instance into the Context.
 func ProdCoordinatorService(c *router.Context, next router.Handler) {
-        c.Context = WithServices(c.Context, &prodServicesInst{})
+        c.Context = WithServices(c.Context, &prodServicesInst{
+                req: c.Request,
+        })
         next(c)
 }
 
 // prodServicesInst is a Service exposing production faciliites. A unique
 // instance is bound to each each request.
 type prodServicesInst struct {
         sync.Mutex
 
+        // req is the request that is being serviced. This is populated when the
+        // service is installed via ProdCoordinatorService.
+        req *http.Request
+
         // 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.
@@ skipping 214 matching lines @@
                 return nil, errors.New("invalid archival topic")
         }
         project, topic := fullTopic.Split()
 
         // Create an authenticated Pub/Sub client.
         creds, err := auth.GetPerRPCCredentials(auth.AsSelf, auth.WithScopes(pubsub.PublisherScopes...))
         if err != nil {
                 log.WithError(err).Errorf(c, "Failed to create Pub/Sub credentials.")
                 return nil, errors.New("failed to create Pub/Sub credentials")
         }
-        // Don't pass gRPC metadata to PubSub.
-        psClient, err := gcps.NewClient(
-                metadata.NewContext(c, nil), project,
+
+        // Create a new AppEngine context. Don't pass gRPC metadata to PubSub, since
+        // we don't want any caller RPC to be forwarded to the backend service.
+        aeCtx := appengine.WithContext(metadata.NewContext(c, nil), s.req)
+
+        psClient, err := gcps.NewClient(aeCtx, project,
                 option.WithGRPCDialOption(grpc.WithPerRPCCredentials(creds)))
         if err != nil {
                 log.WithError(err).Errorf(c, "Failed to create Pub/Sub client.")
                 return nil, errors.New("failed to create Pub/Sub client")
         }
 
         return &pubsubArchivalPublisher{
                 client:           psClient,
                 topic:            psClient.Topic(topic),
                 publishIndexFunc: s.nextArchiveIndex,
@@ skipping 144 matching lines @@
 
         // 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
 }