email-exporter: Initial implementation

Makefile

@@ -6,7 +6,7 @@ VERSION ?= 1.0.0
 EPOCH ?= 1
 MAINTAINER ?= "Community"
 
-CMDS = admin boulder ceremony ct-test-srv
+CMDS = admin boulder ceremony ct-test-srv pardot-test-srv
 CMD_BINS = $(addprefix bin/, $(CMDS) )
 OBJECTS = $(CMD_BINS)
 

cmd/boulder-wfe2/main.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/letsencrypt/boulder/cmd"
 	"github.com/letsencrypt/boulder/config"
+	emailpb "github.com/letsencrypt/boulder/email/proto"
 	"github.com/letsencrypt/boulder/features"
 	"github.com/letsencrypt/boulder/goodkey"
 	"github.com/letsencrypt/boulder/goodkey/sagoodkey"
@@ -59,8 +60,9 @@ type Config struct {
 
 		TLS cmd.TLSConfig
 
-		RAService *cmd.GRPCClientConfig
-		SAService *cmd.GRPCClientConfig
+		RAService     *cmd.GRPCClientConfig
+		SAService     *cmd.GRPCClientConfig
+		EmailExporter *cmd.GRPCClientConfig
 
 		// GetNonceService is a gRPC config which contains a single SRV name
 		// used to lookup nonce-service instances used exclusively for nonce
@@ -285,6 +287,13 @@ func main() {
 	cmd.FailOnError(err, "Failed to load credentials and create gRPC connection to SA")
 	sac := sapb.NewStorageAuthorityReadOnlyClient(saConn)
 
+	var eec emailpb.ExporterClient
+	if c.WFE.EmailExporter != nil {
+		emailExporterConn, err := bgrpc.ClientSetup(c.WFE.EmailExporter, tlsConfig, stats, clk)
+		cmd.FailOnError(err, "Failed to load credentials and create gRPC connection to SA")
+		eec = emailpb.NewExporterClient(emailExporterConn)
+	}
+
 	if c.WFE.RedeemNonceService == nil {
 		cmd.Fail("'redeemNonceService' must be configured.")
 	}
@@ -351,6 +360,7 @@ func main() {
 		c.WFE.StaleTimeout.Duration,
 		rac,
 		sac,
+		eec,
 		gnc,
 		rnc,
 		noncePrefixKey,

cmd/boulder/main.go

@@ -19,6 +19,7 @@ import (
 	_ "github.com/letsencrypt/boulder/cmd/crl-checker"
 	_ "github.com/letsencrypt/boulder/cmd/crl-storer"
 	_ "github.com/letsencrypt/boulder/cmd/crl-updater"
+	_ "github.com/letsencrypt/boulder/cmd/email-exporter"
 	_ "github.com/letsencrypt/boulder/cmd/expiration-mailer"
 	_ "github.com/letsencrypt/boulder/cmd/id-exporter"
 	_ "github.com/letsencrypt/boulder/cmd/log-validator"

cmd/email-exporter/main.go

@@ -0,0 +1,107 @@
+package notmain
+
+import (
+	"context"
+	"flag"
+	"os"
+
+	"github.com/letsencrypt/boulder/cmd"
+	"github.com/letsencrypt/boulder/email"
+	emailpb "github.com/letsencrypt/boulder/email/proto"
+	bgrpc "github.com/letsencrypt/boulder/grpc"
+)
+
+// Config holds the configuration for the email-exporter service.
+type Config struct {
+	EmailExporter struct {
+		cmd.ServiceConfig
+
+		// PerDayLimit is our daily limit as determined by the tier of our
+		// Salesforce account. For more information, see:
+		// https://developer.salesforce.com/docs/marketing/pardot/guide/overview.html?q=rate%20limits
+		PerDayLimit float64 `validate:"required,min=1"`
+
+		// PardotBusinessUnit is the Pardot business unit to use.
+		PardotBusinessUnit string `validate:"required"`
+
+		// ClientId is the OAuth API client ID provided by Salesforce.
+		ClientId cmd.PasswordConfig
+
+		// ClientSecret is the OAuth API client secret provided by Salesforce.
+		ClientSecret cmd.PasswordConfig
+
+		// SalesforceBaseURL is the base URL for the Salesforce API. (e.g.,
+		// "https://login.salesforce.com")
+		SalesforceBaseURL string `validate:"required"`
+
+		// PardotBaseURL is the base URL for the Pardot API. (e.g.,
+		// "https://pi.pardot.com")
+		PardotBaseURL string `validate:"required"`
+	}
+	Syslog        cmd.SyslogConfig
+	OpenTelemetry cmd.OpenTelemetryConfig
+}
+
+func main() {
+	configFile := flag.String("config", "", "Path to configuration file")
+	grpcAddr := flag.String("addr", "", "gRPC listen address override")
+	debugAddr := flag.String("debug-addr", "", "Debug server address override")
+	flag.Parse()
+
+	if *configFile == "" {
+		flag.Usage()
+		os.Exit(1)
+	}
+
+	var c Config
+	err := cmd.ReadConfigFile(*configFile, &c)
+	cmd.FailOnError(err, "Reading JSON config file into config structure")
+
+	if *grpcAddr != "" {
+		c.EmailExporter.ServiceConfig.GRPC.Address = *grpcAddr
+	}
+	if *debugAddr != "" {
+		c.EmailExporter.ServiceConfig.DebugAddr = *debugAddr
+	}
+
+	scope, logger, oTelShutdown := cmd.StatsAndLogging(c.Syslog, c.OpenTelemetry, c.EmailExporter.ServiceConfig.DebugAddr)
+	defer oTelShutdown(context.Background())
+
+	logger.Info(cmd.VersionString())
+
+	clk := cmd.Clock()
+	clientId, err := c.EmailExporter.ClientId.Pass()
+	cmd.FailOnError(err, "Loading client ID")
+	clientSecret, err := c.EmailExporter.ClientSecret.Pass()
+	cmd.FailOnError(err, "Loading client secret")
+
+	pardotClient, err := email.NewPardotClientImpl(
+		clk,
+		c.EmailExporter.PardotBusinessUnit,
+		clientId,
+		clientSecret,
+		c.EmailExporter.SalesforceBaseURL,
+		c.EmailExporter.PardotBaseURL,
+	)
+	cmd.FailOnError(err, "Creating Pardot client")
+	exporterServer := email.NewExporterImpl(pardotClient, c.EmailExporter.PerDayLimit, scope, logger)
+
+	tlsConfig, err := c.EmailExporter.TLS.Load(scope)
+	cmd.FailOnError(err, "Loading TLS config")
+
+	daemonCtx, shutdownExporterServer := context.WithCancel(context.Background())
+	go exporterServer.Start(daemonCtx)
+
+	start, err := bgrpc.NewServer(c.EmailExporter.GRPC, logger).Add(
+		&emailpb.Exporter_ServiceDesc, exporterServer).Build(tlsConfig, scope, clk)
+	cmd.FailOnError(err, "Configuring gRPC server")
+
+	err = start()
+	shutdownExporterServer()
+	exporterServer.Drain()
+	cmd.FailOnError(err, "email-exporter gRPC service failed to start")
+}
+
+func init() {
+	cmd.RegisterCommand("email-exporter", main, &cmd.ConfigValidator{Config: &Config{}})
+}

email/exporter.go

@@ -0,0 +1,161 @@
+package email
+
+import (
+	"context"
+	"errors"
+	"sync"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"golang.org/x/time/rate"
+	"google.golang.org/protobuf/types/known/emptypb"
+
+	"github.com/letsencrypt/boulder/core"
+	emailpb "github.com/letsencrypt/boulder/email/proto"
+	berrors "github.com/letsencrypt/boulder/errors"
+	blog "github.com/letsencrypt/boulder/log"
+)
+
+const (
+	// five is the number of concurrent workers processing the email queue. This
+	// number was chosen specifically to match the number of concurrent
+	// connections allowed by the Pardot API.
+	five = 5
+
+	// queueCap enforces a maximum stack size to prevent unbounded growth.
+	queueCap = 10000
+)
+
+var ErrQueueFull = errors.New("email-exporter queue is full")
+
+// ExporterImpl implements the gRPC server and processes email exports.
+type ExporterImpl struct {
+	emailpb.UnsafeExporterServer
+
+	sync.Mutex
+	drainWG sync.WaitGroup
+	wake    *sync.Cond
+
+	limiter              *rate.Limiter
+	toSend               []string
+	client               PardotClient
+	emailsHandledCounter prometheus.Counter
+	log                  blog.Logger
+}
+
+var _ emailpb.ExporterServer = (*ExporterImpl)(nil)
+
+// NewExporterImpl creates a new ExporterImpl.
+func NewExporterImpl(client PardotClient, perDayLimit float64, scope prometheus.Registerer, logger blog.Logger) *ExporterImpl {
+	// This limiter enforces the daily Pardot API limit and restricts
+	// concurrency to the maximum of 5 requests specified in their
+	// documentation. For more details see:
+	// https://developer.salesforce.com/docs/marketing/pardot/guide/overview.html?q=rate%20limits
+	limiter := rate.NewLimiter(rate.Limit(perDayLimit/86400.0), 5)
+
+	emailsHandledCounter := prometheus.NewCounter(prometheus.CounterOpts{
+		Name: "email_exporter_emails_handled",
+		Help: "Total number of emails handled by the email exporter",
+	})
+	scope.MustRegister(emailsHandledCounter)
+
+	impl := &ExporterImpl{
+		limiter:              limiter,
+		toSend:               make([]string, 0, queueCap),
+		client:               client,
+		emailsHandledCounter: emailsHandledCounter,
+		log:                  logger,
+	}
+	impl.wake = sync.NewCond(&impl.Mutex)
+
+	queueGauge := prometheus.NewGaugeFunc(prometheus.GaugeOpts{
+		Name: "email_exporter_queue_length",
+		Help: "Current length of the email export queue",
+	}, func() float64 {
+		impl.Lock()
+		defer impl.Unlock()
+		return float64(len(impl.toSend))
+	})
+	scope.MustRegister(queueGauge)
+
+	return impl
+}
+
+// CreateProspects enqueues the provided email addresses. If the queue cannot
+// accommodate the new emails, an ErrQueueFull is returned.
+func (impl *ExporterImpl) CreateProspects(ctx context.Context, req *emailpb.CreateProspectsRequest) (*emptypb.Empty, error) {
+	if core.IsAnyNilOrZero(req, req.Emails) {
+		return nil, berrors.InternalServerError("Incomplete UpsertEmails request")
+	}
+
+	impl.Lock()
+	spotsLeft := queueCap - len(impl.toSend)
+	if spotsLeft < len(req.Emails) {
+		return nil, ErrQueueFull
+	}
+	impl.toSend = append(impl.toSend, req.Emails...)
+	impl.Unlock()
+	// Wake waiting workers to process the new emails.
+	impl.wake.Broadcast()
+
+	return &emptypb.Empty{}, nil
+}
+
+// Start begins asynchronous processing of the email queue. When the parent
+// daemonCtx is cancelled the queue will be drained and the workers will exit.
+func (impl *ExporterImpl) Start(daemonCtx context.Context) {
+	go func() {
+		<-daemonCtx.Done()
+		impl.Lock()
+		// Wake waiting workers to exit.
+		impl.wake.Broadcast()
+		impl.Unlock()
+	}()
+
+	worker := func() {
+		defer impl.drainWG.Done()
+		for {
+			impl.Lock()
+
+			for len(impl.toSend) == 0 && daemonCtx.Err() == nil {
+				// Wait for the queue to be updated or the daemon to exit.
+				impl.wake.Wait()
+			}
+
+			if len(impl.toSend) == 0 && daemonCtx.Err() != nil {
+				// No more emails to process, exit.
+				impl.Unlock()
+				return
+			}
+
+			// Dequeue and dispatch an email.
+			last := len(impl.toSend) - 1
+			email := impl.toSend[last]
+			impl.toSend = impl.toSend[:last]
+			impl.Unlock()
+
+			err := impl.limiter.Wait(daemonCtx)
+			if err != nil {
+				if !errors.Is(err, context.Canceled) {
+					impl.log.Errf("Unexpected limiter.Wait() error: %s", err)
+					continue
+				}
+			}
+
+			err = impl.client.CreateProspect(email)
+			if err != nil {
+				impl.log.Errf("Sending Prospect to Pardot: %s", err)
+			}
+			impl.emailsHandledCounter.Inc()
+		}
+	}
+
+	for range five {
+		impl.drainWG.Add(1)
+		go worker()
+	}
+}
+
+// Drain blocks until all workers have finished processing the email queue.
+func (impl *ExporterImpl) Drain() {
+	impl.drainWG.Wait()
+}