Aggregator: Add mutation and validation handlers (#92036)

pkg/aggregator/apiserver/plugin/admission.go

@@ -0,0 +1,127 @@
+package plugin
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+
+	"github.com/grafana/grafana-plugin-sdk-go/backend"
+	"github.com/grafana/grafana/pkg/aggregator/apiserver/plugin/admission"
+	"github.com/grafana/grafana/pkg/aggregator/apiserver/util"
+	grafanasemconv "github.com/grafana/grafana/pkg/semconv"
+	"k8s.io/component-base/tracing"
+	"k8s.io/klog/v2"
+)
+
+func (h *PluginHandler) AdmissionMutationHandler() http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		ctx := r.Context()
+		span := tracing.SpanFromContext(ctx)
+		span.AddEvent("AdmissionMutationHandler")
+		responder := &util.Responder{ResponseWriter: w}
+		ar, err := admission.ParseRequest(h.admissionCodecs, r)
+		if err != nil {
+			responder.Error(w, r, err)
+			return
+		}
+
+		span.AddEvent("GetPluginContext",
+			grafanasemconv.GrafanaPluginId(h.dataplaneService.Spec.PluginID),
+		)
+		pluginContext, err := h.pluginContextProvider.GetPluginContext(ctx, h.dataplaneService.Spec.PluginID, "")
+		if err != nil {
+			responder.Error(w, r, fmt.Errorf("unable to get plugin context: %w", err))
+			return
+		}
+
+		req, err := admission.ToAdmissionRequest(pluginContext, ar)
+		if err != nil {
+			responder.Error(w, r, fmt.Errorf("unable to convert admission request: %w", err))
+			return
+		}
+
+		ctx = backend.WithGrafanaConfig(ctx, pluginContext.GrafanaConfig)
+		span.AddEvent("MutateAdmission start")
+		rsp, err := h.client.MutateAdmission(ctx, req)
+		if err != nil {
+			responder.Error(w, r, err)
+			return
+		}
+		span.AddEvent("MutateAdmission end")
+
+		span.AddEvent("FromMutationResponse start")
+		res, err := admission.FromMutationResponse(ar.Request.Object.Raw, rsp)
+		if err != nil {
+			responder.Error(w, r, err)
+			return
+		}
+		res.SetGroupVersionKind(ar.GroupVersionKind())
+		res.Response.UID = ar.Request.UID
+
+		respBytes, err := json.Marshal(res)
+		if err != nil {
+			klog.Error(err)
+			responder.Error(w, r, err)
+			return
+		}
+
+		w.Header().Set("Content-Type", "application/json")
+		if _, err := w.Write(respBytes); err != nil {
+			klog.Error(err)
+		}
+	})
+}
+
+func (h *PluginHandler) AdmissionValidationHandler() http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		ctx := r.Context()
+		span := tracing.SpanFromContext(ctx)
+		span.AddEvent("AdmissionValidationHandler")
+		responder := &util.Responder{ResponseWriter: w}
+		ar, err := admission.ParseRequest(h.admissionCodecs, r)
+		if err != nil {
+			responder.Error(w, r, err)
+			return
+		}
+
+		span.AddEvent("GetPluginContext",
+			grafanasemconv.GrafanaPluginId(h.dataplaneService.Spec.PluginID),
+		)
+		pluginContext, err := h.pluginContextProvider.GetPluginContext(ctx, h.dataplaneService.Spec.PluginID, "")
+		if err != nil {
+			responder.Error(w, r, fmt.Errorf("unable to get plugin context: %w", err))
+			return
+		}
+
+		req, err := admission.ToAdmissionRequest(pluginContext, ar)
+		if err != nil {
+			responder.Error(w, r, fmt.Errorf("unable to convert admission request: %w", err))
+			return
+		}
+
+		ctx = backend.WithGrafanaConfig(ctx, pluginContext.GrafanaConfig)
+		span.AddEvent("ValidateAdmission start")
+		rsp, err := h.client.ValidateAdmission(ctx, req)
+		if err != nil {
+			responder.Error(w, r, err)
+			return
+		}
+		span.AddEvent("ValidateAdmission end")
+
+		res := admission.FromValidationResponse(rsp)
+		res.SetGroupVersionKind(ar.GroupVersionKind())
+		res.Response.UID = ar.Request.UID
+
+		respBytes, err := json.Marshal(res)
+		if err != nil {
+			klog.Error(err)
+			responder.Error(w, r, err)
+			return
+		}
+
+		w.Header().Set("Content-Type", "application/json")
+		if _, err := w.Write(respBytes); err != nil {
+			klog.Error(err)
+		}
+	})
+}

pkg/aggregator/apiserver/plugin/admission/admission.go

@@ -0,0 +1,78 @@
+package admission
+
+import (
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+
+	"github.com/grafana/grafana-plugin-sdk-go/backend"
+	admissionv1 "k8s.io/api/admission/v1"
+	"k8s.io/apimachinery/pkg/runtime/serializer"
+)
+
+func ToAdmissionRequest(pluginCtx backend.PluginContext, a *admissionv1.AdmissionReview) (*backend.AdmissionRequest, error) {
+	if a.Request == nil {
+		return nil, errors.New("admission review request is nil")
+	}
+	op, err := ToAdmissionOperation(a.Request.Operation)
+	if err != nil {
+		return nil, err
+	}
+
+	return &backend.AdmissionRequest{
+		PluginContext: pluginCtx,
+		Operation:     op,
+		Kind: backend.GroupVersionKind{
+			Group:   a.Request.Kind.Group,
+			Version: a.Request.Kind.Version,
+			Kind:    a.Request.Kind.Kind,
+		},
+		ObjectBytes:    a.Request.Object.Raw,
+		OldObjectBytes: a.Request.OldObject.Raw,
+	}, nil
+}
+
+func ToAdmissionOperation(o admissionv1.Operation) (backend.AdmissionRequestOperation, error) {
+	switch o {
+	case admissionv1.Create:
+		return backend.AdmissionRequestCreate, nil
+	case admissionv1.Delete:
+		return backend.AdmissionRequestDelete, nil
+	case admissionv1.Update:
+		return backend.AdmissionRequestUpdate, nil
+	case admissionv1.Connect:
+		// TODO: CONNECT is missing from the plugin SDK
+		return 3, nil
+	}
+	return 0, errors.New("unknown admission review operation")
+}
+
+func ParseRequest(codecs serializer.CodecFactory, r *http.Request) (*admissionv1.AdmissionReview, error) {
+	var body []byte
+	if r.Body != nil {
+		if data, err := io.ReadAll(r.Body); err == nil {
+			body = data
+		}
+	}
+
+	contentType := r.Header.Get("Content-Type")
+	if contentType != "application/json" {
+		return nil, fmt.Errorf("contentType=%s, expect application/json", contentType)
+	}
+
+	deserializer := codecs.UniversalDeserializer()
+	obj, gvk, err := deserializer.Decode(body, nil, nil)
+	if err != nil {
+		return nil, fmt.Errorf("failed to decode request: %v", err)
+	}
+
+	ar, ok := obj.(*admissionv1.AdmissionReview)
+	if !ok {
+		return nil, fmt.Errorf("expected AdmissionReview v1, got %T", obj)
+	}
+
+	ar.SetGroupVersionKind(*gvk)
+
+	return ar, nil
+}

pkg/aggregator/apiserver/plugin/admission/admission_test.go

@@ -0,0 +1,97 @@
+package admission_test
+
+import (
+	"bytes"
+	"encoding/json"
+	"net/http"
+	"testing"
+
+	admissionv1 "k8s.io/api/admission/v1"
+	v1 "k8s.io/api/authentication/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	example "k8s.io/apiserver/pkg/apis/example/v1"
+
+	"github.com/grafana/grafana-plugin-sdk-go/backend"
+	"github.com/grafana/grafana/pkg/aggregator/apiserver/plugin/admission"
+	"github.com/stretchr/testify/require"
+)
+
+func TestParseRequest(t *testing.T) {
+	exampleObj := example.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "example",
+		},
+		Spec: example.PodSpec{
+			ServiceAccountName: "example",
+		},
+	}
+
+	raw, err := json.Marshal(exampleObj)
+	require.NoError(t, err)
+
+	expectedAR := &admissionv1.AdmissionReview{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "AdmissionReview",
+			APIVersion: admissionv1.SchemeGroupVersion.String(),
+		},
+		Request: &admissionv1.AdmissionRequest{
+			UID:       "1234",
+			Kind:      metav1.GroupVersionKind{Group: "example.k8s.io", Version: "v1", Kind: "Pod"},
+			Resource:  metav1.GroupVersionResource{Group: "example.k8s.io", Version: "v1", Resource: "pods"},
+			Operation: admissionv1.Create,
+			UserInfo:  v1.UserInfo{},
+			Object:    runtime.RawExtension{Raw: raw},
+			OldObject: runtime.RawExtension{},
+			DryRun:    new(bool),
+		},
+	}
+
+	body, err := json.Marshal(expectedAR)
+	require.NoError(t, err)
+
+	t.Run("should parse request", func(t *testing.T) {
+		req, err := http.NewRequest("POST", "/admission", bytes.NewBuffer(body))
+		require.NoError(t, err)
+		req.Header.Set("Content-Type", "application/json")
+
+		ar, err := admission.ParseRequest(admission.GetCodecs(), req)
+		if err != nil {
+			t.Fatalf("failed to parse request: %v", err)
+		}
+
+		require.Equal(t, expectedAR, ar)
+	})
+}
+
+func TestToAdmissionRequest(t *testing.T) {
+	pluginCtx := backend.PluginContext{}
+	admissionReview := &admissionv1.AdmissionReview{
+		Request: &admissionv1.AdmissionRequest{
+			Operation: admissionv1.Update,
+			Kind: metav1.GroupVersionKind{
+				Group:   "example.k8s.io",
+				Version: "v1",
+				Kind:    "Pod",
+			},
+			Object: runtime.RawExtension{
+				Raw: []byte(`{"foo":"bar"}`),
+			},
+			OldObject: runtime.RawExtension{
+				Raw: []byte(`{"bar":"foo"}`),
+			},
+		},
+	}
+
+	expectedAdmissionRequest := &backend.AdmissionRequest{
+		PluginContext:  pluginCtx,
+		Operation:      backend.AdmissionRequestUpdate,
+		Kind:           backend.GroupVersionKind{Group: "example.k8s.io", Version: "v1", Kind: "Pod"},
+		ObjectBytes:    []byte(`{"foo":"bar"}`),
+		OldObjectBytes: []byte(`{"bar":"foo"}`),
+	}
+
+	admissionRequest, err := admission.ToAdmissionRequest(pluginCtx, admissionReview)
+	require.NoError(t, err)
+	require.Equal(t, expectedAdmissionRequest, admissionRequest)
+}

pkg/aggregator/apiserver/plugin/admission/mutation.go

@@ -0,0 +1,56 @@
+package admission
+
+import (
+	"encoding/json"
+	"errors"
+	"net/http"
+
+	"github.com/grafana/grafana-plugin-sdk-go/backend"
+	"github.com/mattbaird/jsonpatch"
+	admissionv1 "k8s.io/api/admission/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func FromMutationResponse(current []byte, r *backend.MutationResponse) (*admissionv1.AdmissionReview, error) {
+	res := &admissionv1.AdmissionReview{
+		Response: &admissionv1.AdmissionResponse{
+			Allowed:  r.Allowed,
+			Warnings: r.Warnings,
+		},
+	}
+
+	if !r.Allowed {
+		res.Response.Result = &metav1.Status{
+			Status:  metav1.StatusFailure,
+			Message: "Internal error",
+			Reason:  metav1.StatusReasonInternalError,
+			Code:    http.StatusInternalServerError,
+		}
+		if r.Result != nil {
+			res.Response.Result.Message = r.Result.Message
+			res.Response.Result.Reason = metav1.StatusReason(r.Result.Reason)
+			res.Response.Result.Code = r.Result.Code
+		}
+		return res, nil
+	}
+
+	if r.Allowed && len(r.ObjectBytes) == 0 {
+		return nil, errors.New("empty mutation response object bytes")
+	}
+
+	patch, err := jsonpatch.CreatePatch(current, r.ObjectBytes)
+	if err != nil {
+		return nil, err
+	}
+
+	raw, err := json.Marshal(patch)
+	if err != nil {
+		return nil, err
+	}
+
+	res.Response.Patch = raw
+	pt := admissionv1.PatchTypeJSONPatch
+	res.Response.PatchType = &pt
+
+	return res, nil
+}

pkg/aggregator/apiserver/plugin/admission/mutation_test.go

@@ -0,0 +1,99 @@
+package admission_test
+
+import (
+	"testing"
+
+	"github.com/grafana/grafana-plugin-sdk-go/backend"
+	"github.com/grafana/grafana/pkg/aggregator/apiserver/plugin/admission"
+	"github.com/stretchr/testify/require"
+	admissionv1 "k8s.io/api/admission/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func TestFromMutationResponse(t *testing.T) {
+	warnings := []string{"warning 1", "warning 2"}
+
+	exampleObj := []byte(`{"key": "value"}`)
+
+	result := &backend.StatusResult{
+		Status:  "Failure",
+		Message: "message",
+		Reason:  "reason",
+		Code:    200,
+	}
+
+	t.Run("should return expected patch", func(t *testing.T) {
+		response := &backend.MutationResponse{
+			Allowed:     true,
+			Warnings:    warnings,
+			ObjectBytes: []byte(`{"key": "value2"}`),
+		}
+		pt := admissionv1.PatchTypeJSONPatch
+		expectedAdmissionResponse := &admissionv1.AdmissionResponse{
+			Allowed:   true,
+			Warnings:  warnings,
+			Patch:     []byte(`[{"op":"replace","path":"/key","value":"value2"}]`),
+			PatchType: &pt,
+		}
+		expectedAdmissionReview := &admissionv1.AdmissionReview{
+			Response: expectedAdmissionResponse,
+		}
+		actualAdmissionReview, err := admission.FromMutationResponse(exampleObj, response)
+		require.NoError(t, err)
+		require.Equal(t, expectedAdmissionReview, actualAdmissionReview)
+	})
+
+	t.Run("should error if MutationResponse has empty ObjectBytes", func(t *testing.T) {
+		response := &backend.MutationResponse{
+			Allowed:  true,
+			Warnings: warnings,
+		}
+		_, err := admission.FromMutationResponse(exampleObj, response)
+		require.Error(t, err)
+	})
+
+	t.Run("should include Result in AdmissionResponse when not allowed", func(t *testing.T) {
+		response := &backend.MutationResponse{
+			Allowed:  false,
+			Warnings: warnings,
+			Result:   result,
+		}
+		expectedAdmissionResponse := &admissionv1.AdmissionResponse{
+			Allowed:  false,
+			Warnings: warnings,
+			Result: &metav1.Status{
+				Status:  result.Status,
+				Message: result.Message,
+				Reason:  metav1.StatusReason(result.Reason),
+				Code:    result.Code,
+			},
+		}
+		expectedAdmissionReview := &admissionv1.AdmissionReview{
+			Response: expectedAdmissionResponse,
+		}
+		actualAdmissionReview, err := admission.FromMutationResponse(exampleObj, response)
+		require.NoError(t, err)
+		require.Equal(t, expectedAdmissionReview, actualAdmissionReview)
+	})
+
+	t.Run("should handle nil Warnings and Result", func(t *testing.T) {
+		response := &backend.MutationResponse{
+			Allowed: false,
+		}
+		expectedAdmissionResponse := &admissionv1.AdmissionResponse{
+			Allowed: false,
+			Result: &metav1.Status{
+				Status:  metav1.StatusFailure,
+				Message: "Internal error",
+				Reason:  metav1.StatusReasonInternalError,
+				Code:    500,
+			},
+		}
+		expectedAdmissionReview := &admissionv1.AdmissionReview{
+			Response: expectedAdmissionResponse,
+		}
+		actualAdmissionReview, err := admission.FromMutationResponse(exampleObj, response)
+		require.NoError(t, err)
+		require.Equal(t, expectedAdmissionReview, actualAdmissionReview)
+	})
+}

pkg/aggregator/apiserver/plugin/admission/scheme.go

@@ -0,0 +1,15 @@
+package admission
+
+import (
+	admissionv1 "k8s.io/api/admission/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/serializer"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+)
+
+func GetCodecs() serializer.CodecFactory {
+	scheme := runtime.NewScheme()
+	codecs := serializer.NewCodecFactory(scheme)
+	utilruntime.Must(admissionv1.AddToScheme(scheme))
+	return codecs
+}

pkg/aggregator/apiserver/plugin/admission/validation.go

@@ -0,0 +1,36 @@
+package admission
+
+import (
+	"net/http"
+
+	"github.com/grafana/grafana-plugin-sdk-go/backend"
+	admissionv1 "k8s.io/api/admission/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func FromValidationResponse(r *backend.ValidationResponse) *admissionv1.AdmissionReview {
+	res := &admissionv1.AdmissionResponse{
+		Allowed:  r.Allowed,
+		Warnings: r.Warnings,
+	}
+
+	if !r.Allowed {
+		res.Result = &metav1.Status{
+			Status:  metav1.StatusFailure,
+			Message: "Internal error",
+			Reason:  metav1.StatusReasonInternalError,
+			Code:    http.StatusInternalServerError,
+		}
+		if r.Result != nil {
+			res.Result.Message = r.Result.Message
+			res.Result.Reason = metav1.StatusReason(r.Result.Reason)
+			res.Result.Code = r.Result.Code
+		}
+	}
+
+	resAR := &admissionv1.AdmissionReview{
+		Response: res,
+	}
+
+	return resAR
+}

pkg/aggregator/apiserver/plugin/admission/validation_test.go

@@ -0,0 +1,82 @@
+package admission_test
+
+import (
+	"testing"
+
+	"github.com/grafana/grafana-plugin-sdk-go/backend"
+	"github.com/grafana/grafana/pkg/aggregator/apiserver/plugin/admission"
+	"github.com/stretchr/testify/require"
+	admissionv1 "k8s.io/api/admission/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func TestFromValidationResponse(t *testing.T) {
+	warnings := []string{"warning 1", "warning 2"}
+
+	result := &backend.StatusResult{
+		Status:  "Failure",
+		Message: "message",
+		Reason:  "reason",
+		Code:    200,
+	}
+
+	t.Run("should include Result in AdmissionResponse when not allowed", func(t *testing.T) {
+		response := &backend.ValidationResponse{
+			Allowed:  false,
+			Warnings: warnings,
+			Result:   result,
+		}
+		expectedAdmissionResponse := &admissionv1.AdmissionResponse{
+			Allowed:  false,
+			Warnings: warnings,
+			Result: &metav1.Status{
+				Status:  result.Status,
+				Message: result.Message,
+				Reason:  metav1.StatusReason(result.Reason),
+				Code:    result.Code,
+			},
+		}
+		expectedAdmissionReview := &admissionv1.AdmissionReview{
+			Response: expectedAdmissionResponse,
+		}
+		actualAdmissionReview := admission.FromValidationResponse(response)
+		require.Equal(t, expectedAdmissionReview, actualAdmissionReview)
+	})
+
+	t.Run("should not include Result in AdmissionResponse when allowed", func(t *testing.T) {
+		response := &backend.ValidationResponse{
+			Allowed:  true,
+			Warnings: warnings,
+			Result:   result,
+		}
+		expectedAdmissionResponse := &admissionv1.AdmissionResponse{
+			Allowed:  true,
+			Warnings: warnings,
+		}
+		expectedAdmissionReview := &admissionv1.AdmissionReview{
+			Response: expectedAdmissionResponse,
+		}
+		actualAdmissionReview := admission.FromValidationResponse(response)
+		require.Equal(t, expectedAdmissionReview, actualAdmissionReview)
+	})
+
+	t.Run("should handle nil Warnings and Result", func(t *testing.T) {
+		response := &backend.ValidationResponse{
+			Allowed: false,
+		}
+		expectedAdmissionResponse := &admissionv1.AdmissionResponse{
+			Allowed: false,
+			Result: &metav1.Status{
+				Status:  metav1.StatusFailure,
+				Message: "Internal error",
+				Reason:  metav1.StatusReasonInternalError,
+				Code:    500,
+			},
+		}
+		expectedAdmissionReview := &admissionv1.AdmissionReview{
+			Response: expectedAdmissionResponse,
+		}
+		actualAdmissionReview := admission.FromValidationResponse(response)
+		require.Equal(t, expectedAdmissionReview, actualAdmissionReview)
+	})
+}

pkg/aggregator/apiserver/plugin/admission_test.go

@@ -0,0 +1,218 @@
+package plugin
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/grafana/grafana-plugin-sdk-go/backend"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	admissionv1 "k8s.io/api/admission/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+
+	"github.com/grafana/grafana/pkg/aggregator/apis/aggregation/v0alpha1"
+	"github.com/grafana/grafana/pkg/aggregator/apiserver/plugin/fakes"
+)
+
+func TestAdmissionMutation(t *testing.T) {
+	dps := v0alpha1.DataPlaneService{
+		Spec: v0alpha1.DataPlaneServiceSpec{
+			PluginID: "testds",
+			Group:    "testds.example.com",
+			Version:  "v1",
+			Services: []v0alpha1.Service{
+				{
+					Type: v0alpha1.AdmissionControlServiceType,
+				},
+			},
+		},
+	}
+
+	pluginContext := backend.PluginContext{
+		DataSourceInstanceSettings: &backend.DataSourceInstanceSettings{
+			ID: 1,
+		},
+	}
+	contextProvider := &fakes.FakePluginContextProvider{
+		PluginContext: pluginContext,
+	}
+
+	admissionReview := &admissionv1.AdmissionReview{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "AdmissionReview",
+			APIVersion: admissionv1.SchemeGroupVersion.String(),
+		},
+		Request: &admissionv1.AdmissionRequest{
+			UID:       "1234",
+			Operation: admissionv1.Update,
+			Kind: metav1.GroupVersionKind{
+				Group:   "example.k8s.io",
+				Version: "v1",
+				Kind:    "Pod",
+			},
+			Object: runtime.RawExtension{
+				Raw: []byte(`{"foo":"bar"}`),
+			},
+			OldObject: runtime.RawExtension{
+				Raw: []byte(`{"bar":"foo"}`),
+			},
+		},
+	}
+
+	pluginRes := &backend.MutationResponse{
+		Allowed:     true,
+		ObjectBytes: []byte(`{"foo": "foo"}`),
+	}
+
+	jsonAdmissionReview, err := json.Marshal(admissionReview)
+	require.NoError(t, err)
+
+	pc := &fakes.FakePluginClient{
+		MutateAdmissionFunc: newFakeMutateAdmissionHandler(pluginRes, nil),
+	}
+
+	delegate := fakes.NewFakeHTTPHandler(http.StatusNotFound, []byte(`Not Found`))
+	handler := NewPluginHandler(pc, dps, contextProvider, delegate)
+
+	t.Run("should return mutation response", func(t *testing.T) {
+		req, err := http.NewRequest("POST", "/apis/testds.example.com/v1/admission/mutate", bytes.NewBuffer(jsonAdmissionReview))
+		assert.NoError(t, err)
+		req.Header.Set("Content-Type", "application/json")
+
+		rr := httptest.NewRecorder()
+		handler.ServeHTTP(rr, req)
+
+		assert.Equal(t, http.StatusOK, rr.Code)
+		pt := admissionv1.PatchTypeJSONPatch
+		expectedRes := &admissionv1.AdmissionReview{
+			TypeMeta: metav1.TypeMeta{
+				Kind:       "AdmissionReview",
+				APIVersion: admissionv1.SchemeGroupVersion.String(),
+			},
+			Response: &admissionv1.AdmissionResponse{
+				UID:       admissionReview.Request.UID,
+				Allowed:   true,
+				Patch:     []byte(`[{"op":"replace","path":"/foo","value":"foo"}]`),
+				PatchType: &pt,
+			},
+		}
+		actualRes := &admissionv1.AdmissionReview{}
+		assert.NoError(t, json.NewDecoder(rr.Body).Decode(actualRes))
+		require.Equal(t, expectedRes, actualRes)
+	})
+}
+
+func TestAdmissionValidation(t *testing.T) {
+	dps := v0alpha1.DataPlaneService{
+		Spec: v0alpha1.DataPlaneServiceSpec{
+			PluginID: "testds",
+			Group:    "testds.example.com",
+			Version:  "v1",
+			Services: []v0alpha1.Service{
+				{
+					Type: v0alpha1.AdmissionControlServiceType,
+				},
+			},
+		},
+	}
+
+	pluginContext := backend.PluginContext{
+		DataSourceInstanceSettings: &backend.DataSourceInstanceSettings{
+			ID: 1,
+		},
+	}
+	contextProvider := &fakes.FakePluginContextProvider{
+		PluginContext: pluginContext,
+	}
+
+	admissionReview := &admissionv1.AdmissionReview{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "AdmissionReview",
+			APIVersion: admissionv1.SchemeGroupVersion.String(),
+		},
+		Request: &admissionv1.AdmissionRequest{
+			UID:       "1234",
+			Operation: admissionv1.Update,
+			Kind: metav1.GroupVersionKind{
+				Group:   "example.k8s.io",
+				Version: "v1",
+				Kind:    "Pod",
+			},
+			Object: runtime.RawExtension{
+				Raw: []byte(`{"foo":"bar"}`),
+			},
+			OldObject: runtime.RawExtension{
+				Raw: []byte(`{"bar":"foo"}`),
+			},
+		},
+	}
+
+	pluginRes := &backend.ValidationResponse{
+		Allowed: false,
+		Result: &backend.StatusResult{
+			Status:  "Failure",
+			Message: "message",
+			Reason:  "NotFound",
+			Code:    404,
+		},
+		Warnings: []string{"warning 1", "warning 2"},
+	}
+
+	jsonAdmissionReview, err := json.Marshal(admissionReview)
+	require.NoError(t, err)
+
+	pc := &fakes.FakePluginClient{
+		ValidateAdmissionFunc: newFakeValidateAdmissionHandler(pluginRes, nil),
+	}
+
+	delegate := fakes.NewFakeHTTPHandler(http.StatusNotFound, []byte(`Not Found`))
+	handler := NewPluginHandler(pc, dps, contextProvider, delegate)
+
+	t.Run("should return validation response", func(t *testing.T) {
+		req, err := http.NewRequest("POST", "/apis/testds.example.com/v1/admission/validate", bytes.NewBuffer(jsonAdmissionReview))
+		assert.NoError(t, err)
+		req.Header.Set("Content-Type", "application/json")
+
+		rr := httptest.NewRecorder()
+		handler.ServeHTTP(rr, req)
+
+		assert.Equal(t, http.StatusOK, rr.Code)
+		expectedRes := &admissionv1.AdmissionReview{
+			TypeMeta: metav1.TypeMeta{
+				Kind:       "AdmissionReview",
+				APIVersion: admissionv1.SchemeGroupVersion.String(),
+			},
+			Response: &admissionv1.AdmissionResponse{
+				UID:     admissionReview.Request.UID,
+				Allowed: false,
+				Result: &metav1.Status{
+					Status:  metav1.StatusFailure,
+					Message: "message",
+					Reason:  metav1.StatusReasonNotFound,
+					Code:    404,
+				},
+				Warnings: pluginRes.Warnings,
+			},
+		}
+		actualRes := &admissionv1.AdmissionReview{}
+		assert.NoError(t, json.NewDecoder(rr.Body).Decode(actualRes))
+		require.Equal(t, expectedRes, actualRes)
+	})
+}
+
+func newFakeMutateAdmissionHandler(response *backend.MutationResponse, err error) backend.MutateAdmissionFunc {
+	return func(ctx context.Context, req *backend.AdmissionRequest) (*backend.MutationResponse, error) {
+		return response, err
+	}
+}
+
+func newFakeValidateAdmissionHandler(response *backend.ValidationResponse, err error) backend.ValidateAdmissionFunc {
+	return func(ctx context.Context, req *backend.AdmissionRequest) (*backend.ValidationResponse, error) {
+		return response, err
+	}
+}

pkg/aggregator/apiserver/plugin/fakes/client.go

@@ -0,0 +1,38 @@
+package fakes
+
+import (
+	"context"
+	"errors"
+
+	"github.com/grafana/grafana-plugin-sdk-go/backend"
+)
+
+type FakePluginClient struct {
+	backend.QueryDataHandlerFunc
+	backend.MutateAdmissionFunc
+	backend.ValidateAdmissionFunc
+}
+
+func (pc *FakePluginClient) QueryData(ctx context.Context, req *backend.QueryDataRequest) (*backend.QueryDataResponse, error) {
+	if pc.QueryDataHandlerFunc != nil {
+		return pc.QueryDataHandlerFunc(ctx, req)
+	}
+
+	return nil, errors.New("QueryDataHandlerFunc not implemented")
+}
+
+func (pc *FakePluginClient) ValidateAdmission(ctx context.Context, req *backend.AdmissionRequest) (*backend.ValidationResponse, error) {
+	if pc.ValidateAdmissionFunc != nil {
+		return pc.ValidateAdmissionFunc(ctx, req)
+	}
+
+	return nil, errors.New("ValidateAdmissionFunc not implemented")
+}
+
+func (pc *FakePluginClient) MutateAdmission(ctx context.Context, req *backend.AdmissionRequest) (*backend.MutationResponse, error) {
+	if pc.MutateAdmissionFunc != nil {
+		return pc.MutateAdmissionFunc(ctx, req)
+	}
+
+	return nil, errors.New("MutateAdmissionFunc not implemented")
+}

pkg/aggregator/apiserver/plugin/fakes/http.go

@@ -0,0 +1,13 @@
+package fakes
+
+import "net/http"
+
+func NewFakeHTTPHandler(status int, res []byte) http.HandlerFunc {
+	return func(w http.ResponseWriter, req *http.Request) {
+		w.WriteHeader(status)
+		_, err := w.Write(res)
+		if err != nil {
+			panic(err)
+		}
+	}
+}

pkg/aggregator/apiserver/plugin/fakes/plugin_context.go

@@ -0,0 +1,16 @@
+package fakes
+
+import (
+	"context"
+
+	"github.com/grafana/grafana-plugin-sdk-go/backend"
+)
+
+type FakePluginContextProvider struct {
+	PluginContext backend.PluginContext
+	Err           error
+}
+
+func (f FakePluginContextProvider) GetPluginContext(ctx context.Context, pluginID, dsUID string) (backend.PluginContext, error) {
+	return f.PluginContext, f.Err
+}

pkg/aggregator/apiserver/plugin/handler.go

@@ -6,15 +6,15 @@ import (
 	"path"
 
 	"github.com/grafana/grafana-plugin-sdk-go/backend"
+	"k8s.io/apimachinery/pkg/runtime/serializer"
 
 	aggregationv0alpha1 "github.com/grafana/grafana/pkg/aggregator/apis/aggregation/v0alpha1"
+	"github.com/grafana/grafana/pkg/aggregator/apiserver/plugin/admission"
 )
 
 type PluginClient interface {
 	backend.QueryDataHandler
-	backend.StreamHandler
 	backend.AdmissionHandler
-	backend.CallResourceHandler
 }
 
 type PluginContextProvider interface {
@@ -29,6 +29,8 @@ type PluginHandler struct {
 	pluginContextProvider PluginContextProvider
 
 	dataplaneService aggregationv0alpha1.DataPlaneService
+
+	admissionCodecs serializer.CodecFactory
 }
 
 func NewPluginHandler(
@@ -43,6 +45,7 @@ func NewPluginHandler(
 		client:                client,
 		pluginContextProvider: pluginContextProvider,
 		dataplaneService:      dataplaneService,
+		admissionCodecs:       admission.GetCodecs(),
 	}
 	h.registerRoutes()
 	return h
@@ -54,7 +57,8 @@ func (h *PluginHandler) registerRoutes() {
 	for _, service := range h.dataplaneService.Spec.Services {
 		switch service.Type {
 		case aggregationv0alpha1.AdmissionControlServiceType:
-			// TODO: implement in future PR
+			h.mux.Handle(proxyPath("/admission/mutate"), h.AdmissionMutationHandler())
+			h.mux.Handle(proxyPath("/admission/validate"), h.AdmissionValidationHandler())
 		case aggregationv0alpha1.ConversionServiceType:
 			// TODO: implement in future PR
 		case aggregationv0alpha1.DataSourceProxyServiceType:

pkg/aggregator/apiserver/plugin/query_test.go

@@ -14,7 +14,7 @@ import (
 	"github.com/grafana/grafana-plugin-sdk-go/data"
 	datav0alpha1 "github.com/grafana/grafana-plugin-sdk-go/experimental/apis/data/v0alpha1"
 	"github.com/grafana/grafana/pkg/aggregator/apis/aggregation/v0alpha1"
-	"github.com/grafana/grafana/pkg/plugins/manager/fakes"
+	"github.com/grafana/grafana/pkg/aggregator/apiserver/plugin/fakes"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -38,8 +38,8 @@ func TestQueryDataHandler(t *testing.T) {
 			ID: 1,
 		},
 	}
-	contextProvider := &fakePluginContextProvider{
-		pluginContext: pluginContext,
+	contextProvider := &fakes.FakePluginContextProvider{
+		PluginContext: pluginContext,
 	}
 
 	res := &backend.QueryDataResponse{
@@ -58,7 +58,7 @@ func TestQueryDataHandler(t *testing.T) {
 		QueryDataHandlerFunc: newfakeQueryDataHandler(res, nil),
 	}
 
-	delegate := newFakeHTTPHandler(http.StatusNotFound, []byte(`Not Found`))
+	delegate := fakes.NewFakeHTTPHandler(http.StatusNotFound, []byte(`Not Found`))
 	handler := NewPluginHandler(pc, dps, contextProvider, delegate)
 
 	qdr := datav0alpha1.QueryDataRequest{
@@ -176,27 +176,8 @@ func TestQueryDataHandler(t *testing.T) {
 	})
 }
 
-type fakePluginContextProvider struct {
-	pluginContext backend.PluginContext
-	err           error
-}
-
-func (f fakePluginContextProvider) GetPluginContext(ctx context.Context, pluginID, dsUID string) (backend.PluginContext, error) {
-	return f.pluginContext, f.err
-}
-
 func newfakeQueryDataHandler(res *backend.QueryDataResponse, err error) backend.QueryDataHandlerFunc {
 	return func(ctx context.Context, req *backend.QueryDataRequest) (*backend.QueryDataResponse, error) {
 		return res, err
 	}
 }
-
-func newFakeHTTPHandler(status int, res []byte) http.HandlerFunc {
-	return func(w http.ResponseWriter, req *http.Request) {
-		w.WriteHeader(status)
-		_, err := w.Write(res)
-		if err != nil {
-			panic(err)
-		}
-	}
-}