Handle requests with very large payload

pkg/broker/ingress/handler.go

@@ -47,6 +47,9 @@ const (
 	// TODO(liu-cong) configurable timeout
 	decoupleSinkTimeout = 30 * time.Second
 
+	// Limit for request payload in bytes (100Mb)
+	maxRequestBodyBytes = 100000000
+
 	// EventArrivalTime is used to access the metadata stored on a
 	// CloudEvent to measure the time difference between when an event is
 	// received on a broker and before it is dispatched to the trigger function.
@@ -119,7 +122,6 @@ func (h *Handler) ServeHTTP(response nethttp.ResponseWriter, request *nethttp.Re
 		response.WriteHeader(nethttp.StatusOK)
 		return
 	}
-
 	ctx := request.Context()
 	ctx = logging.WithLogger(ctx, h.logger)
 	ctx = tracing.WithLogging(ctx, trace.FromContext(ctx))
@@ -129,6 +131,11 @@ func (h *Handler) ServeHTTP(response nethttp.ResponseWriter, request *nethttp.Re
 		return
 	}
 
+	if request.ContentLength > maxRequestBodyBytes {
+		response.WriteHeader(nethttp.StatusRequestEntityTooLarge)
+		return
+	}
+
 	broker, err := ConvertPathToNamespacedName(request.URL.Path)
 	ctx = logging.With(ctx, zap.Stringer("broker", broker))
 	if err != nil {

pkg/broker/ingress/handler_test.go

@@ -26,6 +26,8 @@ import (
 	"testing"
 	"time"
 
+	"math/rand"
+
 	"cloud.google.com/go/pubsub"
 	"cloud.google.com/go/pubsub/pstest"
 	cepubsub "github.com/cloudevents/sdk-go/protocol/pubsub/v2"
@@ -175,6 +177,13 @@ func TestHandler(t *testing.T) {
 			event:    createTestEvent("test-event"),
 			wantCode: nethttp.StatusMethodNotAllowed,
 		},
+		{
+			name:     "an event with a very large payload",
+			method:   "POST",
+			path:     "/ns1/broker1",
+			event:    createTestEventWithPayloadSize("test-event", 110000000),
+			wantCode: nethttp.StatusRequestEntityTooLarge,
+		},
 		{
 			name:     "malformed path",
 			path:     "/ns1/broker1/and/something/else",
@@ -466,6 +475,14 @@ func createTestEvent(id string) *cloudevents.Event {
 	return &event
 }
 
+func createTestEventWithPayloadSize(id string, payloadSizeBytes int) *cloudevents.Event {
+	testEvent := createTestEvent(id)
+	payload := make([]byte, payloadSizeBytes)
+	rand.Read(payload)
+	testEvent.SetData("application/octet-stream", payload)
+	return testEvent
+}
+
 // createRequest creates an http request from the test case. If event is specified, it converts the event to a request.
 func createRequest(tc testCase, url string) *nethttp.Request {
 	method := "POST"