Update dependencies to v0.20.3 tag

Godeps/Godeps.json

@@ -36,7 +36,7 @@
 		},
 		{
 			"ImportPath": "github.com/Azure/go-autorest/autorest",
-			"Rev": "v0.11.12"
+			"Rev": "v0.11.1"
 		},
 		{
 			"ImportPath": "github.com/Azure/go-autorest/autorest/adal",
@@ -106,6 +106,10 @@
 			"ImportPath": "github.com/davecgh/go-spew",
 			"Rev": "v1.1.1"
 		},
+		{
+			"ImportPath": "github.com/dgrijalva/jwt-go",
+			"Rev": "v3.2.0"
+		},
 		{
 			"ImportPath": "github.com/docker/spdystream",
 			"Rev": "449fdfce4d96"
@@ -460,11 +464,11 @@
 		},
 		{
 			"ImportPath": "k8s.io/api",
-			"Rev": "v0.21.0-alpha.2"
+			"Rev": "v0.20.3"
 		},
 		{
 			"ImportPath": "k8s.io/apimachinery",
-			"Rev": "v0.21.0-alpha.2"
+			"Rev": "v0.20.3"
 		},
 		{
 			"ImportPath": "k8s.io/gengo",

README.md

@@ -82,15 +82,14 @@ We will backport bugfixes--but not new features--into older versions of
 
 #### Compatibility matrix
 
-|                               | Kubernetes 1.15 | Kubernetes 1.16 | Kubernetes 1.17 | Kubernetes 1.18 | Kubernetes 1.19 | Kubernetes 1.20 |
-|-------------------------------|-----------------|-----------------|-----------------|-----------------|-----------------|-----------------|
-| `kubernetes-1.15.0`           | ✓               | +-              | +-              | +-              | +-              | +-              |
-| `kubernetes-1.16.0`           | +-              | ✓               | +-              | +-              | +-              | +-              |
-| `kubernetes-1.17.0`/`v0.17.0` | +-              | +-              | ✓               | +-              | +-              | +-              |
-| `kubernetes-1.18.0`/`v0.18.0` | +-              | +-              | +-              | ✓               | +-              | +-              |
-| `kubernetes-1.19.0`/`v0.19.0` | +-              | +-              | +-              | +-              | ✓               | +-              |
-| `kubernetes-1.20.0`/`v0.20.0` | +-              | +-              | +-              | +-              | +-              | ✓               |
-| `HEAD`                        | +-              | +-              | +-              | +-              | +-              | +-              |
+|                               | Kubernetes 1.15 | Kubernetes 1.16 | Kubernetes 1.17 | Kubernetes 1.18 | Kubernetes 1.19 |
+|-------------------------------|-----------------|-----------------|-----------------|-----------------|-----------------|
+| `kubernetes-1.15.0`           | ✓               | +-              | +-              | +-              | +-              |
+| `kubernetes-1.16.0`           | +-              | ✓               | +-              | +-              | +-              |
+| `kubernetes-1.17.0`/`v0.17.0` | +-              | +-              | ✓               | +-              | +-              |
+| `kubernetes-1.18.0`/`v0.18.0` | +-              | +-              | +-              | ✓               | +-              |
+| `kubernetes-1.19.0`/`v0.19.0` | +-              | +-              | +-              | +-              | ✓               |
+| `HEAD`                        | +-              | +-              | +-              | +-              | +-              |
 
 Key:
 
@@ -124,11 +123,10 @@ between client-go versions.
 | `release-10.0` | Kubernetes main repo, 1.13 branch    | =-                            |
 | `release-11.0` | Kubernetes main repo, 1.14 branch    | =-                            |
 | `release-12.0` | Kubernetes main repo, 1.15 branch    | =-                            |
-| `release-13.0` | Kubernetes main repo, 1.16 branch    | =-                            |
+| `release-13.0` | Kubernetes main repo, 1.16 branch    | ✓                             |
 | `release-14.0` | Kubernetes main repo, 1.17 branch    | ✓                             |
 | `release-1.18` | Kubernetes main repo, 1.18 branch    | ✓                             |
 | `release-1.19` | Kubernetes main repo, 1.19 branch    | ✓                             |
-| `release-1.20` | Kubernetes main repo, 1.20 branch    | ✓                             |
 | client-go HEAD | Kubernetes main repo, master branch  | ✓                             |
 
 Key:
@@ -178,7 +176,7 @@ you care about backwards compatibility.
 Use go1.11+ and fetch the desired version using the `go get` command. For example:
 
 ```
-go get k8s.io/client-go@v0.20.0
+go get k8s.io/client-go@v0.19.0
 ```
 
 See [INSTALL.md](/INSTALL.md) for detailed instructions.

go.mod

@@ -6,7 +6,7 @@ go 1.15
 
 require (
 	cloud.google.com/go v0.54.0 // indirect
-	github.com/Azure/go-autorest/autorest v0.11.12
+	github.com/Azure/go-autorest/autorest v0.11.1
 	github.com/Azure/go-autorest/autorest/adal v0.9.5
 	github.com/davecgh/go-spew v1.1.1
 	github.com/evanphx/json-patch v4.9.0+incompatible
@@ -26,14 +26,14 @@ require (
 	golang.org/x/net v0.0.0-20201110031124-69a78807bb2b
 	golang.org/x/oauth2 v0.0.0-20200107190931-bf48bf16ab8d
 	golang.org/x/time v0.0.0-20200630173020-3af7569d3a1e
-	k8s.io/api v0.21.0-alpha.2
-	k8s.io/apimachinery v0.21.0-alpha.2
+	k8s.io/api v0.20.3
+	k8s.io/apimachinery v0.20.3
 	k8s.io/klog/v2 v2.4.0
 	k8s.io/utils v0.0.0-20201110183641-67b214c5f920
 	sigs.k8s.io/yaml v1.2.0
 )
 
 replace (
-	k8s.io/api => k8s.io/api v0.21.0-alpha.2
-	k8s.io/apimachinery => k8s.io/apimachinery v0.21.0-alpha.2
+	k8s.io/api => k8s.io/api v0.20.3
+	k8s.io/apimachinery => k8s.io/apimachinery v0.20.3
 )

go.sum

@@ -24,12 +24,16 @@ cloud.google.com/go/storage v1.6.0/go.mod h1:N7U0C8pVQ/+NIKOBQyamJIeKQKkZ+mxpohl
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
 github.com/Azure/go-autorest v14.2.0+incompatible h1:V5VMDjClD3GiElqLWO7mz2MxNAK/vTfRHdAubSIPRgs=
 github.com/Azure/go-autorest v14.2.0+incompatible/go.mod h1:r+4oMnoxhatjLLJ6zxSWATqVooLgysK6ZNox3g/xq24=
-github.com/Azure/go-autorest/autorest v0.11.12 h1:gI8ytXbxMfI+IVbI9mP2JGCTXIuhHLgRlvQ9X4PsnHE=
-github.com/Azure/go-autorest/autorest v0.11.12/go.mod h1:eipySxLmqSyC5s5k1CLupqet0PSENBEDP93LQ9a8QYw=
+github.com/Azure/go-autorest/autorest v0.11.1 h1:eVvIXUKiTgv++6YnWb42DUA1YL7qDugnKP0HljexdnQ=
+github.com/Azure/go-autorest/autorest v0.11.1/go.mod h1:JFgpikqFJ/MleTTxwepExTKnFUKKszPS8UavbQYUMuw=
+github.com/Azure/go-autorest/autorest/adal v0.9.0 h1:SigMbuFNuKgc1xcGhaeapbh+8fgsu+GxgDRFyg7f5lM=
+github.com/Azure/go-autorest/autorest/adal v0.9.0/go.mod h1:/c022QCutn2P7uY+/oQWWNcK9YU+MH96NgK+jErpbcg=
 github.com/Azure/go-autorest/autorest/adal v0.9.5 h1:Y3bBUV4rTuxenJJs41HU3qmqsb+auo+a3Lz+PlJPpL0=
 github.com/Azure/go-autorest/autorest/adal v0.9.5/go.mod h1:B7KF7jKIeC9Mct5spmyCB/A8CG/sEz1vwIRGv/bbw7A=
 github.com/Azure/go-autorest/autorest/date v0.3.0 h1:7gUk1U5M/CQbp9WoqinNzJar+8KY+LPI6wiWrP/myHw=
 github.com/Azure/go-autorest/autorest/date v0.3.0/go.mod h1:BI0uouVdmngYNUzGWeSYnokU+TrmwEsOqdt8Y6sso74=
+github.com/Azure/go-autorest/autorest/mocks v0.4.0 h1:z20OWOSG5aCye0HEkDp6TPmP17ZcfeMxPi6HnSALa8c=
+github.com/Azure/go-autorest/autorest/mocks v0.4.0/go.mod h1:LTp+uSrOhSkaKrUy935gNZuuIPPVsHlr9DSOxSayd+k=
 github.com/Azure/go-autorest/autorest/mocks v0.4.1 h1:K0laFcLE6VLTOwNgSxaGbUcLPuGXlNkbVvq4cW4nIHk=
 github.com/Azure/go-autorest/autorest/mocks v0.4.1/go.mod h1:LTp+uSrOhSkaKrUy935gNZuuIPPVsHlr9DSOxSayd+k=
 github.com/Azure/go-autorest/logger v0.2.0 h1:e4RVHVZKC5p6UANLJHkM4OfR1UKZPj8Wt8Pcx+3oqrE=
@@ -50,6 +54,8 @@ github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDk
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/dgrijalva/jwt-go v3.2.0+incompatible h1:7qlOGliEKZXTDg6OTjfoBKDXWrumCAMpl/TFQ4/5kLM=
+github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
 github.com/docker/spdystream v0.0.0-20160310174837-449fdfce4d96 h1:cenwrSVm+Z7QLSV/BsnenAOcDXdX4cMv4wP0B/5QbPg=
 github.com/docker/spdystream v0.0.0-20160310174837-449fdfce4d96/go.mod h1:Qh8CwZgvJUkLughtfhJv5dyTYa91l1fOUCrgjqmcifM=
 github.com/docopt/docopt-go v0.0.0-20180111231733-ee0de3bc6815/go.mod h1:WwZ+bS3ebgob9U8Nd0kOddGdZWjyMGR8Wziv+TBNwSE=
@@ -427,8 +433,8 @@ honnef.co/go/tools v0.0.0-20190418001031-e561f6794a2a/go.mod h1:rf3lG4BRIbNafJWh
 honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.1-2019.2.3/go.mod h1:a3bituU0lyd329TUQxRnasdCoJDkEUEAqEt0JzvZhAg=
 honnef.co/go/tools v0.0.1-2020.1.3/go.mod h1:X/FiERA/W4tHapMX5mGpAtMSVEeEUOyHaw9vFzvIQ3k=
-k8s.io/api v0.21.0-alpha.2/go.mod h1:v7pfXujF1vdxa3lGFot+93rftqAVF32CTmPTp5Ub7EE=
-k8s.io/apimachinery v0.21.0-alpha.2/go.mod h1:WlLqWAHZGg07AeltaI0MV5uk1Omp8xaN0JGLY6gkRpU=
+k8s.io/api v0.20.3/go.mod h1:9/N1PKffb5ioImFknLewyktqAKCAh6xnhc7JFMhX6zg=
+k8s.io/apimachinery v0.20.3/go.mod h1:WlLqWAHZGg07AeltaI0MV5uk1Omp8xaN0JGLY6gkRpU=
 k8s.io/gengo v0.0.0-20200413195148-3a45101e95ac/go.mod h1:ezvh/TsK7cY6rbqRK0oQQ8IAqLxYwwyPxAX1Pzy0ii0=
 k8s.io/klog/v2 v2.0.0/go.mod h1:PBfzABfn139FHAV07az/IF9Wp1bkk3vpT2XSJ76fSDE=
 k8s.io/klog/v2 v2.4.0 h1:7+X0fUguPyrKEC4WjH8iGDg3laWgMo5tMnRTIGTTxGQ=

plugin/pkg/client/auth/exec/exec.go

@@ -18,6 +18,7 @@ package exec
 
 import (
 	"bytes"
+	"context"
 	"crypto/tls"
 	"crypto/x509"
 	"errors"
@@ -51,6 +52,7 @@ import (
 )
 
 const execInfoEnv = "KUBERNETES_EXEC_INFO"
+const onRotateListWarningLength = 1000
 const installHintVerboseHelp = `
 
 It looks like you are trying to use a client-go credential plugin that is not installed.
@@ -175,12 +177,6 @@ func newAuthenticator(c *cache, config *api.ExecConfig, cluster *clientauthentic
 		return nil, fmt.Errorf("exec plugin: invalid apiVersion %q", config.APIVersion)
 	}
 
-	connTracker := connrotation.NewConnectionTracker()
-	defaultDialer := connrotation.NewDialerWithTracker(
-		(&net.Dialer{Timeout: 30 * time.Second, KeepAlive: 30 * time.Second}).DialContext,
-		connTracker,
-	)
-
 	a := &Authenticator{
 		cmd:                config.Command,
 		args:               config.Args,
@@ -200,9 +196,6 @@ func newAuthenticator(c *cache, config *api.ExecConfig, cluster *clientauthentic
 		interactive: terminal.IsTerminal(int(os.Stdout.Fd())),
 		now:         time.Now,
 		environ:     os.Environ,
-
-		defaultDialer: defaultDialer,
-		connTracker:   connTracker,
 	}
 
 	for _, env := range config.Env {
@@ -236,11 +229,6 @@ type Authenticator struct {
 	now         func() time.Time
 	environ     func() []string
 
-	// defaultDialer is used for clients which don't specify a custom dialer
-	defaultDialer *connrotation.Dialer
-	// connTracker tracks all connections opened that we need to close when rotating a client certificate
-	connTracker *connrotation.ConnectionTracker
-
 	// Cached results.
 	//
 	// The mutex also guards calling the plugin. Since the plugin could be
@@ -248,6 +236,8 @@ type Authenticator struct {
 	mu          sync.Mutex
 	cachedCreds *credentials
 	exp         time.Time
+
+	onRotateList []func()
 }
 
 type credentials struct {
@@ -276,12 +266,20 @@ func (a *Authenticator) UpdateTransportConfig(c *transport.Config) error {
 	}
 	c.TLS.GetCert = a.cert
 
-	var d *connrotation.Dialer
+	var dial func(ctx context.Context, network, addr string) (net.Conn, error)
 	if c.Dial != nil {
-		// if c has a custom dialer, we have to wrap it
-		d = connrotation.NewDialerWithTracker(c.Dial, a.connTracker)
+		dial = c.Dial
 	} else {
-		d = a.defaultDialer
+		dial = (&net.Dialer{Timeout: 30 * time.Second, KeepAlive: 30 * time.Second}).DialContext
+	}
+	d := connrotation.NewDialer(dial)
+
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	a.onRotateList = append(a.onRotateList, d.CloseAll)
+	onRotateListLength := len(a.onRotateList)
+	if onRotateListLength > onRotateListWarningLength {
+		klog.Warningf("constructing many client instances from the same exec auth config can cause performance problems during cert rotation and can exhaust available network connections; %d clients constructed calling %q", onRotateListLength, a.cmd)
 	}
 
 	c.Dial = d.DialContext
@@ -460,7 +458,9 @@ func (a *Authenticator) refreshCredsLocked(r *clientauthentication.Response) err
 		if oldCreds.cert != nil && oldCreds.cert.Leaf != nil {
 			metrics.ClientCertRotationAge.Observe(time.Now().Sub(oldCreds.cert.Leaf.NotBefore))
 		}
-		a.connTracker.CloseAll()
+		for _, onRotate := range a.onRotateList {
+			onRotate()
+		}
 	}
 
 	expiry := time.Time{}

rest/plugin.go

@@ -47,13 +47,6 @@ type AuthProviderConfigPersister interface {
 	Persist(map[string]string) error
 }
 
-type noopPersister struct{}
-
-func (n *noopPersister) Persist(_ map[string]string) error {
-	// no operation persister
-	return nil
-}
-
 // All registered auth provider plugins.
 var pluginsLock sync.Mutex
 var plugins = make(map[string]Factory)
@@ -76,8 +69,5 @@ func GetAuthProvider(clusterAddress string, apc *clientcmdapi.AuthProviderConfig
 	if !ok {
 		return nil, fmt.Errorf("no Auth Provider found for name %q", apc.Name)
 	}
-	if persister == nil {
-		persister = &noopPersister{}
-	}
 	return p(clusterAddress, apc.Config, persister)
 }

rest/plugin_test.go

@@ -171,28 +171,6 @@ func TestAuthPluginPersist(t *testing.T) {
 
 }
 
-func Test_WhenNilPersister_NoOpPersisterIsAssigned(t *testing.T) {
-
-	if err := RegisterAuthProviderPlugin("anyPlugin", pluginPersistProvider); err != nil {
-		t.Errorf("unexpected error: failed to register 'anyPlugin': %v", err)
-	}
-	cfg := &clientcmdapi.AuthProviderConfig{
-		Name:   "anyPlugin",
-		Config: nil,
-	}
-	plugin, err := GetAuthProvider("127.0.0.1", cfg, nil)
-	if err != nil {
-		t.Errorf("unexpected error: failed to get 'anyPlugin': %v", err)
-	}
-
-	anyPlugin := plugin.(*pluginPersist)
-
-	if _, ok := anyPlugin.persister.(*noopPersister); !ok {
-		t.Errorf("expected to be No Operation persister")
-	}
-
-}
-
 // emptyTransport provides an empty http.Response with an initialized header
 // to allow wrapping RoundTrippers to set header values.
 type emptyTransport struct{}

tools/cache/delta_fifo.go

@@ -26,6 +26,54 @@ import (
 	"k8s.io/klog/v2"
 )
 
+// NewDeltaFIFO returns a Queue which can be used to process changes to items.
+//
+// keyFunc is used to figure out what key an object should have. (It is
+// exposed in the returned DeltaFIFO's KeyOf() method, with additional handling
+// around deleted objects and queue state).
+//
+// 'knownObjects' may be supplied to modify the behavior of Delete,
+// Replace, and Resync.  It may be nil if you do not need those
+// modifications.
+//
+// TODO: consider merging keyLister with this object, tracking a list of
+//       "known" keys when Pop() is called. Have to think about how that
+//       affects error retrying.
+// NOTE: It is possible to misuse this and cause a race when using an
+//       external known object source.
+//       Whether there is a potential race depends on how the consumer
+//       modifies knownObjects. In Pop(), process function is called under
+//       lock, so it is safe to update data structures in it that need to be
+//       in sync with the queue (e.g. knownObjects).
+//
+//       Example:
+//       In case of sharedIndexInformer being a consumer
+//       (https://github.com/kubernetes/kubernetes/blob/0cdd940f/staging/
+//       src/k8s.io/client-go/tools/cache/shared_informer.go#L192),
+//       there is no race as knownObjects (s.indexer) is modified safely
+//       under DeltaFIFO's lock. The only exceptions are GetStore() and
+//       GetIndexer() methods, which expose ways to modify the underlying
+//       storage. Currently these two methods are used for creating Lister
+//       and internal tests.
+//
+// Also see the comment on DeltaFIFO.
+//
+// Warning: This constructs a DeltaFIFO that does not differentiate between
+// events caused by a call to Replace (e.g., from a relist, which may
+// contain object updates), and synthetic events caused by a periodic resync
+// (which just emit the existing object). See https://issue.k8s.io/86015 for details.
+//
+// Use `NewDeltaFIFOWithOptions(DeltaFIFOOptions{..., EmitDeltaTypeReplaced: true})`
+// instead to receive a `Replaced` event depending on the type.
+//
+// Deprecated: Equivalent to NewDeltaFIFOWithOptions(DeltaFIFOOptions{KeyFunction: keyFunc, KnownObjects: knownObjects})
+func NewDeltaFIFO(keyFunc KeyFunc, knownObjects KeyListerGetter) *DeltaFIFO {
+	return NewDeltaFIFOWithOptions(DeltaFIFOOptions{
+		KeyFunction:  keyFunc,
+		KnownObjects: knownObjects,
+	})
+}
+
 // DeltaFIFOOptions is the configuration parameters for DeltaFIFO. All are
 // optional.
 type DeltaFIFOOptions struct {
@@ -51,6 +99,25 @@ type DeltaFIFOOptions struct {
 	EmitDeltaTypeReplaced bool
 }
 
+// NewDeltaFIFOWithOptions returns a Queue which can be used to process changes to
+// items. See also the comment on DeltaFIFO.
+func NewDeltaFIFOWithOptions(opts DeltaFIFOOptions) *DeltaFIFO {
+	if opts.KeyFunction == nil {
+		opts.KeyFunction = MetaNamespaceKeyFunc
+	}
+
+	f := &DeltaFIFO{
+		items:        map[string]Deltas{},
+		queue:        []string{},
+		keyFunc:      opts.KeyFunction,
+		knownObjects: opts.KnownObjects,
+
+		emitDeltaTypeReplaced: opts.EmitDeltaTypeReplaced,
+	}
+	f.cond.L = &f.lock
+	return f
+}
+
 // DeltaFIFO is like FIFO, but differs in two ways.  One is that the
 // accumulator associated with a given object's key is not that object
 // but rather a Deltas, which is a slice of Delta values for that
@@ -61,11 +128,8 @@ type DeltaFIFOOptions struct {
 // Deleted if the older Deleted's object is a
 // DeletedFinalStateUnknown.
 //
-// The other difference is that DeltaFIFO has two additional ways that
-// an object can be applied to an accumulator: Replaced and Sync.
-// If EmitDeltaTypeReplaced is not set to true, Sync will be used in
-// replace events for backwards compatibility.  Sync is used for periodic
-// resync events.
+// The other difference is that DeltaFIFO has an additional way that
+// an object can be applied to an accumulator, called Sync.
 //
 // DeltaFIFO is a producer-consumer queue, where a Reflector is
 // intended to be the producer, and the consumer is whatever calls
@@ -129,107 +193,6 @@ type DeltaFIFO struct {
 	emitDeltaTypeReplaced bool
 }
 
-// DeltaType is the type of a change (addition, deletion, etc)
-type DeltaType string
-
-// Change type definition
-const (
-	Added   DeltaType = "Added"
-	Updated DeltaType = "Updated"
-	Deleted DeltaType = "Deleted"
-	// Replaced is emitted when we encountered watch errors and had to do a
-	// relist. We don't know if the replaced object has changed.
-	//
-	// NOTE: Previous versions of DeltaFIFO would use Sync for Replace events
-	// as well. Hence, Replaced is only emitted when the option
-	// EmitDeltaTypeReplaced is true.
-	Replaced DeltaType = "Replaced"
-	// Sync is for synthetic events during a periodic resync.
-	Sync DeltaType = "Sync"
-)
-
-// Delta is a member of Deltas (a list of Delta objects) which
-// in its turn is the type stored by a DeltaFIFO. It tells you what
-// change happened, and the object's state after* that change.
-//
-// [*] Unless the change is a deletion, and then you'll get the final
-//     state of the object before it was deleted.
-type Delta struct {
-	Type   DeltaType
-	Object interface{}
-}
-
-// Deltas is a list of one or more 'Delta's to an individual object.
-// The oldest delta is at index 0, the newest delta is the last one.
-type Deltas []Delta
-
-// NewDeltaFIFO returns a Queue which can be used to process changes to items.
-//
-// keyFunc is used to figure out what key an object should have. (It is
-// exposed in the returned DeltaFIFO's KeyOf() method, with additional handling
-// around deleted objects and queue state).
-//
-// 'knownObjects' may be supplied to modify the behavior of Delete,
-// Replace, and Resync.  It may be nil if you do not need those
-// modifications.
-//
-// TODO: consider merging keyLister with this object, tracking a list of
-//       "known" keys when Pop() is called. Have to think about how that
-//       affects error retrying.
-// NOTE: It is possible to misuse this and cause a race when using an
-//       external known object source.
-//       Whether there is a potential race depends on how the consumer
-//       modifies knownObjects. In Pop(), process function is called under
-//       lock, so it is safe to update data structures in it that need to be
-//       in sync with the queue (e.g. knownObjects).
-//
-//       Example:
-//       In case of sharedIndexInformer being a consumer
-//       (https://github.com/kubernetes/kubernetes/blob/0cdd940f/staging/
-//       src/k8s.io/client-go/tools/cache/shared_informer.go#L192),
-//       there is no race as knownObjects (s.indexer) is modified safely
-//       under DeltaFIFO's lock. The only exceptions are GetStore() and
-//       GetIndexer() methods, which expose ways to modify the underlying
-//       storage. Currently these two methods are used for creating Lister
-//       and internal tests.
-//
-// Also see the comment on DeltaFIFO.
-//
-// Warning: This constructs a DeltaFIFO that does not differentiate between
-// events caused by a call to Replace (e.g., from a relist, which may
-// contain object updates), and synthetic events caused by a periodic resync
-// (which just emit the existing object). See https://issue.k8s.io/86015 for details.
-//
-// Use `NewDeltaFIFOWithOptions(DeltaFIFOOptions{..., EmitDeltaTypeReplaced: true})`
-// instead to receive a `Replaced` event depending on the type.
-//
-// Deprecated: Equivalent to NewDeltaFIFOWithOptions(DeltaFIFOOptions{KeyFunction: keyFunc, KnownObjects: knownObjects})
-func NewDeltaFIFO(keyFunc KeyFunc, knownObjects KeyListerGetter) *DeltaFIFO {
-	return NewDeltaFIFOWithOptions(DeltaFIFOOptions{
-		KeyFunction:  keyFunc,
-		KnownObjects: knownObjects,
-	})
-}
-
-// NewDeltaFIFOWithOptions returns a Queue which can be used to process changes to
-// items. See also the comment on DeltaFIFO.
-func NewDeltaFIFOWithOptions(opts DeltaFIFOOptions) *DeltaFIFO {
-	if opts.KeyFunction == nil {
-		opts.KeyFunction = MetaNamespaceKeyFunc
-	}
-
-	f := &DeltaFIFO{
-		items:        map[string]Deltas{},
-		queue:        []string{},
-		keyFunc:      opts.KeyFunction,
-		knownObjects: opts.KnownObjects,
-
-		emitDeltaTypeReplaced: opts.EmitDeltaTypeReplaced,
-	}
-	f.cond.L = &f.lock
-	return f
-}
-
 var (
 	_ = Queue(&DeltaFIFO{}) // DeltaFIFO is a Queue
 )
@@ -609,7 +572,7 @@ func (f *DeltaFIFO) Replace(list []interface{}, resourceVersion string) error {
 			f.populated = true
 			// While there shouldn't be any queued deletions in the initial
 			// population of the queue, it's better to be on the safe side.
-			f.initialPopulationCount = keys.Len() + queuedDeletions
+			f.initialPopulationCount = len(list) + queuedDeletions
 		}
 
 		return nil
@@ -639,7 +602,7 @@ func (f *DeltaFIFO) Replace(list []interface{}, resourceVersion string) error {
 
 	if !f.populated {
 		f.populated = true
-		f.initialPopulationCount = keys.Len() + queuedDeletions
+		f.initialPopulationCount = len(list) + queuedDeletions
 	}
 
 	return nil
@@ -710,6 +673,39 @@ type KeyGetter interface {
 	GetByKey(key string) (value interface{}, exists bool, err error)
 }
 
+// DeltaType is the type of a change (addition, deletion, etc)
+type DeltaType string
+
+// Change type definition
+const (
+	Added   DeltaType = "Added"
+	Updated DeltaType = "Updated"
+	Deleted DeltaType = "Deleted"
+	// Replaced is emitted when we encountered watch errors and had to do a
+	// relist. We don't know if the replaced object has changed.
+	//
+	// NOTE: Previous versions of DeltaFIFO would use Sync for Replace events
+	// as well. Hence, Replaced is only emitted when the option
+	// EmitDeltaTypeReplaced is true.
+	Replaced DeltaType = "Replaced"
+	// Sync is for synthetic events during a periodic resync.
+	Sync DeltaType = "Sync"
+)
+
+// Delta is the type stored by a DeltaFIFO. It tells you what change
+// happened, and the object's state after* that change.
+//
+// [*] Unless the change is a deletion, and then you'll get the final
+//     state of the object before it was deleted.
+type Delta struct {
+	Type   DeltaType
+	Object interface{}
+}
+
+// Deltas is a list of one or more 'Delta's to an individual object.
+// The oldest delta is at index 0, the newest delta is the last one.
+type Deltas []Delta
+
 // Oldest is a convenience function that returns the oldest delta, or
 // nil if there are no deltas.
 func (d Deltas) Oldest() *Delta {

tools/cache/delta_fifo_test.go

@@ -633,15 +633,6 @@ func TestDeltaFIFO_HasSynced(t *testing.T) {
 			},
 			expectedSynced: true,
 		},
-		{
-			// This test case won't happen in practice since a Reflector, the only producer for delta_fifo today, always passes a complete snapshot consistent in time;
-			// there cannot be duplicate keys in the list or apiserver is broken.
-			actions: []func(f *DeltaFIFO){
-				func(f *DeltaFIFO) { f.Replace([]interface{}{mkFifoObj("a", 1), mkFifoObj("a", 2)}, "0") },
-				func(f *DeltaFIFO) { Pop(f) },
-			},
-			expectedSynced: true,
-		},
 	}
 
 	for i, test := range tests {

tools/cache/index_test.go

@@ -76,7 +76,7 @@ func TestMultiIndexKeys(t *testing.T) {
 	expected["bert"] = sets.NewString("one", "two")
 	expected["elmo"] = sets.NewString("tre")
 	expected["oscar"] = sets.NewString("two")
-	expected["elmo1"] = sets.NewString()
+	expected["elmo"] = sets.NewString() // let's just make sure we don't get anything back in this case
 	{
 		for k, v := range expected {
 			found := sets.String{}
@@ -87,8 +87,9 @@ func TestMultiIndexKeys(t *testing.T) {
 			for _, item := range indexResults {
 				found.Insert(item.(*v1.Pod).Name)
 			}
-			if !found.Equal(v) {
-				t.Errorf("missing items, index %s, expected %v but found %v", k, v.List(), found.List())
+			items := v.List()
+			if !found.HasAll(items...) {
+				t.Errorf("missing items, index %s, expected %v but found %v", k, items, found.List())
 			}
 		}
 	}

tools/clientcmd/client_config.go

@@ -198,13 +198,13 @@ func (config *DirectClientConfig) ClientConfig() (*restclient.Config, error) {
 		if err != nil {
 			return nil, err
 		}
-		mergo.Merge(clientConfig, userAuthPartialConfig, mergo.WithOverride)
+		mergo.MergeWithOverwrite(clientConfig, userAuthPartialConfig)
 
 		serverAuthPartialConfig, err := getServerIdentificationPartialConfig(configAuthInfo, configClusterInfo)
 		if err != nil {
 			return nil, err
 		}
-		mergo.Merge(clientConfig, serverAuthPartialConfig, mergo.WithOverride)
+		mergo.MergeWithOverwrite(clientConfig, serverAuthPartialConfig)
 	}
 
 	return clientConfig, nil
@@ -225,7 +225,7 @@ func getServerIdentificationPartialConfig(configAuthInfo clientcmdapi.AuthInfo,
 	configClientConfig.CAData = configClusterInfo.CertificateAuthorityData
 	configClientConfig.Insecure = configClusterInfo.InsecureSkipTLSVerify
 	configClientConfig.ServerName = configClusterInfo.TLSServerName
-	mergo.Merge(mergedConfig, configClientConfig, mergo.WithOverride)
+	mergo.MergeWithOverwrite(mergedConfig, configClientConfig)
 
 	return mergedConfig, nil
 }
@@ -294,8 +294,8 @@ func (config *DirectClientConfig) getUserIdentificationPartialConfig(configAuthI
 		promptedConfig := makeUserIdentificationConfig(*promptedAuthInfo)
 		previouslyMergedConfig := mergedConfig
 		mergedConfig = &restclient.Config{}
-		mergo.Merge(mergedConfig, promptedConfig, mergo.WithOverride)
-		mergo.Merge(mergedConfig, previouslyMergedConfig, mergo.WithOverride)
+		mergo.MergeWithOverwrite(mergedConfig, promptedConfig)
+		mergo.MergeWithOverwrite(mergedConfig, previouslyMergedConfig)
 		config.promptedCredentials.username = mergedConfig.Username
 		config.promptedCredentials.password = mergedConfig.Password
 	}
@@ -463,12 +463,12 @@ func (config *DirectClientConfig) getContext() (clientcmdapi.Context, error) {
 
 	mergedContext := clientcmdapi.NewContext()
 	if configContext, exists := contexts[contextName]; exists {
-		mergo.Merge(mergedContext, configContext, mergo.WithOverride)
+		mergo.MergeWithOverwrite(mergedContext, configContext)
 	} else if required {
 		return clientcmdapi.Context{}, fmt.Errorf("context %q does not exist", contextName)
 	}
 	if config.overrides != nil {
-		mergo.Merge(mergedContext, config.overrides.Context, mergo.WithOverride)
+		mergo.MergeWithOverwrite(mergedContext, config.overrides.Context)
 	}
 
 	return *mergedContext, nil
@@ -481,12 +481,12 @@ func (config *DirectClientConfig) getAuthInfo() (clientcmdapi.AuthInfo, error) {
 
 	mergedAuthInfo := clientcmdapi.NewAuthInfo()
 	if configAuthInfo, exists := authInfos[authInfoName]; exists {
-		mergo.Merge(mergedAuthInfo, configAuthInfo, mergo.WithOverride)
+		mergo.MergeWithOverwrite(mergedAuthInfo, configAuthInfo)
 	} else if required {
 		return clientcmdapi.AuthInfo{}, fmt.Errorf("auth info %q does not exist", authInfoName)
 	}
 	if config.overrides != nil {
-		mergo.Merge(mergedAuthInfo, config.overrides.AuthInfo, mergo.WithOverride)
+		mergo.MergeWithOverwrite(mergedAuthInfo, config.overrides.AuthInfo)
 	}
 
 	return *mergedAuthInfo, nil
@@ -499,15 +499,15 @@ func (config *DirectClientConfig) getCluster() (clientcmdapi.Cluster, error) {
 
 	mergedClusterInfo := clientcmdapi.NewCluster()
 	if config.overrides != nil {
-		mergo.Merge(mergedClusterInfo, config.overrides.ClusterDefaults, mergo.WithOverride)
+		mergo.MergeWithOverwrite(mergedClusterInfo, config.overrides.ClusterDefaults)
 	}
 	if configClusterInfo, exists := clusterInfos[clusterInfoName]; exists {
-		mergo.Merge(mergedClusterInfo, configClusterInfo, mergo.WithOverride)
+		mergo.MergeWithOverwrite(mergedClusterInfo, configClusterInfo)
 	} else if required {
 		return clientcmdapi.Cluster{}, fmt.Errorf("cluster %q does not exist", clusterInfoName)
 	}
 	if config.overrides != nil {
-		mergo.Merge(mergedClusterInfo, config.overrides.ClusterInfo, mergo.WithOverride)
+		mergo.MergeWithOverwrite(mergedClusterInfo, config.overrides.ClusterInfo)
 	}
 
 	// * An override of --insecure-skip-tls-verify=true and no accompanying CA/CA data should clear already-set CA/CA data
@@ -548,12 +548,11 @@ func (config *inClusterClientConfig) RawConfig() (clientcmdapi.Config, error) {
 }
 
 func (config *inClusterClientConfig) ClientConfig() (*restclient.Config, error) {
-	inClusterConfigProvider := config.inClusterConfigProvider
-	if inClusterConfigProvider == nil {
-		inClusterConfigProvider = restclient.InClusterConfig
+	if config.inClusterConfigProvider == nil {
+		config.inClusterConfigProvider = restclient.InClusterConfig
 	}
 
-	icc, err := inClusterConfigProvider()
+	icc, err := config.inClusterConfigProvider()
 	if err != nil {
 		return nil, err
 	}
@@ -573,7 +572,7 @@ func (config *inClusterClientConfig) ClientConfig() (*restclient.Config, error)
 		}
 	}
 
-	return icc, nil
+	return icc, err
 }
 
 func (config *inClusterClientConfig) Namespace() (string, bool, error) {
@@ -612,7 +611,7 @@ func (config *inClusterClientConfig) Possible() bool {
 // to the default config.
 func BuildConfigFromFlags(masterUrl, kubeconfigPath string) (*restclient.Config, error) {
 	if kubeconfigPath == "" && masterUrl == "" {
-		klog.Warning("Neither --kubeconfig nor --master was specified.  Using the inClusterConfig.  This might not work.")
+		klog.Warningf("Neither --kubeconfig nor --master was specified.  Using the inClusterConfig.  This might not work.")
 		kubeconfig, err := restclient.InClusterConfig()
 		if err == nil {
 			return kubeconfig, nil

tools/clientcmd/client_config_test.go

@@ -63,7 +63,7 @@ func TestMergoSemantics(t *testing.T) {
 		},
 	}
 	for _, data := range testDataStruct {
-		err := mergo.Merge(&data.dst, &data.src, mergo.WithOverride)
+		err := mergo.MergeWithOverwrite(&data.dst, &data.src)
 		if err != nil {
 			t.Errorf("error while merging: %s", err)
 		}
@@ -92,7 +92,7 @@ func TestMergoSemantics(t *testing.T) {
 		},
 	}
 	for _, data := range testDataMap {
-		err := mergo.Merge(&data.dst, &data.src, mergo.WithOverride)
+		err := mergo.MergeWithOverwrite(&data.dst, &data.src)
 		if err != nil {
 			t.Errorf("error while merging: %s", err)
 		}

tools/clientcmd/loader.go

@@ -18,8 +18,10 @@ package clientcmd
 
 import (
 	"fmt"
+	"io"
 	"io/ioutil"
 	"os"
+	"path"
 	"path/filepath"
 	"reflect"
 	goruntime "runtime"
@@ -46,24 +48,24 @@ const (
 )
 
 var (
-	RecommendedConfigDir  = filepath.Join(homedir.HomeDir(), RecommendedHomeDir)
-	RecommendedHomeFile   = filepath.Join(RecommendedConfigDir, RecommendedFileName)
-	RecommendedSchemaFile = filepath.Join(RecommendedConfigDir, RecommendedSchemaName)
+	RecommendedConfigDir  = path.Join(homedir.HomeDir(), RecommendedHomeDir)
+	RecommendedHomeFile   = path.Join(RecommendedConfigDir, RecommendedFileName)
+	RecommendedSchemaFile = path.Join(RecommendedConfigDir, RecommendedSchemaName)
 )
 
 // currentMigrationRules returns a map that holds the history of recommended home directories used in previous versions.
 // Any future changes to RecommendedHomeFile and related are expected to add a migration rule here, in order to make
 // sure existing config files are migrated to their new locations properly.
 func currentMigrationRules() map[string]string {
-	var oldRecommendedHomeFileName string
+	oldRecommendedHomeFile := path.Join(os.Getenv("HOME"), "/.kube/.kubeconfig")
+	oldRecommendedWindowsHomeFile := path.Join(os.Getenv("HOME"), RecommendedHomeDir, RecommendedFileName)
+
+	migrationRules := map[string]string{}
+	migrationRules[RecommendedHomeFile] = oldRecommendedHomeFile
 	if goruntime.GOOS == "windows" {
-		oldRecommendedHomeFileName = RecommendedFileName
-	} else {
-		oldRecommendedHomeFileName = ".kubeconfig"
-	}
-	return map[string]string{
-		RecommendedHomeFile: filepath.Join(os.Getenv("HOME"), RecommendedHomeDir, oldRecommendedHomeFileName),
+		migrationRules[RecommendedHomeFile] = oldRecommendedWindowsHomeFile
 	}
+	return migrationRules
 }
 
 type ClientConfigLoader interface {
@@ -225,7 +227,7 @@ func (rules *ClientConfigLoadingRules) Load() (*clientcmdapi.Config, error) {
 	mapConfig := clientcmdapi.NewConfig()
 
 	for _, kubeconfig := range kubeconfigs {
-		mergo.Merge(mapConfig, kubeconfig, mergo.WithOverride)
+		mergo.MergeWithOverwrite(mapConfig, kubeconfig)
 	}
 
 	// merge all of the struct values in the reverse order so that priority is given correctly
@@ -233,14 +235,14 @@ func (rules *ClientConfigLoadingRules) Load() (*clientcmdapi.Config, error) {
 	nonMapConfig := clientcmdapi.NewConfig()
 	for i := len(kubeconfigs) - 1; i >= 0; i-- {
 		kubeconfig := kubeconfigs[i]
-		mergo.Merge(nonMapConfig, kubeconfig, mergo.WithOverride)
+		mergo.MergeWithOverwrite(nonMapConfig, kubeconfig)
 	}
 
 	// since values are overwritten, but maps values are not, we can merge the non-map config on top of the map config and
 	// get the values we expect.
 	config := clientcmdapi.NewConfig()
-	mergo.Merge(config, mapConfig, mergo.WithOverride)
-	mergo.Merge(config, nonMapConfig, mergo.WithOverride)
+	mergo.MergeWithOverwrite(config, mapConfig)
+	mergo.MergeWithOverwrite(config, nonMapConfig)
 
 	if rules.ResolvePaths() {
 		if err := ResolveLocalPaths(config); err != nil {
@@ -281,15 +283,20 @@ func (rules *ClientConfigLoadingRules) Migrate() error {
 			return fmt.Errorf("cannot migrate %v to %v because it is a directory", source, destination)
 		}
 
-		data, err := ioutil.ReadFile(source)
+		in, err := os.Open(source)
 		if err != nil {
 			return err
 		}
-		// destination is created with mode 0666 before umask
-		err = ioutil.WriteFile(destination, data, 0666)
+		defer in.Close()
+		out, err := os.Create(destination)
 		if err != nil {
 			return err
 		}
+		defer out.Close()
+
+		if _, err = io.Copy(out, in); err != nil {
+			return err
+		}
 	}
 
 	return nil

tools/leaderelection/resourcelock/interface.go

@@ -145,9 +145,6 @@ func New(lockType string, ns string, name string, coreClient corev1.CoreV1Interf
 }
 
 // NewFromKubeconfig will create a lock of a given type according to the input parameters.
-// Timeout set for a client used to contact to Kubernetes should be lower than
-// RenewDeadline to keep a single hung request from forcing a leader loss.
-// Setting it to max(time.Second, RenewDeadline/2) as a reasonable heuristic.
 func NewFromKubeconfig(lockType string, ns string, name string, rlc ResourceLockConfig, kubeconfig *restclient.Config, renewDeadline time.Duration) (Interface, error) {
 	// shallow copy, do not modify the kubeconfig
 	config := *kubeconfig

tools/record/event.go

@@ -155,21 +155,21 @@ func (a *EventRecorderAdapter) Eventf(regarding, _ runtime.Object, eventtype, re
 // Creates a new event broadcaster.
 func NewBroadcaster() EventBroadcaster {
 	return &eventBroadcasterImpl{
-		Broadcaster:   watch.NewLongQueueBroadcaster(maxQueuedEvents, watch.DropIfChannelFull),
+		Broadcaster:   watch.NewBroadcaster(maxQueuedEvents, watch.DropIfChannelFull),
 		sleepDuration: defaultSleepDuration,
 	}
 }
 
 func NewBroadcasterForTests(sleepDuration time.Duration) EventBroadcaster {
 	return &eventBroadcasterImpl{
-		Broadcaster:   watch.NewLongQueueBroadcaster(maxQueuedEvents, watch.DropIfChannelFull),
+		Broadcaster:   watch.NewBroadcaster(maxQueuedEvents, watch.DropIfChannelFull),
 		sleepDuration: sleepDuration,
 	}
 }
 
 func NewBroadcasterWithCorrelatorOptions(options CorrelatorOptions) EventBroadcaster {
 	return &eventBroadcasterImpl{
-		Broadcaster:   watch.NewLongQueueBroadcaster(maxQueuedEvents, watch.DropIfChannelFull),
+		Broadcaster:   watch.NewBroadcaster(maxQueuedEvents, watch.DropIfChannelFull),
 		sleepDuration: defaultSleepDuration,
 		options:       options,
 	}
@@ -323,7 +323,7 @@ type recorderImpl struct {
 	clock clock.Clock
 }
 
-func (recorder *recorderImpl) generateEvent(object runtime.Object, annotations map[string]string, eventtype, reason, message string) {
+func (recorder *recorderImpl) generateEvent(object runtime.Object, annotations map[string]string, timestamp metav1.Time, eventtype, reason, message string) {
 	ref, err := ref.GetReference(recorder.scheme, object)
 	if err != nil {
 		klog.Errorf("Could not construct reference to: '%#v' due to: '%v'. Will not report event: '%v' '%v' '%v'", object, err, eventtype, reason, message)
@@ -338,18 +338,15 @@ func (recorder *recorderImpl) generateEvent(object runtime.Object, annotations m
 	event := recorder.makeEvent(ref, annotations, eventtype, reason, message)
 	event.Source = recorder.source
 
-	// NOTE: events should be a non-blocking operation, but we also need to not
-	// put this in a goroutine, otherwise we'll race to write to a closed channel
-	// when we go to shut down this broadcaster.  Just drop events if we get overloaded,
-	// and log an error if that happens (we've configured the broadcaster to drop
-	// outgoing events anyway).
-	if sent := recorder.ActionOrDrop(watch.Added, event); !sent {
-		klog.Errorf("unable to record event: too many queued events, dropped event %#v", event)
-	}
+	go func() {
+		// NOTE: events should be a non-blocking operation
+		defer utilruntime.HandleCrash()
+		recorder.Action(watch.Added, event)
+	}()
 }
 
 func (recorder *recorderImpl) Event(object runtime.Object, eventtype, reason, message string) {
-	recorder.generateEvent(object, nil, eventtype, reason, message)
+	recorder.generateEvent(object, nil, metav1.Now(), eventtype, reason, message)
 }
 
 func (recorder *recorderImpl) Eventf(object runtime.Object, eventtype, reason, messageFmt string, args ...interface{}) {
@@ -357,7 +354,7 @@ func (recorder *recorderImpl) Eventf(object runtime.Object, eventtype, reason, m
 }
 
 func (recorder *recorderImpl) AnnotatedEventf(object runtime.Object, annotations map[string]string, eventtype, reason, messageFmt string, args ...interface{}) {
-	recorder.generateEvent(object, annotations, eventtype, reason, fmt.Sprintf(messageFmt, args...))
+	recorder.generateEvent(object, annotations, metav1.Now(), eventtype, reason, fmt.Sprintf(messageFmt, args...))
 }
 
 func (recorder *recorderImpl) makeEvent(ref *v1.ObjectReference, annotations map[string]string, eventtype, reason, message string) *v1.Event {

tools/record/event_test.go

@@ -21,7 +21,6 @@ import (
 	"fmt"
 	"net/http"
 	"strconv"
-	"sync"
 	"testing"
 	"time"
 
@@ -102,29 +101,6 @@ func OnPatchFactory(testCache map[string]*v1.Event, patchEvent chan<- *v1.Event)
 	}
 }
 
-func TestNonRacyShutdown(t *testing.T) {
-	// Attempt to simulate previously racy conditions, and ensure that no race
-	// occurs: Nominally, calling "Eventf" *followed by* shutdown from the same
-	// thread should be a safe operation, but it's not if we launch recorder.Action
-	// in a goroutine.
-
-	caster := NewBroadcasterForTests(0)
-	clock := clock.NewFakeClock(time.Now())
-	recorder := recorderWithFakeClock(v1.EventSource{Component: "eventTest"}, caster, clock)
-
-	var wg sync.WaitGroup
-	wg.Add(100)
-	for i := 0; i < 100; i++ {
-		go func() {
-			defer wg.Done()
-			recorder.Eventf(&v1.ObjectReference{}, v1.EventTypeNormal, "Started", "blah")
-		}()
-	}
-
-	wg.Wait()
-	caster.Shutdown()
-}
-
 func TestEventf(t *testing.T) {
 	testPod := &v1.Pod{
 		ObjectMeta: metav1.ObjectMeta{

tools/remotecommand/remotecommand_test.go

@@ -1,266 +0,0 @@
-/*
-Copyright 2020 The Kubernetes 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 remotecommand
-
-import (
-	"encoding/json"
-	"errors"
-	"io"
-	"io/ioutil"
-	v1 "k8s.io/api/core/v1"
-	apierrors "k8s.io/apimachinery/pkg/api/errors"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/util/httpstream"
-	"k8s.io/apimachinery/pkg/util/httpstream/spdy"
-	remotecommandconsts "k8s.io/apimachinery/pkg/util/remotecommand"
-	"k8s.io/apimachinery/pkg/util/wait"
-	"k8s.io/client-go/rest"
-	"net/http"
-	"net/http/httptest"
-	"net/url"
-	"strings"
-	"testing"
-	"time"
-)
-
-type AttachFunc func(in io.Reader, out, err io.WriteCloser, tty bool, resize <-chan TerminalSize) error
-type streamContext struct {
-	conn         io.Closer
-	stdinStream  io.ReadCloser
-	stdoutStream io.WriteCloser
-	stderrStream io.WriteCloser
-	writeStatus  func(status *apierrors.StatusError) error
-}
-
-type streamAndReply struct {
-	httpstream.Stream
-	replySent <-chan struct{}
-}
-
-type fakeMassiveDataPty struct{}
-
-func (s *fakeMassiveDataPty) Read(p []byte) (int, error) {
-	time.Sleep(time.Duration(1) * time.Second)
-	return copy(p, []byte{}), errors.New("client crashed after 1 second")
-}
-
-func (s *fakeMassiveDataPty) Write(p []byte) (int, error) {
-	time.Sleep(time.Duration(1) * time.Second)
-	return len(p), errors.New("return err")
-}
-
-func fakeMassiveDataAttacher(stdin io.Reader, stdout, stderr io.WriteCloser, tty bool, resize <-chan TerminalSize) error {
-
-	copyDone := make(chan struct{}, 3)
-
-	if stdin == nil {
-		return errors.New("stdin is requested") // we need stdin to notice the conn break
-	}
-
-	go func() {
-		io.Copy(ioutil.Discard, stdin)
-		copyDone <- struct{}{}
-	}()
-
-	go func() {
-		if stdout == nil {
-			return
-		}
-		copyDone <- writeMassiveData(stdout)
-	}()
-
-	go func() {
-		if stderr == nil {
-			return
-		}
-		copyDone <- writeMassiveData(stderr)
-	}()
-
-	select {
-	case <-copyDone:
-		return nil
-	}
-}
-
-func writeMassiveData(stdStream io.Writer) struct{} { // write to stdin or stdout
-	for {
-		_, err := io.Copy(stdStream, strings.NewReader("something"))
-		if err != nil && err.Error() != "EOF" {
-			break
-		}
-	}
-	return struct{}{}
-}
-
-func TestSPDYExecutorStream(t *testing.T) {
-	tests := []struct {
-		name        string
-		options     StreamOptions
-		expectError string
-		attacher    AttachFunc
-	}{
-		{
-			name: "stdoutBlockTest",
-			options: StreamOptions{
-				Stdin:  &fakeMassiveDataPty{},
-				Stdout: &fakeMassiveDataPty{},
-			},
-			expectError: "",
-			attacher:    fakeMassiveDataAttacher,
-		},
-		{
-			name: "stderrBlockTest",
-			options: StreamOptions{
-				Stdin:  &fakeMassiveDataPty{},
-				Stderr: &fakeMassiveDataPty{},
-			},
-			expectError: "",
-			attacher:    fakeMassiveDataAttacher,
-		},
-	}
-
-	for _, test := range tests {
-		server := newTestHTTPServer(test.attacher, &test.options)
-
-		err := attach2Server(server.URL, test.options)
-		gotError := ""
-		if err != nil {
-			gotError = err.Error()
-		}
-		if test.expectError != gotError {
-			t.Errorf("%s: expected [%v], got [%v]", test.name, test.expectError, gotError)
-		}
-
-		server.Close()
-	}
-
-}
-
-func newTestHTTPServer(f AttachFunc, options *StreamOptions) *httptest.Server {
-	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
-		ctx, err := createHTTPStreams(writer, request, options)
-		if err != nil {
-			return
-		}
-		defer ctx.conn.Close()
-
-		// handle input output
-		err = f(ctx.stdinStream, ctx.stdoutStream, ctx.stderrStream, false, nil)
-		if err != nil {
-			ctx.writeStatus(apierrors.NewInternalError(err))
-		} else {
-			ctx.writeStatus(&apierrors.StatusError{ErrStatus: metav1.Status{
-				Status: metav1.StatusSuccess,
-			}})
-		}
-	}))
-	return server
-}
-
-func attach2Server(rawURL string, options StreamOptions) error {
-	uri, _ := url.Parse(rawURL)
-	exec, err := NewSPDYExecutor(&rest.Config{Host: uri.Host}, "POST", uri)
-	if err != nil {
-		return err
-	}
-
-	e := make(chan error)
-	go func(e chan error) {
-		e <- exec.Stream(options)
-	}(e)
-	select {
-	case err := <-e:
-		return err
-	case <-time.After(wait.ForeverTestTimeout):
-		return errors.New("execute timeout")
-	}
-}
-
-// simplify createHttpStreams , only support StreamProtocolV4Name
-func createHTTPStreams(w http.ResponseWriter, req *http.Request, opts *StreamOptions) (*streamContext, error) {
-	_, err := httpstream.Handshake(req, w, []string{remotecommandconsts.StreamProtocolV4Name})
-	if err != nil {
-		return nil, err
-	}
-
-	upgrader := spdy.NewResponseUpgrader()
-	streamCh := make(chan streamAndReply)
-	conn := upgrader.UpgradeResponse(w, req, func(stream httpstream.Stream, replySent <-chan struct{}) error {
-		streamCh <- streamAndReply{Stream: stream, replySent: replySent}
-		return nil
-	})
-	ctx := &streamContext{
-		conn: conn,
-	}
-
-	// wait for stream
-	replyChan := make(chan struct{}, 4)
-	defer close(replyChan)
-	receivedStreams := 0
-	expectedStreams := 1
-	if opts.Stdout != nil {
-		expectedStreams++
-	}
-	if opts.Stdin != nil {
-		expectedStreams++
-	}
-	if opts.Stderr != nil {
-		expectedStreams++
-	}
-WaitForStreams:
-	for {
-		select {
-		case stream := <-streamCh:
-			streamType := stream.Headers().Get(v1.StreamType)
-			switch streamType {
-			case v1.StreamTypeError:
-				replyChan <- struct{}{}
-				ctx.writeStatus = v4WriteStatusFunc(stream)
-			case v1.StreamTypeStdout:
-				replyChan <- struct{}{}
-				ctx.stdoutStream = stream
-			case v1.StreamTypeStdin:
-				replyChan <- struct{}{}
-				ctx.stdinStream = stream
-			case v1.StreamTypeStderr:
-				replyChan <- struct{}{}
-				ctx.stderrStream = stream
-			default:
-				// add other stream ...
-				return nil, errors.New("unimplemented stream type")
-			}
-		case <-replyChan:
-			receivedStreams++
-			if receivedStreams == expectedStreams {
-				break WaitForStreams
-			}
-		}
-	}
-
-	return ctx, nil
-}
-
-func v4WriteStatusFunc(stream io.Writer) func(status *apierrors.StatusError) error {
-	return func(status *apierrors.StatusError) error {
-		bs, err := json.Marshal(status.Status())
-		if err != nil {
-			return err
-		}
-		_, err = stream.Write(bs)
-		return err
-	}
-}

tools/remotecommand/v2.go

@@ -142,10 +142,6 @@ func (p *streamProtocolV2) copyStdout(wg *sync.WaitGroup) {
 	go func() {
 		defer runtime.HandleCrash()
 		defer wg.Done()
-		// make sure, packet in queue can be consumed.
-		// block in queue may lead to deadlock in conn.server
-		// issue: https://github.com/kubernetes/kubernetes/issues/96339
-		defer io.Copy(ioutil.Discard, p.remoteStdout)
 
 		if _, err := io.Copy(p.Stdout, p.remoteStdout); err != nil {
 			runtime.HandleError(err)
@@ -162,7 +158,6 @@ func (p *streamProtocolV2) copyStderr(wg *sync.WaitGroup) {
 	go func() {
 		defer runtime.HandleCrash()
 		defer wg.Done()
-		defer io.Copy(ioutil.Discard, p.remoteStderr)
 
 		if _, err := io.Copy(p.Stderr, p.remoteStderr); err != nil {
 			runtime.HandleError(err)

tools/remotecommand/v4_test.go

@@ -18,7 +18,6 @@ package remotecommand
 
 import (
 	"fmt"
-	"strings"
 	"testing"
 )
 
@@ -37,7 +36,7 @@ func TestV4ErrorDecoder(t *testing.T) {
 		},
 		{
 			message: "{",
-			err:     "unexpected end of JSON input in \"{\"",
+			err:     "error stream protocol error: unexpected end of JSON input in \"{\"",
 		},
 		{
 			message: `{"status": "Success" }`,
@@ -65,7 +64,7 @@ func TestV4ErrorDecoder(t *testing.T) {
 		if want == "" {
 			want = "<nil>"
 		}
-		if got := fmt.Sprintf("%v", err); !strings.Contains(got, want) {
+		if got := fmt.Sprintf("%v", err); got != want {
 			t.Errorf("wrong error for message %q: want=%q, got=%q", test.message, want, got)
 		}
 	}

tools/watch/informerwatcher.go

@@ -127,7 +127,7 @@ func NewIndexerInformerWatcher(lw cache.ListerWatcher, objType runtime.Object) (
 				// We have no means of passing the additional information down using
 				// watch API based on watch.Event but the caller can filter such
 				// objects by checking if metadata.deletionTimestamp is set
-				obj = staleObj.Obj
+				obj = staleObj
 			}
 
 			e.push(watch.Event{

tools/watch/informerwatcher_test.go

@@ -27,7 +27,6 @@ import (
 	"github.com/davecgh/go-spew/spew"
 
 	corev1 "k8s.io/api/core/v1"
-	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -277,88 +276,3 @@ func TestNewInformerWatcher(t *testing.T) {
 	}
 
 }
-
-// TestInformerWatcherDeletedFinalStateUnknown tests the code path when `DeleteFunc`
-// in `NewIndexerInformerWatcher` receives a `cache.DeletedFinalStateUnknown`
-// object from the underlying `DeltaFIFO`. The triggering condition is described
-// at https://github.com/kubernetes/kubernetes/blob/dc39ab2417bfddcec37be4011131c59921fdbe98/staging/src/k8s.io/client-go/tools/cache/delta_fifo.go#L736-L739.
-//
-// Code from @liggitt
-func TestInformerWatcherDeletedFinalStateUnknown(t *testing.T) {
-	listCalls := 0
-	watchCalls := 0
-	lw := &cache.ListWatch{
-		ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
-			retval := &corev1.SecretList{}
-			if listCalls == 0 {
-				// Return a list with items in it
-				retval.ResourceVersion = "1"
-				retval.Items = []corev1.Secret{{ObjectMeta: metav1.ObjectMeta{Name: "secret1", Namespace: "ns1", ResourceVersion: "123"}}}
-			} else {
-				// Return empty lists after the first call
-				retval.ResourceVersion = "2"
-			}
-			listCalls++
-			return retval, nil
-		},
-		WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
-			w := watch.NewFake()
-			if options.ResourceVersion == "1" {
-				go func() {
-					// Close with a "Gone" error when trying to start a watch from the first list
-					w.Error(&apierrors.NewGone("gone").ErrStatus)
-					w.Stop()
-				}()
-			}
-			watchCalls++
-			return w, nil
-		},
-	}
-	_, _, w, done := NewIndexerInformerWatcher(lw, &corev1.Secret{})
-
-	// Expect secret add
-	select {
-	case event, ok := <-w.ResultChan():
-		if !ok {
-			t.Fatal("unexpected close")
-		}
-		if event.Type != watch.Added {
-			t.Fatalf("expected Added event, got %#v", event)
-		}
-		if event.Object.(*corev1.Secret).ResourceVersion != "123" {
-			t.Fatalf("expected added Secret with rv=123, got %#v", event.Object)
-		}
-	case <-time.After(time.Second * 10):
-		t.Fatal("timeout")
-	}
-
-	// Expect secret delete because the relist was missing the secret
-	select {
-	case event, ok := <-w.ResultChan():
-		if !ok {
-			t.Fatal("unexpected close")
-		}
-		if event.Type != watch.Deleted {
-			t.Fatalf("expected Deleted event, got %#v", event)
-		}
-		if event.Object.(*corev1.Secret).ResourceVersion != "123" {
-			t.Fatalf("expected deleted Secret with rv=123, got %#v", event.Object)
-		}
-	case <-time.After(time.Second * 10):
-		t.Fatal("timeout")
-	}
-
-	w.Stop()
-	select {
-	case <-done:
-	case <-time.After(time.Second * 10):
-		t.Fatal("timeout")
-	}
-
-	if listCalls < 2 {
-		t.Fatalf("expected at least 2 list calls, got %d", listCalls)
-	}
-	if watchCalls < 1 {
-		t.Fatalf("expected at least 1 watch call, got %d", watchCalls)
-	}
-}

transport/round_trippers.go

@@ -426,7 +426,7 @@ func (rt *debuggingRoundTripper) RoundTrip(req *http.Request) (*http.Response, e
 
 	}
 	if rt.levels[debugRequestHeaders] {
-		klog.Info("Request Headers:")
+		klog.Infof("Request Headers:")
 		for key, values := range reqInfo.RequestHeaders {
 			for _, value := range values {
 				value = maskValue(key, value)
@@ -448,7 +448,7 @@ func (rt *debuggingRoundTripper) RoundTrip(req *http.Request) (*http.Response, e
 		klog.Infof("Response Status: %s in %d milliseconds", reqInfo.ResponseStatus, reqInfo.Duration.Nanoseconds()/int64(time.Millisecond))
 	}
 	if rt.levels[debugResponseHeaders] {
-		klog.Info("Response Headers:")
+		klog.Infof("Response Headers:")
 		for key, values := range reqInfo.ResponseHeaders {
 			for _, value := range values {
 				klog.Infof("    %s: %s", key, value)

transport/spdy/spdy.go

@@ -20,7 +20,6 @@ import (
 	"fmt"
 	"net/http"
 	"net/url"
-	"time"
 
 	"k8s.io/apimachinery/pkg/util/httpstream"
 	"k8s.io/apimachinery/pkg/util/httpstream/spdy"
@@ -43,13 +42,7 @@ func RoundTripperFor(config *restclient.Config) (http.RoundTripper, Upgrader, er
 	if config.Proxy != nil {
 		proxy = config.Proxy
 	}
-	upgradeRoundTripper := spdy.NewRoundTripperWithConfig(spdy.RoundTripperConfig{
-		TLS:                      tlsConfig,
-		FollowRedirects:          true,
-		RequireSameHostRedirects: false,
-		Proxier:                  proxy,
-		PingPeriod:               time.Second * 5,
-	})
+	upgradeRoundTripper := spdy.NewRoundTripperWithProxy(tlsConfig, true, false, proxy)
 	wrapper, err := restclient.HTTPWrappersForConfig(config, upgradeRoundTripper)
 	if err != nil {
 		return nil, nil, err

util/connrotation/connrotation.go

@@ -33,40 +33,18 @@ type DialFunc func(ctx context.Context, network, address string) (net.Conn, erro
 // Dialer opens connections through Dial and tracks them.
 type Dialer struct {
 	dial DialFunc
-	*ConnectionTracker
-}
 
-// NewDialer creates a new Dialer instance.
-// Equivalent to NewDialerWithTracker(dial, nil).
-func NewDialer(dial DialFunc) *Dialer {
-	return NewDialerWithTracker(dial, nil)
-}
-
-// NewDialerWithTracker creates a new Dialer instance.
-//
-// If dial is not nil, it will be used to create new underlying connections.
-// Otherwise net.DialContext is used.
-// If tracker is not nil, it will be used to track new underlying connections.
-// Otherwise NewConnectionTracker() is used.
-func NewDialerWithTracker(dial DialFunc, tracker *ConnectionTracker) *Dialer {
-	if tracker == nil {
-		tracker = NewConnectionTracker()
-	}
-	return &Dialer{
-		dial:              dial,
-		ConnectionTracker: tracker,
-	}
-}
-
-// ConnectionTracker keeps track of opened connections
-type ConnectionTracker struct {
 	mu    sync.Mutex
 	conns map[*closableConn]struct{}
 }
 
-// NewConnectionTracker returns a connection tracker for use with NewDialerWithTracker
-func NewConnectionTracker() *ConnectionTracker {
-	return &ConnectionTracker{
+// NewDialer creates a new Dialer instance.
+//
+// If dial is not nil, it will be used to create new underlying connections.
+// Otherwise net.DialContext is used.
+func NewDialer(dial DialFunc) *Dialer {
+	return &Dialer{
+		dial:  dial,
 		conns: make(map[*closableConn]struct{}),
 	}
 }
@@ -74,40 +52,17 @@ func NewConnectionTracker() *ConnectionTracker {
 // CloseAll forcibly closes all tracked connections.
 //
 // Note: new connections may get created before CloseAll returns.
-func (c *ConnectionTracker) CloseAll() {
-	c.mu.Lock()
-	conns := c.conns
-	c.conns = make(map[*closableConn]struct{})
-	c.mu.Unlock()
+func (d *Dialer) CloseAll() {
+	d.mu.Lock()
+	conns := d.conns
+	d.conns = make(map[*closableConn]struct{})
+	d.mu.Unlock()
 
 	for conn := range conns {
 		conn.Close()
 	}
 }
 
-// Track adds the connection to the list of tracked connections,
-// and returns a wrapped copy of the connection that stops tracking the connection
-// when it is closed.
-func (c *ConnectionTracker) Track(conn net.Conn) net.Conn {
-	closable := &closableConn{Conn: conn}
-
-	// When the connection is closed, remove it from the map. This will
-	// be no-op if the connection isn't in the map, e.g. if CloseAll()
-	// is called.
-	closable.onClose = func() {
-		c.mu.Lock()
-		delete(c.conns, closable)
-		c.mu.Unlock()
-	}
-
-	// Start tracking the connection
-	c.mu.Lock()
-	c.conns[closable] = struct{}{}
-	c.mu.Unlock()
-
-	return closable
-}
-
 // Dial creates a new tracked connection.
 func (d *Dialer) Dial(network, address string) (net.Conn, error) {
 	return d.DialContext(context.Background(), network, address)
@@ -119,7 +74,24 @@ func (d *Dialer) DialContext(ctx context.Context, network, address string) (net.
 	if err != nil {
 		return nil, err
 	}
-	return d.ConnectionTracker.Track(conn), nil
+
+	closable := &closableConn{Conn: conn}
+
+	// When the connection is closed, remove it from the map. This will
+	// be no-op if the connection isn't in the map, e.g. if CloseAll()
+	// is called.
+	closable.onClose = func() {
+		d.mu.Lock()
+		delete(d.conns, closable)
+		d.mu.Unlock()
+	}
+
+	// Start tracking the connection
+	d.mu.Lock()
+	d.conns[closable] = struct{}{}
+	d.mu.Unlock()
+
+	return closable, nil
 }
 
 type closableConn struct {

util/workqueue/queue.go

@@ -55,13 +55,7 @@ func newQueue(c clock.Clock, metrics queueMetrics, updatePeriod time.Duration) *
 		metrics:                    metrics,
 		unfinishedWorkUpdatePeriod: updatePeriod,
 	}
-
-	// Don't start the goroutine for a type of noMetrics so we don't consume
-	// resources unnecessarily
-	if _, ok := metrics.(noMetrics); !ok {
-		go t.updateUnfinishedWorkLoop()
-	}
-
+	go t.updateUnfinishedWorkLoop()
 	return t
 }