Merge remote-tracking branch 'origin/master' into release-1.14

api/openapi-spec/swagger.json

@@ -17546,6 +17546,10 @@
     "io.k8s.apimachinery.pkg.apis.meta.v1.Preconditions": {
       "description": "Preconditions must be fulfilled before an operation (update, delete, etc.) is carried out.",
       "properties": {
+        "resourceVersion": {
+          "description": "Specifies the target ResourceVersion",
+          "type": "string"
+        },
         "uid": {
           "description": "Specifies the target UID.",
           "type": "string"

build/root/WORKSPACE

@@ -19,16 +19,6 @@ http_archive(
     urls = mirror("https://github.com/kubernetes/repo-infra/archive/b461270ab6ccfb94ff2d78df96d26f669376d660.tar.gz"),
 )
 
-ETCD_VERSION = "3.3.10"
-
-http_archive(
-    name = "com_coreos_etcd",
-    build_file = "@//third_party:etcd.BUILD",
-    sha256 = "1620a59150ec0a0124a65540e23891243feb2d9a628092fb1edcc23974724a45",
-    strip_prefix = "etcd-v%s-linux-amd64" % ETCD_VERSION,
-    urls = mirror("https://github.com/coreos/etcd/releases/download/v%s/etcd-v%s-linux-amd64.tar.gz" % (ETCD_VERSION, ETCD_VERSION)),
-)
-
 http_archive(
     name = "io_bazel_rules_go",
     sha256 = "6776d68ebb897625dead17ae510eac3d5f6342367327875210df44dbe2aeeb19",

build/workspace.bzl

@@ -14,7 +14,7 @@
 
 load("//build:platforms.bzl", "SERVER_PLATFORMS")
 load("//build:workspace_mirror.bzl", "mirror")
-load("@bazel_tools//tools/build_defs/repo:http.bzl", "http_file")
+load("@bazel_tools//tools/build_defs/repo:http.bzl", "http_archive", "http_file")
 load("@io_bazel_rules_docker//container:container.bzl", "container_pull")
 
 CNI_VERSION = "0.6.0"
@@ -35,6 +35,13 @@ _CRI_TARBALL_ARCH_SHA256 = {
     "s390x": "814aa9cd496be416612c2653097a1c9eb5784e38aa4889034b44ebf888709057",
 }
 
+ETCD_VERSION = "3.3.10"
+_ETCD_TARBALL_ARCH_SHA256 = {
+    "amd64": "1620a59150ec0a0124a65540e23891243feb2d9a628092fb1edcc23974724a45",
+    "arm64": "5ec97b0b872adce275b8130d19db314f7f2b803aeb24c4aae17a19e2d66853c4",
+    "ppc64le": "148fe96f0ec1813c5db9916199e96a913174304546bc8447a2d2f9fee4b8f6c2",
+}
+
 # Note that these are digests for the manifest list. We resolve the manifest
 # list to each of its platform-specific images in
 # debian_image_dependencies().
@@ -48,6 +55,7 @@ def release_dependencies():
     cni_tarballs()
     cri_tarballs()
     debian_image_dependencies()
+    etcd_tarballs()
 
 def cni_tarballs():
     for arch, sha in _CNI_TARBALL_ARCH_SHA256.items():
@@ -92,3 +100,13 @@ def debian_image_dependencies():
             registry = "k8s.gcr.io",
             repository = "debian-hyperkube-base",
         )
+
+def etcd_tarballs():
+    for arch, sha in _ETCD_TARBALL_ARCH_SHA256.items():
+        http_archive(
+            name = "com_coreos_etcd_%s" % arch,
+            build_file = "@//third_party:etcd.BUILD",
+            sha256 = sha,
+            strip_prefix = "etcd-v%s-linux-%s" % (ETCD_VERSION, arch),
+            urls = mirror("https://github.com/coreos/etcd/releases/download/v%s/etcd-v%s-linux-%s.tar.gz" % (ETCD_VERSION, ETCD_VERSION, arch)),
+        )

pkg/cloudprovider/providers/azure/azure_loadbalancer.go

@@ -169,27 +169,47 @@ func (az *Cloud) UpdateLoadBalancer(ctx context.Context, clusterName string, ser
 func (az *Cloud) EnsureLoadBalancerDeleted(ctx context.Context, clusterName string, service *v1.Service) error {
 	isInternal := requiresInternalLoadBalancer(service)
 	serviceName := getServiceName(service)
-	klog.V(5).Infof("delete(%s): START clusterName=%q", serviceName, clusterName)
+	klog.V(5).Infof("Delete service (%s): START clusterName=%q", serviceName, clusterName)
+
+	ignoreErrors := func(err error) error {
+		if ignoreStatusNotFoundFromError(err) == nil {
+			klog.V(5).Infof("EnsureLoadBalancerDeleted: ignoring StatusNotFound error because the resource doesn't exist (%v)", err)
+			return nil
+		}
+
+		if ignoreStatusForbiddenFromError(err) == nil {
+			klog.V(5).Infof("EnsureLoadBalancerDeleted: ignoring StatusForbidden error (%v). This may be caused by wrong configuration via service annotations", err)
+			return nil
+		}
+
+		return err
+	}
 
 	serviceIPToCleanup, err := az.findServiceIPAddress(ctx, clusterName, service, isInternal)
-	if err != nil {
+	if ignoreErrors(err) != nil {
 		return err
 	}
 
 	klog.V(2).Infof("EnsureLoadBalancerDeleted: reconciling security group for service %q with IP %q, wantLb = false", serviceName, serviceIPToCleanup)
 	if _, err := az.reconcileSecurityGroup(clusterName, service, &serviceIPToCleanup, false /* wantLb */); err != nil {
-		return err
+		if ignoreErrors(err) != nil {
+			return err
+		}
 	}
 
 	if _, err := az.reconcileLoadBalancer(clusterName, service, nil, false /* wantLb */); err != nil {
-		return err
+		if ignoreErrors(err) != nil {
+			return err
+		}
 	}
 
 	if _, err := az.reconcilePublicIP(clusterName, service, nil, false /* wantLb */); err != nil {
-		return err
+		if ignoreErrors(err) != nil {
+			return err
+		}
 	}
 
-	klog.V(2).Infof("delete(%s): FINISH", serviceName)
+	klog.V(2).Infof("Delete service (%s): FINISH", serviceName)
 	return nil
 }
 

pkg/cloudprovider/providers/azure/azure_loadbalancer_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package azure
 
 import (
+	"context"
 	"fmt"
 	"reflect"
 	"testing"
@@ -308,3 +309,62 @@ func TestSubnet(t *testing.T) {
 		assert.Equal(t, c.expected, real, fmt.Sprintf("TestCase[%d]: %s", i, c.desc))
 	}
 }
+
+func TestEnsureLoadBalancerDeleted(t *testing.T) {
+	const vmCount = 8
+	const availabilitySetCount = 4
+	const serviceCount = 9
+
+	tests := []struct {
+		desc              string
+		service           v1.Service
+		expectCreateError bool
+	}{
+		{
+			desc:    "external service should be created and deleted successfully",
+			service: getTestService("test1", v1.ProtocolTCP, 80),
+		},
+		{
+			desc:    "internal service should be created and deleted successfully",
+			service: getInternalTestService("test2", 80),
+		},
+		{
+			desc:    "annotated service with same resourceGroup should be created and deleted successfully",
+			service: getResourceGroupTestService("test3", "rg", "", 80),
+		},
+		{
+			desc:              "annotated service with different resourceGroup shouldn't be created but should be deleted successfully",
+			service:           getResourceGroupTestService("test4", "random-rg", "1.2.3.4", 80),
+			expectCreateError: true,
+		},
+	}
+
+	az := getTestCloud()
+	for i, c := range tests {
+		clusterResources := getClusterResources(az, vmCount, availabilitySetCount)
+		getTestSecurityGroup(az)
+		if c.service.Annotations[ServiceAnnotationLoadBalancerInternal] == "true" {
+			addTestSubnet(t, az, &c.service)
+		}
+
+		// create the service first.
+		lbStatus, err := az.EnsureLoadBalancer(context.TODO(), testClusterName, &c.service, clusterResources.nodes)
+		if c.expectCreateError {
+			assert.NotNil(t, err, "TestCase[%d]: %s", i, c.desc)
+		} else {
+			assert.Nil(t, err, "TestCase[%d]: %s", i, c.desc)
+			assert.NotNil(t, lbStatus, "TestCase[%d]: %s", i, c.desc)
+			result, err := az.LoadBalancerClient.List(context.TODO(), az.Config.ResourceGroup)
+			assert.Nil(t, err, "TestCase[%d]: %s", i, c.desc)
+			assert.Equal(t, len(result), 1, "TestCase[%d]: %s", i, c.desc)
+			assert.Equal(t, len(*result[0].LoadBalancingRules), 1, "TestCase[%d]: %s", i, c.desc)
+		}
+
+		// finally, delete it.
+		err = az.EnsureLoadBalancerDeleted(context.TODO(), testClusterName, &c.service)
+		assert.Nil(t, err, "TestCase[%d]: %s", i, c.desc)
+		result, err := az.LoadBalancerClient.List(context.Background(), az.Config.ResourceGroup)
+		assert.Nil(t, err, "TestCase[%d]: %s", i, c.desc)
+		assert.Equal(t, len(result), 0, "TestCase[%d]: %s", i, c.desc)
+	}
+}

pkg/cloudprovider/providers/azure/azure_test.go

@@ -1137,6 +1137,13 @@ func getInternalTestService(identifier string, requestedPorts ...int32) v1.Servi
 	return svc
 }
 
+func getResourceGroupTestService(identifier, resourceGroup, loadBalancerIP string, requestedPorts ...int32) v1.Service {
+	svc := getTestService(identifier, v1.ProtocolTCP, requestedPorts...)
+	svc.Spec.LoadBalancerIP = loadBalancerIP
+	svc.Annotations[ServiceAnnotationLoadBalancerResourceGroup] = resourceGroup
+	return svc
+}
+
 func setLoadBalancerModeAnnotation(service *v1.Service, lbMode string) {
 	service.Annotations[ServiceAnnotationLoadBalancerMode] = lbMode
 }

pkg/cloudprovider/providers/azure/azure_wrap.go

@@ -71,6 +71,19 @@ func ignoreStatusNotFoundFromError(err error) error {
 	return err
 }
 
+// ignoreStatusForbiddenFromError returns nil if the status code is StatusForbidden.
+// This happens when AuthorizationFailed is reported from Azure API.
+func ignoreStatusForbiddenFromError(err error) error {
+	if err == nil {
+		return nil
+	}
+	v, ok := err.(autorest.DetailedError)
+	if ok && v.StatusCode == http.StatusForbidden {
+		return nil
+	}
+	return err
+}
+
 /// getVirtualMachine calls 'VirtualMachinesClient.Get' with a timed cache
 /// The service side has throttling control that delays responses if there're multiple requests onto certain vm
 /// resource request in short period.

pkg/proxy/iptables/proxier.go

@@ -369,6 +369,7 @@ var iptablesJumpChains = []iptablesJumpChain{
 	{utiliptables.TableFilter, kubeExternalServicesChain, utiliptables.ChainInput, "kubernetes externally-visible service portals", []string{"-m", "conntrack", "--ctstate", "NEW"}},
 	{utiliptables.TableFilter, kubeServicesChain, utiliptables.ChainForward, "kubernetes service portals", []string{"-m", "conntrack", "--ctstate", "NEW"}},
 	{utiliptables.TableFilter, kubeServicesChain, utiliptables.ChainOutput, "kubernetes service portals", []string{"-m", "conntrack", "--ctstate", "NEW"}},
+	{utiliptables.TableFilter, kubeServicesChain, utiliptables.ChainInput, "kubernetes service portals", []string{"-m", "conntrack", "--ctstate", "NEW"}},
 	{utiliptables.TableFilter, kubeForwardChain, utiliptables.ChainForward, "kubernetes forwarding rules", nil},
 	{utiliptables.TableNAT, kubeServicesChain, utiliptables.ChainOutput, "kubernetes service portals", nil},
 	{utiliptables.TableNAT, kubeServicesChain, utiliptables.ChainPrerouting, "kubernetes service portals", nil},
@@ -847,6 +848,7 @@ func (proxier *Proxier) syncProxyRules() {
 			}
 			writeLine(proxier.natRules, append(args, "-j", string(svcChain))...)
 		} else {
+			// No endpoints.
 			writeLine(proxier.filterRules,
 				"-A", string(kubeServicesChain),
 				"-m", "comment", "--comment", fmt.Sprintf(`"%s has no endpoints"`, svcNameString),
@@ -917,6 +919,7 @@ func (proxier *Proxier) syncProxyRules() {
 				// This covers cases like GCE load-balancers which get added to the local routing table.
 				writeLine(proxier.natRules, append(dstLocalOnlyArgs, "-j", string(svcChain))...)
 			} else {
+				// No endpoints.
 				writeLine(proxier.filterRules,
 					"-A", string(kubeExternalServicesChain),
 					"-m", "comment", "--comment", fmt.Sprintf(`"%s has no endpoints"`, svcNameString),
@@ -929,10 +932,10 @@ func (proxier *Proxier) syncProxyRules() {
 		}
 
 		// Capture load-balancer ingress.
-		if hasEndpoints {
-			fwChain := svcInfo.serviceFirewallChainName
-			for _, ingress := range svcInfo.LoadBalancerStatus.Ingress {
-				if ingress.IP != "" {
+		fwChain := svcInfo.serviceFirewallChainName
+		for _, ingress := range svcInfo.LoadBalancerStatus.Ingress {
+			if ingress.IP != "" {
+				if hasEndpoints {
 					// create service firewall chain
 					if chain, ok := existingNATChains[fwChain]; ok {
 						writeBytesLine(proxier.natChains, chain)
@@ -993,10 +996,19 @@ func (proxier *Proxier) syncProxyRules() {
 					// If the packet was able to reach the end of firewall chain, then it did not get DNATed.
 					// It means the packet cannot go thru the firewall, then mark it for DROP
 					writeLine(proxier.natRules, append(args, "-j", string(KubeMarkDropChain))...)
+				} else {
+					// No endpoints.
+					writeLine(proxier.filterRules,
+						"-A", string(kubeServicesChain),
+						"-m", "comment", "--comment", fmt.Sprintf(`"%s has no endpoints"`, svcNameString),
+						"-m", protocol, "-p", protocol,
+						"-d", utilproxy.ToCIDR(net.ParseIP(ingress.IP)),
+						"--dport", strconv.Itoa(svcInfo.Port),
+						"-j", "REJECT",
+					)
 				}
 			}
 		}
-		// FIXME: do we need REJECT rules for load-balancer ingress if !hasEndpoints?
 
 		// Capture nodeports.  If we had more than 2 rules it might be
 		// worthwhile to make a new per-service chain for nodeport rules, but
@@ -1078,6 +1090,7 @@ func (proxier *Proxier) syncProxyRules() {
 					writeLine(proxier.natRules, append(args, "-j", string(svcXlbChain))...)
 				}
 			} else {
+				// No endpoints.
 				writeLine(proxier.filterRules,
 					"-A", string(kubeExternalServicesChain),
 					"-m", "comment", "--comment", fmt.Sprintf(`"%s has no endpoints"`, svcNameString),

pkg/registry/core/namespace/storage/storage.go

@@ -147,6 +147,14 @@ func (r *REST) Delete(ctx context.Context, name string, options *metav1.DeleteOp
 		)
 		return nil, false, err
 	}
+	if options.Preconditions.ResourceVersion != nil && *options.Preconditions.ResourceVersion != namespace.ResourceVersion {
+		err = apierrors.NewConflict(
+			api.Resource("namespaces"),
+			name,
+			fmt.Errorf("Precondition failed: ResourceVersion in precondition: %v, ResourceVersion in object meta: %v", *options.Preconditions.ResourceVersion, namespace.ResourceVersion),
+		)
+		return nil, false, err
+	}
 
 	// upon first request to delete, we switch the phase to start namespace termination
 	// TODO: enhance graceful deletion's calls to DeleteStrategy to allow phase change and finalizer patterns
@@ -156,7 +164,7 @@ func (r *REST) Delete(ctx context.Context, name string, options *metav1.DeleteOp
 			return nil, false, err
 		}
 
-		preconditions := storage.Preconditions{UID: options.Preconditions.UID}
+		preconditions := storage.Preconditions{UID: options.Preconditions.UID, ResourceVersion: options.Preconditions.ResourceVersion}
 
 		out := r.store.NewFunc()
 		err = r.store.Storage.GuaranteedUpdate(

staging/src/k8s.io/apiextensions-apiserver/pkg/registry/customresourcedefinition/etcd.go

@@ -92,6 +92,14 @@ func (r *REST) Delete(ctx context.Context, name string, options *metav1.DeleteOp
 		)
 		return nil, false, err
 	}
+	if options.Preconditions.ResourceVersion != nil && *options.Preconditions.ResourceVersion != crd.ResourceVersion {
+		err = apierrors.NewConflict(
+			apiextensions.Resource("customresourcedefinitions"),
+			name,
+			fmt.Errorf("Precondition failed: ResourceVersion in precondition: %v, ResourceVersion in object meta: %v", *options.Preconditions.ResourceVersion, crd.ResourceVersion),
+		)
+		return nil, false, err
+	}
 
 	// upon first request to delete, add our finalizer and then delegate
 	if crd.DeletionTimestamp.IsZero() {
@@ -100,7 +108,7 @@ func (r *REST) Delete(ctx context.Context, name string, options *metav1.DeleteOp
 			return nil, false, err
 		}
 
-		preconditions := storage.Preconditions{UID: options.Preconditions.UID}
+		preconditions := storage.Preconditions{UID: options.Preconditions.UID, ResourceVersion: options.Preconditions.ResourceVersion}
 
 		out := r.Store.NewFunc()
 		err = r.Store.Storage.GuaranteedUpdate(

staging/src/k8s.io/apimachinery/pkg/apis/meta/v1/generated.proto

@@ -754,6 +754,10 @@ message Preconditions {
   // Specifies the target UID.
   // +optional
   optional string uid = 1;
+
+  // Specifies the target ResourceVersion
+  // +optional
+  optional string resourceVersion = 2;
 }
 
 // RootPaths lists the paths available at root.

staging/src/k8s.io/apimachinery/pkg/apis/meta/v1/helpers.go

@@ -228,6 +228,12 @@ func NewUIDPreconditions(uid string) *Preconditions {
 	return &Preconditions{UID: &u}
 }
 
+// NewRVDeletionPrecondition returns a DeleteOptions with a ResourceVersion precondition set.
+func NewRVDeletionPrecondition(rv string) *DeleteOptions {
+	p := Preconditions{ResourceVersion: &rv}
+	return &DeleteOptions{Preconditions: &p}
+}
+
 // HasObjectMetaSystemFieldValues returns true if fields that are managed by the system on ObjectMeta have values.
 func HasObjectMetaSystemFieldValues(meta Object) bool {
 	return !meta.GetCreationTimestamp().Time.IsZero() ||

staging/src/k8s.io/apimachinery/pkg/apis/meta/v1/types.go

@@ -575,6 +575,9 @@ type Preconditions struct {
 	// Specifies the target UID.
 	// +optional
 	UID *types.UID `json:"uid,omitempty" protobuf:"bytes,1,opt,name=uid,casttype=k8s.io/apimachinery/pkg/types.UID"`
+	// Specifies the target ResourceVersion
+	// +optional
+	ResourceVersion *string `json:"resourceVersion,omitempty" protobuf:"bytes,2,opt,name=resourceVersion"`
 }
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

staging/src/k8s.io/apimachinery/pkg/apis/meta/v1/types_swagger_doc_generated.go

@@ -294,8 +294,9 @@ func (PatchOptions) SwaggerDoc() map[string]string {
 }
 
 var map_Preconditions = map[string]string{
-	"":    "Preconditions must be fulfilled before an operation (update, delete, etc.) is carried out.",
-	"uid": "Specifies the target UID.",
+	"":                "Preconditions must be fulfilled before an operation (update, delete, etc.) is carried out.",
+	"uid":             "Specifies the target UID.",
+	"resourceVersion": "Specifies the target ResourceVersion",
 }
 
 func (Preconditions) SwaggerDoc() map[string]string {

staging/src/k8s.io/apimachinery/pkg/apis/meta/v1/zz_generated.deepcopy.go

@@ -831,6 +831,11 @@ func (in *Preconditions) DeepCopyInto(out *Preconditions) {
 		*out = new(types.UID)
 		**out = **in
 	}
+	if in.ResourceVersion != nil {
+		in, out := &in.ResourceVersion, &out.ResourceVersion
+		*out = new(string)
+		**out = **in
+	}
 	return
 }
 

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/fieldmanager/BUILD

@@ -7,14 +7,17 @@ go_library(
     importpath = "k8s.io/apiserver/pkg/endpoints/handlers/fieldmanager",
     visibility = ["//visibility:public"],
     deps = [
+        "//staging/src/k8s.io/apimachinery/pkg/api/errors:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/api/meta:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1/unstructured:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/runtime:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/endpoints/handlers/fieldmanager/internal:go_default_library",
         "//vendor/k8s.io/kube-openapi/pkg/util/proto:go_default_library",
         "//vendor/sigs.k8s.io/structured-merge-diff/fieldpath:go_default_library",
         "//vendor/sigs.k8s.io/structured-merge-diff/merge:go_default_library",
+        "//vendor/sigs.k8s.io/yaml:go_default_library",
     ],
 )
 
@@ -41,6 +44,7 @@ go_test(
     embed = [":go_default_library"],
     deps = [
         "//staging/src/k8s.io/api/core/v1:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/api/errors:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/api/meta:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/runtime:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/fieldmanager/fieldmanager.go

@@ -20,14 +20,17 @@ import (
 	"fmt"
 	"time"
 
+	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apiserver/pkg/endpoints/handlers/fieldmanager/internal"
 	openapiproto "k8s.io/kube-openapi/pkg/util/proto"
 	"sigs.k8s.io/structured-merge-diff/fieldpath"
 	"sigs.k8s.io/structured-merge-diff/merge"
+	"sigs.k8s.io/yaml"
 )
 
 // FieldManager updates the managed fields and merge applied
@@ -149,9 +152,20 @@ func (f *FieldManager) Apply(liveObj runtime.Object, patch []byte, fieldManager
 	if err != nil {
 		return nil, fmt.Errorf("failed to decode managed fields: %v", err)
 	}
-	// We can assume that patchObj is already on the proper version:
-	// it shouldn't have to be converted so that it's not defaulted.
-	// TODO (jennybuckley): Explicitly checkt that patchObj is in the proper version.
+	// Check that the patch object has the same version as the live object
+	patchObj := &unstructured.Unstructured{Object: map[string]interface{}{}}
+
+	if err := yaml.Unmarshal(patch, &patchObj.Object); err != nil {
+		return nil, fmt.Errorf("error decoding YAML: %v", err)
+	}
+	if patchObj.GetAPIVersion() != f.groupVersion.String() {
+		return nil,
+			errors.NewBadRequest(
+				fmt.Sprintf("Incorrect version specified in apply patch. "+
+					"Specified patch version: %s, expected: %s",
+					patchObj.GetAPIVersion(), f.groupVersion.String()))
+	}
+
 	liveObjVersioned, err := f.toVersioned(liveObj)
 	if err != nil {
 		return nil, fmt.Errorf("failed to convert live object to proper version: %v", err)

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/fieldmanager/fieldmanager_test.go

@@ -18,9 +18,11 @@ package fieldmanager_test
 
 import (
 	"errors"
+	"net/http"
 	"testing"
 
 	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -72,7 +74,7 @@ func TestApplyStripsFields(t *testing.T) {
 	obj := &corev1.Pod{}
 
 	newObj, err := f.Apply(obj, []byte(`{
-		"apiVersion": "v1",
+		"apiVersion": "apps/v1",
 		"kind": "Deployment",
 		"metadata": {
 			"name": "b",
@@ -85,7 +87,7 @@ func TestApplyStripsFields(t *testing.T) {
 			"managedFields": [{
 					"manager": "apply",
 					"operation": "Apply",
-					"apiVersion": "v1",
+					"apiVersion": "apps/v1",
 					"fields": {
 						"f:metadata": {
 							"f:labels": {
@@ -111,13 +113,46 @@ func TestApplyStripsFields(t *testing.T) {
 	}
 }
 
+func TestVersionCheck(t *testing.T) {
+	f := NewTestFieldManager(t)
+
+	obj := &corev1.Pod{}
+
+	// patch has 'apiVersion: apps/v1' and live version is apps/v1 -> no errors
+	_, err := f.Apply(obj, []byte(`{
+		"apiVersion": "apps/v1",
+		"kind": "Deployment",
+	}`), "fieldmanager_test", false)
+	if err != nil {
+		t.Fatalf("failed to apply object: %v", err)
+	}
+
+	// patch has 'apiVersion: apps/v2' but live version is apps/v1 -> error
+	_, err = f.Apply(obj, []byte(`{
+		"apiVersion": "apps/v2",
+		"kind": "Deployment",
+	}`), "fieldmanager_test", false)
+	if err == nil {
+		t.Fatalf("expected an error from mismatched patch and live versions")
+	}
+	switch typ := err.(type) {
+	default:
+		t.Fatalf("expected error to be of type %T was %T", apierrors.StatusError{}, typ)
+	case apierrors.APIStatus:
+		if typ.Status().Code != http.StatusBadRequest {
+			t.Fatalf("expected status code to be %d but was %d",
+				http.StatusBadRequest, typ.Status().Code)
+		}
+	}
+}
+
 func TestApplyDoesNotStripLabels(t *testing.T) {
 	f := NewTestFieldManager(t)
 
 	obj := &corev1.Pod{}
 
 	newObj, err := f.Apply(obj, []byte(`{
-		"apiVersion": "v1",
+		"apiVersion": "apps/v1",
 		"kind": "Pod",
 		"metadata": {
 			"labels": {

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/fieldmanager/internal/typeconverter.go

@@ -105,7 +105,12 @@ func (c *typeConverter) YAMLToTyped(from []byte) (typed.TypedValue, error) {
 		return nil, fmt.Errorf("error decoding YAML: %v", err)
 	}
 
-	return c.ObjectToTyped(unstructured)
+	gvk := unstructured.GetObjectKind().GroupVersionKind()
+	t := c.parser.Type(gvk)
+	if t == nil {
+		return nil, fmt.Errorf("no corresponding type for %v", gvk)
+	}
+	return t.FromYAML(typed.YAMLObject(string(from)))
 }
 
 func (c *typeConverter) TypedToObject(value typed.TypedValue) (runtime.Object, error) {

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/fieldmanager/internal/typeconverter_test.go

@@ -68,7 +68,6 @@ metadata:
   labels:
     app: nginx
 spec:
-  replicas: 3
   selector:
     matchLabels:
       app: nginx
@@ -91,7 +90,6 @@ metadata:
   labels:
     app: nginx
 spec:
-  replicas: 3
   selector:
     matchLabels:
       app: nginx

staging/src/k8s.io/apiserver/pkg/registry/generic/registry/store.go

@@ -448,6 +448,7 @@ func (e *Store) Update(ctx context.Context, name string, objInfo rest.UpdatedObj
 	storagePreconditions := &storage.Preconditions{}
 	if preconditions := objInfo.Preconditions(); preconditions != nil {
 		storagePreconditions.UID = preconditions.UID
+		storagePreconditions.ResourceVersion = preconditions.ResourceVersion
 	}
 
 	out := e.NewFunc()
@@ -879,6 +880,7 @@ func (e *Store) Delete(ctx context.Context, name string, options *metav1.DeleteO
 	var preconditions storage.Preconditions
 	if options.Preconditions != nil {
 		preconditions.UID = options.Preconditions.UID
+		preconditions.ResourceVersion = options.Preconditions.ResourceVersion
 	}
 	graceful, pendingGraceful, err := rest.BeforeDelete(e.DeleteStrategy, ctx, obj, options)
 	if err != nil {

staging/src/k8s.io/apiserver/pkg/registry/rest/delete.go

@@ -77,8 +77,13 @@ func BeforeDelete(strategy RESTDeleteStrategy, ctx context.Context, obj runtime.
 		return false, false, errors.NewInvalid(schema.GroupKind{Group: metav1.GroupName, Kind: "DeleteOptions"}, "", errs)
 	}
 	// Checking the Preconditions here to fail early. They'll be enforced later on when we actually do the deletion, too.
-	if options.Preconditions != nil && options.Preconditions.UID != nil && *options.Preconditions.UID != objectMeta.GetUID() {
-		return false, false, errors.NewConflict(schema.GroupResource{Group: gvk.Group, Resource: gvk.Kind}, objectMeta.GetName(), fmt.Errorf("the UID in the precondition (%s) does not match the UID in record (%s). The object might have been deleted and then recreated", *options.Preconditions.UID, objectMeta.GetUID()))
+	if options.Preconditions != nil {
+		if options.Preconditions.UID != nil && *options.Preconditions.UID != objectMeta.GetUID() {
+			return false, false, errors.NewConflict(schema.GroupResource{Group: gvk.Group, Resource: gvk.Kind}, objectMeta.GetName(), fmt.Errorf("the UID in the precondition (%s) does not match the UID in record (%s). The object might have been deleted and then recreated", *options.Preconditions.UID, objectMeta.GetUID()))
+		}
+		if options.Preconditions.ResourceVersion != nil && *options.Preconditions.ResourceVersion != objectMeta.GetResourceVersion() {
+			return false, false, errors.NewConflict(schema.GroupResource{Group: gvk.Group, Resource: gvk.Kind}, objectMeta.GetName(), fmt.Errorf("the ResourceVersion in the precondition (%s) does not match the ResourceVersion in record (%s). The object might have been modified", *options.Preconditions.ResourceVersion, objectMeta.GetResourceVersion()))
+		}
 	}
 	gracefulStrategy, ok := strategy.(RESTGracefulDeleteStrategy)
 	if !ok {

staging/src/k8s.io/apiserver/pkg/registry/rest/resttest/resttest.go

@@ -909,6 +909,45 @@ func (t *Tester) testDeleteWithUID(obj runtime.Object, createFn CreateFunc, getF
 	}
 }
 
+//  This test the fast-fail path. We test that the precondition gets verified
+//  again before deleting the object in tests of pkg/storage/etcd.
+func (t *Tester) testDeleteWithResourceVersion(obj runtime.Object, createFn CreateFunc, getFn GetFunc, isNotFoundFn IsErrorFunc, opts metav1.DeleteOptions) {
+	ctx := t.TestContext()
+
+	foo := obj.DeepCopyObject()
+	t.setObjectMeta(foo, t.namer(1))
+	objectMeta := t.getObjectMetaOrFail(foo)
+	objectMeta.SetResourceVersion("RV0000")
+	if err := createFn(ctx, foo); err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+	opts.Preconditions = metav1.NewRVDeletionPrecondition("RV1111").Preconditions
+	obj, wasDeleted, err := t.storage.(rest.GracefulDeleter).Delete(ctx, objectMeta.GetName(), &opts)
+	if err == nil || !errors.IsConflict(err) {
+		t.Errorf("unexpected error: %v", err)
+	}
+	if wasDeleted {
+		t.Errorf("unexpected, object %s should not have been deleted immediately", objectMeta.GetName())
+	}
+	obj, _, err = t.storage.(rest.GracefulDeleter).Delete(ctx, objectMeta.GetName(), metav1.NewRVDeletionPrecondition("RV0000"))
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	if !t.returnDeletedObject {
+		if status, ok := obj.(*metav1.Status); !ok {
+			t.Errorf("expected status of delete, got %v", status)
+		} else if status.Status != metav1.StatusSuccess {
+			t.Errorf("expected success, got: %v", status.Status)
+		}
+	}
+
+	_, err = getFn(ctx, foo)
+	if err == nil || !isNotFoundFn(err) {
+		t.Errorf("unexpected error: %v", err)
+	}
+}
+
 // =============================================================================
 // Graceful Deletion tests.
 

staging/src/k8s.io/apiserver/pkg/storage/interfaces.go

@@ -98,6 +98,9 @@ type Preconditions struct {
 	// Specifies the target UID.
 	// +optional
 	UID *types.UID `json:"uid,omitempty"`
+	// Specifies the target ResourceVersion
+	// +optional
+	ResourceVersion *string `json:"resourceVersion,omitempty"`
 }
 
 // NewUIDPreconditions returns a Preconditions with UID set.
@@ -125,8 +128,14 @@ func (p *Preconditions) Check(key string, obj runtime.Object) error {
 			objMeta.GetUID())
 		return NewInvalidObjError(key, err)
 	}
+	if p.ResourceVersion != nil && *p.ResourceVersion != objMeta.GetResourceVersion() {
+		err := fmt.Sprintf(
+			"Precondition failed: ResourceVersion in precondition: %v, ResourceVersion in object meta: %v",
+			*p.ResourceVersion,
+			objMeta.GetResourceVersion())
+		return NewInvalidObjError(key, err)
+	}
 	return nil
-
 }
 
 // Interface offers a common interface for object marshaling/unmarshaling operations and

test/e2e/framework/ingress/ingress_utils.go

@@ -667,12 +667,26 @@ func (j *TestJig) pollIngressWithCert(ing *extensions.Ingress, address string, k
 }
 
 // WaitForIngress waits for the Ingress to get an address.
+// WaitForIngress returns when it gets the first 200 response
 func (j *TestJig) WaitForIngress(waitForNodePort bool) {
 	if err := j.WaitForGivenIngressWithTimeout(j.Ingress, waitForNodePort, framework.LoadBalancerPollTimeout); err != nil {
 		framework.Failf("error in waiting for ingress to get an address: %s", err)
 	}
 }
 
+// WaitForIngressToStable waits for the LB return 100 consecutive 200 responses.
+func (j *TestJig) WaitForIngressToStable() {
+	if err := wait.Poll(10*time.Second, framework.LoadBalancerCreateTimeoutDefault, func() (bool, error) {
+		_, err := j.GetDistinctResponseFromIngress()
+		if err != nil {
+			return false, nil
+		}
+		return true, nil
+	}); err != nil {
+		framework.Failf("error in waiting for ingress to stablize: %v", err)
+	}
+}
+
 // WaitForGivenIngressWithTimeout waits till the ingress acquires an IP,
 // then waits for its hosts/urls to respond to a protocol check (either
 // http or https). If waitForNodePort is true, the NodePort of the Service

test/e2e/framework/service_util.go

@@ -623,6 +623,7 @@ func (j *ServiceTestJig) waitForConditionOrFail(namespace, name string, timeout
 // name as the jig and runs the "netexec" container.
 func (j *ServiceTestJig) newRCTemplate(namespace string) *v1.ReplicationController {
 	var replicas int32 = 1
+	var grace int64 = 3 // so we don't race with kube-proxy when scaling up/down
 
 	rc := &v1.ReplicationController{
 		ObjectMeta: metav1.ObjectMeta{
@@ -654,7 +655,7 @@ func (j *ServiceTestJig) newRCTemplate(namespace string) *v1.ReplicationControll
 							},
 						},
 					},
-					TerminationGracePeriodSeconds: new(int64),
+					TerminationGracePeriodSeconds: &grace,
 				},
 			},
 		},
@@ -737,6 +738,28 @@ func (j *ServiceTestJig) RunOrFail(namespace string, tweak func(rc *v1.Replicati
 	return result
 }
 
+func (j *ServiceTestJig) Scale(namespace string, replicas int) {
+	rc := j.Name
+	scale, err := j.Client.CoreV1().ReplicationControllers(namespace).GetScale(rc, metav1.GetOptions{})
+	if err != nil {
+		Failf("Failed to get scale for RC %q: %v", rc, err)
+	}
+
+	scale.Spec.Replicas = int32(replicas)
+	_, err = j.Client.CoreV1().ReplicationControllers(namespace).UpdateScale(rc, scale)
+	if err != nil {
+		Failf("Failed to scale RC %q: %v", rc, err)
+	}
+	pods, err := j.waitForPodsCreated(namespace, replicas)
+	if err != nil {
+		Failf("Failed waiting for pods: %v", err)
+	}
+	if err := j.waitForPodsReady(namespace, pods); err != nil {
+		Failf("Failed waiting for pods to be running: %v", err)
+	}
+	return
+}
+
 func (j *ServiceTestJig) waitForPdbReady(namespace string) error {
 	timeout := 2 * time.Minute
 	for start := time.Now(); time.Since(start) < timeout; time.Sleep(2 * time.Second) {
@@ -875,9 +898,19 @@ func (j *ServiceTestJig) TestReachableHTTP(host string, port int, timeout time.D
 }
 
 func (j *ServiceTestJig) TestReachableHTTPWithRetriableErrorCodes(host string, port int, retriableErrCodes []int, timeout time.Duration) {
-	if err := wait.PollImmediate(Poll, timeout, func() (bool, error) {
-		return TestReachableHTTPWithRetriableErrorCodes(host, port, "/echo?msg=hello", "hello", retriableErrCodes)
-	}); err != nil {
+	pollfn := func() (bool, error) {
+		result := PokeHTTP(host, port, "/echo?msg=hello",
+			&HTTPPokeParams{
+				BodyContains:   "hello",
+				RetriableCodes: retriableErrCodes,
+			})
+		if result.Status == HTTPSuccess {
+			return true, nil
+		}
+		return false, nil // caller can retry
+	}
+
+	if err := wait.PollImmediate(Poll, timeout, pollfn); err != nil {
 		if err == wait.ErrWaitTimeout {
 			Failf("Could not reach HTTP service through %v:%v after %v", host, port, timeout)
 		} else {
@@ -887,36 +920,87 @@ func (j *ServiceTestJig) TestReachableHTTPWithRetriableErrorCodes(host string, p
 }
 
 func (j *ServiceTestJig) TestNotReachableHTTP(host string, port int, timeout time.Duration) {
-	if err := wait.PollImmediate(Poll, timeout, func() (bool, error) { return TestNotReachableHTTP(host, port) }); err != nil {
-		Failf("Could still reach HTTP service through %v:%v after %v: %v", host, port, timeout, err)
+	pollfn := func() (bool, error) {
+		result := PokeHTTP(host, port, "/", nil)
+		if result.Code == 0 {
+			return true, nil
+		}
+		return false, nil // caller can retry
+	}
+
+	if err := wait.PollImmediate(Poll, timeout, pollfn); err != nil {
+		Failf("HTTP service %v:%v reachable after %v: %v", host, port, timeout, err)
+	}
+}
+
+func (j *ServiceTestJig) TestRejectedHTTP(host string, port int, timeout time.Duration) {
+	pollfn := func() (bool, error) {
+		result := PokeHTTP(host, port, "/", nil)
+		if result.Status == HTTPRefused {
+			return true, nil
+		}
+		return false, nil // caller can retry
+	}
+
+	if err := wait.PollImmediate(Poll, timeout, pollfn); err != nil {
+		Failf("HTTP service %v:%v not rejected: %v", host, port, err)
 	}
 }
 
 func (j *ServiceTestJig) TestReachableUDP(host string, port int, timeout time.Duration) {
-	if err := wait.PollImmediate(Poll, timeout, func() (bool, error) { return TestReachableUDP(host, port, "echo hello", "hello") }); err != nil {
+	pollfn := func() (bool, error) {
+		result := PokeUDP(host, port, "echo hello", &UDPPokeParams{
+			Timeout:  3 * time.Second,
+			Response: "hello",
+		})
+		if result.Status == UDPSuccess {
+			return true, nil
+		}
+		return false, nil // caller can retry
+	}
+
+	if err := wait.PollImmediate(Poll, timeout, pollfn); err != nil {
 		Failf("Could not reach UDP service through %v:%v after %v: %v", host, port, timeout, err)
 	}
 }
 
 func (j *ServiceTestJig) TestNotReachableUDP(host string, port int, timeout time.Duration) {
-	if err := wait.PollImmediate(Poll, timeout, func() (bool, error) { return TestNotReachableUDP(host, port, "echo hello") }); err != nil {
-		Failf("Could still reach UDP service through %v:%v after %v: %v", host, port, timeout, err)
+	pollfn := func() (bool, error) {
+		result := PokeUDP(host, port, "echo hello", &UDPPokeParams{Timeout: 3 * time.Second})
+		if result.Status != UDPSuccess && result.Status != UDPError {
+			return true, nil
+		}
+		return false, nil // caller can retry
+	}
+	if err := wait.PollImmediate(Poll, timeout, pollfn); err != nil {
+		Failf("UDP service %v:%v reachable after %v: %v", host, port, timeout, err)
+	}
+}
+
+func (j *ServiceTestJig) TestRejectedUDP(host string, port int, timeout time.Duration) {
+	pollfn := func() (bool, error) {
+		result := PokeUDP(host, port, "echo hello", &UDPPokeParams{Timeout: 3 * time.Second})
+		if result.Status == UDPRefused {
+			return true, nil
+		}
+		return false, nil // caller can retry
+	}
+	if err := wait.PollImmediate(Poll, timeout, pollfn); err != nil {
+		Failf("UDP service %v:%v not rejected: %v", host, port, err)
 	}
 }
 
 func (j *ServiceTestJig) GetHTTPContent(host string, port int, timeout time.Duration, url string) bytes.Buffer {
 	var body bytes.Buffer
-	var err error
 	if pollErr := wait.PollImmediate(Poll, timeout, func() (bool, error) {
-		var result bool
-		result, err = TestReachableHTTPWithContent(host, port, url, "", &body)
-		if err != nil {
-			Logf("Error hitting %v:%v%v, retrying: %v", host, port, url, err)
-			return false, nil
+		result := PokeHTTP(host, port, url, nil)
+		if result.Status == HTTPSuccess {
+			body.Write(result.Body)
+			return true, nil
 		}
-		return result, nil
+		return false, nil
 	}); pollErr != nil {
-		Failf("Could not reach HTTP service through %v:%v%v after %v: %v", host, port, url, timeout, err)
+		Failf("Could not reach HTTP service through %v:%v%v after %v: %v", host, port, url, timeout, pollErr)
 	}
 	return body
 }
@@ -929,7 +1013,7 @@ func testHTTPHealthCheckNodePort(ip string, port int, request string) (bool, err
 		return false, fmt.Errorf("Invalid input ip or port")
 	}
 	Logf("Testing HTTP health check on %v", url)
-	resp, err := httpGetNoConnectionPool(url)
+	resp, err := httpGetNoConnectionPoolTimeout(url, 5*time.Second)
 	if err != nil {
 		Logf("Got error testing for reachability of %s: %v", url, err)
 		return false, err

test/e2e/network/firewall.go

@@ -188,11 +188,11 @@ var _ = SIGDescribe("Firewall rule", func() {
 })
 
 func assertNotReachableHTTPTimeout(ip string, port int, timeout time.Duration) {
-	unreachable, err := framework.TestNotReachableHTTPTimeout(ip, port, timeout)
-	if err != nil {
-		framework.Failf("Unexpected error checking for reachability of %s:%d: %v", ip, port, err)
+	result := framework.PokeHTTP(ip, port, "/", &framework.HTTPPokeParams{Timeout: timeout})
+	if result.Status == framework.HTTPError {
+		framework.Failf("Unexpected error checking for reachability of %s:%d: %v", ip, port, result.Error)
 	}
-	if !unreachable {
+	if result.Code != 0 {
 		framework.Failf("Was unexpectedly able to reach %s:%d", ip, port)
 	}
 }

test/e2e/network/ingress.go

@@ -529,9 +529,10 @@ var _ = SIGDescribe("Loadbalancing: L7", func() {
 				_, err = f.ClientSet.CoreV1().Services(ns).Update(&svc)
 				Expect(err).NotTo(HaveOccurred())
 			}
-			wait.Poll(5*time.Second, framework.LoadBalancerPollTimeout, func() (bool, error) {
-				return gceController.BackendServiceUsingIG(jig.GetServicePorts(true))
+			err = wait.Poll(5*time.Second, framework.LoadBalancerPollTimeout, func() (bool, error) {
+				return gceController.BackendServiceUsingIG(jig.GetServicePorts(false))
 			})
+			Expect(err).NotTo(HaveOccurred(), "Expect backend service to target IG, but failed to observe")
 			jig.WaitForIngress(true)
 
 			By("Switch backend service to use NEG")
@@ -542,9 +543,10 @@ var _ = SIGDescribe("Loadbalancing: L7", func() {
 				_, err = f.ClientSet.CoreV1().Services(ns).Update(&svc)
 				Expect(err).NotTo(HaveOccurred())
 			}
-			wait.Poll(5*time.Second, framework.LoadBalancerPollTimeout, func() (bool, error) {
+			err = wait.Poll(5*time.Second, framework.LoadBalancerPollTimeout, func() (bool, error) {
 				return gceController.BackendServiceUsingNEG(jig.GetServicePorts(false))
 			})
+			Expect(err).NotTo(HaveOccurred(), "Expect backend service to target NEG, but failed to observe")
 			jig.WaitForIngress(true)
 		})
 
@@ -556,7 +558,7 @@ var _ = SIGDescribe("Loadbalancing: L7", func() {
 			svcPorts := jig.GetServicePorts(false)
 			usingNEG, err := gceController.BackendServiceUsingNEG(svcPorts)
 			Expect(err).NotTo(HaveOccurred())
-			Expect(usingNEG).To(BeTrue())
+			Expect(usingNEG).To(BeTrue(), "Expect backend service to be using NEG. But not.")
 
 			// ClusterIP ServicePorts have no NodePort
 			for _, sp := range svcPorts {
@@ -574,18 +576,21 @@ var _ = SIGDescribe("Loadbalancing: L7", func() {
 					_, err = f.ClientSet.AppsV1().Deployments(ns).UpdateScale(name, scale)
 					Expect(err).NotTo(HaveOccurred())
 				}
-				wait.Poll(10*time.Second, negUpdateTimeout, func() (bool, error) {
+				err = wait.Poll(10*time.Second, negUpdateTimeout, func() (bool, error) {
 					res, err := jig.GetDistinctResponseFromIngress()
 					if err != nil {
 						return false, nil
 					}
+					framework.Logf("Expecting %d backends, got %d", num, res.Len())
 					return res.Len() == num, nil
 				})
+				Expect(err).NotTo(HaveOccurred())
 			}
 
 			By("Create a basic HTTP ingress using NEG")
 			jig.CreateIngress(filepath.Join(ingress.IngressManifestPath, "neg"), ns, map[string]string{}, map[string]string{})
 			jig.WaitForIngress(true)
+			jig.WaitForIngressToStable()
 			usingNEG, err := gceController.BackendServiceUsingNEG(jig.GetServicePorts(false))
 			Expect(err).NotTo(HaveOccurred())
 			Expect(usingNEG).To(BeTrue())
@@ -611,6 +616,7 @@ var _ = SIGDescribe("Loadbalancing: L7", func() {
 			By("Create a basic HTTP ingress using NEG")
 			jig.CreateIngress(filepath.Join(ingress.IngressManifestPath, "neg"), ns, map[string]string{}, map[string]string{})
 			jig.WaitForIngress(true)
+			jig.WaitForIngressToStable()
 			usingNEG, err := gceController.BackendServiceUsingNEG(jig.GetServicePorts(false))
 			Expect(err).NotTo(HaveOccurred())
 			Expect(usingNEG).To(BeTrue())
@@ -621,13 +627,15 @@ var _ = SIGDescribe("Loadbalancing: L7", func() {
 			scale.Spec.Replicas = int32(replicas)
 			_, err = f.ClientSet.AppsV1().Deployments(ns).UpdateScale(name, scale)
 			Expect(err).NotTo(HaveOccurred())
-			wait.Poll(10*time.Second, framework.LoadBalancerPollTimeout, func() (bool, error) {
+
+			err = wait.Poll(10*time.Second, framework.LoadBalancerPollTimeout, func() (bool, error) {
 				res, err := jig.GetDistinctResponseFromIngress()
 				if err != nil {
 					return false, nil
 				}
 				return res.Len() == replicas, nil
 			})
+			Expect(err).NotTo(HaveOccurred())
 
 			By("Trigger rolling update and observe service disruption")
 			deploy, err := f.ClientSet.AppsV1().Deployments(ns).Get(name, metav1.GetOptions{})
@@ -637,7 +645,7 @@ var _ = SIGDescribe("Loadbalancing: L7", func() {
 			deploy.Spec.Template.Spec.TerminationGracePeriodSeconds = &gracePeriod
 			_, err = f.ClientSet.AppsV1().Deployments(ns).Update(deploy)
 			Expect(err).NotTo(HaveOccurred())
-			wait.Poll(10*time.Second, framework.LoadBalancerPollTimeout, func() (bool, error) {
+			err = wait.Poll(10*time.Second, framework.LoadBalancerPollTimeout, func() (bool, error) {
 				res, err := jig.GetDistinctResponseFromIngress()
 				Expect(err).NotTo(HaveOccurred())
 				deploy, err := f.ClientSet.AppsV1().Deployments(ns).Get(name, metav1.GetOptions{})
@@ -655,6 +663,7 @@ var _ = SIGDescribe("Loadbalancing: L7", func() {
 					return false, nil
 				}
 			})
+			Expect(err).NotTo(HaveOccurred())
 		})
 
 		It("should sync endpoints for both Ingress-referenced NEG and standalone NEG", func() {
@@ -669,7 +678,7 @@ var _ = SIGDescribe("Loadbalancing: L7", func() {
 					_, err = f.ClientSet.AppsV1().Deployments(ns).UpdateScale(name, scale)
 					Expect(err).NotTo(HaveOccurred())
 				}
-				wait.Poll(10*time.Second, negUpdateTimeout, func() (bool, error) {
+				err = wait.Poll(10*time.Second, negUpdateTimeout, func() (bool, error) {
 					svc, err := f.ClientSet.CoreV1().Services(ns).Get(name, metav1.GetOptions{})
 					Expect(err).NotTo(HaveOccurred())
 
@@ -716,6 +725,7 @@ var _ = SIGDescribe("Loadbalancing: L7", func() {
 
 					return true, nil
 				})
+				Expect(err).NotTo(HaveOccurred())
 			}
 
 			By("Create a basic HTTP ingress using NEG")
@@ -1108,7 +1118,7 @@ func detectHttpVersionAndSchemeTest(f *framework.Framework, jig *ingress.TestJig
 }
 
 func detectNegAnnotation(f *framework.Framework, jig *ingress.TestJig, gceController *gce.GCEIngressController, ns, name string, negs int) {
-	wait.Poll(5*time.Second, framework.LoadBalancerPollTimeout, func() (bool, error) {
+	if err := wait.Poll(5*time.Second, negUpdateTimeout, func() (bool, error) {
 		svc, err := f.ClientSet.CoreV1().Services(ns).Get(name, metav1.GetOptions{})
 		if err != nil {
 			return false, nil
@@ -1150,5 +1160,7 @@ func detectNegAnnotation(f *framework.Framework, jig *ingress.TestJig, gceContro
 		}
 
 		return gceController.BackendServiceUsingNEG(jig.GetServicePorts(false))
-	})
+	}); err != nil {
+		Expect(err).NotTo(HaveOccurred())
+	}
 }

test/e2e/network/service.go

@@ -791,11 +791,47 @@ var _ = SIGDescribe("Services", func() {
 		jig.TestReachableUDP(nodeIP, udpNodePort, framework.KubeProxyLagTimeout)
 
 		By("hitting the TCP service's LoadBalancer")
-		jig.TestReachableHTTP(tcpIngressIP, svcPort, loadBalancerCreateTimeout) // this may actually recreate the LB
+		jig.TestReachableHTTP(tcpIngressIP, svcPort, loadBalancerCreateTimeout)
 
 		if loadBalancerSupportsUDP {
 			By("hitting the UDP service's LoadBalancer")
-			jig.TestReachableUDP(udpIngressIP, svcPort, loadBalancerCreateTimeout) // this may actually recreate the LB)
+			jig.TestReachableUDP(udpIngressIP, svcPort, loadBalancerCreateTimeout)
+		}
+
+		By("Scaling the pods to 0")
+		jig.Scale(ns1, 0)
+		jig.Scale(ns2, 0)
+
+		By("looking for ICMP REJECT on the TCP service's NodePort")
+		jig.TestRejectedHTTP(nodeIP, tcpNodePort, framework.KubeProxyLagTimeout)
+
+		By("looking for ICMP REJECT on the UDP service's NodePort")
+		jig.TestRejectedUDP(nodeIP, udpNodePort, framework.KubeProxyLagTimeout)
+
+		By("looking for ICMP REJECT on the TCP service's LoadBalancer")
+		jig.TestRejectedHTTP(tcpIngressIP, svcPort, loadBalancerCreateTimeout)
+
+		if loadBalancerSupportsUDP {
+			By("looking for ICMP REJECT on the UDP service's LoadBalancer")
+			jig.TestRejectedUDP(udpIngressIP, svcPort, loadBalancerCreateTimeout)
+		}
+
+		By("Scaling the pods to 1")
+		jig.Scale(ns1, 1)
+		jig.Scale(ns2, 1)
+
+		By("hitting the TCP service's NodePort")
+		jig.TestReachableHTTP(nodeIP, tcpNodePort, framework.KubeProxyLagTimeout)
+
+		By("hitting the UDP service's NodePort")
+		jig.TestReachableUDP(nodeIP, udpNodePort, framework.KubeProxyLagTimeout)
+
+		By("hitting the TCP service's LoadBalancer")
+		jig.TestReachableHTTP(tcpIngressIP, svcPort, loadBalancerCreateTimeout)
+
+		if loadBalancerSupportsUDP {
+			By("hitting the UDP service's LoadBalancer")
+			jig.TestReachableUDP(udpIngressIP, svcPort, loadBalancerCreateTimeout)
 		}
 
 		// Change the services back to ClusterIP.
@@ -2063,14 +2099,18 @@ var _ = SIGDescribe("ESIPP [Slow] [DisabledForLargeClusters]", func() {
 		for nodeName, nodeIPs := range endpointNodeMap {
 			By(fmt.Sprintf("checking kube-proxy health check fails on node with endpoint (%s), public IP %s", nodeName, nodeIPs[0]))
 			var body bytes.Buffer
-			var result bool
-			var err error
-			if pollErr := wait.PollImmediate(framework.Poll, framework.ServiceTestTimeout, func() (bool, error) {
-				result, err = framework.TestReachableHTTPWithContent(nodeIPs[0], healthCheckNodePort, "/healthz", "", &body)
-				return !result, nil
-			}); pollErr != nil {
-				framework.Failf("Kube-proxy still exposing health check on node %v:%v, after ESIPP was turned off. Last err %v, last body %v",
-					nodeName, healthCheckNodePort, err, body.String())
+			pollfn := func() (bool, error) {
+				result := framework.PokeHTTP(nodeIPs[0], healthCheckNodePort, "/healthz", nil)
+				if result.Code == 0 {
+					return true, nil
+				}
+				body.Reset()
+				body.Write(result.Body)
+				return false, nil
+			}
+			if pollErr := wait.PollImmediate(framework.Poll, framework.ServiceTestTimeout, pollfn); pollErr != nil {
+				framework.Failf("Kube-proxy still exposing health check on node %v:%v, after ESIPP was turned off. body %s",
+					nodeName, healthCheckNodePort, body.String())
 			}
 		}
 

test/e2e/storage/csi_mock_volume.go

@@ -379,7 +379,7 @@ var _ = utils.SIGDescribe("CSI mock volume", func() {
 			_, _, pod3 := createPod()
 			Expect(pod3).NotTo(BeNil(), "while creating third pod")
 			err = waitForMaxVolumeCondition(pod3, m.cs)
-			Expect(err).NotTo(HaveOccurred(), "while waiting for max volume condition")
+			Expect(err).NotTo(HaveOccurred(), "while waiting for max volume condition on pod : %+v", pod3)
 		})
 	})
 

test/e2e/storage/drivers/csi.go

@@ -367,7 +367,7 @@ func (g *gcePDCSIDriver) SkipUnsupportedTest(pattern testpatterns.TestPattern) {
 	if pattern.FsType == "xfs" {
 		framework.SkipUnlessNodeOSDistroIs("ubuntu", "custom")
 	}
-	if pattern.FeatureTag == "sig-windows" {
+	if pattern.FeatureTag == "[sig-windows]" {
 		framework.Skipf("Skipping tests for windows since CSI does not support it yet")
 	}
 }
@@ -473,7 +473,7 @@ func (g *gcePDExternalCSIDriver) SkipUnsupportedTest(pattern testpatterns.TestPa
 	if pattern.FsType == "xfs" {
 		framework.SkipUnlessNodeOSDistroIs("ubuntu", "custom")
 	}
-	if pattern.FeatureTag == "sig-windows" {
+	if pattern.FeatureTag == "[sig-windows]" {
 		framework.Skipf("Skipping tests for windows since CSI does not support it yet")
 	}
 }

test/e2e/storage/drivers/in_tree.go

@@ -1106,21 +1106,18 @@ var _ testsuites.DynamicPVTestDriver = &gcePdDriver{}
 
 // InitGcePdDriver returns gcePdDriver that implements TestDriver interface
 func InitGcePdDriver() testsuites.TestDriver {
-	var supportedTypes sets.String
-	var capFsGroup bool
-	if framework.NodeOSDistroIs("windows") {
-		supportedTypes = sets.NewString("ntfs")
-		capFsGroup = false
-	} else {
-		supportedTypes = sets.NewString(
-			"", // Default fsType
-			"ext2",
-			"ext3",
-			"ext4",
-			"xfs",
-		)
-		capFsGroup = true
-	}
+	// In current test structure, it first initialize the driver and then set up
+	// the new framework, so we cannot get the correct OS here. So here set to
+	// support all fs types including both linux and windows. We have code to check Node OS later
+	// during test.
+	supportedTypes := sets.NewString(
+		"", // Default fsType
+		"ext2",
+		"ext3",
+		"ext4",
+		"xfs",
+		"ntfs",
+	)
 	return &gcePdDriver{
 		driverInfo: testsuites.DriverInfo{
 			Name:                 "gcepd",
@@ -1129,7 +1126,7 @@ func InitGcePdDriver() testsuites.TestDriver {
 			SupportedMountOption: sets.NewString("debug", "nouid32"),
 			Capabilities: map[testsuites.Capability]bool{
 				testsuites.CapPersistence: true,
-				testsuites.CapFsGroup:     capFsGroup,
+				testsuites.CapFsGroup:     true,
 				testsuites.CapBlock:       true,
 				testsuites.CapExec:        true,
 			},
@@ -1143,7 +1140,7 @@ func (g *gcePdDriver) GetDriverInfo() *testsuites.DriverInfo {
 
 func (g *gcePdDriver) SkipUnsupportedTest(pattern testpatterns.TestPattern) {
 	framework.SkipUnlessProviderIs("gce", "gke")
-	if pattern.FeatureTag == "sig-windows" {
+	if pattern.FeatureTag == "[sig-windows]" {
 		framework.SkipUnlessNodeOSDistroIs("windows")
 	}
 }

test/e2e/storage/testpatterns/testpattern.go

@@ -152,21 +152,21 @@ var (
 		Name:       "Inline-volume (ntfs)",
 		VolType:    InlineVolume,
 		FsType:     "ntfs",
-		FeatureTag: "sig-windows",
+		FeatureTag: "[sig-windows]",
 	}
 	// NtfsPreprovisionedPV is TestPattern for "Pre-provisioned PV (ntfs)"
 	NtfsPreprovisionedPV = TestPattern{
 		Name:       "Pre-provisioned PV (ntfs)",
 		VolType:    PreprovisionedPV,
 		FsType:     "ntfs",
-		FeatureTag: "sig-windows",
+		FeatureTag: "[sig-windows]",
 	}
 	// NtfsDynamicPV is TestPattern for "Dynamic PV (xfs)"
 	NtfsDynamicPV = TestPattern{
 		Name:       "Dynamic PV (ntfs)",
 		VolType:    DynamicPV,
 		FsType:     "ntfs",
-		FeatureTag: "sig-windows",
+		FeatureTag: "[sig-windows]",
 	}
 
 	// Definitions for Filesystem volume mode

test/e2e/storage/testsuites/volume_io.go

@@ -125,7 +125,7 @@ func (t *volumeIOTestSuite) defineTests(driver TestDriver, pattern testpatterns.
 		fileSizes := createFileSizes(dInfo.MaxFileSize)
 		testFile := fmt.Sprintf("%s_io_test_%s", dInfo.Name, f.Namespace.Name)
 		var fsGroup *int64
-		if dInfo.Capabilities[CapFsGroup] {
+		if !framework.NodeOSDistroIs("windows") && dInfo.Capabilities[CapFsGroup] {
 			fsGroupVal := int64(1234)
 			fsGroup = &fsGroupVal
 		}

test/e2e/storage/testsuites/volumes.go

@@ -152,7 +152,7 @@ func (t *volumesTestSuite) defineTests(driver TestDriver, pattern testpatterns.T
 		}
 		config := convertTestConfig(l.config)
 		var fsGroup *int64
-		if dInfo.Capabilities[CapFsGroup] {
+		if framework.NodeOSDistroIs("windows") && dInfo.Capabilities[CapFsGroup] {
 			fsGroupVal := int64(1234)
 			fsGroup = &fsGroupVal
 		}
@@ -185,7 +185,12 @@ func testScriptInPod(
 	)
 	suffix := generateSuffixForPodName(volumeType)
 	fileName := fmt.Sprintf("test-%s", suffix)
-	content := fmt.Sprintf("ls %s", volPath)
+	var content string
+	if framework.NodeOSDistroIs("windows") {
+		content = fmt.Sprintf("ls -n %s", volPath)
+	} else {
+		content = fmt.Sprintf("ls %s", volPath)
+	}
 	command := framework.GenerateWriteandExecuteScriptFileCmd(content, fileName, volPath)
 	pod := &v1.Pod{
 		ObjectMeta: metav1.ObjectMeta{

test/e2e_node/device_plugin.go

@@ -45,17 +45,12 @@ const (
 )
 
 // Serial because the test restarts Kubelet
-var _ = framework.KubeDescribe("Device Plugin [Feature:DevicePlugin][NodeFeature:DevicePlugin][Serial]", func() {
-	f := framework.NewDefaultFramework("device-plugin-errors")
-	testDevicePlugin(f, false, pluginapi.DevicePluginPath)
-})
-
 var _ = framework.KubeDescribe("Device Plugin [Feature:DevicePluginProbe][NodeFeature:DevicePluginProbe][Serial]", func() {
 	f := framework.NewDefaultFramework("device-plugin-errors")
-	testDevicePlugin(f, true, "/var/lib/kubelet/plugins_registry")
+	testDevicePlugin(f, "/var/lib/kubelet/plugins_registry")
 })
 
-func testDevicePlugin(f *framework.Framework, enablePluginWatcher bool, pluginSockDir string) {
+func testDevicePlugin(f *framework.Framework, pluginSockDir string) {
 	pluginSockDir = filepath.Join(pluginSockDir) + "/"
 	Context("DevicePlugin", func() {
 		By("Enabling support for Kubelet Plugins Watcher")
@@ -63,7 +58,6 @@ func testDevicePlugin(f *framework.Framework, enablePluginWatcher bool, pluginSo
 			if initialConfig.FeatureGates == nil {
 				initialConfig.FeatureGates = map[string]bool{}
 			}
-			initialConfig.FeatureGates[string(features.KubeletPluginsWatcher)] = enablePluginWatcher
 			initialConfig.FeatureGates[string(features.KubeletPodResources)] = true
 		})
 		It("Verifies the Kubelet device plugin functionality.", func() {

test/integration/apiserver/apply/apply_test.go

@@ -119,6 +119,19 @@ func TestApplyAlsoCreates(t *testing.T) {
 		if err != nil {
 			t.Fatalf("Failed to retrieve object: %v", err)
 		}
+
+		// Test that we can re apply with a different field manager and don't get conflicts
+		_, err = client.CoreV1().RESTClient().Patch(types.ApplyPatchType).
+			Namespace("default").
+			Resource(tc.resource).
+			Name(tc.name).
+			Param("fieldManager", "apply_test_2").
+			Body([]byte(tc.body)).
+			Do().
+			Get()
+		if err != nil {
+			t.Fatalf("Failed to re-apply object using Apply patch: %v", err)
+		}
 	}
 }
 

test/integration/framework/BUILD

@@ -4,6 +4,7 @@ load(
     "@io_bazel_rules_go//go:def.bzl",
     "go_library",
 )
+load("//build:platforms.bzl", "go_platform_constraint")
 
 go_library(
     name = "go_default_library",
@@ -15,9 +16,23 @@ go_library(
         "test_server.go",
         "util.go",
     ],
-    data = [
-        "@com_coreos_etcd//:etcd",
-    ],
+    data = select({
+        go_platform_constraint(
+            arch = "arm64",
+            os = "linux",
+        ): [
+            "@com_coreos_etcd_arm64//:etcd",
+        ],
+        go_platform_constraint(
+            arch = "ppc64le",
+            os = "linux",
+        ): [
+            "@com_coreos_etcd_ppc64le//:etcd",
+        ],
+        "//conditions:default": [
+            "@com_coreos_etcd_amd64//:etcd",
+        ],
+    }),
     importpath = "k8s.io/kubernetes/test/integration/framework",
     deps = [
         "//cmd/kube-apiserver/app:go_default_library",

test/integration/framework/etcd.go

@@ -24,6 +24,7 @@ import (
 	"os"
 	"os/exec"
 	"path/filepath"
+	"runtime"
 	"strings"
 
 	"k8s.io/klog"
@@ -43,7 +44,8 @@ You can use 'hack/install-etcd.sh' to install a copy in third_party/.
 
 // getEtcdPath returns a path to an etcd executable.
 func getEtcdPath() (string, error) {
-	bazelPath := filepath.Join(os.Getenv("RUNFILES_DIR"), "com_coreos_etcd/etcd")
+	bazelPath := filepath.Join(os.Getenv("RUNFILES_DIR"), fmt.Sprintf("com_coreos_etcd_%s", runtime.GOARCH), "etcd")
+
 	p, err := exec.LookPath(bazelPath)
 	if err == nil {
 		return p, nil