Merge pull request #5 from benjlevesque/gcr-pubsub

Add support for GCR

Author: bigkevmcd

README.md

@@ -17,6 +17,7 @@ Usage:
 Available Commands:
   help        Help about any command
   http        update repositories in response to image hooks
+  pubsub      update repositories in response to gcr pubsub events
   update      update a repository configuration
 
 Flags:
@@ -25,9 +26,9 @@ Flags:
 Use "image-updater [command] --help" for more information about a command.
 ```
 
-There are two sub-commands, `http` and `update`.
+There are three sub-commands, `http`, `pubsub` and `update`.
 
-`http` provides a Webhook service, and `update` will perform the same
+`http` provides a Webhook service, `pubsub` subscribes to pubsub events and `update` will perform the same
 functionality from the command-line.
 
 ## Update tool
@@ -57,15 +58,22 @@ This is a micro-service for updating Git Repos when a hook is received indicatin
 
 This currently supports receiving hooks from Docker and Quay.io.
 
-## WARNING
+### WARNING
 
 Neither Docker Hub nor Quay.io provide a way for receivers to authenticate Webhooks, which makes this insecure, a malicious user could trigger the creation of pull requests in your git hosting service.
 
 Please understand the risks of using this component.
 
+## Pubsub Service
+Similarly to the Webhook service, the pubsub services allows to update Git Repos when a pubsub Event is received. 
+
+This currently supports Events from [Google Cloud Registry](https://cloud.google.com/container-registry/docs/configuring-notifications).
+
+It requires two arguments `--project-id` and `--subscription-name`. See [below](#google-container-registry-setup) for more details on how to setup the subscription.
+
 ## Configuration
 
-This service uses a really simple configuration:
+Both the Webhook and Pubsub service uses a really simple configuration:
 
 ```yaml
 repositories:
@@ -133,6 +141,7 @@ changed to support Quay.io.
 The `--parser` command-line option chooses which of the supported (Quay, Docker)
 hook formats to parse.
 
+
 ## Exposing the Handler
 
 The Service exposes a Hook handler at `/` on port 8080 that handles the
@@ -143,6 +152,22 @@ configured hook type.
 A Tekton task is provided in [./tekton](./tekton) which allows you to apply
 updates to repos from a Tekton pipeline run.
 
+
+## Google Container registry setup
+```bash
+gcloud pubsub topics create gcr
+gcloud pubsub subscriptions create gcr-image-updater --topic projects/$GOOGLE_PROJECT/topics/gcr
+
+gcloud iam service-accounts create 
+gcloud iam service-accounts keys create credentials.json \
+  --iam-account $SA_NAME@$GOOGLE_PROJECT.iam.gserviceaccount.com
+
+gcloud pubsub subscriptions add-iam-policy-binding gcr-image-updater \
+--member=serviceAccount:$SA_NAME@$GOOGLE_PROJECT.iam.gserviceaccount.com --role=roles/pubsub.subscriber
+```
+
+You then need to set the `GOOGLE_APPLICATION_CREDENTIALS` environment variable to the path the generated  `credentials.json` file.
+
 ## Building
 
 A `Dockerfile` is provided for building a container, but otherwise:

go.mod

@@ -3,6 +3,7 @@ module github.com/gitops-tools/image-updater
 go 1.14
 
 require (
+	cloud.google.com/go/pubsub v1.0.1
 	github.com/gitops-tools/pkg v0.0.0-20200823054310-42f81b2b396d
 	github.com/go-logr/logr v0.1.0
 	github.com/go-logr/zapr v0.1.0

go.sum

@@ -4,10 +4,12 @@ cloud.google.com/go v0.38.0/go.mod h1:990N+gfupTy94rShfmMCWGDn0LpTmnzTp2qbd1dvSR
 cloud.google.com/go v0.44.1/go.mod h1:iSa0KzasP4Uvy3f1mN/7PiObzGgflwredwwASm/v6AU=
 cloud.google.com/go v0.44.2/go.mod h1:60680Gw3Yr4ikxnPRS/oxxkBccT6SA1yMk63TGekxKY=
 cloud.google.com/go v0.45.1/go.mod h1:RpBamKRgapWJb87xiFSdk4g1CME7QZg3uwTez+TSTjc=
+cloud.google.com/go v0.46.3 h1:AVXDdKsrtX33oR9fbCMu/+c1o8Ofjq6Ku/MInaLVg5Y=
 cloud.google.com/go v0.46.3/go.mod h1:a6bKKbmY7er1mI7TEI4lsAkts/mkhTSZK8w33B4RAg0=
 cloud.google.com/go/bigquery v1.0.1/go.mod h1:i/xbL2UlR5RvWAURpBYZTtm/cXjCha9lbfbpx4poX+o=
 cloud.google.com/go/datastore v1.0.0/go.mod h1:LXYbyblFSglQ5pkeyhO+Qmw7ukd3C+pD7TKLgZqpHYE=
 cloud.google.com/go/firestore v1.1.0/go.mod h1:ulACoGHTpvq5r8rxGJ4ddJZBZqakUQqClKRT5SZwBmk=
+cloud.google.com/go/pubsub v1.0.1 h1:W9tAK3E57P75u0XLLR82LZyw8VpAnhmyTOxW9qzmyj8=
 cloud.google.com/go/pubsub v1.0.1/go.mod h1:R0Gpsv3s54REJCy4fxDixWD93lHJMoZTyQ2kNxGRt3I=
 cloud.google.com/go/storage v1.0.0/go.mod h1:IhtSnM/ZTZV8YYJWCY8RULGVqBDmpoyjwiyrjsg+URw=
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
@@ -192,6 +194,7 @@ github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm4
 github.com/google/uuid v1.0.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+vpHVxEJEs9eg=
+github.com/googleapis/gax-go/v2 v2.0.5 h1:sjZBwGj9Jlw33ImPtvFviGYvseOtDM7hkSKB7+Tv3SM=
 github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5mhpdKc/us6bOk=
 github.com/googleapis/gnostic v0.0.0-20170729233727-0c5108395e2d/go.mod h1:sJBsCZ4ayReDTBIg8b9dl28c5xFWyhBTVRp3pOg5EKY=
 github.com/googleapis/gnostic v0.1.0/go.mod h1:sJBsCZ4ayReDTBIg8b9dl28c5xFWyhBTVRp3pOg5EKY=
@@ -227,6 +230,7 @@ github.com/hashicorp/go-uuid v1.0.1/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/b
 github.com/hashicorp/go.net v0.0.1/go.mod h1:hjKkEWcCURg++eb33jQU7oqQcI9XDCnUzHA0oac0k90=
 github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
+github.com/hashicorp/golang-lru v0.5.4 h1:YDjusn29QI/Das2iO9M0BHnIbxPeyuCHsjMW+lJfyTc=
 github.com/hashicorp/golang-lru v0.5.4/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
 github.com/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4=
 github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
@@ -422,6 +426,7 @@ go.mongodb.org/mongo-driver v1.0.3/go.mod h1:u7ryQJ+DOzQmeO7zB6MHyr8jkEQvC8vH7qL
 go.mongodb.org/mongo-driver v1.1.1/go.mod h1:u7ryQJ+DOzQmeO7zB6MHyr8jkEQvC8vH7qLUO4lqsUM=
 go.mongodb.org/mongo-driver v1.1.2/go.mod h1:u7ryQJ+DOzQmeO7zB6MHyr8jkEQvC8vH7qLUO4lqsUM=
 go.opencensus.io v0.21.0/go.mod h1:mSImk1erAIZhrmZN+AvHh14ztQfjbGwt4TtuofqLduU=
+go.opencensus.io v0.22.0 h1:C9hSCOW830chIVkdja34wa6Ky+IzWllkUinR+BtRZd4=
 go.opencensus.io v0.22.0/go.mod h1:+kGneAE2xo2IficOXnaByMWTGM9T73dGwxeWcUqIpI8=
 go.uber.org/atomic v1.3.2/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
 go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
@@ -504,6 +509,7 @@ golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190227155943-e225da77a7e6/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e h1:vcxGaoTs7kV8m5Np9uUNQin4BrLOthgV7252N8V+FwY=
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20170830134202-bb24a47a89ea/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -582,6 +588,7 @@ google.golang.org/api v0.4.0/go.mod h1:8k5glujaEP+g9n7WNsDg8QP6cUVNI86fCNMcbazEt
 google.golang.org/api v0.7.0/go.mod h1:WtwebWUNSVBH/HAw79HIFXZNqEvBhG+Ra+ax0hx3E3M=
 google.golang.org/api v0.8.0/go.mod h1:o4eAsZoiT+ibD93RtjEohWalFOjRDx6CVaqeizhEnKg=
 google.golang.org/api v0.9.0/go.mod h1:o4eAsZoiT+ibD93RtjEohWalFOjRDx6CVaqeizhEnKg=
+google.golang.org/api v0.13.0 h1:Q3Ui3V3/CVinFWFiW39Iw0kMuVrRzYX0wN6OPFp0lTA=
 google.golang.org/api v0.13.0/go.mod h1:iLdEw5Ide6rF15KTC1Kkl0iskquN2gFfn9o9XIsbkAI=
 google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM=
 google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
@@ -604,6 +611,7 @@ google.golang.org/grpc v1.21.0/go.mod h1:oYelfM1adQP15Ek0mdvEgi9Df8B9CZIaU1084ij
 google.golang.org/grpc v1.21.1/go.mod h1:oYelfM1adQP15Ek0mdvEgi9Df8B9CZIaU1084ijfRaM=
 google.golang.org/grpc v1.23.0/go.mod h1:Y5yQAOtifL1yxbo5wqy6BxZv8vAUGQwXBOALyacEbxg=
 google.golang.org/grpc v1.23.1/go.mod h1:Y5yQAOtifL1yxbo5wqy6BxZv8vAUGQwXBOALyacEbxg=
+google.golang.org/grpc v1.26.0 h1:2dTRdpdFEEhJYQD8EMLB61nnrzSCTbG38PhqdhvOltg=
 google.golang.org/grpc v1.26.0/go.mod h1:qbnxyOmOxrQa7FizSgH+ReBfzJrCY1pSN7KXBS8abTk=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
 google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=

pkg/cmd/pubsub.go

@@ -0,0 +1,110 @@
+package cmd
+
+import (
+	"context"
+	"fmt"
+	"os"
+
+	"cloud.google.com/go/pubsub"
+	"github.com/gitops-tools/image-updater/pkg/applier"
+	"github.com/gitops-tools/image-updater/pkg/config"
+	"github.com/gitops-tools/image-updater/pkg/hooks/gcr"
+	"github.com/gitops-tools/image-updater/pkg/pubsubhandler"
+	"github.com/gitops-tools/pkg/client"
+	"github.com/go-logr/zapr"
+	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
+	"go.uber.org/zap"
+)
+
+const (
+	projectIDFlag        = "project-id"
+	subscriptionNameFlag = "subscription-name"
+)
+
+type message struct {
+	data []byte
+}
+
+func (m *message) Ack()         {}
+func (m *message) Data() []byte { return m.data }
+
+func makePubsubCmd() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "pubsub",
+		Short: "update repositories in response to gcr pubsub events",
+		RunE: func(cmd *cobra.Command, args []string) error {
+			zapl, _ := zap.NewProduction()
+			defer func() {
+				_ = zapl.Sync() // flushes buffer, if any
+			}()
+			logger := zapr.NewLogger(zapl)
+			scmClient, err := createClientFromViper()
+			if err != nil {
+				return fmt.Errorf("failed to create a git driver: %s", err)
+			}
+			f, err := os.Open(viper.GetString("config"))
+			if err != nil {
+				return err
+			}
+			defer f.Close()
+			repos, err := config.Parse(f)
+			if err != nil {
+				return err
+			}
+			applier := applier.New(logger, client.New(scmClient), repos)
+
+			sub, err := createSubscriptionFromViper()
+			if err != nil {
+				return err
+			}
+
+			handler := pubsubhandler.New(logger, applier, gcr.Parse)
+
+			return sub.Receive(context.Background(), func(ctx context.Context, msg *pubsub.Message) {
+				handler.Handle(ctx, &message{
+					data: msg.Data,
+				})
+			})
+		},
+	}
+
+	cmd.Flags().String(
+		"config",
+		"/etc/image-updater/config.yaml",
+		"repository configuration",
+	)
+	logIfError(viper.BindPFlag("config", cmd.Flags().Lookup("config")))
+
+	cmd.Flags().String(
+		projectIDFlag,
+		"",
+		"GCP project ID",
+	)
+	logIfError(viper.BindPFlag(projectIDFlag, cmd.Flags().Lookup(projectIDFlag)))
+	logIfError(cmd.MarkFlagRequired(projectIDFlag))
+
+	cmd.Flags().String(
+		subscriptionNameFlag,
+		"",
+		"GCP subscription name",
+	)
+	logIfError(viper.BindPFlag(subscriptionNameFlag, cmd.Flags().Lookup(subscriptionNameFlag)))
+	logIfError(cmd.MarkFlagRequired(subscriptionNameFlag))
+
+	return cmd
+}
+
+func createSubscriptionFromViper() (*pubsub.Subscription, error) {
+	ctx := context.Background()
+	projectID := viper.GetString(projectIDFlag)
+	subscriptionName := viper.GetString(subscriptionNameFlag)
+
+	client, err := pubsub.NewClient(ctx, projectID)
+	if err != nil {
+		return nil, err
+	}
+
+	sub := client.Subscription(subscriptionName)
+	return sub, nil
+}

pkg/cmd/root.go

@@ -60,6 +60,7 @@ func makeRootCmd() *cobra.Command {
 
 	cmd.AddCommand(makeHTTPCmd())
 	cmd.AddCommand(makeUpdateCmd())
+	cmd.AddCommand(makePubsubCmd())
 	return cmd
 }
 

pkg/handler/handler.go

@@ -1,6 +1,7 @@
 package handler
 
 import (
+	"io/ioutil"
 	"net/http"
 
 	"github.com/go-logr/logr"
@@ -17,14 +18,12 @@ type Handler struct {
 }
 
 func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
-	h.log.Info("processing hook request")
-	hook, err := h.parser(r)
+	hook, err := h.parse(r)
 	if err != nil {
 		h.log.Error(err, "failed to parse request")
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 		return
 	}
-
 	err = h.applier.UpdateFromHook(r.Context(), hook)
 
 	if err != nil {
@@ -34,6 +33,17 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+func (h *Handler) parse(r *http.Request) (hooks.PushEvent, error) {
+	h.log.Info("processing hook request")
+	// TODO: LimitReader
+	data, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		h.log.Error(err, "failed to read request body")
+		return nil, err
+	}
+	return h.parser(data)
+}
+
 // New creates and returns a new Handler.
 func New(logger logr.Logger, u *applier.Applier, p hooks.PushEventParser) *Handler {
 	return &Handler{log: logger, applier: u, parser: p}

pkg/handler/handler_test.go

@@ -49,7 +49,7 @@ func TestHandler(t *testing.T) {
 }
 
 func TestHandlerWithParseFailure(t *testing.T) {
-	badParser := func(*http.Request) (hooks.PushEvent, error) {
+	badParser := func(payload []byte) (hooks.PushEvent, error) {
 		return nil, errors.New("failed")
 	}
 	logger := zapr.NewLogger(zaptest.NewLogger(t, zaptest.Level(zap.WarnLevel)))
@@ -85,6 +85,36 @@ func TestHandlerWithFailureToUpdate(t *testing.T) {
 	}
 }
 
+func TestParseWithNoBody(t *testing.T) {
+	logger := zapr.NewLogger(zaptest.NewLogger(t, zaptest.Level(zap.WarnLevel)))
+	m := mock.New(t)
+	applier := applier.New(logger, m, createConfigs(), updater.NameGenerator(stubNameGenerator{"a"}))
+	h := New(logger, applier, quay.Parse)
+	bodyErr := errors.New("just a test error")
+
+	req := httptest.NewRequest("POST", "/", failingReader{err: bodyErr})
+
+	_, err := h.parse(req)
+	if err != bodyErr {
+		t.Fatal("expected an error")
+	}
+}
+
+func TestParseWithUnparseableBody(t *testing.T) {
+	logger := zapr.NewLogger(zaptest.NewLogger(t, zaptest.Level(zap.WarnLevel)))
+	m := mock.New(t)
+	applier := applier.New(logger, m, createConfigs(), updater.NameGenerator(stubNameGenerator{"a"}))
+	h := New(logger, applier, quay.Parse)
+
+	req := httptest.NewRequest("POST", "/", nil)
+
+	_, err := h.parse(req)
+
+	if err == nil {
+		t.Fatal("expected an error")
+	}
+}
+
 func makeHookRequest(t *testing.T, fixture string) *http.Request {
 	t.Helper()
 	b, err := ioutil.ReadFile(fixture)
@@ -118,3 +148,14 @@ type stubNameGenerator struct {
 func (s stubNameGenerator) PrefixedName(p string) string {
 	return p + s.name
 }
+
+type failingReader struct {
+	err error
+}
+
+func (f failingReader) Read(p []byte) (n int, err error) {
+	return 0, f.err
+}
+func (f failingReader) Close() error {
+	return f.err
+}

pkg/hooks/docker/hook.go

@@ -3,21 +3,14 @@ package docker
 import (
 	"encoding/json"
 	"fmt"
-	"io/ioutil"
-	"net/http"
 
 	"github.com/gitops-tools/image-updater/pkg/hooks"
 )
 
-// Parse takes an http.Request and parses it into a Docker webhook event.
-func Parse(req *http.Request) (hooks.PushEvent, error) {
-	// TODO: LimitReader
-	data, err := ioutil.ReadAll(req.Body)
-	if err != nil {
-		return nil, err
-	}
+// Parse parses a payload into a Docker webhook event.
+func Parse(payload []byte) (hooks.PushEvent, error) {
 	h := &Webhook{}
-	err = json.Unmarshal(data, h)
+	err := json.Unmarshal(payload, h)
 	if err != nil {
 		return nil, err
 	}