Add autoregistration package to manage APIService.

Signed-off-by: Matt Moyer <moyerm@vmware.com>

go.mod

@@ -13,5 +13,7 @@ require (
 	k8s.io/apimachinery v0.19.0-rc.0
 	k8s.io/apiserver v0.19.0-rc.0
 	k8s.io/client-go v0.19.0-rc.0
+	k8s.io/kube-aggregator v0.19.0-rc.0
+	k8s.io/utils v0.0.0-20200619165400-6e3d28b6ed19
 	sigs.k8s.io/yaml v1.2.0
 )

go.sum

@@ -101,6 +101,7 @@ github.com/emicklei/go-restful v0.0.0-20170410110728-ff4f55a20633/go.mod h1:otzb
 github.com/emicklei/go-restful v2.9.5+incompatible/go.mod h1:otzb+WCGbkyDHkqmQmT5YD2WR4BBwUdeQoFo8l/7tVs=
 github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
+github.com/evanphx/json-patch v0.0.0-20190815234213-e83c0a1c26c8 h1:DM7gHzQfHwIj+St8zaPOI6iQEPAxOwIkskvw6s9rDaM=
 github.com/evanphx/json-patch v0.0.0-20190815234213-e83c0a1c26c8/go.mod h1:pmLOTb3x90VhIKxsA9yeQG5yfOkkKnkk1h+Ql8NDYDw=
 github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
 github.com/fatih/color v1.9.0 h1:8xPHl4/q1VyqGIPif1F+1V3Y3lSmrq01EabUW3CoW5s=
@@ -302,6 +303,8 @@ github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCV
 github.com/json-iterator/go v1.1.7/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
 github.com/json-iterator/go v1.1.9 h1:9yzud/Ht36ygwatGx56VwCZtlI/2AD15T1X2sjSuGns=
 github.com/json-iterator/go v1.1.9/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
+github.com/json-iterator/go v1.1.10 h1:Kz6Cvnvv2wGdaG/V8yMvfkmNiXq9Ya2KUv4rouJJr68=
+github.com/json-iterator/go v1.1.10/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
 github.com/jstemmer/go-junit-report v0.0.0-20190106144839-af01ea7f8024/go.mod h1:6v2b51hI/fHJwM22ozAgKL4VKDeJcHhJFhtBdhmNjmU=
 github.com/jstemmer/go-junit-report v0.9.1/go.mod h1:Brl9GWCQeLvo8nXZwPNNblvFj/XSXhF0NWZEnDohbsk=
 github.com/jtolds/gls v4.20.0+incompatible h1:xdiiI2gbIgH/gLH7ADydsJ1uDOEzR8yvV7C0MuV77Wo=
@@ -715,6 +718,7 @@ golang.org/x/tools v0.0.0-20200414032229-332987a829c3/go.mod h1:EkVYQZoAsY45+roY
 golang.org/x/tools v0.0.0-20200422022333-3d57cf2e726e/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
 golang.org/x/tools v0.0.0-20200428185508-e9a00ec82136/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
 golang.org/x/tools v0.0.0-20200519015757-0d0afa43d58a/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
+golang.org/x/tools v0.0.0-20200602230032-c00d67ef29d0/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20200625211823-6506e20df31f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
 golang.org/x/tools v0.0.0-20200702044944-0cc1aa72b347 h1:/e4fNMHdLn7SQSxTrRZTma2xjQW6ELdxcnpqMhpo9X4=
 golang.org/x/tools v0.0.0-20200702044944-0cc1aa72b347/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
@@ -817,12 +821,17 @@ k8s.io/apiserver v0.19.0-rc.0 h1:SaF/gMgUeDPbQDKHTMvB2yynBUZpp6s4HYQIOx/LdDQ=
 k8s.io/apiserver v0.19.0-rc.0/go.mod h1:yEjU524zw/pxiG6nOsgY5Hu/akAg7tH/J/tKrLUp/mo=
 k8s.io/client-go v0.19.0-rc.0 h1:6WW8MElhoLeYcLiN4ky1159XG5E39KYdmLCrV/6lNiE=
 k8s.io/client-go v0.19.0-rc.0/go.mod h1:3kWGD05F7c58atlk7ep9ob1hg2Yu9NSz8gJxCNNTHhc=
+k8s.io/code-generator v0.19.0-rc.0/go.mod h1:2jgaU9hVSqti1GiO69UFSoTZcL5XAvZSrXaNnK5RVA0=
 k8s.io/component-base v0.19.0-rc.0 h1:S/jt6xey1Wg5i5A9/BCkPYekpjJ5zlfuSCCVlNSJ/Yc=
 k8s.io/component-base v0.19.0-rc.0/go.mod h1:8cHxNUQdeDIIcORXOrMABUPbuEmbbHRtEweSSk8Il4g=
 k8s.io/gengo v0.0.0-20200413195148-3a45101e95ac/go.mod h1:ezvh/TsK7cY6rbqRK0oQQ8IAqLxYwwyPxAX1Pzy0ii0=
+k8s.io/gengo v0.0.0-20200428234225-8167cfdcfc14/go.mod h1:ezvh/TsK7cY6rbqRK0oQQ8IAqLxYwwyPxAX1Pzy0ii0=
 k8s.io/klog/v2 v2.0.0/go.mod h1:PBfzABfn139FHAV07az/IF9Wp1bkk3vpT2XSJ76fSDE=
 k8s.io/klog/v2 v2.2.0 h1:XRvcwJozkgZ1UQJmfMGpvRthQHOvihEhYtDfAaxMz/A=
 k8s.io/klog/v2 v2.2.0/go.mod h1:Od+F08eJP+W3HUb4pSrPpgp9DGU4GzlpG/TmITuYh/Y=
+k8s.io/kube-aggregator v0.19.0-rc.0 h1:+u9y1c0R2GF8fuaEnlJrdUtxoEmQOON98oatycSquOA=
+k8s.io/kube-aggregator v0.19.0-rc.0/go.mod h1:DCq8Korz9XUEZVsq0wAGIAyJW79xdcYhIBtvWNTsTkc=
+k8s.io/kube-openapi v0.0.0-20200427153329-656914f816f9 h1:5NC2ITmvg8RoxoH0wgmL4zn4VZqXGsKbxrikjaQx6s4=
 k8s.io/kube-openapi v0.0.0-20200427153329-656914f816f9/go.mod h1:bfCVj+qXcEaE5SCvzBaqpOySr6tuCcpPKqF6HD8nyCw=
 k8s.io/utils v0.0.0-20200619165400-6e3d28b6ed19 h1:7Nu2dTj82c6IaWvL7hImJzcXoTPz1MsSCH7r+0m6rfo=
 k8s.io/utils v0.0.0-20200619165400-6e3d28b6ed19/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=

internal/autoregistration/autoregistration.go

@@ -0,0 +1,138 @@
+/*
+Copyright 2020 VMware, Inc.
+SPDX-License-Identifier: Apache-2.0
+*/
+
+// Package autoregistration registers a Kubernetes APIService pointing at the current pod.
+package autoregistration
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	corev1 "k8s.io/api/core/v1"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	corev1client "k8s.io/client-go/kubernetes/typed/core/v1"
+	"k8s.io/client-go/util/retry"
+	apiregistrationv1 "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1"
+	aggregatationv1client "k8s.io/kube-aggregator/pkg/client/clientset_generated/clientset"
+)
+
+// ErrInvalidServiceTemplate is returned by Setup when the provided ServiceTemplate is not valid.
+var ErrInvalidServiceTemplate = errors.New("invalid service template")
+
+// SetupOptions specifies the inputs for Setup().
+type SetupOptions struct {
+	CoreV1             corev1client.CoreV1Interface
+	AggregationV1      aggregatationv1client.Interface
+	Namespace          string
+	ServiceTemplate    corev1.Service
+	APIServiceTemplate apiregistrationv1.APIService
+}
+
+// Setup registers a Kubernetes Service, and an aggregation APIService which points to it.
+func Setup(ctx context.Context, options SetupOptions) error {
+	// Get the namespace so we can use its UID set owner references on other objects.
+	ns, err := options.CoreV1.Namespaces().Get(ctx, options.Namespace, metav1.GetOptions{})
+	if err != nil {
+		return fmt.Errorf("could not get namespace: %w", err)
+	}
+
+	// Make a copy of the Service template.
+	svc := options.ServiceTemplate.DeepCopy()
+	svc.Namespace = ns.Name
+
+	// Validate that the Service meets our expectations.
+	if len(svc.Spec.Ports) != 1 {
+		return fmt.Errorf("%w: must have 1 port (found %d)", ErrInvalidServiceTemplate, len(svc.Spec.Ports))
+	}
+	if port := svc.Spec.Ports[0]; port.Protocol != corev1.ProtocolTCP || port.Port != 443 {
+		return fmt.Errorf("%w: must expose TCP/443 (found %s/%d)", ErrInvalidServiceTemplate, port.Protocol, port.Port)
+	}
+
+	// Create or update the Service.
+	if err := createOrUpdateService(ctx, options.CoreV1, svc); err != nil {
+		return err
+	}
+
+	apiSvc := options.APIServiceTemplate.DeepCopy()
+	apiSvc.Spec.Service = &apiregistrationv1.ServiceReference{
+		Namespace: ns.Name,
+		Name:      svc.Name,
+		Port:      &svc.Spec.Ports[0].Port,
+	}
+	apiSvc.ObjectMeta.OwnerReferences = []metav1.OwnerReference{{
+		APIVersion: ns.APIVersion,
+		Kind:       ns.Kind,
+		UID:        ns.UID,
+		Name:       ns.Name,
+	}}
+	if err := createOrUpdateAPIService(ctx, options.AggregationV1, apiSvc); err != nil {
+		return err
+	}
+	return nil
+}
+
+func createOrUpdateService(ctx context.Context, client corev1client.CoreV1Interface, svc *corev1.Service) error {
+	services := client.Services(svc.Namespace)
+
+	_, err := services.Create(ctx, svc, metav1.CreateOptions{})
+	if err == nil {
+		return nil
+	}
+	if !k8serrors.IsAlreadyExists(err) {
+		return fmt.Errorf("could not create service: %w", err)
+	}
+
+	if err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
+		// Retrieve the latest version of the Service before attempting update
+		// RetryOnConflict uses exponential backoff to avoid exhausting the apiserver
+		result, err := services.Get(ctx, svc.Name, metav1.GetOptions{})
+		if err != nil {
+			return fmt.Errorf("could not get existing version of service: %w", err)
+		}
+
+		// Update just the fields we care about.
+		result.Spec.Ports = svc.Spec.Ports
+		result.Spec.Selector = svc.Spec.Selector
+
+		_, updateErr := services.Update(ctx, result, metav1.UpdateOptions{})
+		return updateErr
+	}); err != nil {
+		return fmt.Errorf("could not update service: %w", err)
+	}
+	return nil
+}
+
+func createOrUpdateAPIService(ctx context.Context, client aggregatationv1client.Interface, apiSvc *apiregistrationv1.APIService) error {
+	apiServices := client.ApiregistrationV1().APIServices()
+
+	_, err := apiServices.Create(ctx, apiSvc, metav1.CreateOptions{})
+	if err == nil {
+		return nil
+	}
+	if !k8serrors.IsAlreadyExists(err) {
+		return fmt.Errorf("could not create API service: %w", err)
+	}
+
+	if err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
+		// Retrieve the latest version of the Service before attempting update
+		// RetryOnConflict uses exponential backoff to avoid exhausting the apiserver
+		result, err := apiServices.Get(ctx, apiSvc.Name, metav1.GetOptions{})
+		if err != nil {
+			return fmt.Errorf("could not get existing version of API service: %w", err)
+		}
+
+		// Update just the fields we care about.
+		apiSvc.Spec.DeepCopyInto(&result.Spec)
+		apiSvc.OwnerReferences = result.OwnerReferences
+
+		_, updateErr := apiServices.Update(ctx, result, metav1.UpdateOptions{})
+		return updateErr
+	}); err != nil {
+		return fmt.Errorf("could not update API service: %w", err)
+	}
+	return nil
+}

internal/autoregistration/autoregistration_test.go

@@ -0,0 +1,540 @@
+/*
+Copyright 2020 VMware, Inc.
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package autoregistration
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/intstr"
+	kubefake "k8s.io/client-go/kubernetes/fake"
+	kubetesting "k8s.io/client-go/testing"
+	apiregistrationv1 "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1"
+	aggregationv1fake "k8s.io/kube-aggregator/pkg/client/clientset_generated/clientset/fake"
+	"k8s.io/utils/pointer"
+)
+
+func TestSetup(t *testing.T) {
+	tests := []struct {
+		name            string
+		input           SetupOptions
+		mocks           func(*kubefake.Clientset, *aggregationv1fake.Clientset)
+		wantErr         string
+		wantServices    []corev1.Service
+		wantAPIServices []apiregistrationv1.APIService
+	}{
+		{
+			name: "no such namespace",
+			input: SetupOptions{
+				Namespace: "foo",
+			},
+			wantErr: `could not get namespace: namespaces "foo" not found`,
+		},
+		{
+			name: "service template missing port",
+			input: SetupOptions{
+				Namespace: "test-namespace",
+			},
+			mocks: func(kube *kubefake.Clientset, agg *aggregationv1fake.Clientset) {
+				_ = kube.Tracker().Add(&corev1.Namespace{
+					ObjectMeta: metav1.ObjectMeta{Name: "test-namespace", UID: "test-namespace-uid"},
+				})
+			},
+			wantErr: `invalid service template: must have 1 port (found 0)`,
+		},
+		{
+			name: "service template missing port",
+			input: SetupOptions{
+				Namespace: "test-namespace",
+				ServiceTemplate: corev1.Service{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "test-service",
+						Namespace: "replaceme",
+					},
+					Spec: corev1.ServiceSpec{
+						Ports: []corev1.ServicePort{
+							{
+								Protocol:   "UDP",
+								Port:       1234,
+								TargetPort: intstr.IntOrString{IntVal: 1234},
+							},
+						},
+					},
+				},
+			},
+			mocks: func(kube *kubefake.Clientset, agg *aggregationv1fake.Clientset) {
+				_ = kube.Tracker().Add(&corev1.Namespace{
+					ObjectMeta: metav1.ObjectMeta{Name: "test-namespace", UID: "test-namespace-uid"},
+				})
+			},
+			wantErr: `invalid service template: must expose TCP/443 (found UDP/1234)`,
+		},
+		{
+			name: "fail to create service",
+			input: SetupOptions{
+				Namespace: "test-namespace",
+				ServiceTemplate: corev1.Service{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "test-service",
+						Namespace: "replaceme",
+					},
+					Spec: corev1.ServiceSpec{
+						Ports: []corev1.ServicePort{
+							{
+								Protocol:   "TCP",
+								Port:       443,
+								TargetPort: intstr.IntOrString{IntVal: 1234},
+							},
+						},
+					},
+				},
+			},
+			mocks: func(kube *kubefake.Clientset, agg *aggregationv1fake.Clientset) {
+				_ = kube.Tracker().Add(&corev1.Namespace{
+					ObjectMeta: metav1.ObjectMeta{Name: "test-namespace", UID: "test-namespace-uid"},
+				})
+				kube.PrependReactor("create", "services", func(_ kubetesting.Action) (bool, runtime.Object, error) {
+					return true, nil, fmt.Errorf("some Service creation failure")
+				})
+			},
+			wantErr: `could not create service: some Service creation failure`,
+		},
+		{
+			name: "fail to create API service",
+			input: SetupOptions{
+				Namespace: "test-namespace",
+				ServiceTemplate: corev1.Service{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "test-service",
+						Namespace: "replaceme",
+					},
+					Spec: corev1.ServiceSpec{
+						Ports: []corev1.ServicePort{
+							{
+								Protocol:   "TCP",
+								Port:       443,
+								TargetPort: intstr.IntOrString{IntVal: 1234},
+							},
+						},
+					},
+				},
+			},
+			mocks: func(kube *kubefake.Clientset, agg *aggregationv1fake.Clientset) {
+				_ = kube.Tracker().Add(&corev1.Namespace{
+					ObjectMeta: metav1.ObjectMeta{Name: "test-namespace", UID: "test-namespace-uid"},
+				})
+				agg.PrependReactor("create", "apiservices", func(_ kubetesting.Action) (bool, runtime.Object, error) {
+					return true, nil, fmt.Errorf("some APIService creation failure")
+				})
+			},
+			wantErr: `could not create API service: some APIService creation failure`,
+		},
+		{
+			name: "success",
+			input: SetupOptions{
+				Namespace: "test-namespace",
+				ServiceTemplate: corev1.Service{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "test-service",
+						Namespace: "replaceme",
+					},
+					Spec: corev1.ServiceSpec{
+						Ports: []corev1.ServicePort{
+							{
+								Protocol:   "TCP",
+								Port:       443,
+								TargetPort: intstr.IntOrString{IntVal: 1234},
+							},
+						},
+					},
+				},
+				APIServiceTemplate: apiregistrationv1.APIService{
+					ObjectMeta: metav1.ObjectMeta{Name: "test-api-service"},
+					Spec: apiregistrationv1.APIServiceSpec{
+						Group:                "test-api-group",
+						Version:              "test-version",
+						CABundle:             []byte("test-ca-bundle"),
+						GroupPriorityMinimum: 1234,
+						VersionPriority:      4321,
+					},
+				},
+			},
+			mocks: func(kube *kubefake.Clientset, agg *aggregationv1fake.Clientset) {
+				_ = kube.Tracker().Add(&corev1.Namespace{
+					TypeMeta:   metav1.TypeMeta{APIVersion: "v1", Kind: "Namespace"},
+					ObjectMeta: metav1.ObjectMeta{Name: "test-namespace", UID: "test-namespace-uid"},
+				})
+			},
+			wantServices: []corev1.Service{{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-service",
+					Namespace: "test-namespace",
+				},
+				Spec: corev1.ServiceSpec{
+					Ports: []corev1.ServicePort{
+						{
+							Protocol:   "TCP",
+							Port:       443,
+							TargetPort: intstr.IntOrString{IntVal: 1234},
+						},
+					},
+				},
+			}},
+			wantAPIServices: []apiregistrationv1.APIService{{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "test-api-service",
+					OwnerReferences: []metav1.OwnerReference{{
+						APIVersion: "v1",
+						Kind:       "Namespace",
+						Name:       "test-namespace",
+						UID:        "test-namespace-uid",
+					}},
+				},
+				Spec: apiregistrationv1.APIServiceSpec{
+					Service: &apiregistrationv1.ServiceReference{
+						Namespace: "test-namespace",
+						Name:      "test-service",
+						Port:      pointer.Int32Ptr(443),
+					},
+					Group:                "test-api-group",
+					Version:              "test-version",
+					CABundle:             []byte("test-ca-bundle"),
+					GroupPriorityMinimum: 1234,
+					VersionPriority:      4321,
+				},
+			}},
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			ctx := context.Background()
+
+			kubeClient := kubefake.NewSimpleClientset()
+			aggregationClient := aggregationv1fake.NewSimpleClientset()
+			if tt.mocks != nil {
+				tt.mocks(kubeClient, aggregationClient)
+			}
+
+			tt.input.CoreV1 = kubeClient.CoreV1()
+			tt.input.AggregationV1 = aggregationClient
+			err := Setup(context.Background(), tt.input)
+			if tt.wantErr != "" {
+				require.EqualError(t, err, tt.wantErr)
+				return
+			}
+			require.NoError(t, err)
+			if tt.wantServices != nil {
+				objects, err := kubeClient.CoreV1().Services(tt.input.Namespace).List(ctx, metav1.ListOptions{})
+				require.NoError(t, err)
+				require.Equal(t, tt.wantServices, objects.Items)
+			}
+			if tt.wantAPIServices != nil {
+				objects, err := aggregationClient.ApiregistrationV1().APIServices().List(ctx, metav1.ListOptions{})
+				require.NoError(t, err)
+				require.Equal(t, tt.wantAPIServices, objects.Items)
+			}
+		})
+	}
+}
+
+func TestCreateOrUpdateService(t *testing.T) {
+	tests := []struct {
+		name        string
+		input       *corev1.Service
+		mocks       func(*kubefake.Clientset)
+		wantObjects []corev1.Service
+		wantErr     string
+	}{
+		{
+			name: "error on create",
+			input: &corev1.Service{
+				ObjectMeta: metav1.ObjectMeta{Name: "foo", Namespace: "ns"},
+				Spec: corev1.ServiceSpec{
+					Type:      corev1.ServiceTypeClusterIP,
+					ClusterIP: "1.2.3.4",
+				},
+			},
+			mocks: func(c *kubefake.Clientset) {
+				c.PrependReactor("create", "services", func(_ kubetesting.Action) (bool, runtime.Object, error) {
+					return true, nil, fmt.Errorf("error on create")
+				})
+			},
+			wantErr: "could not create service: error on create",
+		},
+		{
+			name: "new",
+			input: &corev1.Service{
+				ObjectMeta: metav1.ObjectMeta{Name: "foo", Namespace: "ns"},
+				Spec: corev1.ServiceSpec{
+					Type:      corev1.ServiceTypeClusterIP,
+					ClusterIP: "1.2.3.4",
+				},
+			},
+			wantObjects: []corev1.Service{{
+				ObjectMeta: metav1.ObjectMeta{Name: "foo", Namespace: "ns"},
+				Spec: corev1.ServiceSpec{
+					Type:      corev1.ServiceTypeClusterIP,
+					ClusterIP: "1.2.3.4",
+				},
+			}},
+		},
+		{
+			name: "update",
+			mocks: func(c *kubefake.Clientset) {
+				_ = c.Tracker().Add(&corev1.Service{
+					ObjectMeta: metav1.ObjectMeta{Name: "foo", Namespace: "ns"},
+					Spec: corev1.ServiceSpec{
+						Type:      corev1.ServiceTypeClusterIP,
+						ClusterIP: "1.2.3.4",
+					},
+				})
+			},
+			input: &corev1.Service{
+				ObjectMeta: metav1.ObjectMeta{Name: "foo", Namespace: "ns"},
+				Spec: corev1.ServiceSpec{
+					Type:      corev1.ServiceTypeClusterIP,
+					ClusterIP: "1.2.3.4",
+				},
+			},
+			wantObjects: []corev1.Service{{
+				ObjectMeta: metav1.ObjectMeta{Name: "foo", Namespace: "ns"},
+				Spec: corev1.ServiceSpec{
+					Type:      corev1.ServiceTypeClusterIP,
+					ClusterIP: "1.2.3.4",
+				},
+			}},
+		},
+		{
+			name: "error on get",
+			mocks: func(c *kubefake.Clientset) {
+				_ = c.Tracker().Add(&corev1.Service{
+					ObjectMeta: metav1.ObjectMeta{Name: "foo"},
+					Spec: corev1.ServiceSpec{
+						Type:      corev1.ServiceTypeClusterIP,
+						ClusterIP: "1.2.3.4",
+					},
+				})
+				c.PrependReactor("get", "services", func(_ kubetesting.Action) (bool, runtime.Object, error) {
+					return true, nil, fmt.Errorf("error on get")
+				})
+			},
+			input: &corev1.Service{
+				ObjectMeta: metav1.ObjectMeta{Name: "foo"},
+				Spec: corev1.ServiceSpec{
+					Type:      corev1.ServiceTypeClusterIP,
+					ClusterIP: "1.2.3.4",
+				},
+			},
+			wantErr: "could not update service: could not get existing version of service: error on get",
+		},
+		{
+			name: "error on get, successful retry",
+			mocks: func(c *kubefake.Clientset) {
+				_ = c.Tracker().Add(&corev1.Service{
+					ObjectMeta: metav1.ObjectMeta{Name: "foo"},
+					Spec: corev1.ServiceSpec{
+						Type:      corev1.ServiceTypeClusterIP,
+						ClusterIP: "1.2.3.4",
+					},
+				})
+
+				hit := false
+				c.PrependReactor("get", "services", func(_ kubetesting.Action) (bool, runtime.Object, error) {
+					// Return an error on the first call, then fall through to the default (successful) response.
+					if !hit {
+						hit = true
+						return true, nil, fmt.Errorf("error on get")
+					}
+					return false, nil, nil
+				})
+			},
+			input: &corev1.Service{
+				ObjectMeta: metav1.ObjectMeta{Name: "foo"},
+				Spec: corev1.ServiceSpec{
+					Type:      corev1.ServiceTypeClusterIP,
+					ClusterIP: "1.2.3.4",
+				},
+			},
+			wantErr: "could not update service: could not get existing version of service: error on get",
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			ctx := context.Background()
+
+			client := kubefake.NewSimpleClientset()
+			if tt.mocks != nil {
+				tt.mocks(client)
+			}
+
+			err := createOrUpdateService(ctx, client.CoreV1(), tt.input)
+
+			if tt.wantErr != "" {
+				require.EqualError(t, err, tt.wantErr)
+				return
+			}
+			require.NoError(t, err)
+			if tt.wantObjects != nil {
+				objects, err := client.CoreV1().Services(tt.input.ObjectMeta.Namespace).List(ctx, metav1.ListOptions{})
+				require.NoError(t, err)
+				require.Equal(t, tt.wantObjects, objects.Items)
+			}
+		})
+	}
+}
+
+func TestCreateOrUpdateAPIService(t *testing.T) {
+	tests := []struct {
+		name        string
+		input       *apiregistrationv1.APIService
+		mocks       func(*aggregationv1fake.Clientset)
+		wantObjects []apiregistrationv1.APIService
+		wantErr     string
+	}{
+		{
+			name: "error on create",
+			input: &apiregistrationv1.APIService{
+				ObjectMeta: metav1.ObjectMeta{Name: "foo"},
+				Spec: apiregistrationv1.APIServiceSpec{
+					GroupPriorityMinimum: 123,
+					VersionPriority:      456,
+				},
+			},
+			mocks: func(c *aggregationv1fake.Clientset) {
+				c.PrependReactor("create", "apiservices", func(_ kubetesting.Action) (bool, runtime.Object, error) {
+					return true, nil, fmt.Errorf("error on create")
+				})
+			},
+			wantErr: "could not create API service: error on create",
+		},
+		{
+			name: "new",
+			input: &apiregistrationv1.APIService{
+				ObjectMeta: metav1.ObjectMeta{Name: "foo"},
+				Spec: apiregistrationv1.APIServiceSpec{
+					GroupPriorityMinimum: 123,
+					VersionPriority:      456,
+				},
+			},
+			wantObjects: []apiregistrationv1.APIService{{
+				ObjectMeta: metav1.ObjectMeta{Name: "foo"},
+				Spec: apiregistrationv1.APIServiceSpec{
+					GroupPriorityMinimum: 123,
+					VersionPriority:      456,
+				},
+			}},
+		},
+		{
+			name: "update",
+			mocks: func(c *aggregationv1fake.Clientset) {
+				_ = c.Tracker().Add(&apiregistrationv1.APIService{
+					ObjectMeta: metav1.ObjectMeta{Name: "foo"},
+					Spec: apiregistrationv1.APIServiceSpec{
+						GroupPriorityMinimum: 999,
+						VersionPriority:      999,
+					},
+				})
+			},
+			input: &apiregistrationv1.APIService{
+				ObjectMeta: metav1.ObjectMeta{Name: "foo"},
+				Spec: apiregistrationv1.APIServiceSpec{
+					GroupPriorityMinimum: 123,
+					VersionPriority:      456,
+				},
+			},
+			wantObjects: []apiregistrationv1.APIService{{
+				ObjectMeta: metav1.ObjectMeta{Name: "foo"},
+				Spec: apiregistrationv1.APIServiceSpec{
+					GroupPriorityMinimum: 123,
+					VersionPriority:      456,
+				},
+			}},
+		},
+		{
+			name: "error on get",
+			mocks: func(c *aggregationv1fake.Clientset) {
+				_ = c.Tracker().Add(&apiregistrationv1.APIService{
+					ObjectMeta: metav1.ObjectMeta{Name: "foo"},
+					Spec: apiregistrationv1.APIServiceSpec{
+						GroupPriorityMinimum: 999,
+						VersionPriority:      999,
+					},
+				})
+				c.PrependReactor("get", "apiservices", func(_ kubetesting.Action) (bool, runtime.Object, error) {
+					return true, nil, fmt.Errorf("error on get")
+				})
+			},
+			input: &apiregistrationv1.APIService{
+				ObjectMeta: metav1.ObjectMeta{Name: "foo"},
+				Spec: apiregistrationv1.APIServiceSpec{
+					GroupPriorityMinimum: 123,
+					VersionPriority:      456,
+				},
+			},
+			wantErr: "could not update API service: could not get existing version of API service: error on get",
+		},
+		{
+			name: "error on get, successful retry",
+			mocks: func(c *aggregationv1fake.Clientset) {
+				_ = c.Tracker().Add(&apiregistrationv1.APIService{
+					ObjectMeta: metav1.ObjectMeta{Name: "foo"},
+					Spec: apiregistrationv1.APIServiceSpec{
+						GroupPriorityMinimum: 999,
+						VersionPriority:      999,
+					},
+				})
+
+				hit := false
+				c.PrependReactor("get", "apiservices", func(_ kubetesting.Action) (bool, runtime.Object, error) {
+					// Return an error on the first call, then fall through to the default (successful) response.
+					if !hit {
+						hit = true
+						return true, nil, fmt.Errorf("error on get")
+					}
+					return false, nil, nil
+				})
+			},
+			input: &apiregistrationv1.APIService{
+				ObjectMeta: metav1.ObjectMeta{Name: "foo"},
+				Spec: apiregistrationv1.APIServiceSpec{
+					GroupPriorityMinimum: 123,
+					VersionPriority:      456,
+				},
+			},
+			wantErr: "could not update API service: could not get existing version of API service: error on get",
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			ctx := context.Background()
+
+			client := aggregationv1fake.NewSimpleClientset()
+			if tt.mocks != nil {
+				tt.mocks(client)
+			}
+
+			err := createOrUpdateAPIService(ctx, client, tt.input)
+			if tt.wantErr != "" {
+				require.EqualError(t, err, tt.wantErr)
+				return
+			}
+			require.NoError(t, err)
+			if tt.wantObjects != nil {
+				objects, err := client.ApiregistrationV1().APIServices().List(ctx, metav1.ListOptions{})
+				require.NoError(t, err)
+				require.Equal(t, tt.wantObjects, objects.Items)
+			}
+		})
+	}
+}