[Collector e2e] Capture ocgrpc self observability metrics

exporter/collector/internal/integrationtest/diff.go

@@ -15,6 +15,8 @@
 package integrationtest
 
 import (
+	"testing"
+
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
 	"google.golang.org/protobuf/proto"
@@ -29,11 +31,11 @@ var (
 )
 
 // Diff uses cmp.Diff(), protocmp, and some custom options to compare two protobuf messages.
-func DiffProtos(x, y *MetricExpectFixture) string {
+func DiffProtos(t testing.TB, x, y *MetricExpectFixture) string {
 	x = proto.Clone(x).(*MetricExpectFixture)
 	y = proto.Clone(y).(*MetricExpectFixture)
-	normalizeFixture(x)
-	normalizeFixture(y)
+	normalizeFixture(t, x)
+	normalizeFixture(t, y)
 
 	return cmp.Diff(x, y, cmpOptions...)
 }

exporter/collector/internal/integrationtest/inmemoryocexporter.go

@@ -38,8 +38,7 @@ type InMemoryOCExporter struct {
 func getViews() []*view.View {
 	views := []*view.View{}
 	views = append(views, collector.MetricViews()...)
-	// TODO: enable this
-	// views = append(views, ocgrpc.DefaultClientViews...)
+	views = append(views, ocgrpc.DefaultClientViews...)
 	return views
 }
 
@@ -76,9 +75,6 @@ func (i *InMemoryOCExporter) Shutdown(ctx context.Context) error {
 	i.testServer.Shutdown()
 
 	view.Unregister(getViews()...)
-	// TODO: remove this special case
-	view.Unregister(ocgrpc.DefaultClientViews...)
-
 	return err
 }
 
@@ -90,9 +86,6 @@ func NewInMemoryOCViewExporter() (*InMemoryOCExporter, error) {
 	view.Unregister(views...)
 	view.Register(views...)
 
-	// TODO: remove this special case
-	view.Unregister(ocgrpc.DefaultClientViews...)
-
 	testServer, err := NewMetricTestServer()
 	if err != nil {
 		return nil, err

exporter/collector/internal/integrationtest/metrics_test.go

@@ -60,6 +60,7 @@ func TestMetrics(t *testing.T) {
 			selfObsMetrics, err := inMemoryOCExporter.Proto(ctx)
 			require.NoError(t, err)
 			diff := DiffProtos(
+				t,
 				&MetricExpectFixture{
 					CreateTimeSeriesRequests:        testServer.CreateTimeSeriesRequests(),
 					CreateMetricDescriptorRequests:  testServer.CreateMetricDescriptorRequests(),

exporter/collector/internal/integrationtest/testcase.go

@@ -17,11 +17,16 @@ package integrationtest
 import (
 	"bytes"
 	"encoding/json"
+	"fmt"
 	"io/ioutil"
 	"os"
+	"sort"
 	"testing"
 	"time"
 
+	distributionpb "google.golang.org/genproto/googleapis/api/distribution"
+	monitoringpb "google.golang.org/genproto/googleapis/monitoring/v3"
+
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/model/otlp"
 	"go.opentelemetry.io/collector/model/pdata"
@@ -31,6 +36,17 @@ import (
 	"github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/collector"
 )
 
+var (
+	// selfObsMetricsToNormalize is the set of self-observability metrics which may not record
+	// the same value every time due to side effects. The values of these metrics get cleared
+	// and are not checked in the fixture. Their labels and types are still checked.
+	selfObsMetricsToNormalize = map[string]struct{}{
+		"custom.googleapis.com/opencensus/grpc.io/client/roundtrip_latency":      {},
+		"custom.googleapis.com/opencensus/grpc.io/client/sent_bytes_per_rpc":     {},
+		"custom.googleapis.com/opencensus/grpc.io/client/received_bytes_per_rpc": {},
+	}
+)
+
 type MetricsTestCase struct {
 	// Name of the test case
 	Name string
@@ -156,7 +172,7 @@ func (m *MetricsTestCase) SaveRecordedFixtures(
 	t testing.TB,
 	fixture *MetricExpectFixture,
 ) {
-	normalizeFixture(fixture)
+	normalizeFixture(t, fixture)
 
 	jsonBytes, err := protojson.Marshal(fixture)
 	require.NoError(t, err)
@@ -169,26 +185,20 @@ func (m *MetricsTestCase) SaveRecordedFixtures(
 
 // Normalizes timestamps and removes project ID fields which create noise in the fixture
 // because they can vary each test run
-func normalizeFixture(fixture *MetricExpectFixture) {
-	timeSeriesReqs := append(
-		fixture.GetCreateTimeSeriesRequests(),
-		fixture.GetCreateServiceTimeSeriesRequests()...,
-	)
-	timeSeriesReqs = append(
-		timeSeriesReqs,
-		fixture.GetSelfObservabilityMetrics().GetCreateTimeSeriesRequests()...,
-	)
-	metricDescriptorReqs := append(
-		fixture.GetCreateMetricDescriptorRequests(),
-		fixture.GetSelfObservabilityMetrics().GetCreateMetricDescriptorRequests()...,
-	)
+func normalizeFixture(t testing.TB, fixture *MetricExpectFixture) {
+	normalizeTimeSeriesReqs(t, fixture.CreateTimeSeriesRequests...)
+	normalizeTimeSeriesReqs(t, fixture.CreateServiceTimeSeriesRequests...)
+	normalizeMetricDescriptorReqs(t, fixture.CreateMetricDescriptorRequests...)
+	normalizeSelfObs(t, fixture.SelfObservabilityMetrics)
+}
 
-	for _, req := range timeSeriesReqs {
+func normalizeTimeSeriesReqs(t testing.TB, reqs ...*monitoringpb.CreateTimeSeriesRequest) {
+	for _, req := range reqs {
 		// clear project ID
 		req.Name = ""
 
-		for _, ts := range req.GetTimeSeries() {
-			for _, p := range ts.GetPoints() {
+		for _, ts := range req.TimeSeries {
+			for _, p := range ts.Points {
 				// Normalize timestamps if they are set
 				if p.GetInterval().GetStartTime() != nil {
 					p.GetInterval().StartTime = &timestamppb.Timestamp{}
@@ -202,13 +212,61 @@ func normalizeFixture(fixture *MetricExpectFixture) {
 			delete(ts.GetResource().GetLabels(), "project_id")
 		}
 	}
+}
 
-	for _, req := range metricDescriptorReqs {
+func normalizeMetricDescriptorReqs(t testing.TB, reqs ...*monitoringpb.CreateMetricDescriptorRequest) {
+	for _, req := range reqs {
 		req.Name = ""
-		if md := req.GetMetricDescriptor(); md != nil {
-			md.Name = ""
+		if req.MetricDescriptor != nil {
+			req.MetricDescriptor.Name = ""
+		}
+	}
+}
+
+func normalizeSelfObs(t testing.TB, selfObs *SelfObservabilityMetric) {
+	for _, req := range selfObs.CreateTimeSeriesRequests {
+		normalizeTimeSeriesReqs(t, req)
+		tss := req.TimeSeries
+		for _, ts := range tss {
+			if _, ok := selfObsMetricsToNormalize[ts.Metric.Type]; ok {
+				// zero out the specific value type
+				switch value := ts.Points[0].Value.Value.(type) {
+				case *monitoringpb.TypedValue_Int64Value:
+					value.Int64Value = 0
+				case *monitoringpb.TypedValue_DoubleValue:
+					value.DoubleValue = 0
+				case *monitoringpb.TypedValue_DistributionValue:
+					// Only preserve the bucket options and zeroed out counts
+					for i := range value.DistributionValue.BucketCounts {
+						value.DistributionValue.BucketCounts[i] = 0
+					}
+					value.DistributionValue = &distributionpb.Distribution{
+						BucketOptions: value.DistributionValue.BucketOptions,
+						BucketCounts:  value.DistributionValue.BucketCounts,
+					}
+				default:
+					t.Logf("Do not know how to normalize typed value type %T", value)
+				}
+			}
 		}
+		// sort time series by (type, labelset)
+		sort.Slice(tss, func(i, j int) bool {
+			iMetric := tss[i].Metric
+			jMetric := tss[j].Metric
+			if iMetric.Type == jMetric.Type {
+				// Doesn't need to sorted correctly, just consistently
+				return fmt.Sprint(iMetric.Labels) < fmt.Sprint(jMetric.Labels)
+			}
+			return iMetric.Type < jMetric.Type
+		})
 	}
+
+	normalizeMetricDescriptorReqs(t, selfObs.CreateMetricDescriptorRequests...)
+	// sort descriptors by type
+	sort.Slice(selfObs.CreateMetricDescriptorRequests, func(i, j int) bool {
+		return selfObs.CreateMetricDescriptorRequests[i].MetricDescriptor.Type <
+			selfObs.CreateMetricDescriptorRequests[j].MetricDescriptor.Type
+	})
 }
 
 func (m *MetricsTestCase) SkipIfNeeded(t testing.TB) {