[logdog] Replace Tumble with push queues.

logdog/appengine/coordinator/service.go

 // Copyright 2015 The LUCI Authors.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
 //      http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ skipping 17 matching lines @@
         "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/luci_config/common/cfgtypes"
         "github.com/luci/luci-go/server/auth"
         "github.com/luci/luci-go/server/router"
 
-        gcps "cloud.google.com/go/pubsub"
+        vkit "cloud.google.com/go/pubsub/apiv1"
         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"
+
+        "golang.org/x/net/context"
 )
 
 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.
         //
         // AppEngine's "urlfetch" has a limit of 32MB:
@@ skipping 63 matching lines @@
 
         // gcfg is the cached global configuration.
         gcfg           *config.Config
         projectConfigs map[cfgtypes.ProjectName]*cachedProjectConfig
 
         // archivalIndex is the atomically-manipulated archival index for the
         // ArchivalPublisher. This is shared between all ArchivalPublisher instances
         // from this service.
         archivalIndex int32
 
-        // pubSubClients is a map of Pub/Sub client singletons generated during this
-        // request. Each client is associated with its project, and will be
-        // initialized the first time it is requested by getPubSubClient.
+        // pubSubClient is a Pub/Sub client generated during this request.
         //
-        // All clients will be closed on "close".
-        pubSubClients map[string]*gcps.Client
-
+        // It will be closed on "close".
+        pubSubClient *vkit.PublisherClient
         // signer is the signer instance to use.
         signer gaesigner.Signer
 }
 
 func (s *prodServicesInst) close(c context.Context) {
         s.Lock()
         defer s.Unlock()
 
-        for proj, client := range s.pubSubClients {
+        if client := s.pubSubClient; client != nil {
                 if err := client.Close(); err != nil {
-                        log.Fields{
-                                log.ErrorKey: err,
-                                "project":    proj,
-                        }.Errorf(c, "Failed to close Pub/Sub client singleton.")
+                        log.WithError(err).Errorf(c, "Failed to close Pub/Sub client singleton.")
                 }
+                s.pubSubClient = nil
         }
 }
-
 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)
                 if err != nil {
                         return nil, err
@@ skipping 193 matching lines @@
         }
 
         fullTopic := pubsub.Topic(cfg.Coordinator.ArchiveTopic)
         if err := fullTopic.Validate(); err != nil {
                 log.Fields{
                         log.ErrorKey: err,
                         "topic":      fullTopic,
                 }.Errorf(c, "Failed to validate archival topic.")
                 return nil, errors.New("invalid archival topic")
         }
-        project, topic := fullTopic.Split()
 
-        // Get a Pub/Sub client (maybe re-use).
-        psClient, err := s.getPubSubClient(project)
+        psClient, err := s.getPubSubClient()
         if err != nil {
-                log.WithError(err).Errorf(c, "Failed to get Pub/Sub client.")
                 return nil, err
         }
 
         // Create a Topic, and configure it to not bundle messages.
-        psTopic := psClient.Topic(topic)
-        pubsub.DisableTopicBundling(psTopic)
-
         return &pubsubArchivalPublisher{
-                client:           psClient,
-                topic:            psTopic,
+                publisher: &pubsub.UnbufferedPublisher{
+                        Topic:  fullTopic,
+                        Client: psClient,
+                },
                 publishIndexFunc: s.nextArchiveIndex,
         }, nil
 }
 
-func (s *prodServicesInst) getPubSubClient(proj string) (*gcps.Client, error) {
+func (s *prodServicesInst) getPubSubClient() (*vkit.PublisherClient, error) {
+        s.Lock()
+        defer s.Unlock()
+
+        if s.pubSubClient != nil {
+                return s.pubSubClient, nil
+        }
+
         // 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.
         c := metadata.NewContext(s.aeCtx, nil)
 
-        s.Lock()
-        defer s.Unlock()
-
-        client := s.pubSubClients[proj]
-        if client != nil {
-                return client, nil
-        }
-
-        // Create an authenticated Pub/Sub client.
+        // Create an authenticated unbuffered Pub/Sub Publisher.
         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")
         }
 
-        psClient, err := gcps.NewClient(c, proj,
+        psClient, err := vkit.NewPublisherClient(c,
                 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")
         }
 
-        // Retain this client for the duration of our request.
-        if s.pubSubClients == nil {
-                s.pubSubClients = make(map[string]*gcps.Client)
-        }
-        s.pubSubClients[proj] = psClient
+        s.pubSubClient = psClient
         return psClient, nil
 }
 
 func (s *prodServicesInst) nextArchiveIndex() uint64 {
         // We use a 32-bit value for this because it avoids atomic memory bounary
         // issues. Furthermore, we constrain it to be positive, using a negative
         // value as a sentinel that the archival index has wrapped.
         //
         // This is reasonable, as it is very unlikely that a single request will issue
         // more than MaxInt32 archival tasks.
@@ skipping 134 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
 }