Bug 1619470: Add support for listening to hg.mozilla.org pulse messsages.

go.mod

@@ -4,8 +4,11 @@ go 1.14
 
 require (
 	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/pborman/uuid v1.2.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/streadway/amqp v0.0.0-20200108173154-1c71cc93ed71
 	github.com/stretchr/testify v1.1.4-0.20160524234229-8d64eb7173c7
+	github.com/taskcluster/pulse-go v1.0.0
 	github.com/urfave/cli v1.17.1-0.20160608151511-fa949b48f384
 	go.mozilla.org/mozlog v0.0.0-20160610165107-cd74695caf44
 )

go.sum

@@ -1,9 +1,17 @@
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/google/uuid v1.0.0 h1:b4Gk+7WdP/d3HZH8EJsZpvV7EtDOgaZLtnaNGIu1adA=
+github.com/google/uuid v1.0.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/pborman/uuid v1.2.0 h1:J7Q5mO4ysT1dv8hyrUGHb9+ooztCXu1D8MY8DZYsu3g=
+github.com/pborman/uuid v1.2.0/go.mod h1:X/NO0urCmaxf9VXbdlT7C2Yzkj2IKimNn4k+gtPdI/k=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/streadway/amqp v0.0.0-20200108173154-1c71cc93ed71 h1:2MR0pKUzlP3SGgj5NYJe/zRYDwOu9ku6YHy+Iw7l5DM=
+github.com/streadway/amqp v0.0.0-20200108173154-1c71cc93ed71/go.mod h1:AZpEONHx3DKn8O/DFsRAY58/XVQiIPMTMB1SddzLXVw=
 github.com/stretchr/testify v1.1.4-0.20160524234229-8d64eb7173c7 h1:5KNW+VDCZxdFJasHPW1AehzhTIm69pHQq2psutjjXRk=
 github.com/stretchr/testify v1.1.4-0.20160524234229-8d64eb7173c7/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
+github.com/taskcluster/pulse-go v1.0.0 h1:ys4ZUNp5TYiV5LSMxge4YF/AtuBUNH9StAA/bkno+r0=
+github.com/taskcluster/pulse-go v1.0.0/go.mod h1:uuaqnRQj9XqouabCEKjnrlJiC6UT9Gurx2oSe6s+irM=
 github.com/urfave/cli v1.17.1-0.20160608151511-fa949b48f384 h1:nYZrQfefkHIWNPwwIToQOnTeZ5nQt4nUHiS4wHQBWSA=
 github.com/urfave/cli v1.17.1-0.20160608151511-fa949b48f384/go.mod h1:70zkFmudgCuE/ngEzBv17Jvp/497gISqfk5gWijbERA=
 go.mozilla.org/mozlog v0.0.0-20160610165107-cd74695caf44 h1:ZwI2d4V8jjame21x82EXAGuKKStymcxBsNt0hkN4/5E=

main.go

@@ -9,6 +9,8 @@ import (
 	"go.mozilla.org/mozlog"
 
 	"github.com/urfave/cli"
+
+	"github.com/taskcluster/pulse-go/pulse"
 )
 
 func init() {
@@ -47,6 +49,34 @@ func main() {
 			Usage:  "Password for authing against jenkins",
 			EnvVar: "JENKINS_PASSWORD",
 		},
+		cli.StringFlag{
+			Name:   "pulse-username",
+			Usage:  "Username for authing against pulse",
+			EnvVar: "PULSE_USERNAME",
+		},
+		cli.StringFlag{
+			Name:   "pulse-password",
+			Usage:  "Password for authing against pulse",
+			EnvVar: "PULSE_PASSWORD",
+		},
+		cli.StringFlag{
+			Name:   "pulse-host",
+			Usage:  "Pulse host to connect to",
+			Value:  "",
+			EnvVar: "PULSE_HOST",
+		},
+		cli.StringSliceFlag{
+			Name:   "hgmo-repo",
+			Usage:  "hg.mozilla.org repo to listend to pushes from (can be used multiple times)",
+			Value:  &cli.StringSlice{"ci/ci-admin", "ci/ci-configuration"},
+			EnvVar: "HGMO_REPO",
+		},
+		cli.StringFlag{
+			Name:   "hgmo-pulse-queue",
+			Usage:  "Pulse queue to listen to.",
+			Value:  "hgmo",
+			EnvVar: "HGMO_PULSE_QUEUE",
+		},
 	}
 
 	app.Action = func(c *cli.Context) error {
@@ -74,6 +104,25 @@ func main() {
 			w.Write([]byte("OK"))
 		})
 
+		if c.String("pulse-host") != "" {
+			pulse := pulse.NewConnection(
+				c.String("pulse-username"),
+				c.String("pulse-password"),
+				c.String("pulse-host"),
+			)
+
+			hgmoPulseHandler := proxyservice.NewHgmoPulseHandler(
+				jenkins,
+				&pulse,
+				c.String("hgmo-pulse-queue"),
+				c.StringSlice("hgmo-repo")...,
+			)
+
+			if err := hgmoPulseHandler.Consume(); err != nil {
+				return cli.NewExitError(fmt.Sprintf("Could not listen to hgmo pulse: %v", err), 1)
+			}
+		}
+
 		server := &http.Server{
 			Addr:    c.String("addr"),
 			Handler: mux,
@@ -94,6 +143,20 @@ func validateCliContext(c *cli.Context) error {
 		}
 	}
 
+	pulseSpecified := false
+	pulseMissing := false
+	pulseOptions := []string{"pulse-username", "pulse-password", "pulse-host"}
+	for _, s := range pulseOptions {
+		if c.String(s) == "" {
+			pulseMissing = true
+		} else {
+			pulseSpecified = true
+		}
+	}
+	if pulseMissing && pulseSpecified {
+		cErrors = append(cErrors, fmt.Errorf("All or none of %s must be set", pulseOptions))
+	}
+
 	if len(cErrors) > 0 {
 		return cli.NewMultiError(cErrors...)
 	}

proxyservice/fixtures/hgmo_central_changegroup.json

@@ -0,0 +1,27 @@
+{
+  "_meta": {
+    "exchange": "exchange/hgpushes/v2",
+    "routing_key": "ci/ci-admin",
+    "sent": "2020-04-30T19:02:53.272543",
+    "serializer": "json"
+  },
+  "payload": {
+    "data": {
+      "heads": [
+        "9442967f483a7c61b520c3f559a8db9fb29aa573"
+      ],
+      "pushlog_pushes": [
+        {
+          "push_full_json_url": "https://hg.mozilla.org/mozilla-central/json-pushes?version=2&full=1&startID=37345&endID=37346",
+          "push_json_url": "https://hg.mozilla.org/mozilla-central/json-pushes?version=2&startID=37345&endID=37346",
+          "pushid": 37346,
+          "time": 1587752487,
+          "user": "mozilla@hocat.ca"
+        }
+      ],
+      "repo_url": "https://hg.mozilla.org/mozilla-central",
+      "source": "serve"
+    },
+    "type": "changegroup.1"
+  }
+}

proxyservice/fixtures/hgmo_changegroup.json

@@ -0,0 +1,27 @@
+{
+  "_meta": {
+    "exchange": "exchange/hgpushes/v2",
+    "routing_key": "ci/ci-admin",
+    "sent": "2020-04-30T19:02:53.272543",
+    "serializer": "json"
+  },
+  "payload": {
+    "data": {
+      "heads": [
+        "9c9a898b351909b2e0fe8420ac9d649ded523af3"
+      ],
+      "pushlog_pushes": [
+        {
+          "push_full_json_url": "https://hg.mozilla.org/ci/ci-admin/json-pushes?version=2&full=1&startID=157&endID=158",
+          "push_json_url": "https://hg.mozilla.org/ci/ci-admin/json-pushes?version=2&startID=157&endID=158",
+          "pushid": 158,
+          "time": 1588273363,
+          "user": "mozilla@hocat.ca"
+        }
+      ],
+      "repo_url": "https://hg.mozilla.org/ci/ci-admin",
+      "source": "serve"
+    },
+    "type": "changegroup.1"
+  }
+}

proxyservice/fixtures/hgmo_hg-extra_changegroup.json

@@ -0,0 +1,27 @@
+{
+  "_meta": {
+    "exchange": "exchange/hgpushes/v2",
+    "routing_key": "ci/ci-admin",
+    "sent": "2020-04-30T19:02:53.272543",
+    "serializer": "json"
+  },
+  "payload": {
+    "data": {
+      "heads": [
+        "1262b7fc3aecbe4e192f29e9e619eef8ba6e3f35"
+      ],
+      "pushlog_pushes": [
+        {
+          "push_full_json_url": "https://hg.mozilla.org/users/mozilla_hocat.ca/hg-extra/json-pushes?version=2&full=1&startID=1&endID=2",
+          "push_json_url": "https://hg.mozilla.org/users/mozilla_hocat.ca/hg-extra/json-pushes?version=2&startID=1&endID=2",
+          "pushid": 2,
+          "time": 1587713752,
+          "user": "mozilla@hocat.ca"
+        }
+      ],
+      "repo_url": "https://hg.mozilla.org/users/mozilla_hocat.ca/hg-extra",
+      "source": "serve"
+    },
+    "type": "changegroup.1"
+  }
+}

proxyservice/helpers_test.go

@@ -1,7 +1,10 @@
 package proxyservice
 
 import (
+	"bytes"
 	"io"
+	"io/ioutil"
+	"log"
 	"net/http"
 	"net/http/httptest"
 	"net/url"
@@ -34,3 +37,29 @@ func sendRequest(method, url string, body io.Reader, h http.Handler) *httptest.R
 	h.ServeHTTP(recorder, req)
 	return recorder
 }
+
+func loadFixture(path string) []byte {
+	data, err := ioutil.ReadFile(path)
+	if err != nil {
+		log.Fatalf("loadFixture: %s", err)
+		panic(err)
+	}
+	return data
+}
+
+type LogCapture struct {
+	Messages  bytes.Buffer
+	oldWriter io.Writer
+}
+
+func (l *LogCapture) Start() {
+	l.oldWriter = log.Writer()
+	log.SetOutput(io.MultiWriter(&l.Messages, l.oldWriter))
+}
+
+func (l *LogCapture) Reset() {
+	if l.oldWriter != nil {
+		log.SetOutput(l.oldWriter)
+		l.oldWriter = nil
+	}
+}