Repository: knative/pkg
Pull request #1591 (2020-08-06 20:17:53)

Use real synchronization to make sure we get four exports in the OpenCensus test
Somehow, the timing on my machine always "wins" this race even without the sleep, but switch from a 10ms sleep to real `sync.WaitGroup`s since the race can apparently be lost, making the test flaky: https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative_pkg/1588/pull-knative-pkg-unit-tests/1291443127265005569

/assign @vagababov 

knative-prow-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative/pkg/pull/1591#" title="Author self-approved">evankanderson</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative%2Fpkg).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[metrics/OWNERS](https://github.com/knative/pkg/blob/master/metrics/OWNERS)~~ [evankanderson]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
evankanderson: The chain goes:

* set up `oc.exports` with 4 items
* wait for that to hit 0 due to the test executing:
  * then shut down the streaming RPC server
  * this causes the infinite loop of reading exports to exit
* once the grpc server has exited **and all the infinite-loop streaming threads**:
  * close the channel
* This causes the `for ... range` loop to terminate reading all the metrics
knative-prow-robot: @evankanderson: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-pkg-unit-tests | 191aaaf7f88fb1c8c486a38429e6fd77ef9da152 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative_pkg/1591/pull-knative-pkg-unit-tests/1291468922133942273) | `/test pull-knative-pkg-unit-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative_pkg/1591). [Your PR dashboard](https://gubernator.knative.dev/pr/evankanderson).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
evankanderson: Obviously, that didn't work on CI.
evankanderson: /hold
evankanderson: ```
panic: sync: negative WaitGroup counter
goroutine 667 [running]:
=== RUN   TestGetURIDestinationV1/nil_url
sync.(*WaitGroup).Add(0xc0004c8618, 0xffffffffffffffff)
	/root/.gvm/gos/go1.14.6/src/sync/waitgroup.go:74 +0x2c0
sync.(*WaitGroup).Done(...)
	/root/.gvm/gos/go1.14.6/src/sync/waitgroup.go:99
knative.dev/pkg/metrics.(*openCensusFake).Export(0xc0004c8600, 0x2bcc1e0, 0xc0003cb490, 0x0, 0x0)
	/home/prow/go/src/knative.dev/pkg/metrics/resource_view_test.go:651 +0x24b
github.com/census-instrumentation/opencensus-proto/gen-go/agent/metrics/v1._MetricsService_Export_Handler(0x243fe00, 0xc0004c8600, 0x2bc1380, 0xc00084a000, 0x3ba9658, 0xc000242500)
	/home/prow/go/src/knative.dev/pkg/vendor/github.com/census-instrumentation/opencensus-proto/gen-go/agent/metrics/v1/metrics_service.pb.go:237 +0xce
google.golang.org/grpc.(*Server).processStreamingRPC(0xc0009c2000, 0x2bd2b40, 0xc000a70600, 0xc000242500, 0xc0009be2a0, 0x3b56220, 0x0, 0x0, 0x0)
	/home/prow/go/src/knative.dev/pkg/vendor/google.golang.org/grpc/server.go:1427 +0x15c4
google.golang.org/grpc.(*Server).handleStream(0xc0009c2000, 0x2bd2b40, 0xc000a70600, 0xc000242500, 0x0)
	/home/prow/go/src/knative.dev/pkg/vendor/google.golang.org/grpc/server.go:1507 +0x1302
google.golang.org/grpc.(*Server).serveStreams.func1.2(0xc000788550, 0xc0009c2000, 0x2bd2b40, 0xc000a70600, 0xc000242500)
	/home/prow/go/src/knative.dev/pkg/vendor/google.golang.org/grpc/server.go:843 +0xd9
created by google.golang.org/grpc.(*Server).serveStreams.func1
	/home/prow/go/src/knative.dev/pkg/vendor/google.golang.org/grpc/server.go:841 +0x342
```

Patch:
diff --git a/configmap/informed_watcher.go b/configmap/informed_watcher.go
index 3231e7d28..16447b634 100644
--- a/configmap/informed_watcher.go
+++ b/configmap/informed_watcher.go
@@ -33,13 +33,6 @@ import (
 	"k8s.io/client-go/tools/cache"
 )
 
-// NewDefaultWatcher creates a new default configmap.Watcher instance.
-//
-// Deprecated: Use NewInformedWatcher
-func NewDefaultWatcher(kc kubernetes.Interface, namespace string) *InformedWatcher {
-	return NewInformedWatcher(kc, namespace)
-}
-
 // NewInformedWatcherFromFactory watches a Kubernetes namespace for ConfigMap changes.
 func NewInformedWatcherFromFactory(sif informers.SharedInformerFactory, namespace string) *InformedWatcher {
 	return &InformedWatcher{
diff --git a/controller/controller.go b/controller/controller.go
index e32db29bd..121db8567 100644
--- a/controller/controller.go
+++ b/controller/controller.go
@@ -262,7 +262,9 @@ func (c *Impl) EnqueueAfter(obj interface{}, after time.Duration) {
 // and enqueues that key in the slow lane.
 func (c *Impl) EnqueueSlowKey(key types.NamespacedName) {
 	c.workQueue.SlowLane().Add(key)
-	c.logger.Debugf("Adding to the slow queue %s (depth(total/slow): %d/%d)", safeKey(key), c.workQueue.Len(), c.workQueue.SlowLane().Len())
+	c.logger.With(zap.Object(logkey.Key, logging.NamespacedName(key))).
+		Debug("Adding to the slow queue %s (depth(total/slow): %d/%d)",
+			safeKey(key), c.workQueue.Len(), c.workQueue.SlowLane().Len())
 }
 
 // EnqueueSlow extracts namesspeced name from the object and enqueues it on the slow
@@ -388,7 +390,8 @@ func (c *Impl) EnqueueNamespaceOf(obj interface{}) {
 // EnqueueKey takes a namespace/name string and puts it onto the work queue.
 func (c *Impl) EnqueueKey(key types.NamespacedName) {
 	c.workQueue.Add(key)
-	c.logger.Debugf("Adding to queue %s (depth: %d)", safeKey(key), c.workQueue.Len())
+	c.logger.With(zap.Object(logkey.Key, logging.NamespacedName(key))).
+		Debugf("Adding to queue %s (depth: %d)", safeKey(key), c.workQueue.Len())
 }
 
 // MaybeEnqueueBucketKey takes a Bucket and namespace/name string and puts it onto
@@ -403,7 +406,8 @@ func (c *Impl) MaybeEnqueueBucketKey(bkt reconciler.Bucket, key types.Namespaced
 // the work queue after given delay.
 func (c *Impl) EnqueueKeyAfter(key types.NamespacedName, delay time.Duration) {
 	c.workQueue.AddAfter(key, delay)
-	c.logger.Debugf("Adding to queue %s (delay: %v, depth: %d)", safeKey(key), delay, c.workQueue.Len())
+	c.logger.With(zap.Object(logkey.Key, logging.NamespacedName(key))).
+		Debugf("Adding to queue %s (delay: %v, depth: %d)", safeKey(key), delay, c.workQueue.Len())
 }
 
 // RunContext starts the controller's worker threads, the number of which is threadiness.
diff --git a/metrics/opencensus_exporter.go b/metrics/opencensus_exporter.go
index 651d3ce66..d94b53e69 100644
--- a/metrics/opencensus_exporter.go
+++ b/metrics/opencensus_exporter.go
@@ -52,11 +52,15 @@ func newOpenCensusExporter(config *metricsConfig, logger *zap.SugaredLogger) (vi
 	}
 	logger.Infow("Created OpenCensus exporter with config:", zap.Any("config", *config))
 	view.RegisterExporter(e)
-	return e, getFactory(opts), nil
+	return e, getFactory(e, opts), nil
 }
 
-func getFactory(stored []ocagent.ExporterOption) ResourceExporterFactory {
+func getFactory(defaultExporter view.Exporter, stored []ocagent.ExporterOption) ResourceExporterFactory {
 	return func(r *resource.Resource) (view.Exporter, error) {
+		if r == nil || (r.Type == "" && len(r.Labels) == 0) {
+			// Don't create duplicate exporters for the default exporter.
+			return defaultExporter, nil
+		}
 		opts := append(stored, ocagent.WithResourceDetector(
 			func(context.Context) (*resource.Resource, error) {
 				return r, nil
diff --git a/metrics/resource_view.go b/metrics/resource_view.go
index 2fcf24ef9..7d69ae94e 100644
--- a/metrics/resource_view.go
+++ b/metrics/resource_view.go
@@ -177,6 +177,21 @@ func flushResourceExporters() {
 	}
 }
 
+// ClearMetersForTest clears the internal set of metrics being exported,
+// including cleaning up background threads.
+func ClearMetersForTest() {
+	allMeters.lock.Lock()
+	defer allMeters.lock.Unlock()
+
+	for k, meter := range allMeters.meters {
+		if k == "" {
+			continue
+		}
+		meter.m.Stop()
+		delete(allMeters.meters, k)
+	}
+}
+
 func meterExporterForResource(r *resource.Resource) *meterExporter {
 	key := resourceToKey(r)
 	mE := allMeters.meters[key]
diff --git a/metrics/stackdriver_exporter.go b/metrics/stackdriver_exporter.go
index 44ae68b68..befb95dba 100644
--- a/metrics/stackdriver_exporter.go
+++ b/metrics/stackdriver_exporter.go
@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"path"
 	"sync"
+	"time"
 
 	sd "contrib.go.opencensus.io/exporter/stackdriver"
 	"go.opencensus.io/resource"
@@ -49,6 +50,8 @@ const (
 	StackdriverSecretNameDefault = "stackdriver-service-account-key"
 	// secretDataFieldKey is the name of the k8s Secret field that contains the Secret's key.
 	secretDataFieldKey = "key.json"
+	// stackdriverApiTimeout is the timeout value of Stackdriver API service side.
+	stackdriverApiTimeout = 12 * time.Second
 )
 
 var (
@@ -186,6 +189,7 @@ func newStackdriverExporter(config *metricsConfig, logger *zap.SugaredLogger) (v
 		GetMetricPrefix:         mpf,
 		ReportingInterval:       config.reportingPeriod,
 		DefaultMonitoringLabels: &sd.Labels{},
+		Timeout:                 stackdriverApiTimeout,
 	})
 	if err != nil {
 		logger.Errorw("Failed to create the Stackdriver exporter: ", zap.Error(err))


Test patch:
diff --git a/metrics/resource_view_test.go b/metrics/resource_view_test.go
index 72d09817e..e579b16a3 100644
--- a/metrics/resource_view_test.go
+++ b/metrics/resource_view_test.go
@@ -29,7 +29,6 @@ import (
 	"strings"
 	"sync"
 	"testing"
-	"time"
 
 	sd "contrib.go.opencensus.io/exporter/stackdriver"
 	ocmetrics "github.com/census-instrumentation/opencensus-proto/gen-go/agent/metrics/v1"
@@ -326,7 +325,7 @@ testComponent_testing_value{project="p1",revision="r2"} 1
 	}, {
 		name: "OpenCensus",
 		init: func() error {
-			if err := ocFake.start(); err != nil {
+			if err := ocFake.start(len(resources) + 1); err != nil {
 				return err
 			}
 			t.Logf("Created exporter at %s", ocFake.address)
@@ -339,10 +338,7 @@ testComponent_testing_value{project="p1",revision="r2"} 1
 			// [new duration] in the future.
 			view.Unregister(globalCounter)
 			UnregisterResourceView(gaugeView, resourceCounter)
-			FlushExporter()
 
-			time.Sleep(10 * time.Millisecond)
-			ocFake.srv.Stop() // Force close connections
 			records := []metricExtract{}
 			for record := range ocFake.published {
 				for _, m := range record.Metrics {
@@ -399,6 +395,7 @@ testComponent_testing_value{project="p1",revision="r2"} 1
 
 	for _, c := range harnesses {
 		t.Run(c.name, func(t *testing.T) {
+			ClearMetersForTest()
 			sdFake.t = t
 			if err := c.init(); err != nil {
 				t.Fatalf("unable to init: %+v", err)
@@ -599,11 +596,12 @@ func TestStackDriverExports(t *testing.T) {
 type openCensusFake struct {
 	address   string
 	srv       *grpc.Server
+	exports   sync.WaitGroup
 	wg        sync.WaitGroup
 	published chan ocmetrics.ExportMetricsServiceRequest
 }
 
-func (oc *openCensusFake) start() error {
+func (oc *openCensusFake) start(expectedStreams int) error {
 	ln, err := net.Listen("tcp", oc.address)
 	if err != nil {
 		return err
@@ -619,13 +617,21 @@ func (oc *openCensusFake) start() error {
 		oc.wg.Wait()
 		close(oc.published)
 	}()
+	oc.exports.Add(expectedStreams)
+	go oc.stop()
 	return nil
 }
 
+func (oc *openCensusFake) stop() {
+	oc.exports.Wait()
+	oc.srv.Stop()
+}
+
 func (oc *openCensusFake) Export(stream ocmetrics.MetricsService_ExportServer) error {
 	var streamResource *ocresource.Resource
 	oc.wg.Add(1)
 	defer oc.wg.Done()
+	metricSeen := false
 	for {
 		in, err := stream.Recv()
 		if err == io.EOF {
@@ -643,6 +649,10 @@ func (oc *openCensusFake) Export(stream ocmetrics.MetricsService_ExportServer) e
 				in.Resource = streamResource
 			}
 			oc.published <- *in
+			if !metricSeen {
+				oc.exports.Done()
+				metricSeen = true
+			}
 		}
 	}
 }