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"
         "time"
 
@@ skipping 34 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 map[string]pubsubClient
+
 // 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 {
+                if client, ok := clients[projectID]; ok {
+                        return client, nil
+                }
+                return nil, fmt.Errorf("no pubsub clients installed for" + projectID)

[nodir, 2017/07/12 21:31:12]

space missing

...for %s"

[Ryan Tseng, 2017/07/12 22:26:08]

Done.

         }
         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 = pubsubClients{}
+                c = context.WithValue(c, &pubSubClientKey, clients)
+        }
+        clients[projectID] = &prodPubSubClient{c, client}

[nodir, 2017/07/12 21:31:12]

it used to be gouroutine safe, but now it is not. Typically functions that deal
with context are goroutine safe.

You need to RWMutex on the map

[Ryan Tseng, 2017/07/12 22:26:08]

Done.

+        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)
                 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)
-                }
+                logging.WithError(err).Errorf(c, "could not create subscription %#v", sub)

[nodir, 2017/07/12 21:31:12]

why do we both log error and return it? the caller is likely to log it too, so
the error would be double-logged.

consider adding the context ("could not create subscription...") to the error
itself (perhaps using errors.Annotate) and return it without logging.

[Ryan Tseng, 2017/07/12 22:26:08]

Done.

                 return err
         }
         // Success!
         logging.Infof(c, "successfully created subscription %#v", newSub)
         return nil
 }