Merge pull request #219 from xuzhu-591/feat-mutating-admission-webhook

feat: support for muatating admission webhook

.gitignore

@@ -16,4 +16,6 @@ bin/
 
 # makefile
 tmp/
-tools/
+tools/
+
+config-dev.yaml

core/controller/cloudevent/controller.go

@@ -73,6 +73,10 @@ func (c *controller) CloudEvent(ctx context.Context, wpr *WrappedPipelineRun) (e
 
 	horizonMetaData, err := c.getHorizonMetaData(ctx, wpr)
 	if err != nil {
+		if _, ok := perror.Cause(err).(*herrors.HorizonErrNotFound); ok {
+			log.Warningf(ctx, "horizon resource not found, err: %v", err.Error())
+			return nil
+		}
 		return err
 	}
 	log.Infof(ctx, "got cloudEvent of pipelineRun %v, event id: %v",
@@ -180,6 +184,9 @@ func (c *controller) getHorizonMetaData(ctx context.Context, wpr *WrappedPipelin
 	if err != nil {
 		return nil, err
 	}
+	if pipelinerun == nil {
+		return nil, herrors.NewErrNotFound(herrors.PipelinerunInDB, "cannot find pipelinerun by eventID")
+	}
 	cluster, err := c.clusterMgr.GetByID(ctx, pipelinerun.ClusterID)
 	if err != nil {
 		return nil, err

core/errors/horizonerrors.go

@@ -213,6 +213,7 @@ var (
 	ErrGenerateRandomID        = errors.New("failed to generate random id")
 	ErrDisabled                = errors.New("entity is disabled")
 	ErrDuplicatedKey           = errors.New("duplicated keys")
+	ErrMutatingFailed          = errors.New("mutating failed")
 	ErrValidatingFailed        = errors.New("validating failed")
 	ErrUnsupportedResourceType = errors.New("unsupported resource type")
 

core/middleware/admission/admission.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/gin-gonic/gin"
 	"github.com/gin-gonic/gin/binding"
+
 	"github.com/horizoncd/horizon/core/common"
 	"github.com/horizoncd/horizon/core/middleware"
 	admissionwebhook "github.com/horizoncd/horizon/pkg/admission"
@@ -63,7 +64,11 @@ func Middleware(skippers ...middleware.Skipper) gin.HandlerFunc {
 		queries := c.Request.URL.Query()
 		options := make(map[string]interface{}, len(queries))
 		for k, v := range queries {
-			options[k] = v
+			if len(v) == 1 {
+				options[k] = v[0]
+			} else {
+				options[k] = v
+			}
 		}
 		admissionRequest := &admissionwebhook.Request{
 			Operation:   admissionmodels.Operation(attr.GetVerb()),
@@ -74,6 +79,12 @@ func Middleware(skippers ...middleware.Skipper) gin.HandlerFunc {
 			Object:      object,
 			Options:     options,
 		}
+		admissionRequest, err = admissionwebhook.Mutating(c, admissionRequest)
+		if err != nil {
+			response.AbortWithRPCError(c,
+				rpcerror.ParamError.WithErrMsg(fmt.Sprintf("admission mutating failed: %v", err)))
+			return
+		}
 		if err := admissionwebhook.Validating(c, admissionRequest); err != nil {
 			response.AbortWithRPCError(c,
 				rpcerror.ParamError.WithErrMsg(fmt.Sprintf("admission validating failed: %v", err)))

go.mod

@@ -38,7 +38,7 @@ require (
 	golang.org/x/net v0.0.0-20220107192237-5cfca573fb4d
 	golang.org/x/oauth2 v0.0.0-20200107190931-bf48bf16ab8d
 	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
-	gopkg.in/evanphx/json-patch.v5 v5.6.0
+	gopkg.in/evanphx/json-patch.v5 v5.9.0
 	gopkg.in/igm/sockjs-go.v3 v3.0.1
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0
 	gopkg.in/yaml.v3 v3.0.1

go.sum

@@ -2475,8 +2475,8 @@ gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntN
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
 gopkg.in/cheggaaa/pb.v1 v1.0.25/go.mod h1:V/YB90LKu/1FcN3WVnfiiE5oMCibMjukxqG/qStrOgw=
 gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI=
-gopkg.in/evanphx/json-patch.v5 v5.6.0 h1:BMT6KIwBD9CaU91PJCZIe46bDmBWa9ynTQgJIOpfQBk=
-gopkg.in/evanphx/json-patch.v5 v5.6.0/go.mod h1:/kvTRh1TVm5wuM6OkHxqXtE/1nUZZpihg29RtuIyfvk=
+gopkg.in/evanphx/json-patch.v5 v5.9.0 h1:hx1VU2SGj4F8r9b8GUwJLdc8DNO8sy79ZGui0G05GLo=
+gopkg.in/evanphx/json-patch.v5 v5.9.0/go.mod h1:/kvTRh1TVm5wuM6OkHxqXtE/1nUZZpihg29RtuIyfvk=
 gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
 gopkg.in/gcfg.v1 v1.2.0/go.mod h1:yesOnuUOFQAhST5vPY4nbZsb/huCgGGXlipJsBn0b3o=
 gopkg.in/gcfg.v1 v1.2.3/go.mod h1:yesOnuUOFQAhST5vPY4nbZsb/huCgGGXlipJsBn0b3o=

pkg/admission/http.go

@@ -13,6 +13,8 @@ import (
 	"strings"
 	"time"
 
+	"github.com/mattbaird/jsonpatch"
+
 	herrors "github.com/horizoncd/horizon/core/errors"
 	"github.com/horizoncd/horizon/pkg/admission/models"
 	config "github.com/horizoncd/horizon/pkg/config/admission"
@@ -181,6 +183,8 @@ type HTTPAdmissionWebhook struct {
 func NewHTTPWebhooks(config config.Admission) {
 	for _, webhook := range config.Webhooks {
 		switch webhook.Kind {
+		case models.KindMutating:
+			Register(models.KindMutating, NewHTTPWebhook(webhook))
 		case models.KindValidating:
 			Register(models.KindValidating, NewHTTPWebhook(webhook))
 		}
@@ -216,23 +220,24 @@ func (m *HTTPAdmissionWebhook) Interest(req *Request) bool {
 	return m.matchers.Match(req)
 }
 
-type DummyValidatingWebhookServer struct {
+type DummyWebhookServer struct {
 	server *httptest.Server
 }
 
 // NewDummyWebhookServer creates a dummy validating webhook server for testing
-func NewDummyWebhookServer() *DummyValidatingWebhookServer {
-	webhook := &DummyValidatingWebhookServer{}
+func NewDummyWebhookServer() *DummyWebhookServer {
+	webhook := &DummyWebhookServer{}
 
 	mux := http.NewServeMux()
+	mux.HandleFunc("/mutate", webhook.Mutating)
 	mux.HandleFunc("/validate", webhook.Validating)
 
 	server := httptest.NewServer(mux)
 	webhook.server = server
 	return webhook
 }
 
-func (*DummyValidatingWebhookServer) ReadAndResponse(resp http.ResponseWriter,
+func (*DummyWebhookServer) ReadAndResponse(resp http.ResponseWriter,
 	req *http.Request, fn func(Request, *Response)) {
 	bodyBytes, _ := ioutil.ReadAll(req.Body)
 
@@ -247,11 +252,40 @@ func (*DummyValidatingWebhookServer) ReadAndResponse(resp http.ResponseWriter,
 	_, _ = resp.Write(respBytes)
 }
 
-func (w *DummyValidatingWebhookServer) Validating(resp http.ResponseWriter, req *http.Request) {
+func (w *DummyWebhookServer) Mutating(resp http.ResponseWriter, req *http.Request) {
+	w.ReadAndResponse(resp, req, w.mutating)
+}
+
+func (w *DummyWebhookServer) mutating(req Request, resp *Response) {
+	obj := req.Object.(map[string]interface{})
+
+	jsonObj, _ := json.Marshal(obj)
+
+	var newObj map[string]interface{}
+	_ = json.Unmarshal(jsonObj, &newObj)
+	if obj["tags"] != nil {
+		tags := obj["tags"].([]interface{})
+		tags = append(tags, map[string]interface{}{"key": "scope", "value": "online/hz"})
+		newObj["tags"] = tags
+	}
+
+	newObj["name"] = fmt.Sprintf("%v-%s", obj["name"], "mutated")
+
+	jsonNewObj, _ := json.Marshal(newObj)
+
+	patch, _ := jsonpatch.CreatePatch(jsonObj, jsonNewObj)
+
+	patchJSON, _ := json.Marshal(patch)
+
+	resp.Patch = patchJSON
+	resp.PatchType = models.PatchTypeJSONPatch
+}
+
+func (w *DummyWebhookServer) Validating(resp http.ResponseWriter, req *http.Request) {
 	w.ReadAndResponse(resp, req, w.validating)
 }
 
-func (w *DummyValidatingWebhookServer) validating(req Request, resp *Response) {
+func (w *DummyWebhookServer) validating(req Request, resp *Response) {
 	obj := req.Object.(map[string]interface{})
 
 	if req.Operation.Eq(models.OperationCreate) {
@@ -297,10 +331,14 @@ func (w *DummyValidatingWebhookServer) validating(req Request, resp *Response) {
 	resp.Allowed = common.BoolPtr(true)
 }
 
-func (w *DummyValidatingWebhookServer) ValidatingURL() string {
+func (w *DummyWebhookServer) MutatingURL() string {
+	return w.server.URL + "/mutate"
+}
+
+func (w *DummyWebhookServer) ValidatingURL() string {
 	return w.server.URL + "/validate"
 }
 
-func (w *DummyValidatingWebhookServer) Stop() {
+func (w *DummyWebhookServer) Stop() {
 	w.server.Close()
 }

pkg/admission/models/models.go

@@ -20,9 +20,12 @@ func (o Operation) Eq(other Operation) bool {
 
 const (
 	KindValidating Kind = "validating"
+	KindMutating   Kind = "mutating"
 
 	MatchAll string = "*"
 
 	OperationCreate Operation = "create"
 	OperationUpdate Operation = "update"
+
+	PatchTypeJSONPatch = "JSONPatch"
 )

pkg/admission/webhook.go

@@ -2,21 +2,26 @@ package admission
 
 import (
 	"context"
+	"encoding/json"
 
 	"k8s.io/apimachinery/pkg/util/runtime"
 
 	herrors "github.com/horizoncd/horizon/core/errors"
 	"github.com/horizoncd/horizon/pkg/admission/models"
 	perror "github.com/horizoncd/horizon/pkg/errors"
 	"github.com/horizoncd/horizon/pkg/util/log"
+	jsonpatch "gopkg.in/evanphx/json-patch.v5"
 )
 
 var (
+	mutatingWebhooks   []Webhook
 	validatingWebhooks []Webhook
 )
 
 func Register(kind models.Kind, webhook Webhook) {
 	switch kind {
+	case models.KindMutating:
+		mutatingWebhooks = append(mutatingWebhooks, webhook)
 	case models.KindValidating:
 		validatingWebhooks = append(validatingWebhooks, webhook)
 	}
@@ -40,8 +45,10 @@ type Request struct {
 }
 
 type Response struct {
-	Allowed *bool  `json:"allowed"`
-	Result  string `json:"result,omitempty"`
+	Allowed   *bool  `json:"allowed"`
+	Result    string `json:"result,omitempty"`
+	Patch     []byte `json:"patch,omitempty"`
+	PatchType string `json:"patchType,omitempty"`
 }
 
 type Webhook interface {
@@ -50,6 +57,30 @@ type Webhook interface {
 	Interest(*Request) bool
 }
 
+func Mutating(ctx context.Context, request *Request) (*Request, error) {
+	ctx, cancelFunc := context.WithCancel(ctx)
+	defer cancelFunc()
+	if request.Object == nil {
+		return request, nil
+	}
+	for _, webhook := range mutatingWebhooks {
+		if !webhook.Interest(request) {
+			continue
+		}
+		response, err := webhook.Handle(ctx, request)
+		if err = loggingError(ctx, err, webhook); err != nil {
+			return nil, err
+		}
+		if response != nil && response.Patch != nil {
+			request.Object, err = jsonPatch(request.Object, response.Patch)
+			if err = loggingError(ctx, err, webhook); err != nil {
+				return nil, err
+			}
+		}
+	}
+	return request, nil
+}
+
 func Validating(ctx context.Context, request *Request) error {
 	if len(validatingWebhooks) < 1 {
 		return nil
@@ -108,3 +139,36 @@ func Validating(ctx context.Context, request *Request) error {
 
 	return nil
 }
+
+func loggingError(ctx context.Context, err error, webhook Webhook) error {
+	if err != nil {
+		if webhook.IgnoreError() {
+			log.Warningf(ctx, "failed to admit request: %v", err.Error())
+			return nil
+		}
+		log.Errorf(ctx, "failed to admit request: %v", err.Error())
+		return err
+	}
+	return nil
+}
+
+func jsonPatch(obj interface{}, patchJSON []byte) (interface{}, error) {
+	objJSON, err := json.Marshal(obj)
+	if err != nil {
+		return nil, err
+	}
+	patch, err := jsonpatch.DecodePatch(patchJSON)
+	if err != nil {
+		return nil, err
+	}
+
+	objPatched, err := patch.Apply(objJSON)
+	if err != nil {
+		return nil, err
+	}
+	err = json.Unmarshal(objPatched, &obj)
+	if err != nil {
+		return nil, err
+	}
+	return obj, nil
+}