Skip to content

K
k3s
Commit ec392597 authored by hzxuzhonghu's avatar hzxuzhonghu
Browse files
Options

add wait ready for mutating/validating webhook configuration

parent f2875274
Hide whitespace changes
Inline Side-by-side
Showing with 72 additions and 136 deletions
staging/src/k8s.io/apiserver/pkg/admission/configuration/mutating_webhook_manager.go
View file @ ec392597
......@@ -31,17 +31,13 @@ import (
// MutatingWebhookConfigurationManager collects the mutating webhook objects so that they can be called.
type MutatingWebhookConfigurationManager struct {
ready int32
configuration *atomic.Value
hasSynced func() bool
lister admissionregistrationlisters.MutatingWebhookConfigurationLister
}
func NewMutatingWebhookConfigurationManager(informer admissionregistrationinformers.MutatingWebhookConfigurationInformer) *MutatingWebhookConfigurationManager {
manager := &MutatingWebhookConfigurationManager{
ready: 0,
configuration: &atomic.Value{},
hasSynced: informer.Informer().HasSynced,
lister: informer.Lister(),
}
......@@ -59,16 +55,8 @@ func NewMutatingWebhookConfigurationManager(informer admissionregistrationinform
}
// Webhooks returns the merged MutatingWebhookConfiguration.
func (m *MutatingWebhookConfigurationManager) Webhooks() (*v1beta1.MutatingWebhookConfiguration, error) {
if atomic.LoadInt32(&m.ready) == 0 {
if !m.hasSynced() {
// Return an error until we've synced
return nil, fmt.Errorf("mutating webhook configuration is not ready")
}
// Remember we're ready
atomic.StoreInt32(&m.ready, 1)
}
return m.configuration.Load().(*v1beta1.MutatingWebhookConfiguration), nil
func (m *MutatingWebhookConfigurationManager) Webhooks() *v1beta1.MutatingWebhookConfiguration {
return m.configuration.Load().(*v1beta1.MutatingWebhookConfiguration)
}
func (m *MutatingWebhookConfigurationManager) updateConfiguration() {
......
staging/src/k8s.io/apiserver/pkg/admission/configuration/mutating_webhook_manager_test.go
View file @ ec392597
......@@ -43,7 +43,6 @@ func (f *fakeMutatingWebhookConfigSharedInformer) Lister() admissionregistration
type fakeMutatingWebhookConfigInformer struct {
eventHandler cache.ResourceEventHandler
hasSynced bool
}
func (f *fakeMutatingWebhookConfigInformer) AddEventHandler(handler cache.ResourceEventHandler) {
......@@ -63,7 +62,7 @@ func (f *fakeMutatingWebhookConfigInformer) Run(stopCh <-chan struct{}) {
panic("unsupported")
}
func (f *fakeMutatingWebhookConfigInformer) HasSynced() bool {
return f.hasSynced
panic("unsupported")
}
func (f *fakeMutatingWebhookConfigInformer) LastSyncResourceVersion() string {
panic("unsupported")
......@@ -92,43 +91,33 @@ func TestGetMutatingWebhookConfig(t *testing.T) {
lister: &fakeMutatingWebhookConfigLister{},
}
// unsynced, error retrieving list
informer.informer.hasSynced = false
// no configurations
informer.lister.list = nil
informer.lister.err = fmt.Errorf("mutating webhook configuration is not ready")
manager := NewMutatingWebhookConfigurationManager(informer)
if _, err := manager.Webhooks(); err == nil {
t.Errorf("expected err, but got none")
}
// list found, still unsynced
informer.informer.hasSynced = false
informer.lister.list = []*v1beta1.MutatingWebhookConfiguration{}
informer.lister.err = nil
if _, err := manager.Webhooks(); err == nil {
t.Errorf("expected err, but got none")
if configurations := manager.Webhooks(); len(configurations.Webhooks) != 0 {
t.Errorf("expected empty webhooks, but got %v", configurations.Webhooks)
}
// items populated, still unsynced
webhookContainer := &v1beta1.MutatingWebhookConfiguration{
// list err
webhookConfiguration := &v1beta1.MutatingWebhookConfiguration{
ObjectMeta: metav1.ObjectMeta{Name: "webhook1"},
Webhooks: []v1beta1.Webhook{{Name: "webhook1.1"}},
}
informer.informer.hasSynced = false
informer.lister.list = []*v1beta1.MutatingWebhookConfiguration{webhookContainer.DeepCopy()}
informer.lister.err = nil
informer.informer.eventHandler.OnAdd(webhookContainer.DeepCopy())
if _, err := manager.Webhooks(); err == nil {
t.Errorf("expected err, but got none")
informer.lister.list = []*v1beta1.MutatingWebhookConfiguration{webhookConfiguration.DeepCopy()}
informer.lister.err = fmt.Errorf("mutating webhook configuration list error")
informer.informer.eventHandler.OnAdd(webhookConfiguration.DeepCopy())
if configurations := manager.Webhooks(); len(configurations.Webhooks) != 0 {
t.Errorf("expected empty webhooks, but got %v", configurations.Webhooks)
}
// sync completed
informer.informer.hasSynced = true
hooks, err := manager.Webhooks()
if err != nil {
t.Errorf("unexpected err: %v", err)
// configuration populated
informer.lister.err = nil
informer.informer.eventHandler.OnAdd(webhookConfiguration.DeepCopy())
configurations := manager.Webhooks()
if len(configurations.Webhooks) == 0 {
t.Errorf("expected non empty webhooks")
}
if !reflect.DeepEqual(hooks.Webhooks, webhookContainer.Webhooks) {
t.Errorf("Expected\n%#v\ngot\n%#v", webhookContainer.Webhooks, hooks.Webhooks)
if !reflect.DeepEqual(configurations.Webhooks, webhookConfiguration.Webhooks) {
t.Errorf("Expected\n%#v\ngot\n%#v", webhookConfiguration.Webhooks, configurations.Webhooks)
}
}
staging/src/k8s.io/apiserver/pkg/admission/configuration/validating_webhook_manager.go
View file @ ec392597
......@@ -31,17 +31,13 @@ import (
// ValidatingWebhookConfigurationManager collects the validating webhook objects so that they can be called.
type ValidatingWebhookConfigurationManager struct {
ready int32
configuration *atomic.Value
hasSynced func() bool
lister admissionregistrationlisters.ValidatingWebhookConfigurationLister
}
func NewValidatingWebhookConfigurationManager(informer admissionregistrationinformers.ValidatingWebhookConfigurationInformer) *ValidatingWebhookConfigurationManager {
manager := &ValidatingWebhookConfigurationManager{
ready: 0,
configuration: &atomic.Value{},
hasSynced: informer.Informer().HasSynced,
lister: informer.Lister(),
}
......@@ -59,16 +55,8 @@ func NewValidatingWebhookConfigurationManager(informer admissionregistrationinfo
}
// Webhooks returns the merged ValidatingWebhookConfiguration.
func (v *ValidatingWebhookConfigurationManager) Webhooks() (*v1beta1.ValidatingWebhookConfiguration, error) {
if atomic.LoadInt32(&v.ready) == 0 {
if !v.hasSynced() {
// Return an error until we've synced
return nil, fmt.Errorf("validating webhook configuration is not ready")
}
// Remember we're ready
atomic.StoreInt32(&v.ready, 1)
}
return v.configuration.Load().(*v1beta1.ValidatingWebhookConfiguration), nil
func (v *ValidatingWebhookConfigurationManager) Webhooks() *v1beta1.ValidatingWebhookConfiguration {
return v.configuration.Load().(*v1beta1.ValidatingWebhookConfiguration)
}
func (v *ValidatingWebhookConfigurationManager) updateConfiguration() {
......
staging/src/k8s.io/apiserver/pkg/admission/configuration/validating_webhook_manager_test.go
View file @ ec392597
......@@ -43,7 +43,6 @@ func (f *fakeValidatingWebhookConfigSharedInformer) Lister() admissionregistrati
type fakeValidatingWebhookConfigInformer struct {
eventHandler cache.ResourceEventHandler
hasSynced bool
}
func (f *fakeValidatingWebhookConfigInformer) AddEventHandler(handler cache.ResourceEventHandler) {
......@@ -63,7 +62,7 @@ func (f *fakeValidatingWebhookConfigInformer) Run(stopCh <-chan struct{}) {
panic("unsupported")
}
func (f *fakeValidatingWebhookConfigInformer) HasSynced() bool {
return f.hasSynced
panic("unsupported")
}
func (f *fakeValidatingWebhookConfigInformer) LastSyncResourceVersion() string {
panic("unsupported")
......@@ -92,43 +91,33 @@ func TestGettValidatingWebhookConfig(t *testing.T) {
lister: &fakeValidatingWebhookConfigLister{},
}
// unsynced, error retrieving list
informer.informer.hasSynced = false
// no configurations
informer.lister.list = nil
informer.lister.err = fmt.Errorf("validating webhook configuration is not ready")
manager := NewValidatingWebhookConfigurationManager(informer)
if _, err := manager.Webhooks(); err == nil {
t.Errorf("expected err, but got none")
}
// list found, still unsynced
informer.informer.hasSynced = false
informer.lister.list = []*v1beta1.ValidatingWebhookConfiguration{}
informer.lister.err = nil
if _, err := manager.Webhooks(); err == nil {
t.Errorf("expected err, but got none")
if configurations := manager.Webhooks(); len(configurations.Webhooks) != 0 {
t.Errorf("expected empty webhooks, but got %v", configurations.Webhooks)
}
// items populated, still unsynced
webhookContainer := &v1beta1.ValidatingWebhookConfiguration{
// list error
webhookConfiguration := &v1beta1.ValidatingWebhookConfiguration{
ObjectMeta: metav1.ObjectMeta{Name: "webhook1"},
Webhooks: []v1beta1.Webhook{{Name: "webhook1.1"}},
}
informer.informer.hasSynced = false
informer.lister.list = []*v1beta1.ValidatingWebhookConfiguration{webhookContainer.DeepCopy()}
informer.lister.err = nil
informer.informer.eventHandler.OnAdd(webhookContainer.DeepCopy())
if _, err := manager.Webhooks(); err == nil {
t.Errorf("expected err, but got none")
informer.lister.list = []*v1beta1.ValidatingWebhookConfiguration{webhookConfiguration.DeepCopy()}
informer.lister.err = fmt.Errorf("validating webhook configuration list error")
informer.informer.eventHandler.OnAdd(webhookConfiguration.DeepCopy())
if configurations := manager.Webhooks(); len(configurations.Webhooks) != 0 {
t.Errorf("expected empty webhooks, but got %v", configurations.Webhooks)
}
// sync completed
informer.informer.hasSynced = true
hooks, err := manager.Webhooks()
if err != nil {
t.Errorf("unexpected err: %v", err)
// configuration populated
informer.lister.err = nil
informer.informer.eventHandler.OnAdd(webhookConfiguration.DeepCopy())
configurations := manager.Webhooks()
if len(configurations.Webhooks) == 0 {
t.Errorf("expected non empty webhooks")
}
if !reflect.DeepEqual(hooks.Webhooks, webhookContainer.Webhooks) {
t.Errorf("Expected\n%#v\ngot\n%#v", webhookContainer.Webhooks, hooks.Webhooks)
if !reflect.DeepEqual(configurations.Webhooks, webhookConfiguration.Webhooks) {
t.Errorf("Expected\n%#v\ngot\n%#v", webhookConfiguration.Webhooks, configurations.Webhooks)
}
}
staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/mutating/BUILD
View file @ ec392597
......@@ -14,7 +14,6 @@ go_library(
"//vendor/k8s.io/api/admission/v1beta1:go_default_library",
"//vendor/k8s.io/api/admissionregistration/v1beta1:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/api/errors:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/runtime/serializer:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/runtime/serializer/json:go_default_library",
......
staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/mutating/admission.go
View file @ ec392597
......@@ -30,7 +30,6 @@ import (
admissionv1beta1 "k8s.io/api/admission/v1beta1"
"k8s.io/api/admissionregistration/v1beta1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/serializer"
"k8s.io/apimachinery/pkg/runtime/serializer/json"
......@@ -68,7 +67,7 @@ func Register(plugins *admission.Plugins) {
// WebhookSource can list dynamic webhook plugins.
type WebhookSource interface {
Webhooks() (*v1beta1.MutatingWebhookConfiguration, error)
Webhooks() *v1beta1.MutatingWebhookConfiguration
}
// NewMutatingWebhook returns a generic admission webhook plugin.
......@@ -150,8 +149,11 @@ func (a *MutatingWebhook) SetExternalKubeClientSet(client clientset.Interface) {
func (a *MutatingWebhook) SetExternalKubeInformerFactory(f informers.SharedInformerFactory) {
namespaceInformer := f.Core().V1().Namespaces()
a.namespaceMatcher.NamespaceLister = namespaceInformer.Lister()
a.SetReadyFunc(namespaceInformer.Informer().HasSynced)
a.hookSource = configuration.NewMutatingWebhookConfigurationManager(f.Admissionregistration().V1beta1().MutatingWebhookConfigurations())
mutatingWebhookConfigurationsInformer := f.Admissionregistration().V1beta1().MutatingWebhookConfigurations()
a.hookSource = configuration.NewMutatingWebhookConfigurationManager(mutatingWebhookConfigurationsInformer)
a.SetReadyFunc(func() bool {
return namespaceInformer.Informer().HasSynced() && mutatingWebhookConfigurationsInformer.Informer().HasSynced()
})
}
// ValidateInitialization implements the InitializationValidator interface.
......@@ -177,27 +179,18 @@ func (a *MutatingWebhook) ValidateInitialization() error {
return nil
}
func (a *MutatingWebhook) loadConfiguration(attr admission.Attributes) (*v1beta1.MutatingWebhookConfiguration, error) {
hookConfig, err := a.hookSource.Webhooks()
if err != nil {
e := apierrors.NewServerTimeout(attr.GetResource().GroupResource(), string(attr.GetOperation()), 1)
e.ErrStatus.Message = fmt.Sprintf("Unable to refresh the Webhook configuration: %v", err)
e.ErrStatus.Reason = "LoadingConfiguration"
e.ErrStatus.Details.Causes = append(e.ErrStatus.Details.Causes, metav1.StatusCause{
Type: "MutatingWebhookConfigurationFailure",
Message: "An error has occurred while refreshing the MutatingWebhook configuration, no resources can be created/updated/deleted/connected until a refresh succeeds.",
})
return nil, e
}
return hookConfig, nil
func (a *MutatingWebhook) loadConfiguration(attr admission.Attributes) *v1beta1.MutatingWebhookConfiguration {
hookConfig := a.hookSource.Webhooks()
return hookConfig
}
// Admit makes an admission decision based on the request attributes.
func (a *MutatingWebhook) Admit(attr admission.Attributes) error {
hookConfig, err := a.loadConfiguration(attr)
if err != nil {
return err
if !a.WaitForReady() {
return admission.NewForbidden(attr, fmt.Errorf("not yet ready to handle request"))
}
hookConfig := a.loadConfiguration(attr)
hooks := hookConfig.Webhooks
ctx := context.TODO()
......
staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/mutating/admission_test.go
View file @ ec392597
......@@ -47,16 +47,16 @@ type fakeHookSource struct {
err error
}
func (f *fakeHookSource) Webhooks() (*registrationv1beta1.MutatingWebhookConfiguration, error) {
func (f *fakeHookSource) Webhooks() *registrationv1beta1.MutatingWebhookConfiguration {
if f.err != nil {
return nil, f.err
return nil
}
for i, h := range f.hooks {
if h.NamespaceSelector == nil {
f.hooks[i].NamespaceSelector = &metav1.LabelSelector{}
}
}
return &registrationv1beta1.MutatingWebhookConfiguration{Webhooks: f.hooks}, nil
return &registrationv1beta1.MutatingWebhookConfiguration{Webhooks: f.hooks}
}
func (f *fakeHookSource) Run(stopCh <-chan struct{}) {}
......
staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/validating/BUILD
View file @ ec392597
......@@ -13,7 +13,6 @@ go_library(
"//vendor/k8s.io/api/admission/v1beta1:go_default_library",
"//vendor/k8s.io/api/admissionregistration/v1beta1:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/api/errors:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/runtime/serializer:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/util/runtime:go_default_library",
......
staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/validating/admission.go
View file @ ec392597
......@@ -30,7 +30,6 @@ import (
admissionv1beta1 "k8s.io/api/admission/v1beta1"
"k8s.io/api/admissionregistration/v1beta1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/serializer"
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
......@@ -67,7 +66,7 @@ func Register(plugins *admission.Plugins) {
// WebhookSource can list dynamic webhook plugins.
type WebhookSource interface {
Webhooks() (*v1beta1.ValidatingWebhookConfiguration, error)
Webhooks() *v1beta1.ValidatingWebhookConfiguration
}
// NewValidatingAdmissionWebhook returns a generic admission webhook plugin.
......@@ -145,8 +144,11 @@ func (a *ValidatingAdmissionWebhook) SetExternalKubeClientSet(client clientset.I
func (a *ValidatingAdmissionWebhook) SetExternalKubeInformerFactory(f informers.SharedInformerFactory) {
namespaceInformer := f.Core().V1().Namespaces()
a.namespaceMatcher.NamespaceLister = namespaceInformer.Lister()
a.SetReadyFunc(namespaceInformer.Informer().HasSynced)
a.hookSource = configuration.NewValidatingWebhookConfigurationManager(f.Admissionregistration().V1beta1().ValidatingWebhookConfigurations())
validatingWebhookConfigurationsInformer := f.Admissionregistration().V1beta1().ValidatingWebhookConfigurations()
a.hookSource = configuration.NewValidatingWebhookConfigurationManager(validatingWebhookConfigurationsInformer)
a.SetReadyFunc(func() bool {
return namespaceInformer.Informer().HasSynced() && validatingWebhookConfigurationsInformer.Informer().HasSynced()
})
}
// ValidateInitialization implements the InitializationValidator interface.
......@@ -166,27 +168,16 @@ func (a *ValidatingAdmissionWebhook) ValidateInitialization() error {
return nil
}
func (a *ValidatingAdmissionWebhook) loadConfiguration(attr admission.Attributes) (*v1beta1.ValidatingWebhookConfiguration, error) {
hookConfig, err := a.hookSource.Webhooks()
if err != nil {
e := apierrors.NewServerTimeout(attr.GetResource().GroupResource(), string(attr.GetOperation()), 1)
e.ErrStatus.Message = fmt.Sprintf("Unable to refresh the Webhook configuration: %v", err)
e.ErrStatus.Reason = "LoadingConfiguration"
e.ErrStatus.Details.Causes = append(e.ErrStatus.Details.Causes, metav1.StatusCause{
Type: "ValidatingWebhookConfigurationFailure",
Message: "An error has occurred while refreshing the ValidatingWebhook configuration, no resources can be created/updated/deleted/connected until a refresh succeeds.",
})
return nil, e
}
return hookConfig, nil
func (a *ValidatingAdmissionWebhook) loadConfiguration(attr admission.Attributes) *v1beta1.ValidatingWebhookConfiguration {
return a.hookSource.Webhooks()
}
// Validate makes an admission decision based on the request attributes.
func (a *ValidatingAdmissionWebhook) Validate(attr admission.Attributes) error {
hookConfig, err := a.loadConfiguration(attr)
if err != nil {
return err
if !a.WaitForReady() {
return admission.NewForbidden(attr, fmt.Errorf("not yet ready to handle request"))
}
hookConfig := a.loadConfiguration(attr)
hooks := hookConfig.Webhooks
ctx := context.TODO()
......
staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/validating/admission_test.go
View file @ ec392597
......@@ -47,16 +47,16 @@ type fakeHookSource struct {
err error
}
func (f *fakeHookSource) Webhooks() (*registrationv1beta1.ValidatingWebhookConfiguration, error) {
func (f *fakeHookSource) Webhooks() *registrationv1beta1.ValidatingWebhookConfiguration {
if f.err != nil {
return nil, f.err
return nil
}
for i, h := range f.hooks {
if h.NamespaceSelector == nil {
f.hooks[i].NamespaceSelector = &metav1.LabelSelector{}
}
}
return &registrationv1beta1.ValidatingWebhookConfiguration{Webhooks: f.hooks}, nil
return &registrationv1beta1.ValidatingWebhookConfiguration{Webhooks: f.hooks}
}
func (f *fakeHookSource) Run(stopCh <-chan struct{}) {}
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment