Limit the values of event_type in Broker metrics

pkg/broker/ingress/handler_test.go

@@ -59,7 +59,7 @@ const (
 
 	traceID = "4bf92f3577b34da6a3ce929d0e0e4736"
 
-	eventType = "test-event-type"
+	eventType = "com.google.cloud.pubsub.topic.publish"
 	pod       = "testpod"
 	container = "testcontainer"
 )

pkg/metrics/delivery_reporter.go

@@ -155,11 +155,11 @@ func (r *DeliveryReporter) reportEventProcessingTime(ctx context.Context, end ti
 }
 
 func filterTypeValue(v string) string {
-	if v != "" {
-		return v
+	if v == "" {
+		// the default value if the filter attributes are empty.
+		return "any"
 	}
-	// the default value if the filter attributes are empty.
-	return "any"
+	return EventTypeMetricValue(v)
 }
 
 func (r *DeliveryReporter) AddTags(ctx context.Context) (context.Context, error) {

pkg/metrics/delivery_reporter_test.go

@@ -36,7 +36,7 @@ func TestReportEventDispatchTime(t *testing.T) {
 		metricskey.LabelNamespaceName:     "testns",
 		metricskey.LabelBrokerName:        "testbroker",
 		metricskey.LabelTriggerName:       "testtrigger",
-		metricskey.LabelFilterType:        "testeventtype",
+		metricskey.LabelFilterType:        "com.google.cloud.pubsub.topic.publish",
 		metricskey.LabelResponseCode:      "202",
 		metricskey.LabelResponseCodeClass: "2xx",
 		metricskey.PodName:                "testpod",
@@ -57,7 +57,7 @@ func TestReportEventDispatchTime(t *testing.T) {
 		Broker:    "testbroker",
 		Name:      "testtrigger",
 		FilterAttributes: map[string]string{
-			"type": "testeventtype",
+			"type": "com.google.cloud.pubsub.topic.publish",
 		},
 	})
 	if err != nil {
@@ -82,7 +82,7 @@ func TestReportEventProcessingTime(t *testing.T) {
 		metricskey.LabelNamespaceName: "testns",
 		metricskey.LabelBrokerName:    "testbroker",
 		metricskey.LabelTriggerName:   "testtrigger",
-		metricskey.LabelFilterType:    "testeventtype",
+		metricskey.LabelFilterType:    "com.google.cloud.pubsub.topic.publish",
 		metricskey.PodName:            "testpod",
 		metricskey.ContainerName:      "testcontainer",
 	}
@@ -100,7 +100,7 @@ func TestReportEventProcessingTime(t *testing.T) {
 		Broker:    "testbroker",
 		Name:      "testtrigger",
 		FilterAttributes: map[string]string{
-			"type": "testeventtype",
+			"type": "com.google.cloud.pubsub.topic.publish",
 		},
 	})
 	if err != nil {
@@ -124,6 +124,47 @@ func TestReportEventProcessingTime(t *testing.T) {
 	metricstest.CheckDistributionData(t, "event_processing_latencies", wantTags, 2, 1100.0, 9100.0)
 }
 
+func TestMetricsWithCustomFilterType(t *testing.T) {
+	reportertest.ResetDeliveryMetrics()
+
+	wantTags := map[string]string{
+		metricskey.LabelNamespaceName:     "testns",
+		metricskey.LabelBrokerName:        "testbroker",
+		metricskey.LabelTriggerName:       "testtrigger",
+		metricskey.LabelFilterType:        "custom", // Expects this to be "custom" instead of filter type
+		metricskey.LabelResponseCode:      "202",
+		metricskey.LabelResponseCodeClass: "2xx",
+		metricskey.PodName:                "testpod",
+		metricskey.ContainerName:          "testcontainer",
+	}
+
+	r, err := NewDeliveryReporter("testpod", "testcontainer")
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, err := r.AddTags(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, err = AddTargetTags(ctx, &config.Target{
+		Namespace: "testns",
+		Broker:    "testbroker",
+		Name:      "testtrigger",
+		FilterAttributes: map[string]string{
+			"type": "testeventtype",
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	reportertest.ExpectMetrics(t, func() error {
+		r.ReportEventDispatchTime(ctx, 1100*time.Millisecond, 202)
+		return nil
+	})
+	metricstest.CheckCountData(t, "event_count", wantTags, 1)
+}
+
 func TestMetricsWithEmptySourceAndTypeFilter(t *testing.T) {
 	reportertest.ResetDeliveryMetrics()
 

pkg/metrics/ingress_reporter.go

@@ -93,7 +93,7 @@ func (r *IngressReporter) ReportEventCount(ctx context.Context, args IngressRepo
 		tag.Insert(ContainerNameKey, string(r.containerName)),
 		tag.Insert(NamespaceNameKey, args.Namespace),
 		tag.Insert(BrokerNameKey, args.Broker),
-		tag.Insert(EventTypeKey, args.EventType),
+		tag.Insert(EventTypeKey, EventTypeMetricValue(args.EventType)),
 		tag.Insert(ResponseCodeKey, strconv.Itoa(args.ResponseCode)),
 		tag.Insert(ResponseCodeClassKey, metrics.ResponseCodeClass(args.ResponseCode)),
 	)

pkg/metrics/ingress_reporter_test.go

@@ -30,6 +30,40 @@ import (
 func TestStatsReporter(t *testing.T) {
 	reportertest.ResetIngressMetrics()
 
+	args := IngressReportArgs{
+		Namespace:    "testns",
+		Broker:       "testbroker",
+		EventType:    "com.google.cloud.scheduler.job.execute",
+		ResponseCode: 202,
+	}
+	wantTags := map[string]string{
+		metricskey.LabelNamespaceName:     "testns",
+		metricskey.LabelBrokerName:        "testbroker",
+		metricskey.LabelEventType:         "com.google.cloud.scheduler.job.execute",
+		metricskey.LabelResponseCode:      "202",
+		metricskey.LabelResponseCodeClass: "2xx",
+		metricskey.ContainerName:          "testcontainer",
+		metricskey.PodName:                "testpod",
+	}
+
+	r, err := NewIngressReporter(PodName("testpod"), ContainerName("testcontainer"))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// test ReportEventCount
+	reportertest.ExpectMetrics(t, func() error {
+		return r.ReportEventCount(context.Background(), args)
+	})
+	reportertest.ExpectMetrics(t, func() error {
+		return r.ReportEventCount(context.Background(), args)
+	})
+	metricstest.CheckCountData(t, "event_count", wantTags, 2)
+}
+
+func TestStatsReporterWithCustomEventType(t *testing.T) {
+	reportertest.ResetIngressMetrics()
+
 	args := IngressReportArgs{
 		Namespace:    "testns",
 		Broker:       "testbroker",
@@ -39,7 +73,7 @@ func TestStatsReporter(t *testing.T) {
 	wantTags := map[string]string{
 		metricskey.LabelNamespaceName:     "testns",
 		metricskey.LabelBrokerName:        "testbroker",
-		metricskey.LabelEventType:         "testeventtype",
+		metricskey.LabelEventType:         "custom",
 		metricskey.LabelResponseCode:      "202",
 		metricskey.LabelResponseCodeClass: "2xx",
 		metricskey.ContainerName:          "testcontainer",

pkg/metrics/tags.go

@@ -17,6 +17,9 @@ limitations under the License.
 package metrics
 
 import (
+	"strings"
+
+	"github.com/google/knative-gcp/test/e2e/lib"
 	"go.opencensus.io/tag"
 	"knative.dev/pkg/metrics/metricskey"
 )
@@ -43,3 +46,27 @@ var (
 	PodNameKey       = tag.MustNewKey(metricskey.PodName)
 	ContainerNameKey = tag.MustNewKey(metricskey.ContainerName)
 )
+
+var (
+	allowedEventTypes = []string{
+		lib.E2EDummyEventType,
+		lib.E2EDummyRespEventType,
+	}
+)
+
+// Stackdriver has a limit on the cardinality of fields on a metric, and event types can be any custom string. We're
+// reducing the allowable values on this field to be limited to GCP event types and defaulting everything else to
+// "custom".
+func EventTypeMetricValue(eventType string) string {
+	if strings.HasPrefix(eventType, "com.google.cloud") {
+		return eventType
+	}
+
+	for _, allowed := range allowedEventTypes {
+		if eventType == allowed {
+			return eventType
+		}
+	}
+
+	return "custom"
+}

pkg/metrics/tags_test.go

@@ -0,0 +1,52 @@
+/*
+Copyright 2020 Google LLC
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package metrics
+
+import (
+	"testing"
+)
+
+func TestEventTypeMetricValue(t *testing.T) {
+	tests := []struct {
+		event string
+		want  string
+	}{
+		// Current Supported GCP event types
+		{"com.google.cloud.auditlog.event", "com.google.cloud.auditlog.event"},
+		{"com.google.cloud.pubsub.topic.publish", "com.google.cloud.pubsub.topic.publish"},
+		{"com.google.cloud.scheduler.job.execute", "com.google.cloud.scheduler.job.execute"},
+		{"com.google.cloud.storage.object.archive", "com.google.cloud.storage.object.archive"},
+		{"com.google.cloud.storage.object.delete", "com.google.cloud.storage.object.delete"},
+		{"com.google.cloud.storage.object.finalize", "com.google.cloud.storage.object.finalize"},
+		{"com.google.cloud.storage.object.metadataUpdate", "com.google.cloud.storage.object.metadataUpdate"},
+
+		{"com.google.cloud.some.newly.added.event", "com.google.cloud.some.newly.added.event"},
+
+		{"some.custom.event", "custom"},
+		{"", "custom"},
+
+		// Used in E2E tests
+		{"e2e-dummy-event-type", "e2e-dummy-event-type"},
+		{"e2e-testing-resp-event-type-dummy", "e2e-testing-resp-event-type-dummy"},
+	}
+
+	for _, test := range tests {
+		if got := EventTypeMetricValue(test.event); got != test.want {
+			t.Errorf("EventTypeMetricValue(%q) = %v; want %v", test.event, got, test.want)
+		}
+	}
+}