Taskcluster support

go.mod

@@ -5,7 +5,11 @@ go 1.14
 require (
 	github.com/davecgh/go-spew v1.1.1 // 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/taskcluster/taskcluster/clients/client-go/v23 v23.0.0
 	github.com/urfave/cli v1.17.1-0.20160608151511-fa949b48f384
 	go.mozilla.org/mozlog v0.0.0-20160610165107-cd74695caf44
+	launchpad.net/gocheck v0.0.0-20140225173054-000000000087 // indirect
 )

go.sum

@@ -1,10 +1,54 @@
+github.com/cenkalti/backoff/v3 v3.0.0 h1:ske+9nBpD9qZsTBoF41nW5L+AIuFBKMeze18XQ3eG1c=
+github.com/cenkalti/backoff/v3 v3.0.0/go.mod h1:cIeZDE3IrqwwJl6VUwCN6trj1oXrTS4rc0ij+ULvLYs=
 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/docopt/docopt-go v0.0.0-20180111231733-ee0de3bc6815/go.mod h1:WwZ+bS3ebgob9U8Nd0kOddGdZWjyMGR8Wziv+TBNwSE=
+github.com/fatih/camelcase v1.0.0 h1:hxNvNX/xYBp0ovncs8WyWZrOrpBNub/JfaMvbURyft8=
+github.com/fatih/camelcase v1.0.0/go.mod h1:yN2Sb0lFhZJUdVvtELVWefmrXpuZESvPmqwoZc+/fpc=
+github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
+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-20190827072141-edfb9018d271/go.mod h1:AZpEONHx3DKn8O/DFsRAY58/XVQiIPMTMB1SddzLXVw=
+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/httpbackoff/v3 v3.0.0 h1:Zh2BCW2iA3fzBBuZo2E4MvwyPSB6aimyI4EreeK3TRM=
+github.com/taskcluster/httpbackoff/v3 v3.0.0/go.mod h1:99ubellEC0fmRj7wnGkPftT2xfCY7NmbjT3gzn2ZPUM=
+github.com/taskcluster/jsonschema2go v1.0.0 h1:ZEDj2NKh8Sceq36zyLhSV6ann/aNXKZIe9cAXq7CDdk=
+github.com/taskcluster/jsonschema2go v1.0.0/go.mod h1:jhsT3XPj3iLNRx0efJVfFzZBZgxeYE7IHfZAai8wuKQ=
+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/taskcluster/slugid-go v1.1.0 h1:SWsUplliyamdYzOKVM4+lDohZKuL63fKreGkvIKJ9aI=
+github.com/taskcluster/slugid-go v1.1.0/go.mod h1:5sOAcPHjqso1UkKxSl77CkKgOwha0D9X0msBKBj0AOg=
+github.com/taskcluster/taskcluster-base-go v1.0.0 h1:Jh2R/J7+a23LjtYEHQtkFV04QBxx6EVX8E0PrzUqJo4=
+github.com/taskcluster/taskcluster-base-go v1.0.0/go.mod h1:ByyzyqqufsfZTrAHUw+0Grp8FwZAizZOKzVE1IpDXxQ=
+github.com/taskcluster/taskcluster-lib-urls v12.0.0+incompatible h1:57WLzh7B04y6ahTOJ8wjvdkbwYqnyJkwLXQ1Tu4E/DU=
+github.com/taskcluster/taskcluster-lib-urls v12.0.0+incompatible/go.mod h1:ALqTgi15AmJGEGubRKM0ydlLAFatlQPrQrmal9YZpQs=
+github.com/taskcluster/taskcluster/clients/client-go/v23 v23.0.0 h1:QQDQUy8zHLdfMF+EQEkA+NRGfjdykhHxNcPMakr+sm8=
+github.com/taskcluster/taskcluster/clients/client-go/v23 v23.0.0/go.mod h1:5NZbdHaaF2/yjkHtlcEKm4AXs3EH3M5Kih0wAEGRlMs=
+github.com/tent/hawk-go v0.0.0-20161026210932-d341ea318957 h1:6Fre/uvwovW5YY4nfHZk66cAg9HjT9YdFSAJHUUgOyQ=
+github.com/tent/hawk-go v0.0.0-20161026210932-d341ea318957/go.mod h1:dch7ywQEefE1ibFqBG1erFibrdUIwovcwQjksYuHuP4=
 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=
+github.com/xeipuuv/gojsonpointer v0.0.0-20190905194746-02993c407bfb/go.mod h1:N2zxlSyiKSe5eX1tZViRH5QA0qijqEDrYZiPEAiq3wU=
+github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415/go.mod h1:GwrjFmJcFw6At/Gs6z4yjiIwzuJ1/+UwLxMQDVQXShQ=
+github.com/xeipuuv/gojsonschema v1.1.0/go.mod h1:5yf86TLmAcydyeJq5YvxkGPE2fm/u4myDekKRoLuqhs=
 go.mozilla.org/mozlog v0.0.0-20160610165107-cd74695caf44 h1:ZwI2d4V8jjame21x82EXAGuKKStymcxBsNt0hkN4/5E=
 go.mozilla.org/mozlog v0.0.0-20160610165107-cd74695caf44/go.mod h1:jHoPAGnDrCy6kaI2tAze5Prf0Nr0w/oNkROt2lw3n3o=
+golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/tools v0.0.0-20190719005602-e377ae9d6386/go.mod h1:jcCCGcm9btYwXyDqrUWc6MKQKKGJCWEQ3AfLSRIbEuI=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v2 v2.2.2 h1:ZCJp+EgiOT7lHqUV2J862kp8Qj64Jo6az82+3Td9dZw=
+gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+launchpad.net/gocheck v0.0.0-20140225173054-000000000087 h1:Izowp2XBH6Ya6rv+hqbceQyw/gSGoXfH/UPoTGduL54=
+launchpad.net/gocheck v0.0.0-20140225173054-000000000087/go.mod h1:hj7XX3B/0A+80Vse0e+BUHsHMTEhd0O4cpUHr/e/BUM=

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:   "cloudops-pulse-prefix",
+			Usage:  "Pulse route to listen to.",
+			Value:  "cloudops.v1.deploy",
+			EnvVar: "CLOUDOPS_PULSE_PREFIX",
+		},
+		cli.StringFlag{
+			Name:   "pulse-queue",
+			Usage:  "Pulse quue to listen to.",
+			Value:  "deploy-proxy",
+			EnvVar: "PULSE_QUEUE",
+		},
+		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-password",
+			Usage:  "Password for authing against pulse",
+			Value:  "",
+			EnvVar: "PULSE_HOST",
+		},
 	}
 
 	app.Action = func(c *cli.Context) error {
@@ -74,6 +104,21 @@ func main() {
 			w.Write([]byte("OK"))
 		})
 
+		pulse := pulse.NewConnection(
+			c.String("pulse-username"),
+			c.String("pulse-password"),
+			c.String("pulse-host"),
+		)
+		taskclusterPulseHandler := proxyservice.NewTaskclusterPulseHandler(
+			jenkins,
+			&pulse,
+			c.String("cloudops-pulse-prefix"),
+		)
+
+		if err := taskclusterPulseHandler.Consume(); err != nil {
+			return cli.NewExitError(fmt.Sprintf("Could not listen to pulse: %v", err), 1)
+		}
+
 		server := &http.Server{
 			Addr:    c.String("addr"),
 			Handler: mux,

proxyservice/jenkins.go

@@ -116,3 +116,25 @@ func (j *Jenkins) TriggerDockerhubJob(data *DockerHubWebhookData) error {
 	params.Set("RawJSON", string(rawJSON))
 	return j.TriggerJob(path, params)
 }
+
+// triggers a jenkins job given
+func (j *Jenkins) TriggerTaskclusterJob(taskID string, route string, data *TaskCompletedMessage) error {
+	if !regexp.MustCompile(`^[A-Za-z0-9_-]{8}[Q-T][A-Za-z0-9_-][CGKOSWaeimquy26-][A-Za-z0-9_-]{10}[AQgw]$`).MatchString(taskID) {
+		return fmt.Errorf("Invalid taskID: %s", taskID)
+	}
+	// FIXME: This should probably be split on .
+	if !regexp.MustCompile(`^[a-zA-Z0-9_\-]{2,255}$`).MatchString(route) {
+		return fmt.Errorf("Invalid data.Repository.Namespace: %s", route)
+	}
+
+	rawJSON, err := json.Marshal(data)
+	if err != nil {
+		return fmt.Errorf("Error marshaling data: %v", err)
+	}
+	path := path.Join("/job/taskcluster/job",
+		route, "job", route)
+	params := url.Values{}
+	params.Set("TASK_ID", taskID)
+	params.Set("RawJSON", string(rawJSON))
+	return j.TriggerJob(path, params)
+}

proxyservice/taskcluster.go

@@ -0,0 +1,70 @@
+package proxyservice
+
+import (
+	"fmt"
+	"log"
+	"strings"
+
+	"github.com/streadway/amqp"
+	"github.com/taskcluster/pulse-go/pulse"
+	"github.com/taskcluster/taskcluster/clients/client-go/v23/tcqueueevents"
+)
+
+type TaskCompletedMessage = tcqueueevents.TaskCompletedMessage
+
+type routeTaskCompleted struct {
+	Route string
+}
+
+func (binding routeTaskCompleted) RoutingKey() string {
+	return fmt.Sprintf("route.%s", binding.Route)
+}
+
+func (binding routeTaskCompleted) ExchangeName() string {
+	return "exchange/taskcluster-queue/v1/task-completed"
+}
+
+func (binding routeTaskCompleted) NewPayloadObject() interface{} {
+	return new(tcqueueevents.TaskCompletedMessage)
+}
+
+type TaskclusterPulseHandler struct {
+	Jenkins          *Jenkins
+	Pulse            *pulse.Connection
+	PulseRoutePrefix string
+}
+
+func NewTaskclusterPulseHandler(jenkins *Jenkins, pulse *pulse.Connection, routePrefix string) *TaskclusterPulseHandler {
+	return &TaskclusterPulseHandler{
+		Jenkins:          jenkins,
+		Pulse:            pulse,
+		PulseRoutePrefix: routePrefix,
+	}
+}
+
+func (handler *TaskclusterPulseHandler) handleMessage(message interface{}, delivery amqp.Delivery) {
+	routingKeyPrefix := "route." + handler.PulseRoutePrefix
+	switch t := message.(type) {
+	case *tcqueueevents.TaskCompletedMessage:
+		if strings.HasPrefix(delivery.RoutingKey, routingKeyPrefix) {
+			route := strings.TrimPrefix(delivery.RoutingKey, routingKeyPrefix)
+			if err := handler.Jenkins.TriggerTaskclusterJob(t.Status.TaskID, route, t); err != nil {
+				log.Printf("Error triggering taskcluster job: %s", err)
+			}
+		}
+	}
+	delivery.Ack(false) // acknowledge message *after* processing
+
+}
+
+func (handler *TaskclusterPulseHandler) Consume() error {
+	routingKeyPrefix := "route." + handler.PulseRoutePrefix
+	_, err := handler.Pulse.Consume(
+		"", // queue name
+		handler.handleMessage,
+		1,     // prefetch 1 message at a time
+		false, // don't autoacknowledge messages
+		routeTaskCompleted{Route: routingKeyPrefix + ".#"},
+	)
+	return err
+}