feat: Support publishing notifications for new log entries

Signed-off-by: James Alseth <[email protected]>

cmd/rekor-server/app/root.go

@@ -73,7 +73,7 @@ func init() {
 	rootCmd.PersistentFlags().StringVar(&logType, "log_type", "dev", "logger type to use (dev/prod)")
 	rootCmd.PersistentFlags().BoolVar(&enablePprof, "enable_pprof", false, "enable pprof for profiling on port 6060")
 	rootCmd.PersistentFlags().Bool("enable_killswitch", false, "enable killswitch for TESTING ONLY on port 2345")
-	_ = rootCmd.PersistentFlags().MarkHidden("enable_killswitch")
+	rootCmd.PersistentFlags().MarkHidden("enable_killswitch")
 
 	rootCmd.PersistentFlags().String("trillian_log_server.address", "127.0.0.1", "Trillian log server address")
 	rootCmd.PersistentFlags().Uint16("trillian_log_server.port", 8090, "Trillian log server port")
@@ -95,6 +95,8 @@ func init() {
 Memory and file-based signers should only be used for testing.`)
 	rootCmd.PersistentFlags().String("rekor_server.signer-passwd", "", "Password to decrypt signer private key")
 
+	rootCmd.PersistentFlags().String("rekor_server.new_entry_publisher", "", "URL for pub/sub queue to send messages to when new entries are added to the log. Ignored if not set.")
+
 	rootCmd.PersistentFlags().Uint16("port", 3000, "Port to bind to")
 
 	rootCmd.PersistentFlags().Bool("enable_retrieve_api", true, "enables Redis-based index API endpoint")

go.mod

@@ -181,7 +181,7 @@ require (
 	golang.org/x/term v0.9.0 // indirect
 	golang.org/x/text v0.10.0 // indirect
 	golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2 // indirect
-	google.golang.org/api v0.128.0 // indirect
+	google.golang.org/api v0.128.0
 	google.golang.org/appengine v1.6.7 // indirect
 	gopkg.in/square/go-jose.v2 v2.6.0 // indirect
 	gopkg.in/yaml.v2 v2.4.0

pkg/api/api.go

@@ -31,6 +31,7 @@ import (
 	"google.golang.org/grpc/credentials/insecure"
 
 	"github.com/sigstore/rekor/pkg/log"
+	"github.com/sigstore/rekor/pkg/pubsub"
 	"github.com/sigstore/rekor/pkg/sharding"
 	"github.com/sigstore/rekor/pkg/signer"
 	"github.com/sigstore/rekor/pkg/storage"
@@ -63,6 +64,9 @@ type API struct {
 	signer     signature.Signer
 	// stops checkpoint publishing
 	checkpointPublishCancel context.CancelFunc
+	// Publishes notifications when new entries are added to the log. May be
+	// nil if no publisher is configured.
+	newEntryPublisher pubsub.Publisher
 }
 
 func NewAPI(treeID uint) (*API, error) {
@@ -112,6 +116,11 @@ func NewAPI(treeID uint) (*API, error) {
 
 	pubkey := cryptoutils.PEMEncode(cryptoutils.PublicKeyPEMType, b)
 
+	publisher, err := pubsub.Get(ctx, viper.GetString("rekor_server.new_entry_publisher"))
+	if err != nil {
+		return nil, fmt.Errorf("init event publisher: %w", err)
+	}
+
 	return &API{
 		// Transparency Log Stuff
 		logClient: logClient,
@@ -121,6 +130,8 @@ func NewAPI(treeID uint) (*API, error) {
 		pubkey:     string(pubkey),
 		pubkeyHash: hex.EncodeToString(pubkeyHashBytes[:]),
 		signer:     rekorSigner,
+		// Utility functionality not required for operation of the core service
+		newEntryPublisher: publisher,
 	}, nil
 }
 
@@ -165,5 +176,8 @@ func ConfigureAPI(treeID uint) {
 }
 
 func StopAPI() {
+	if api.newEntryPublisher != nil {
+		api.newEntryPublisher.Close()
+	}
 	api.checkpointPublishCancel()
 }

pkg/api/entries.go

@@ -301,6 +301,16 @@ func createLogEntry(params entries.CreateLogEntryParams) (models.LogEntry, middl
 	logEntry := models.LogEntry{
 		entryID: logEntryAnon,
 	}
+
+	if api.newEntryPublisher != nil {
+		// Publishing notifications should not block the API response.
+		go func() {
+			if err := api.newEntryPublisher.Publish(context.Background(), logEntryAnon); err != nil {
+				log.ContextLogger(ctx).Error(err)
+			}
+		}()
+	}
+
 	return logEntry, nil
 }