Repository: knative/serving
Pull request #10060 (2020-11-06 23:59:45)

Use full holder identifier in forwarding metric logs for better debugging

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

This pull-request has been approved by: *<a href="https://github.com/knative/serving/pull/10060#" title="Author self-approved">yanweiguo</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **vagababov** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @vagababov` 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%2Fserving).

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

- **[pkg/autoscaler/OWNERS](https://github.com/knative/serving/blob/master/pkg/autoscaler/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":["vagababov"]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative/serving/pull/10060?src=pr&el=h1) Report
> Merging [#10060](https://codecov.io/gh/knative/serving/pull/10060?src=pr&el=desc) (5ac3b70) into [master](https://codecov.io/gh/knative/serving/commit/63b0d153719dc395e024e188955684980671b33f?el=desc) (63b0d15) will **increase** coverage by `0.07%`.
> The diff coverage is `90.90%`.

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

```diff
@@            Coverage Diff             @@
##           master   #10060      +/-   ##
==========================================
+ Coverage   87.96%   88.03%   +0.07%     
==========================================
  Files         185      185              
  Lines        8683     8685       +2     
==========================================
+ Hits         7638     7646       +8     
+ Misses        796      792       -4     
+ Partials      249      247       -2     
```


| [Impacted Files](https://codecov.io/gh/knative/serving/pull/10060?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [pkg/autoscaler/statforwarder/forwarder.go](https://codecov.io/gh/knative/serving/pull/10060/diff?src=pr&el=tree#diff-cGtnL2F1dG9zY2FsZXIvc3RhdGZvcndhcmRlci9mb3J3YXJkZXIuZ28=) | `82.44% <88.88%> (+1.26%)` | :arrow_up: |
| [pkg/autoscaler/statforwarder/processor.go](https://codecov.io/gh/knative/serving/pull/10060/diff?src=pr&el=tree#diff-cGtnL2F1dG9zY2FsZXIvc3RhdGZvcndhcmRlci9wcm9jZXNzb3IuZ28=) | `83.33% <100.00%> (ø)` | |
| [pkg/activator/net/revision\_backends.go](https://codecov.io/gh/knative/serving/pull/10060/diff?src=pr&el=tree#diff-cGtnL2FjdGl2YXRvci9uZXQvcmV2aXNpb25fYmFja2VuZHMuZ28=) | `91.40% <0.00%> (+0.90%)` | :arrow_up: |
| [pkg/reconciler/autoscaling/kpa/scaler.go](https://codecov.io/gh/knative/serving/pull/10060/diff?src=pr&el=tree#diff-cGtnL3JlY29uY2lsZXIvYXV0b3NjYWxpbmcva3BhL3NjYWxlci5nbw==) | `90.00% <0.00%> (+1.42%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative/serving/pull/10060?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/serving/pull/10060?src=pr&el=footer). Last update [63b0d15...5ac3b70](https://codecov.io/gh/knative/serving/pull/10060?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Patch:
diff --git a/pkg/autoscaler/statforwarder/forwarder.go b/pkg/autoscaler/statforwarder/forwarder.go
index 5894fbb75..15f0527d0 100644
--- a/pkg/autoscaler/statforwarder/forwarder.go
+++ b/pkg/autoscaler/statforwarder/forwarder.go
@@ -166,12 +166,13 @@ func (f *Forwarder) filterFunc(namespace string) func(interface{}) bool {
 		}
 
 		// The holder identity is in format of <POD-NAME>_<POD-IP>.
-		ip, err := f.extractIP(*l.Spec.HolderIdentity)
+		holder := *l.Spec.HolderIdentity
+		_, err := f.extractIP(holder)
 		if err != nil {
 			f.logger.Warn("Found invalid Lease holder identify ", *l.Spec.HolderIdentity)
 			return false
 		}
-		if p := f.getProcessor(l.Name); p != nil && p.ip == ip {
+		if p := f.getProcessor(l.Name); p != nil && p.holder == holder {
 			// Already up-to-date.
 			return false
 		}
@@ -195,15 +196,16 @@ func (f *Forwarder) leaseUpdated(obj interface{}) {
 	// Close existing connection if there's one. The target address won't
 	// be the same as the IP has changed.
 	f.shutdown(f.getProcessor(n))
+	holder := *l.Spec.HolderIdentity
 	// The map should have the IP because of the operations in the filter when queueing.
-	ip, ok := f.id2ip[*l.Spec.HolderIdentity]
+	ip, ok := f.id2ip[holder]
 	if !ok {
 		// This shouldn't happen because we store the value into the map in the filter
 		// when queueing. Add a log here in case something unexpected happens.
-		f.logger.Warn("IP not found in cached map for ", *l.Spec.HolderIdentity)
+		f.logger.Warn("IP not found in cached map for ", holder)
 		return
 	}
-	f.setProcessor(n, f.createProcessor(ns, n, ip))
+	f.setProcessor(n, f.createProcessor(ns, n, ip, holder))
 
 	if ip != f.selfIP {
 		// Skip creating/updating Service and Endpoints if not the leader.
@@ -333,17 +335,17 @@ func (f *Forwarder) setProcessor(bkt string, p *bucketProcessor) {
 	f.processors[bkt] = p
 }
 
-func (f *Forwarder) createProcessor(ns, bkt, ip string) *bucketProcessor {
+func (f *Forwarder) createProcessor(ns, bkt, ip, holder string) *bucketProcessor {
 	if ip == f.selfIP {
 		return &bucketProcessor{
 			bkt:    bkt,
 			logger: f.logger.With(zap.String("bucket", bkt)),
-			ip:     ip,
+			holder: holder,
 			accept: f.accept,
 		}
 	}
 
-	return newForwardProcessor(f.logger.With(zap.String("bucket", bkt)), bkt, ip,
+	return newForwardProcessor(f.logger.With(zap.String("bucket", bkt)), bkt, holder,
 		fmt.Sprintf("ws://%s:%d", ip, autoscalerPort),
 		fmt.Sprintf("ws://%s.%s.%s", bkt, ns, svcURLSuffix))
 }
diff --git a/pkg/autoscaler/statforwarder/processor.go b/pkg/autoscaler/statforwarder/processor.go
index 5e32a099f..a2f6402a0 100644
--- a/pkg/autoscaler/statforwarder/processor.go
+++ b/pkg/autoscaler/statforwarder/processor.go
@@ -39,8 +39,8 @@ type bucketProcessor struct {
 	logger *zap.SugaredLogger
 	// The name of the bucket
 	bkt string
-	// ip is the Pod IP extracted from the HolderIdentity.
-	ip       string
+	// holder is the HolderIdentity of a Lease for a bucket.
+	holder   string
 	svcDNS   string
 	connLock sync.RWMutex
 	// conn is the WebSocket connection to the holder pod.
@@ -50,7 +50,7 @@ type bucketProcessor struct {
 	accept statProcessor
 }
 
-func newForwardProcessor(logger *zap.SugaredLogger, bkt, ip, podDNS, svcDNS string) *bucketProcessor {
+func newForwardProcessor(logger *zap.SugaredLogger, bkt, holder, podDNS, svcDNS string) *bucketProcessor {
 	// First try to connect via Pod IP address synchronously. If the connection can
 	// not be established within `establishTimeout`, we assume the pods can not be
 	// accessed by IP address. Then try to connect via Pod IP address synchronously.
@@ -63,7 +63,7 @@ func newForwardProcessor(logger *zap.SugaredLogger, bkt, ip, podDNS, svcDNS stri
 	return &bucketProcessor{
 		logger: logger,
 		bkt:    bkt,
-		ip:     ip,
+		holder: holder,
 		conn:   c,
 		svcDNS: svcDNS,
 	}
@@ -89,7 +89,7 @@ func (p *bucketProcessor) process(sm asmetrics.StatMessage) error {
 		return nil
 	}
 
-	l.Debugf("Forward stat of bucket %s to the holder %s", p.bkt, p.ip)
+	l.Debugf("Forward stat of bucket %s to the holder %s", p.bkt, p.holder)
 	wsms := asmetrics.ToWireStatMessages([]asmetrics.StatMessage{sm})
 	b, err := wsms.Marshal()
 	if err != nil {
diff --git a/pkg/reconciler/route/traffic/rollout.go b/pkg/reconciler/route/traffic/rollout.go
index 744464490..61ffb5256 100644
--- a/pkg/reconciler/route/traffic/rollout.go
+++ b/pkg/reconciler/route/traffic/rollout.go
@@ -21,7 +21,9 @@ limitations under the License.
 
 package traffic
 
-import "sort"
+import (
+	"sort"
+)
 
 // Rollout encapsulates the current rollout state of the system.
 // Since the route might reference more than one configuration.
@@ -112,7 +114,8 @@ func (cur *Rollout) Step(prev *Rollout) *Rollout {
 				i++
 			case ccfgs[i].ConfigurationName == pcfgs[j].ConfigurationName:
 				// Config might have 0% traffic assigned, if it is a tag only route (i.e.
-				// receives no traffic via default tag).
+				// receives no traffic via default tag). So just skip it from the rollout
+				// altogether.
 				if ccfgs[i].Percent != 0 {
 					ret = append(ret, *stepConfig(ccfgs[i], pcfgs[j]))
 				}
@@ -120,7 +123,10 @@ func (cur *Rollout) Step(prev *Rollout) *Rollout {
 				j++
 			case ccfgs[i].ConfigurationName < pcfgs[j].ConfigurationName:
 				// A new config, has been added. No action for rollout though.
-				ret = append(ret, *ccfgs[i])
+				// Keep it if it will receive traffic.
+				if ccfgs[i].Percent != 0 {
+					ret = append(ret, *ccfgs[i])
+				}
 				i++
 			default: // cur > prev.
 				// A config has been removed during this update.
@@ -147,7 +153,9 @@ func stepConfig(goal, prev *ConfigurationRollout) *ConfigurationRollout {
 		Percent:           goal.Percent,
 	}
 	// goal will always have just one revision in the list – the current desired revision.
-	if goal.Revisions[0].RevisionName == prev.Revisions[pc-1].RevisionName {
+	// If it matches the last revision of the previous rollout state (or there were no revisions)
+	// then no new rollout hast begun for this configuration.
+	if len(prev.Revisions) == 0 || goal.Revisions[0].RevisionName == prev.Revisions[pc-1].RevisionName {
 		// TODO(vagababov): here would go the logic to compute new percentages for the rollout,
 		// i.e step function, so return value will change, depending on that.
 		// TODO(vagababov): percentage might change, so this should trigger recompute of existing


Test patch:
diff --git a/pkg/autoscaler/statforwarder/forwarder_test.go b/pkg/autoscaler/statforwarder/forwarder_test.go
index 44398dc67..4d7fb930b 100644
--- a/pkg/autoscaler/statforwarder/forwarder_test.go
+++ b/pkg/autoscaler/statforwarder/forwarder_test.go
@@ -439,7 +439,7 @@ func TestProcess(t *testing.T) {
 	if err := wait.PollImmediate(10*time.Millisecond, 10*time.Second, func() (bool, error) {
 		p1 := f.getProcessor(bucket1)
 		p2 := f.getProcessor(bucket2)
-		return p1 != nil && p2 != nil && p1.ip == testIP1 && p2.ip == testIP2, nil
+		return p1 != nil && p2 != nil && p1.holder == testHolder1 && p2.holder == testHolder2, nil
 	}); err != nil {
 		t.Fatalf("Timeout waiting f.processors got updated")
 	}
diff --git a/pkg/reconciler/route/traffic/rollout_test.go b/pkg/reconciler/route/traffic/rollout_test.go
index 929a304d0..6e4a0b5f5 100644
--- a/pkg/reconciler/route/traffic/rollout_test.go
+++ b/pkg/reconciler/route/traffic/rollout_test.go
@@ -27,7 +27,7 @@ import (
 	"github.com/google/go-cmp/cmp"
 )
 
-func TestRoll(t *testing.T) {
+func TestStep(t *testing.T) {
 	tests := []struct {
 		name            string
 		prev, cur, want *Rollout
@@ -36,6 +36,34 @@ func TestRoll(t *testing.T) {
 		cur:  &Rollout{},
 		prev: nil,
 		want: &Rollout{},
+	}, {
+		name: "prev is empty",
+		cur: &Rollout{
+			Configurations: []ConfigurationRollout{{
+				ConfigurationName: "mick",
+				Percent:           100,
+				Revisions: []RevisionRollout{{
+					RevisionName: "let-it-bleed",
+					Percent:      100,
+				}},
+			}},
+		},
+		prev: &Rollout{
+			Configurations: []ConfigurationRollout{{
+				ConfigurationName: "mick",
+				Percent:           100,
+			}},
+		},
+		want: &Rollout{
+			Configurations: []ConfigurationRollout{{
+				ConfigurationName: "mick",
+				Percent:           100,
+				Revisions: []RevisionRollout{{
+					RevisionName: "let-it-bleed",
+					Percent:      100,
+				}},
+			}},
+		},
 	}, {
 		name: "simplest, same",
 		cur: &Rollout{
@@ -68,6 +96,68 @@ func TestRoll(t *testing.T) {
 				}},
 			}},
 		},
+	}, {
+		name: "when new revision becomes 0%",
+		cur: &Rollout{
+			Configurations: []ConfigurationRollout{{
+				ConfigurationName: "charlie",
+				Percent:           0,
+				Revisions: []RevisionRollout{{
+					RevisionName: "aftermath",
+					Percent:      0,
+				}},
+			}},
+		},
+		prev: &Rollout{
+			Configurations: []ConfigurationRollout{{
+				ConfigurationName: "charlie",
+				Percent:           100,
+				Revisions: []RevisionRollout{{
+					RevisionName: "your-satanic-majesties-request",
+					Percent:      100,
+				}},
+			}},
+		},
+		want: &Rollout{},
+	}, {
+		name: "when new config is added but it's 0%",
+		cur: &Rollout{
+			Configurations: []ConfigurationRollout{{
+				ConfigurationName: "charlie",
+				Percent:           0,
+				Revisions: []RevisionRollout{{
+					RevisionName: "aftermath",
+					Percent:      0,
+				}},
+			}, {
+				ConfigurationName: "mick",
+				Percent:           100,
+				Revisions: []RevisionRollout{{
+					RevisionName: "between-the-buttons",
+					Percent:      100,
+				}},
+			}},
+		},
+		prev: &Rollout{
+			Configurations: []ConfigurationRollout{{
+				ConfigurationName: "mick",
+				Percent:           100,
+				Revisions: []RevisionRollout{{
+					RevisionName: "between-the-buttons",
+					Percent:      100,
+				}},
+			}},
+		},
+		want: &Rollout{
+			Configurations: []ConfigurationRollout{{
+				ConfigurationName: "mick",
+				Percent:           100,
+				Revisions: []RevisionRollout{{
+					RevisionName: "between-the-buttons",
+					Percent:      100,
+				}},
+			}},
+		},
 	}, {
 		name: "simplest, roll",
 		cur: &Rollout{