Call CreateServiceTimeSeries for service time series (#294)

Co-authored-by: David Ashpole <dashpole@google.com>
Co-authored-by: Punya Biswal <punya@google.com>

Author: 3 people

equivalence_test.go

@@ -372,6 +372,7 @@ type fakeMetricsServer struct {
 	monitoringpb.MetricServiceServer
 	mu                           sync.RWMutex
 	stackdriverTimeSeries        []*monitoringpb.CreateTimeSeriesRequest
+	stackdriverServiceTimeSeries []*monitoringpb.CreateTimeSeriesRequest
 	stackdriverMetricDescriptors []*monitoringpb.CreateMetricDescriptorRequest
 }
 
@@ -437,3 +438,10 @@ func (server *fakeMetricsServer) CreateTimeSeries(ctx context.Context, req *moni
 	server.mu.Unlock()
 	return new(empty.Empty), nil
 }
+
+func (server *fakeMetricsServer) CreateServiceTimeSeries(ctx context.Context, req *monitoringpb.CreateTimeSeriesRequest) (*empty.Empty, error) {
+	server.mu.Lock()
+	server.stackdriverServiceTimeSeries = append(server.stackdriverServiceTimeSeries, req)
+	server.mu.Unlock()
+	return new(empty.Empty), nil
+}

go.mod

@@ -3,22 +3,22 @@ module contrib.go.opencensus.io/exporter/stackdriver
 go 1.15
 
 require (
-	cloud.google.com/go v0.75.0
+	cloud.google.com/go v0.97.0
+	cloud.google.com/go/monitoring v1.1.0
+	cloud.google.com/go/trace v1.0.0
 	github.com/aws/aws-sdk-go v1.37.0
 	github.com/census-instrumentation/opencensus-proto v0.3.0
-	github.com/golang/protobuf v1.4.3
-	github.com/google/go-cmp v0.5.4
+	github.com/golang/protobuf v1.5.2
+	github.com/google/go-cmp v0.5.6
 	github.com/jstemmer/go-junit-report v0.9.1
-	go.opencensus.io v0.22.6
-	golang.org/x/lint v0.0.0-20201208152925-83fdc39ff7b5
-	golang.org/x/net v0.0.0-20210119194325-5f4716e94777
-	golang.org/x/oauth2 v0.0.0-20210126194326-f9ce19ea3013
-	golang.org/x/sys v0.0.0-20210124154548-22da62e12c0c // indirect
-	golang.org/x/text v0.3.5 // indirect
-	golang.org/x/tools v0.0.0-20210108195828-e2f9c7f1fc8e
-	google.golang.org/api v0.37.0
-	google.golang.org/genproto v0.0.0-20210126160654-44e461bb6506
-	google.golang.org/grpc v1.35.0
-	google.golang.org/protobuf v1.25.0
+	go.opencensus.io v0.23.0
+	golang.org/x/lint v0.0.0-20210508222113-6edffad5e616
+	golang.org/x/net v0.0.0-20210503060351-7fd8e65b6420
+	golang.org/x/oauth2 v0.0.0-20211005180243-6b3c2da341f1
+	golang.org/x/tools v0.1.5
+	google.golang.org/api v0.59.0
+	google.golang.org/genproto v0.0.0-20211018162055-cf77aa76bad2
+	google.golang.org/grpc v1.40.0
+	google.golang.org/protobuf v1.27.1
 	honnef.co/go/tools v0.0.1-2020.1.4
 )

go.sum

@@ -111,11 +111,24 @@ func (se *statsExporter) uploadMetrics(metrics []*metricdata.Metric) error {
 			end = len(allTimeSeries)
 		}
 		batch := allTimeSeries[start:end]
-		ctsreql := se.combineTimeSeriesToCreateTimeSeriesRequest(batch)
-		for _, ctsreq := range ctsreql {
-			if err := createTimeSeries(ctx, se.c, ctsreq); err != nil {
-				span.SetStatus(trace.Status{Code: trace.StatusCodeUnknown, Message: err.Error()})
-				errors = append(errors, err)
+		serviceTsBatch, nonServiceTsBatch := splitTimeSeries(batch)
+
+		if len(nonServiceTsBatch) > 0 {
+			nonServiceReql := se.combineTimeSeriesToCreateTimeSeriesRequest(nonServiceTsBatch)
+			for _, ctsreq := range nonServiceReql {
+				if err := createTimeSeries(ctx, se.c, ctsreq); err != nil {
+					span.SetStatus(trace.Status{Code: trace.StatusCodeUnknown, Message: err.Error()})
+					errors = append(errors, err)
+				}
+			}
+		}
+		if len(serviceTsBatch) > 0 {
+			serviceReql := se.combineTimeSeriesToCreateTimeSeriesRequest(serviceTsBatch)
+			for _, ctsreq := range serviceReql {
+				if err := createServiceTimeSeries(ctx, se.c, ctsreq); err != nil {
+					span.SetStatus(trace.Status{Code: trace.StatusCodeUnknown, Message: err.Error()})
+					errors = append(errors, err)
+				}
 			}
 		}
 	}

metrics.go

@@ -138,20 +138,36 @@ var timeSeriesErrRegex = regexp.MustCompile(`: timeSeries\[([0-9]+(?:-[0-9]+)?(?
 
 // sendReq sends create time series requests to Stackdriver,
 // and returns the count of dropped time series and error.
-func sendReq(ctx context.Context, c *monitoring.MetricClient, req *monitoringpb.CreateTimeSeriesRequest) (int, error) {
+func sendReq(ctx context.Context, c *monitoring.MetricClient, req *monitoringpb.CreateTimeSeriesRequest) (int, []error) {
 	// c == nil only happens in unit tests where we don't make real calls to Stackdriver server
 	if c == nil {
 		return 0, nil
 	}
 
-	err := createTimeSeries(ctx, c, req)
-	if err == nil {
-		return 0, nil
+	dropped := 0
+	errors := []error{}
+	serviceReq, nonServiceReq := splitCreateTimeSeriesRequest(req)
+	if nonServiceReq != nil {
+		err := createTimeSeries(ctx, c, nonServiceReq)
+		if err != nil {
+			dropped += droppedTimeSeriesFromMonitoringAPIError(nonServiceReq, err)
+			errors = append(errors, err)
+		}
 	}
+	if serviceReq != nil {
+		err := createServiceTimeSeries(ctx, c, serviceReq)
+		if err != nil {
+			dropped += droppedTimeSeriesFromMonitoringAPIError(serviceReq, err)
+			errors = append(errors, err)
+		}
+	}
+	return dropped, errors
+}
 
-	droppedTimeSeriesRangeMatches := timeSeriesErrRegex.FindAllStringSubmatch(err.Error(), -1)
-	if !strings.HasPrefix(err.Error(), "One or more TimeSeries could not be written:") || len(droppedTimeSeriesRangeMatches) == 0 {
-		return len(req.TimeSeries), err
+func droppedTimeSeriesFromMonitoringAPIError(req *monitoringpb.CreateTimeSeriesRequest, monitoringAPIerr error) int {
+	droppedTimeSeriesRangeMatches := timeSeriesErrRegex.FindAllStringSubmatch(monitoringAPIerr.Error(), -1)
+	if !strings.HasPrefix(monitoringAPIerr.Error(), "One or more TimeSeries could not be written:") || len(droppedTimeSeriesRangeMatches) == 0 {
+		return len(req.TimeSeries)
 	}
 
 	dropped := 0
@@ -171,7 +187,7 @@ func sendReq(ctx context.Context, c *monitoring.MetricClient, req *monitoringpb.
 			}
 		}
 	}
-	return dropped, err
+	return dropped
 }
 
 type worker struct {
@@ -219,10 +235,10 @@ func (w *worker) sendReqWithTimeout(req *monitoringpb.CreateTimeSeriesRequest) {
 	w.recordDroppedTimeseries(sendReq(ctx, w.mc, req))
 }
 
-func (w *worker) recordDroppedTimeseries(numTimeSeries int, err error) {
+func (w *worker) recordDroppedTimeseries(numTimeSeries int, errors []error) {
 	w.resp.droppedTimeSeries += numTimeSeries
-	if err != nil {
-		w.resp.errs = append(w.resp.errs, err)
+	if len(errors) > 0 {
+		w.resp.errs = append(w.resp.errs, errors...)
 	}
 }
 

metrics_batcher.go

@@ -44,10 +44,7 @@ func TestWorkers(t *testing.T) {
 	}
 	m2 := newMetricsBatcher(ctx, "test", 2, c2, defaultTimeout) // batcher with 2 workers
 
-	tss := make([]*monitoringpb.TimeSeries, 0, 500) // make 500 time series, should be split to 3 reqs
-	for i := 0; i < 500; i++ {
-		tss = append(tss, makeTs(i))
-	}
+	tss := makeTs(500, false) // make 500 time series, should be split to 3 reqs
 
 	for _, ts := range tss {
 		m1.addTimeSeries(ts)
@@ -89,69 +86,97 @@ func makeClient(addr string) (*monitoring.MetricClient, error) {
 	return monitoring.NewMetricClient(context.Background(), option.WithGRPCConn(conn))
 }
 
-func makeTs(i int) *monitoringpb.TimeSeries {
-	return &monitoringpb.TimeSeries{
-		Metric: &googlemetricpb.Metric{
-			Type: fmt.Sprintf("custom.googleapis.com/opencensus/test/metric/%v", i),
-			Labels: map[string]string{
-				"key": fmt.Sprintf("test_%v", i),
+// makeTs returns a list of n *monitoringpb.TimeSeries. The metric type (service/non-service)
+// is determined by serviceMetric
+func makeTs(n int, serviceMetric bool) []*monitoringpb.TimeSeries {
+	var tsl []*monitoringpb.TimeSeries
+	for i := 0; i < n; i++ {
+		metricType := fmt.Sprintf("custom.googleapis.com/opencensus/test/metric/%v", i)
+		if serviceMetric {
+			metricType = fmt.Sprintf("kubernetes.io/opencensus/test/metric/%v", i)
+		}
+		tsl = append(tsl, &monitoringpb.TimeSeries{
+			Metric: &googlemetricpb.Metric{
+				Type: metricType,
+				Labels: map[string]string{
+					"key": fmt.Sprintf("test_%v", i),
+				},
 			},
-		},
-		MetricKind: googlemetricpb.MetricDescriptor_CUMULATIVE,
-		ValueType:  googlemetricpb.MetricDescriptor_INT64,
-		Points: []*monitoringpb.Point{
-			{
-				Value: &monitoringpb.TypedValue{
-					Value: &monitoringpb.TypedValue_Int64Value{
-						Int64Value: int64(i),
+			MetricKind: googlemetricpb.MetricDescriptor_CUMULATIVE,
+			ValueType:  googlemetricpb.MetricDescriptor_INT64,
+			Points: []*monitoringpb.Point{
+				{
+					Value: &monitoringpb.TypedValue{
+						Value: &monitoringpb.TypedValue_Int64Value{
+							Int64Value: int64(i),
+						},
 					},
 				},
 			},
-		},
+		})
 	}
+	return tsl
 }
 
 func TestSendReqAndParseDropped(t *testing.T) {
 	type testCase struct {
-		name                 string
-		timeseriesCount      int
-		createTimeSeriesFunc func(ctx context.Context, c *monitoring.MetricClient, ts *monitoringpb.CreateTimeSeriesRequest) error
-		expectedErr          bool
-		expectedDropped      int
+		name                        string
+		nonServiceTimeSeriesCount   int
+		serviceTimeSeriesCount      int
+		createTimeSeriesFunc        func(ctx context.Context, c *monitoring.MetricClient, ts *monitoringpb.CreateTimeSeriesRequest) error
+		createServiceTimeSeriesFunc func(ctx context.Context, c *monitoring.MetricClient, ts *monitoringpb.CreateTimeSeriesRequest) error
+		expectedErr                 bool
+		expectedDropped             int
 	}
 
 	testCases := []testCase{
 		{
-			name:            "No error",
-			timeseriesCount: 75,
+			name:                      "No error",
+			serviceTimeSeriesCount:    75,
+			nonServiceTimeSeriesCount: 75,
 			createTimeSeriesFunc: func(ctx context.Context, c *monitoring.MetricClient, ts *monitoringpb.CreateTimeSeriesRequest) error {
 				return nil
 			},
+			createServiceTimeSeriesFunc: func(ctx context.Context, c *monitoring.MetricClient, ts *monitoringpb.CreateTimeSeriesRequest) error {
+				return nil
+			},
 			expectedErr:     false,
 			expectedDropped: 0,
 		},
 		{
-			name:            "Partial error",
-			timeseriesCount: 75,
+			name:                      "Partial error",
+			serviceTimeSeriesCount:    75,
+			nonServiceTimeSeriesCount: 75,
 			createTimeSeriesFunc: func(ctx context.Context, c *monitoring.MetricClient, ts *monitoringpb.CreateTimeSeriesRequest) error {
 				return errors.New("One or more TimeSeries could not be written: Internal error encountered. Please retry after a few seconds. If internal errors persist, contact support at https://cloud.google.com/support/docs.: timeSeries[0-16,25-44,46-74]; Unknown metric: agent.googleapis.com/system.swap.page_faults: timeSeries[45]")
 			},
+			createServiceTimeSeriesFunc: func(ctx context.Context, c *monitoring.MetricClient, ts *monitoringpb.CreateTimeSeriesRequest) error {
+				return errors.New("One or more TimeSeries could not be written: Internal error encountered. Please retry after a few seconds. If internal errors persist, contact support at https://cloud.google.com/support/docs.: timeSeries[0-16,25-44,46-74]; Unknown metric: agent.googleapis.com/system.swap.page_faults: timeSeries[45]")
+			},
 			expectedErr:     true,
-			expectedDropped: 67,
+			expectedDropped: 67 * 2,
 		},
 		{
-			name:            "Incorrectly formatted error",
-			timeseriesCount: 75,
+			name:                      "Incorrectly formatted error",
+			nonServiceTimeSeriesCount: 75,
+			serviceTimeSeriesCount:    75,
 			createTimeSeriesFunc: func(ctx context.Context, c *monitoring.MetricClient, ts *monitoringpb.CreateTimeSeriesRequest) error {
 				return errors.New("One or more TimeSeries could not be written: Internal error encountered. Please retry after a few seconds. If internal errors persist, contact support at https://cloud.google.com/support/docs.: timeSeries[0-16,25-44,,46-74]; Unknown metric: agent.googleapis.com/system.swap.page_faults: timeSeries[45x]")
 			},
+			createServiceTimeSeriesFunc: func(ctx context.Context, c *monitoring.MetricClient, ts *monitoringpb.CreateTimeSeriesRequest) error {
+				return nil
+			},
 			expectedErr:     true,
 			expectedDropped: 75,
 		},
 		{
-			name:            "Unexpected error format",
-			timeseriesCount: 75,
+			name:                      "Unexpected error format",
+			nonServiceTimeSeriesCount: 75,
+			serviceTimeSeriesCount:    75,
 			createTimeSeriesFunc: func(ctx context.Context, c *monitoring.MetricClient, ts *monitoringpb.CreateTimeSeriesRequest) error {
+				return nil
+			},
+			createServiceTimeSeriesFunc: func(ctx context.Context, c *monitoring.MetricClient, ts *monitoringpb.CreateTimeSeriesRequest) error {
 				return errors.New("err1")
 			},
 			expectedErr:     true,
@@ -162,21 +187,28 @@ func TestSendReqAndParseDropped(t *testing.T) {
 	for _, test := range testCases {
 		t.Run(test.name, func(t *testing.T) {
 			persistedCreateTimeSeries := createTimeSeries
+			persistedCreateServiceTimeSeries := createServiceTimeSeries
 			createTimeSeries = test.createTimeSeriesFunc
+			createServiceTimeSeries = test.createServiceTimeSeriesFunc
+			defer func() {
+				createTimeSeries = persistedCreateTimeSeries
+				createServiceTimeSeries = persistedCreateServiceTimeSeries
+			}()
 
 			mc, _ := monitoring.NewMetricClient(context.Background())
-			d, err := sendReq(context.Background(), mc, &monitoringpb.CreateTimeSeriesRequest{TimeSeries: make([]*monitoringpb.TimeSeries, test.timeseriesCount)})
-			if !test.expectedErr && err != nil {
-				t.Fatal("Expected no err")
+			var tsl []*monitoringpb.TimeSeries
+			tsl = append(tsl, makeTs(test.serviceTimeSeriesCount, true)...)
+			tsl = append(tsl, makeTs(test.nonServiceTimeSeriesCount, false)...)
+			d, errors := sendReq(context.Background(), mc, &monitoringpb.CreateTimeSeriesRequest{TimeSeries: tsl})
+			if !test.expectedErr && len(errors) > 0 {
+				t.Fatalf("Expected no errors, got %v", errors)
 			}
-			if test.expectedErr && err == nil {
-				t.Fatal("Expected noerr")
+			if test.expectedErr && len(errors) == 0 {
+				t.Fatalf("Expected errors, got %v", errors)
 			}
 			if d != test.expectedDropped {
 				t.Fatalf("Want %v dropped, got %v", test.expectedDropped, d)
 			}
-
-			createTimeSeries = persistedCreateTimeSeries
 		})
 	}
 }

metrics_batcher_test.go

@@ -398,7 +398,9 @@ func executeTestCase(t *testing.T, tc *testCases, se *sd.Exporter, server *fakeM
 	server.forEachStackdriverTimeSeries(func(sdt *monitoringpb.CreateTimeSeriesRequest) {
 		gotTimeSeries = append(gotTimeSeries, sdt)
 	})
-
+	server.forEachServiceStackdriverTimeSeries(func(sdt *monitoringpb.CreateTimeSeriesRequest) {
+		gotTimeSeries = append(gotTimeSeries, sdt)
+	})
 	if diff, idx := requireTimeSeriesRequestEqual(t, gotTimeSeries, tc.outTSR); diff != "" {
 		t.Errorf("Name[%s], TimeSeries[%d], Error: -got +want %s\n", tc.name, idx, diff)
 	}
@@ -413,6 +415,7 @@ func executeTestCase(t *testing.T, tc *testCases, se *sd.Exporter, server *fakeM
 	}
 	server.resetStackdriverMetricDescriptors()
 	server.resetStackdriverTimeSeries()
+	server.resetStackdriverServiceTimeSeries()
 }
 
 func readTestCaseFromFiles(t *testing.T, filename string) *testCases {
@@ -514,6 +517,7 @@ type fakeMetricsServer struct {
 	monitoringpb.MetricServiceServer
 	mu                           sync.RWMutex
 	stackdriverTimeSeries        []*monitoringpb.CreateTimeSeriesRequest
+	stackdriverServiceTimeSeries []*monitoringpb.CreateTimeSeriesRequest
 	stackdriverMetricDescriptors []*monitoringpb.CreateMetricDescriptorRequest
 }
 
@@ -548,6 +552,15 @@ func (server *fakeMetricsServer) forEachStackdriverTimeSeries(fn func(sdt *monit
 	}
 }
 
+func (server *fakeMetricsServer) forEachServiceStackdriverTimeSeries(fn func(sdt *monitoringpb.CreateTimeSeriesRequest)) {
+	server.mu.RLock()
+	defer server.mu.RUnlock()
+
+	for _, sdt := range server.stackdriverServiceTimeSeries {
+		fn(sdt)
+	}
+}
+
 func (server *fakeMetricsServer) forEachStackdriverMetricDescriptor(fn func(sdmd *monitoringpb.CreateMetricDescriptorRequest)) {
 	server.mu.RLock()
 	defer server.mu.RUnlock()
@@ -563,6 +576,12 @@ func (server *fakeMetricsServer) resetStackdriverTimeSeries() {
 	server.mu.Unlock()
 }
 
+func (server *fakeMetricsServer) resetStackdriverServiceTimeSeries() {
+	server.mu.Lock()
+	server.stackdriverServiceTimeSeries = server.stackdriverServiceTimeSeries[:0]
+	server.mu.Unlock()
+}
+
 func (server *fakeMetricsServer) resetStackdriverMetricDescriptors() {
 	server.mu.Lock()
 	server.stackdriverMetricDescriptors = server.stackdriverMetricDescriptors[:0]
@@ -583,6 +602,13 @@ func (server *fakeMetricsServer) CreateTimeSeries(ctx context.Context, req *moni
 	return new(empty.Empty), nil
 }
 
+func (server *fakeMetricsServer) CreateServiceTimeSeries(ctx context.Context, req *monitoringpb.CreateTimeSeriesRequest) (*empty.Empty, error) {
+	server.mu.Lock()
+	server.stackdriverServiceTimeSeries = append(server.stackdriverServiceTimeSeries, req)
+	server.mu.Unlock()
+	return new(empty.Empty), nil
+}
+
 func requireTimeSeriesRequestEqual(t *testing.T, got, want []*monitoringpb.CreateTimeSeriesRequest) (string, int) {
 	diff := ""
 	i := 0