Report event_count metric in Broker Ingress for invalid events

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 {

pkg/broker/ingress/handler_test.go

@@ -191,6 +191,14 @@ func TestHandler(t *testing.T) {
 			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",
@@ -203,6 +211,14 @@ func TestHandler(t *testing.T) {
 			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",

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_": {},
 	}
 )
 

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"},