pin deps to 0.21

go.mod

@@ -27,9 +27,9 @@ require (
 	k8s.io/client-go v0.19.7
 	k8s.io/code-generator v0.19.7
 	k8s.io/kube-openapi v0.0.0-20200805222855-6aeccd4b50c6
-	knative.dev/eventing v0.21.1-0.20210302004649-f22066824960
+	knative.dev/eventing v0.21.0
 	knative.dev/hack v0.0.0-20210203173706-8368e1f6eacf
-	knative.dev/pkg v0.0.0-20210226182947-9039dc189ced
+	knative.dev/pkg v0.0.0-20210216013737-584933f8280b
 )
 
 replace (

go.sum

@@ -1217,15 +1217,14 @@ k8s.io/kube-openapi v0.0.0-20200805222855-6aeccd4b50c6 h1:+WnxoVtG8TMiudHBSEtrVL
 k8s.io/kube-openapi v0.0.0-20200805222855-6aeccd4b50c6/go.mod h1:UuqjUnNftUyPE5H64/qeyjQoUZhGpeFDVdxjTeEVN2o=
 k8s.io/utils v0.0.0-20200729134348-d5654de09c73 h1:uJmqzgNWG7XyClnU/mLPBWwfKKF1K8Hf8whTseBgJcg=
 k8s.io/utils v0.0.0-20200729134348-d5654de09c73/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
-knative.dev/eventing v0.21.1-0.20210302004649-f22066824960 h1:E5lX5F6PNv2GwBcrJyrq+uHls2SnKLX4TLH1BeLq6j4=
-knative.dev/eventing v0.21.1-0.20210302004649-f22066824960/go.mod h1:KtN2FEhPzxyqXKxpJufdq0lzL3Iu+5qjFqBal8Lg2tQ=
+knative.dev/eventing v0.21.0 h1:oZoL0lXXslUClNvoi5mLO2YI6zn2C8qg80mz8Y6Vmkg=
+knative.dev/eventing v0.21.0/go.mod h1:JjbVEOTJJHqo9CTxbTfrMn018hG8fOr3UfBoCJ7KWaA=
 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/pkg v0.0.0-20210215165523-84c98f3c3e7a/go.mod h1:TJSdebQOWX5N2bszohOYVi0H1QtXbtlYLuMghAFBMhY=
-knative.dev/pkg v0.0.0-20210225221047-07b5ddfaea60/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=
-knative.dev/reconciler-test v0.0.0-20210223181346-209f32c8cdc4/go.mod h1:RP/K5xJylB72Go6eAsXYEsQHp4zCCNMNjmsqhvq7wko=
+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/reconciler-test v0.0.0-20210216030508-77f50054d024/go.mod h1:RP/K5xJylB72Go6eAsXYEsQHp4zCCNMNjmsqhvq7wko=
 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,9 +271,6 @@ 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,9 +363,6 @@ 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,13 +152,6 @@ 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,9 +35,6 @@ 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,6 +21,7 @@ import (
 	"errors"
 	"fmt"
 	"strings"
+	"sync"
 
 	"go.opencensus.io/resource"
 	"go.opencensus.io/stats/view"
@@ -31,14 +32,9 @@ import (
 var (
 	curMetricsExporter view.Exporter
 	curMetricsConfig   *metricsConfig
-	mWorker            *metricsWorker
+	metricsMux         sync.RWMutex
 )
 
-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)
@@ -158,14 +154,28 @@ 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.
-	updateCmd := &updateMetricsConfigWithExporter{
-		ctx:       ctx,
-		newConfig: newConfig,
-		done:      make(chan error),
+	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)
 	}
-	mWorker.c <- updateCmd
-	err = <-updateCmd.done
-	return err
+
+	setCurMetricsConfigUnlocked(newConfig)
+	return nil
 }
 
 // isNewExporterRequired compares the non-nil newConfig against curMetricsConfig. When backend changes,
@@ -218,35 +228,27 @@ func newMetricsExporter(config *metricsConfig, logger *zap.SugaredLogger) (view.
 }
 
 func getCurMetricsExporter() view.Exporter {
-	readCmd := &readExporter{done: make(chan *view.Exporter)}
-	mWorker.c <- readCmd
-	e := <-readCmd.done
-	return *e
+	metricsMux.RLock()
+	defer metricsMux.RUnlock()
+	return curMetricsExporter
 }
 
 func setCurMetricsExporter(e view.Exporter) {
-	setCmd := &setExporter{
-		newExporter: &e,
-		done:        make(chan struct{}),
-	}
-	mWorker.c <- setCmd
-	<-setCmd.done
+	metricsMux.Lock()
+	defer metricsMux.Unlock()
+	curMetricsExporter = e
 }
 
 func getCurMetricsConfig() *metricsConfig {
-	readCmd := &readMetricsConfig{done: make(chan *metricsConfig)}
-	mWorker.c <- readCmd
-	cfg := <-readCmd.done
-	return cfg
+	metricsMux.RLock()
+	defer metricsMux.RUnlock()
+	return curMetricsConfig
 }
 
 func setCurMetricsConfig(c *metricsConfig) {
-	setCmd := &setMetricsConfig{
-		newConfig: c,
-		done:      make(chan struct{}),
-	}
-	mWorker.c <- setCmd
-	<-setCmd.done
+	metricsMux.Lock()
+	defer metricsMux.Unlock()
+	setCurMetricsConfigUnlocked(c)
 }
 
 func setCurMetricsConfigUnlocked(c *metricsConfig) {

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

@@ -82,16 +82,10 @@ 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()
 		}
 	}
 }
@@ -145,7 +139,7 @@ func RegisterResourceView(views ...*view.View) error {
 	return nil
 }
 
-// UnregisterResourceView is similar to view.Unregister(), except that it will
+// UnregisterResourceView is similar to view.Unregiste(), 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,7 +17,6 @@ limitations under the License.
 package handlers
 
 import (
-	"fmt"
 	"net/http"
 	"sync"
 	"time"
@@ -91,11 +90,11 @@ func (d *Drainer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		}
 		return
 	}
-	if isKProbe(r) {
+	if network.IsKProbe(r) {
 		if d.draining() {
 			http.Error(w, "shutting down", http.StatusServiceUnavailable)
 		} else {
-			serveKProbe(w, r)
+			network.ServeKProbe(w, r)
 		}
 		return
 	}
@@ -147,21 +146,3 @@ 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)
-}