基于策略的 kubernetes admission webhook · 一

1
2
3
4
5
6
7

# match with all the pod which contains label webhook:enabled
resourceSelectors:
  - apiVersion: v1
    kind: Pod
    labelSelector:
      matchLabels:
        webhook: enabled

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

validateRules:
# 这段代码表示，检查当前资源的 label 如果有不可删除的标识，则拒绝这次删除操作
- cue: |-
    object: _ @tag(object)
    reject: object.metadata.labels != null && object.metadata.labels["xxx.io/no-delete"] == "true"
    validate: {
        if reject{
                reason: "operation rejected"
        }
        if !reject{
                reason: ""
        }
        valid: !reject
    }    
  targetOperations:
   - DELETE

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

# 表示 存在 no-delete annotation 时 拒绝本次删除操作
validateRules:
- targetOperations:
    - DELETE
    template:
    type: condition # 当前只有 condition 一个类型，后续扩展
    condition:
        affectMode: reject # 表示命中该规则是拒绝，可以设置为 allow 表示只有命中时才准入，默认是 reject
        cond: Exist # 支持存在 不存在 大于 小于 等操作
        message: "cannot delete this ns" # 命中时返回的信息
        dataRef: # 校验的数据来源
          from: current # current 表示当前 object 中获取，支持从当前集群的其他资源或者通过 http 请求获取数据
          path: "/metadata/annotations/no-delete" # 数据 path

1
2
3
4
5
6
7
8

  overrideRules:
    - targetOperations:
        - CREATE
      overriders:
        plaintext: # 为数组，可同时修改多个字段值，会直接 apply 到对象上
          - path: /metadata/annotations/added-by
            op: add
            value: op

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

  overrideRules:
    - targetOperations:
        - CREATE
      overriders:
        # cue 部分，可以写相关逻辑，可以先判断当前 object 的值再进行操作，只要写入到 patches 的数组即可，可有多个
        cue: |-
          object: _ @tag(object)
          patches: [
            if object.metadata.annotations == _|_ {
              {
                op: "add"
                path: "/metadata/annotations"
                value: {}
              }
            },
            {
              op: "add"
              path: "/metadata/annotations/added-by"
              value: "cue"
            }
          ]          

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# 对当前资源进行超卖，即修改资源的request为 limit * factor，从而可以达到集群内资源超卖的效果
overrideRules:
- targetOperations:
    - CREATE
    overriders:
        template:
            type: resourcesOversell # 以超售为例
            operation: replace # set remove to reset oversell
            resourcesOversell:
                cpuFactor: "0.5" # use half of limit / set 0 as placeholder when it needed remove
                memoryFactor: "0.2" # use 1/5 of limit
                diskFactor: "0.1" # use 1/10 of limit

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

// Run runs the webhook server with options. This should never exit.
func Run(ctx context.Context, opts *options.Options) error {
    klog.InfoS("kinitiras webhook starting.", "version", version.Get())
    config, err := controllerruntime.GetConfig()
    if err != nil {
        panic(err)
    }
    config.QPS, config.Burst = opts.KubeAPIQPS, opts.KubeAPIBurst

    hookManager, err := controllerruntime.NewManager(config, controllerruntime.Options{
        // ... set options
    })
    if err != nil {
        klog.ErrorS(err, "failed to build webhook server.")
        return err
    }

    // init clients, informer, lister
    sm := &setupManager{}
    if err := sm.init(hookManager, ctx.Done()); err != nil {
        klog.ErrorS(err, "init setup manager failed")
        return err
    }

    if err := sm.waitForCacheSync(ctx); err != nil {
        klog.ErrorS(err, "wait for cache sync failed")
        return err
    }

    if err := sm.setupInterrupter(); err != nil {
        klog.ErrorS(err, "setup interrupter failed")
        return err
    }

    setupCh, err := cert.SetupCertRotator(hookManager, cert.Options{
        // ... set options
    })
    if err != nil {
        klog.ErrorS(err, "failed to setup cert rotator controller.")
        return err
    }

    go func() {
        <-setupCh

        // register handler here
        hookServer := hookManager.GetWebhookServer()
        hookServer.Register("/mutate", &webhook.Admission{Handler: pkgwebhook.NewMutatingAdmissionHandler(sm.overrideManager, sm.policyInterrupterManager)})
        hookServer.Register("/validate", &webhook.Admission{Handler: pkgwebhook.NewValidatingAdmissionHandler(sm.validateManager, sm.policyInterrupterManager)})
        hookServer.WebhookMux.Handle("/readyz", http.StripPrefix("/readyz", &healthz.Handler{}))
    }()

    // blocks until the context is done.
    if err := hookManager.Start(ctx); err != nil {
        klog.ErrorS(err, "webhook server exits unexpectedly.")
        return err
    }

    // never reach here
    return nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

func (s *setupManager) setupInterrupter() error {
    // 初始化模板 -- 模板渲染是基于 go.tmpl 实现的，因此这里初始化 tmpl
    otm, err := templatemanager.NewOverrideTemplateManager(&templatemanager.TemplateSource{
        Content:      templates.OverrideTemplate,
        TemplateName: "BaseTemplate",
    })
    if err != nil {
        klog.ErrorS(err, "failed to setup mutating template manager.")
        return err
    }

    // 初始化模板 -- 模板渲染是基于 go.tmpl 实现的，因此这里初始化 tmpl
    vtm, err := templatemanager.NewValidateTemplateManager(&templatemanager.TemplateSource{
        Content:      templates.ValidateTemplate,
        TemplateName: "BaseTemplate",
    })
    if err != nil {
        klog.ErrorS(err, "failed to setup validate template manager.")
        return err
    }

    // base
    baseInterrupter := interrupter.NewBaseInterrupter(otm, vtm, templatemanager.NewCueManager())

    // op
    overridePolicyInterrupter := interrupter.NewOverridePolicyInterrupter(baseInterrupter, s.tokenManager, s.client, s.opLister)
    // register interrupter to manager
    s.policyInterrupterManager.AddInterrupter(schema.GroupVersionKind{
        Group:   policyv1alpha1.SchemeGroupVersion.Group,
        Version: policyv1alpha1.SchemeGroupVersion.Version,
        Kind:    "OverridePolicy",
    }, overridePolicyInterrupter)
    // cop
    s.policyInterrupterManager.AddInterrupter(schema.GroupVersionKind{
        Group:   policyv1alpha1.SchemeGroupVersion.Group,
        Version: policyv1alpha1.SchemeGroupVersion.Version,
        Kind:    "ClusterOverridePolicy",
    }, interrupter.NewClusterOverridePolicyInterrupter(overridePolicyInterrupter, s.copLister))
    // cvp
    s.policyInterrupterManager.AddInterrupter(schema.GroupVersionKind{
        Group:   policyv1alpha1.SchemeGroupVersion.Group,
        Version: policyv1alpha1.SchemeGroupVersion.Version,
        Kind:    "ClusterValidatePolicy",
    }, interrupter.NewClusterValidatePolicyInterrupter(baseInterrupter, s.tokenManager, s.client, s.cvpLister))

    return s.policyInterrupterManager.OnStartUp()
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

func (a *MutatingAdmission) Handle(ctx context.Context, req admission.Request) admission.Response {
    obj, oldObj, err := decodeObj(a.decoder, req)
    if err != nil {
        return admission.Errored(http.StatusBadRequest, err)
    }

    newObj := obj.DeepCopy()
    // if obj is known policy, then run policy interrupter
    // 这里先调用拦截器逻辑，内部识别是否为已知的策略 crd 以及是否需要对其进行模版渲染等
    patches, err := a.policyInterrupterManager.OnMutating(newObj, oldObj, req.Operation)
    if err != nil {
        return admission.Errored(http.StatusInternalServerError, err)
    }

    // 如果有需要修改的信息，则以 patches 的形式返回，这里就确定是已知的 crd，打 patch
    if len(patches) != 0 {
        klog.V(4).InfoS("patches for policy", "policy", obj.GroupVersionKind(), "patchesCount", len(patches))
        // patch data
        patchedObj, err := json.Marshal(newObj)
        if err != nil {
            return admission.Errored(http.StatusInternalServerError, err)
        }

        return admission.PatchResponseFromRaw(req.Object.Raw, patchedObj)
    }

    // 其他资源或当前策略crd
    // 这里是另一个核心点，匹配&&执行 策略
    cops, ops, err := a.overrideManager.ApplyOverridePolicies(newObj, oldObj, req.Operation)
    if err != nil {
        return admission.Errored(http.StatusInternalServerError, err)
    }

    if req.Operation == admissionv1.Delete {
        return admission.Allowed("")
    }

    patchedObj, err := json.Marshal(newObj)
    if err != nil {
        return admission.Errored(http.StatusInternalServerError, err)
    }

    return admission.PatchResponseFromRaw(req.Object.Raw, patchedObj)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

// PolicyInterrupterManager manage multi PolicyInterrupter and decide which one to use by gvk.
type PolicyInterrupterManager interface {
    PolicyInterrupter
    // AddInterrupter add a PolicyInterrupter to manager,
    //  it will replace interrupter if already add with same gvk.s
    AddInterrupter(gvk schema.GroupVersionKind, pi PolicyInterrupter)
}

// PolicyInterrupter defines interrupt process for policy change
// It validate and mutate policy.
type PolicyInterrupter interface {
    // OnMutating called on "/mutating" api to complete policy
    // return nil means obj is not defined policy
    OnMutating(obj, oldObj *unstructured.Unstructured, operation admissionv1.Operation) ([]jsonpatchv2.JsonPatchOperation, error)
    // OnValidating called on "/validating" api to validate policy
    // return nil means obj is not defined policy or no invalid field
    OnValidating(obj, oldObj *unstructured.Unstructured, operation admissionv1.Operation) error
    // OnStartUp called when webhook process initialize
    // return error if initial phase get any error
    OnStartUp() error
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

type policyInterrupterManagerImpl struct {
    interrupters sync.Map
}

func (p *policyInterrupterManagerImpl) OnValidating(obj, oldObj *unstructured.Unstructured, operation admissionv1.Operation) error {
    if interrupter := p.getInterrupter(obj); interrupter != nil {
        return interrupter.OnValidating(obj, oldObj, operation)
    }

    return nil
}

func (p *policyInterrupterManagerImpl) getInterrupter(obj *unstructured.Unstructured) PolicyInterrupter {
    if !p.isKnownPolicy(obj) {
        klog.V(5).InfoS("unknown policy", "gvk", obj.GroupVersionKind())
        return nil
    }

    i, ok := p.interrupters.Load(obj.GroupVersionKind())
    if ok {
        klog.V(4).InfoS("sub interrupter found", "gvk", obj.GroupVersionKind())
        return i.(PolicyInterrupter)
    }

    return nil
}

func (p *policyInterrupterManagerImpl) isKnownPolicy(obj *unstructured.Unstructured) bool {
    group := strings.Split(obj.GetAPIVersion(), "/")[0]
    return group == policyv1alpha1.SchemeGroupVersion.Group
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

// ResourceMatchSelectors tells if the specific resource matches the selectors.
func ResourceMatchSelectors(resource *unstructured.Unstructured, selectors ...policyv1alpha1.ResourceSelector) bool {
    for _, rs := range selectors {
        // 一个策略可以配置多个 selector，只要其中任意一个命中即可
        if ResourceMatches(resource, rs) {
            return true
        }
    }
    return false
}

// ResourceMatches tells if the specific resource matches the selector.
func ResourceMatches(resource *unstructured.Unstructured, rs policyv1alpha1.ResourceSelector) bool {
    if resource.GetAPIVersion() != rs.APIVersion ||
        resource.GetKind() != rs.Kind ||
        (len(rs.Namespace) > 0 && resource.GetNamespace() != rs.Namespace) {
        return false
    }

    // name not empty, don't need to consult selector.
    if len(rs.Name) > 0 {
        return rs.Name == resource.GetName()
    }

    // all empty, matches all
    if rs.LabelSelector == nil && rs.FieldSelector == nil {
        return true
    }

    // matches with field selector
    if rs.FieldSelector != nil {
        match, err := rs.FieldSelector.MatchObject(resource)
        if err != nil {
            klog.ErrorS(err, "match fields failed")
            return false
        }

        if !match {
            // return false if not match
            return false
        }
    }

    // matches with selector
    if rs.LabelSelector != nil {
        var s labels.Selector
        var err error
        if s, err = metav1.LabelSelectorAsSelector(rs.LabelSelector); err != nil {
            // should not happen because all resource selector should be fully validated by webhook.
            klog.ErrorS(err, "match labels failed")
            return false
        }

        return s.Matches(labels.Set(resource.GetLabels()))
    }

    return true
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

// applyPolicyOverriders applies OverridePolicy/ClusterOverridePolicy overriders to target object
func (o *overrideManagerImpl) applyPolicyOverriders(rawObj, oldObj *unstructured.Unstructured, p policyOverriders) error {
    policyName := p.name
    if p.namespace != "" {
        policyName = p.namespace + "/" + p.name
    }
    if p.overriders.Template != nil && p.overriders.RenderedCue != "" {
        cp, err := cue.BuildCueParamsViaOverridePolicy(o.dynamicLister, rawObj, p.overriders.Template)
        if err != nil {
            metrics.PolicyGotError(policyName, rawObj.GroupVersionKind(), metrics.ErrTypePrepareCueParams)
            return fmt.Errorf("BuildCueParamsViaOverridePolicy error=%w", err)
        }
        cp.Object = rawObj
        cp.OldObject = oldObj
        if cp.OldObject == nil {
            cp.OldObject = &unstructured.Unstructured{Object: map[string]interface{}{}}
        }
        params := []cue.Parameter{
            {
                Object: cp,
                Name:   utils.DataParameterName,
            },
        } // 该参数将传参到 cue 中，从而达到 cue 内引入外部数据

        patches, err := executeCueV2(p.overriders.RenderedCue, params)
        if err != nil {
            metrics.PolicyGotError(policyName, rawObj.GroupVersionKind(), metrics.ErrorTypeCueExecute)
            return err
        }

        // 执行后可获取cue 内部的所有定义的数据，我们只取 patches 这个数组
        if len(patches) > 0 {
            metrics.OverridePolicyOverride(policyName, rawObj.GroupVersionKind())
        }

        if err := applyJSONPatch(rawObj, patches); err != nil {
            return err
        }
    }

    // ... ignore code
}    

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

// simple cue code
data: _ @tag(data)
object := data.object

patches: [
    if object.metadata.annotations == _|_ {
        {
        op: "add"
        path: "/metadata/annotations"
        value: {}
        }
    },
    {
        op: "add"
        path: "/metadata/annotations/added-by"
        value: "cue"
    }
]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

// applyPolicyOverriders applies OverridePolicy/ClusterOverridePolicy overriders to target object
func (o *overrideManagerImpl) applyPolicyOverriders(rawObj, oldObj *unstructured.Unstructured, p policyOverriders) error {
    // ...ignore code
    if p.overriders.Cue != "" {
        // 用户自定义 cue 脚本只传参当前 object 不支持引用外部 object
        patches, err := executeCue(rawObj, p.overriders.Cue)
        if err != nil {
            metrics.PolicyGotError(policyName, rawObj.GroupVersionKind(), metrics.ErrorTypeCueExecute)
            return err
        }
        if patches != nil && len(*patches) > 0 {
            metrics.OverridePolicyOverride(policyName, rawObj.GroupVersionKind())
        }
        if err := applyJSONPatch(rawObj, *patches); err != nil {
            return err
        }
    }

    // ...ignore code
}

1
2
3
4
5
6

// applyPolicyOverriders applies OverridePolicy/ClusterOverridePolicy overriders to target object
func (o *overrideManagerImpl) applyPolicyOverriders(rawObj, oldObj *unstructured.Unstructured, p policyOverriders) error {
    // ...ignore code
    return applyJSONPatch(rawObj, parseJSONPatchesByPlaintext(p.overriders.Plaintext))
}