Repository: knative/pkg
Pull request #1741 (2020-09-23 21:33:43)

Add TTL expiry for items in allMeters.meters map
Currently, pkg/metrics keeps one meterExporter for each resource the system is sending metrics about.

As resources  fall off the system and no longer send metrics (deleting a knative revision, for example), a slow leak occurs as orphaned meterExporters are never cleaned up.

This PR adds a TTL expiry bit for each meterExporter, and cleans them up when they expire.

knative-prow-robot: Skipping CI for Draft Pull Request.
If you want CI signal for your change, please convert it to an actual PR.
You can still manually trigger a test run with `/test all`
knative-prow-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative/pkg/pull/1741#" title="Author self-approved">MontyCarter</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **yanweiguo**
You can assign the PR to them by writing `/assign @yanweiguo` in a comment when ready.

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

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

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

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":["yanweiguo"]} -->
MontyCarter: /assign yanweiguo
codecov[bot]: # [Codecov](https://codecov.io/gh/knative/pkg/pull/1741?src=pr&el=h1) Report
> Merging [#1741](https://codecov.io/gh/knative/pkg/pull/1741?src=pr&el=desc) into [master](https://codecov.io/gh/knative/pkg/commit/5fbbde31b33a121e934f65911f62b58eb413f7a9?el=desc) will **increase** coverage by `0.04%`.
> The diff coverage is `85.18%`.

[![Impacted file tree graph](https://codecov.io/gh/knative/pkg/pull/1741/graphs/tree.svg?width=650&height=150&src=pr&token=rlqpDnfxbO)](https://codecov.io/gh/knative/pkg/pull/1741?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##           master    #1741      +/-   ##
==========================================
+ Coverage   68.44%   68.49%   +0.04%     
==========================================
  Files         215      215              
  Lines        9204     9223      +19     
==========================================
+ Hits         6300     6317      +17     
- Misses       2587     2589       +2     
  Partials      317      317              
```


| [Impacted Files](https://codecov.io/gh/knative/pkg/pull/1741?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [metrics/resource\_view.go](https://codecov.io/gh/knative/pkg/pull/1741/diff?src=pr&el=tree#diff-bWV0cmljcy9yZXNvdXJjZV92aWV3Lmdv) | `83.87% <85.18%> (-0.69%)` | :arrow_down: |
| [test/gcs/mock/mock.go](https://codecov.io/gh/knative/pkg/pull/1741/diff?src=pr&el=tree#diff-dGVzdC9nY3MvbW9jay9tb2NrLmdv) | `91.39% <0.00%> (+1.07%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative/pkg/pull/1741?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative/pkg/pull/1741?src=pr&el=footer). Last update [5fbbde3...641ac1f](https://codecov.io/gh/knative/pkg/pull/1741?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-pkg-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[metrics/resource_view.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative_pkg/1741/pull-knative-pkg-go-coverage/1308885328861335554/artifacts/line-cov.html#file0) | 91.8% | 91.0% | -0.8

yanweiguo: /assign @evankanderson 

I'm not sure the reason why we need to use a cache to replace map. Evan has more knowledge than me on these codes.
knative-prow-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative/pkg/pull/1741#" title="Author self-approved">MontyCarter</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **evankanderson**
You can assign the PR to them by writing `/assign @evankanderson` in a comment when ready.

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

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

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

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"]} -->
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-pkg-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[metrics/resource_view.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative_pkg/1741/pull-knative-pkg-go-coverage/1308897156404350976/artifacts/line-cov.html#file0) | 91.8% | 92.2% | 0.3

knative-prow-robot: @MontyCarter: The following test **failed**, say `/retest` to rerun all failed tests:

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

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

<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 -->
knative-prow-robot: @MontyCarter: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-pkg-unit-tests | 4508cfa8cc1ce852b5fd876a557a928809bc7af8 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative_pkg/1741/pull-knative-pkg-unit-tests/1311445182331228161) | `/test pull-knative-pkg-unit-tests`

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

<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 -->
knative-prow-robot: @MontyCarter: The following test **failed**, say `/retest` to rerun all failed tests:

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

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

<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 -->
knative-prow-robot: @MontyCarter: The following test **failed**, say `/retest` to rerun all failed tests:

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

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

<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: Two notes:

1. You probably want to update the title of this PR.
2. You probably want a description.

Taking a look now.
evankanderson: > Not a fan of bespoke things likes this, but 🤷

I think Monty has also looked at various lifetime-based caches. Ideally, the set of current `meterExporter`s would be a subset of the current set of items in the Informer cache, which is theoretically unbounded, but at least people expect it to grow as clusters grow, but not over time for the same set of live objects.

The current implementation in `metrics` suffers from a leak as resources are cycled, since it never expires an exporter from the map, though they may quiesce a bit if there are no further metrics recorded.
knative-prow-robot: @MontyCarter: The following test **failed**, say `/retest` to rerun all failed tests:

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

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

<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 -->
MontyCarter: > Two notes:
> 
> 1. You probably want to update the title of this PR.
> 2. You probably want a description.
> 
> Taking a look now.

Done.
MontyCarter: > > Not a fan of bespoke things likes this, but 🤷
> 
> I think Monty has also looked at various lifetime-based caches. Ideally, the set of current `meterExporter`s would be a subset of the current set of items in the Informer cache, which is theoretically unbounded, but at least people expect it to grow as clusters grow, but not over time for the same set of live objects.
> 
> The current implementation in `metrics` suffers from a leak as resources are cycled, since it never expires an exporter from the map, though they may quiesce a bit if there are no further metrics recorded.
> 
> Would you prefer to make the expiring-map construct a library import? I'm of two minds here, particularly with the lack of Go generics. On one hand, it's sort of silly to maintain our own implementation; on the other hand, relying on a third-party implementation which isn't actively maintained may not improve things much. (Guava has the benefit that there's a team continually improving it, which a library posted by an individual may not have.)

There are three candidates that I identified, if we decide to go this way:

- https://github.com/ReneKroon/ttlcache - Good interface, active project, but doesn't have ability to iterate over contents - would have to add
- https://github.com/patrickmn/go-cache - Acceptable interface, seems active enough, but "iterating" creates a new map
- https://github.com/muesli/cache2go - Not as active.  Less mature interface, less active project, but provides Foreach which takes a function pointer to run against each item.
MontyCarter: /retest
vagababov: /lgtm
In general
knative-prow-robot: New changes are detected. LGTM label has been removed.
knative-prow-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative/pkg/pull/1741#pullrequestreview-502282886" title="Approved">evankanderson</a>*, *<a href="https://github.com/knative/pkg/pull/1741#" title="Author self-approved">MontyCarter</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":[]} -->

Patch:
diff --git a/metrics/resource_view.go b/metrics/resource_view.go
index 1e89787f6..57e54a6f0 100644
--- a/metrics/resource_view.go
+++ b/metrics/resource_view.go
@@ -29,6 +29,7 @@ import (
 	"go.opencensus.io/stats"
 	"go.opencensus.io/stats/view"
 	"go.opencensus.io/tag"
+	"k8s.io/apimachinery/pkg/util/clock"
 )
 
 type storedViews struct {
@@ -40,6 +41,7 @@ type meterExporter struct {
 	m view.Meter    // NOTE: DO NOT RETURN THIS DIRECTLY; the view.Meter will not work for the empty Resource
 	o stats.Options // Cache the option to reduce allocations
 	e view.Exporter
+	t time.Time // Time when last access occurred
 }
 
 // ResourceExporterFactory provides a hook for producing separate view.Exporters
@@ -48,17 +50,70 @@ type meterExporter struct {
 // Tags are.
 type ResourceExporterFactory func(*resource.Resource) (view.Exporter, error)
 type meters struct {
-	meters  map[string]*meterExporter
-	factory ResourceExporterFactory
-	lock    sync.Mutex
+	meters         map[string]*meterExporter
+	factory        ResourceExporterFactory
+	lock           sync.Mutex
+	clock          clock.Clock
+	ticker         clock.Ticker
+	tickerStopChan chan struct{}
 }
 
 // Lock regime: lock allMeters before resourceViews. The critical path is in
 // optionForResource, which must lock allMeters, but only needs to lock
 // resourceViews if a new meter needs to be created.
-var resourceViews = storedViews{}
-var allMeters = meters{
-	meters: map[string]*meterExporter{"": &defaultMeter},
+var (
+	resourceViews = storedViews{}
+	allMeters     = meters{
+		meters: map[string]*meterExporter{"": &defaultMeter},
+		clock:  clock.Clock(clock.RealClock{}),
+	}
+
+	cleanupOnce                = new(sync.Once)
+	meterExporterScanFrequency = 1 * time.Minute
+	maxMeterExporterAge        = 10 * time.Minute
+)
+
+// cleanup looks through allMeter's meterExporter and prunes any that were last visited
+// more than maxMeterExporterAge ago. This will clean up any exporters for resources
+// that no longer exist on the system (a replaced revision, e.g.).  All meterExporter
+// lookups (and creations) run meterExporterForResource, which updates the timestamp
+// on each access.
+func cleanup() {
+	expiryCutoff := allMeters.clock.Now().Add(-1 * maxMeterExporterAge)
+	allMeters.lock.Lock()
+	defer allMeters.lock.Unlock()
+	for key, meter := range allMeters.meters {
+		if key != "" && meter.t.Before(expiryCutoff) {
+			flushGivenExporter(meter.e)
+			delete(allMeters.meters, key)
+		}
+	}
+}
+
+// startCleanup is configured to only run once in production.  For testing,
+// calling startCleanup again will terminate old cleanup threads,
+// and restart anew.
+func startCleanup() {
+	if allMeters.tickerStopChan != nil {
+		close(allMeters.tickerStopChan)
+	}
+
+	curClock := allMeters.clock
+	newTicker := curClock.NewTicker(meterExporterScanFrequency)
+	newStopChan := make(chan struct{})
+	allMeters.tickerStopChan = newStopChan
+	allMeters.ticker = newTicker
+	go func() {
+		defer newTicker.Stop()
+		for {
+			select {
+			case <-newTicker.C():
+				cleanup()
+			case <-newStopChan:
+				return
+			}
+		}
+	}()
 }
 
 // RegisterResourceView is similar to view.Register(), except that it will
@@ -174,7 +229,9 @@ func flushResourceExporters() {
 }
 
 // ClearMetersForTest clears the internal set of metrics being exported,
-// including cleaning up background threads.
+// including cleaning up background threads, and restarts cleanup thread.
+//
+// If a replacement clock is desired, it should be in allMeters.clock before calling.
 func ClearMetersForTest() {
 	allMeters.lock.Lock()
 	defer allMeters.lock.Unlock()
@@ -186,6 +243,7 @@ func ClearMetersForTest() {
 		meter.m.Stop()
 		delete(allMeters.meters, k)
 	}
+	startCleanup()
 }
 
 func meterExporterForResource(r *resource.Resource) *meterExporter {
@@ -195,6 +253,7 @@ func meterExporterForResource(r *resource.Resource) *meterExporter {
 		mE = &meterExporter{}
 		allMeters.meters[key] = mE
 	}
+	mE.t = allMeters.clock.Now()
 	if mE.o != nil {
 		return mE
 	}
@@ -219,6 +278,9 @@ func meterExporterForResource(r *resource.Resource) *meterExporter {
 // optionForResource finds or creates a stats exporter for the resource, and
 // returns a stats.Option indicating which meter to record to.
 func optionForResource(r *resource.Resource) (stats.Options, error) {
+	// Start the allMeters cleanup thread, if not already started
+	cleanupOnce.Do(startCleanup)
+
 	allMeters.lock.Lock()
 	defer allMeters.lock.Unlock()
 
@@ -313,6 +375,7 @@ var _ view.Meter = (*defaultMeterImpl)(nil)
 var defaultMeter = meterExporter{
 	m: &defaultMeterImpl{},
 	o: stats.WithRecorder(nil),
+	t: time.Now(), // time.Now() here is ok - defaultMeter never expires, so this won't be checked
 }
 
 func (*defaultMeterImpl) Record(*tag.Map, interface{}, map[string]interface{}) {


Test patch:
diff --git a/metrics/resource_view_test.go b/metrics/resource_view_test.go
index 3c5a8e241..e527672b8 100644
--- a/metrics/resource_view_test.go
+++ b/metrics/resource_view_test.go
@@ -37,6 +37,7 @@ import (
 	"go.opencensus.io/stats"
 	"go.opencensus.io/stats/view"
 	"go.opencensus.io/tag"
+	"k8s.io/apimachinery/pkg/util/clock"
 
 	emptypb "github.com/golang/protobuf/ptypes/empty"
 	"github.com/google/go-cmp/cmp"
@@ -144,6 +145,70 @@ func TestSetFactory(t *testing.T) {
 	}
 }
 
+func TestAllMetersExpiration(t *testing.T) {
+	allMeters.clock = clock.Clock(clock.NewFakeClock(time.Now()))
+	var fakeClock *clock.FakeClock = allMeters.clock.(*clock.FakeClock)
+	ClearMetersForTest() // t+0m
+
+	// Add resource123
+	resource123 := r
+	resource123.Labels["id"] = "123"
+	_, err := optionForResource(&resource123)
+	if err != nil {
+		t.Error("Should succeed getting option, instead got error ", err)
+	}
+	// (123=0m, 456=Inf)
+
+	// Bump time to make resource123's expiry offset from resource456
+	fakeClock.Step(90 * time.Second) // t+1.5m
+	// (123=0m, 456=Inf)
+
+	// Add 456
+	resource456 := r
+	resource456.Labels["id"] = "456"
+	_, err = optionForResource(&resource456)
+	if err != nil {
+		t.Error("Should succeed getting option, instead got error ", err)
+	}
+	allMeters.lock.Lock()
+	if len(allMeters.meters) != 3 {
+		t.Errorf("len(allMeters)=%d, want: 3", len(allMeters.meters))
+	}
+	allMeters.lock.Unlock()
+	// (123=1.5m, 456=0m)
+
+	// Warm up the older entry
+	fakeClock.Step(90 * time.Second) //t+3m
+	// (123=4.5m, 456=3m)
+
+	// Refresh the first entry
+	_, err = optionForResource(&resource123)
+	if err != nil {
+		t.Error("Should succeed getting option, instead got error ", err)
+	}
+	// (123=0, 456=1.5m)
+
+	// Expire the second entry
+	fakeClock.Step(9 * time.Minute) // t+12m
+	time.Sleep(1 * time.Second)     // Wait a second on the wallclock, so that the cleanup thread has time to finish a loop
+	allMeters.lock.Lock()
+	if len(allMeters.meters) != 2 {
+		t.Errorf("len(allMeters)=%d, want: 2", len(allMeters.meters))
+	}
+	allMeters.lock.Unlock()
+	// (123=9m, 456=10.5m)
+	// non-expiring defaultMeter was just tested
+
+	// Add resource789
+	resource789 := r
+	resource789.Labels["id"] = "789"
+	_, err = optionForResource(&resource789)
+	if err != nil {
+		t.Error("Should succeed getting option, instead got error ", err)
+	}
+	// (123=9m, 456=evicted, 789=0m)
+}
+
 func TestResourceAsString(t *testing.T) {
 	r1 := &resource.Resource{Type: "foobar", Labels: map[string]string{"k1": "v1", "k3": "v3", "k2": "v2"}}
 	r2 := &resource.Resource{Type: "foobar", Labels: map[string]string{"k2": "v2", "k3": "v3", "k1": "v1"}}