kubernetes
/
apiserver
mirror of https://github.com/kubernetes/apiserver.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

use typedinformer if available 

reduces memory and cpu when things like configmap are used as a param

cannot be shared due to limitatoins of sharedinformerfactory

Kubernetes-commit: b969dfec9fd33f8bfff47e54f2995a4865839ea6
This commit is contained in:
Alexander Zielenski 2023-01-19 10:04:52 -08:00 committed by Kubernetes Publisher
parent 2ea5662b05
commit 1b8963b016
3 changed files with 172 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

141
pkg/admission/plugin/validatingadmissionpolicy/admission_test.go
View File

@ -26,6 +26,7 @@ import (
"github.com/stretchr/testify/require"
"k8s.io/api/admissionregistration/v1alpha1"
v1 "k8s.io/api/core/v1"
k8serrors "k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/api/meta"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@ -59,6 +60,11 @@ var (
if err := v1alpha1.AddToScheme(res); err != nil {
panic(err)
}
if err := fake.AddToScheme(res); err != nil {
panic(err)
}
return res
}()
paramsGVK schema.GroupVersionKind = schema.GroupVersionKind{
@ -78,6 +84,7 @@ var (
res.Add(paramsGVK, meta.RESTScopeNamespace)
res.Add(definitionGVK, meta.RESTScopeRoot)
res.Add(bindingGVK, meta.RESTScopeRoot)
res.Add(v1.SchemeGroupVersion.WithKind("ConfigMap"), meta.RESTScopeNamespace)
return res
}()
@ -388,30 +395,6 @@ func (c *celAdmissionController) getCurrentObject(obj runtime.Object) (runtime.O
defer c.policyController.mutex.RUnlock()
switch obj.(type) {
case *unstructured.Unstructured:
paramSourceGVK := obj.GetObjectKind().GroupVersionKind()
paramKind := v1alpha1.ParamKind{
APIVersion: paramSourceGVK.GroupVersion().String(),
Kind: paramSourceGVK.Kind,
}
var paramInformer generic.Informer[*unstructured.Unstructured]
if paramInfo, ok := c.policyController.paramsCRDControllers[paramKind]; ok {
paramInformer = paramInfo.controller.Informer()
} else {
// Treat unknown CRD the same as not found
return nil, nil
}
// Param type. Just check informer for its GVK
item, err := paramInformer.Get(accessor.GetName())
if err != nil {
if k8serrors.IsNotFound(err) {
return nil, nil
}
return nil, err
}
return item, nil
case *v1alpha1.ValidatingAdmissionPolicyBinding:
nn := getNamespaceName(accessor.GetNamespace(), accessor.GetName())
info, ok := c.policyController.bindingInfos[nn]
@ -429,7 +412,32 @@ func (c *celAdmissionController) getCurrentObject(obj runtime.Object) (runtime.O
return info.lastReconciledValue, nil
default:
panic(fmt.Errorf("unhandled object type: %T", obj))
// If test isn't trying to fetch a policy or binding, assume it is
// fetching a param
paramSourceGVK := obj.GetObjectKind().GroupVersionKind()
paramKind := v1alpha1.ParamKind{
APIVersion: paramSourceGVK.GroupVersion().String(),
Kind: paramSourceGVK.Kind,
}
var paramInformer generic.Informer[runtime.Object]
if paramInfo, ok := c.policyController.paramsCRDControllers[paramKind]; ok {
paramInformer = paramInfo.controller.Informer()
} else {
// Treat unknown CRD the same as not found
return nil, nil
}
// Param type. Just check informer for its GVK
item, err := paramInformer.Get(accessor.GetName())
if err != nil {
if k8serrors.IsNotFound(err) {
return nil, nil
}
return nil, err
}
return item, nil
}
}
@ -1223,3 +1231,86 @@ func TestMultiplePoliciesSharedParamType(t *testing.T) {
require.EqualValues(t, 1, compiles2.Load())
require.EqualValues(t, 2, evaluations2.Load())
}
// Shows that we can refer to native-typed params just fine
// (as opposed to CRD params)
func TestNativeTypeParam(t *testing.T) {
testContext, testContextCancel := context.WithCancel(context.Background())
defer testContextCancel()
compiler := &fakeCompiler{
// Match everything by default
DefaultMatch: true,
}
handler, _, tracker, controller := setupFakeTest(t, compiler)
compiles := atomic.Int64{}
evaluations := atomic.Int64{}
// Use ConfigMap native-typed param
nativeTypeParamPolicy := *denyPolicy
nativeTypeParamPolicy.Spec.ParamKind = &v1alpha1.ParamKind{
APIVersion: "v1",
Kind: "ConfigMap",
}
compiler.RegisterDefinition(&nativeTypeParamPolicy, func(vap *v1alpha1.ValidatingAdmissionPolicy) Validator {
compiles.Add(1)
return validatorFunc(func(a admission.Attributes, o admission.ObjectInterfaces, params runtime.Object, matchKind schema.GroupVersionKind) ([]policyDecision, error) {
evaluations.Add(1)
// show that the passed params was a ConfigMap native type
if _, ok := params.(*v1.ConfigMap); ok {
return []policyDecision{
{
action: actionDeny,
message: "correct type",
},
}, nil
}
return []policyDecision{
{
action: actionDeny,
message: "Incorrect param type",
},
}, nil
})
}, nil)
configMapParam := &v1.ConfigMap{
TypeMeta: metav1.TypeMeta{
APIVersion: "v1",
Kind: "ConfigMap",
},
ObjectMeta: metav1.ObjectMeta{
Name: "replicas-test.example.com",
Namespace: "",
ResourceVersion: "1",
},
Data: map[string]string{
"coolkey": "coolvalue",
},
}
require.NoError(t, tracker.Create(definitionsGVR, &nativeTypeParamPolicy, nativeTypeParamPolicy.Namespace))
require.NoError(t, tracker.Create(bindingsGVR, denyBinding, denyBinding.Namespace))
require.NoError(t, tracker.Add(configMapParam))
// Wait for controller to reconcile given objects
require.NoError(t,
waitForReconcile(
testContext, controller,
denyBinding, denyPolicy, configMapParam))
err := handler.Validate(
testContext,
// Object is irrelevant/unchecked for this test. Just test that
// the evaluator is executed, and returns admit meaning the params
// passed was a configmap
attributeRecord(nil, fakeParams, admission.Create),
&admission.RuntimeObjectInterfaces{},
)
require.ErrorContains(t, err, "correct type")
require.EqualValues(t, 1, compiles.Load())
require.EqualValues(t, 1, evaluations.Load())
}

9
pkg/admission/plugin/validatingadmissionpolicy/controller.go
View File

@ -25,13 +25,13 @@ import (
"time"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apiserver/pkg/admission/plugin/validatingadmissionpolicy/matching"
"k8s.io/api/admissionregistration/v1alpha1"
k8serrors "k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/api/meta"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/apimachinery/pkg/util/sets"
"k8s.io/apimachinery/pkg/util/wait"
@ -65,7 +65,7 @@ type celAdmissionController struct {
// against all of its registered bindings.
type policyData struct {
definitionInfo
paramController generic.Controller[*unstructured.Unstructured]
paramController generic.Controller[runtime.Object]
bindings []bindingInfo
}
@ -96,7 +96,7 @@ type bindingInfo struct {
type paramInfo struct {
// Controller which is watching this param CRD
controller generic.Controller[*unstructured.Unstructured]
controller generic.Controller[runtime.Object]
// Function to call to stop the informer and clean up the controller
stop func()
@ -116,6 +116,7 @@ func NewAdmissionController(
definitions: atomic.Value{},
policyController: newPolicyController(
restMapper,
client,
dynamicClient,
&CELValidatorCompiler{
Matcher: matching.NewMatcher(informerFactory.Core().V1().Namespaces().Lister(), client),
@ -241,7 +242,7 @@ func (c *celAdmissionController) Validate(
continue
}
var param *unstructured.Unstructured
var param runtime.Object
// If definition has paramKind, paramRef is required in binding.
// If definition has no paramKind, paramRef set in binding will be ignored.

65
pkg/admission/plugin/validatingadmissionpolicy/controller_reconcile.go
View File

@ -25,13 +25,15 @@ import (
"k8s.io/api/admissionregistration/v1alpha1"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/meta"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/apimachinery/pkg/util/sets"
celmetrics "k8s.io/apiserver/pkg/admission/cel"
"k8s.io/apiserver/pkg/admission/plugin/validatingadmissionpolicy/internal/generic"
"k8s.io/client-go/dynamic"
"k8s.io/client-go/dynamic/dynamicinformer"
"k8s.io/client-go/informers"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/tools/cache"
)
@ -74,10 +76,13 @@ type policyController struct {
// All keys must have at least one dependent binding
// All binding names MUST exist as a key bindingInfos
definitionsToBindings map[namespacedName]sets.Set[namespacedName]
client kubernetes.Interface
}
func newPolicyController(
restMapper meta.RESTMapper,
client kubernetes.Interface,
dynamicClient dynamic.Interface,
validatorCompiler ValidatorCompiler,
policiesInformer generic.Informer[*v1alpha1.ValidatingAdmissionPolicy],
@ -108,6 +113,7 @@ func newPolicyController(
),
restMapper: restMapper,
dynamicClient: dynamicClient,
client: client,
}
return res
}
@ -237,18 +243,50 @@ func (c *policyController) reconcilePolicyDefinition(namespace, name string, def
} else {
instanceContext, instanceCancel := context.WithCancel(c.context)
// Watch for new instances of this policy
informer := dynamicinformer.NewFilteredDynamicInformer(
c.dynamicClient,
paramsGVR.Resource,
corev1.NamespaceAll,
30*time.Second, // TODO: do we really need to ever resync these?
cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
nil,
)
var informer informers.GenericInformer
// Informer Factory is optional
if c.client != nil {
// Create temporary informer factory
// Cannot use the k8s shared informer factory for dynamic params informer.
// Would leak unnecessary informers when we are done since we would have to
// call informerFactory.Start() with a longer-lived stopCh than necessary.
// SharedInformerFactory does not support temporary usage.
dynamicFactory := informers.NewSharedInformerFactory(c.client, 10*time.Minute)
// Look for a typed informer. If it does not exist
informer, err = dynamicFactory.ForResource(paramsGVR.Resource)
// Ignore error. We fallback to dynamic informer if there is no
// typed informer
if err != nil {
informer = nil
} else {
dynamicFactory.Start(instanceContext.Done())
}
}
if informer == nil {
// Dynamic JSON informer fallback.
// Cannot use shared dynamic informer since it would be impossible
// to clean CRD informers properly with multiple dependents
// (cannot start ahead of time, and cannot track dependencies via stopCh)
informer = dynamicinformer.NewFilteredDynamicInformer(
c.dynamicClient,
paramsGVR.Resource,
corev1.NamespaceAll,
// Use same interval as is used for k8s typed sharedInformerFactory
// https://github.com/kubernetes/kubernetes/blob/7e0923899fed622efbc8679cca6b000d43633e38/cmd/kube-apiserver/app/server.go#L430
10*time.Minute,
cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
nil,
)
go informer.Informer().Run(instanceContext.Done())
}
controller := generic.NewController(
generic.NewInformer[*unstructured.Unstructured](informer.Informer()),
generic.NewInformer[runtime.Object](informer.Informer()),
c.reconcileParams,
generic.ControllerOptions{
Workers: 1,
@ -262,7 +300,6 @@ func (c *policyController) reconcilePolicyDefinition(namespace, name string, def
dependentDefinitions: sets.New(nn),
}
go informer.Informer().Run(instanceContext.Done())
go controller.Run(instanceContext)
}
@ -329,7 +366,7 @@ func (c *policyController) reconcilePolicyBinding(namespace, name string, bindin
return nil
}
func (c *policyController) reconcileParams(namespace, name string, params *unstructured.Unstructured) error {
func (c *policyController) reconcileParams(namespace, name string, params runtime.Object) error {
// Do nothing.
// When we add informational type checking we will need to compile in the
// reconcile loops instead of lazily so we can add compiler errors / type
@ -365,7 +402,7 @@ func (c *policyController) latestPolicyData() []policyData {
bindingInfos = append(bindingInfos, *bindingInfo)
}
var paramController generic.Controller[*unstructured.Unstructured]
var paramController generic.Controller[runtime.Object]
if paramKind := definitionInfo.lastReconciledValue.Spec.ParamKind; paramKind != nil {
if info, ok := c.paramsCRDControllers[*paramKind]; ok {
paramController = info.controller