admission webhook server

Author: Mengqi Yu

example/main.go

@@ -19,12 +19,17 @@ package main
 import (
 	"context"
 	"flag"
+	"fmt"
+	"net/http"
 	"os"
 
 	"github.com/go-logr/logr"
+
+	admissionregistrationv1beta1 "k8s.io/api/admissionregistration/v1beta1"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
+	apitypes "k8s.io/apimachinery/pkg/types"
 	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/config"
@@ -35,11 +40,13 @@ import (
 	logf "sigs.k8s.io/controller-runtime/pkg/runtime/log"
 	"sigs.k8s.io/controller-runtime/pkg/runtime/signals"
 	"sigs.k8s.io/controller-runtime/pkg/source"
+	"sigs.k8s.io/controller-runtime/pkg/webhook"
+	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
+	"sigs.k8s.io/controller-runtime/pkg/webhook/admission/builder"
+	"sigs.k8s.io/controller-runtime/pkg/webhook/types"
 )
 
-var (
-	log = logf.Log.WithName("example-controller")
-)
+var log = logf.Log.WithName("example-controller")
 
 func main() {
 	flag.Parse()
@@ -75,6 +82,64 @@ func main() {
 		os.Exit(1)
 	}
 
+	// Setup webhooks
+	mutatingWebhook, err := builder.NewWebhookBuilder().
+		Name("mutating.k8s.io").
+		Type(types.WebhookTypeMutating).
+		Path("/mutating-pods").
+		Operations(admissionregistrationv1beta1.Create, admissionregistrationv1beta1.Update).
+		WithManager(mgr).
+		ForType(&corev1.Pod{}).
+		Build(&podAnnotator{client: mgr.GetClient(), decoder: mgr.GetAdmissionDecoder()})
+	if err != nil {
+		entryLog.Error(err, "unable to setup mutating webhook")
+		os.Exit(1)
+	}
+
+	validatingWebhook, err := builder.NewWebhookBuilder().
+		Name("validating.k8s.io").
+		Type(types.WebhookTypeValidating).
+		Path("/validating-pods").
+		Operations(admissionregistrationv1beta1.Create, admissionregistrationv1beta1.Update).
+		WithManager(mgr).
+		ForType(&corev1.Pod{}).
+		Build(&podValidator{client: mgr.GetClient(), decoder: mgr.GetAdmissionDecoder()})
+	if err != nil {
+		entryLog.Error(err, "unable to setup validating webhook")
+		os.Exit(1)
+	}
+
+	as, err := webhook.NewServer("foo-admission-server", mgr, webhook.ServerOptions{
+		Port:    443,
+		CertDir: "/tmp/cert",
+		Client:  mgr.GetClient(),
+		KVMap:   map[string]interface{}{"foo": "bar"},
+		BootstrapOptions: &webhook.BootstrapOptions{
+			Secret: &apitypes.NamespacedName{
+				Namespace: "default",
+				Name:      "foo-admission-server-secret",
+			},
+
+			Service: &apitypes.NamespacedName{
+				Namespace: "default",
+				Name:      "foo-admission-server-service",
+			},
+			// Labels should select the pods that runs this webhook server.
+			Labels: map[string]string{
+				"app": "foo-admission-server",
+			},
+		},
+	})
+	if err != nil {
+		entryLog.Error(err, "unable to create a new webhook server")
+		os.Exit(1)
+	}
+	err = as.Register(mutatingWebhook, validatingWebhook)
+	if err != nil {
+		entryLog.Error(err, "unable to register webhooks in the admission server")
+		os.Exit(1)
+	}
+
 	if err := mgr.Start(signals.SetupSignalHandler()); err != nil {
 		entryLog.Error(err, "unable to run manager")
 		os.Exit(1)
@@ -83,6 +148,7 @@ func main() {
 
 // reconcileReplicaSet reconciles ReplicaSets
 type reconcileReplicaSet struct {
+	// client can be used to retrieve objects from the APIServer.
 	client client.Client
 	log    logr.Logger
 }
@@ -128,3 +194,73 @@ func (r *reconcileReplicaSet) Reconcile(request reconcile.Request) (reconcile.Re
 
 	return reconcile.Result{}, nil
 }
+
+// podAnnotator annotates Pods
+type podAnnotator struct {
+	client  client.Client
+	decoder admission.Decoder
+}
+
+// Implement admission.Handler so the controller can handle admission request.
+var _ admission.Handler = &podAnnotator{}
+
+// podAnnotator adds an annotation to every incoming pods.
+func (a *podAnnotator) Handle(_ context.Context, req admission.Request) admission.Response {
+	pod := &corev1.Pod{}
+
+	err := a.decoder.Decode(req, pod)
+	if err != nil {
+		return admission.ErrorResponse(http.StatusBadRequest, err)
+	}
+	copy := pod.DeepCopy()
+
+	err = mutatePodsFn(copy)
+	if err != nil {
+		return admission.ErrorResponse(http.StatusInternalServerError, err)
+	}
+	return admission.PatchResponse(pod, copy)
+}
+
+// mutatePodsFn add an annotation to the given pod
+func mutatePodsFn(pod *corev1.Pod) error {
+	anno := pod.GetAnnotations()
+	anno["example-mutating-admission-webhhok"] = "foo"
+	pod.SetAnnotations(anno)
+	return nil
+}
+
+// podValidator validates Pods
+type podValidator struct {
+	client  client.Client
+	decoder admission.Decoder
+}
+
+// Implement admission.Handler so the controller can handle admission request.
+var _ admission.Handler = &podValidator{}
+
+// podValidator admits a pod iff a specific annotation exists.
+func (v *podValidator) Handle(_ context.Context, req admission.Request) admission.Response {
+	pod := &corev1.Pod{}
+
+	err := v.decoder.Decode(req, pod)
+	if err != nil {
+		return admission.ErrorResponse(http.StatusBadRequest, err)
+	}
+
+	allowed, reason, err := validatePodsFn(pod)
+	if err != nil {
+		return admission.ErrorResponse(http.StatusInternalServerError, err)
+	}
+	return admission.ValidationResponse(allowed, reason)
+}
+
+func validatePodsFn(pod *corev1.Pod) (bool, string, error) {
+	anno := pod.GetAnnotations()
+	key := "example-mutating-admission-webhhok"
+	_, found := anno[key]
+	if found {
+		return found, "", nil
+	} else {
+		return found, fmt.Sprintf("failed to find annotation with key: %v", key), nil
+	}
+}

pkg/client/interfaces.go

@@ -19,6 +19,7 @@ package client
 import (
 	"context"
 
+	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/labels"
@@ -29,6 +30,15 @@ import (
 // ObjectKey identifies a Kubernetes Object.
 type ObjectKey = types.NamespacedName
 
+// ObjectKeyFromObject returns the ObjectKey given a runtime.Object
+func ObjectKeyFromObject(obj runtime.Object) (ObjectKey, error) {
+	accessor, err := meta.Accessor(obj)
+	if err != nil {
+		return ObjectKey{}, err
+	}
+	return ObjectKey{Namespace: accessor.GetNamespace(), Name: accessor.GetName()}, nil
+}
+
 // TODO(directxman12): is there a sane way to deal with get/delete options?
 
 // Reader knows how to read and list Kubernetes objects.

pkg/manager/internal.go

@@ -19,6 +19,7 @@ package manager
 import (
 	"sync"
 
+	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/record"
@@ -27,6 +28,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/recorder"
 	"sigs.k8s.io/controller-runtime/pkg/runtime/inject"
 	logf "sigs.k8s.io/controller-runtime/pkg/runtime/log"
+	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
 )
 
 var log = logf.KBLog.WithName("manager")
@@ -38,6 +40,8 @@ type controllerManager struct {
 	// scheme is the scheme injected into Controllers, EventHandlers, Sources and Predicates.  Defaults
 	// to scheme.scheme.
 	scheme *runtime.Scheme
+	// admissionDecoder is used to decode an admission.Request.
+	admissionDecoder admission.Decoder
 
 	// runnables is the set of Controllers that the controllerManager injects deps into and Starts.
 	runnables []Runnable
@@ -56,6 +60,9 @@ type controllerManager struct {
 	// (and EventHandlers, Sources and Predicates).
 	recorderProvider recorder.Provider
 
+	// mapper is used to map resources to kind, and map kind and version.
+	mapper meta.RESTMapper
+
 	mu      sync.Mutex
 	started bool
 	errChan chan error
@@ -120,6 +127,10 @@ func (cm *controllerManager) GetScheme() *runtime.Scheme {
 	return cm.scheme
 }
 
+func (cm *controllerManager) GetAdmissionDecoder() admission.Decoder {
+	return cm.admissionDecoder
+}
+
 func (cm *controllerManager) GetFieldIndexer() client.FieldIndexer {
 	return cm.fieldIndexes
 }
@@ -132,6 +143,10 @@ func (cm *controllerManager) GetRecorder(name string) record.EventRecorder {
 	return cm.recorderProvider.GetEventRecorderFor(name)
 }
 
+func (cm *controllerManager) GetRESTMapper() meta.RESTMapper {
+	return cm.mapper
+}
+
 func (cm *controllerManager) Start(stop <-chan struct{}) error {
 	func() {
 		cm.mu.Lock()

pkg/manager/manager.go

@@ -31,6 +31,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
 	internalrecorder "sigs.k8s.io/controller-runtime/pkg/internal/recorder"
 	"sigs.k8s.io/controller-runtime/pkg/recorder"
+	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
 )
 
 // Manager initializes shared dependencies such as Caches and Clients, and provides them to Runnables.
@@ -55,6 +56,9 @@ type Manager interface {
 	// GetScheme returns and initialized Scheme
 	GetScheme() *runtime.Scheme
 
+	// GetAdmissionDecoder returns the runtime.Decoder based on the scheme.
+	GetAdmissionDecoder() admission.Decoder
+
 	// GetClient returns a client configured with the Config
 	GetClient() client.Client
 
@@ -66,6 +70,9 @@ type Manager interface {
 
 	// GetRecorder returns a new EventRecorder for the provided name
 	GetRecorder(name string) record.EventRecorder
+
+	// GetRESTMapper returns a RESTMapper
+	GetRESTMapper() meta.RESTMapper
 }
 
 // Options are the arguments for creating a new Manager
@@ -87,6 +94,7 @@ type Options struct {
 	newCache            func(config *rest.Config, opts cache.Options) (cache.Cache, error)
 	newClient           func(config *rest.Config, options client.Options) (client.Client, error)
 	newRecorderProvider func(config *rest.Config, scheme *runtime.Scheme, logger logr.Logger) (recorder.Provider, error)
+	newAdmissionDecoder func(scheme *runtime.Scheme) (admission.Decoder, error)
 }
 
 // Runnable allows a component to be started.
@@ -140,14 +148,21 @@ func New(config *rest.Config, options Options) (Manager, error) {
 		return nil, err
 	}
 
+	admissionDecoder, err := options.newAdmissionDecoder(options.Scheme)
+	if err != nil {
+		return nil, err
+	}
+
 	return &controllerManager{
 		config:           config,
 		scheme:           options.Scheme,
+		admissionDecoder: admissionDecoder,
 		errChan:          make(chan error),
 		cache:            cache,
 		fieldIndexes:     cache,
 		client:           client.DelegatingClient{Reader: cache, Writer: writeObj, StatusClient: writeObj},
 		recorderProvider: recorderProvider,
+		mapper:           mapper,
 	}, nil
 }
 
@@ -177,5 +192,9 @@ func setOptionsDefaults(options Options) Options {
 		options.newRecorderProvider = internalrecorder.NewProvider
 	}
 
+	if options.newAdmissionDecoder == nil {
+		options.newAdmissionDecoder = admission.NewDecoder
+	}
+
 	return options
 }

pkg/patch/doc.go

@@ -0,0 +1,33 @@
+/*
+Copyright 2018 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+/*
+Package patch provides method to calculate JSON patch between 2 k8s objects.
+
+Calculate JSON patch
+
+	oldDeployment := appsv1.Deployment{
+		// some fields
+	}
+	newDeployment := appsv1.Deployment{
+		// some different fields
+	}
+	patch, err := NewJSONPatch(oldDeployment, newDeployment)
+	if err != nil {
+		// handle error
+	}
+*/
+package patch