google · knative-prow-robot · Sep 2, 2020 · Aug 31, 2020 · Sep 1, 2020 · Sep 2, 2020
diff --git a/cmd/broker/ingress/main.go b/cmd/broker/ingress/main.go
@@ -24,13 +24,17 @@ import (
 	"github.com/google/knative-gcp/pkg/utils/clients"
 	"github.com/google/knative-gcp/pkg/utils/mainhelper"
 
+	"cloud.google.com/go/pubsub"
 	"go.uber.org/zap"
 )
 
 type envConfig struct {
 	PodName   string `envconfig:"POD_NAME" required:"true"`
 	Port      int    `envconfig:"PORT" default:"8080"`
 	ProjectID string `envconfig:"PROJECT_ID"`
+
+	// Default 300Mi.
+	PublishBufferedByteLimit int `envconfig:"PUBLISH_BUFFERED_BYTES_LIMIT" default:"314572800"`
 }
 
 const (
@@ -63,6 +67,7 @@ func main() {
 		clients.ProjectID(projectID),
 		metrics.PodName(env.PodName),
 		metrics.ContainerName(component),
+		publishSetting(env),
 	)
 	if err != nil {
 		logger.Desugar().Fatal("Unable to create ingress handler: ", zap.Error(err))
@@ -73,3 +78,11 @@ func main() {
 		logger.Desugar().Fatal("failed to start ingress: ", zap.Error(err))
 	}
 }
+
+func publishSetting(env envConfig) pubsub.PublishSettings {
+	s := pubsub.DefaultPublishSettings
+	if env.PublishBufferedByteLimit > 0 {
+		s.BufferedByteLimit = env.PublishBufferedByteLimit
+	}
+	return s
+}
diff --git a/cmd/broker/ingress/wire.go b/cmd/broker/ingress/wire.go
@@ -21,6 +21,7 @@ package main
 import (
 	"context"
 
+	"cloud.google.com/go/pubsub"
 	"github.com/google/knative-gcp/pkg/broker/config/volume"
 	"github.com/google/knative-gcp/pkg/broker/ingress"
 	"github.com/google/knative-gcp/pkg/metrics"
@@ -34,6 +35,7 @@ func InitializeHandler(
 	projectID clients.ProjectID,
 	podName metrics.PodName,
 	containerName metrics.ContainerName,
+	publishSettings pubsub.PublishSettings,
 ) (*ingress.Handler, error) {
 	panic(wire.Build(
 		ingress.HandlerSet,

diff --git a/cmd/broker/ingress/wire_gen.go b/cmd/broker/ingress/wire_gen.go
diff --git a/pkg/apis/intevents/v1alpha1/brokercell_defaults.go b/pkg/apis/intevents/v1alpha1/brokercell_defaults.go
@@ -32,19 +32,19 @@ const (
 	// usage, HPA could have enough time to kick in.
 	// See: https://github.com/google/knative-gcp/issues/1265
 	avgMemoryUsageFanout  string = "1500Mi"
-	avgMemoryUsageIngress string = "700Mi"
+	avgMemoryUsageIngress string = "1500Mi"
 	avgMemoryUsageRetry   string = "1500Mi"
 	cpuRequestFanout      string = "1500m"
-	cpuRequestIngress     string = "1000m"
+	cpuRequestIngress     string = "2000m"
 	cpuRequestRetry       string = "1000m"
 	cpuLimitFanout        string = ""
 	cpuLimitIngress       string = ""
 	cpuLimitRetry         string = ""
 	memoryRequestFanout   string = "500Mi"
-	memoryRequestIngress  string = "500Mi"
+	memoryRequestIngress  string = "2000Mi"
 	memoryRequestRetry    string = "500Mi"
 	memoryLimitFanout     string = "3000Mi"
-	memoryLimitIngress    string = "1000Mi"
+	memoryLimitIngress    string = "2000Mi"
 	memoryLimitRetry      string = "3000Mi"
 	minReplicas           int32  = 1
 	maxReplicas           int32  = 10

diff --git a/pkg/broker/ingress/handler_test.go b/pkg/broker/ingress/handler_test.go
@@ -334,7 +334,7 @@ func runIngressHandlerBenchmark(b *testing.B, eventSize int) {
 	defer psSrv.Close()
 
 	psClient := createPubsubClient(ctx, b, psSrv)
-	decouple := NewMultiTopicDecoupleSink(ctx, memory.NewTargets(brokerConfig), psClient)
+	decouple := NewMultiTopicDecoupleSink(ctx, memory.NewTargets(brokerConfig), psClient, pubsub.DefaultPublishSettings)
 	statsReporter, err := metrics.NewIngressReporter(metrics.PodName(pod), metrics.ContainerName(container))
 	if err != nil {
 		b.Fatal(err)
@@ -405,7 +405,7 @@ func setupTestReceiver(ctx context.Context, t testing.TB, psSrv *pstest.Server)
 
 // createAndStartIngress creates an ingress and calls its Start() method in a goroutine.
 func createAndStartIngress(ctx context.Context, t testing.TB, psSrv *pstest.Server) string {
-	decouple := NewMultiTopicDecoupleSink(ctx, memory.NewTargets(brokerConfig), createPubsubClient(ctx, t, psSrv))
+	decouple := NewMultiTopicDecoupleSink(ctx, memory.NewTargets(brokerConfig), createPubsubClient(ctx, t, psSrv), pubsub.DefaultPublishSettings)
 
 	receiver := &testHttpMessageReceiver{urlCh: make(chan string)}
 	statsReporter, err := metrics.NewIngressReporter(metrics.PodName(pod), metrics.ContainerName(container))

diff --git a/pkg/broker/ingress/multi_topic_decouple_sink.go b/pkg/broker/ingress/multi_topic_decouple_sink.go
@@ -38,11 +38,17 @@ import (
 const projectEnvKey = "PROJECT_ID"
 
 // NewMultiTopicDecoupleSink creates a new multiTopicDecoupleSink.
-func NewMultiTopicDecoupleSink(ctx context.Context, brokerConfig config.ReadonlyTargets, client *pubsub.Client) *multiTopicDecoupleSink {
+func NewMultiTopicDecoupleSink(
+	ctx context.Context,
+	brokerConfig config.ReadonlyTargets,
+	client *pubsub.Client,
+	publishSettings pubsub.PublishSettings) *multiTopicDecoupleSink {
+
 	return &multiTopicDecoupleSink{
-		logger:       logging.FromContext(ctx),
-		pubsub:       client,
-		brokerConfig: brokerConfig,
+		logger:          logging.FromContext(ctx),
+		pubsub:          client,
+		publishSettings: publishSettings,
+		brokerConfig:    brokerConfig,
 		// TODO(#1118): remove Topic when broker config is removed
 		topics: make(map[types.NamespacedName]*pubsub.Topic),
 	}
@@ -52,7 +58,8 @@ func NewMultiTopicDecoupleSink(ctx context.Context, brokerConfig config.Readonly
 // to the broker to which the events are sent.
 type multiTopicDecoupleSink struct {
 	// pubsub talks to pubsub.
-	pubsub *pubsub.Client
+	pubsub          *pubsub.Client
+	publishSettings pubsub.PublishSettings
 	// map from brokers to topics
 	topics    map[types.NamespacedName]*pubsub.Topic
 	topicsMut sync.RWMutex

diff --git a/pkg/broker/ingress/multi_topic_decouple_sink_test.go b/pkg/broker/ingress/multi_topic_decouple_sink_test.go
@@ -183,7 +183,7 @@ func TestMultiTopicDecoupleSink(t *testing.T) {
 					t.Fatal(err)
 				}
 
-				sink := NewMultiTopicDecoupleSink(ctx, brokerConfig, psClient)
+				sink := NewMultiTopicDecoupleSink(ctx, brokerConfig, psClient, pubsub.DefaultPublishSettings)
 				// Send events
 				event := createTestEvent(uuid.New().String())
 				err = sink.Send(context.Background(), testCase.broker, *event)

diff --git a/pkg/reconciler/brokercell/resources/deployments.go b/pkg/reconciler/brokercell/resources/deployments.go
@@ -43,7 +43,7 @@ func MakeIngressDeployment(args IngressArgs) *appsv1.Deployment {
 				Scheme: corev1.URISchemeHTTP,
 			},
 		},
-		FailureThreshold: 3,
+		FailureThreshold: 5,
 		PeriodSeconds:    2,
 		SuccessThreshold: 1,
 		TimeoutSeconds:   5,
@@ -56,7 +56,7 @@ func MakeIngressDeployment(args IngressArgs) *appsv1.Deployment {
 				Scheme: corev1.URISchemeHTTP,
 			},
 		},
-		FailureThreshold:    3,
+		FailureThreshold:    5,
 		InitialDelaySeconds: 5,
 		PeriodSeconds:       2,
 		SuccessThreshold:    1,

diff --git a/pkg/reconciler/brokercell/testingdata/ingress_deployment.yaml b/pkg/reconciler/brokercell/testingdata/ingress_deployment.yaml
@@ -43,7 +43,7 @@ spec:
       - name: ingress
         image: ingress
         livenessProbe:
-          failureThreshold: 3
+          failureThreshold: 5
           httpGet:
             path: /healthz
             port: 8080
@@ -53,7 +53,7 @@ spec:
           successThreshold: 1
           timeoutSeconds: 5
         readinessProbe:
-          failureThreshold: 3
+          failureThreshold: 5
           httpGet:
             path: /healthz
             port: 8080
@@ -85,10 +85,10 @@ spec:
           mountPath: /var/secrets/google          
         resources:
           limits:
-            memory: 1000Mi
+            memory: 2000Mi
           requests:
-            cpu: 1000m
-            memory: 500Mi
+            cpu: 2000m
+            memory: 2000Mi
         ports:
         - name: metrics
           containerPort: 9090

diff --git a/pkg/reconciler/brokercell/testingdata/ingress_deployment_with_status.yaml b/pkg/reconciler/brokercell/testingdata/ingress_deployment_with_status.yaml
@@ -44,7 +44,7 @@ spec:
       - name: ingress
         image: ingress
         livenessProbe:
-          failureThreshold: 3
+          failureThreshold: 5
           httpGet:
             path: /healthz
             port: 8080
@@ -54,7 +54,7 @@ spec:
           successThreshold: 1
           timeoutSeconds: 5
         readinessProbe:
-          failureThreshold: 3
+          failureThreshold: 5
           httpGet:
             path: /healthz
             port: 8080
@@ -86,10 +86,10 @@ spec:
           mountPath: /var/secrets/google          
         resources:
           limits:
-            memory: 1000Mi
+            memory: 2000Mi
           requests:
-            cpu: 1000m
-            memory: 500Mi
+            cpu: 2000m
+            memory: 2000Mi
         ports:
         - name: metrics
           containerPort: 9090

diff --git a/pkg/reconciler/brokercell/testingdata/ingress_hpa.yaml b/pkg/reconciler/brokercell/testingdata/ingress_hpa.yaml
@@ -44,4 +44,4 @@ spec:
       name: memory
       target:
         type: AverageValue
-        averageValue: 700Mi
+        averageValue: 1500Mi