Update to v1.22.3 (#4354)

go.mod

@@ -39,37 +39,37 @@ replace (
 	google.golang.org/genproto => google.golang.org/genproto v0.0.0-20200224152610-e50cd9704f63
 	google.golang.org/grpc => google.golang.org/grpc v1.38.0
 	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.22.2-k3s1
-	k8s.io/apiextensions-apiserver => github.com/k3s-io/kubernetes/staging/src/k8s.io/apiextensions-apiserver v1.22.2-k3s1
-	k8s.io/apimachinery => github.com/k3s-io/kubernetes/staging/src/k8s.io/apimachinery v1.22.2-k3s1
-	k8s.io/apiserver => github.com/k3s-io/kubernetes/staging/src/k8s.io/apiserver v1.22.2-k3s1
-	k8s.io/cli-runtime => github.com/k3s-io/kubernetes/staging/src/k8s.io/cli-runtime v1.22.2-k3s1
-	k8s.io/client-go => github.com/k3s-io/kubernetes/staging/src/k8s.io/client-go v1.22.2-k3s1
-	k8s.io/cloud-provider => github.com/k3s-io/kubernetes/staging/src/k8s.io/cloud-provider v1.22.2-k3s1
-	k8s.io/cluster-bootstrap => github.com/k3s-io/kubernetes/staging/src/k8s.io/cluster-bootstrap v1.22.2-k3s1
-	k8s.io/code-generator => github.com/k3s-io/kubernetes/staging/src/k8s.io/code-generator v1.22.2-k3s1
-	k8s.io/component-base => github.com/k3s-io/kubernetes/staging/src/k8s.io/component-base v1.22.2-k3s1
-	k8s.io/component-helpers => github.com/k3s-io/kubernetes/staging/src/k8s.io/component-helpers v1.22.2-k3s1
-	k8s.io/controller-manager => github.com/k3s-io/kubernetes/staging/src/k8s.io/controller-manager v1.22.2-k3s1
-	k8s.io/cri-api => github.com/k3s-io/kubernetes/staging/src/k8s.io/cri-api v1.22.2-k3s1
-	k8s.io/csi-translation-lib => github.com/k3s-io/kubernetes/staging/src/k8s.io/csi-translation-lib v1.22.2-k3s1
+	k8s.io/api => github.com/k3s-io/kubernetes/staging/src/k8s.io/api v1.22.3-k3s1
+	k8s.io/apiextensions-apiserver => github.com/k3s-io/kubernetes/staging/src/k8s.io/apiextensions-apiserver v1.22.3-k3s1
+	k8s.io/apimachinery => github.com/k3s-io/kubernetes/staging/src/k8s.io/apimachinery v1.22.3-k3s1
+	k8s.io/apiserver => github.com/k3s-io/kubernetes/staging/src/k8s.io/apiserver v1.22.3-k3s1
+	k8s.io/cli-runtime => github.com/k3s-io/kubernetes/staging/src/k8s.io/cli-runtime v1.22.3-k3s1
+	k8s.io/client-go => github.com/k3s-io/kubernetes/staging/src/k8s.io/client-go v1.22.3-k3s1
+	k8s.io/cloud-provider => github.com/k3s-io/kubernetes/staging/src/k8s.io/cloud-provider v1.22.3-k3s1
+	k8s.io/cluster-bootstrap => github.com/k3s-io/kubernetes/staging/src/k8s.io/cluster-bootstrap v1.22.3-k3s1
+	k8s.io/code-generator => github.com/k3s-io/kubernetes/staging/src/k8s.io/code-generator v1.22.3-k3s1
+	k8s.io/component-base => github.com/k3s-io/kubernetes/staging/src/k8s.io/component-base v1.22.3-k3s1
+	k8s.io/component-helpers => github.com/k3s-io/kubernetes/staging/src/k8s.io/component-helpers v1.22.3-k3s1
+	k8s.io/controller-manager => github.com/k3s-io/kubernetes/staging/src/k8s.io/controller-manager v1.22.3-k3s1
+	k8s.io/cri-api => github.com/k3s-io/kubernetes/staging/src/k8s.io/cri-api v1.22.3-k3s1
+	k8s.io/csi-translation-lib => github.com/k3s-io/kubernetes/staging/src/k8s.io/csi-translation-lib v1.22.3-k3s1
 	k8s.io/klog => github.com/k3s-io/klog v1.0.0-k3s2 // k3s-release-1.x
 	k8s.io/klog/v2 => github.com/k3s-io/klog/v2 v2.9.0-k3s2 // k3s-main
-	k8s.io/kube-aggregator => github.com/k3s-io/kubernetes/staging/src/k8s.io/kube-aggregator v1.22.2-k3s1
-	k8s.io/kube-controller-manager => github.com/k3s-io/kubernetes/staging/src/k8s.io/kube-controller-manager v1.22.2-k3s1
-	k8s.io/kube-proxy => github.com/k3s-io/kubernetes/staging/src/k8s.io/kube-proxy v1.22.2-k3s1
-	k8s.io/kube-scheduler => github.com/k3s-io/kubernetes/staging/src/k8s.io/kube-scheduler v1.22.2-k3s1
-	k8s.io/kubectl => github.com/k3s-io/kubernetes/staging/src/k8s.io/kubectl v1.22.2-k3s1
-	k8s.io/kubelet => github.com/k3s-io/kubernetes/staging/src/k8s.io/kubelet v1.22.2-k3s1
-	k8s.io/kubernetes => github.com/k3s-io/kubernetes v1.22.2-k3s1
-	k8s.io/legacy-cloud-providers => github.com/k3s-io/kubernetes/staging/src/k8s.io/legacy-cloud-providers v1.22.2-k3s1
-	k8s.io/metrics => github.com/k3s-io/kubernetes/staging/src/k8s.io/metrics v1.22.2-k3s1
-	k8s.io/mount-utils => github.com/k3s-io/kubernetes/staging/src/k8s.io/mount-utils v1.22.2-k3s1
-	k8s.io/node-api => github.com/k3s-io/kubernetes/staging/src/k8s.io/node-api v1.22.2-k3s1
-	k8s.io/pod-security-admission => github.com/k3s-io/kubernetes/staging/src/k8s.io/pod-security-admission v1.22.2-k3s1
-	k8s.io/sample-apiserver => github.com/k3s-io/kubernetes/staging/src/k8s.io/sample-apiserver v1.22.2-k3s1
-	k8s.io/sample-cli-plugin => github.com/k3s-io/kubernetes/staging/src/k8s.io/sample-cli-plugin v1.22.2-k3s1
-	k8s.io/sample-controller => github.com/k3s-io/kubernetes/staging/src/k8s.io/sample-controller v1.22.2-k3s1
+	k8s.io/kube-aggregator => github.com/k3s-io/kubernetes/staging/src/k8s.io/kube-aggregator v1.22.3-k3s1
+	k8s.io/kube-controller-manager => github.com/k3s-io/kubernetes/staging/src/k8s.io/kube-controller-manager v1.22.3-k3s1
+	k8s.io/kube-proxy => github.com/k3s-io/kubernetes/staging/src/k8s.io/kube-proxy v1.22.3-k3s1
+	k8s.io/kube-scheduler => github.com/k3s-io/kubernetes/staging/src/k8s.io/kube-scheduler v1.22.3-k3s1
+	k8s.io/kubectl => github.com/k3s-io/kubernetes/staging/src/k8s.io/kubectl v1.22.3-k3s1
+	k8s.io/kubelet => github.com/k3s-io/kubernetes/staging/src/k8s.io/kubelet v1.22.3-k3s1
+	k8s.io/kubernetes => github.com/k3s-io/kubernetes v1.22.3-k3s1
+	k8s.io/legacy-cloud-providers => github.com/k3s-io/kubernetes/staging/src/k8s.io/legacy-cloud-providers v1.22.3-k3s1
+	k8s.io/metrics => github.com/k3s-io/kubernetes/staging/src/k8s.io/metrics v1.22.3-k3s1
+	k8s.io/mount-utils => github.com/k3s-io/kubernetes/staging/src/k8s.io/mount-utils v1.22.3-k3s1
+	k8s.io/node-api => github.com/k3s-io/kubernetes/staging/src/k8s.io/node-api v1.22.3-k3s1
+	k8s.io/pod-security-admission => github.com/k3s-io/kubernetes/staging/src/k8s.io/pod-security-admission v1.22.3-k3s1
+	k8s.io/sample-apiserver => github.com/k3s-io/kubernetes/staging/src/k8s.io/sample-apiserver v1.22.3-k3s1
+	k8s.io/sample-cli-plugin => github.com/k3s-io/kubernetes/staging/src/k8s.io/sample-cli-plugin v1.22.3-k3s1
+	k8s.io/sample-controller => github.com/k3s-io/kubernetes/staging/src/k8s.io/sample-controller v1.22.3-k3s1
 	mvdan.cc/unparam => mvdan.cc/unparam v0.0.0-20210104141923-aac4ce9116a7
 )
 
@@ -128,18 +128,18 @@ require (
 	google.golang.org/grpc v1.40.0
 	gopkg.in/yaml.v2 v2.4.0
 	inet.af/tcpproxy v0.0.0-20200125044825-b6bb9b5b8252
-	k8s.io/api v0.22.2
-	k8s.io/apimachinery v0.22.2
-	k8s.io/apiserver v0.22.2
+	k8s.io/api v0.22.3
+	k8s.io/apimachinery v0.22.3
+	k8s.io/apiserver v0.22.3
 	k8s.io/client-go v11.0.1-0.20190409021438-1a26190bd76a+incompatible
-	k8s.io/cloud-provider v0.22.2
-	k8s.io/component-base v0.22.2
-	k8s.io/controller-manager v0.22.2 // indirect
-	k8s.io/cri-api v0.22.2
+	k8s.io/cloud-provider v0.22.3
+	k8s.io/component-base v0.22.3
+	k8s.io/controller-manager v0.22.3 // indirect
+	k8s.io/cri-api v0.22.3
 	k8s.io/klog v1.0.0
 	k8s.io/klog/v2 v2.9.0
-	k8s.io/kubectl v0.22.2
-	k8s.io/kubernetes v1.22.2
+	k8s.io/kubectl v0.22.3
+	k8s.io/kubernetes v1.22.3
 	k8s.io/utils v0.0.0-20210819203725-bdf08cb9a70a
 	sigs.k8s.io/yaml v1.2.0
 )

vendor/k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1/conversion.go

@@ -17,6 +17,8 @@ limitations under the License.
 package v1
 
 import (
+	"bytes"
+
 	"k8s.io/apiextensions-apiserver/pkg/apis/apiextensions"
 	apiequality "k8s.io/apimachinery/pkg/api/equality"
 	"k8s.io/apimachinery/pkg/conversion"
@@ -36,21 +38,30 @@ func Convert_apiextensions_JSONSchemaProps_To_v1_JSONSchemaProps(in *apiextensio
 	return nil
 }
 
+var nullLiteral = []byte(`null`)
+
 func Convert_apiextensions_JSON_To_v1_JSON(in *apiextensions.JSON, out *JSON, s conversion.Scope) error {
 	raw, err := json.Marshal(*in)
 	if err != nil {
 		return err
 	}
+	if len(raw) == 0 || bytes.Equal(raw, nullLiteral) {
+		// match JSON#UnmarshalJSON treatment of literal nulls
+		out.Raw = nil
+	} else {
 		out.Raw = raw
+	}
 	return nil
 }
 
 func Convert_v1_JSON_To_apiextensions_JSON(in *JSON, out *apiextensions.JSON, s conversion.Scope) error {
 	if in != nil {
 		var i interface{}
+		if len(in.Raw) > 0 && !bytes.Equal(in.Raw, nullLiteral) {
 			if err := json.Unmarshal(in.Raw, &i); err != nil {
 				return err
 			}
+		}
 		*out = i
 	} else {
 		out = nil
@@ -103,7 +114,7 @@ func Convert_apiextensions_CustomResourceDefinitionSpec_To_v1_CustomResourceDefi
 
 func Convert_v1_CustomResourceDefinitionSpec_To_apiextensions_CustomResourceDefinitionSpec(in *CustomResourceDefinitionSpec, out *apiextensions.CustomResourceDefinitionSpec, s conversion.Scope) error {
 	if err := autoConvert_v1_CustomResourceDefinitionSpec_To_apiextensions_CustomResourceDefinitionSpec(in, out, s); err != nil {
-		return nil
+		return err
 	}
 
 	if len(out.Versions) == 0 {

vendor/k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1/marshal.go

@@ -17,6 +17,7 @@ limitations under the License.
 package v1
 
 import (
+	"bytes"
 	"errors"
 
 	"k8s.io/apimachinery/pkg/util/json"
@@ -128,7 +129,7 @@ func (s JSON) MarshalJSON() ([]byte, error) {
 }
 
 func (s *JSON) UnmarshalJSON(data []byte) error {
-	if len(data) > 0 && string(data) != "null" {
+	if len(data) > 0 && !bytes.Equal(data, nullLiteral) {
 		s.Raw = data
 	}
 	return nil

vendor/k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1/conversion.go

@@ -17,6 +17,8 @@ limitations under the License.
 package v1beta1
 
 import (
+	"bytes"
+
 	"k8s.io/apimachinery/pkg/conversion"
 	"k8s.io/apimachinery/pkg/util/json"
 
@@ -36,21 +38,30 @@ func Convert_apiextensions_JSONSchemaProps_To_v1beta1_JSONSchemaProps(in *apiext
 	return nil
 }
 
+var nullLiteral = []byte(`null`)
+
 func Convert_apiextensions_JSON_To_v1beta1_JSON(in *apiextensions.JSON, out *JSON, s conversion.Scope) error {
 	raw, err := json.Marshal(*in)
 	if err != nil {
 		return err
 	}
+	if len(raw) == 0 || bytes.Equal(raw, nullLiteral) {
+		// match JSON#UnmarshalJSON treatment of literal nulls
+		out.Raw = nil
+	} else {
 		out.Raw = raw
+	}
 	return nil
 }
 
 func Convert_v1beta1_JSON_To_apiextensions_JSON(in *JSON, out *apiextensions.JSON, s conversion.Scope) error {
 	if in != nil {
 		var i interface{}
+		if len(in.Raw) > 0 && !bytes.Equal(in.Raw, nullLiteral) {
 			if err := json.Unmarshal(in.Raw, &i); err != nil {
 				return err
 			}
+		}
 		*out = i
 	} else {
 		out = nil

vendor/k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1/marshal.go

@@ -17,6 +17,7 @@ limitations under the License.
 package v1beta1
 
 import (
+	"bytes"
 	"errors"
 
 	"k8s.io/apimachinery/pkg/util/json"
@@ -128,7 +129,7 @@ func (s JSON) MarshalJSON() ([]byte, error) {
 }
 
 func (s *JSON) UnmarshalJSON(data []byte) error {
-	if len(data) > 0 && string(data) != "null" {
+	if len(data) > 0 && !bytes.Equal(data, nullLiteral) {
 		s.Raw = data
 	}
 	return nil

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

@@ -362,6 +362,16 @@ func NewEgressSelector(config *apiserver.EgressSelectorConfiguration) (*EgressSe
 	return cs, nil
 }
 
+// NewEgressSelectorWithMap returns a EgressSelector with the supplied EgressType to DialFunc map.
+func NewEgressSelectorWithMap(m map[EgressType]utilnet.DialFunc) *EgressSelector {
+	if m == nil {
+		m = make(map[EgressType]utilnet.DialFunc)
+	}
+	return &EgressSelector{
+		egressToDialer: m,
+	}
+}
+
 // Lookup gets the dialer function for the network context.
 // This is configured for the Kubernetes API Server at startup.
 func (cs *EgressSelector) Lookup(networkContext NetworkContext) (utilnet.DialFunc, error) {

vendor/k8s.io/apiserver/pkg/util/flowcontrol/apf_controller.go

@@ -67,6 +67,14 @@ const timeFmt = "2006-01-02T15:04:05.999"
 // undesired becomes completely unused, all the config objects are
 // read and processed as a whole.
 
+// The funcs in this package follow the naming convention that the suffix
+// "Locked" means the relevant mutex must be locked at the start of each
+// call and will be locked upon return.  For a configController, the
+// suffix "ReadLocked" stipulates a read lock while just "Locked"
+// stipulates a full lock.  Absence of either suffix means that either
+// (a) the lock must NOT be held at call time and will not be held
+// upon return or (b) locking is irrelevant.
+
 // StartFunction begins the process of handling a request.  If the
 // request gets queued then this function uses the given hashValue as
 // the source of entropy as it shuffle-shards the request into a
@@ -124,10 +132,25 @@ type configController struct {
 	// requestWaitLimit comes from server configuration.
 	requestWaitLimit time.Duration
 
+	// watchTracker implements the necessary WatchTracker interface.
+	WatchTracker
+
+	// the most recent update attempts, ordered by increasing age.
+	// Consumer trims to keep only the last minute's worth of entries.
+	// The controller uses this to limit itself to at most six updates
+	// to a given FlowSchema in any minute.
+	// This may only be accessed from the one and only worker goroutine.
+	mostRecentUpdates []updateAttempt
+
 	// This must be locked while accessing flowSchemas or
-	// priorityLevelStates.  It is the lock involved in
-	// LockingWriteMultiple.
-	lock sync.Mutex
+	// priorityLevelStates.  A lock for writing is needed
+	// for writing to any of the following:
+	// - the flowSchemas field
+	// - the slice held in the flowSchemas field
+	// - the priorityLevelStates field
+	// - the map held in the priorityLevelStates field
+	// - any field of a priorityLevelState held in that map
+	lock sync.RWMutex
 
 	// flowSchemas holds the flow schema objects, sorted by increasing
 	// numerical (decreasing logical) matching precedence.  Every
@@ -138,16 +161,6 @@ type configController struct {
 	// name to the state for that level.  Every name referenced from a
 	// member of `flowSchemas` has an entry here.
 	priorityLevelStates map[string]*priorityLevelState
-
-	// the most recent update attempts, ordered by increasing age.
-	// Consumer trims to keep only the last minute's worth of entries.
-	// The controller uses this to limit itself to at most six updates
-	// to a given FlowSchema in any minute.
-	// This may only be accessed from the one and only worker goroutine.
-	mostRecentUpdates []updateAttempt
-
-	// watchTracker implements the necessary WatchTracker interface.
-	WatchTracker
 }
 
 type updateAttempt struct {
@@ -281,8 +294,8 @@ func (cfgCtlr *configController) MaintainObservations(stopCh <-chan struct{}) {
 }
 
 func (cfgCtlr *configController) updateObservations() {
-	cfgCtlr.lock.Lock()
-	defer cfgCtlr.lock.Unlock()
+	cfgCtlr.lock.RLock()
+	defer cfgCtlr.lock.RUnlock()
 	for _, plc := range cfgCtlr.priorityLevelStates {
 		if plc.queues != nil {
 			plc.queues.UpdateObservations()
@@ -765,8 +778,8 @@ func (immediateRequest) Finish(execute func()) bool {
 // waiting in its queue, or `Time{}` if this did not happen.
 func (cfgCtlr *configController) startRequest(ctx context.Context, rd RequestDigest, queueNoteFn fq.QueueNoteFn) (fs *flowcontrol.FlowSchema, pl *flowcontrol.PriorityLevelConfiguration, isExempt bool, req fq.Request, startWaitingTime time.Time) {
 	klog.V(7).Infof("startRequest(%#+v)", rd)
-	cfgCtlr.lock.Lock()
-	defer cfgCtlr.lock.Unlock()
+	cfgCtlr.lock.RLock()
+	defer cfgCtlr.lock.RUnlock()
 	var selectedFlowSchema, catchAllFlowSchema *flowcontrol.FlowSchema
 	for _, fs := range cfgCtlr.flowSchemas {
 		if matchesFlowSchema(rd, fs) {
@@ -811,7 +824,7 @@ func (cfgCtlr *configController) startRequest(ctx context.Context, rd RequestDig
 	klog.V(7).Infof("startRequest(%#+v) => fsName=%q, distMethod=%#+v, plName=%q, numQueues=%d", rd, selectedFlowSchema.Name, selectedFlowSchema.Spec.DistinguisherMethod, plName, numQueues)
 	req, idle := plState.queues.StartRequest(ctx, &rd.Width, hashValue, flowDistinguisher, selectedFlowSchema.Name, rd.RequestInfo, rd.User, queueNoteFn)
 	if idle {
-		cfgCtlr.maybeReapLocked(plName, plState)
+		cfgCtlr.maybeReapReadLocked(plName, plState)
 	}
 	return selectedFlowSchema, plState.pl, false, req, startWaitingTime
 }
@@ -820,8 +833,8 @@ func (cfgCtlr *configController) startRequest(ctx context.Context, rd RequestDig
 // priority level if it has no more use.  Call this after getting a
 // clue that the given priority level is undesired and idle.
 func (cfgCtlr *configController) maybeReap(plName string) {
-	cfgCtlr.lock.Lock()
-	defer cfgCtlr.lock.Unlock()
+	cfgCtlr.lock.RLock()
+	defer cfgCtlr.lock.RUnlock()
 	plState := cfgCtlr.priorityLevelStates[plName]
 	if plState == nil {
 		klog.V(7).Infof("plName=%s, plState==nil", plName)
@@ -843,7 +856,7 @@ func (cfgCtlr *configController) maybeReap(plName string) {
 // it has no more use.  Call this if both (1) plState.queues is
 // non-nil and reported being idle, and (2) cfgCtlr's lock has not
 // been released since then.
-func (cfgCtlr *configController) maybeReapLocked(plName string, plState *priorityLevelState) {
+func (cfgCtlr *configController) maybeReapReadLocked(plName string, plState *priorityLevelState) {
 	if !(plState.quiescing && plState.numPending == 0) {
 		return
 	}

vendor/k8s.io/cli-runtime/pkg/genericclioptions/command_headers.go

@@ -77,3 +77,15 @@ func (c *CommandHeaderRoundTripper) ParseCommandHeaders(cmd *cobra.Command, args
 		c.Headers[kubectlCommandHeader] = strings.Join(cmdStrs, " ")
 	}
 }
+
+// CancelRequest is propagated to the Delegate RoundTripper within
+// if the wrapped RoundTripper implements this function.
+func (c *CommandHeaderRoundTripper) CancelRequest(req *http.Request) {
+	type canceler interface {
+		CancelRequest(*http.Request)
+	}
+	// If possible, call "CancelRequest" on the wrapped Delegate RoundTripper.
+	if cr, ok := c.Delegate.(canceler); ok {
+		cr.CancelRequest(req)
+	}
+}

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

@@ -3,8 +3,8 @@ package version
 var (
 	gitMajor = "1"
 	gitMinor = "22"
-	gitVersion   = "v1.22.2-k3s1"
-	gitCommit    = "c4ebe1924cba7aeae2a97729df3e0ac4a2ce5459"
+	gitVersion   = "v1.22.3-k3s1"
+	gitCommit    = "5d8c744cb54fc54d261b2f14e8464049aad7e32c"
 	gitTreeState = "clean"
-	buildDate = "2021-09-16T19:43:43Z"
+	buildDate = "2021-10-28T15:23:07Z"
 )

vendor/k8s.io/client-go/tools/clientcmd/api/v1/conversion.go

@@ -165,7 +165,7 @@ func Convert_Map_string_To_runtime_Object_To_Slice_v1_NamedExtension(in *map[str
 		newExtension := (*in)[key]
 		oldExtension := runtime.RawExtension{}
 		if err := runtime.Convert_runtime_Object_To_runtime_RawExtension(&newExtension, &oldExtension, s); err != nil {
-			return nil
+			return err
 		}
 		namedExtension := NamedExtension{key, oldExtension}
 		*out = append(*out, namedExtension)

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

@@ -3,8 +3,8 @@ package version
 var (
 	gitMajor = "1"
 	gitMinor = "22"
-	gitVersion   = "v1.22.2-k3s1"
-	gitCommit    = "c4ebe1924cba7aeae2a97729df3e0ac4a2ce5459"
+	gitVersion   = "v1.22.3-k3s1"
+	gitCommit    = "5d8c744cb54fc54d261b2f14e8464049aad7e32c"
 	gitTreeState = "clean"
-	buildDate = "2021-09-16T19:43:43Z"
+	buildDate = "2021-10-28T15:23:07Z"
 )

vendor/k8s.io/kube-aggregator/pkg/apiserver/handler_proxy.go

@@ -26,7 +26,6 @@ import (
 
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/httpstream"
-	"k8s.io/apimachinery/pkg/util/httpstream/spdy"
 	utilnet "k8s.io/apimachinery/pkg/util/net"
 	"k8s.io/apimachinery/pkg/util/proxy"
 	auditinternal "k8s.io/apiserver/pkg/apis/audit"
@@ -161,23 +160,21 @@ func (r *proxyHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 		return
 	}
 
-	// we need to wrap the roundtripper in another roundtripper which will apply the front proxy headers
-	proxyRoundTripper, upgrade, err := maybeWrapForConnectionUpgrades(handlingInfo.restConfig, handlingInfo.proxyRoundTripper, req)
-	if err != nil {
-		proxyError(w, req, err.Error(), http.StatusInternalServerError)
-		return
-	}
+	proxyRoundTripper := handlingInfo.proxyRoundTripper
+	upgrade := httpstream.IsUpgradeRequest(req)
+
 	proxyRoundTripper = transport.NewAuthProxyRoundTripper(user.GetName(), user.GetGroups(), user.GetExtra(), proxyRoundTripper)
 
-	// if we are upgrading, then the upgrade path tries to use this request with the TLS config we provide, but it does
-	// NOT use the roundtripper.  Its a direct call that bypasses the round tripper.  This means that we have to
-	// attach the "correct" user headers to the request ahead of time.  After the initial upgrade, we'll be back
-	// at the roundtripper flow, so we only have to muck with this request, but we do have to do it.
+	// If we are upgrading, then the upgrade path tries to use this request with the TLS config we provide, but it does
+	// NOT use the proxyRoundTripper.  It's a direct dial that bypasses the proxyRoundTripper.  This means that we have to
+	// attach the "correct" user headers to the request ahead of time.
 	if upgrade {
 		transport.SetAuthProxyHeaders(newReq, user.GetName(), user.GetGroups(), user.GetExtra())
 	}
 
 	handler := proxy.NewUpgradeAwareHandler(location, proxyRoundTripper, true, upgrade, &responder{w: w})
+	handler.InterceptRedirects = utilfeature.DefaultFeatureGate.Enabled(genericfeatures.StreamingProxyRedirects)
+	handler.RequireSameHostRedirects = utilfeature.DefaultFeatureGate.Enabled(genericfeatures.ValidateProxyRedirects)
 	handler.ServeHTTP(w, newReq)
 }
 
@@ -212,27 +209,6 @@ func newRequestForProxy(location *url.URL, req *http.Request) (*http.Request, co
 	return newReq, cancelFn
 }
 
-// maybeWrapForConnectionUpgrades wraps the roundtripper for upgrades.  The bool indicates if it was wrapped
-func maybeWrapForConnectionUpgrades(restConfig *restclient.Config, rt http.RoundTripper, req *http.Request) (http.RoundTripper, bool, error) {
-	if !httpstream.IsUpgradeRequest(req) {
-		return rt, false, nil
-	}
-
-	tlsConfig, err := restclient.TLSConfigFor(restConfig)
-	if err != nil {
-		return nil, true, err
-	}
-	followRedirects := utilfeature.DefaultFeatureGate.Enabled(genericfeatures.StreamingProxyRedirects)
-	requireSameHostRedirects := utilfeature.DefaultFeatureGate.Enabled(genericfeatures.ValidateProxyRedirects)
-	upgradeRoundTripper := spdy.NewRoundTripper(tlsConfig, followRedirects, requireSameHostRedirects)
-	wrappedRT, err := restclient.HTTPWrappersForConfig(restConfig, upgradeRoundTripper)
-	if err != nil {
-		return nil, true, err
-	}
-
-	return wrappedRT, true, nil
-}
-
 // responder implements rest.Responder for assisting a connector in writing objects or errors.
 type responder struct {
 	w http.ResponseWriter

vendor/k8s.io/kubectl/pkg/cmd/cmd.go

@@ -425,8 +425,12 @@ func addCmdHeaderHooks(cmds *cobra.Command, kubeConfigFlags *genericclioptions.C
 	// Wraps CommandHeaderRoundTripper around standard RoundTripper.
 	kubeConfigFlags.WrapConfigFn = func(c *rest.Config) *rest.Config {
 		c.Wrap(func(rt http.RoundTripper) http.RoundTripper {
-			crt.Delegate = rt
-			return crt
+			// Must be separate RoundTripper; not "crt" closure.
+			// Fixes: https://github.com/kubernetes/kubectl/issues/1098
+			return &genericclioptions.CommandHeaderRoundTripper{
+				Delegate: rt,
+				Headers:  crt.Headers,
+			}
 		})
 		return c
 	}

vendor/k8s.io/kubernetes/cmd/kube-controller-manager/app/core.go

@@ -577,7 +577,7 @@ func startPVCProtectionController(ctx ControllerContext) (http.Handler, bool, er
 		ctx.InformerFactory.Core().V1().Pods(),
 		ctx.ClientBuilder.ClientOrDie("pvc-protection-controller"),
 		utilfeature.DefaultFeatureGate.Enabled(features.StorageObjectInUseProtection),
-		utilfeature.DefaultFeatureGate.Enabled(features.StorageObjectInUseProtection),
+		utilfeature.DefaultFeatureGate.Enabled(features.GenericEphemeralVolume),
 	)
 	if err != nil {
 		return nil, true, fmt.Errorf("failed to start the pvc protection controller: %v", err)

vendor/k8s.io/kubernetes/pkg/apis/core/validation/events.go

@@ -21,7 +21,7 @@ import (
 	"reflect"
 	"time"
 
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	eventsv1beta1 "k8s.io/api/events/v1beta1"
 	apimachineryvalidation "k8s.io/apimachinery/pkg/api/validation"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -140,7 +140,7 @@ func legacyValidateEvent(event *core.Event) field.ErrorList {
 		}
 
 	} else {
-		if len(event.InvolvedObject.Namespace) == 0 && event.Namespace != metav1.NamespaceSystem {
+		if len(event.InvolvedObject.Namespace) == 0 && event.Namespace != metav1.NamespaceDefault && event.Namespace != metav1.NamespaceSystem {
 			allErrs = append(allErrs, field.Invalid(field.NewPath("involvedObject", "namespace"), event.InvolvedObject.Namespace, "does not match event.namespace"))
 		}
 		if len(event.ReportingController) == 0 {

vendor/k8s.io/kubernetes/pkg/apis/extensions/v1beta1/conversion.go

@@ -189,7 +189,7 @@ func Convert_networking_IngressBackend_To_v1beta1_IngressBackend(in *networking.
 
 func Convert_v1beta1_IngressSpec_To_networking_IngressSpec(in *extensionsv1beta1.IngressSpec, out *networking.IngressSpec, s conversion.Scope) error {
 	if err := autoConvert_v1beta1_IngressSpec_To_networking_IngressSpec(in, out, s); err != nil {
-		return nil
+		return err
 	}
 	if in.Backend != nil {
 		out.DefaultBackend = &networking.IngressBackend{}
@@ -202,7 +202,7 @@ func Convert_v1beta1_IngressSpec_To_networking_IngressSpec(in *extensionsv1beta1
 
 func Convert_networking_IngressSpec_To_v1beta1_IngressSpec(in *networking.IngressSpec, out *extensionsv1beta1.IngressSpec, s conversion.Scope) error {
 	if err := autoConvert_networking_IngressSpec_To_v1beta1_IngressSpec(in, out, s); err != nil {
-		return nil
+		return err
 	}
 	if in.DefaultBackend != nil {
 		out.Backend = &extensionsv1beta1.IngressBackend{}

vendor/k8s.io/kubernetes/pkg/apis/networking/v1beta1/conversion.go

@@ -52,7 +52,7 @@ func Convert_networking_IngressBackend_To_v1beta1_IngressBackend(in *networking.
 }
 func Convert_v1beta1_IngressSpec_To_networking_IngressSpec(in *v1beta1.IngressSpec, out *networking.IngressSpec, s conversion.Scope) error {
 	if err := autoConvert_v1beta1_IngressSpec_To_networking_IngressSpec(in, out, s); err != nil {
-		return nil
+		return err
 	}
 	if in.Backend != nil {
 		out.DefaultBackend = &networking.IngressBackend{}
@@ -65,7 +65,7 @@ func Convert_v1beta1_IngressSpec_To_networking_IngressSpec(in *v1beta1.IngressSp
 
 func Convert_networking_IngressSpec_To_v1beta1_IngressSpec(in *networking.IngressSpec, out *v1beta1.IngressSpec, s conversion.Scope) error {
 	if err := autoConvert_networking_IngressSpec_To_v1beta1_IngressSpec(in, out, s); err != nil {
-		return nil
+		return err
 	}
 	if in.DefaultBackend != nil {
 		out.Backend = &v1beta1.IngressBackend{}

vendor/k8s.io/kubernetes/pkg/controller/resourcequota/resource_quota_controller.go

@@ -287,7 +287,11 @@ func (rq *Controller) Run(workers int, stopCh <-chan struct{}) {
 		go wait.Until(rq.worker(rq.missingUsageQueue), time.Second, stopCh)
 	}
 	// the timer for how often we do a full recalculation across all quotas
+	if rq.resyncPeriod() > 0 {
 		go wait.Until(func() { rq.enqueueAll() }, rq.resyncPeriod(), stopCh)
+	} else {
+		klog.Warningf("periodic quota controller resync disabled")
+	}
 	<-stopCh
 }
 

vendor/k8s.io/kubernetes/pkg/kubelet/cm/memorymanager/policy_static.go

@@ -936,13 +936,13 @@ func isInitContainer(pod *v1.Pod, container *v1.Container) bool {
 func isNUMAAffinitiesEqual(numaAffinity1, numaAffinity2 []int) bool {
 	bitMask1, err := bitmask.NewBitMask(numaAffinity1...)
 	if err != nil {
-		klog.ErrorS(err, "failed to create bit mask", numaAffinity1)
+		klog.ErrorS(err, "failed to create bit mask", "numaAffinity1", numaAffinity1)
 		return false
 	}
 
 	bitMask2, err := bitmask.NewBitMask(numaAffinity2...)
 	if err != nil {
-		klog.ErrorS(err, "failed to create bit mask", numaAffinity2)
+		klog.ErrorS(err, "failed to create bit mask", "numaAffinity2", numaAffinity2)
 		return false
 	}
 

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

@@ -2227,6 +2227,8 @@ func (kl *Kubelet) HandlePodAdditions(pods []*v1.Pod) {
 		}
 		mirrorPod, _ := kl.podManager.GetMirrorPodByPod(pod)
 		kl.dispatchWork(pod, kubetypes.SyncPodCreate, mirrorPod, start)
+		// TODO: move inside syncPod and make reentrant
+		// https://github.com/kubernetes/kubernetes/issues/105014
 		kl.probeManager.AddPod(pod)
 	}
 }
@@ -2261,6 +2263,9 @@ func (kl *Kubelet) HandlePodRemoves(pods []*v1.Pod) {
 		if err := kl.deletePod(pod); err != nil {
 			klog.V(2).InfoS("Failed to delete pod", "pod", klog.KObj(pod), "err", err)
 		}
+		// TODO: move inside syncTerminatingPod|syncTerminatedPod (we should stop probing
+		// once the pod kill is acknowledged and during eviction)
+		// https://github.com/kubernetes/kubernetes/issues/105014
 		kl.probeManager.RemovePod(pod)
 	}
 }

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

@@ -978,7 +978,8 @@ func (kl *Kubelet) filterOutTerminatedPods(pods []*v1.Pod) []*v1.Pod {
 		if kl.podWorkers.IsPodKnownTerminated(p.UID) {
 			continue
 		}
-		if p.Status.Phase == v1.PodSucceeded || p.Status.Phase == v1.PodFailed {
+		// terminal pods are considered inactive UNLESS they are actively terminating
+		if kl.isAdmittedPodTerminal(p) && !kl.podWorkers.IsPodTerminationRequested(p.UID) {
 			continue
 		}
 		filteredPods = append(filteredPods, p)
@@ -986,6 +987,28 @@ func (kl *Kubelet) filterOutTerminatedPods(pods []*v1.Pod) []*v1.Pod {
 	return filteredPods
 }
 
+// isAdmittedPodTerminal returns true if the provided config source pod is in
+// a terminal phase, or if the Kubelet has already indicated the pod has reached
+// a terminal phase but the config source has not accepted it yet. This method
+// should only be used within the pod configuration loops that notify the pod
+// worker, other components should treat the pod worker as authoritative.
+func (kl *Kubelet) isAdmittedPodTerminal(pod *v1.Pod) bool {
+	// pods are considered inactive if the config source has observed a
+	// terminal phase (if the Kubelet recorded that the pod reached a terminal
+	// phase the pod should never be restarted)
+	if pod.Status.Phase == v1.PodSucceeded || pod.Status.Phase == v1.PodFailed {
+		return true
+	}
+	// a pod that has been marked terminal within the Kubelet is considered
+	// inactive (may have been rejected by Kubelet admision)
+	if status, ok := kl.statusManager.GetPodStatus(pod.UID); ok {
+		if status.Phase == v1.PodSucceeded || status.Phase == v1.PodFailed {
+			return true
+		}
+	}
+	return false
+}
+
 // removeOrphanedPodStatuses removes obsolete entries in podStatus where
 // the pod is no longer considered bound to this node.
 func (kl *Kubelet) removeOrphanedPodStatuses(pods []*v1.Pod, mirrorPods []*v1.Pod) {
@@ -1067,13 +1090,16 @@ func (kl *Kubelet) HandlePodCleanups() error {
 	// cleanup of pod cgroups.
 	runningPods := make(map[types.UID]sets.Empty)
 	possiblyRunningPods := make(map[types.UID]sets.Empty)
+	restartablePods := make(map[types.UID]sets.Empty)
 	for uid, sync := range workingPods {
 		switch sync {
-		case SyncPodWork:
+		case SyncPod:
 			runningPods[uid] = struct{}{}
 			possiblyRunningPods[uid] = struct{}{}
-		case TerminatingPodWork:
+		case TerminatingPod:
 			possiblyRunningPods[uid] = struct{}{}
+		case TerminatedAndRecreatedPod:
+			restartablePods[uid] = struct{}{}
 		}
 	}
 
@@ -1089,8 +1115,8 @@ func (kl *Kubelet) HandlePodCleanups() error {
 		return err
 	}
 	for _, runningPod := range runningRuntimePods {
-		switch workType, ok := workingPods[runningPod.ID]; {
-		case ok && workType == SyncPodWork, ok && workType == TerminatingPodWork:
+		switch workerState, ok := workingPods[runningPod.ID]; {
+		case ok && workerState == SyncPod, ok && workerState == TerminatingPod:
 			// if the pod worker is already in charge of this pod, we don't need to do anything
 			continue
 		default:
@@ -1157,6 +1183,32 @@ func (kl *Kubelet) HandlePodCleanups() error {
 	}
 
 	kl.backOff.GC()
+
+	// If two pods with the same UID are observed in rapid succession, we need to
+	// resynchronize the pod worker after the first pod completes and decide whether
+	// to restart the pod. This happens last to avoid confusing the desired state
+	// in other components and to increase the likelihood transient OS failures during
+	// container start are mitigated. In general only static pods will ever reuse UIDs
+	// since the apiserver uses randomly generated UUIDv4 UIDs with a very low
+	// probability of collision.
+	for uid := range restartablePods {
+		pod, ok := allPodsByUID[uid]
+		if !ok {
+			continue
+		}
+		if kl.isAdmittedPodTerminal(pod) {
+			klog.V(3).InfoS("Pod is restartable after termination due to UID reuse, but pod phase is terminal", "pod", klog.KObj(pod), "podUID", pod.UID)
+			continue
+		}
+		start := kl.clock.Now()
+		mirrorPod, _ := kl.podManager.GetMirrorPodByPod(pod)
+		klog.V(3).InfoS("Pod is restartable after termination due to UID reuse", "pod", klog.KObj(pod), "podUID", pod.UID)
+		kl.dispatchWork(pod, kubetypes.SyncPodCreate, mirrorPod, start)
+		// TODO: move inside syncPod and make reentrant
+		// https://github.com/kubernetes/kubernetes/issues/105014
+		kl.probeManager.AddPod(pod)
+	}
+
 	return nil
 }
 
@@ -1398,6 +1450,20 @@ func (kl *Kubelet) generateAPIPodStatus(pod *v1.Pod, podStatus *kubecontainer.Po
 	allStatus := append(append([]v1.ContainerStatus{}, s.ContainerStatuses...), s.InitContainerStatuses...)
 	s.Phase = getPhase(&pod.Spec, allStatus)
 	klog.V(4).InfoS("Got phase for pod", "pod", klog.KObj(pod), "oldPhase", oldPodStatus.Phase, "phase", s.Phase)
+
+	// Perform a three-way merge between the statuses from the status manager,
+	// runtime, and generated status to ensure terminal status is correctly set.
+	if s.Phase != v1.PodFailed && s.Phase != v1.PodSucceeded {
+		switch {
+		case oldPodStatus.Phase == v1.PodFailed || oldPodStatus.Phase == v1.PodSucceeded:
+			klog.V(4).InfoS("Status manager phase was terminal, updating phase to match", "pod", klog.KObj(pod), "phase", oldPodStatus.Phase)
+			s.Phase = oldPodStatus.Phase
+		case pod.Status.Phase == v1.PodFailed || pod.Status.Phase == v1.PodSucceeded:
+			klog.V(4).InfoS("API phase was terminal, updating phase to match", "pod", klog.KObj(pod), "phase", pod.Status.Phase)
+			s.Phase = pod.Status.Phase
+		}
+	}
+
 	if s.Phase == oldPodStatus.Phase {
 		// preserve the reason and message which is associated with the phase
 		s.Reason = oldPodStatus.Reason

vendor/k8s.io/kubernetes/pkg/kubelet/kuberuntime/fake_kuberuntime_manager.go

@@ -67,6 +67,11 @@ func newFakePodStateProvider() *fakePodStateProvider {
 	}
 }
 
+func (f *fakePodStateProvider) IsPodTerminationRequested(uid types.UID) bool {
+	_, found := f.removed[uid]
+	return found
+}
+
 func (f *fakePodStateProvider) ShouldPodRuntimeBeRemoved(uid types.UID) bool {
 	_, found := f.terminated[uid]
 	return found

vendor/k8s.io/kubernetes/pkg/kubelet/kuberuntime/kuberuntime_manager.go

@@ -78,6 +78,7 @@ var (
 // podStateProvider can determine if none of the elements are necessary to retain (pod content)
 // or if none of the runtime elements are necessary to retain (containers)
 type podStateProvider interface {
+	IsPodTerminationRequested(kubetypes.UID) bool
 	ShouldPodContentBeRemoved(kubetypes.UID) bool
 	ShouldPodRuntimeBeRemoved(kubetypes.UID) bool
 }
@@ -805,12 +806,14 @@ func (m *kubeGenericRuntimeManager) SyncPod(pod *v1.Pod, podStatus *kubecontaine
 			// or CRI if the Pod has been deleted while the POD is
 			// being created. If the pod has been deleted then it's
 			// not a real error.
-			// TODO: this is probably not needed now that termination is part of the sync loop
-			if m.podStateProvider.ShouldPodContentBeRemoved(pod.UID) {
+			//
+			// SyncPod can still be running when we get here, which
+			// means the PodWorker has not acked the deletion.
+			if m.podStateProvider.IsPodTerminationRequested(pod.UID) {
 				klog.V(4).InfoS("Pod was deleted and sandbox failed to be created", "pod", klog.KObj(pod), "podUID", pod.UID)
 				return
 			}
-			metrics.StartedPodsErrorsTotal.WithLabelValues(err.Error()).Inc()
+			metrics.StartedPodsErrorsTotal.Inc()
 			createSandboxResult.Fail(kubecontainer.ErrCreatePodSandbox, msg)
 			klog.ErrorS(err, "CreatePodSandbox for pod failed", "pod", klog.KObj(pod))
 			ref, referr := ref.GetReference(legacyscheme.Scheme, pod)

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

@@ -460,14 +460,13 @@ var (
 		},
 	)
 	// StartedPodsErrorsTotal is a counter that tracks the number of errors creating pod sandboxes
-	StartedPodsErrorsTotal = metrics.NewCounterVec(
+	StartedPodsErrorsTotal = metrics.NewCounter(
 		&metrics.CounterOpts{
 			Subsystem:      KubeletSubsystem,
 			Name:           StartedPodsErrorsTotalKey,
 			Help:           "Cumulative number of errors when starting pods",
 			StabilityLevel: metrics.ALPHA,
 		},
-		[]string{"message"},
 	)
 	// StartedContainersTotal is a counter that tracks the number of container creation operations
 	StartedContainersTotal = metrics.NewCounterVec(

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

@@ -28,6 +28,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/tools/record"
+	runtimeapi "k8s.io/cri-api/pkg/apis/runtime/v1alpha2"
 	"k8s.io/klog/v2"
 	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
 	"k8s.io/kubernetes/pkg/kubelet/events"
@@ -91,7 +92,7 @@ type UpdatePodOptions struct {
 type PodWorkType int
 
 const (
-	// SyncPodSync is when the pod is expected to be started and running.
+	// SyncPodWork is when the pod is expected to be started and running.
 	SyncPodWork PodWorkType = iota
 	// TerminatingPodWork is when the pod is no longer being set up, but some
 	// containers may be running and are being torn down.
@@ -101,6 +102,26 @@ const (
 	TerminatedPodWork
 )
 
+// PodWorkType classifies the status of pod as seen by the pod worker - setup (sync),
+// teardown of containers (terminating), cleanup (terminated), or recreated with the
+// same UID (kill -> create while terminating)
+type PodWorkerState int
+
+const (
+	// SyncPod is when the pod is expected to be started and running.
+	SyncPod PodWorkerState = iota
+	// TerminatingPod is when the pod is no longer being set up, but some
+	// containers may be running and are being torn down.
+	TerminatingPod
+	// TerminatedPod indicates the pod is stopped, can have no more running
+	// containers, and any foreground cleanup can be executed.
+	TerminatedPod
+	// TerminatedAndRecreatedPod indicates that after the pod was terminating a
+	// request to recreate the pod was received. The pod is terminated and can
+	// now be restarted by sending a create event to the pod worker.
+	TerminatedAndRecreatedPod
+)
+
 // podWork is the internal changes
 type podWork struct {
 	// WorkType is the type of sync to perform - sync (create), terminating (stop
@@ -127,8 +148,8 @@ type PodWorkers interface {
 	// and have been terminated for a significant period of time. Once this method
 	// has been called once, the workers are assumed to be fully initialized and
 	// subsequent calls to ShouldPodContentBeRemoved on unknown pods will return
-	// true.
-	SyncKnownPods(desiredPods []*v1.Pod) map[types.UID]PodWorkType
+	// true. It returns a map describing the state of each known pod worker.
+	SyncKnownPods(desiredPods []*v1.Pod) map[types.UID]PodWorkerState
 
 	// IsPodKnownTerminated returns true if the provided pod UID is known by the pod
 	// worker to be terminated. If the pod has been force deleted and the pod worker
@@ -254,6 +275,11 @@ type podSyncStatus struct {
 	// to remove the pod. A terminal pod (Succeeded/Failed) will have
 	// termination status until the pod is deleted.
 	finished bool
+	// restartRequested is true if the pod worker was informed the pod is
+	// expected to exist (update type of create, update, or sync) after
+	// it has been killed. When known pods are synced, any pod that is
+	// terminated and has restartRequested will have its history cleared.
+	restartRequested bool
 	// notifyPostTerminating will be closed once the pod transitions to
 	// terminated. After the pod is in terminated state, nothing should be
 	// added to this list.
@@ -476,6 +502,22 @@ func (p *podWorkers) IsPodForMirrorPodTerminatingByFullName(podFullName string)
 	return ok
 }
 
+func isPodStatusCacheTerminal(status *kubecontainer.PodStatus) bool {
+	runningContainers := 0
+	runningSandboxes := 0
+	for _, container := range status.ContainerStatuses {
+		if container.State == kubecontainer.ContainerStateRunning {
+			runningContainers++
+		}
+	}
+	for _, sb := range status.SandboxStatuses {
+		if sb.State == runtimeapi.PodSandboxState_SANDBOX_READY {
+			runningSandboxes++
+		}
+	}
+	return runningContainers == 0 && runningSandboxes == 0
+}
+
 // UpdatePod carries a configuration change or termination state to a pod. A pod is either runnable,
 // terminating, or terminated, and will transition to terminating if deleted on the apiserver, it is
 // discovered to have a terminal phase (Succeeded or Failed), or if it is evicted by the kubelet.
@@ -511,9 +553,38 @@ func (p *podWorkers) UpdatePod(options UpdatePodOptions) {
 		status = &podSyncStatus{
 			syncedAt: now,
 		}
+		// if this pod is being synced for the first time, we need to make sure it is an active pod
+		if !isRuntimePod && (pod.Status.Phase == v1.PodFailed || pod.Status.Phase == v1.PodSucceeded) {
+			// check to see if the pod is not running and the pod is terminal.
+			// If this succeeds then record in the podWorker that it is terminated.
+			if statusCache, err := p.podCache.Get(pod.UID); err == nil {
+				if isPodStatusCacheTerminal(statusCache) {
+					status = &podSyncStatus{
+						terminatedAt:       now,
+						terminatingAt:      now,
+						syncedAt:           now,
+						startedTerminating: true,
+						finished:           true,
+					}
+				}
+			}
+		}
 		p.podSyncStatuses[uid] = status
 	}
 
+	// if an update is received that implies the pod should be running, but we are already terminating a pod by
+	// that UID, assume that two pods with the same UID were created in close temporal proximity (usually static
+	// pod but it's possible for an apiserver to extremely rarely do something similar) - flag the sync status
+	// to indicate that after the pod terminates it should be reset to "not running" to allow a subsequent add/update
+	// to start the pod worker again
+	if status.IsTerminationRequested() {
+		if options.UpdateType == kubetypes.SyncPodCreate {
+			status.restartRequested = true
+			klog.V(4).InfoS("Pod is terminating but has been requested to restart with same UID, will be reconciled later", "pod", klog.KObj(pod), "podUID", pod.UID)
+			return
+		}
+	}
+
 	// once a pod is terminated by UID, it cannot reenter the pod worker (until the UID is purged by housekeeping)
 	if status.IsFinished() {
 		klog.V(4).InfoS("Pod is finished processing, no further updates", "pod", klog.KObj(pod), "podUID", pod.UID)
@@ -965,8 +1036,8 @@ func (p *podWorkers) contextForWorker(uid types.UID) context.Context {
 // to UpdatePods for new pods. It returns a map of known workers that are not finished
 // with a value of SyncPodTerminated, SyncPodKill, or SyncPodSync depending on whether
 // the pod is terminated, terminating, or syncing.
-func (p *podWorkers) SyncKnownPods(desiredPods []*v1.Pod) map[types.UID]PodWorkType {
-	workers := make(map[types.UID]PodWorkType)
+func (p *podWorkers) SyncKnownPods(desiredPods []*v1.Pod) map[types.UID]PodWorkerState {
+	workers := make(map[types.UID]PodWorkerState)
 	known := make(map[types.UID]struct{})
 	for _, pod := range desiredPods {
 		known[pod.UID] = struct{}{}
@@ -977,16 +1048,20 @@ func (p *podWorkers) SyncKnownPods(desiredPods []*v1.Pod) map[types.UID]PodWorkT
 
 	p.podsSynced = true
 	for uid, status := range p.podSyncStatuses {
-		if _, exists := known[uid]; !exists {
+		if _, exists := known[uid]; !exists || status.restartRequested {
 			p.removeTerminatedWorker(uid)
 		}
 		switch {
 		case !status.terminatedAt.IsZero():
-			workers[uid] = TerminatedPodWork
+			if status.restartRequested {
+				workers[uid] = TerminatedAndRecreatedPod
+			} else {
+				workers[uid] = TerminatedPod
+			}
 		case !status.terminatingAt.IsZero():
-			workers[uid] = TerminatingPodWork
+			workers[uid] = TerminatingPod
 		default:
-			workers[uid] = SyncPodWork
+			workers[uid] = SyncPod
 		}
 	}
 	return workers
@@ -1009,7 +1084,11 @@ func (p *podWorkers) removeTerminatedWorker(uid types.UID) {
 		return
 	}
 
+	if status.restartRequested {
+		klog.V(4).InfoS("Pod has been terminated but another pod with the same UID was created, remove history to allow restart", "podUID", uid)
+	} else {
 		klog.V(4).InfoS("Pod has been terminated and is no longer known to the kubelet, remove all history", "podUID", uid)
+	}
 	delete(p.podSyncStatuses, uid)
 	delete(p.podUpdates, uid)
 	delete(p.lastUndeliveredWorkUpdate, uid)

vendor/k8s.io/kubernetes/pkg/kubelet/status/status_manager.go

@@ -333,7 +333,7 @@ func (m *manager) TerminatePod(pod *v1.Pod) {
 	}
 	status := *oldStatus.DeepCopy()
 	for i := range status.ContainerStatuses {
-		if status.ContainerStatuses[i].State.Terminated != nil || status.ContainerStatuses[i].State.Waiting != nil {
+		if status.ContainerStatuses[i].State.Terminated != nil {
 			continue
 		}
 		status.ContainerStatuses[i].State = v1.ContainerState{
@@ -345,7 +345,7 @@ func (m *manager) TerminatePod(pod *v1.Pod) {
 		}
 	}
 	for i := range status.InitContainerStatuses {
-		if status.InitContainerStatuses[i].State.Terminated != nil || status.InitContainerStatuses[i].State.Waiting != nil {
+		if status.InitContainerStatuses[i].State.Terminated != nil {
 			continue
 		}
 		status.InitContainerStatuses[i].State = v1.ContainerState{

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

@@ -1041,6 +1041,9 @@ func (proxier *Proxier) syncProxyRules() {
 			for _, extIP := range svcInfo.ExternalIPStrings() {
 				staleServices.Insert(extIP)
 			}
+			for _, extIP := range svcInfo.LoadBalancerIPStrings() {
+				staleServices.Insert(extIP)
+			}
 		}
 	}
 

vendor/k8s.io/kubernetes/pkg/registry/core/service/storage/rest.go

@@ -759,6 +759,12 @@ func (rs *REST) handleClusterIPsForUpdatedService(oldService *api.Service, servi
 	}
 
 	// CASE B:
+
+	// if headless service then we bail out early (no clusterIPs management needed)
+	if len(oldService.Spec.ClusterIPs) > 0 && oldService.Spec.ClusterIPs[0] == api.ClusterIPNone {
+		return nil, nil, nil
+	}
+
 	// Update service from non-ExternalName to ExternalName, should release ClusterIP if exists.
 	if oldService.Spec.Type != api.ServiceTypeExternalName && service.Spec.Type == api.ServiceTypeExternalName {
 		toRelease = make(map[api.IPFamily]string)
@@ -775,11 +781,6 @@ func (rs *REST) handleClusterIPsForUpdatedService(oldService *api.Service, servi
 		return nil, toRelease, nil
 	}
 
-	// if headless service then we bail out early (no clusterIPs management needed)
-	if len(oldService.Spec.ClusterIPs) > 0 && oldService.Spec.ClusterIPs[0] == api.ClusterIPNone {
-		return nil, nil, nil
-	}
-
 	// upgrade and downgrade are specific to dualstack
 	if !utilfeature.DefaultFeatureGate.Enabled(features.IPv6DualStack) {
 		return nil, nil, nil

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

@@ -453,5 +453,8 @@ func getValidCreationData(subscriptionID, resourceGroup, sourceResourceID, sourc
 
 func isInstanceNotFoundError(err error) bool {
 	errMsg := strings.ToLower(err.Error())
+	if strings.Contains(errMsg, strings.ToLower(vmssVMNotActiveErrorMessage)) {
+		return true
+	}
 	return strings.Contains(errMsg, errStatusCode400) && strings.Contains(errMsg, errInvalidParameter) && strings.Contains(errMsg, errTargetInstanceIds)
 }

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

@@ -20,6 +20,7 @@ package azure
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"math"
 	"reflect"
@@ -304,7 +305,7 @@ func (az *Cloud) cleanBackendpoolForPrimarySLB(primarySLB *network.LoadBalancer,
 					ipConf := (*bp.BackendIPConfigurations)[i]
 					ipConfigID := to.String(ipConf.ID)
 					_, vmSetName, err := az.VMSet.GetNodeNameByIPConfigurationID(ipConfigID)
-					if err != nil {
+					if err != nil && !errors.Is(err, cloudprovider.InstanceNotFound) {
 						return nil, err
 					}
 					primaryVMSetName := az.VMSet.GetPrimaryVMSetName()
@@ -1129,13 +1130,10 @@ func (az *Cloud) reconcileLoadBalancer(clusterName string, service *v1.Service,
 					for _, ipConf := range *bp.BackendIPConfigurations {
 						ipConfID := to.String(ipConf.ID)
 						nodeName, _, err := az.VMSet.GetNodeNameByIPConfigurationID(ipConfID)
-						if err != nil {
+						if err != nil && !errors.Is(err, cloudprovider.InstanceNotFound) {
 							return nil, err
 						}
-						if nodeName == "" {
-							// VM may under deletion
-							continue
-						}
+
 						// If a node is not supposed to be included in the LB, it
 						// would not be in the `nodes` slice. We need to check the nodes that
 						// have been added to the LB's backendpool, find the unwanted ones and

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

@@ -458,6 +458,7 @@ func (as *availabilitySet) GetInstanceIDByNodeName(name string) (string, error)
 
 	machine, err = as.getVirtualMachine(types.NodeName(name), azcache.CacheReadTypeUnsafe)
 	if err == cloudprovider.InstanceNotFound {
+		klog.Warningf("Unable to find node %s: %v", name, cloudprovider.InstanceNotFound)
 		return "", cloudprovider.InstanceNotFound
 	}
 	if err != nil {
@@ -965,13 +966,16 @@ func (as *availabilitySet) EnsureBackendPoolDeleted(service *v1.Service, backend
 		}
 	}
 	nicUpdaters := make([]func() error, 0)
-	errors := make([]error, 0)
+	allErrs := make([]error, 0)
 	for i := range ipConfigurationIDs {
 		ipConfigurationID := ipConfigurationIDs[i]
 		nodeName, _, err := as.GetNodeNameByIPConfigurationID(ipConfigurationID)
-		if err != nil {
+		if err != nil && !errors.Is(err, cloudprovider.InstanceNotFound) {
 			klog.Errorf("Failed to GetNodeNameByIPConfigurationID(%s): %v", ipConfigurationID, err)
-			errors = append(errors, err)
+			allErrs = append(allErrs, err)
+			continue
+		}
+		if nodeName == "" {
 			continue
 		}
 
@@ -1038,9 +1042,9 @@ func (as *availabilitySet) EnsureBackendPoolDeleted(service *v1.Service, backend
 	if errs != nil {
 		return utilerrors.Flatten(errs)
 	}
-	// Fail if there are other errors.
-	if len(errors) > 0 {
-		return utilerrors.Flatten(utilerrors.NewAggregate(errors))
+	// Fail if there are other allErrs.
+	if len(allErrs) > 0 {
+		return utilerrors.Flatten(utilerrors.NewAggregate(allErrs))
 	}
 
 	isOperationSucceeded = true
@@ -1104,7 +1108,8 @@ func (as *availabilitySet) GetNodeNameByIPConfigurationID(ipConfigurationID stri
 
 	vm, err := as.getVirtualMachine(types.NodeName(vmName), azcache.CacheReadTypeDefault)
 	if err != nil {
-		return "", "", fmt.Errorf("cannot get the virtual machine by node name %s", vmName)
+		klog.Errorf("Unable to get the virtual machine by node name %s: %v", vmName, err)
+		return "", "", err
 	}
 	asID := ""
 	if vm.VirtualMachineProperties != nil && vm.AvailabilitySet != nil {
@@ -1116,7 +1121,7 @@ func (as *availabilitySet) GetNodeNameByIPConfigurationID(ipConfigurationID stri
 
 	asName, err := getAvailabilitySetNameByID(asID)
 	if err != nil {
-		return "", "", fmt.Errorf("cannot get the availability set name by the availability set ID %s", asID)
+		return "", "", fmt.Errorf("cannot get the availability set name by the availability set ID %s: %v", asID, err)
 	}
 	return vmName, strings.ToLower(asName), nil
 }

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

@@ -195,6 +195,7 @@ func (ss *scaleSet) getVmssVMByNodeIdentity(node *nodeIdentity, crt azcache.Azur
 	}
 
 	if !found || vm == nil {
+		klog.Warningf("Unable to find node %s: %v", node.nodeName, cloudprovider.InstanceNotFound)
 		return "", "", nil, cloudprovider.InstanceNotFound
 	}
 	return vmssName, instanceID, vm, nil
@@ -342,6 +343,7 @@ func (ss *scaleSet) GetInstanceIDByNodeName(name string) (string, error) {
 
 	_, _, vm, err := ss.getVmssVM(name, azcache.CacheReadTypeUnsafe)
 	if err != nil {
+		klog.Errorf("Unable to find node %s: %v", name, err)
 		return "", err
 	}
 
@@ -389,6 +391,7 @@ func (ss *scaleSet) GetNodeNameByProviderID(providerID string) (types.NodeName,
 
 	vm, err := ss.getVmssVMByInstanceID(resourceGroup, scaleSetName, instanceID, azcache.CacheReadTypeUnsafe)
 	if err != nil {
+		klog.Errorf("Unable to find node by providerID %s: %v", providerID, err)
 		return "", err
 	}
 
@@ -709,6 +712,7 @@ func (ss *scaleSet) getNodeIdentityByNodeName(nodeName string, crt azcache.Azure
 		return nil, err
 	}
 	if node.vmssName == "" {
+		klog.Warningf("Unable to find node %s: %v", nodeName, cloudprovider.InstanceNotFound)
 		return nil, cloudprovider.InstanceNotFound
 	}
 	return node, nil
@@ -721,7 +725,7 @@ func (ss *scaleSet) listScaleSetVMs(scaleSetName, resourceGroup string) ([]compu
 
 	allVMs, rerr := ss.VirtualMachineScaleSetVMsClient.List(ctx, resourceGroup, scaleSetName, string(compute.InstanceView))
 	if rerr != nil {
-		klog.Errorf("VirtualMachineScaleSetVMsClient.List failed: %v", rerr)
+		klog.Errorf("VirtualMachineScaleSetVMsClient.List(%s, %s) failed: %v", resourceGroup, scaleSetName, rerr)
 		if rerr.IsNotFound() {
 			return nil, cloudprovider.InstanceNotFound
 		}
@@ -960,6 +964,11 @@ func (ss *scaleSet) EnsureHostInPool(service *v1.Service, nodeName types.NodeNam
 	vmName := mapNodeNameToVMName(nodeName)
 	ssName, instanceID, vm, err := ss.getVmssVM(vmName, azcache.CacheReadTypeDefault)
 	if err != nil {
+		if errors.Is(err, cloudprovider.InstanceNotFound) {
+			klog.Infof("EnsureHostInPool: skipping node %s because it is not found", vmName)
+			return "", "", "", nil, nil
+		}
+
 		return "", "", "", nil, err
 	}
 
@@ -1335,6 +1344,11 @@ func (ss *scaleSet) EnsureHostsInPool(service *v1.Service, nodes []*v1.Node, bac
 func (ss *scaleSet) ensureBackendPoolDeletedFromNode(nodeName, backendPoolID string) (string, string, string, *compute.VirtualMachineScaleSetVM, error) {
 	ssName, instanceID, vm, err := ss.getVmssVM(nodeName, azcache.CacheReadTypeDefault)
 	if err != nil {
+		if errors.Is(err, cloudprovider.InstanceNotFound) {
+			klog.Infof("ensureBackendPoolDeletedFromNode: skipping node %s because it is not found", nodeName)
+			return "", "", "", nil, nil
+		}
+
 		return "", "", "", nil, err
 	}
 
@@ -1407,7 +1421,7 @@ func (ss *scaleSet) GetNodeNameByIPConfigurationID(ipConfigurationID string) (st
 
 		klog.V(4).Infof("Can not extract scale set name from ipConfigurationID (%s), assuming it is managed by availability set", ipConfigurationID)
 		name, rg, err := ss.availabilitySet.GetNodeNameByIPConfigurationID(ipConfigurationID)
-		if err != nil {
+		if err != nil && !errors.Is(err, cloudprovider.InstanceNotFound) {
 			klog.Errorf("GetNodeNameByIPConfigurationID: failed to invoke availabilitySet.GetNodeNameByIPConfigurationID: %s", err.Error())
 			return "", "", err
 		}
@@ -1419,6 +1433,7 @@ func (ss *scaleSet) GetNodeNameByIPConfigurationID(ipConfigurationID string) (st
 	instanceID := matches[3]
 	vm, err := ss.getVmssVMByInstanceID(resourceGroup, scaleSetName, instanceID, azcache.CacheReadTypeUnsafe)
 	if err != nil {
+		klog.Errorf("Unable to find node by ipConfigurationID %s: %v", ipConfigurationID, err)
 		return "", "", err
 	}
 
@@ -1580,7 +1595,7 @@ func (ss *scaleSet) EnsureBackendPoolDeleted(service *v1.Service, backendPoolID,
 
 	hostUpdates := make([]func() error, 0, len(ipConfigurationIDs))
 	nodeUpdates := make(map[vmssMetaInfo]map[string]compute.VirtualMachineScaleSetVM)
-	errors := make([]error, 0)
+	allErrs := make([]error, 0)
 	for i := range ipConfigurationIDs {
 		ipConfigurationID := ipConfigurationIDs[i]
 
@@ -1600,14 +1615,16 @@ func (ss *scaleSet) EnsureBackendPoolDeleted(service *v1.Service, backendPoolID,
 			}
 
 			klog.Errorf("Failed to GetNodeNameByIPConfigurationID(%s): %v", ipConfigurationID, err)
-			errors = append(errors, err)
+			allErrs = append(allErrs, err)
 			continue
 		}
 
 		nodeResourceGroup, nodeVMSS, nodeInstanceID, nodeVMSSVM, err := ss.ensureBackendPoolDeletedFromNode(nodeName, backendPoolID)
 		if err != nil {
+			if !errors.Is(err, ErrorNotVmssInstance) { // Do nothing for the VMAS nodes.
 				klog.Errorf("EnsureBackendPoolDeleted(%s): backendPoolID(%s) - failed with error %v", getServiceName(service), backendPoolID, err)
-			errors = append(errors, err)
+				allErrs = append(allErrs, err)
+			}
 			continue
 		}
 
@@ -1653,8 +1670,8 @@ func (ss *scaleSet) EnsureBackendPoolDeleted(service *v1.Service, backendPoolID,
 	}
 
 	// Fail if there are other errors.
-	if len(errors) > 0 {
-		return utilerrors.Flatten(utilerrors.NewAggregate(errors))
+	if len(allErrs) > 0 {
+		return utilerrors.Flatten(utilerrors.NewAggregate(allErrs))
 	}
 
 	// Ensure the backendPoolID is also deleted on VMSS itself.

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

@@ -76,7 +76,7 @@ func (err *Error) Error() error {
 		retryAfterSeconds = int(err.RetryAfter.Sub(curTime) / time.Second)
 	}
 
-	return fmt.Errorf("Retriable: %v, RetryAfter: %ds, HTTPStatusCode: %d, RawError: %v",
+	return fmt.Errorf("Retriable: %v, RetryAfter: %ds, HTTPStatusCode: %d, RawError: %w",
 		err.Retriable, retryAfterSeconds, err.HTTPStatusCode, err.RawError)
 }