Use OC stackdriver exporter to capture self observability metrics as …

…GCM protos

exporter/collector/internal/integrationtest/cmd/recordfixtures/main.go

@@ -71,11 +71,13 @@ func main() {
 			require.NoError(t, testServerExporter.PushMetrics(ctx, metrics), "failed to export metrics to local test server")
 			require.NoError(t, testServerExporter.Shutdown(ctx))
 
+			selfObsMetrics, err := inMemoryOCExporter.Proto(ctx)
+			require.NoError(t, err)
 			fixture := &integrationtest.MetricExpectFixture{
 				CreateMetricDescriptorRequests:  testServer.CreateMetricDescriptorRequests(),
 				CreateTimeSeriesRequests:        testServer.CreateTimeSeriesRequests(),
 				CreateServiceTimeSeriesRequests: testServer.CreateServiceTimeSeriesRequests(),
-				SelfObservabilityMetrics:        inMemoryOCExporter.Proto(),
+				SelfObservabilityMetrics:        selfObsMetrics,
 			}
 			test.SaveRecordedFixtures(t, fixture)
 		}()

exporter/collector/internal/integrationtest/config_test.go → exporter/collector/internal/integrationtest/config/config_test.go

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package integrationtest
+package config
 
 import (
 	"path"
@@ -39,7 +39,7 @@ func TestLoadConfig(t *testing.T) {
 
 	factory := newFactory()
 	factories.Exporters[typeStr] = factory
-	cfg, err := configtest.LoadConfigAndValidate(path.Join(".", "testdata", "config.yaml"), factories)
+	cfg, err := configtest.LoadConfigAndValidate(path.Join("..", "testdata", "config.yaml"), factories)
 
 	require.NoError(t, err)
 	require.NotNil(t, cfg)

exporter/collector/internal/integrationtest/fixtures.proto

@@ -22,14 +22,10 @@ message MetricExpectFixture {
   repeated google.monitoring.v3.CreateTimeSeriesRequest create_time_series_requests = 1;
   repeated google.monitoring.v3.CreateMetricDescriptorRequest create_metric_descriptor_requests = 2;
   repeated google.monitoring.v3.CreateTimeSeriesRequest create_service_time_series_requests = 3;
-  repeated SelfObservabilityMetric self_observability_metrics = 4;
+  SelfObservabilityMetric self_observability_metrics = 4;
 }
 
 message SelfObservabilityMetric {
-  // metric name
-  string name = 1;
-  // stringified value
-  string val = 2;
-  // stringified labels
-  map<string, string> labels = 3;
+  repeated google.monitoring.v3.CreateTimeSeriesRequest create_time_series_requests = 1;
+  repeated google.monitoring.v3.CreateMetricDescriptorRequest create_metric_descriptor_requests = 2;
 }

exporter/collector/internal/integrationtest/go.mod

@@ -3,13 +3,15 @@ module github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/colle
 go 1.16
 
 require (
+	contrib.go.opencensus.io/exporter/stackdriver v0.13.10
 	github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/collector v0.0.0-20220111155622-771af0772772
 	github.com/google/go-cmp v0.5.6
 	github.com/stretchr/testify v1.7.0
 	go.opencensus.io v0.23.0
 	go.opentelemetry.io/collector v0.40.0
 	go.opentelemetry.io/collector/model v0.40.0
 	go.uber.org/zap v1.19.1
+	google.golang.org/api v0.60.0
 	google.golang.org/genproto v0.0.0-20211021150943-2b146023228c
 	google.golang.org/grpc v1.42.0
 	google.golang.org/protobuf v1.27.1

exporter/collector/internal/integrationtest/inmemoryocexporter.go

@@ -16,85 +16,108 @@ package integrationtest
 
 import (
 	"context"
-	"fmt"
-	"sort"
 	"time"
 
-	"go.opencensus.io/metric/metricdata"
+	"contrib.go.opencensus.io/exporter/stackdriver"
 	"go.opencensus.io/metric/metricexport"
 	"go.opencensus.io/plugin/ocgrpc"
 	"go.opencensus.io/stats/view"
+	"google.golang.org/api/option"
+	"google.golang.org/grpc"
 
 	"github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/collector"
 )
 
-var _ metricexport.Exporter = (*InMemoryOCExporter)(nil)
-
 // OC stats/metrics exporter used to capture self observability metrics
 type InMemoryOCExporter struct {
-	c      chan []*metricdata.Metric
-	reader *metricexport.Reader
+	testServer          *MetricsTestServer
+	reader              *metricexport.Reader
+	stackdriverExporter *stackdriver.Exporter
 }
 
-func (i *InMemoryOCExporter) ExportMetrics(ctx context.Context, data []*metricdata.Metric) error {
-	i.c <- data
-	return nil
+func getViews() []*view.View {
+	views := []*view.View{}
+	views = append(views, collector.MetricViews()...)
+	// TODO: enable this
+	// views = append(views, ocgrpc.DefaultClientViews...)
+	return views
 }
 
-func (i *InMemoryOCExporter) Proto() []*SelfObservabilityMetric {
+func (i *InMemoryOCExporter) Proto(ctx context.Context) (*SelfObservabilityMetric, error) {
 	// Hack to flush stats, see https://tinyurl.com/5hfcxzk2
 	view.SetReportingPeriod(time.Minute)
-	i.reader.ReadAndExport(i)
-	var data []*metricdata.Metric
+	i.reader.ReadAndExport(i.stackdriverExporter)
+	ctx, cancel := context.WithTimeout(ctx, time.Millisecond*500)
+	defer cancel()
+
+	done := make(chan struct{}, 1)
+	go func() {
+		i.stackdriverExporter.Flush()
+		close(done)
+	}()
 
 	select {
-	case <-time.NewTimer(time.Millisecond * 100).C:
-		return nil
-	case data = <-i.c:
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	case <-done:
 	}
 
-	selfObsMetrics := []*SelfObservabilityMetric{}
-	for _, d := range data {
-		for _, ts := range d.TimeSeries {
-			labels := make(map[string]string, len(d.Descriptor.LabelKeys))
-			for i := 0; i < len(d.Descriptor.LabelKeys); i++ {
-				labels[d.Descriptor.LabelKeys[i].Key] = ts.LabelValues[i].Value
-			}
-
-			for _, p := range ts.Points {
-				selfObsMetrics = append(selfObsMetrics, &SelfObservabilityMetric{
-					Name:   d.Descriptor.Name,
-					Val:    fmt.Sprint(p.Value),
-					Labels: labels,
-				})
-			}
-		}
-	}
-	sort.Slice(selfObsMetrics, func(i, j int) bool {
-		return selfObsMetrics[i].Name < selfObsMetrics[j].Name
-	})
-	return selfObsMetrics
+	return &SelfObservabilityMetric{
+			CreateTimeSeriesRequests:       i.testServer.CreateTimeSeriesRequests(),
+			CreateMetricDescriptorRequests: i.testServer.CreateMetricDescriptorRequests(),
+		},
+		nil
 }
 
 // Shutdown unregisters the global OpenCensus views to reset state for the next test
 func (i *InMemoryOCExporter) Shutdown(ctx context.Context) error {
-	view.Unregister(collector.MetricViews()...)
+	i.stackdriverExporter.StopMetricsExporter()
+	err := i.stackdriverExporter.Close()
+	i.testServer.Shutdown()
+
+	view.Unregister(getViews()...)
+	// TODO: remove this special case
 	view.Unregister(ocgrpc.DefaultClientViews...)
-	return nil
+
+	return err
 }
 
 // NewInMemoryOCViewExporter creates a new in memory OC exporter for testing. Be sure to defer
 // a call to Shutdown().
 func NewInMemoryOCViewExporter() (*InMemoryOCExporter, error) {
 	// Reset our views in case any tests ran before this
-	view.Unregister(collector.MetricViews()...)
-	view.Register(collector.MetricViews()...)
+	views := getViews()
+	view.Unregister(views...)
+	view.Register(views...)
+
+	// TODO: remove this special case
 	view.Unregister(ocgrpc.DefaultClientViews...)
-	// TODO: Register ocgrpc.DefaultClientViews to test them
+
+	testServer, err := NewMetricTestServer()
+	if err != nil {
+		return nil, err
+	}
+	go testServer.Serve()
+	conn, err := grpc.Dial(testServer.Endpoint, grpc.WithInsecure())
+	if err != nil {
+		return nil, err
+	}
+	clientOpts := []option.ClientOption{option.WithGRPCConn(conn)}
+
+	stackdriverExporter, err := stackdriver.NewExporter(stackdriver.Options{
+		DefaultMonitoringLabels: &stackdriver.Labels{},
+		ProjectID:               "myproject",
+		MonitoringClientOptions: clientOpts,
+		TraceClientOptions:      clientOpts,
+	})
+	if err != nil {
+		return nil, err
+	}
 
 	return &InMemoryOCExporter{
-			c:      make(chan []*metricdata.Metric, 1),
-			reader: metricexport.NewReader(),
+			testServer:          testServer,
+			stackdriverExporter: stackdriverExporter,
+			reader:              metricexport.NewReader(),
 		},
 		nil
 }