Make proxy buffer size configurable

Author: siggisim

docs/config-misc.md

@@ -13,6 +13,7 @@ sidebar_label: Misc
 **Optional**
 
 - `hosts` A list of host strings that BuildBudy should connect and forward events to.
+- `buffer_size` The number of build events to buffer locally when proxying build events.
 
 ## Example section
 
@@ -21,4 +22,5 @@ build_event_proxy:
   hosts:
     - "grpc://localhost:1985"
     - "grpc://events.buildbuddy.io:1985"
+  buffer_size: 1000
 ```

server/build_event_protocol/build_event_proxy/BUILD

@@ -7,6 +7,7 @@ go_library(
     visibility = ["//visibility:public"],
     deps = [
         "//proto:publish_build_event_go_proto",
+        "//server/environment:go_default_library",
         "//server/util/grpc_client:go_default_library",
         "@io_bazel_rules_go//proto/wkt:empty_go_proto",
         "@org_golang_google_grpc//:go_default_library",

server/build_event_protocol/build_event_proxy/build_event_proxy.go

@@ -5,20 +5,20 @@ import (
 	"log"
 	"sync"
 
+	"github.com/buildbuddy-io/buildbuddy/server/environment"
 	"github.com/buildbuddy-io/buildbuddy/server/util/grpc_client"
 	"github.com/golang/protobuf/ptypes/empty"
 	"google.golang.org/grpc"
 
 	pepb "github.com/buildbuddy-io/buildbuddy/proto/publish_build_event"
 )
 
-const eventBufferSize = 100
-
 type BuildEventProxyClient struct {
-	target    string
-	clientMux sync.Mutex // PROTECTS(client)
-	client    pepb.PublishBuildEventClient
-	rootCtx   context.Context
+	target          string
+	clientMux       sync.Mutex // PROTECTS(client)
+	client          pepb.PublishBuildEventClient
+	rootCtx         context.Context
+	eventBufferSize int
 }
 
 func (c *BuildEventProxyClient) reconnectIfNecessary() {
@@ -36,10 +36,15 @@ func (c *BuildEventProxyClient) reconnectIfNecessary() {
 	c.client = pepb.NewPublishBuildEventClient(conn)
 }
 
-func NewBuildEventProxyClient(target string) *BuildEventProxyClient {
+func NewBuildEventProxyClient(env environment.Env, target string) *BuildEventProxyClient {
+	bufferSize := 100
+	if configuredBufferSize := env.GetConfigurator().GetBuildEventProxyBufferSize(); configuredBufferSize != 0 {
+		bufferSize = configuredBufferSize
+	}
 	c := &BuildEventProxyClient{
-		target:  target,
-		rootCtx: context.Background(),
+		target:          target,
+		rootCtx:         context.Background(),
+		eventBufferSize: bufferSize,
 	}
 	c.reconnectIfNecessary()
 	return c
@@ -65,7 +70,7 @@ type asyncStreamProxy struct {
 func (c *BuildEventProxyClient) newAsyncStreamProxy(ctx context.Context, opts ...grpc.CallOption) *asyncStreamProxy {
 	asp := &asyncStreamProxy{
 		ctx:    ctx,
-		events: make(chan pepb.PublishBuildToolEventStreamRequest, eventBufferSize),
+		events: make(chan pepb.PublishBuildToolEventStreamRequest, c.eventBufferSize),
 	}
 	// Start a goroutine that will open the stream and pass along events.
 	go func() {

server/config/config.go

@@ -45,7 +45,8 @@ type appConfig struct {
 }
 
 type buildEventProxy struct {
-	Hosts []string `yaml:"hosts" usage:"The list of hosts to pass build events onto."`
+	Hosts      []string `yaml:"hosts" usage:"The list of hosts to pass build events onto."`
+	BufferSize int      `yaml:"buffer_size" usage:"The number of build events to buffer locally when proxying build events."`
 }
 
 type DatabaseConfig struct {
@@ -379,6 +380,10 @@ func (c *Configurator) GetBuildEventProxyHosts() []string {
 	return c.gc.BuildEventProxy.Hosts
 }
 
+func (c *Configurator) GetBuildEventProxyBufferSize() int {
+	return c.gc.BuildEventProxy.BufferSize
+}
+
 func (c *Configurator) GetCacheMaxSizeBytes() int64 {
 	return c.gc.Cache.MaxSizeBytes
 }

server/libmain/libmain.go

@@ -119,7 +119,7 @@ func GetConfiguredEnvironmentOrDie(configurator *config.Configurator, healthChec
 	for _, target := range configurator.GetBuildEventProxyHosts() {
 		// NB: This can block for up to a second on connecting. This would be a
 		// great place to have our health checker and mark these as optional.
-		buildEventProxyClients = append(buildEventProxyClients, build_event_proxy.NewBuildEventProxyClient(target))
+		buildEventProxyClients = append(buildEventProxyClients, build_event_proxy.NewBuildEventProxyClient(realEnv, target))
 		log.Printf("Proxy: forwarding build events to: %s", target)
 	}
 	realEnv.SetBuildEventProxyClients(buildEventProxyClients)