Milo: Move buildbucket pubsub sub from buildbucket project to milo project

milo/common/pubsub.go

 package common
 
 import (
         "encoding/base64"
         "errors"
         "fmt"
         "net/url"
         "strings"
+        "sync"
         "time"
 
         "cloud.google.com/go/pubsub"
         "golang.org/x/net/context"
 
         "github.com/luci/gae/service/info"
+        luciErrors "github.com/luci/luci-go/common/errors"
         "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/milo/api/config"
 )
 
 var pubSubClientKey = "stores a pubsubClient"
 
 type PubSubMessage struct {
         Attributes map[string]string `json:"attributes"`
         Data       string            `json:"data"`
         MessageID  string            `json:"message_id"`
@@ skipping 20 matching lines @@
         getTopic(string) (*pubsub.Topic, error)
 
         // getSubscription returns the pubsub subscription if it exists,
         // a notExist error if it does not exist, or an error if there was an error.
         getSubscription(string) (*pubsub.Subscription, error)
 
         createSubscription(string, pubsub.SubscriptionConfig) (
                 *pubsub.Subscription, error)
 }
 
+type pubsubClients struct {
+        clients map[string]pubsubClient
+        lock    sync.RWMutex
+}
+
+func newClientsBundle() *pubsubClients {
+        return &pubsubClients{
+                clients: map[string]pubsubClient{},
+                lock:    sync.RWMutex{},

[nodir, 2017/07/12 23:06:58]

this line is noop, unnecessary

[Ryan Tseng, 2017/07/13 01:08:26]

Oh you're right, removed.

+        }
+}
+
 // prodPubSubClient is a wrapper around the production pubsub client.
 type prodPubSubClient struct {
         ctx    context.Context
         client *pubsub.Client
 }
 
 func (pc *prodPubSubClient) getTopic(id string) (*pubsub.Topic, error) {
         topic := pc.client.Topic(id)
         exists, err := topic.Exists(pc.ctx)
         switch {
@@ skipping 17 matching lines @@
         return sub, nil
 }
 
 func (pc *prodPubSubClient) createSubscription(id string, cfg pubsub.SubscriptionConfig) (
         *pubsub.Subscription, error) {
 
         return pc.client.CreateSubscription(pc.ctx, id, cfg)
 }
 
 // getPubSubClient extracts a debug PubSub client out of the context.
-func getPubSubClient(c context.Context) (pubsubClient, error) {
-        if client, ok := c.Value(&pubSubClientKey).(pubsubClient); ok {
-                return client, nil
+func getPubSubClient(c context.Context, projectID string) (pubsubClient, error) {
+        v := c.Value(&pubSubClientKey)
+        if clients, ok := v.(*pubsubClients); ok {
+                clients.lock.RLock()
+                client, ok := clients.clients[projectID]
+                clients.lock.RUnlock()
+                if ok {
+                        return client, nil
+                }
+                return nil, fmt.Errorf("no pubsub clients installed for " + projectID)
         }
         return nil, errors.New("no pubsub clients installed")
 }
 
 // withClient returns a context with a pubsub client instantiated to the
 // given project ID
 func withClient(c context.Context, projectID string) (context.Context, error) {
         if projectID == "" {
-                return nil, errors.New("missing buildbucket project")
+                return nil, errors.New("missing project id")
         }
         client, err := pubsub.NewClient(c, projectID)
         if err != nil {
                 return nil, err
         }
-        return context.WithValue(c, &pubSubClientKey, &prodPubSubClient{c, client}), nil
+        var clients *pubsubClients
+        var ok bool
+        if clients, ok = c.Value(&pubSubClientKey).(*pubsubClients); !ok {
+                clients = newClientsBundle()
+                c = context.WithValue(c, &pubSubClientKey, clients)
+        }
+        clients.lock.Lock()
+        clients.clients[projectID] = &prodPubSubClient{c, client}

[nodir, 2017/07/12 23:06:58]

this may cause a memory leak. Consider:

- there is a chain of contexts: c1 <- c2 <- c3 <- c4
  c1 is some global context while c4 is specific to a request
- c1 contains the client bundle
- c == c4
- the pubsubClient is created with c4 and is put into the bundle. Now the global
c1 has a references to request-specific c4
- future requests use the pubsubClient bound to the c4. This may cause all sorts
of problems

I am not sure prodPubSubClient should store context. Typically context is passed
in functions as the first parameter, as you know. For example, if a client wants
to set a timeout or pass metadata, they create an configured context and pass in
a function call, as opposed to reconfigure entire client object (which can be
global, like in this case).

Consider removing context from the prodPubSubClient and passing context in each
of its methods.

[Ryan Tseng, 2017/07/13 01:08:26]

I see, you're right.  I used this pattern only because I've seen it used in
other places but the concern is valid.

[Ryan Tseng, 2017/07/13 02:42:57]

Actually on second thought, withClient() is called per request, not on instance
startup.  The stored clients will expire between requests, so there is no global
context where the clients are stored.  Am I missing anything?

[nodir, 2017/07/13 05:58:46]

indeed, pubsub client is not global.

my concern with the current patchset is that withClient is written in a generic
way, but cannot be used in a generic way, i.e. one might call withClient during
app initialization, and then call it again with a context-with-request and cause
a memory leak. IMO code must be either generic without assumptions of
non-generic usage, or it must be non-generic. Both are possible in this
particular case.

Personally, I don't see a reason for this generality here, e.g. I don't see a
reason to put the clients into the context. You can achieve the same result by
doing: 
- in ensurePubSubClient: add miloClient and buildbucketClient parameters
- in EnsurePubSubClient: pass prodPubSubClient instances as the parameters
- in tests: pass testPubSubClients instead
so no need for context keys, bundles/map/mutex or functions newClientsBundle,
getTopic, getSubscription, createSubscription. This would be less code and it'd
still do the job.

If you still insist on preserving the generality, it is still achievable in a
non-problematic way. I assume you switched context value from a single client to
a map of clients because you have only one context key. You could declare `type
projectIDCtxKey string` and then use `projectIDCtxKey(projectID)` as a context
key when putting or retrieving a client to/from a context. This way the code is
still goroutine-safe and there is no need for map or mutex.

[Ryan Tseng, 2017/07/13 18:02:30]

I think it is very confusing to have multiple ways to reference services in the
same code (both by putting them in the context and by passing them as
parameters.  We also use a third way in Milo which is to turn functions into
methods where the class contains an overridable service provider that defaults
to a production instance if not overridden.  I think following multiple design
patterns at the same time makes things too confusing.

I think services in context is in general a confusing way to do things because
it involves a ton of code spread across a lot of files and interfaces.  But
because we've already committed to using this pattern everywhere, I thought it
was better to follow this pattern rather than invent yet another pattern, even
if it was more code.

Basically my primary object is to reduce complexity by limiting the number of
design patterns used.

The design pattern you suggested was one that I spent a long time refactoring
out of in another place in the code, which is why I'm a little hesitant to use
it.

[nodir, 2017/07/13 19:38:39]

I understand that you prefer services be in context, in general. Then I have two
suggestions:

1) Define a context key type (not a constant) tied to project id

type projectIDCtxKey string

// putting to context
return ctx.WithValue(projectIDCtxKey(projectId), myNewClient)
// This is goroutine safe

// retrieving
client = ctx.Value(projectIDCtxKey(projectId)).(pubsubClient)

2) I still don't understand why do you need getTopic, getSubscription and
createSubscription wrappers. A caller can do

topic = getPubSubClient(c, project).getTopic("buildbucket")

Removing the wrappers will reduce amount of code by 22 lines. 

Not in this CL though.

[nodir, 2017/07/13 21:55:00]

another issue with the current patchset is:

c1 := context.Background()
c2 := withClient(c1, "a")
withClient(c2, "b") // this mutates c2
getPubSubClient(c2, "b") is not nil

this happens because a mutable object (map) is put into the a context value. 

can be solved by the suggestion (1) above

+        clients.lock.Unlock()
+        return c, nil
 }
 
-func getTopic(c context.Context, id string) (*pubsub.Topic, error) {
-        client, err := getPubSubClient(c)
+func getTopic(c context.Context, project, id string) (*pubsub.Topic, error) {
+        client, err := getPubSubClient(c, project)
         if err != nil {
                 return nil, err
         }
         return client.getTopic(id)
 }
 
-func getSubscription(c context.Context, id string) (*pubsub.Subscription, error) {
-        client, err := getPubSubClient(c)
+func getSubscription(c context.Context, project, id string) (*pubsub.Subscription, error) {
+        client, err := getPubSubClient(c, project)
         if err != nil {
                 return nil, err
         }
         return client.getSubscription(id)
 }
 
-func createSubscription(c context.Context, id string, cfg pubsub.SubscriptionConfig) (
+func createSubscription(c context.Context, project, id string, cfg pubsub.SubscriptionConfig) (
         *pubsub.Subscription, error) {
 
-        client, err := getPubSubClient(c)
+        client, err := getPubSubClient(c, project)
         if err != nil {
                 return nil, err
         }
         return client.createSubscription(id, cfg)
 }
 
 // EnsurePubSubSubscribed makes sure the following subscriptions are in place:
 // * buildbucket, via the settings.Buildbucket.Topic setting
 func EnsurePubSubSubscribed(c context.Context, settings *config.Settings) error {
         if settings.Buildbucket != nil {
                 // Install the production pubsub client pointing to the buildbucket project
-                // into the context.
+                // into the context, so that we can get a reference to the topic.
                 c, err := withClient(c, settings.Buildbucket.Project)
                 if err != nil {
                         return err
                 }
+                // We also need a client for this project for the subscription.
+                c, err = withClient(c, info.AppID(c))
+                if err != nil {
+                        return err
+                }
                 return ensureBuildbucketSubscribed(c, settings.Buildbucket.Project)
         }
         // TODO(hinoka): Ensure buildbot subscribed.
         return nil
 }
 
 // ensureSubscribed is called by a cron job and ensures that the Milo
 // instance is properly subscribed to the buildbucket subscription endpoint.
 func ensureBuildbucketSubscribed(c context.Context, projectID string) error {
         topicID := "builds"
         // Check to see if the topic exists first.
-        topic, err := getTopic(c, topicID)
+        topic, err := getTopic(c, projectID, topicID)
         switch err {
         case errNotExist:
-                logging.WithError(err).Errorf(c, "%s does not exist", topicID)
+                luciErrors.Annotate(err, "%s does not exist", topicID).Err()

[nodir, 2017/07/12 23:06:58]

this is noop because the return value of Err() is not used

it should be returned

[Ryan Tseng, 2017/07/13 01:08:26]

my bad, thanks.

                 return err
         case nil:
                 // continue
         default:
                 if strings.Contains(err.Error(), "PermissionDenied") {
                         URL := "https://console.cloud.google.com/iam-admin/iam/project?project=" + projectID
                         acct, serr := info.ServiceAccount(c)
                         if serr != nil {
                                 acct = fmt.Sprintf("Unknown: %s", serr.Error())
                         }
                         // The documentation is incorrect.  We need Editor permission because
                         // the Subscriber permission does NOT permit attaching subscriptions to
                         // topics or to view topics.
                         logging.WithError(err).Errorf(
                                 c, "please go to %s and add %s as a Pub/Sub Editor", URL, acct)
                 } else {
                         logging.WithError(err).Errorf(c, "could not check topic %#v", topic)
                 }
                 return err
         }
         // Now check to see if the subscription already exists.
-        subID := info.AppID(c)
-        // Get the pubsub module of our app.  We do not want to use info.ModuleHostname()
-        // because it returns a version pinned hostname instead of the default route.
-        sub, err := getSubscription(c, subID)
+        appID := info.AppID(c)
+        sub, err := getSubscription(c, appID, "buildbucket")
         switch err {
         case errNotExist:
                 // continue
         case nil:
                 logging.Infof(c, "subscription %#v exists, no need to update", sub)
                 return nil
         default:
                 logging.WithError(err).Errorf(c, "could not check subscription %#v", sub)
                 return err
         }
+        // Get the pubsub module of our app.  We do not want to use info.ModuleHostname()
+        // because it returns a version pinned hostname instead of the default route.
         pubsubModuleHost := "pubsub." + info.DefaultVersionHostname(c)
 
         // No subscription exists, attach a new subscription to the existing topic.
         endpointURL := url.URL{
                 Scheme: "https",
                 Host:   pubsubModuleHost,
                 Path:   "/_ah/push-handlers/buildbucket",
         }
         subConfig := pubsub.SubscriptionConfig{
                 Topic:       topic,
                 PushConfig:  pubsub.PushConfig{Endpoint: endpointURL.String()},
                 AckDeadline: time.Minute * 10,
         }
-        newSub, err := createSubscription(c, subID, subConfig)
+        newSub, err := createSubscription(c, appID, "buildbucket", subConfig)
         if err != nil {
-                if strings.Contains(err.Error(), "The supplied HTTP URL is not registered") {
-                        registerURL := "https://console.cloud.google.com/apis/credentials/domainverification?project=" + projectID
-                        verifyURL := "https://www.google.com/webmasters/verification/verification?hl=en-GB&siteUrl=http://" + pubsubModuleHost
-                        logging.WithError(err).Errorf(
-                                c, "The domain has to be verified and added.\n\n"+
-                                        "1. Go to %s\n"+
-                                        "2. Verify the domain\n"+
-                                        "3. Go to %s\n"+
-                                        "4. Add %s to allowed domains\n\n",
-                                verifyURL, registerURL, pubsubModuleHost)
-                } else {
-                        logging.WithError(err).Errorf(c, "could not create subscription %#v", sub)
-                }
-                return err
+                return luciErrors.Annotate(err, "could not create subscription %#v", sub).Err()
         }
         // Success!
         logging.Infof(c, "successfully created subscription %#v", newSub)
         return nil
 }