upgrade to latest dependencies (#528)

bumping knative.dev/pkg 584933f...9039dc1:
  > 9039dc1 moved release docs to github.com/knative/release (# 2040)
  > 07b5ddf add demoteFunc controller option (# 2033)
  > 8a8de5f Move IsKProbe to drainer, since it's the only place it's used (# 2038)
  > 5367a43 Fix potential deadlock when k8s client is used (# 2031)
  > b5bf07b adding retry event count metric for source (# 2016)
  > 17a36e7 Format markdown (# 2037)
  > 29be1c1 Add the release badge for all of the repos (# 2035)
  > e90ac15 Remove the system.Clock and it's mock in a different location (# 2034)
  > 8a9bf76 Add symmetric filter helped based on OwnerRefable. (# 2032)
  > 29092fe Format markdown (# 2030)
  > d733bb4 Add nightly badged to release tables (# 2029)
  > 7b5bcc3 Use the constant for UserAgent (# 2028)
  > 86a8236 Format markdown (# 2025)
  > 6c2f94a Unregister views to avoid slow oom issue during meter cleanup (# 2005)
  > 7689cce Update common github actions (# 2024)
  > f4650d8 Add releasability badges for each repo (# 2023)
bumping knative.dev/networking 94bfc01...d5e05ad:
  > d5e05ad Update common github actions (# 357)

Signed-off-by: Knative Automation <automation@knative.team>

go.mod

@@ -14,6 +14,6 @@ require (
 	k8s.io/apimachinery v0.19.7
 	k8s.io/client-go v0.19.7
 	knative.dev/hack v0.0.0-20210203173706-8368e1f6eacf
-	knative.dev/networking v0.0.0-20210216014426-94bfc013982b
-	knative.dev/pkg v0.0.0-20210216013737-584933f8280b
+	knative.dev/networking v0.0.0-20210216193011-d5e05ad323f0
+	knative.dev/pkg v0.0.0-20210226182947-9039dc189ced
 )

go.sum

@@ -1192,11 +1192,11 @@ k8s.io/utils v0.0.0-20200729134348-d5654de09c73 h1:uJmqzgNWG7XyClnU/mLPBWwfKKF1K
 k8s.io/utils v0.0.0-20200729134348-d5654de09c73/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 knative.dev/hack v0.0.0-20210203173706-8368e1f6eacf h1:u4cY4jr2LYvhoz/1HBWEPsMiLkm0HMdDTfmmw1RE8zE=
 knative.dev/hack v0.0.0-20210203173706-8368e1f6eacf/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
-knative.dev/networking v0.0.0-20210216014426-94bfc013982b h1:708bHZC2RPjdx8MLH7OASqGP+fhmarJJxVIZoIy+msY=
-knative.dev/networking v0.0.0-20210216014426-94bfc013982b/go.mod h1:Crdn87hxdFd3Jj6PIyrjzGnr8OGHX35k5xo9jlOrjjA=
+knative.dev/networking v0.0.0-20210216193011-d5e05ad323f0 h1:3TwevWoWJ/xlU1Bjzb3vtRRl1NfJWIJTkZb+1eNdW8k=
+knative.dev/networking v0.0.0-20210216193011-d5e05ad323f0/go.mod h1:Crdn87hxdFd3Jj6PIyrjzGnr8OGHX35k5xo9jlOrjjA=
 knative.dev/pkg v0.0.0-20210215165523-84c98f3c3e7a/go.mod h1:TJSdebQOWX5N2bszohOYVi0H1QtXbtlYLuMghAFBMhY=
-knative.dev/pkg v0.0.0-20210216013737-584933f8280b h1:2v+GBBenjPXhtk3KX/YqxCQVBU6o/AaRkkqE58lgFRQ=
-knative.dev/pkg v0.0.0-20210216013737-584933f8280b/go.mod h1:TJSdebQOWX5N2bszohOYVi0H1QtXbtlYLuMghAFBMhY=
+knative.dev/pkg v0.0.0-20210226182947-9039dc189ced h1:OCtTbXsJhiR35qCDxFj9Xc8sU3i87ePbc7YlHCt0al4=
+knative.dev/pkg v0.0.0-20210226182947-9039dc189ced/go.mod h1:TJSdebQOWX5N2bszohOYVi0H1QtXbtlYLuMghAFBMhY=
 pgregory.net/rapid v0.3.3/go.mod h1:UYpPVyjFHzYBGHIxLFoupi8vwk6rXNzRY9OMvVxFIOU=
 rsc.io/binaryregexp v0.2.0/go.mod h1:qTv7/COck+e2FymRvadv62gMdZztPaShugOCi3I+8D8=
 rsc.io/quote/v3 v3.1.0/go.mod h1:yEA65RcK8LyAZtP9Kv3t0HmxON59tX3rD+tICJqUlj0=

vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/reconciler_controller.go

@@ -271,6 +271,9 @@ func NewImpl(ctx {{.contextContext|raw}}, r Interface{{if .hasClass}}, classValu
 			rec.skipStatusUpdates = true
 		}
 		{{- end}}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
 	}
 
 	rec.Recorder = createRecorder(ctx, agentName)

vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/reconciler_reconciler.go

@@ -363,6 +363,9 @@ func NewReconciler(ctx {{.contextContext|raw}}, logger *{{.zapSugaredLogger|raw}
 			rec.skipStatusUpdates = true
 		}
 		{{- end}}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
 	}
 
 	return rec

vendor/knative.dev/pkg/controller/controller.go

@@ -152,6 +152,13 @@ func FilterControllerGK(gk schema.GroupKind) func(obj interface{}) bool {
 	}
 }
 
+// FilterController makes it simple to create FilterFunc's for use with
+// cache.FilteringResourceEventHandler that filter based on the
+// controlling resource.
+func FilterController(r kmeta.OwnerRefable) func(obj interface{}) bool {
+	return FilterControllerGK(r.GetGroupVersionKind().GroupKind())
+}
+
 // FilterWithName makes it simple to create FilterFunc's for use with
 // cache.FilteringResourceEventHandler that filter based on a name.
 func FilterWithName(name string) func(obj interface{}) bool {

vendor/knative.dev/pkg/controller/options.go

@@ -35,6 +35,9 @@ type Options struct {
 	// SkipStatusUpdates configures this reconciler to either do automated status
 	// updates (default) or skip them if this is set to true.
 	SkipStatusUpdates bool
+
+	// DemoteFunc configures the demote function this reconciler uses
+	DemoteFunc func(b reconciler.Bucket)
 }
 
 // OptionsFn is a callback method signature that accepts an Impl and returns

vendor/knative.dev/pkg/metrics/exporter.go

@@ -21,7 +21,6 @@ import (
 	"errors"
 	"fmt"
 	"strings"
-	"sync"
 
 	"go.opencensus.io/resource"
 	"go.opencensus.io/stats/view"
@@ -32,9 +31,14 @@ import (
 var (
 	curMetricsExporter view.Exporter
 	curMetricsConfig   *metricsConfig
-	metricsMux         sync.RWMutex
+	mWorker            *metricsWorker
 )
 
+func init() {
+	mWorker = newMetricsWorker()
+	go mWorker.start()
+}
+
 // SecretFetcher is a function (extracted from SecretNamespaceLister) for fetching
 // a specific Secret. This avoids requiring global or namespace list in controllers.
 type SecretFetcher func(string) (*corev1.Secret, error)
@@ -154,28 +158,14 @@ func UpdateExporter(ctx context.Context, ops ExporterOptions, logger *zap.Sugare
 
 	// Updating the metrics config and the metrics exporters needs to be atomic to
 	// avoid using an outdated metrics config with new exporters.
-	metricsMux.Lock()
-	defer metricsMux.Unlock()
-
-	if isNewExporterRequired(newConfig) {
-		logger.Info("Flushing the existing exporter before setting up the new exporter.")
-		flushGivenExporter(curMetricsExporter)
-		e, f, err := newMetricsExporter(newConfig, logger)
-		if err != nil {
-			logger.Errorw("Failed to update a new metrics exporter based on metric config", zap.Error(err), "config", newConfig)
-			return err
-		}
-		existingConfig := curMetricsConfig
-		curMetricsExporter = e
-		if err := setFactory(f); err != nil {
-			logger.Errorw("Failed to update metrics factory when loading metric config", zap.Error(err), "config", newConfig)
-			return err
-		}
-		logger.Infof("Successfully updated the metrics exporter; old config: %v; new config %v", existingConfig, newConfig)
+	updateCmd := &updateMetricsConfigWithExporter{
+		ctx:       ctx,
+		newConfig: newConfig,
+		done:      make(chan error),
 	}
-
-	setCurMetricsConfigUnlocked(newConfig)
-	return nil
+	mWorker.c <- updateCmd
+	err = <-updateCmd.done
+	return err
 }
 
 // isNewExporterRequired compares the non-nil newConfig against curMetricsConfig. When backend changes,
@@ -228,27 +218,35 @@ func newMetricsExporter(config *metricsConfig, logger *zap.SugaredLogger) (view.
 }
 
 func getCurMetricsExporter() view.Exporter {
-	metricsMux.RLock()
-	defer metricsMux.RUnlock()
-	return curMetricsExporter
+	readCmd := &readExporter{done: make(chan *view.Exporter)}
+	mWorker.c <- readCmd
+	e := <-readCmd.done
+	return *e
 }
 
 func setCurMetricsExporter(e view.Exporter) {
-	metricsMux.Lock()
-	defer metricsMux.Unlock()
-	curMetricsExporter = e
+	setCmd := &setExporter{
+		newExporter: &e,
+		done:        make(chan struct{}),
+	}
+	mWorker.c <- setCmd
+	<-setCmd.done
 }
 
 func getCurMetricsConfig() *metricsConfig {
-	metricsMux.RLock()
-	defer metricsMux.RUnlock()
-	return curMetricsConfig
+	readCmd := &readMetricsConfig{done: make(chan *metricsConfig)}
+	mWorker.c <- readCmd
+	cfg := <-readCmd.done
+	return cfg
 }
 
 func setCurMetricsConfig(c *metricsConfig) {
-	metricsMux.Lock()
-	defer metricsMux.Unlock()
-	setCurMetricsConfigUnlocked(c)
+	setCmd := &setMetricsConfig{
+		newConfig: c,
+		done:      make(chan struct{}),
+	}
+	mWorker.c <- setCmd
+	<-setCmd.done
 }
 
 func setCurMetricsConfigUnlocked(c *metricsConfig) {

vendor/knative.dev/pkg/metrics/metrics_worker.go

@@ -0,0 +1,111 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package metrics
+
+import (
+	"context"
+
+	"go.opencensus.io/stats/view"
+	"go.uber.org/zap"
+	"knative.dev/pkg/logging"
+)
+
+type metricsWorker struct {
+	c chan command
+}
+
+func newMetricsWorker() *metricsWorker {
+	return &metricsWorker{c: make(chan command)}
+}
+
+type command interface {
+	handleCommand(w *metricsWorker)
+}
+
+type readExporter struct {
+	done chan *view.Exporter
+}
+
+type setExporter struct {
+	newExporter *view.Exporter
+	done        chan struct{}
+}
+
+type readMetricsConfig struct {
+	done chan *metricsConfig
+}
+
+type updateMetricsConfigWithExporter struct {
+	ctx       context.Context
+	newConfig *metricsConfig
+	done      chan error
+}
+
+type setMetricsConfig struct {
+	newConfig *metricsConfig
+	done      chan struct{}
+}
+
+func (cmd *readMetricsConfig) handleCommand(w *metricsWorker) {
+	cmd.done <- curMetricsConfig
+}
+
+func (cmd *setMetricsConfig) handleCommand(w *metricsWorker) {
+	setCurMetricsConfigUnlocked(cmd.newConfig)
+	cmd.done <- struct{}{}
+}
+
+func (cmd *updateMetricsConfigWithExporter) handleCommand(w *metricsWorker) {
+	ctx := cmd.ctx
+	logger := logging.FromContext(ctx)
+	if isNewExporterRequired(cmd.newConfig) {
+		logger.Info("Flushing the existing exporter before setting up the new exporter.")
+		flushGivenExporter(curMetricsExporter)
+		e, f, err := newMetricsExporter(cmd.newConfig, logger)
+		if err != nil {
+			logger.Errorw("Failed to update a new metrics exporter based on metric config", zap.Error(err), "config", cmd.newConfig)
+			cmd.done <- err
+			return
+		}
+		existingConfig := curMetricsConfig
+		curMetricsExporter = e
+		if err := setFactory(f); err != nil {
+			logger.Errorw("Failed to update metrics factory when loading metric config", zap.Error(err), "config", cmd.newConfig)
+			cmd.done <- err
+			return
+		}
+		logger.Infof("Successfully updated the metrics exporter; old config: %v; new config %v", existingConfig, cmd.newConfig)
+	}
+	setCurMetricsConfigUnlocked(cmd.newConfig)
+	cmd.done <- nil
+}
+
+func (w *metricsWorker) start() {
+	for {
+		cmd := <-w.c
+		cmd.handleCommand(w)
+	}
+}
+
+func (cmd *setExporter) handleCommand(w *metricsWorker) {
+	curMetricsExporter = *cmd.newExporter
+	cmd.done <- struct{}{}
+}
+
+func (cmd *readExporter) handleCommand(w *metricsWorker) {
+	cmd.done <- &curMetricsExporter
+}

vendor/knative.dev/pkg/metrics/resource_view.go

@@ -82,10 +82,16 @@ func cleanup() {
 	expiryCutoff := allMeters.clock.Now().Add(-1 * maxMeterExporterAge)
 	allMeters.lock.Lock()
 	defer allMeters.lock.Unlock()
+	resourceViews.lock.Lock()
+	defer resourceViews.lock.Unlock()
 	for key, meter := range allMeters.meters {
 		if key != "" && meter.t.Before(expiryCutoff) {
 			flushGivenExporter(meter.e)
+			// Make a copy of views to avoid data races
+			viewsCopy := copyViews(resourceViews.views)
+			meter.m.Unregister(viewsCopy...)
 			delete(allMeters.meters, key)
+			meter.m.Stop()
 		}
 	}
 }
@@ -139,7 +145,7 @@ func RegisterResourceView(views ...*view.View) error {
 	return nil
 }
 
-// UnregisterResourceView is similar to view.Unregiste(), except that it will
+// UnregisterResourceView is similar to view.Unregister(), except that it will
 // unregister the view across all Resources tracked byt he system, rather than
 // simply the default view.
 func UnregisterResourceView(views ...*view.View) {

vendor/knative.dev/pkg/network/handlers/drain.go

@@ -17,6 +17,7 @@ limitations under the License.
 package handlers
 
 import (
+	"fmt"
 	"net/http"
 	"sync"
 	"time"
@@ -90,11 +91,11 @@ func (d *Drainer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		}
 		return
 	}
-	if network.IsKProbe(r) {
+	if isKProbe(r) {
 		if d.draining() {
 			http.Error(w, "shutting down", http.StatusServiceUnavailable)
 		} else {
-			network.ServeKProbe(w, r)
+			serveKProbe(w, r)
 		}
 		return
 	}
@@ -146,3 +147,21 @@ func (d *Drainer) draining() bool {
 	defer d.RUnlock()
 	return d.timer != nil
 }
+
+// isKProbe returns true if the request is a knatvie probe.
+func isKProbe(r *http.Request) bool {
+	return r.Header.Get(network.ProbeHeaderName) == network.ProbeHeaderValue
+}
+
+// serveKProbe serve KProbe requests.
+func serveKProbe(w http.ResponseWriter, r *http.Request) {
+	hh := r.Header.Get(network.HashHeaderName)
+	if hh == "" {
+		http.Error(w,
+			fmt.Sprintf("a probe request must contain a non-empty %q header", network.HashHeaderName),
+			http.StatusBadRequest)
+		return
+	}
+	w.Header().Set(network.HashHeaderName, hh)
+	w.WriteHeader(http.StatusOK)
+}