Milo: Buildbucket PubSub ingestion outline

milo/buildsource/buildbot/pubsub.go

 // Copyright 2016 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 buildbot
 
 import (
         "bytes"
         "compress/gzip"
         "compress/zlib"
-        "encoding/base64"
         "encoding/json"
         "fmt"
         "net/http"
         "strings"
         "time"
 
         ds "github.com/luci/gae/service/datastore"
         "github.com/luci/luci-go/common/clock"
         "github.com/luci/luci-go/common/iotools"
         "github.com/luci/luci-go/common/logging"
@@ skipping 22 matching lines @@
         masterCounter = metric.NewCounter(
                 "luci/milo/buildbot_pubsub/masters",
                 "The number of buildbot master jsons received by Milo from PubSub",
                 nil,
                 field.Bool("internal"),
                 field.String("master"),
                 // Status can be one of 2 options.  "success", "failure".
                 field.String("status"))
 )
 
-type pubSubMessage struct {
-        Attributes map[string]string `json:"attributes"`
-        Data       string            `json:"data"`
-        MessageID  string            `json:"message_id"`
-}
-
-type pubSubSubscription struct {
-        Message      pubSubMessage `json:"message"`
-        Subscription string        `json:"subscription"`
-}
-
 type buildMasterMsg struct {
         Master *buildbotMaster  `json:"master"`
         Builds []*buildbotBuild `json:"builds"`
 }
 
 // buildbotMasterEntry is a container for a marshaled and packed buildbot
 // master json.
 type buildbotMasterEntry struct {
         // Name of the buildbot master.
         Name string `gae:"$id"`
@@ skipping 30 matching lines @@
         if err := e.Encode(master); err != nil {
                 return err
         }
         gsw.Close()
         entry.Data = gzbs.Bytes()
         logging.Debugf(c, "Length of json data: %d", cw.Count)
         logging.Debugf(c, "Length of gzipped data: %d", len(entry.Data))
         return ds.Put(c, &entry)
 }
 
-// GetData returns the expanded form of Data (decoded from base64).
-func (m *pubSubSubscription) GetData() ([]byte, error) {
-        return base64.StdEncoding.DecodeString(m.Message.Data)
-}
-
 // unmarshal a gzipped byte stream into a list of buildbot builds and masters.
 func unmarshal(
         c context.Context, msg []byte) ([]*buildbotBuild, *buildbotMaster, error) {
         bm := buildMasterMsg{}
         if len(msg) == 0 {
                 return bm.Builds, bm.Master, nil
         }
         reader, err := zlib.NewReader(bytes.NewReader(msg))
         if err != nil {
                 logging.WithError(err).Errorf(c, "gzip decompression error")
@@ skipping 156 matching lines @@
 // PubSubHandler is a webhook that stores the builds coming in from pubsub.
 func PubSubHandler(ctx *router.Context) {
         statusCode := pubSubHandlerImpl(ctx.Context, ctx.Request)
         ctx.Writer.WriteHeader(statusCode)
 }
 
 // This is the actual implementation of the pubsub handler.  Returns
 // a status code.  StatusOK (200) for okay (ACK implied, don't retry).
 // Anything else will signal to pubsub to retry.
 func pubSubHandlerImpl(c context.Context, r *http.Request) int {
-        msg := pubSubSubscription{}
+        msg := common.PubSubSubscription{}
         now := int(clock.Now(c).Unix())
         defer r.Body.Close()
         dec := json.NewDecoder(r.Body)
         if err := dec.Decode(&msg); err != nil {
                 logging.WithError(err).Errorf(
                         c, "Could not decode message.  %s", err)
                 return http.StatusOK // This is a hard failure, we don't want PubSub to retry.
         }
         internal := true
         // Get the name of the subscription on luci-config
@@ skipping 91 matching lines @@
 
         }
         if master != nil {
                 code := doMaster(c, master, internal)
                 if code != 0 {
                         return code
                 }
         }
         return http.StatusOK
 }