Update Kubernetes to v1.21.4-k3s1

go.mod

@@ -36,34 +36,34 @@ replace (
 	google.golang.org/genproto => google.golang.org/genproto v0.0.0-20200513103714-09dca8ec2884
 	google.golang.org/grpc => google.golang.org/grpc v1.27.1
 	gopkg.in/square/go-jose.v2 => gopkg.in/square/go-jose.v2 v2.2.2
-	k8s.io/api => github.com/k3s-io/kubernetes/staging/src/k8s.io/api v1.21.3-k3s1
-	k8s.io/apiextensions-apiserver => github.com/k3s-io/kubernetes/staging/src/k8s.io/apiextensions-apiserver v1.21.3-k3s1
-	k8s.io/apimachinery => github.com/k3s-io/kubernetes/staging/src/k8s.io/apimachinery v1.21.3-k3s1
-	k8s.io/apiserver => github.com/k3s-io/kubernetes/staging/src/k8s.io/apiserver v1.21.3-k3s1
-	k8s.io/cli-runtime => github.com/k3s-io/kubernetes/staging/src/k8s.io/cli-runtime v1.21.3-k3s1
-	k8s.io/client-go => github.com/k3s-io/kubernetes/staging/src/k8s.io/client-go v1.21.3-k3s1
-	k8s.io/cloud-provider => github.com/k3s-io/kubernetes/staging/src/k8s.io/cloud-provider v1.21.3-k3s1
-	k8s.io/cluster-bootstrap => github.com/k3s-io/kubernetes/staging/src/k8s.io/cluster-bootstrap v1.21.3-k3s1
-	k8s.io/code-generator => github.com/k3s-io/kubernetes/staging/src/k8s.io/code-generator v1.21.3-k3s1
-	k8s.io/component-base => github.com/k3s-io/kubernetes/staging/src/k8s.io/component-base v1.21.3-k3s1
-	k8s.io/component-helpers => github.com/k3s-io/kubernetes/staging/src/k8s.io/component-helpers v1.21.3-k3s1
-	k8s.io/controller-manager => github.com/k3s-io/kubernetes/staging/src/k8s.io/controller-manager v1.21.3-k3s1
-	k8s.io/cri-api => github.com/k3s-io/kubernetes/staging/src/k8s.io/cri-api v1.21.3-k3s1
-	k8s.io/csi-translation-lib => github.com/k3s-io/kubernetes/staging/src/k8s.io/csi-translation-lib v1.21.3-k3s1
-	k8s.io/kube-aggregator => github.com/k3s-io/kubernetes/staging/src/k8s.io/kube-aggregator v1.21.3-k3s1
-	k8s.io/kube-controller-manager => github.com/k3s-io/kubernetes/staging/src/k8s.io/kube-controller-manager v1.21.3-k3s1
-	k8s.io/kube-proxy => github.com/k3s-io/kubernetes/staging/src/k8s.io/kube-proxy v1.21.3-k3s1
-	k8s.io/kube-scheduler => github.com/k3s-io/kubernetes/staging/src/k8s.io/kube-scheduler v1.21.3-k3s1
-	k8s.io/kubectl => github.com/k3s-io/kubernetes/staging/src/k8s.io/kubectl v1.21.3-k3s1
-	k8s.io/kubelet => github.com/k3s-io/kubernetes/staging/src/k8s.io/kubelet v1.21.3-k3s1
-	k8s.io/kubernetes => github.com/k3s-io/kubernetes v1.21.3-k3s1
-	k8s.io/legacy-cloud-providers => github.com/k3s-io/kubernetes/staging/src/k8s.io/legacy-cloud-providers v1.21.3-k3s1
-	k8s.io/metrics => github.com/k3s-io/kubernetes/staging/src/k8s.io/metrics v1.21.3-k3s1
-	k8s.io/mount-utils => github.com/k3s-io/kubernetes/staging/src/k8s.io/mount-utils v1.21.3-k3s1
-	k8s.io/node-api => github.com/k3s-io/kubernetes/staging/src/k8s.io/node-api v1.21.3-k3s1
-	k8s.io/sample-apiserver => github.com/k3s-io/kubernetes/staging/src/k8s.io/sample-apiserver v1.21.3-k3s1
-	k8s.io/sample-cli-plugin => github.com/k3s-io/kubernetes/staging/src/k8s.io/sample-cli-plugin v1.21.3-k3s1
-	k8s.io/sample-controller => github.com/k3s-io/kubernetes/staging/src/k8s.io/sample-controller v1.21.3-k3s1
+	k8s.io/api => github.com/k3s-io/kubernetes/staging/src/k8s.io/api v1.21.4-k3s1
+	k8s.io/apiextensions-apiserver => github.com/k3s-io/kubernetes/staging/src/k8s.io/apiextensions-apiserver v1.21.4-k3s1
+	k8s.io/apimachinery => github.com/k3s-io/kubernetes/staging/src/k8s.io/apimachinery v1.21.4-k3s1
+	k8s.io/apiserver => github.com/k3s-io/kubernetes/staging/src/k8s.io/apiserver v1.21.4-k3s1
+	k8s.io/cli-runtime => github.com/k3s-io/kubernetes/staging/src/k8s.io/cli-runtime v1.21.4-k3s1
+	k8s.io/client-go => github.com/k3s-io/kubernetes/staging/src/k8s.io/client-go v1.21.4-k3s1
+	k8s.io/cloud-provider => github.com/k3s-io/kubernetes/staging/src/k8s.io/cloud-provider v1.21.4-k3s1
+	k8s.io/cluster-bootstrap => github.com/k3s-io/kubernetes/staging/src/k8s.io/cluster-bootstrap v1.21.4-k3s1
+	k8s.io/code-generator => github.com/k3s-io/kubernetes/staging/src/k8s.io/code-generator v1.21.4-k3s1
+	k8s.io/component-base => github.com/k3s-io/kubernetes/staging/src/k8s.io/component-base v1.21.4-k3s1
+	k8s.io/component-helpers => github.com/k3s-io/kubernetes/staging/src/k8s.io/component-helpers v1.21.4-k3s1
+	k8s.io/controller-manager => github.com/k3s-io/kubernetes/staging/src/k8s.io/controller-manager v1.21.4-k3s1
+	k8s.io/cri-api => github.com/k3s-io/kubernetes/staging/src/k8s.io/cri-api v1.21.4-k3s1
+	k8s.io/csi-translation-lib => github.com/k3s-io/kubernetes/staging/src/k8s.io/csi-translation-lib v1.21.4-k3s1
+	k8s.io/kube-aggregator => github.com/k3s-io/kubernetes/staging/src/k8s.io/kube-aggregator v1.21.4-k3s1
+	k8s.io/kube-controller-manager => github.com/k3s-io/kubernetes/staging/src/k8s.io/kube-controller-manager v1.21.4-k3s1
+	k8s.io/kube-proxy => github.com/k3s-io/kubernetes/staging/src/k8s.io/kube-proxy v1.21.4-k3s1
+	k8s.io/kube-scheduler => github.com/k3s-io/kubernetes/staging/src/k8s.io/kube-scheduler v1.21.4-k3s1
+	k8s.io/kubectl => github.com/k3s-io/kubernetes/staging/src/k8s.io/kubectl v1.21.4-k3s1
+	k8s.io/kubelet => github.com/k3s-io/kubernetes/staging/src/k8s.io/kubelet v1.21.4-k3s1
+	k8s.io/kubernetes => github.com/k3s-io/kubernetes v1.21.4-k3s1
+	k8s.io/legacy-cloud-providers => github.com/k3s-io/kubernetes/staging/src/k8s.io/legacy-cloud-providers v1.21.4-k3s1
+	k8s.io/metrics => github.com/k3s-io/kubernetes/staging/src/k8s.io/metrics v1.21.4-k3s1
+	k8s.io/mount-utils => github.com/k3s-io/kubernetes/staging/src/k8s.io/mount-utils v1.21.4-k3s1
+	k8s.io/node-api => github.com/k3s-io/kubernetes/staging/src/k8s.io/node-api v1.21.4-k3s1
+	k8s.io/sample-apiserver => github.com/k3s-io/kubernetes/staging/src/k8s.io/sample-apiserver v1.21.4-k3s1
+	k8s.io/sample-cli-plugin => github.com/k3s-io/kubernetes/staging/src/k8s.io/sample-cli-plugin v1.21.4-k3s1
+	k8s.io/sample-controller => github.com/k3s-io/kubernetes/staging/src/k8s.io/sample-controller v1.21.4-k3s1
 	mvdan.cc/unparam => mvdan.cc/unparam v0.0.0-20210104141923-aac4ce9116a7
 )
 
@@ -123,18 +123,18 @@ require (
 	google.golang.org/grpc v1.37.0
 	gopkg.in/yaml.v2 v2.4.0
 	inet.af/tcpproxy v0.0.0-20200125044825-b6bb9b5b8252
-	k8s.io/api v0.21.2
-	k8s.io/apimachinery v0.21.2
-	k8s.io/apiserver v0.21.2
+	k8s.io/api v0.21.4
+	k8s.io/apimachinery v0.21.4
+	k8s.io/apiserver v0.21.4
 	k8s.io/client-go v11.0.1-0.20190409021438-1a26190bd76a+incompatible
-	k8s.io/cloud-provider v0.21.2
-	k8s.io/component-base v0.21.2
-	k8s.io/controller-manager v0.21.2
-	k8s.io/cri-api v0.21.2
+	k8s.io/cloud-provider v0.21.4
+	k8s.io/component-base v0.21.4
+	k8s.io/controller-manager v0.21.4
+	k8s.io/cri-api v0.21.4
 	k8s.io/klog v1.0.0
 	k8s.io/klog/v2 v2.8.0
-	k8s.io/kubectl v0.21.2
-	k8s.io/kubernetes v1.21.3
+	k8s.io/kubectl v0.21.4
+	k8s.io/kubernetes v1.21.4
 	k8s.io/utils v0.0.0-20201110183641-67b214c5f920
 	sigs.k8s.io/yaml v1.2.0
 )

vendor/k8s.io/apiserver/pkg/endpoints/handlers/fieldmanager/admission.go

@@ -60,7 +60,10 @@ func (admit *managedFieldsValidatingAdmissionController) Admit(ctx context.Conte
 	}
 	objectMeta, err := meta.Accessor(a.GetObject())
 	if err != nil {
-		return err
+		// the object we are dealing with doesn't have object metadata defined
+		// in that case we don't have to keep track of the managedField
+		// just call the wrapped admission
+		return mutationInterface.Admit(ctx, a, o)
 	}
 	managedFieldsBeforeAdmission := objectMeta.GetManagedFields()
 	if err := mutationInterface.Admit(ctx, a, o); err != nil {

vendor/k8s.io/apiserver/pkg/endpoints/metrics/metrics.go

@@ -315,7 +315,7 @@ func RecordRequestAbort(req *http.Request, requestInfo *request.RequestInfo) {
 	}
 
 	scope := CleanScope(requestInfo)
-	reportedVerb := cleanVerb(canonicalVerb(strings.ToUpper(req.Method), scope), req)
+	reportedVerb := cleanVerb(canonicalVerb(strings.ToUpper(req.Method), scope), "", req)
 	resource := requestInfo.Resource
 	subresource := requestInfo.Subresource
 	group := requestInfo.APIGroup
@@ -338,7 +338,7 @@ func RecordRequestTermination(req *http.Request, requestInfo *request.RequestInf
 	// InstrumentRouteFunc which is registered in installer.go with predefined
 	// list of verbs (different than those translated to RequestInfo).
 	// However, we need to tweak it e.g. to differentiate GET from LIST.
-	reportedVerb := cleanVerb(canonicalVerb(strings.ToUpper(req.Method), scope), req)
+	reportedVerb := cleanVerb(canonicalVerb(strings.ToUpper(req.Method), scope), "", req)
 
 	if requestInfo.IsResourceRequest {
 		requestTerminationsTotal.WithContext(req.Context()).WithLabelValues(reportedVerb, requestInfo.APIGroup, requestInfo.APIVersion, requestInfo.Resource, requestInfo.Subresource, scope, component, codeToString(code)).Inc()
@@ -360,7 +360,7 @@ func RecordLongRunning(req *http.Request, requestInfo *request.RequestInfo, comp
 	// InstrumentRouteFunc which is registered in installer.go with predefined
 	// list of verbs (different than those translated to RequestInfo).
 	// However, we need to tweak it e.g. to differentiate GET from LIST.
-	reportedVerb := cleanVerb(canonicalVerb(strings.ToUpper(req.Method), scope), req)
+	reportedVerb := cleanVerb(canonicalVerb(strings.ToUpper(req.Method), scope), "", req)
 
 	if requestInfo.IsResourceRequest {
 		g = longRunningRequestGauge.WithContext(req.Context()).WithLabelValues(reportedVerb, requestInfo.APIGroup, requestInfo.APIVersion, requestInfo.Resource, requestInfo.Subresource, scope, component)
@@ -379,7 +379,7 @@ func MonitorRequest(req *http.Request, verb, group, version, resource, subresour
 	// InstrumentRouteFunc which is registered in installer.go with predefined
 	// list of verbs (different than those translated to RequestInfo).
 	// However, we need to tweak it e.g. to differentiate GET from LIST.
-	reportedVerb := cleanVerb(canonicalVerb(strings.ToUpper(req.Method), scope), req)
+	reportedVerb := cleanVerb(canonicalVerb(strings.ToUpper(req.Method), scope), verb, req)
 
 	dryRun := cleanDryRun(req.URL)
 	elapsedSeconds := elapsed.Seconds()
@@ -483,8 +483,15 @@ func canonicalVerb(verb string, scope string) string {
 	}
 }
 
-func cleanVerb(verb string, request *http.Request) string {
+func cleanVerb(verb, suggestedVerb string, request *http.Request) string {
 	reportedVerb := verb
+	// CanonicalVerb (being an input for this function) doesn't handle correctly the
+	// deprecated path pattern for watch of:
+	//   GET /api/{version}/watch/{resource}
+	// We correct it manually based on the pass verb from the installer.
+	if suggestedVerb == "WATCH" || suggestedVerb == "WATCHLIST" {
+		reportedVerb = "WATCH"
+	}
 	if verb == "LIST" {
 		// see apimachinery/pkg/runtime/conversion.go Convert_Slice_string_To_bool
 		if values := request.URL.Query()["watch"]; len(values) > 0 {

vendor/k8s.io/apiserver/pkg/registry/generic/registry/decorated_watcher.go

@@ -29,8 +29,8 @@ type decoratedWatcher struct {
 	resultCh  chan watch.Event
 }
 
-func newDecoratedWatcher(w watch.Interface, decorator func(runtime.Object)) *decoratedWatcher {
-	ctx, cancel := context.WithCancel(context.Background())
+func newDecoratedWatcher(ctx context.Context, w watch.Interface, decorator func(runtime.Object)) *decoratedWatcher {
+	ctx, cancel := context.WithCancel(ctx)
 	d := &decoratedWatcher{
 		w:         w,
 		decorator: decorator,
@@ -41,14 +41,18 @@ func newDecoratedWatcher(w watch.Interface, decorator func(runtime.Object)) *dec
 	return d
 }
 
+// run decorates watch events from the underlying watcher until its result channel
+// is closed or the passed in context is done.
+// When run() returns, decoratedWatcher#resultCh is closed.
 func (d *decoratedWatcher) run(ctx context.Context) {
 	var recv, send watch.Event
 	var ok bool
+	defer close(d.resultCh)
 	for {
 		select {
 		case recv, ok = <-d.w.ResultChan():
-			// The underlying channel may be closed after timeout.
 			if !ok {
+				// The underlying channel was closed, cancel our context
 				d.cancel()
 				return
 			}
@@ -61,20 +65,24 @@ func (d *decoratedWatcher) run(ctx context.Context) {
 			}
 			select {
 			case d.resultCh <- send:
-				if send.Type == watch.Error {
-					d.cancel()
-				}
+				// propagated event successfully
 			case <-ctx.Done():
+				// context timed out or was cancelled, stop the underlying watcher
+				d.w.Stop()
+				return
 			}
 		case <-ctx.Done():
+			// context timed out or was cancelled, stop the underlying watcher
 			d.w.Stop()
-			close(d.resultCh)
 			return
 		}
 	}
 }
 
 func (d *decoratedWatcher) Stop() {
+	// stop the underlying watcher
+	d.w.Stop()
+	// cancel our context
 	d.cancel()
 }
 

vendor/k8s.io/apiserver/pkg/registry/generic/registry/store.go

@@ -1226,7 +1226,7 @@ func (e *Store) WatchPredicate(ctx context.Context, p storage.SelectionPredicate
 				return nil, err
 			}
 			if e.Decorator != nil {
-				return newDecoratedWatcher(w, e.Decorator), nil
+				return newDecoratedWatcher(ctx, w, e.Decorator), nil
 			}
 			return w, nil
 		}
@@ -1239,7 +1239,7 @@ func (e *Store) WatchPredicate(ctx context.Context, p storage.SelectionPredicate
 		return nil, err
 	}
 	if e.Decorator != nil {
-		return newDecoratedWatcher(w, e.Decorator), nil
+		return newDecoratedWatcher(ctx, w, e.Decorator), nil
 	}
 	return w, nil
 }

vendor/k8s.io/apiserver/pkg/server/egressselector/egress_selector.go

@@ -134,7 +134,7 @@ func tunnelHTTPConnect(proxyConn net.Conn, proxyAddress, addr string) (net.Conn,
 
 type proxier interface {
 	// proxy returns a connection to addr.
-	proxy(addr string) (net.Conn, error)
+	proxy(ctx context.Context, addr string) (net.Conn, error)
 }
 
 var _ proxier = &httpConnectProxier{}
@@ -144,7 +144,7 @@ type httpConnectProxier struct {
 	proxyAddress string
 }
 
-func (t *httpConnectProxier) proxy(addr string) (net.Conn, error) {
+func (t *httpConnectProxier) proxy(ctx context.Context, addr string) (net.Conn, error) {
 	return tunnelHTTPConnect(t.conn, t.proxyAddress, addr)
 }
 
@@ -154,8 +154,8 @@ type grpcProxier struct {
 	tunnel client.Tunnel
 }
 
-func (g *grpcProxier) proxy(addr string) (net.Conn, error) {
-	return g.tunnel.Dial("tcp", addr)
+func (g *grpcProxier) proxy(ctx context.Context, addr string) (net.Conn, error) {
+	return g.tunnel.DialContext(ctx, "tcp", addr)
 }
 
 type proxyServerConnector interface {
@@ -203,7 +203,8 @@ func (u *udsGRPCConnector) connect() (proxier, error) {
 		return c, err
 	})
 
-	tunnel, err := client.CreateSingleUseGrpcTunnel(udsName, dialOption, grpc.WithInsecure())
+	ctx := context.TODO()
+	tunnel, err := client.CreateSingleUseGrpcTunnel(ctx, udsName, dialOption, grpc.WithInsecure())
 	if err != nil {
 		return nil, err
 	}
@@ -234,7 +235,7 @@ func (d *dialerCreator) createDialer() utilnet.DialFunc {
 			egressmetrics.Metrics.ObserveDialFailure(d.options.protocol, d.options.transport, egressmetrics.StageConnect)
 			return nil, err
 		}
-		conn, err := proxier.proxy(addr)
+		conn, err := proxier.proxy(ctx, addr)
 		if err != nil {
 			egressmetrics.Metrics.ObserveDialFailure(d.options.protocol, d.options.transport, egressmetrics.StageProxy)
 			return nil, err

vendor/k8s.io/apiserver/pkg/storage/etcd3/store.go

@@ -230,12 +230,22 @@ func (s *store) conditionalDelete(
 				}
 
 				// It's possible we're working with stale data.
+				// Remember the revision of the potentially stale data and the resulting update error
+				cachedRev := origState.rev
+				cachedUpdateErr := err
+
 				// Actually fetch
 				origState, err = getCurrentState()
 				if err != nil {
 					return err
 				}
 				origStateIsCurrent = true
+
+				// it turns out our cached data was not stale, return the error
+				if cachedRev == origState.rev {
+					return cachedUpdateErr
+				}
+
 				// Retry
 				continue
 			}
@@ -246,12 +256,22 @@ func (s *store) conditionalDelete(
 			}
 
 			// It's possible we're working with stale data.
+			// Remember the revision of the potentially stale data and the resulting update error
+			cachedRev := origState.rev
+			cachedUpdateErr := err
+
 			// Actually fetch
 			origState, err = getCurrentState()
 			if err != nil {
 				return err
 			}
 			origStateIsCurrent = true
+
+			// it turns out our cached data was not stale, return the error
+			if cachedRev == origState.rev {
+				return cachedUpdateErr
+			}
+
 			// Retry
 			continue
 		}
@@ -345,12 +365,22 @@ func (s *store) GuaranteedUpdate(
 			}
 
 			// It's possible we were working with stale data
+			// Remember the revision of the potentially stale data and the resulting update error
+			cachedRev := origState.rev
+			cachedUpdateErr := err
+
 			// Actually fetch
 			origState, err = getCurrentState()
 			if err != nil {
 				return err
 			}
 			origStateIsCurrent = true
+
+			// it turns out our cached data was not stale, return the error
+			if cachedRev == origState.rev {
+				return cachedUpdateErr
+			}
+
 			// Retry
 			continue
 		}

vendor/k8s.io/client-go/pkg/version/base.go

@@ -3,8 +3,8 @@ package version
 var (
 	gitMajor = "1"
 	gitMinor = "21"
-	gitVersion   = "v1.21.3-k3s1"
-	gitCommit    = "cd1e339ca1143f54627539d1abc7345ffd0ce4c4"
+	gitVersion   = "v1.21.4-k3s1"
+	gitCommit    = "24d3954c6cd8b757150dd3b0ebf02ad005f07523"
 	gitTreeState = "clean"
-	buildDate = "2021-07-15T23:18:03Z"
+	buildDate = "2021-08-12T22:35:54Z"
 )

vendor/k8s.io/cloud-provider/go.sum

@@ -679,8 +679,8 @@ k8s.io/utils v0.0.0-20201110183641-67b214c5f920/go.mod h1:jPW/WVKK9YHAvNhRxK0md/
 rsc.io/binaryregexp v0.2.0/go.mod h1:qTv7/COck+e2FymRvadv62gMdZztPaShugOCi3I+8D8=
 rsc.io/quote/v3 v3.1.0/go.mod h1:yEA65RcK8LyAZtP9Kv3t0HmxON59tX3rD+tICJqUlj0=
 rsc.io/sampler v1.3.0/go.mod h1:T1hPZKmBbMNahiBKFy5HrXp6adAjACjK9JXDnKaTXpA=
-sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.0.19 h1:0jaDAAxtqIrrqas4vtTqxct4xS5kHfRNycTRLTyJmVM=
-sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.0.19/go.mod h1:LEScyzhFmoF5pso/YSeBstl57mOzx9xlU9n85RGrDQg=
+sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.0.22 h1:fmRfl9WJ4ApJn7LxNuED4m0t18qivVQOxP6aAYG9J6c=
+sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.0.22/go.mod h1:LEScyzhFmoF5pso/YSeBstl57mOzx9xlU9n85RGrDQg=
 sigs.k8s.io/structured-merge-diff/v4 v4.0.2/go.mod h1:bJZC9H9iH24zzfZ/41RGcq60oK1F7G282QMXDPYydCw=
 sigs.k8s.io/structured-merge-diff/v4 v4.1.2 h1:Hr/htKFmJEbtMgS/UD0N+gtgctAqz81t3nu+sPzynno=
 sigs.k8s.io/structured-merge-diff/v4 v4.1.2/go.mod h1:j/nl6xW8vLS49O8YvXW1ocPhZawJtm+Yrr7PPRQ0Vg4=

vendor/k8s.io/component-base/version/base.go

@@ -3,8 +3,8 @@ package version
 var (
 	gitMajor = "1"
 	gitMinor = "21"
-	gitVersion   = "v1.21.3-k3s1"
-	gitCommit    = "cd1e339ca1143f54627539d1abc7345ffd0ce4c4"
+	gitVersion   = "v1.21.4-k3s1"
+	gitCommit    = "24d3954c6cd8b757150dd3b0ebf02ad005f07523"
 	gitTreeState = "clean"
-	buildDate = "2021-07-15T23:18:03Z"
+	buildDate = "2021-08-12T22:35:54Z"
 )

vendor/k8s.io/kubectl/pkg/polymorphichelpers/logsforobject.go

@@ -76,31 +76,34 @@ func logsForObjectWithClient(clientset corev1client.CoreV1Interface, object, opt
 	case *corev1.Pod:
 		// if allContainers is true, then we're going to locate all containers and then iterate through them. At that point, "allContainers" is false
 		if !allContainers {
+			currOpts := new(corev1.PodLogOptions)
+			if opts != nil {
+				opts.DeepCopyInto(currOpts)
+			}
 			// in case the "kubectl.kubernetes.io/default-container" annotation is present, we preset the opts.Containers to default to selected
 			// container. This gives users ability to preselect the most interesting container in pod.
-			if annotations := t.GetAnnotations(); annotations != nil && len(opts.Container) == 0 {
-				var containerName string
+			if annotations := t.GetAnnotations(); annotations != nil && currOpts.Container == "" {
+				var defaultContainer string
 				if len(annotations[podcmd.DefaultContainerAnnotationName]) > 0 {
-					containerName = annotations[podcmd.DefaultContainerAnnotationName]
+					defaultContainer = annotations[podcmd.DefaultContainerAnnotationName]
 				} else if len(annotations[defaultLogsContainerAnnotationName]) > 0 {
 					// Only log deprecation if we have only the old annotation. This allows users to
 					// set both to support multiple versions of kubectl; if they are setting both
 					// they must already know it is deprecated, so we don't need to add noisy
 					// warnings.
-					containerName = annotations[defaultLogsContainerAnnotationName]
+					defaultContainer = annotations[defaultLogsContainerAnnotationName]
 					fmt.Fprintf(os.Stderr, "Using deprecated annotation `kubectl.kubernetes.io/default-logs-container` in pod/%v. Please use `kubectl.kubernetes.io/default-container` instead\n", t.Name)
 				}
-				if len(containerName) > 0 {
-					if exists, _ := podcmd.FindContainerByName(t, containerName); exists != nil {
-						opts.Container = containerName
+				if len(defaultContainer) > 0 {
+					if exists, _ := podcmd.FindContainerByName(t, defaultContainer); exists == nil {
+						fmt.Fprintf(os.Stderr, "Default container name %q not found in pod %s\n", defaultContainer, t.Name)
 					} else {
-						fmt.Fprintf(os.Stderr, "Default container name %q not found in a pod\n", containerName)
+						currOpts.Container = defaultContainer
 					}
 				}
 			}
 
-			var containerName string
-			if opts == nil || len(opts.Container) == 0 {
+			if currOpts.Container == "" {
 				// We don't know container name. In this case we expect only one container to be present in the pod (ignoring InitContainers).
 				// If there is more than one container, we should return an error showing all container names.
 				if len(t.Spec.Containers) != 1 {
@@ -117,14 +120,12 @@ func logsForObjectWithClient(clientset corev1client.CoreV1Interface, object, opt
 
 					return nil, errors.New(err)
 				}
-				containerName = t.Spec.Containers[0].Name
-			} else {
-				containerName = opts.Container
+				currOpts.Container = t.Spec.Containers[0].Name
 			}
 
-			container, fieldPath := podcmd.FindContainerByName(t, containerName)
+			container, fieldPath := podcmd.FindContainerByName(t, currOpts.Container)
 			if container == nil {
-				return nil, fmt.Errorf("container %s is not valid for pod %s", opts.Container, t.Name)
+				return nil, fmt.Errorf("container %s is not valid for pod %s", currOpts.Container, t.Name)
 			}
 			ref, err := reference.GetPartialReference(scheme.Scheme, t, fieldPath)
 			if err != nil {
@@ -132,7 +133,7 @@ func logsForObjectWithClient(clientset corev1client.CoreV1Interface, object, opt
 			}
 
 			ret := make(map[corev1.ObjectReference]rest.ResponseWrapper, 1)
-			ret[*ref] = clientset.Pods(t.Namespace).GetLogs(t.Name, opts)
+			ret[*ref] = clientset.Pods(t.Namespace).GetLogs(t.Name, currOpts)
 			return ret, nil
 		}
 

vendor/k8s.io/kubernetes/pkg/controller/certificates/cleaner/cleaner.go

@@ -104,11 +104,7 @@ func (ccc *CSRCleanerController) worker() {
 }
 
 func (ccc *CSRCleanerController) handle(csr *capi.CertificateSigningRequest) error {
-	isIssuedExpired, err := isIssuedExpired(csr)
-	if err != nil {
-		return err
-	}
-	if isIssuedPastDeadline(csr) || isDeniedPastDeadline(csr) || isFailedPastDeadline(csr) || isPendingPastDeadline(csr) || isIssuedExpired {
+	if isIssuedPastDeadline(csr) || isDeniedPastDeadline(csr) || isFailedPastDeadline(csr) || isPendingPastDeadline(csr) || isIssuedExpired(csr) {
 		if err := ccc.csrClient.Delete(context.TODO(), csr.Name, metav1.DeleteOptions{}); err != nil {
 			return fmt.Errorf("unable to delete CSR %q: %v", csr.Name, err)
 		}
@@ -118,18 +114,14 @@ func (ccc *CSRCleanerController) handle(csr *capi.CertificateSigningRequest) err
 
 // isIssuedExpired checks if the CSR has been issued a certificate and if the
 // expiration of the certificate (the NotAfter value) has passed.
-func isIssuedExpired(csr *capi.CertificateSigningRequest) (bool, error) {
-	isExpired, err := isExpired(csr)
-	if err != nil {
-		return false, err
-	}
+func isIssuedExpired(csr *capi.CertificateSigningRequest) bool {
 	for _, c := range csr.Status.Conditions {
-		if c.Type == capi.CertificateApproved && isIssued(csr) && isExpired {
+		if c.Type == capi.CertificateApproved && isIssued(csr) && isExpired(csr) {
 			klog.Infof("Cleaning CSR %q as the associated certificate is expired.", csr.Name)
-			return true, nil
+			return true
 		}
 	}
-	return false, nil
+	return false
 }
 
 // isPendingPastDeadline checks if the certificate has a Pending status and the
@@ -198,20 +190,20 @@ func isIssued(csr *capi.CertificateSigningRequest) bool {
 
 // isExpired checks if the CSR has a certificate and the date in the `NotAfter`
 // field has gone by.
-func isExpired(csr *capi.CertificateSigningRequest) (bool, error) {
+func isExpired(csr *capi.CertificateSigningRequest) bool {
 	if len(csr.Status.Certificate) == 0 {
-		return false, nil
+		return false
 	}
 	block, _ := pem.Decode(csr.Status.Certificate)
 	if block == nil {
-		return false, fmt.Errorf("expected the certificate associated with the CSR to be PEM encoded")
+		return false
 	}
 	certs, err := x509.ParseCertificates(block.Bytes)
 	if err != nil {
-		return false, fmt.Errorf("unable to parse certificate data: %v", err)
+		return false
 	}
 	if len(certs) == 0 {
-		return false, fmt.Errorf("no certificates found")
+		return false
 	}
-	return time.Now().After(certs[0].NotAfter), nil
+	return time.Now().After(certs[0].NotAfter)
 }

vendor/k8s.io/kubernetes/pkg/kubelet/kubelet_volumes.go

@@ -18,6 +18,8 @@ package kubelet
 
 import (
 	"fmt"
+	"io/ioutil"
+	"path/filepath"
 	"syscall"
 
 	v1 "k8s.io/api/core/v1"
@@ -90,6 +92,57 @@ func (kl *Kubelet) newVolumeMounterFromPlugins(spec *volume.Spec, pod *v1.Pod, o
 	return physicalMounter, nil
 }
 
+// removeOrphanedPodVolumeDirs attempts to remove the pod volumes directory and
+// its subdirectories. There should be no files left under normal conditions
+// when this is called, so it effectively does a recursive rmdir instead of
+// RemoveAll to ensure it only removes directories and not regular files.
+func (kl *Kubelet) removeOrphanedPodVolumeDirs(uid types.UID) []error {
+	orphanVolumeErrors := []error{}
+
+	// If there are still volume directories, attempt to rmdir them
+	volumePaths, err := kl.getPodVolumePathListFromDisk(uid)
+	if err != nil {
+		orphanVolumeErrors = append(orphanVolumeErrors, fmt.Errorf("orphaned pod %q found, but error %v occurred during reading volume dir from disk", uid, err))
+		return orphanVolumeErrors
+	}
+	if len(volumePaths) > 0 {
+		for _, volumePath := range volumePaths {
+			if err := syscall.Rmdir(volumePath); err != nil {
+				orphanVolumeErrors = append(orphanVolumeErrors, fmt.Errorf("orphaned pod %q found, but failed to rmdir() volume at path %v: %v", uid, volumePath, err))
+			} else {
+				klog.InfoS("Cleaned up orphaned volume from pod", "podUID", uid, "path", volumePath)
+			}
+		}
+	}
+
+	// If there are any volume-subpaths, attempt to rmdir them
+	subpathVolumePaths, err := kl.getPodVolumeSubpathListFromDisk(uid)
+	if err != nil {
+		orphanVolumeErrors = append(orphanVolumeErrors, fmt.Errorf("orphaned pod %q found, but error %v occurred during reading of volume-subpaths dir from disk", uid, err))
+		return orphanVolumeErrors
+	}
+	if len(subpathVolumePaths) > 0 {
+		for _, subpathVolumePath := range subpathVolumePaths {
+			if err := syscall.Rmdir(subpathVolumePath); err != nil {
+				orphanVolumeErrors = append(orphanVolumeErrors, fmt.Errorf("orphaned pod %q found, but failed to rmdir() subpath at path %v: %v", uid, subpathVolumePath, err))
+			} else {
+				klog.InfoS("Cleaned up orphaned volume subpath from pod", "podUID", uid, "path", subpathVolumePath)
+			}
+		}
+	}
+
+	// Remove any remaining subdirectories along with the volumes directory itself.
+	// Fail if any regular files are encountered.
+	podVolDir := kl.getPodVolumesDir(uid)
+	if err := removeall.RemoveDirsOneFilesystem(kl.mounter, podVolDir); err != nil {
+		orphanVolumeErrors = append(orphanVolumeErrors, fmt.Errorf("orphaned pod %q found, but error %v occurred when trying to remove the volumes dir", uid, err))
+	} else {
+		klog.InfoS("Cleaned up orphaned pod volumes dir", "podUID", uid, "path", podVolDir)
+	}
+
+	return orphanVolumeErrors
+}
+
 // cleanupOrphanedPodDirs removes the volumes of pods that should not be
 // running and that have no containers running.  Note that we roll up logs here since it runs in the main loop.
 func (kl *Kubelet) cleanupOrphanedPodDirs(pods []*v1.Pod, runningPods []*kubecontainer.Pod) error {
@@ -122,55 +175,48 @@ func (kl *Kubelet) cleanupOrphanedPodDirs(pods []*v1.Pod, runningPods []*kubecon
 			continue
 		}
 
-		allVolumesCleanedUp := true
-
-		// If there are still volume directories, attempt to rmdir them
-		volumePaths, err := kl.getPodVolumePathListFromDisk(uid)
-		if err != nil {
-			orphanVolumeErrors = append(orphanVolumeErrors, fmt.Errorf("orphaned pod %q found, but error %v occurred during reading volume dir from disk", uid, err))
+		// Attempt to remove the pod volumes directory and its subdirs
+		podVolumeErrors := kl.removeOrphanedPodVolumeDirs(uid)
+		if len(podVolumeErrors) > 0 {
+			orphanVolumeErrors = append(orphanVolumeErrors, podVolumeErrors...)
+			// Not all volumes were removed, so don't clean up the pod directory yet. It is likely
+			// that there are still mountpoints or files left which could cause removal of the pod
+			// directory to fail below.
+			// Errors for all removal operations have already been recorded, so don't add another
+			// one here.
 			continue
 		}
-		if len(volumePaths) > 0 {
-			for _, volumePath := range volumePaths {
-				if err := syscall.Rmdir(volumePath); err != nil {
-					orphanVolumeErrors = append(orphanVolumeErrors, fmt.Errorf("orphaned pod %q found, but failed to rmdir() volume at path %v: %v", uid, volumePath, err))
-					allVolumesCleanedUp = false
-				} else {
-					klog.InfoS("Cleaned up orphaned volume from pod", "podUID", uid, "path", volumePath)
-				}
-			}
-		}
 
-		// If there are any volume-subpaths, attempt to rmdir them
-		subpathVolumePaths, err := kl.getPodVolumeSubpathListFromDisk(uid)
+		// Call RemoveAllOneFilesystem for remaining subdirs under the pod directory
+		podDir := kl.getPodDir(uid)
+		podSubdirs, err := ioutil.ReadDir(podDir)
 		if err != nil {
-			orphanVolumeErrors = append(orphanVolumeErrors, fmt.Errorf("orphaned pod %q found, but error %v occurred during reading of volume-subpaths dir from disk", uid, err))
+			klog.ErrorS(err, "Could not read directory", "path", podDir)
+			orphanRemovalErrors = append(orphanRemovalErrors, fmt.Errorf("orphaned pod %q found, but error %v occurred during reading the pod dir from disk", uid, err))
 			continue
 		}
-		if len(subpathVolumePaths) > 0 {
-			for _, subpathVolumePath := range subpathVolumePaths {
-				if err := syscall.Rmdir(subpathVolumePath); err != nil {
-					orphanVolumeErrors = append(orphanVolumeErrors, fmt.Errorf("orphaned pod %q found, but failed to rmdir() subpath at path %v: %v", uid, subpathVolumePath, err))
-					allVolumesCleanedUp = false
-				} else {
-					klog.InfoS("Cleaned up orphaned volume subpath from pod", "podUID", uid, "path", subpathVolumePath)
-				}
+		for _, podSubdir := range podSubdirs {
+			podSubdirName := podSubdir.Name()
+			podSubdirPath := filepath.Join(podDir, podSubdirName)
+			// Never attempt RemoveAllOneFilesystem on the volumes directory,
+			// as this could lead to data loss in some situations. The volumes
+			// directory should have been removed by removeOrphanedPodVolumeDirs.
+			if podSubdirName == "volumes" {
+				err := fmt.Errorf("volumes subdir was found after it was removed")
+				klog.ErrorS(err, "Orphaned pod found, but failed to remove volumes subdir", "podUID", uid, "path", podSubdirPath)
+				continue
+			}
+			if err := removeall.RemoveAllOneFilesystem(kl.mounter, podSubdirPath); err != nil {
+				klog.ErrorS(err, "Failed to remove orphaned pod subdir", "podUID", uid, "path", podSubdirPath)
+				orphanRemovalErrors = append(orphanRemovalErrors, fmt.Errorf("orphaned pod %q found, but error %v occurred when trying to remove subdir %q", uid, err, podSubdirPath))
 			}
 		}
 
-		if !allVolumesCleanedUp {
-			// Not all volumes were removed, so don't clean up the pod directory yet. It is likely
-			// that there are still mountpoints left which could stall RemoveAllOneFilesystem which
-			// would otherwise be called below.
-			// Errors for all removal operations have already been recorded, so don't add another
-			// one here.
-			continue
-		}
-
+		// Rmdir the pod dir, which should be empty if everything above was successful
 		klog.V(3).InfoS("Orphaned pod found, removing", "podUID", uid)
-		if err := removeall.RemoveAllOneFilesystem(kl.mounter, kl.getPodDir(uid)); err != nil {
+		if err := syscall.Rmdir(podDir); err != nil {
 			klog.ErrorS(err, "Failed to remove orphaned pod dir", "podUID", uid)
-			orphanRemovalErrors = append(orphanRemovalErrors, err)
+			orphanRemovalErrors = append(orphanRemovalErrors, fmt.Errorf("orphaned pod %q found, but error %v occurred when trying to remove the pod directory", uid, err))
 		}
 	}
 

vendor/k8s.io/kubernetes/pkg/scheduler/factory.go

@@ -144,6 +144,7 @@ func (c *Configurator) create() (*Scheduler, error) {
 		frameworkruntime.WithCaptureProfile(frameworkruntime.CaptureProfile(c.frameworkCapturer)),
 		frameworkruntime.WithClusterEventMap(clusterEventMap),
 		frameworkruntime.WithParallelism(int(c.parallellism)),
+		frameworkruntime.WithExtenders(extenders),
 	)
 	if err != nil {
 		return nil, fmt.Errorf("initializing profiles: %v", err)

vendor/k8s.io/kubernetes/pkg/util/removeall/OWNERS

+# See the OWNERS docs at https://go.k8s.io/owners
+
+approvers:
+- sig-storage-approvers
+reviewers:
+- sig-storage-reviewers
+labels:
+- sig/storage

vendor/k8s.io/kubernetes/pkg/util/removeall/removeall.go

@@ -25,16 +25,16 @@ import (
 	"k8s.io/mount-utils"
 )
 
-// RemoveAllOneFilesystem removes path and any children it contains.
-// It removes everything it can but returns the first error
-// it encounters. If the path does not exist, RemoveAll
+// RemoveAllOneFilesystemCommon removes the path and any children it contains,
+// using the provided remove function. It removes everything it can but returns
+// the first error it encounters. If the path does not exist, RemoveAll
 // returns nil (no error).
 // It makes sure it does not cross mount boundary, i.e. it does *not* remove
 // files from another filesystems. Like 'rm -rf --one-file-system'.
 // It is copied from RemoveAll() sources, with IsLikelyNotMountPoint
-func RemoveAllOneFilesystem(mounter mount.Interface, path string) error {
+func RemoveAllOneFilesystemCommon(mounter mount.Interface, path string, remove func(string) error) error {
 	// Simple case: if Remove works, we're done.
-	err := os.Remove(path)
+	err := remove(path)
 	if err == nil || os.IsNotExist(err) {
 		return nil
 	}
@@ -48,7 +48,7 @@ func RemoveAllOneFilesystem(mounter mount.Interface, path string) error {
 		return serr
 	}
 	if !dir.IsDir() {
-		// Not a directory; return the error from Remove.
+		// Not a directory; return the error from remove.
 		return err
 	}
 
@@ -76,7 +76,7 @@ func RemoveAllOneFilesystem(mounter mount.Interface, path string) error {
 	for {
 		names, err1 := fd.Readdirnames(100)
 		for _, name := range names {
-			err1 := RemoveAllOneFilesystem(mounter, path+string(os.PathSeparator)+name)
+			err1 := RemoveAllOneFilesystemCommon(mounter, path+string(os.PathSeparator)+name, remove)
 			if err == nil {
 				err = err1
 			}
@@ -97,7 +97,7 @@ func RemoveAllOneFilesystem(mounter mount.Interface, path string) error {
 	fd.Close()
 
 	// Remove directory.
-	err1 := os.Remove(path)
+	err1 := remove(path)
 	if err1 == nil || os.IsNotExist(err1) {
 		return nil
 	}
@@ -106,3 +106,23 @@ func RemoveAllOneFilesystem(mounter mount.Interface, path string) error {
 	}
 	return err
 }
+
+// RemoveAllOneFilesystem removes the path and any children it contains, using
+// the os.Remove function. It makes sure it does not cross mount boundaries,
+// i.e. it returns an error rather than remove files from another filesystem.
+// It removes everything it can but returns the first error it encounters.
+// If the path does not exist, it returns nil (no error).
+func RemoveAllOneFilesystem(mounter mount.Interface, path string) error {
+	return RemoveAllOneFilesystemCommon(mounter, path, os.Remove)
+}
+
+// RemoveDirsOneFilesystem removes the path and any empty subdirectories it
+// contains, using the syscall.Rmdir function. Unlike RemoveAllOneFilesystem,
+// RemoveDirsOneFilesystem will remove only directories and returns an error if
+// it encounters any files in the directory tree. It makes sure it does not
+// cross mount boundaries, i.e. it returns an error rather than remove dirs
+// from another filesystem. It removes everything it can but returns the first
+// error it encounters. If the path does not exist, it returns nil (no error).
+func RemoveDirsOneFilesystem(mounter mount.Interface, path string) error {
+	return RemoveAllOneFilesystemCommon(mounter, path, syscall.Rmdir)
+}

vendor/k8s.io/legacy-cloud-providers/azure/azure.go

@@ -274,6 +274,8 @@ type Cloud struct {
 	ipv6DualStackEnabled bool
 	// Lock for access to node caches, includes nodeZones, nodeResourceGroups, and unmanagedNodes.
 	nodeCachesLock sync.RWMutex
+	// nodeNames holds current nodes for tracking added nodes in VM caches.
+	nodeNames sets.String
 	// nodeZones is a mapping from Zone to a sets.String of Node's names in the Zone
 	// it is updated by the nodeInformer
 	nodeZones map[string]sets.String
@@ -342,6 +344,7 @@ func NewCloudWithoutFeatureGates(configReader io.Reader) (*Cloud, error) {
 	}
 
 	az := &Cloud{
+		nodeNames:          sets.NewString(),
 		nodeZones:          map[string]sets.String{},
 		nodeResourceGroups: map[string]string{},
 		unmanagedNodes:     sets.NewString(),
@@ -782,6 +785,9 @@ func (az *Cloud) updateNodeCaches(prevNode, newNode *v1.Node) {
 	defer az.nodeCachesLock.Unlock()
 
 	if prevNode != nil {
+		// Remove from nodeNames cache.
+		az.nodeNames.Delete(prevNode.ObjectMeta.Name)
+
 		// Remove from nodeZones cache.
 		prevZone, ok := prevNode.ObjectMeta.Labels[LabelFailureDomainBetaZone]
 		if ok && az.isAvailabilityZone(prevZone) {
@@ -805,6 +811,9 @@ func (az *Cloud) updateNodeCaches(prevNode, newNode *v1.Node) {
 	}
 
 	if newNode != nil {
+		// Add to nodeNames cache.
+		az.nodeNames.Insert(newNode.ObjectMeta.Name)
+
 		// Add to nodeZones cache.
 		newZone, ok := newNode.ObjectMeta.Labels[LabelFailureDomainBetaZone]
 		if ok && az.isAvailabilityZone(newZone) {
@@ -876,6 +885,22 @@ func (az *Cloud) GetNodeResourceGroup(nodeName string) (string, error) {
 	return az.ResourceGroup, nil
 }
 
+// GetNodeNames returns a set of all node names in the k8s cluster.
+func (az *Cloud) GetNodeNames() (sets.String, error) {
+	// Kubelet won't set az.nodeInformerSynced, return nil.
+	if az.nodeInformerSynced == nil {
+		return nil, nil
+	}
+
+	az.nodeCachesLock.RLock()
+	defer az.nodeCachesLock.RUnlock()
+	if !az.nodeInformerSynced() {
+		return nil, fmt.Errorf("node informer is not synced when trying to GetNodeNames")
+	}
+
+	return sets.NewString(az.nodeNames.List()...), nil
+}
+
 // GetResourceGroups returns a set of resource groups that all nodes are running on.
 func (az *Cloud) GetResourceGroups() (sets.String, error) {
 	// Kubelet won't set az.nodeInformerSynced, always return configured resourceGroup.

vendor/k8s.io/legacy-cloud-providers/azure/azure_loadbalancer.go

@@ -1258,13 +1258,6 @@ func (az *Cloud) reconcileLoadBalancer(clusterName string, service *v1.Service,
 			// construct FrontendIPConfigurationPropertiesFormat
 			var fipConfigurationProperties *network.FrontendIPConfigurationPropertiesFormat
 			if isInternal {
-				// azure does not support ILB for IPv6 yet.
-				// TODO: remove this check when ILB supports IPv6 *and* the SDK
-				// have been rev'ed to 2019* version
-				if utilnet.IsIPv6String(service.Spec.ClusterIP) {
-					return nil, fmt.Errorf("ensure(%s): lb(%s) - internal load balancers does not support IPv6", serviceName, lbName)
-				}
-
 				subnetName := subnet(service)
 				if subnetName == nil {
 					subnetName = &az.SubnetName
@@ -1282,6 +1275,10 @@ func (az *Cloud) reconcileLoadBalancer(clusterName string, service *v1.Service,
 					Subnet: &subnet,
 				}
 
+				if utilnet.IsIPv6String(service.Spec.ClusterIP) {
+					configProperties.PrivateIPAddressVersion = network.IPv6
+				}
+
 				loadBalancerIP := service.Spec.LoadBalancerIP
 				if loadBalancerIP != "" {
 					configProperties.PrivateIPAllocationMethod = network.Static

vendor/k8s.io/legacy-cloud-providers/azure/azure_managedDiskController.go

@@ -21,6 +21,7 @@ package azure
 import (
 	"context"
 	"fmt"
+	"net/http"
 	"path"
 	"strconv"
 	"strings"
@@ -236,6 +237,10 @@ func (c *ManagedDiskController) DeleteManagedDisk(diskURI string) error {
 
 	rerr = c.common.cloud.DisksClient.Delete(ctx, resourceGroup, diskName)
 	if rerr != nil {
+		if rerr.HTTPStatusCode == http.StatusNotFound {
+			klog.V(2).Infof("azureDisk - disk(%s) is already deleted", diskURI)
+			return nil
+		}
 		return rerr.Error()
 	}
 	// We don't need poll here, k8s will immediately stop referencing the disk

vendor/k8s.io/legacy-cloud-providers/azure/azure_routes.go

@@ -149,12 +149,17 @@ func (d *delayedRouteUpdater) updateRoutes() {
 	}
 
 	// reconcile routes.
-	dirty := false
+	dirty, onlyUpdateTags := false, true
 	routes := []network.Route{}
 	if routeTable.Routes != nil {
 		routes = *routeTable.Routes
 	}
-	onlyUpdateTags := true
+
+	routes, dirty = d.cleanupOutdatedRoutes(routes)
+	if dirty {
+		onlyUpdateTags = false
+	}
+
 	for _, rt := range d.routesToUpdate {
 		if rt.operation == routeTableOperationUpdateTags {
 			routeTable.Tags = rt.routeTableTags
@@ -204,6 +209,34 @@ func (d *delayedRouteUpdater) updateRoutes() {
 	}
 }
 
+// cleanupOutdatedRoutes deletes all non-dualstack routes when dualstack is enabled,
+// and deletes all dualstack routes when dualstack is not enabled.
+func (d *delayedRouteUpdater) cleanupOutdatedRoutes(existingRoutes []network.Route) (routes []network.Route, changed bool) {
+	for i := len(existingRoutes) - 1; i >= 0; i-- {
+		existingRouteName := to.String(existingRoutes[i].Name)
+		split := strings.Split(existingRouteName, routeNameSeparator)
+
+		// filter out unmanaged routes
+		deleteRoute := false
+		if d.az.nodeNames.Has(split[0]) {
+			if d.az.ipv6DualStackEnabled && len(split) == 1 {
+				klog.V(2).Infof("cleanupOutdatedRoutes: deleting outdated non-dualstack route %s", existingRouteName)
+				deleteRoute = true
+			} else if !d.az.ipv6DualStackEnabled && len(split) == 2 {
+				klog.V(2).Infof("cleanupOutdatedRoutes: deleting outdated dualstack route %s", existingRouteName)
+				deleteRoute = true
+			}
+
+			if deleteRoute {
+				existingRoutes = append(existingRoutes[:i], existingRoutes[i+1:]...)
+				changed = true
+			}
+		}
+	}
+
+	return existingRoutes, changed
+}
+
 // addRouteOperation adds the routeOperation to delayedRouteUpdater and returns a delayedRouteOperation.
 func (d *delayedRouteUpdater) addRouteOperation(operation routeOperation, route network.Route) (*delayedRouteOperation, error) {
 	d.lock.Lock()

vendor/k8s.io/legacy-cloud-providers/azure/azure_standard.go

@@ -1034,6 +1034,11 @@ func (as *availabilitySet) EnsureBackendPoolDeleted(service *v1.Service, backend
 }
 
 func getAvailabilitySetNameByID(asID string) (string, error) {
+	// for standalone VM
+	if asID == "" {
+		return "", nil
+	}
+
 	matches := vmasIDRE.FindStringSubmatch(asID)
 	if len(matches) != 2 {
 		return "", fmt.Errorf("getAvailabilitySetNameByID: failed to parse the VMAS ID %s", asID)

vendor/k8s.io/legacy-cloud-providers/azure/azure_vmss.go

@@ -74,7 +74,8 @@ type scaleSet struct {
 	*Cloud
 
 	// availabilitySet is also required for scaleSet because some instances
-	// (e.g. master nodes) may not belong to any scale sets.
+	// (e.g. control plane nodes) may not belong to any scale sets.
+	// this also allows for clusters with both VM and VMSS nodes.
 	availabilitySet VMSet
 
 	vmssCache                 *azcache.TimedCache
@@ -1376,8 +1377,17 @@ func (ss *scaleSet) ensureBackendPoolDeletedFromNode(nodeName, backendPoolID str
 func (ss *scaleSet) GetNodeNameByIPConfigurationID(ipConfigurationID string) (string, string, error) {
 	matches := vmssIPConfigurationRE.FindStringSubmatch(ipConfigurationID)
 	if len(matches) != 4 {
+		if ss.DisableAvailabilitySetNodes {
+			return "", "", ErrorNotVmssInstance
+		}
+
 		klog.V(4).Infof("Can not extract scale set name from ipConfigurationID (%s), assuming it is managed by availability set", ipConfigurationID)
-		return "", "", ErrorNotVmssInstance
+		name, rg, err := ss.availabilitySet.GetNodeNameByIPConfigurationID(ipConfigurationID)
+		if err != nil {
+			klog.Errorf("GetNodeNameByIPConfigurationID: failed to invoke availabilitySet.GetNodeNameByIPConfigurationID: %s", err.Error())
+			return "", "", err
+		}
+		return name, rg, nil
 	}
 
 	resourceGroup := matches[1]

vendor/k8s.io/legacy-cloud-providers/azure/azure_vmss_cache.go

@@ -58,6 +58,11 @@ type vmssEntry struct {
 	lastUpdate    time.Time
 }
 
+type availabilitySetEntry struct {
+	vmNames   sets.String
+	nodeNames sets.String
+}
+
 func (ss *scaleSet) newVMSSCache() (*azcache.TimedCache, error) {
 	getter := func(key string) (interface{}, error) {
 		localCache := &sync.Map{} // [vmssName]*vmssEntry
@@ -278,7 +283,7 @@ func (ss *scaleSet) deleteCacheForNode(nodeName string) error {
 
 func (ss *scaleSet) newAvailabilitySetNodesCache() (*azcache.TimedCache, error) {
 	getter := func(key string) (interface{}, error) {
-		localCache := sets.NewString()
+		vmNames := sets.NewString()
 		resourceGroups, err := ss.GetResourceGroups()
 		if err != nil {
 			return nil, err
@@ -292,11 +297,22 @@ func (ss *scaleSet) newAvailabilitySetNodesCache() (*azcache.TimedCache, error)
 
 			for _, vm := range vmList {
 				if vm.Name != nil {
-					localCache.Insert(*vm.Name)
+					vmNames.Insert(*vm.Name)
 				}
 			}
 		}
 
+		// store all the node names in the cluster when the cache data was created.
+		nodeNames, err := ss.GetNodeNames()
+		if err != nil {
+			return nil, err
+		}
+
+		localCache := availabilitySetEntry{
+			vmNames:   vmNames,
+			nodeNames: nodeNames,
+		}
+
 		return localCache, nil
 	}
 
@@ -318,6 +334,16 @@ func (ss *scaleSet) isNodeManagedByAvailabilitySet(nodeName string, crt azcache.
 		return false, err
 	}
 
-	availabilitySetNodes := cached.(sets.String)
-	return availabilitySetNodes.Has(nodeName), nil
+	cachedNodes := cached.(availabilitySetEntry).nodeNames
+	// if the node is not in the cache, assume the node has joined after the last cache refresh and attempt to refresh the cache.
+	if !cachedNodes.Has(nodeName) {
+		klog.V(2).Infof("Node %s has joined the cluster since the last VM cache refresh, refreshing the cache", nodeName)
+		cached, err = ss.availabilitySetNodesCache.Get(availabilitySetNodesKey, azcache.CacheReadTypeForceRefresh)
+		if err != nil {
+			return false, err
+		}
+	}
+
+	cachedVMs := cached.(availabilitySetEntry).vmNames
+	return cachedVMs.Has(nodeName), nil
 }

vendor/sigs.k8s.io/apiserver-network-proxy/konnectivity-client/pkg/client/client.go

@@ -35,7 +35,7 @@ import (
 type Tunnel interface {
 	// Dial connects to the address on the named network, similar to
 	// what net.Dial does. The only supported protocol is tcp.
-	Dial(protocol, address string) (net.Conn, error)
+	DialContext(ctx context.Context, protocol, address string) (net.Conn, error)
 }
 
 type dialResult struct {
@@ -66,15 +66,15 @@ var _ clientConn = &grpc.ClientConn{}
 // gRPC based proxy service.
 // Currently, a single tunnel supports a single connection, and the tunnel is closed when the connection is terminated
 // The Dial() method of the returned tunnel should only be called once
-func CreateSingleUseGrpcTunnel(address string, opts ...grpc.DialOption) (Tunnel, error) {
-	c, err := grpc.Dial(address, opts...)
+func CreateSingleUseGrpcTunnel(ctx context.Context, address string, opts ...grpc.DialOption) (Tunnel, error) {
+	c, err := grpc.DialContext(ctx, address, opts...)
 	if err != nil {
 		return nil, err
 	}
 
 	grpcClient := client.NewProxyServiceClient(c)
 
-	stream, err := grpcClient.Proxy(context.Background())
+	stream, err := grpcClient.Proxy(ctx)
 	if err != nil {
 		return nil, err
 	}
@@ -175,7 +175,7 @@ func (t *grpcTunnel) serve(c clientConn) {
 
 // Dial connects to the address on the named network, similar to
 // what net.Dial does. The only supported protocol is tcp.
-func (t *grpcTunnel) Dial(protocol, address string) (net.Conn, error) {
+func (t *grpcTunnel) DialContext(ctx context.Context, protocol, address string) (net.Conn, error) {
 	if protocol != "tcp" {
 		return nil, errors.New("protocol not supported")
 	}
@@ -224,7 +224,9 @@ func (t *grpcTunnel) Dial(protocol, address string) (net.Conn, error) {
 		t.conns[res.connid] = c
 		t.connsLock.Unlock()
 	case <-time.After(30 * time.Second):
-		return nil, errors.New("dial timeout")
+		return nil, errors.New("dial timeout, backstop")
+	case <-ctx.Done():
+		return nil, errors.New("dial timeout, context")
 	}
 
 	return c, nil