diff --git a/staging/src/k8s.io/apiserver/pkg/admission/configuration/mutating_webhook_manager.go b/staging/src/k8s.io/apiserver/pkg/admission/configuration/mutating_webhook_manager.go index fbde83b7d82..3c0990699a4 100644 --- a/staging/src/k8s.io/apiserver/pkg/admission/configuration/mutating_webhook_manager.go +++ b/staging/src/k8s.io/apiserver/pkg/admission/configuration/mutating_webhook_manager.go @@ -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() { diff --git a/staging/src/k8s.io/apiserver/pkg/admission/configuration/mutating_webhook_manager_test.go b/staging/src/k8s.io/apiserver/pkg/admission/configuration/mutating_webhook_manager_test.go index e6b50aa2320..d6f4f1a4512 100644 --- a/staging/src/k8s.io/apiserver/pkg/admission/configuration/mutating_webhook_manager_test.go +++ b/staging/src/k8s.io/apiserver/pkg/admission/configuration/mutating_webhook_manager_test.go @@ -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") + if configurations := manager.Webhooks(); len(configurations.Webhooks) != 0 { + t.Errorf("expected empty webhooks, but got %v", configurations.Webhooks) } - // 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") - } - - // 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) } } diff --git a/staging/src/k8s.io/apiserver/pkg/admission/configuration/validating_webhook_manager.go b/staging/src/k8s.io/apiserver/pkg/admission/configuration/validating_webhook_manager.go index f93068b8037..33644f57fed 100644 --- a/staging/src/k8s.io/apiserver/pkg/admission/configuration/validating_webhook_manager.go +++ b/staging/src/k8s.io/apiserver/pkg/admission/configuration/validating_webhook_manager.go @@ -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() { diff --git a/staging/src/k8s.io/apiserver/pkg/admission/configuration/validating_webhook_manager_test.go b/staging/src/k8s.io/apiserver/pkg/admission/configuration/validating_webhook_manager_test.go index 929d7b2cfcf..6505b2b9b4b 100644 --- a/staging/src/k8s.io/apiserver/pkg/admission/configuration/validating_webhook_manager_test.go +++ b/staging/src/k8s.io/apiserver/pkg/admission/configuration/validating_webhook_manager_test.go @@ -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") + if configurations := manager.Webhooks(); len(configurations.Webhooks) != 0 { + t.Errorf("expected empty webhooks, but got %v", configurations.Webhooks) } - // 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") - } - - // 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) } } diff --git a/staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/mutating/BUILD b/staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/mutating/BUILD index 0d46b5d7627..6ed322b2500 100644 --- a/staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/mutating/BUILD +++ b/staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/mutating/BUILD @@ -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", diff --git a/staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/mutating/admission.go b/staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/mutating/admission.go index 0f50edf6a3e..73e213a6f10 100644 --- a/staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/mutating/admission.go +++ b/staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/mutating/admission.go @@ -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() diff --git a/staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/mutating/admission_test.go b/staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/mutating/admission_test.go index 9f92fad1126..523e03762d0 100644 --- a/staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/mutating/admission_test.go +++ b/staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/mutating/admission_test.go @@ -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 ®istrationv1beta1.MutatingWebhookConfiguration{Webhooks: f.hooks}, nil + return ®istrationv1beta1.MutatingWebhookConfiguration{Webhooks: f.hooks} } func (f *fakeHookSource) Run(stopCh <-chan struct{}) {} diff --git a/staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/validating/BUILD b/staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/validating/BUILD index 4226a13912c..a3b6a155937 100644 --- a/staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/validating/BUILD +++ b/staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/validating/BUILD @@ -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", diff --git a/staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/validating/admission.go b/staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/validating/admission.go index f14d65083de..8e8c7448fcb 100644 --- a/staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/validating/admission.go +++ b/staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/validating/admission.go @@ -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() diff --git a/staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/validating/admission_test.go b/staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/validating/admission_test.go index 9a190f41239..77871f01942 100644 --- a/staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/validating/admission_test.go +++ b/staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/validating/admission_test.go @@ -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 ®istrationv1beta1.ValidatingWebhookConfiguration{Webhooks: f.hooks}, nil + return ®istrationv1beta1.ValidatingWebhookConfiguration{Webhooks: f.hooks} } func (f *fakeHookSource) Run(stopCh <-chan struct{}) {}