Update dependencies to v0.21.0-alpha.2 tag

Don't spawn a goroutine for every event recording

Kubernetes-commit: 9d99dbc357ed85bcc963fc4bab8a4a3089c910b2

Godeps/Godeps.json

@@ -36,7 +36,7 @@
 		},
 		{
 			"ImportPath": "github.com/Azure/go-autorest/autorest",
-			"Rev": "v0.11.1"
+			"Rev": "v0.11.12"
 		},
 		{
 			"ImportPath": "github.com/Azure/go-autorest/autorest/adal",
@@ -106,10 +106,6 @@
 			"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"
@@ -464,11 +460,11 @@
 		},
 		{
 			"ImportPath": "k8s.io/api",
-			"Rev": "v0.20.4"
+			"Rev": "v0.21.0-alpha.2"
 		},
 		{
 			"ImportPath": "k8s.io/apimachinery",
-			"Rev": "v0.20.4"
+			"Rev": "v0.21.0-alpha.2"
 		},
 		{
 			"ImportPath": "k8s.io/gengo",

README.md

@@ -82,14 +82,15 @@ 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.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`                        | +-              | +-              | +-              | +-              | +-              |
+|                               | 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`                        | +-              | +-              | +-              | +-              | +-              | +-              |
 
 Key:
 
@@ -123,10 +124,11 @@ 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:
@@ -176,7 +178,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.19.0
+go get k8s.io/client-go@v0.20.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.1
+	github.com/Azure/go-autorest/autorest v0.11.12
 	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.20.4
-	k8s.io/apimachinery v0.20.4
+	k8s.io/api v0.21.0-alpha.2
+	k8s.io/apimachinery v0.21.0-alpha.2
 	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.20.4
-	k8s.io/apimachinery => k8s.io/apimachinery v0.20.4
+	k8s.io/api => k8s.io/api v0.21.0-alpha.2
+	k8s.io/apimachinery => k8s.io/apimachinery v0.21.0-alpha.2
 )

go.sum

@@ -24,16 +24,12 @@ 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.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 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/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=
@@ -54,8 +50,6 @@ 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=
@@ -433,8 +427,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.20.4/go.mod h1:++lNL1AJMkDymriNniQsWRkMDzRaX2Y/POTUi8yvqYQ=
-k8s.io/apimachinery v0.20.4/go.mod h1:WlLqWAHZGg07AeltaI0MV5uk1Omp8xaN0JGLY6gkRpU=
+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/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,7 +18,6 @@ package exec
 
 import (
 	"bytes"
-	"context"
 	"crypto/tls"
 	"crypto/x509"
 	"errors"
@@ -52,7 +51,6 @@ 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.
@@ -177,6 +175,12 @@ 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,
@@ -196,6 +200,9 @@ 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 {
@@ -229,6 +236,11 @@ 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
@@ -236,8 +248,6 @@ type Authenticator struct {
 	mu          sync.Mutex
 	cachedCreds *credentials
 	exp         time.Time
-
-	onRotateList []func()
 }
 
 type credentials struct {
@@ -266,20 +276,12 @@ func (a *Authenticator) UpdateTransportConfig(c *transport.Config) error {
 	}
 	c.TLS.GetCert = a.cert
 
-	var dial func(ctx context.Context, network, addr string) (net.Conn, error)
+	var d *connrotation.Dialer
 	if c.Dial != nil {
-		dial = c.Dial
+		// if c has a custom dialer, we have to wrap it
+		d = connrotation.NewDialerWithTracker(c.Dial, a.connTracker)
 	} else {
-		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)
+		d = a.defaultDialer
 	}
 
 	c.Dial = d.DialContext
@@ -458,9 +460,7 @@ 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))
 		}
-		for _, onRotate := range a.onRotateList {
-			onRotate()
-		}
+		a.connTracker.CloseAll()
 	}
 
 	expiry := time.Time{}

rest/plugin.go

@@ -47,6 +47,13 @@ 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)
@@ -69,5 +76,8 @@ 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,6 +171,28 @@ 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,54 +26,6 @@ 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 {
@@ -99,25 +51,6 @@ 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
@@ -128,8 +61,11 @@ func NewDeltaFIFOWithOptions(opts DeltaFIFOOptions) *DeltaFIFO {
 // Deleted if the older Deleted's object is a
 // DeletedFinalStateUnknown.
 //
-// The other difference is that DeltaFIFO has an additional way that
-// an object can be applied to an accumulator, called Sync.
+// 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.
 //
 // DeltaFIFO is a producer-consumer queue, where a Reflector is
 // intended to be the producer, and the consumer is whatever calls
@@ -193,6 +129,107 @@ 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
 )
@@ -572,7 +609,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 = len(list) + queuedDeletions
+			f.initialPopulationCount = keys.Len() + queuedDeletions
 		}
 
 		return nil
@@ -602,7 +639,7 @@ func (f *DeltaFIFO) Replace(list []interface{}, resourceVersion string) error {
 
 	if !f.populated {
 		f.populated = true
-		f.initialPopulationCount = len(list) + queuedDeletions
+		f.initialPopulationCount = keys.Len() + queuedDeletions
 	}
 
 	return nil
@@ -673,39 +710,6 @@ 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,6 +633,15 @@ 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["elmo"] = sets.NewString() // let's just make sure we don't get anything back in this case
+	expected["elmo1"] = sets.NewString()
 	{
 		for k, v := range expected {
 			found := sets.String{}
@@ -87,9 +87,8 @@ func TestMultiIndexKeys(t *testing.T) {
 			for _, item := range indexResults {
 				found.Insert(item.(*v1.Pod).Name)
 			}
-			items := v.List()
-			if !found.HasAll(items...) {
-				t.Errorf("missing items, index %s, expected %v but found %v", k, items, found.List())
+			if !found.Equal(v) {
+				t.Errorf("missing items, index %s, expected %v but found %v", k, v.List(), 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.MergeWithOverwrite(clientConfig, userAuthPartialConfig)
+		mergo.Merge(clientConfig, userAuthPartialConfig, mergo.WithOverride)
 
 		serverAuthPartialConfig, err := getServerIdentificationPartialConfig(configAuthInfo, configClusterInfo)
 		if err != nil {
 			return nil, err
 		}
-		mergo.MergeWithOverwrite(clientConfig, serverAuthPartialConfig)
+		mergo.Merge(clientConfig, serverAuthPartialConfig, mergo.WithOverride)
 	}
 
 	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.MergeWithOverwrite(mergedConfig, configClientConfig)
+	mergo.Merge(mergedConfig, configClientConfig, mergo.WithOverride)
 
 	return mergedConfig, nil
 }
@@ -294,8 +294,8 @@ func (config *DirectClientConfig) getUserIdentificationPartialConfig(configAuthI
 		promptedConfig := makeUserIdentificationConfig(*promptedAuthInfo)
 		previouslyMergedConfig := mergedConfig
 		mergedConfig = &restclient.Config{}
-		mergo.MergeWithOverwrite(mergedConfig, promptedConfig)
-		mergo.MergeWithOverwrite(mergedConfig, previouslyMergedConfig)
+		mergo.Merge(mergedConfig, promptedConfig, mergo.WithOverride)
+		mergo.Merge(mergedConfig, previouslyMergedConfig, mergo.WithOverride)
 		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.MergeWithOverwrite(mergedContext, configContext)
+		mergo.Merge(mergedContext, configContext, mergo.WithOverride)
 	} else if required {
 		return clientcmdapi.Context{}, fmt.Errorf("context %q does not exist", contextName)
 	}
 	if config.overrides != nil {
-		mergo.MergeWithOverwrite(mergedContext, config.overrides.Context)
+		mergo.Merge(mergedContext, config.overrides.Context, mergo.WithOverride)
 	}
 
 	return *mergedContext, nil
@@ -481,12 +481,12 @@ func (config *DirectClientConfig) getAuthInfo() (clientcmdapi.AuthInfo, error) {
 
 	mergedAuthInfo := clientcmdapi.NewAuthInfo()
 	if configAuthInfo, exists := authInfos[authInfoName]; exists {
-		mergo.MergeWithOverwrite(mergedAuthInfo, configAuthInfo)
+		mergo.Merge(mergedAuthInfo, configAuthInfo, mergo.WithOverride)
 	} else if required {
 		return clientcmdapi.AuthInfo{}, fmt.Errorf("auth info %q does not exist", authInfoName)
 	}
 	if config.overrides != nil {
-		mergo.MergeWithOverwrite(mergedAuthInfo, config.overrides.AuthInfo)
+		mergo.Merge(mergedAuthInfo, config.overrides.AuthInfo, mergo.WithOverride)
 	}
 
 	return *mergedAuthInfo, nil
@@ -499,15 +499,15 @@ func (config *DirectClientConfig) getCluster() (clientcmdapi.Cluster, error) {
 
 	mergedClusterInfo := clientcmdapi.NewCluster()
 	if config.overrides != nil {
-		mergo.MergeWithOverwrite(mergedClusterInfo, config.overrides.ClusterDefaults)
+		mergo.Merge(mergedClusterInfo, config.overrides.ClusterDefaults, mergo.WithOverride)
 	}
 	if configClusterInfo, exists := clusterInfos[clusterInfoName]; exists {
-		mergo.MergeWithOverwrite(mergedClusterInfo, configClusterInfo)
+		mergo.Merge(mergedClusterInfo, configClusterInfo, mergo.WithOverride)
 	} else if required {
 		return clientcmdapi.Cluster{}, fmt.Errorf("cluster %q does not exist", clusterInfoName)
 	}
 	if config.overrides != nil {
-		mergo.MergeWithOverwrite(mergedClusterInfo, config.overrides.ClusterInfo)
+		mergo.Merge(mergedClusterInfo, config.overrides.ClusterInfo, mergo.WithOverride)
 	}
 
 	// * An override of --insecure-skip-tls-verify=true and no accompanying CA/CA data should clear already-set CA/CA data
@@ -548,11 +548,12 @@ func (config *inClusterClientConfig) RawConfig() (clientcmdapi.Config, error) {
 }
 
 func (config *inClusterClientConfig) ClientConfig() (*restclient.Config, error) {
-	if config.inClusterConfigProvider == nil {
-		config.inClusterConfigProvider = restclient.InClusterConfig
+	inClusterConfigProvider := config.inClusterConfigProvider
+	if inClusterConfigProvider == nil {
+		inClusterConfigProvider = restclient.InClusterConfig
 	}
 
-	icc, err := config.inClusterConfigProvider()
+	icc, err := inClusterConfigProvider()
 	if err != nil {
 		return nil, err
 	}
@@ -572,7 +573,7 @@ func (config *inClusterClientConfig) ClientConfig() (*restclient.Config, error)
 		}
 	}
 
-	return icc, err
+	return icc, nil
 }
 
 func (config *inClusterClientConfig) Namespace() (string, bool, error) {
@@ -611,7 +612,7 @@ func (config *inClusterClientConfig) Possible() bool {
 // to the default config.
 func BuildConfigFromFlags(masterUrl, kubeconfigPath string) (*restclient.Config, error) {
 	if kubeconfigPath == "" && masterUrl == "" {
-		klog.Warningf("Neither --kubeconfig nor --master was specified.  Using the inClusterConfig.  This might not work.")
+		klog.Warning("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.MergeWithOverwrite(&data.dst, &data.src)
+		err := mergo.Merge(&data.dst, &data.src, mergo.WithOverride)
 		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.MergeWithOverwrite(&data.dst, &data.src)
+		err := mergo.Merge(&data.dst, &data.src, mergo.WithOverride)
 		if err != nil {
 			t.Errorf("error while merging: %s", err)
 		}

tools/clientcmd/loader.go

@@ -18,10 +18,8 @@ package clientcmd
 
 import (
 	"fmt"
-	"io"
 	"io/ioutil"
 	"os"
-	"path"
 	"path/filepath"
 	"reflect"
 	goruntime "runtime"
@@ -48,24 +46,24 @@ const (
 )
 
 var (
-	RecommendedConfigDir  = path.Join(homedir.HomeDir(), RecommendedHomeDir)
-	RecommendedHomeFile   = path.Join(RecommendedConfigDir, RecommendedFileName)
-	RecommendedSchemaFile = path.Join(RecommendedConfigDir, RecommendedSchemaName)
+	RecommendedConfigDir  = filepath.Join(homedir.HomeDir(), RecommendedHomeDir)
+	RecommendedHomeFile   = filepath.Join(RecommendedConfigDir, RecommendedFileName)
+	RecommendedSchemaFile = filepath.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 {
-	oldRecommendedHomeFile := path.Join(os.Getenv("HOME"), "/.kube/.kubeconfig")
-	oldRecommendedWindowsHomeFile := path.Join(os.Getenv("HOME"), RecommendedHomeDir, RecommendedFileName)
-
-	migrationRules := map[string]string{}
-	migrationRules[RecommendedHomeFile] = oldRecommendedHomeFile
+	var oldRecommendedHomeFileName string
 	if goruntime.GOOS == "windows" {
-		migrationRules[RecommendedHomeFile] = oldRecommendedWindowsHomeFile
+		oldRecommendedHomeFileName = RecommendedFileName
+	} else {
+		oldRecommendedHomeFileName = ".kubeconfig"
+	}
+	return map[string]string{
+		RecommendedHomeFile: filepath.Join(os.Getenv("HOME"), RecommendedHomeDir, oldRecommendedHomeFileName),
 	}
-	return migrationRules
 }
 
 type ClientConfigLoader interface {
@@ -227,7 +225,7 @@ func (rules *ClientConfigLoadingRules) Load() (*clientcmdapi.Config, error) {
 	mapConfig := clientcmdapi.NewConfig()
 
 	for _, kubeconfig := range kubeconfigs {
-		mergo.MergeWithOverwrite(mapConfig, kubeconfig)
+		mergo.Merge(mapConfig, kubeconfig, mergo.WithOverride)
 	}
 
 	// merge all of the struct values in the reverse order so that priority is given correctly
@@ -235,14 +233,14 @@ func (rules *ClientConfigLoadingRules) Load() (*clientcmdapi.Config, error) {
 	nonMapConfig := clientcmdapi.NewConfig()
 	for i := len(kubeconfigs) - 1; i >= 0; i-- {
 		kubeconfig := kubeconfigs[i]
-		mergo.MergeWithOverwrite(nonMapConfig, kubeconfig)
+		mergo.Merge(nonMapConfig, kubeconfig, mergo.WithOverride)
 	}
 
 	// 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.MergeWithOverwrite(config, mapConfig)
-	mergo.MergeWithOverwrite(config, nonMapConfig)
+	mergo.Merge(config, mapConfig, mergo.WithOverride)
+	mergo.Merge(config, nonMapConfig, mergo.WithOverride)
 
 	if rules.ResolvePaths() {
 		if err := ResolveLocalPaths(config); err != nil {
@@ -283,20 +281,15 @@ func (rules *ClientConfigLoadingRules) Migrate() error {
 			return fmt.Errorf("cannot migrate %v to %v because it is a directory", source, destination)
 		}
 
-		in, err := os.Open(source)
+		data, err := ioutil.ReadFile(source)
 		if err != nil {
 			return err
 		}
-		defer in.Close()
-		out, err := os.Create(destination)
+		// destination is created with mode 0666 before umask
+		err = ioutil.WriteFile(destination, data, 0666)
 		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,6 +145,9 @@ 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.NewBroadcaster(maxQueuedEvents, watch.DropIfChannelFull),
+		Broadcaster:   watch.NewLongQueueBroadcaster(maxQueuedEvents, watch.DropIfChannelFull),
 		sleepDuration: defaultSleepDuration,
 	}
 }
 
 func NewBroadcasterForTests(sleepDuration time.Duration) EventBroadcaster {
 	return &eventBroadcasterImpl{
-		Broadcaster:   watch.NewBroadcaster(maxQueuedEvents, watch.DropIfChannelFull),
+		Broadcaster:   watch.NewLongQueueBroadcaster(maxQueuedEvents, watch.DropIfChannelFull),
 		sleepDuration: sleepDuration,
 	}
 }
 
 func NewBroadcasterWithCorrelatorOptions(options CorrelatorOptions) EventBroadcaster {
 	return &eventBroadcasterImpl{
-		Broadcaster:   watch.NewBroadcaster(maxQueuedEvents, watch.DropIfChannelFull),
+		Broadcaster:   watch.NewLongQueueBroadcaster(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, timestamp metav1.Time, eventtype, reason, message string) {
+func (recorder *recorderImpl) generateEvent(object runtime.Object, annotations map[string]string, 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,15 +338,18 @@ func (recorder *recorderImpl) generateEvent(object runtime.Object, annotations m
 	event := recorder.makeEvent(ref, annotations, eventtype, reason, message)
 	event.Source = recorder.source
 
-	go func() {
-		// NOTE: events should be a non-blocking operation
-		defer utilruntime.HandleCrash()
-		recorder.Action(watch.Added, event)
-	}()
+	// 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)
+	}
 }
 
 func (recorder *recorderImpl) Event(object runtime.Object, eventtype, reason, message string) {
-	recorder.generateEvent(object, nil, metav1.Now(), eventtype, reason, message)
+	recorder.generateEvent(object, nil, eventtype, reason, message)
 }
 
 func (recorder *recorderImpl) Eventf(object runtime.Object, eventtype, reason, messageFmt string, args ...interface{}) {
@@ -354,7 +357,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, metav1.Now(), eventtype, reason, fmt.Sprintf(messageFmt, args...))
+	recorder.generateEvent(object, annotations, 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,6 +21,7 @@ import (
 	"fmt"
 	"net/http"
 	"strconv"
+	"sync"
 	"testing"
 	"time"
 
@@ -101,6 +102,29 @@ 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

@@ -0,0 +1,266 @@
+/*
+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,6 +142,10 @@ 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)
@@ -158,6 +162,7 @@ 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,6 +18,7 @@ package remotecommand
 
 import (
 	"fmt"
+	"strings"
 	"testing"
 )
 
@@ -36,7 +37,7 @@ func TestV4ErrorDecoder(t *testing.T) {
 		},
 		{
 			message: "{",
-			err:     "error stream protocol error: unexpected end of JSON input in \"{\"",
+			err:     "unexpected end of JSON input in \"{\"",
 		},
 		{
 			message: `{"status": "Success" }`,
@@ -64,7 +65,7 @@ func TestV4ErrorDecoder(t *testing.T) {
 		if want == "" {
 			want = "<nil>"
 		}
-		if got := fmt.Sprintf("%v", err); got != want {
+		if got := fmt.Sprintf("%v", err); !strings.Contains(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 = staleObj.Obj
 			}
 
 			e.push(watch.Event{

tools/watch/informerwatcher_test.go

@@ -27,6 +27,7 @@ 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"
@@ -276,3 +277,88 @@ 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.Infof("Request Headers:")
+		klog.Info("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.Infof("Response Headers:")
+		klog.Info("Response Headers:")
 		for key, values := range reqInfo.ResponseHeaders {
 			for _, value := range values {
 				klog.Infof("    %s: %s", key, value)

transport/spdy/spdy.go

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

util/connrotation/connrotation.go

@@ -33,18 +33,40 @@ 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{}
 }
 
-// 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,
+// NewConnectionTracker returns a connection tracker for use with NewDialerWithTracker
+func NewConnectionTracker() *ConnectionTracker {
+	return &ConnectionTracker{
 		conns: make(map[*closableConn]struct{}),
 	}
 }
@@ -52,17 +74,40 @@ func NewDialer(dial DialFunc) *Dialer {
 // CloseAll forcibly closes all tracked connections.
 //
 // Note: new connections may get created before CloseAll returns.
-func (d *Dialer) CloseAll() {
-	d.mu.Lock()
-	conns := d.conns
-	d.conns = make(map[*closableConn]struct{})
-	d.mu.Unlock()
+func (c *ConnectionTracker) CloseAll() {
+	c.mu.Lock()
+	conns := c.conns
+	c.conns = make(map[*closableConn]struct{})
+	c.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)
@@ -74,24 +119,7 @@ func (d *Dialer) DialContext(ctx context.Context, network, address string) (net.
 	if err != nil {
 		return nil, err
 	}
-
-	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
+	return d.ConnectionTracker.Track(conn), nil
 }
 
 type closableConn struct {

util/workqueue/queue.go

@@ -55,7 +55,13 @@ func newQueue(c clock.Clock, metrics queueMetrics, updatePeriod time.Duration) *
 		metrics:                    metrics,
 		unfinishedWorkUpdatePeriod: updatePeriod,
 	}
-	go t.updateUnfinishedWorkLoop()
+
+	// 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()
+	}
+
 	return t
 }