Handle APIService speically when ServerSide apply is enabled

go.sum

@@ -703,10 +703,6 @@ sigs.k8s.io/controller-runtime v0.6.0 h1:Fzna3DY7c4BIP6KwfSlrfnj20DJ+SeMBK8HSFvO
 sigs.k8s.io/controller-runtime v0.6.0/go.mod h1:CpYf5pdNY/B352A1TFLAS2JVSlnGQ5O2cftPHndTroo=
 sigs.k8s.io/kustomize v2.0.3+incompatible h1:JUufWFNlI44MdtnjUqVnvh29rR37PQFzPbLXqhyOyX0=
 sigs.k8s.io/kustomize v2.0.3+incompatible/go.mod h1:MkjgH3RdOWrievjo6c9T245dYlB5QeXV4WCbnt/PEpU=
-sigs.k8s.io/kustomize/kyaml v0.10.6 h1:xUJxc/k8JoWqHUahaB8DTqY0KwEPxTbTGStvW8TOcDc=
-sigs.k8s.io/kustomize/kyaml v0.10.6/go.mod h1:K9yg1k/HB/6xNOf5VH3LhTo1DK9/5ykSZO5uIv+Y/1k=
-sigs.k8s.io/kustomize/kyaml v0.10.7 h1:r0r8UEL0bL7X56HKUmhJZ+TP+nvRNGrDHHSLO7izlcQ=
-sigs.k8s.io/kustomize/kyaml v0.10.7/go.mod h1:K9yg1k/HB/6xNOf5VH3LhTo1DK9/5ykSZO5uIv+Y/1k=
 sigs.k8s.io/kustomize/kyaml v0.10.9 h1:n3WNdvPPReRNDxW+XXd2JlyZ8EII721I21D1DBpBVBE=
 sigs.k8s.io/kustomize/kyaml v0.10.9/go.mod h1:K9yg1k/HB/6xNOf5VH3LhTo1DK9/5ykSZO5uIv+Y/1k=
 sigs.k8s.io/structured-merge-diff/v3 v3.0.0-20200116222232-67a7b8c61874/go.mod h1:PlARxl6Hbt/+BC80dRLi1qAmnMqwqDg62YvvVkZjemw=

pkg/apply/task/apply_task.go

@@ -6,6 +6,7 @@ package task
 import (
 	"context"
 	"io/ioutil"
+	"strings"
 
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
@@ -184,6 +185,12 @@ func (a *ApplyTask) Start(taskContext *taskrunner.TaskContext) {
 			ao.SetObjects([]*resource.Info{info})
 			klog.V(5).Infof("applying %s/%s...", info.Namespace, info.Name)
 			err = ao.Run()
+			if err != nil && a.ServerSideOptions.ServerSideApply && isAPIService(obj) && isStreamError(err) {
+				// Server-side Apply doesn't work with APIService before k8s 1.21
+				// https://github.com/kubernetes/kubernetes/issues/89264
+				// Thus APIService is handled specially using client-side apply.
+				err = clientSideApply(info, taskContext.EventChannel(), a.DryRunStrategy, a.Factory)
+			}
 			if err != nil {
 				if klog.V(4) {
 					klog.Errorf("error applying (%s/%s) %s", info.Namespace, info.Name, err)
@@ -388,3 +395,23 @@ func sendBatchApplyEvents(taskContext *taskrunner.TaskContext, objects []*unstru
 		taskContext.CaptureResourceFailure(id)
 	}
 }
+
+func isAPIService(obj *unstructured.Unstructured) bool {
+	gk := obj.GroupVersionKind().GroupKind()
+	return gk.Group == "apiregistration.k8s.io" && gk.Kind == "APIService"
+}
+
+// isStreamError checks if the error is a StreamError. Since kubectl wraps the actual StreamError,
+// we can't check the error type.
+func isStreamError(err error) bool {
+	return strings.Contains(err.Error(), "stream error: stream ID ")
+}
+
+func clientSideApply(info *resource.Info, eventChannel chan event.Event, strategy common.DryRunStrategy, factory util.Factory) error {
+	ao, _, err := applyOptionsFactoryFunc(eventChannel, common.ServerSideOptions{ServerSideApply: false}, strategy, factory)
+	if err != nil {
+		return err
+	}
+	ao.SetObjects([]*resource.Info{info})
+	return ao.Run()
+}

test/e2e/common_test.go

@@ -128,6 +128,30 @@ func deploymentManifest(namespace string) *unstructured.Unstructured {
 	}
 }
 
+func apiserviceManifest() *unstructured.Unstructured {
+	apiservice := &unstructured.Unstructured{
+		Object: map[string]interface{}{
+			"apiVersion": "apiregistration.k8s.io/v1",
+			"kind":       "APIService",
+			"metadata": map[string]interface{}{
+				"name": "v1beta1.custom.metrics.k8s.io",
+			},
+			"spec": map[string]interface{}{
+				"insecureSkipTLSVerify": true,
+				"group":                 "custom.metrics.k8s.io",
+				"groupPriorityMinimum":  100,
+				"versionPriority":       100,
+				"service": map[string]interface{}{
+					"name":      "custom-metrics-stackdriver-adapter",
+					"namespace": "custome-metrics",
+				},
+				"version": "v1beta1",
+			},
+		},
+	}
+	return apiservice
+}
+
 func manifestToUnstructured(manifest []byte) *unstructured.Unstructured {
 	u := make(map[string]interface{})
 	err := yaml.Unmarshal(manifest, &u)

test/e2e/e2e_test.go

@@ -18,6 +18,7 @@ import (
 	"k8s.io/kubectl/pkg/cmd/util"
 	"k8s.io/kubectl/pkg/scheme"
 	"sigs.k8s.io/cli-utils/pkg/apply"
+	"sigs.k8s.io/cli-utils/pkg/common"
 	"sigs.k8s.io/cli-utils/pkg/inventory"
 	"sigs.k8s.io/cli-utils/pkg/provider"
 	"sigs.k8s.io/cli-utils/pkg/util/factory"
@@ -109,6 +110,10 @@ var _ = Describe("Applier", func() {
 				It("Apply continues on error", func() {
 					continueOnErrorTest(c, invConfig, inventoryName, namespace.GetName())
 				})
+
+				It("Server-Side Apply", func() {
+					serversideApplyTest(c, invConfig, inventoryName, namespace.GetName())
+				})
 			})
 
 			Context("Inventory policy", func() {
@@ -205,7 +210,7 @@ func defaultInvSizeVerifyFunc(c client.Client, name, namespace string, count int
 
 func defaultInvCountVerifyFunc(c client.Client, namespace string, count int) {
 	var cmList v1.ConfigMapList
-	err := c.List(context.TODO(), &cmList, client.InNamespace(namespace))
+	err := c.List(context.TODO(), &cmList, client.InNamespace(namespace), client.HasLabels{common.InventoryLabel})
 	Expect(err).NotTo(HaveOccurred())
 	Expect(len(cmList.Items)).To(Equal(count))
 }

test/e2e/serverside_apply_test.go

@@ -0,0 +1,71 @@
+// Copyright 2020 The Kubernetes Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package e2e
+
+import (
+	"context"
+	"time"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	appsv1 "k8s.io/api/apps/v1"
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/cli-utils/pkg/apply"
+	"sigs.k8s.io/cli-utils/pkg/common"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+func serversideApplyTest(c client.Client, invConfig InventoryConfig, inventoryName, namespaceName string) {
+	By("Apply a Deployment and an APIService by server-side apply")
+	applier := invConfig.ApplierFactoryFunc()
+
+	inv := invConfig.InvWrapperFunc(invConfig.InventoryFactoryFunc(inventoryName, namespaceName, "test"))
+	firstResources := []*unstructured.Unstructured{
+		deploymentManifest(namespaceName),
+		apiserviceManifest(),
+	}
+
+	runWithNoErr(applier.Run(context.TODO(), inv, firstResources, apply.Options{
+		ReconcileTimeout: 2 * time.Minute,
+		EmitStatusEvents: true,
+		ServerSideOptions: common.ServerSideOptions{
+			ServerSideApply: true,
+			ForceConflicts:  true,
+			FieldManager:    "test",
+		},
+	}))
+
+	By("Verify deployment is server-side applied")
+	var d appsv1.Deployment
+	err := c.Get(context.TODO(), types.NamespacedName{
+		Namespace: namespaceName,
+		Name:      deploymentManifest(namespaceName).GetName(),
+	}, &d)
+	Expect(err).NotTo(HaveOccurred())
+	_, found := d.ObjectMeta.Annotations[v1.LastAppliedConfigAnnotation]
+	Expect(found).To(Equal(false))
+	fields := d.GetManagedFields()
+	Expect(fields[0].Manager).To(Equal("test"))
+
+	By("Verify APIService is client-side applied")
+	var apiService = &unstructured.Unstructured{}
+	apiService.SetGroupVersionKind(
+		schema.GroupVersionKind{
+			Group:   "apiregistration.k8s.io",
+			Version: "v1",
+			Kind:    "APIService",
+		},
+	)
+	err = c.Get(context.TODO(), types.NamespacedName{
+		Name: "v1beta1.custom.metrics.k8s.io",
+	}, apiService)
+	Expect(err).NotTo(HaveOccurred())
+	_, found2 := apiService.GetAnnotations()[v1.LastAppliedConfigAnnotation]
+	Expect(found2).To(Equal(true))
+	fields2 := apiService.GetManagedFields()
+	Expect(len(fields2)).To(Equal(0))
+}