From dc36dacf0328313abff3fe05d84e1de5fb72e174 Mon Sep 17 00:00:00 2001
From: Alexandra Roatis <aroatis@google.com>
Date: Wed, 30 Sep 2020 16:50:57 -0400
Subject: [PATCH] Report event_count metric in Broker Ingress for invalid
 events

---
 pkg/broker/ingress/handler.go      |  7 +++---
 pkg/broker/ingress/handler_test.go | 38 +++++++++++++++++++++---------
 pkg/metrics/tags.go                |  1 +
 pkg/metrics/tags_test.go           |  3 +++
 4 files changed, 35 insertions(+), 14 deletions(-)

diff --git a/pkg/broker/ingress/handler.go b/pkg/broker/ingress/handler.go
index 1f523e5807..d69d1d6ffa 100644
--- a/pkg/broker/ingress/handler.go
+++ b/pkg/broker/ingress/handler.go
@@ -161,6 +161,7 @@ func (h *Handler) ServeHTTP(response nethttp.ResponseWriter, request *nethttp.Re
 			httpStatus = nethttp.StatusRequestEntityTooLarge
 		}
 		nethttp.Error(response, err.Error(), httpStatus)
+		h.reportMetrics(ctx, "_invalid_cloud_event_", httpStatus)
 		return
 	}
 
@@ -186,7 +187,7 @@ func (h *Handler) ServeHTTP(response nethttp.ResponseWriter, request *nethttp.Re
 	statusCode := nethttp.StatusAccepted
 	ctx, cancel := context.WithTimeout(ctx, decoupleSinkTimeout)
 	defer cancel()
-	defer func() { h.reportMetrics(ctx, event, statusCode) }()
+	defer func() { h.reportMetrics(ctx, event.Type(), statusCode) }()
 	if res := h.decouple.Send(ctx, broker, *event); !cev2.IsACK(res) {
 		logging.FromContext(ctx).Error("Error publishing to PubSub", zap.Error(res))
 		statusCode = nethttp.StatusInternalServerError
@@ -230,9 +231,9 @@ func (h *Handler) toEvent(ctx context.Context, request *nethttp.Request) (*cev2.
 	return event, nil
 }
 
-func (h *Handler) reportMetrics(ctx context.Context, event *cev2.Event, statusCode int) {
+func (h *Handler) reportMetrics(ctx context.Context, eventType string, statusCode int) {
 	args := metrics.IngressReportArgs{
-		EventType:    event.Type(),
+		EventType:    eventType,
 		ResponseCode: statusCode,
 	}
 	if err := h.reporter.ReportEventCount(ctx, args); err != nil {
diff --git a/pkg/broker/ingress/handler_test.go b/pkg/broker/ingress/handler_test.go
index 89b60ec5ba..f552afe02f 100644
--- a/pkg/broker/ingress/handler_test.go
+++ b/pkg/broker/ingress/handler_test.go
@@ -184,13 +184,21 @@ func TestHandler(t *testing.T) {
 			wantCode: nethttp.StatusRequestEntityTooLarge,
 		},
 		{
-			name:          "malicious requests or requests with unknown Content-Length and a very large payload",
-			method:        "POST",
-			path:          "/ns1/broker1",
-			event:         createTestEventWithPayloadSize("test-event", 11000000), // 11Mb
-			contentLength: ptr.Int64(-1),
-			wantCode:      nethttp.StatusRequestEntityTooLarge,
-			timeout:       10 * time.Second,
+			name:           "malicious requests or requests with unknown Content-Length and a very large payload",
+			method:         "POST",
+			path:           "/ns1/broker1",
+			event:          createTestEventWithPayloadSize("test-event", 11000000), // 11Mb
+			contentLength:  ptr.Int64(-1),
+			wantCode:       nethttp.StatusRequestEntityTooLarge,
+			timeout:        10 * time.Second,
+			wantEventCount: 1,
+			wantMetricTags: map[string]string{
+				metricskey.LabelEventType:         "_invalid_cloud_event_",
+				metricskey.LabelResponseCode:      "413",
+				metricskey.LabelResponseCodeClass: "4xx",
+				metricskey.PodName:                pod,
+				metricskey.ContainerName:          container,
+			},
 		},
 		{
 			name:     "malformed path",
@@ -199,10 +207,18 @@ func TestHandler(t *testing.T) {
 			wantCode: nethttp.StatusNotFound,
 		},
 		{
-			name:     "request is not an event",
-			path:     "/ns1/broker1",
-			wantCode: nethttp.StatusBadRequest,
-			header:   nethttp.Header{},
+			name:           "request is not an event",
+			path:           "/ns1/broker1",
+			wantCode:       nethttp.StatusBadRequest,
+			header:         nethttp.Header{},
+			wantEventCount: 1,
+			wantMetricTags: map[string]string{
+				metricskey.LabelEventType:         "_invalid_cloud_event_",
+				metricskey.LabelResponseCode:      "400",
+				metricskey.LabelResponseCodeClass: "4xx",
+				metricskey.PodName:                pod,
+				metricskey.ContainerName:          container,
+			},
 		},
 		{
 			name:           "wrong path - broker doesn't exist in given namespace",
diff --git a/pkg/metrics/tags.go b/pkg/metrics/tags.go
index d3464126d1..e8e9f998bd 100644
--- a/pkg/metrics/tags.go
+++ b/pkg/metrics/tags.go
@@ -58,6 +58,7 @@ var (
 	allowedEventTypes = map[string]struct{}{
 		"e2e-dummy-event-type":              {},
 		"e2e-testing-resp-event-type-dummy": {},
+		"_invalid_cloud_event_":             {},
 	}
 )
 
diff --git a/pkg/metrics/tags_test.go b/pkg/metrics/tags_test.go
index 70f0364cc0..e1cf8a67e9 100644
--- a/pkg/metrics/tags_test.go
+++ b/pkg/metrics/tags_test.go
@@ -39,6 +39,9 @@ func TestEventTypeMetricValue(t *testing.T) {
 		{"some.custom.event", "custom"},
 		{"", "custom"},
 
+		// Used to mark invalid cloud events
+		{"_invalid_cloud_event_", "_invalid_cloud_event_"},
+
 		// Used in E2E tests
 		{"e2e-dummy-event-type", "e2e-dummy-event-type"},
 		{"e2e-testing-resp-event-type-dummy", "e2e-testing-resp-event-type-dummy"},