Update to v1.19.12 (#3476)

go.mod

@@ -34,31 +34,31 @@ replace (
 	google.golang.org/genproto => google.golang.org/genproto v0.0.0-20200224152610-e50cd9704f63
 	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.19.11-k3s1
-	k8s.io/apiextensions-apiserver => github.com/k3s-io/kubernetes/staging/src/k8s.io/apiextensions-apiserver v1.19.11-k3s1
-	k8s.io/apimachinery => github.com/k3s-io/kubernetes/staging/src/k8s.io/apimachinery v1.19.11-k3s1
-	k8s.io/apiserver => github.com/k3s-io/kubernetes/staging/src/k8s.io/apiserver v1.19.11-k3s1
-	k8s.io/cli-runtime => github.com/k3s-io/kubernetes/staging/src/k8s.io/cli-runtime v1.19.11-k3s1
-	k8s.io/client-go => github.com/k3s-io/kubernetes/staging/src/k8s.io/client-go v1.19.11-k3s1
-	k8s.io/cloud-provider => github.com/k3s-io/kubernetes/staging/src/k8s.io/cloud-provider v1.19.11-k3s1
-	k8s.io/cluster-bootstrap => github.com/k3s-io/kubernetes/staging/src/k8s.io/cluster-bootstrap v1.19.11-k3s1
-	k8s.io/code-generator => github.com/k3s-io/kubernetes/staging/src/k8s.io/code-generator v1.19.11-k3s1
-	k8s.io/component-base => github.com/k3s-io/kubernetes/staging/src/k8s.io/component-base v1.19.11-k3s1
-	k8s.io/cri-api => github.com/k3s-io/kubernetes/staging/src/k8s.io/cri-api v1.19.11-k3s1
-	k8s.io/csi-translation-lib => github.com/k3s-io/kubernetes/staging/src/k8s.io/csi-translation-lib v1.19.11-k3s1
-	k8s.io/kube-aggregator => github.com/k3s-io/kubernetes/staging/src/k8s.io/kube-aggregator v1.19.11-k3s1
-	k8s.io/kube-controller-manager => github.com/k3s-io/kubernetes/staging/src/k8s.io/kube-controller-manager v1.19.11-k3s1
-	k8s.io/kube-proxy => github.com/k3s-io/kubernetes/staging/src/k8s.io/kube-proxy v1.19.11-k3s1
-	k8s.io/kube-scheduler => github.com/k3s-io/kubernetes/staging/src/k8s.io/kube-scheduler v1.19.11-k3s1
-	k8s.io/kubectl => github.com/k3s-io/kubernetes/staging/src/k8s.io/kubectl v1.19.11-k3s1
-	k8s.io/kubelet => github.com/k3s-io/kubernetes/staging/src/k8s.io/kubelet v1.19.11-k3s1
-	k8s.io/kubernetes => github.com/k3s-io/kubernetes v1.19.11-k3s1
-	k8s.io/legacy-cloud-providers => github.com/k3s-io/kubernetes/staging/src/k8s.io/legacy-cloud-providers v1.19.11-k3s1
-	k8s.io/metrics => github.com/k3s-io/kubernetes/staging/src/k8s.io/metrics v1.19.11-k3s1
-	k8s.io/node-api => github.com/k3s-io/kubernetes/staging/src/k8s.io/node-api v1.19.11-k3s1
-	k8s.io/sample-apiserver => github.com/k3s-io/kubernetes/staging/src/k8s.io/sample-apiserver v1.19.11-k3s1
-	k8s.io/sample-cli-plugin => github.com/k3s-io/kubernetes/staging/src/k8s.io/sample-cli-plugin v1.19.11-k3s1
-	k8s.io/sample-controller => github.com/k3s-io/kubernetes/staging/src/k8s.io/sample-controller v1.19.11-k3s1
+	k8s.io/api => github.com/k3s-io/kubernetes/staging/src/k8s.io/api v1.19.12-k3s1
+	k8s.io/apiextensions-apiserver => github.com/k3s-io/kubernetes/staging/src/k8s.io/apiextensions-apiserver v1.19.12-k3s1
+	k8s.io/apimachinery => github.com/k3s-io/kubernetes/staging/src/k8s.io/apimachinery v1.19.12-k3s1
+	k8s.io/apiserver => github.com/k3s-io/kubernetes/staging/src/k8s.io/apiserver v1.19.12-k3s1
+	k8s.io/cli-runtime => github.com/k3s-io/kubernetes/staging/src/k8s.io/cli-runtime v1.19.12-k3s1
+	k8s.io/client-go => github.com/k3s-io/kubernetes/staging/src/k8s.io/client-go v1.19.12-k3s1
+	k8s.io/cloud-provider => github.com/k3s-io/kubernetes/staging/src/k8s.io/cloud-provider v1.19.12-k3s1
+	k8s.io/cluster-bootstrap => github.com/k3s-io/kubernetes/staging/src/k8s.io/cluster-bootstrap v1.19.12-k3s1
+	k8s.io/code-generator => github.com/k3s-io/kubernetes/staging/src/k8s.io/code-generator v1.19.12-k3s1
+	k8s.io/component-base => github.com/k3s-io/kubernetes/staging/src/k8s.io/component-base v1.19.12-k3s1
+	k8s.io/cri-api => github.com/k3s-io/kubernetes/staging/src/k8s.io/cri-api v1.19.12-k3s1
+	k8s.io/csi-translation-lib => github.com/k3s-io/kubernetes/staging/src/k8s.io/csi-translation-lib v1.19.12-k3s1
+	k8s.io/kube-aggregator => github.com/k3s-io/kubernetes/staging/src/k8s.io/kube-aggregator v1.19.12-k3s1
+	k8s.io/kube-controller-manager => github.com/k3s-io/kubernetes/staging/src/k8s.io/kube-controller-manager v1.19.12-k3s1
+	k8s.io/kube-proxy => github.com/k3s-io/kubernetes/staging/src/k8s.io/kube-proxy v1.19.12-k3s1
+	k8s.io/kube-scheduler => github.com/k3s-io/kubernetes/staging/src/k8s.io/kube-scheduler v1.19.12-k3s1
+	k8s.io/kubectl => github.com/k3s-io/kubernetes/staging/src/k8s.io/kubectl v1.19.12-k3s1
+	k8s.io/kubelet => github.com/k3s-io/kubernetes/staging/src/k8s.io/kubelet v1.19.12-k3s1
+	k8s.io/kubernetes => github.com/k3s-io/kubernetes v1.19.12-k3s1
+	k8s.io/legacy-cloud-providers => github.com/k3s-io/kubernetes/staging/src/k8s.io/legacy-cloud-providers v1.19.12-k3s1
+	k8s.io/metrics => github.com/k3s-io/kubernetes/staging/src/k8s.io/metrics v1.19.12-k3s1
+	k8s.io/node-api => github.com/k3s-io/kubernetes/staging/src/k8s.io/node-api v1.19.12-k3s1
+	k8s.io/sample-apiserver => github.com/k3s-io/kubernetes/staging/src/k8s.io/sample-apiserver v1.19.12-k3s1
+	k8s.io/sample-cli-plugin => github.com/k3s-io/kubernetes/staging/src/k8s.io/sample-cli-plugin v1.19.12-k3s1
+	k8s.io/sample-controller => github.com/k3s-io/kubernetes/staging/src/k8s.io/sample-controller v1.19.12-k3s1
 	mvdan.cc/unparam => mvdan.cc/unparam v0.0.0-20190209190245-fbb59629db34
 )
 
@@ -109,15 +109,15 @@ require (
 	golang.org/x/sys v0.0.0-20201201145000-ef89a241ccb3
 	google.golang.org/grpc v1.31.1
 	gopkg.in/yaml.v2 v2.3.0
-	k8s.io/api v0.19.11
-	k8s.io/apimachinery v0.19.11
-	k8s.io/apiserver v0.19.11
+	k8s.io/api v0.19.12
+	k8s.io/apimachinery v0.19.12
+	k8s.io/apiserver v0.19.12
 	k8s.io/client-go v11.0.1-0.20190409021438-1a26190bd76a+incompatible
-	k8s.io/cloud-provider v0.19.11
-	k8s.io/component-base v0.19.11
-	k8s.io/cri-api v0.19.11
+	k8s.io/cloud-provider v0.19.12
+	k8s.io/component-base v0.19.12
+	k8s.io/cri-api v0.19.12
 	k8s.io/klog v1.0.0
-	k8s.io/kubernetes v1.19.11
+	k8s.io/kubernetes v1.19.12
 	k8s.io/utils v0.0.0-20200729134348-d5654de09c73
 	sigs.k8s.io/yaml v1.2.0
 )

vendor/k8s.io/apimachinery/pkg/util/httpstream/httpstream.go

@@ -78,8 +78,6 @@ type Connection interface {
 	// SetIdleTimeout sets the amount of time the connection may remain idle before
 	// it is automatically closed.
 	SetIdleTimeout(timeout time.Duration)
-	// RemoveStreams can be used to remove a set of streams from the Connection.
-	RemoveStreams(streams ...Stream)
 }
 
 // Stream represents a bidirectional communications channel that is part of an

vendor/k8s.io/apimachinery/pkg/util/httpstream/spdy/connection.go

@@ -31,7 +31,7 @@ import (
 // streams.
 type connection struct {
 	conn             *spdystream.Connection
-	streams          map[uint32]httpstream.Stream
+	streams          []httpstream.Stream
 	streamLock       sync.Mutex
 	newStreamHandler httpstream.NewStreamHandler
 }
@@ -64,11 +64,7 @@ func NewServerConnection(conn net.Conn, newStreamHandler httpstream.NewStreamHan
 // will be invoked when the server receives a newly created stream from the
 // client.
 func newConnection(conn *spdystream.Connection, newStreamHandler httpstream.NewStreamHandler) httpstream.Connection {
-	c := &connection{
-		conn:             conn,
-		newStreamHandler: newStreamHandler,
-		streams:          make(map[uint32]httpstream.Stream),
-	}
+	c := &connection{conn: conn, newStreamHandler: newStreamHandler}
 	go conn.Serve(c.newSpdyStream)
 	return c
 }
@@ -85,7 +81,7 @@ func (c *connection) Close() error {
 		// calling Reset instead of Close ensures that all streams are fully torn down
 		s.Reset()
 	}
-	c.streams = make(map[uint32]httpstream.Stream, 0)
+	c.streams = make([]httpstream.Stream, 0)
 	c.streamLock.Unlock()
 
 	// now that all streams are fully torn down, it's safe to call close on the underlying connection,
@@ -94,15 +90,6 @@ func (c *connection) Close() error {
 	return c.conn.Close()
 }
 
-// RemoveStreams can be used to removes a set of streams from the Connection.
-func (c *connection) RemoveStreams(streams ...httpstream.Stream) {
-	c.streamLock.Lock()
-	for _, stream := range streams {
-		delete(c.streams, stream.Identifier())
-	}
-	c.streamLock.Unlock()
-}
-
 // CreateStream creates a new stream with the specified headers and registers
 // it with the connection.
 func (c *connection) CreateStream(headers http.Header) (httpstream.Stream, error) {
@@ -122,7 +109,7 @@ func (c *connection) CreateStream(headers http.Header) (httpstream.Stream, error
 // it owns.
 func (c *connection) registerStream(s httpstream.Stream) {
 	c.streamLock.Lock()
-	c.streams[s.Identifier()] = s
+	c.streams = append(c.streams, s)
 	c.streamLock.Unlock()
 }
 

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

@@ -25,6 +25,8 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 )
 
+const totalAnnotationSizeLimitB int64 = 256 * (1 << 10) // 256 kB
+
 type lastAppliedUpdater struct {
 	fieldManager Manager
 }
@@ -78,8 +80,8 @@ func hasLastApplied(obj runtime.Object) bool {
 	if annotations == nil {
 		return false
 	}
-	_, ok := annotations[corev1.LastAppliedConfigAnnotation]
-	return ok
+	lastApplied, ok := annotations[corev1.LastAppliedConfigAnnotation]
+	return ok && len(lastApplied) > 0
 }
 
 func setLastApplied(obj runtime.Object, value string) error {
@@ -92,6 +94,9 @@ func setLastApplied(obj runtime.Object, value string) error {
 		annotations = map[string]string{}
 	}
 	annotations[corev1.LastAppliedConfigAnnotation] = value
+	if isAnnotationsValid(annotations) != nil {
+		delete(annotations, corev1.LastAppliedConfigAnnotation)
+	}
 	accessor.SetAnnotations(annotations)
 	return nil
 }
@@ -115,3 +120,14 @@ func buildLastApplied(obj runtime.Object) (string, error) {
 	}
 	return string(lastApplied), nil
 }
+
+func isAnnotationsValid(annotations map[string]string) error {
+	var totalSize int64
+	for k, v := range annotations {
+		totalSize += (int64)(len(k)) + (int64)(len(v))
+	}
+	if totalSize > (int64)(totalAnnotationSizeLimitB) {
+		return fmt.Errorf("annotations size %d is larger than limit %d", totalSize, totalAnnotationSizeLimitB)
+	}
+	return nil
+}

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

@@ -3,8 +3,8 @@ package version
 var (
 	gitMajor = "1"
 	gitMinor = "19"
-	gitVersion   = "v1.19.11-k3s1"
-	gitCommit    = "4230b2ac644cf72064bf7cacdf2ffa13292b3c1f"
+	gitVersion   = "v1.19.12-k3s1"
+	gitCommit    = "3e32ea1663306f5e5b7f386fc190f0afefc80454"
 	gitTreeState = "clean"
-	buildDate = "2021-05-14T11:17:50Z"
+	buildDate = "2021-06-17T18:50:24Z"
 )

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

@@ -3,8 +3,8 @@ package version
 var (
 	gitMajor = "1"
 	gitMinor = "19"
-	gitVersion   = "v1.19.11-k3s1"
-	gitCommit    = "4230b2ac644cf72064bf7cacdf2ffa13292b3c1f"
+	gitVersion   = "v1.19.12-k3s1"
+	gitCommit    = "3e32ea1663306f5e5b7f386fc190f0afefc80454"
 	gitTreeState = "clean"
-	buildDate = "2021-05-14T11:17:50Z"
+	buildDate = "2021-06-17T18:50:24Z"
 )

vendor/k8s.io/kube-aggregator/pkg/apiserver/BUILD

@@ -12,6 +12,7 @@ go_test(
         "handler_apis_test.go",
         "handler_proxy_test.go",
     ],
+    data = glob(["testdata/**"]),
     embed = [":go_default_library"],
     deps = [
         "//staging/src/k8s.io/apimachinery/pkg/api/equality:go_default_library",

vendor/k8s.io/kubernetes/cmd/kubelet/app/server_linux.go

@@ -29,6 +29,7 @@ func watchForLockfileContention(path string, done chan struct{}) error {
 	}
 	if err = watcher.AddWatch(path, inotify.InOpen|inotify.InDeleteSelf); err != nil {
 		klog.Errorf("unable to watch lockfile: %v", err)
+		watcher.Close()
 		return err
 	}
 	go func() {
@@ -39,6 +40,7 @@ func watchForLockfileContention(path string, done chan struct{}) error {
 			klog.Errorf("inotify watcher error: %v", err)
 		}
 		close(done)
+		watcher.Close()
 	}()
 	return nil
 }

vendor/k8s.io/kubernetes/pkg/controller/volume/attachdetach/BUILD

@@ -20,6 +20,7 @@ go_library(
         "//pkg/controller/volume/common:go_default_library",
         "//pkg/features:go_default_library",
         "//pkg/volume:go_default_library",
+        "//pkg/volume/csi:go_default_library",
         "//pkg/volume/csimigration:go_default_library",
         "//pkg/volume/util:go_default_library",
         "//pkg/volume/util/operationexecutor:go_default_library",
@@ -59,13 +60,18 @@ go_test(
         "//pkg/controller:go_default_library",
         "//pkg/controller/volume/attachdetach/cache:go_default_library",
         "//pkg/controller/volume/attachdetach/testing:go_default_library",
+        "//pkg/features:go_default_library",
         "//pkg/volume:go_default_library",
+        "//pkg/volume/csi:go_default_library",
+        "//pkg/volume/util:go_default_library",
         "//staging/src/k8s.io/api/core/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/labels:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/util/feature:go_default_library",
         "//staging/src/k8s.io/client-go/informers:go_default_library",
         "//staging/src/k8s.io/client-go/tools/cache:go_default_library",
+        "//staging/src/k8s.io/component-base/featuregate/testing:go_default_library",
     ],
 )
 

vendor/k8s.io/kubernetes/pkg/controller/volume/attachdetach/attach_detach_controller.go

@@ -56,6 +56,7 @@ import (
 	"k8s.io/kubernetes/pkg/controller/volume/common"
 	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/pkg/volume"
+	"k8s.io/kubernetes/pkg/volume/csi"
 	"k8s.io/kubernetes/pkg/volume/csimigration"
 	volumeutil "k8s.io/kubernetes/pkg/volume/util"
 	"k8s.io/kubernetes/pkg/volume/util/operationexecutor"
@@ -709,17 +710,45 @@ func (adc *attachDetachController) processVolumeAttachments() error {
 			klog.Errorf("Unable to lookup pv object for: %q, err: %v", *pvName, err)
 			continue
 		}
+
+		var plugin volume.AttachableVolumePlugin
 		volumeSpec := volume.NewSpecFromPersistentVolume(pv, false)
-		plugin, err := adc.volumePluginMgr.FindAttachablePluginBySpec(volumeSpec)
-		if err != nil || plugin == nil {
-			// Currently VA objects are created for CSI volumes only. nil plugin is unexpected, generate a warning
-			klog.Warningf(
-				"Skipping processing the volume %q on nodeName: %q, no attacher interface found. err=%v",
-				*pvName,
-				nodeName,
-				err)
-			continue
+
+		// Consult csiMigratedPluginManager first before querying the plugins registered during runtime in volumePluginMgr.
+		// In-tree plugins that provisioned PVs will not be registered anymore after migration to CSI, once the respective
+		// feature gate is enabled.
+		if inTreePluginName, err := adc.csiMigratedPluginManager.GetInTreePluginNameFromSpec(pv, nil); err == nil {
+			if adc.csiMigratedPluginManager.IsMigrationEnabledForPlugin(inTreePluginName) {
+				// PV is migrated and should be handled by the CSI plugin instead of the in-tree one
+				plugin, _ = adc.volumePluginMgr.FindAttachablePluginByName(csi.CSIPluginName)
+				// podNamespace is not needed here for Azurefile as the volumeName generated will be the same with or without podNamespace
+				volumeSpec, err = csimigration.TranslateInTreeSpecToCSI(volumeSpec, "" /* podNamespace */, adc.intreeToCSITranslator)
+				if err != nil {
+					klog.Errorf(
+						"Failed to translate intree volumeSpec to CSI volumeSpec for volume:%q, va.Name:%q, nodeName:%q: %s. Error: %v",
+						*pvName,
+						va.Name,
+						nodeName,
+						inTreePluginName,
+						err)
+					continue
+				}
+			}
 		}
+
+		if plugin == nil {
+			plugin, err = adc.volumePluginMgr.FindAttachablePluginBySpec(volumeSpec)
+			if err != nil || plugin == nil {
+				// Currently VA objects are created for CSI volumes only. nil plugin is unexpected, generate a warning
+				klog.Warningf(
+					"Skipping processing the volume %q on nodeName: %q, no attacher interface found. err=%v",
+					*pvName,
+					nodeName,
+					err)
+				continue
+			}
+		}
+
 		volumeName, err := volumeutil.GetUniqueVolumeNameFromSpec(plugin, volumeSpec)
 		if err != nil {
 			klog.Errorf(

vendor/k8s.io/kubernetes/pkg/kubelet/cri/streaming/portforward/httpstream.go

@@ -163,10 +163,6 @@ func (h *httpStreamHandler) removeStreamPair(requestID string) {
 	h.streamPairsLock.Lock()
 	defer h.streamPairsLock.Unlock()
 
-	if h.conn != nil {
-		pair := h.streamPairs[requestID]
-		h.conn.RemoveStreams(pair.dataStream, pair.errorStream)
-	}
 	delete(h.streamPairs, requestID)
 }
 

vendor/k8s.io/kubernetes/pkg/kubelet/dockershim/docker_stats_windows.go

@@ -20,6 +20,7 @@ package dockershim
 
 import (
 	"context"
+	"strings"
 	"time"
 
 	"github.com/Microsoft/hcsshim"
@@ -39,7 +40,7 @@ func (ds *dockerService) getContainerStats(containerID string) (*runtimeapi.Cont
 		// That will typically happen with init-containers in Exited state. Docker still knows about them but the HCS does not.
 		// As we don't want to block stats retrieval for other containers, we only log errors.
 		if !hcsshim.IsNotExist(err) && !hcsshim.IsAlreadyStopped(err) {
-			klog.Errorf("Error opening container (stats will be missing) '%s': %v", containerID, err)
+			klog.V(4).Infof("Error opening container (stats will be missing) '%s': %v", containerID, err)
 		}
 		return nil, nil
 	}
@@ -52,6 +53,16 @@ func (ds *dockerService) getContainerStats(containerID string) (*runtimeapi.Cont
 
 	stats, err := hcsshim_container.Statistics()
 	if err != nil {
+		if strings.Contains(err.Error(), "0x5") || strings.Contains(err.Error(), "0xc0370105") {
+			// When the container is just created, querying for stats causes access errors because it hasn't started yet
+			// This is transient; skip container for now
+			//
+			// These hcs errors do not have helpers exposed in public package so need to query for the known codes
+			// https://github.com/microsoft/hcsshim/blob/master/internal/hcs/errors.go
+			// PR to expose helpers in hcsshim: https://github.com/microsoft/hcsshim/pull/933
+			klog.V(4).Infof("Container is not in a state that stats can be accessed '%s': %v. This occurs when the container is created but not started.", containerID, err)
+			return nil, nil
+		}
 		return nil, err
 	}
 

vendor/k8s.io/kubernetes/pkg/proxy/winkernel/proxier.go

@@ -317,7 +317,7 @@ func newSourceVIP(hns HostNetworkService, network string, ip string, mac string,
 
 func (ep *endpointsInfo) Cleanup() {
 	Log(ep, "Endpoint Cleanup", 3)
-	if ep.refCount != nil {
+	if !ep.GetIsLocal() && ep.refCount != nil {
 		*ep.refCount--
 
 		// Remove the remote hns endpoint, if no service is referring it
@@ -1112,10 +1112,10 @@ func (proxier *Proxier) syncProxyRules() {
 			} else {
 				// We only share the refCounts for remote endpoints
 				ep.refCount = proxier.endPointsRefCount.getRefCount(newHnsEndpoint.hnsID)
+				*ep.refCount++
 			}
 
 			ep.hnsID = newHnsEndpoint.hnsID
-			*ep.refCount++
 
 			Log(ep, "Endpoint resource found", 3)
 		}

vendor/k8s.io/kubernetes/pkg/registry/rbac/rest/storage_rbac.go

@@ -165,6 +165,7 @@ func (p *PolicyData) EnsureRBACPolicy() genericapiserver.PostStartHookFunc {
 		// initializing roles is really important.  On some e2e runs, we've seen cases where etcd is down when the server
 		// starts, the roles don't initialize, and nothing works.
 		err := wait.Poll(1*time.Second, 30*time.Second, func() (done bool, err error) {
+			failedReconciliation := false
 
 			coreclientset, err := corev1client.NewForConfig(hookContext.LoopbackClientConfig)
 			if err != nil {
@@ -224,6 +225,7 @@ func (p *PolicyData) EnsureRBACPolicy() genericapiserver.PostStartHookFunc {
 				if err != nil {
 					// don't fail on failures, try to create as many as you can
 					utilruntime.HandleError(fmt.Errorf("unable to reconcile clusterrole.%s/%s: %v", rbac.GroupName, clusterRole.Name, err))
+					failedReconciliation = true
 				}
 			}
 
@@ -254,6 +256,7 @@ func (p *PolicyData) EnsureRBACPolicy() genericapiserver.PostStartHookFunc {
 				if err != nil {
 					// don't fail on failures, try to create as many as you can
 					utilruntime.HandleError(fmt.Errorf("unable to reconcile clusterrolebinding.%s/%s: %v", rbac.GroupName, clusterRoleBinding.Name, err))
+					failedReconciliation = true
 				}
 			}
 
@@ -283,6 +286,7 @@ func (p *PolicyData) EnsureRBACPolicy() genericapiserver.PostStartHookFunc {
 					if err != nil {
 						// don't fail on failures, try to create as many as you can
 						utilruntime.HandleError(fmt.Errorf("unable to reconcile role.%s/%s in %v: %v", rbac.GroupName, role.Name, namespace, err))
+						failedReconciliation = true
 					}
 				}
 			}
@@ -315,9 +319,14 @@ func (p *PolicyData) EnsureRBACPolicy() genericapiserver.PostStartHookFunc {
 					if err != nil {
 						// don't fail on failures, try to create as many as you can
 						utilruntime.HandleError(fmt.Errorf("unable to reconcile rolebinding.%s/%s in %v: %v", rbac.GroupName, roleBinding.Name, namespace, err))
+						failedReconciliation = true
 					}
 				}
 			}
+			// failed to reconcile some objects, retry
+			if failedReconciliation {
+				return false, nil
+			}
 
 			return true, nil
 		})

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

@@ -158,7 +158,7 @@ func (c *Configurator) create() (*Scheduler, error) {
 	}
 
 	// The nominator will be passed all the way to framework instantiation.
-	nominator := internalqueue.NewPodNominator()
+	nominator := internalqueue.NewSafePodNominator(c.informerFactory.Core().V1().Pods().Lister())
 	profiles, err := profile.NewMap(c.profiles, c.buildFramework, c.recorderFactory,
 		frameworkruntime.WithPodNominator(nominator))
 	if err != nil {

vendor/k8s.io/kubernetes/pkg/scheduler/framework/plugins/volumezone/BUILD

@@ -10,6 +10,7 @@ go_library(
         "//pkg/scheduler/framework/v1alpha1:go_default_library",
         "//staging/src/k8s.io/api/core/v1:go_default_library",
         "//staging/src/k8s.io/api/storage/v1:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/api/errors:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/runtime:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/sets:go_default_library",
         "//staging/src/k8s.io/client-go/listers/core/v1:go_default_library",

vendor/k8s.io/kubernetes/pkg/scheduler/framework/plugins/volumezone/volume_zone.go

@@ -22,6 +22,7 @@ import (
 
 	v1 "k8s.io/api/core/v1"
 	storage "k8s.io/api/storage/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/sets"
 	corelisters "k8s.io/client-go/listers/core/v1"
@@ -108,47 +109,38 @@ func (pl *VolumeZone) Filter(ctx context.Context, _ *framework.CycleState, pod *
 		}
 		pvcName := volume.PersistentVolumeClaim.ClaimName
 		if pvcName == "" {
-			return framework.NewStatus(framework.Error, "PersistentVolumeClaim had no name")
+			return framework.NewStatus(framework.UnschedulableAndUnresolvable, "PersistentVolumeClaim had no name")
 		}
 		pvc, err := pl.pvcLister.PersistentVolumeClaims(pod.Namespace).Get(pvcName)
-		if err != nil {
-			return framework.NewStatus(framework.Error, err.Error())
-		}
-
-		if pvc == nil {
-			return framework.NewStatus(framework.Error, fmt.Sprintf("PersistentVolumeClaim was not found: %q", pvcName))
+		if s := getErrorAsStatus(err); !s.IsSuccess() {
+			return s
 		}
 
 		pvName := pvc.Spec.VolumeName
 		if pvName == "" {
 			scName := v1helper.GetPersistentVolumeClaimClass(pvc)
 			if len(scName) == 0 {
-				return framework.NewStatus(framework.Error, fmt.Sprint("PersistentVolumeClaim had no pv name and storageClass name"))
+				return framework.NewStatus(framework.UnschedulableAndUnresolvable, "PersistentVolumeClaim had no pv name and storageClass name")
 			}
 
-			class, _ := pl.scLister.Get(scName)
-			if class == nil {
-				return framework.NewStatus(framework.Error, fmt.Sprintf("StorageClass %q claimed by PersistentVolumeClaim %q not found", scName, pvcName))
-
+			class, err := pl.scLister.Get(scName)
+			if s := getErrorAsStatus(err); !s.IsSuccess() {
+				return s
 			}
 			if class.VolumeBindingMode == nil {
-				return framework.NewStatus(framework.Error, fmt.Sprintf("VolumeBindingMode not set for StorageClass %q", scName))
+				return framework.NewStatus(framework.UnschedulableAndUnresolvable, fmt.Sprintf("VolumeBindingMode not set for StorageClass %q", scName))
 			}
 			if *class.VolumeBindingMode == storage.VolumeBindingWaitForFirstConsumer {
 				// Skip unbound volumes
 				continue
 			}
 
-			return framework.NewStatus(framework.Error, fmt.Sprint("PersistentVolume had no name"))
+			return framework.NewStatus(framework.UnschedulableAndUnresolvable, "PersistentVolume had no name")
 		}
 
 		pv, err := pl.pvLister.Get(pvName)
-		if err != nil {
-			return framework.NewStatus(framework.Error, err.Error())
-		}
-
-		if pv == nil {
-			return framework.NewStatus(framework.Error, fmt.Sprintf("PersistentVolume was not found: %q", pvName))
+		if s := getErrorAsStatus(err); !s.IsSuccess() {
+			return s
 		}
 
 		for k, v := range pv.ObjectMeta.Labels {
@@ -171,6 +163,16 @@ func (pl *VolumeZone) Filter(ctx context.Context, _ *framework.CycleState, pod *
 	return nil
 }
 
+func getErrorAsStatus(err error) *framework.Status {
+	if err != nil {
+		if errors.IsNotFound(err) {
+			return framework.NewStatus(framework.UnschedulableAndUnresolvable, err.Error())
+		}
+		return framework.NewStatus(framework.Error, err.Error())
+	}
+	return nil
+}
+
 // New initializes a new plugin and returns it.
 func New(_ runtime.Object, handle framework.FrameworkHandle) (framework.Plugin, error) {
 	informerFactory := handle.SharedInformerFactory()

vendor/k8s.io/kubernetes/pkg/scheduler/internal/queue/BUILD

@@ -17,6 +17,7 @@ go_library(
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/wait:go_default_library",
+        "//staging/src/k8s.io/client-go/listers/core/v1:go_default_library",
         "//staging/src/k8s.io/client-go/tools/cache:go_default_library",
         "//vendor/k8s.io/klog/v2:go_default_library",
     ],
@@ -36,7 +37,11 @@ go_test(
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/clock:go_default_library",
+        "//staging/src/k8s.io/client-go/informers:go_default_library",
+        "//staging/src/k8s.io/client-go/kubernetes/fake:go_default_library",
         "//staging/src/k8s.io/component-base/metrics/testutil:go_default_library",
+        "//vendor/github.com/google/go-cmp/cmp:go_default_library",
+        "//vendor/github.com/google/go-cmp/cmp/cmpopts:go_default_library",
     ],
 )
 

vendor/k8s.io/kubernetes/pkg/scheduler/internal/queue/scheduling_queue.go

@@ -29,6 +29,7 @@ import (
 	"sync"
 	"time"
 
+	listersv1 "k8s.io/client-go/listers/core/v1"
 	"k8s.io/klog/v2"
 
 	v1 "k8s.io/api/core/v1"
@@ -721,6 +722,8 @@ func newUnschedulablePodsMap(metricRecorder metrics.MetricRecorder) *Unschedulab
 // may be different than what scheduler has here. We should be able to find pods
 // by their UID and update/delete them.
 type nominatedPodMap struct {
+	// podLister is used to verify if the given pod is alive.
+	podLister listersv1.PodLister
 	// nominatedPods is a map keyed by a node name and the value is a list of
 	// pods which are nominated to run on the node. These are pods which can be in
 	// the activeQ or unschedulableQ.
@@ -744,6 +747,14 @@ func (npm *nominatedPodMap) add(p *v1.Pod, nodeName string) {
 			return
 		}
 	}
+	if npm.podLister != nil {
+		// If the pod is not alive, don't contain it.
+		if _, err := npm.podLister.Pods(p.Namespace).Get(p.Name); err != nil {
+			klog.V(4).InfoS("Pod  %v/%v doesn't exist in podLister, aborting adding it to the nominated map", p.Namespace, p.Name)
+			return
+		}
+	}
+
 	npm.nominatedPodToNode[p.UID] = nnn
 	for _, np := range npm.nominatedPods[nnn] {
 		if np.UID == p.UID {
@@ -796,8 +807,17 @@ func (npm *nominatedPodMap) UpdateNominatedPod(oldPod, newPod *v1.Pod) {
 }
 
 // NewPodNominator creates a nominatedPodMap as a backing of framework.PodNominator.
+// DEPRECATED: use NewSafePodNominator() instead.
 func NewPodNominator() framework.PodNominator {
+	return NewSafePodNominator(nil)
+}
+
+// NewSafePodNominator creates a nominatedPodMap as a backing of framework.PodNominator.
+// Unlike NewPodNominator, it passes in a podLister so as to check if the pod is alive
+// before adding its nominatedNode info.
+func NewSafePodNominator(podLister listersv1.PodLister) framework.PodNominator {
 	return &nominatedPodMap{
+		podLister:          podLister,
 		nominatedPods:      make(map[string][]*v1.Pod),
 		nominatedPodToNode: make(map[ktypes.UID]string),
 	}

vendor/k8s.io/legacy-cloud-providers/aws/aws_loadbalancer.go

@@ -32,9 +32,9 @@ import (
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/aws/aws-sdk-go/service/elb"
 	"github.com/aws/aws-sdk-go/service/elbv2"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/klog/v2"
 
-	"k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
 )
@@ -60,6 +60,10 @@ var (
 	defaultHCUnhealthyThreshold = int64(6)
 	defaultHCTimeout            = int64(5)
 	defaultHCInterval           = int64(10)
+
+	// Defaults for ELB Target operations
+	defaultRegisterTargetsChunkSize   = 100
+	defaultDeregisterTargetsChunkSize = 100
 )
 
 func isNLB(annotations map[string]string) bool {
@@ -540,6 +544,7 @@ func (c *Cloud) deleteListenerV2(listener *elbv2.Listener) error {
 // ensureTargetGroup creates a target group with a set of instances.
 func (c *Cloud) ensureTargetGroup(targetGroup *elbv2.TargetGroup, serviceName types.NamespacedName, mapping nlbPortMapping, instances []string, vpcID string, tags map[string]string) (*elbv2.TargetGroup, error) {
 	dirty := false
+	expectedTargets := c.computeTargetGroupExpectedTargets(instances, mapping.TrafficPort)
 	if targetGroup == nil {
 		targetType := "instance"
 		name := c.buildTargetGroupName(serviceName, mapping.FrontendPort, mapping.TrafficPort, mapping.TrafficProtocol, targetType)
@@ -596,86 +601,23 @@ func (c *Cloud) ensureTargetGroup(targetGroup *elbv2.TargetGroup, serviceName ty
 			}
 		}
 
-		registerInput := &elbv2.RegisterTargetsInput{
-			TargetGroupArn: result.TargetGroups[0].TargetGroupArn,
-			Targets:        []*elbv2.TargetDescription{},
-		}
-		for _, instanceID := range instances {
-			registerInput.Targets = append(registerInput.Targets, &elbv2.TargetDescription{
-				Id:   aws.String(string(instanceID)),
-				Port: aws.Int64(mapping.TrafficPort),
-			})
-		}
-
-		_, err = c.elbv2.RegisterTargets(registerInput)
-		if err != nil {
-			return nil, fmt.Errorf("error registering targets for load balancer: %q", err)
+		tg := result.TargetGroups[0]
+		tgARN := aws.StringValue(tg.TargetGroupArn)
+		if err := c.ensureTargetGroupTargets(tgARN, expectedTargets, nil); err != nil {
+			return nil, err
 		}
-
-		return result.TargetGroups[0], nil
+		return tg, nil
 	}
 
 	// handle instances in service
 	{
-		healthResponse, err := c.elbv2.DescribeTargetHealth(&elbv2.DescribeTargetHealthInput{TargetGroupArn: targetGroup.TargetGroupArn})
+		tgARN := aws.StringValue(targetGroup.TargetGroupArn)
+		actualTargets, err := c.obtainTargetGroupActualTargets(tgARN)
 		if err != nil {
-			return nil, fmt.Errorf("error describing target group health: %q", err)
-		}
-		actualIDs := []string{}
-		for _, healthDescription := range healthResponse.TargetHealthDescriptions {
-			if aws.StringValue(healthDescription.TargetHealth.State) == elbv2.TargetHealthStateEnumHealthy {
-				actualIDs = append(actualIDs, *healthDescription.Target.Id)
-			} else if healthDescription.TargetHealth.Reason != nil {
-				switch aws.StringValue(healthDescription.TargetHealth.Reason) {
-				case elbv2.TargetHealthReasonEnumTargetDeregistrationInProgress:
-					// We don't need to count this instance in service if it is
-					// on its way out
-				default:
-					actualIDs = append(actualIDs, *healthDescription.Target.Id)
-				}
-			}
-		}
-
-		actual := sets.NewString(actualIDs...)
-		expected := sets.NewString(instances...)
-
-		additions := expected.Difference(actual)
-		removals := actual.Difference(expected)
-
-		if len(additions) > 0 {
-			registerInput := &elbv2.RegisterTargetsInput{
-				TargetGroupArn: targetGroup.TargetGroupArn,
-				Targets:        []*elbv2.TargetDescription{},
-			}
-			for instanceID := range additions {
-				registerInput.Targets = append(registerInput.Targets, &elbv2.TargetDescription{
-					Id:   aws.String(instanceID),
-					Port: aws.Int64(mapping.TrafficPort),
-				})
-			}
-			_, err := c.elbv2.RegisterTargets(registerInput)
-			if err != nil {
-				return nil, fmt.Errorf("error registering new targets in target group: %q", err)
-			}
-			dirty = true
+			return nil, err
 		}
-
-		if len(removals) > 0 {
-			deregisterInput := &elbv2.DeregisterTargetsInput{
-				TargetGroupArn: targetGroup.TargetGroupArn,
-				Targets:        []*elbv2.TargetDescription{},
-			}
-			for instanceID := range removals {
-				deregisterInput.Targets = append(deregisterInput.Targets, &elbv2.TargetDescription{
-					Id:   aws.String(instanceID),
-					Port: aws.Int64(mapping.TrafficPort),
-				})
-			}
-			_, err := c.elbv2.DeregisterTargets(deregisterInput)
-			if err != nil {
-				return nil, fmt.Errorf("error trying to deregister targets in target group: %q", err)
-			}
-			dirty = true
+		if err := c.ensureTargetGroupTargets(tgARN, expectedTargets, actualTargets); err != nil {
+			return nil, err
 		}
 	}
 
@@ -723,6 +665,101 @@ func (c *Cloud) ensureTargetGroup(targetGroup *elbv2.TargetGroup, serviceName ty
 	return targetGroup, nil
 }
 
+func (c *Cloud) ensureTargetGroupTargets(tgARN string, expectedTargets []*elbv2.TargetDescription, actualTargets []*elbv2.TargetDescription) error {
+	targetsToRegister, targetsToDeregister := c.diffTargetGroupTargets(expectedTargets, actualTargets)
+	if len(targetsToRegister) > 0 {
+		targetsToRegisterChunks := c.chunkTargetDescriptions(targetsToRegister, defaultRegisterTargetsChunkSize)
+		for _, targetsChunk := range targetsToRegisterChunks {
+			req := &elbv2.RegisterTargetsInput{
+				TargetGroupArn: aws.String(tgARN),
+				Targets:        targetsChunk,
+			}
+			if _, err := c.elbv2.RegisterTargets(req); err != nil {
+				return fmt.Errorf("error trying to register targets in target group: %q", err)
+			}
+		}
+	}
+	if len(targetsToDeregister) > 0 {
+		targetsToDeregisterChunks := c.chunkTargetDescriptions(targetsToDeregister, defaultDeregisterTargetsChunkSize)
+		for _, targetsChunk := range targetsToDeregisterChunks {
+			req := &elbv2.DeregisterTargetsInput{
+				TargetGroupArn: aws.String(tgARN),
+				Targets:        targetsChunk,
+			}
+			if _, err := c.elbv2.DeregisterTargets(req); err != nil {
+				return fmt.Errorf("error trying to deregister targets in target group: %q", err)
+			}
+		}
+	}
+	return nil
+}
+
+func (c *Cloud) computeTargetGroupExpectedTargets(instanceIDs []string, port int64) []*elbv2.TargetDescription {
+	expectedTargets := make([]*elbv2.TargetDescription, 0, len(instanceIDs))
+	for _, instanceID := range instanceIDs {
+		expectedTargets = append(expectedTargets, &elbv2.TargetDescription{
+			Id:   aws.String(instanceID),
+			Port: aws.Int64(port),
+		})
+	}
+	return expectedTargets
+}
+
+func (c *Cloud) obtainTargetGroupActualTargets(tgARN string) ([]*elbv2.TargetDescription, error) {
+	req := &elbv2.DescribeTargetHealthInput{
+		TargetGroupArn: aws.String(tgARN),
+	}
+	resp, err := c.elbv2.DescribeTargetHealth(req)
+	if err != nil {
+		return nil, fmt.Errorf("error describing target group health: %q", err)
+	}
+	actualTargets := make([]*elbv2.TargetDescription, 0, len(resp.TargetHealthDescriptions))
+	for _, targetDesc := range resp.TargetHealthDescriptions {
+		if targetDesc.TargetHealth.Reason != nil && aws.StringValue(targetDesc.TargetHealth.Reason) == elbv2.TargetHealthReasonEnumTargetDeregistrationInProgress {
+			continue
+		}
+		actualTargets = append(actualTargets, targetDesc.Target)
+	}
+	return actualTargets, nil
+}
+
+// diffTargetGroupTargets computes the targets to register and targets to deregister based on existingTargets and desired instances.
+func (c *Cloud) diffTargetGroupTargets(expectedTargets []*elbv2.TargetDescription, actualTargets []*elbv2.TargetDescription) (targetsToRegister []*elbv2.TargetDescription, targetsToDeregister []*elbv2.TargetDescription) {
+	expectedTargetsByUID := make(map[string]*elbv2.TargetDescription, len(expectedTargets))
+	for _, target := range expectedTargets {
+		targetUID := fmt.Sprintf("%v:%v", aws.StringValue(target.Id), aws.Int64Value(target.Port))
+		expectedTargetsByUID[targetUID] = target
+	}
+	actualTargetsByUID := make(map[string]*elbv2.TargetDescription, len(actualTargets))
+	for _, target := range actualTargets {
+		targetUID := fmt.Sprintf("%v:%v", aws.StringValue(target.Id), aws.Int64Value(target.Port))
+		actualTargetsByUID[targetUID] = target
+	}
+
+	expectedTargetsUIDs := sets.StringKeySet(expectedTargetsByUID)
+	actualTargetsUIDs := sets.StringKeySet(actualTargetsByUID)
+	for _, targetUID := range expectedTargetsUIDs.Difference(actualTargetsUIDs).List() {
+		targetsToRegister = append(targetsToRegister, expectedTargetsByUID[targetUID])
+	}
+	for _, targetUID := range actualTargetsUIDs.Difference(expectedTargetsUIDs).List() {
+		targetsToDeregister = append(targetsToDeregister, actualTargetsByUID[targetUID])
+	}
+	return targetsToRegister, targetsToDeregister
+}
+
+// chunkTargetDescriptions will split slice of TargetDescription into chunks
+func (c *Cloud) chunkTargetDescriptions(targets []*elbv2.TargetDescription, chunkSize int) [][]*elbv2.TargetDescription {
+	var chunks [][]*elbv2.TargetDescription
+	for i := 0; i < len(targets); i += chunkSize {
+		end := i + chunkSize
+		if end > len(targets) {
+			end = len(targets)
+		}
+		chunks = append(chunks, targets[i:end])
+	}
+	return chunks
+}
+
 // updateInstanceSecurityGroupsForNLB will adjust securityGroup's settings to allow inbound traffic into instances from clientCIDRs and portMappings.
 // TIP: if either instances or clientCIDRs or portMappings are nil, then the securityGroup rules for lbName are cleared.
 func (c *Cloud) updateInstanceSecurityGroupsForNLB(lbName string, instances map[InstanceID]*ec2.Instance, subnetCIDRs []string, clientCIDRs []string, portMappings []nlbPortMapping) error {

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_vmss.go

@@ -1162,7 +1162,6 @@ func (ss *scaleSet) ensureVMSSInPool(service *v1.Service, nodes []*v1.Node, back
 			})
 		primaryIPConfig.LoadBalancerBackendAddressPools = &loadBalancerBackendAddressPools
 		newVMSS := compute.VirtualMachineScaleSet{
-			Sku:      vmss.Sku,
 			Location: vmss.Location,
 			VirtualMachineScaleSetProperties: &compute.VirtualMachineScaleSetProperties{
 				VirtualMachineProfile: &compute.VirtualMachineScaleSetVMProfile{
@@ -1468,7 +1467,6 @@ func (ss *scaleSet) ensureBackendPoolDeletedFromVMSS(service *v1.Service, backen
 			// Compose a new vmss with added backendPoolID.
 			primaryIPConfig.LoadBalancerBackendAddressPools = &newBackendPools
 			newVMSS := compute.VirtualMachineScaleSet{
-				Sku:      vmss.Sku,
 				Location: vmss.Location,
 				VirtualMachineScaleSetProperties: &compute.VirtualMachineScaleSetProperties{
 					VirtualMachineProfile: &compute.VirtualMachineScaleSetVMProfile{

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

@@ -193,6 +193,11 @@ func (ss *scaleSet) newVMSSVirtualMachinesCache(resourceGroupName, vmssName, cac
 			}
 
 			computerName := strings.ToLower(*vm.OsProfile.ComputerName)
+			if vm.NetworkProfile == nil || vm.NetworkProfile.NetworkInterfaces == nil {
+				klog.Warningf("skip caching vmssVM %s since its network profile hasn't initialized yet (probably still under creating)", computerName)
+				continue
+			}
+
 			vmssVMCacheEntry := &vmssVirtualMachinesEntry{
 				resourceGroup:  resourceGroupName,
 				vmssName:       vmssName,

vendor/k8s.io/legacy-cloud-providers/vsphere/BUILD

@@ -12,6 +12,7 @@ go_library(
         "credentialmanager.go",
         "doc.go",
         "nodemanager.go",
+        "shared_datastore.go",
         "vsphere.go",
         "vsphere_util.go",
         "vsphere_util_linux.go",

vendor/k8s.io/legacy-cloud-providers/vsphere/shared_datastore.go

+// +build !providerless
+
+/*
+Copyright 2021 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 vsphere
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/vmware/govmomi/property"
+	"github.com/vmware/govmomi/vim25/mo"
+	"github.com/vmware/govmomi/vim25/types"
+	"k8s.io/klog/v2"
+	"k8s.io/legacy-cloud-providers/vsphere/vclib"
+)
+
+type sharedDatastore struct {
+	nodeManager         *NodeManager
+	candidateDatastores []*vclib.DatastoreInfo
+}
+
+type hostInfo struct {
+	hostUUID   string
+	hostMOID   string
+	datacenter string
+}
+
+const (
+	summary       = "summary"
+	runtimeHost   = "summary.runtime.host"
+	hostsProperty = "host"
+	nameProperty  = "name"
+)
+
+func (shared *sharedDatastore) getSharedDatastore(ctcx context.Context) (*vclib.DatastoreInfo, error) {
+	nodes := shared.nodeManager.getNodes()
+
+	// Segregate nodes according to VC-DC
+	dcNodes := make(map[string][]NodeInfo)
+	nodeHosts := make(map[string]hostInfo)
+
+	for nodeName, node := range nodes {
+		nodeInfo, err := shared.nodeManager.GetNodeInfoWithNodeObject(node)
+		if err != nil {
+			return nil, fmt.Errorf("unable to find node %s: %v", nodeName, err)
+		}
+		vcDC := nodeInfo.vcServer + nodeInfo.dataCenter.String()
+		dcNodes[vcDC] = append(dcNodes[vcDC], nodeInfo)
+	}
+
+	for vcDC, nodes := range dcNodes {
+		var hostInfos []hostInfo
+		var err error
+		hostInfos, err = shared.getNodeHosts(ctcx, nodes, vcDC)
+		if err != nil {
+			if vclib.IsManagedObjectNotFoundError(err) {
+				klog.Warningf("SharedHost.getSharedDatastore: batch fetching of hosts failed - switching to fetching them individually.")
+				hostInfos, err = shared.getEachNodeHost(ctcx, nodes, vcDC)
+				if err != nil {
+					klog.Errorf("SharedHost.getSharedDatastore: error fetching node hosts individually: %v", err)
+					return nil, err
+				}
+			} else {
+				return nil, err
+			}
+		}
+		for _, host := range hostInfos {
+			hostDCName := fmt.Sprintf("%s/%s", host.datacenter, host.hostMOID)
+			nodeHosts[hostDCName] = host
+		}
+	}
+
+	if len(nodeHosts) < 1 {
+		msg := fmt.Sprintf("SharedHost.getSharedDatastore unable to find hosts associated with nodes")
+		klog.Error(msg)
+		return nil, fmt.Errorf("")
+	}
+
+	for _, datastoreInfo := range shared.candidateDatastores {
+		dataStoreHosts, err := shared.getAttachedHosts(ctcx, datastoreInfo.Datastore)
+		if err != nil {
+			msg := fmt.Sprintf("error finding attached hosts to datastore %s: %v", datastoreInfo.Name(), err)
+			klog.Error(msg)
+			return nil, fmt.Errorf(msg)
+		}
+		if shared.isIncluded(dataStoreHosts, nodeHosts) {
+			return datastoreInfo, nil
+		}
+	}
+	return nil, fmt.Errorf("SharedHost.getSharedDatastore: unable to find any shared datastores")
+}
+
+// check if all of the nodeHosts are included in the dataStoreHosts
+func (shared *sharedDatastore) isIncluded(dataStoreHosts []hostInfo, nodeHosts map[string]hostInfo) bool {
+	result := true
+	for _, host := range nodeHosts {
+		hostFound := false
+		for _, targetHost := range dataStoreHosts {
+			if host.hostUUID == targetHost.hostUUID && host.hostMOID == targetHost.hostMOID {
+				hostFound = true
+			}
+		}
+		if !hostFound {
+			result = false
+		}
+	}
+	return result
+}
+
+func (shared *sharedDatastore) getEachNodeHost(ctx context.Context, nodes []NodeInfo, dcVC string) ([]hostInfo, error) {
+	var hosts []hostInfo
+	for _, node := range nodes {
+		host, err := node.vm.GetHost(ctx)
+		if err != nil {
+			klog.Errorf("SharedHost.getEachNodeHost: unable to find host for vm %s: %v", node.vm.InventoryPath, err)
+			return nil, err
+		}
+		hosts = append(hosts, hostInfo{
+			hostUUID:   host.Summary.Hardware.Uuid,
+			hostMOID:   host.Summary.Host.String(),
+			datacenter: node.dataCenter.String(),
+		})
+	}
+	return hosts, nil
+}
+
+func (shared *sharedDatastore) getNodeHosts(ctx context.Context, nodes []NodeInfo, dcVC string) ([]hostInfo, error) {
+	var vmRefs []types.ManagedObjectReference
+	if len(nodes) < 1 {
+		return nil, fmt.Errorf("no nodes found for dc-vc: %s", dcVC)
+	}
+	var nodeInfo NodeInfo
+	for _, n := range nodes {
+		nodeInfo = n
+		vmRefs = append(vmRefs, n.vm.Reference())
+	}
+	pc := property.DefaultCollector(nodeInfo.dataCenter.Client())
+	var vmoList []mo.VirtualMachine
+	err := pc.Retrieve(ctx, vmRefs, []string{nameProperty, runtimeHost}, &vmoList)
+	if err != nil {
+		klog.Errorf("SharedHost.getNodeHosts: unable to fetch vms from datacenter %s: %w", nodeInfo.dataCenter.String(), err)
+		return nil, err
+	}
+	var hostMoList []mo.HostSystem
+	var hostRefs []types.ManagedObjectReference
+	for _, vmo := range vmoList {
+		if vmo.Summary.Runtime.Host == nil {
+			msg := fmt.Sprintf("SharedHost.getNodeHosts: no host associated with vm %s", vmo.Name)
+			klog.Error(msg)
+			return nil, fmt.Errorf(msg)
+		}
+		hostRefs = append(hostRefs, vmo.Summary.Runtime.Host.Reference())
+	}
+	pc = property.DefaultCollector(nodeInfo.dataCenter.Client())
+	err = pc.Retrieve(ctx, hostRefs, []string{summary}, &hostMoList)
+	if err != nil {
+		klog.Errorf("SharedHost.getNodeHosts: unable to fetch hosts from datacenter %s: %w", nodeInfo.dataCenter.String(), err)
+		return nil, err
+	}
+	var hosts []hostInfo
+	for _, host := range hostMoList {
+		hosts = append(hosts, hostInfo{hostMOID: host.Summary.Host.String(), hostUUID: host.Summary.Hardware.Uuid, datacenter: nodeInfo.dataCenter.String()})
+	}
+	return hosts, nil
+}
+
+func (shared *sharedDatastore) getAttachedHosts(ctx context.Context, datastore *vclib.Datastore) ([]hostInfo, error) {
+	var ds mo.Datastore
+
+	pc := property.DefaultCollector(datastore.Client())
+	err := pc.RetrieveOne(ctx, datastore.Reference(), []string{hostsProperty}, &ds)
+	if err != nil {
+		return nil, err
+	}
+
+	mounts := make(map[types.ManagedObjectReference]types.DatastoreHostMount)
+	var refs []types.ManagedObjectReference
+	for _, host := range ds.Host {
+		refs = append(refs, host.Key)
+		mounts[host.Key] = host
+	}
+
+	var hs []mo.HostSystem
+	err = pc.Retrieve(ctx, refs, []string{summary}, &hs)
+	if err != nil {
+		return nil, err
+	}
+	var hosts []hostInfo
+	for _, h := range hs {
+		hosts = append(hosts, hostInfo{hostUUID: h.Summary.Hardware.Uuid, hostMOID: h.Summary.Host.String()})
+	}
+	return hosts, nil
+
+}

vendor/k8s.io/legacy-cloud-providers/vsphere/vclib/virtualmachine.go

@@ -166,6 +166,24 @@ func (vm *VirtualMachine) AttachDisk(ctx context.Context, vmDiskPath string, vol
 	return diskUUID, nil
 }
 
+// GetHost returns host of the virtual machine
+func (vm *VirtualMachine) GetHost(ctx context.Context) (mo.HostSystem, error) {
+	host, err := vm.HostSystem(ctx)
+	var hostSystemMo mo.HostSystem
+	if err != nil {
+		klog.Errorf("Failed to get host system for VM: %q. err: %+v", vm.InventoryPath, err)
+		return hostSystemMo, err
+	}
+
+	s := object.NewSearchIndex(vm.Client())
+	err = s.Properties(ctx, host.Reference(), []string{"summary"}, &hostSystemMo)
+	if err != nil {
+		klog.Errorf("Failed to retrieve datastores for host: %+v. err: %+v", host, err)
+		return hostSystemMo, err
+	}
+	return hostSystemMo, nil
+}
+
 // DetachDisk detaches the disk specified by vmDiskPath
 func (vm *VirtualMachine) DetachDisk(ctx context.Context, vmDiskPath string) error {
 	vmDiskPath = RemoveStorageClusterORFolderNameFromVDiskPath(vmDiskPath)

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

@@ -1340,13 +1340,20 @@ func (vs *VSphere) CreateVolume(volumeOptions *vclib.VolumeOptions) (canonicalVo
 				if len(zonesToSearch) == 0 {
 					// If zone is not provided, get the shared datastore across all node VMs.
 					klog.V(4).Infof("Validating if datastore %s is shared across all node VMs", datastoreName)
-					sharedDsList, err = getSharedDatastoresInK8SCluster(ctx, vs.nodeManager)
+					sharedDSFinder := &sharedDatastore{
+						nodeManager:         vs.nodeManager,
+						candidateDatastores: candidateDatastoreInfos,
+					}
+					datastoreInfo, err = sharedDSFinder.getSharedDatastore(ctx)
 					if err != nil {
 						klog.Errorf("Failed to get shared datastore: %+v", err)
 						return "", err
 					}
-					// Prepare error msg to be used later, if required.
-					err = fmt.Errorf("The specified datastore %s is not a shared datastore across node VMs", datastoreName)
+					if datastoreInfo == nil {
+						err = fmt.Errorf("The specified datastore %s is not a shared datastore across node VMs", datastoreName)
+						klog.Error(err)
+						return "", err
+					}
 				} else {
 					// If zone is provided, get the shared datastores in that zone.
 					klog.V(4).Infof("Validating if datastore %s is in zone %s ", datastoreName, zonesToSearch)
@@ -1355,21 +1362,19 @@ func (vs *VSphere) CreateVolume(volumeOptions *vclib.VolumeOptions) (canonicalVo
 						klog.Errorf("Failed to find a shared datastore matching zone %s. err: %+v", zonesToSearch, err)
 						return "", err
 					}
-					// Prepare error msg to be used later, if required.
-					err = fmt.Errorf("The specified datastore %s does not match the provided zones : %s", datastoreName, zonesToSearch)
-				}
-				found := false
-				// Check if the selected datastore belongs to the list of shared datastores computed.
-				for _, sharedDs := range sharedDsList {
-					if datastoreInfo, found = candidateDatastores[sharedDs.Info.Url]; found {
-						klog.V(4).Infof("Datastore validation succeeded")
-						found = true
-						break
+					found := false
+					for _, sharedDs := range sharedDsList {
+						if datastoreInfo, found = candidateDatastores[sharedDs.Info.Url]; found {
+							klog.V(4).Infof("Datastore validation succeeded")
+							found = true
+							break
+						}
+					}
+					if !found {
+						err = fmt.Errorf("The specified datastore %s does not match the provided zones : %s", datastoreName, zonesToSearch)
+						klog.Error(err)
+						return "", err
 					}
-				}
-				if !found {
-					klog.Error(err)
-					return "", err
 				}
 			}
 		}