gddo-server: publish crawl events to a Pub/Sub feed

Change-Id: I55d798a8be5ca1b05375d57d011eda3444b0b415
Reviewed-on: https://go-review.googlesource.com/114775
Reviewed-by: Tuo Shan <shantuo@google.com>

gddo-server/config.go

diff --git a/gddo-server/config.go b/gddo-server/config.go
index 199f262..7317ddf 100644
--- a/gddo-server/config.go
+++ b/gddo-server/config.go
@@ -65,6 +65,9 @@
 	ConfigGithubToken        = "github_token"
 	ConfigGithubClientID     = "github_client_id"
 	ConfigGithubClientSecret = "github_client_secret"
+
+	// Pub/Sub Config
+	ConfigCrawlPubSubTopic = "crawl-events"
 )
 
 func loadConfig(ctx context.Context, args []string) (*viper.Viper, error) {

gddo-server/crawl.go

diff --git a/gddo-server/crawl.go b/gddo-server/crawl.go
index c79e27a..d85352e 100644
--- a/gddo-server/crawl.go
+++ b/gddo-server/crawl.go
@@ -8,12 +8,15 @@
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"log"
 	"regexp"
 	"strings"
 	"time"
 
+	"cloud.google.com/go/pubsub"
+
 	"github.com/golang/gddo/doc"
 	"github.com/golang/gddo/gosrc"
 )
@@ -22,6 +25,27 @@
 	testdataPat = regexp.MustCompile(`/testdata(?:/|$)`)
 )
 
+// crawlNote is a message sent to Pub/Sub when a crawl occurs.
+// It is encoded with encoding/gob, so changes should match its
+// compatibility requirements.
+type crawlNote struct {
+	ImportPath string
+}
+
+func (s *server) publishCrawl(ctx context.Context, importPath string) {
+	if s.crawlTopic == nil {
+		return
+	}
+
+	note := &crawlNote{ImportPath: importPath}
+	b, err := json.Marshal(note)
+	if err != nil {
+		log.Printf("Encoding crawlNote: %v", err)
+		return
+	}
+	s.crawlTopic.Publish(ctx, &pubsub.Message{Data: b})
+}
+
 // crawlDoc fetches the package documentation from the VCS and updates the database.
 func (s *server) crawlDoc(ctx context.Context, source string, importPath string, pdoc *doc.Package, hasSubdirs bool, nextCrawl time.Time) (*doc.Package, error) {
 	message := []interface{}{source}
@@ -85,6 +109,7 @@
 		if err := s.put(ctx, pdoc, nextCrawl); err != nil {
 			log.Println(err)
 		}
+		s.publishCrawl(ctx, importPath)
 		return pdoc, nil
 	} else if e, ok := err.(gosrc.NotModifiedError); ok {
 		if pdoc.Status == gosrc.Active && !s.isActivePkg(importPath, e.Status) {
@@ -103,6 +128,7 @@
 				log.Printf("ERROR db.SetNextCrawl(%q): %v", importPath, err)
 			}
 		}
+		s.publishCrawl(ctx, importPath)
 		return pdoc, nil
 	} else if e, ok := err.(gosrc.NotFoundError); ok {
 		message = append(message, "notfound:", e)

gddo-server/main.go

diff --git a/gddo-server/main.go b/gddo-server/main.go
index 2f080fe..2137cb9 100644
--- a/gddo-server/main.go
+++ b/gddo-server/main.go
@@ -29,6 +29,7 @@
 	"time"
 
 	"cloud.google.com/go/logging"
+	"cloud.google.com/go/pubsub"
 	"cloud.google.com/go/trace"
 	"github.com/spf13/viper"
 
@@ -838,6 +839,7 @@
 	gceLogger   *GCELogger
 	templates   templateMap
 	traceClient *trace.Client
+	crawlTopic  *pubsub.Topic
 
 	statusPNG http.Handler
 	statusSVG http.Handler
@@ -861,6 +863,13 @@
 			return nil, err
 		}
 		s.traceClient.SetSamplingPolicy(sp)
+
+		// This topic should be created in the cloud console.
+		ps, err := pubsub.NewClient(ctx, proj)
+		if err != nil {
+			return nil, err
+		}
+		s.crawlTopic = ps.Topic(ConfigCrawlPubSubTopic)
 	}
 
 	assets := v.GetString(ConfigAssetsDir)