diff --git a/pkg/shell-operator/operator.go b/pkg/shell-operator/operator.go index d68bda1a..1b3c556c 100644 --- a/pkg/shell-operator/operator.go +++ b/pkg/shell-operator/operator.go @@ -107,6 +107,9 @@ type ShellOperator struct { // can wait for them to drain. wg sync.WaitGroup + // reloadMu serialises ReloadHooks calls. + reloadMu sync.Mutex + // startOnce / shutdownOnce make Start and Shutdown idempotent. startOnce sync.Once startErr error diff --git a/pkg/shell-operator/reload.go b/pkg/shell-operator/reload.go new file mode 100644 index 00000000..0cab6ca8 --- /dev/null +++ b/pkg/shell-operator/reload.go @@ -0,0 +1,380 @@ +// Copyright 2026 Flant JSC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package shell_operator + +import ( + "context" + "errors" + "fmt" + "log/slog" + "time" + + "github.com/deckhouse/deckhouse/pkg/log" + apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/wait" + + pkg "github.com/flant/shell-operator/pkg" + hook_types "github.com/flant/shell-operator/pkg/hook/types" + "github.com/flant/shell-operator/pkg/webhook/admission" + "github.com/flant/shell-operator/pkg/webhook/conversion" +) + +const ( + enableKubernetesBindingsRetryTimeout = 2 * time.Minute + enableKubernetesBindingsRetrySteps = 30 +) + +// ReloadHooks re-discovers hooks from disk so that the operator picks up new or +// removed hook configurations without a process restart. It is safe for +// concurrent use: calls are serialised, because overlapping HookManager.Init() + +// Enable*Bindings sequences would operate on stale/overwritten hook indices. +// +// HookManager.Init() replaces the hook index atomically, but newly loaded +// hooks have uninitialised AdmissionLinks/ConversionLinks maps. We must +// call EnableAdmissionBindings / EnableConversionBindings on every hook that +// carries validating/mutating/conversion configs so that +// CanHandleAdmissionEvent / CanHandleConversionEvent can match incoming +// requests to the right hook. +// +// AdmissionWebhookManager.Init() and ConversionWebhookManager.Init() are +// deliberately NOT called here because they recreate HTTP servers and would +// either fail with "address already in use" or silently orphan the old +// listeners. As a consequence a webhook server that was never started (no +// validating/mutating hook existed at bootstrap) is not started by a reload +// either — callers that need the server up must have at least one such hook +// on disk before NewShellOperator. +func (op *ShellOperator) ReloadHooks(ctx context.Context) error { + op.reloadMu.Lock() + defer op.reloadMu.Unlock() + + if err := ctx.Err(); err != nil { + return fmt.Errorf("context cancelled before reload: %w", err) + } + + op.logger.Info("reloading hooks") + + if op.HookManager == nil { + return fmt.Errorf("hook manager is not initialized") + } + + // Snapshot the current registrations before any mutation. These are + // compared against the rebuilt state later to determine which + // ValidatingWebhookConfigurations / MutatingWebhookConfigurations / CRD + // conversion settings are stale and must be unregistered. + oldValidatingResources := make(map[string]*admission.ValidatingWebhookResource) + oldMutatingResources := make(map[string]*admission.MutatingWebhookResource) + oldConversionClientConfigs := make(map[string]*conversion.CrdClientConfig) + + if op.AdmissionWebhookManager != nil { + for confID, resource := range op.AdmissionWebhookManager.ValidatingResources { + oldValidatingResources[confID] = resource + } + for confID, resource := range op.AdmissionWebhookManager.MutatingResources { + oldMutatingResources[confID] = resource + } + } + + if op.ConversionWebhookManager != nil { + for crdName, cfg := range op.ConversionWebhookManager.ClientConfigs { + oldConversionClientConfigs[crdName] = cfg + } + } + + // Re-discover hooks from disk. This is the step most likely to fail + // (e.g. malformed hook file), so it MUST happen before we clear the + // current registration maps — otherwise a failed Init leaves the + // operator with empty maps and no way to unregister stale entries. + if err := op.HookManager.Init(); err != nil { + return fmt.Errorf("re-init hook manager: %w", err) + } + + // Clear the current registration state AFTER Init has succeeded. + // Rebuilding from scratch avoids stale webhook resources after hook + // removals. The Enable*Bindings calls below repopulate these maps with + // fresh entries for every discovered hook. + if op.AdmissionWebhookManager != nil { + op.AdmissionWebhookManager.ValidatingResources = make(map[string]*admission.ValidatingWebhookResource) + op.AdmissionWebhookManager.MutatingResources = make(map[string]*admission.MutatingWebhookResource) + } + + if op.ConversionWebhookManager != nil { + op.ConversionWebhookManager.ClientConfigs = make(map[string]*conversion.CrdClientConfig) + } + + // Enable admission bindings on every newly loaded hook so that + // AdmissionLinks are populated and CanHandleEvent works. + validatingHookNames, err := op.HookManager.GetHooksInOrder(hook_types.KubernetesValidating) + if err != nil { + return fmt.Errorf("get validating hooks: %w", err) + } + mutatingHookNames, err := op.HookManager.GetHooksInOrder(hook_types.KubernetesMutating) + if err != nil { + return fmt.Errorf("get mutating hooks: %w", err) + } + for _, name := range append(validatingHookNames, mutatingHookNames...) { + if h := op.HookManager.GetHook(name); h != nil && h.HookController != nil { + h.HookController.EnableAdmissionBindings() + } + } + + // Enable conversion bindings on every newly loaded hook so that + // CanHandleConversionEvent works. + conversionHookNames, err := op.HookManager.GetHooksInOrder(hook_types.KubernetesConversion) + if err != nil { + return fmt.Errorf("get conversion hooks: %w", err) + } + for _, name := range conversionHookNames { + if h := op.HookManager.GetHook(name); h != nil && h.HookController != nil { + h.HookController.EnableConversionBindings() + } + } + + // Re-enable kubernetes bindings on every newly loaded hook so that the + // monitors that maintain object caches (snapshots) are recreated and + // wired to the freshly rebuilt HookController. + // + // HookManager.Init() above replaces every Hook (and therefore every + // HookController) with a brand new instance whose KubernetesController + // has no registered monitors, and every kubernetes binding gets a fresh + // random MonitorId. Admission/conversion hooks that pull data via + // 'includeSnapshotsFrom' resolve their snapshot through the *current* + // hook index, i.e. these freshly rebuilt controllers. If the monitors + // for this new index are not wired, the 'snapshots' field is delivered + // empty and validating webhooks decide on missing data (e.g. always + // denying because a ModuleConfig snapshot looks absent). + // + // This MUST be driven to completion here rather than relying on the + // caller to retry on error. The main queue enables bindings on the + // *startup* hook index, but every ReloadHooks() call replaces that index + // with new MonitorIds; the main queue's monitors are then orphaned + // relative to the current index. If ReloadHooks() returned before wiring + // the current index (e.g. a transient apiserver outage) and no further + // reload were triggered, the current index would stay unwired and + // snapshots empty indefinitely. So enableKubernetesBindings retries with + // backoff until every hook is wired (or ctx/deadline ends). + kubernetesHookNames, err := op.HookManager.GetHooksInOrder(hook_types.OnKubernetesEvent) + if err != nil { + return fmt.Errorf("get kubernetes hooks: %w", err) + } + if err := op.enableKubernetesBindings(ctx, kubernetesHookNames); err != nil { + return err + } + + if err := op.syncAdmissionWebhookConfigurations(ctx, oldValidatingResources, oldMutatingResources); err != nil { + return fmt.Errorf("sync admission webhook configurations: %w", err) + } + + if err := op.syncConversionWebhookConfigurations(ctx, oldConversionClientConfigs); err != nil { + return fmt.Errorf("sync conversion webhook configurations: %w", err) + } + + return nil +} + +// enableKubernetesBindings wires the monitors (snapshot caches) for every +// kubernetes hook in the *current* hook index, retrying hooks that fail with +// bounded backoff until all succeed or the deadline/context ends. +func (op *ShellOperator) enableKubernetesBindings(ctx context.Context, hookNames []string) error { + // Track hooks still needing a successful enable so each is wired exactly + // once regardless of how many retry passes it takes. + pending := make(map[string]struct{}, len(hookNames)) + for _, name := range hookNames { + pending[name] = struct{}{} + } + + // enableOne wires a single hook by name against the *current* hook index. + // Use the same HookController entry point as the main queue, but pass nil + // to avoid creating synchronization hook-run tasks during reload. + enableOne := func(name string) error { + h := op.HookManager.GetHook(name) + if h == nil { + return fmt.Errorf("hook %q not found in hook manager", name) + } + if h.HookController == nil { + return fmt.Errorf("hook controller for hook %q is nil", name) + } + + if err := h.HookController.HandleEnableKubernetesBindings(ctx, nil); err != nil { + // Leave it pending and do NOT unlock events: the cache is not + // filled, so emitting events would be premature. + return err + } + + // Cache is filled; allow the monitors to emit future events. + h.HookController.UnlockKubernetesEvents() + return nil + } + + var lastErr error + attempt := func() bool { + var errs []error + for name := range pending { + if err := enableOne(name); err != nil { + errs = append(errs, fmt.Errorf("enable kubernetes bindings for hook %q: %w", name, err)) + continue + } + delete(pending, name) + } + lastErr = errors.Join(errs...) + return len(pending) == 0 + } + + // First pass: attempt every hook once. Continuing past a single failing + // hook ensures one transient failure cannot starve the rest. + if attempt() { + return nil + } + + op.logger.Warn("some kubernetes hooks failed to enable, retrying until wired", + slog.Int("pending", len(pending)), + log.Err(lastErr)) + + // Retry only the still-pending hooks with capped exponential backoff until + // they are all wired, the per-reload deadline is reached, or ctx is done. + retryCtx, cancel := context.WithTimeout(ctx, enableKubernetesBindingsRetryTimeout) + defer cancel() + + waitErr := wait.ExponentialBackoffWithContext(retryCtx, wait.Backoff{ + Duration: 500 * time.Millisecond, + Factor: 2.0, + Jitter: 0.1, + Steps: enableKubernetesBindingsRetrySteps, + Cap: 10 * time.Second, + }, func(context.Context) (bool, error) { + return attempt(), nil + }) + if waitErr != nil { + // Deadline or context ended with hooks still unwired. Prefer the + // underlying enable errors so the caller retries with the real cause; + // fall back to the wait error if there is no enable error. + if lastErr == nil { + lastErr = waitErr + } + return fmt.Errorf("enable kubernetes bindings (%d hook(s) still unwired after %s): %w", + len(pending), enableKubernetesBindingsRetryTimeout, lastErr) + } + + op.logger.Info("all kubernetes hooks enabled and wired") + return nil +} + +// syncAdmissionWebhookConfigurations (re)registers the rebuilt +// Validating/MutatingWebhookConfigurations and removes the ones that no +// longer have a hook behind them. +func (op *ShellOperator) syncAdmissionWebhookConfigurations( + ctx context.Context, + oldValidatingResources map[string]*admission.ValidatingWebhookResource, + oldMutatingResources map[string]*admission.MutatingWebhookResource, +) error { + if op.AdmissionWebhookManager == nil { + return nil + } + + for confID, resource := range op.AdmissionWebhookManager.ValidatingResources { + if err := resource.Register(ctx); err != nil { + return fmt.Errorf("register validating webhook configuration %q: %w", confID, err) + } + } + + for confID, resource := range op.AdmissionWebhookManager.MutatingResources { + if err := resource.Register(ctx); err != nil { + return fmt.Errorf("register mutating webhook configuration %q: %w", confID, err) + } + } + + for confID, resource := range oldValidatingResources { + if _, stillPresent := op.AdmissionWebhookManager.ValidatingResources[confID]; stillPresent { + continue + } + + if err := resource.Unregister(); err != nil && !apierrors.IsNotFound(err) { + return fmt.Errorf("delete stale validating webhook configuration %q: %w", confID, err) + } + } + + for confID, resource := range oldMutatingResources { + if _, stillPresent := op.AdmissionWebhookManager.MutatingResources[confID]; stillPresent { + continue + } + + if err := resource.Unregister(); err != nil && !apierrors.IsNotFound(err) { + return fmt.Errorf("delete stale mutating webhook configuration %q: %w", confID, err) + } + } + + return nil +} + +// syncConversionWebhookConfigurations points every CRD that still has a +// conversion hook at this operator's webhook and resets the conversion +// strategy of CRDs whose hook disappeared. +func (op *ShellOperator) syncConversionWebhookConfigurations( + ctx context.Context, + oldConversionClientConfigs map[string]*conversion.CrdClientConfig, +) error { + if op.ConversionWebhookManager == nil { + return nil + } + + for crdName, cfg := range op.ConversionWebhookManager.ClientConfigs { + if err := cfg.Update(ctx); err != nil { + return fmt.Errorf("update conversion client config for crd %q: %w", crdName, err) + } + } + + for crdName := range oldConversionClientConfigs { + if _, stillPresent := op.ConversionWebhookManager.ClientConfigs[crdName]; stillPresent { + continue + } + + if err := op.resetCRDConversionToNone(ctx, crdName); err != nil { + return fmt.Errorf("cleanup stale conversion webhook config for crd %q: %w", crdName, err) + } + } + + return nil +} + +// resetCRDConversionToNone switches a CRD back to the None conversion strategy +// so the apiserver stops calling a webhook that no longer has a hook behind it. +func (op *ShellOperator) resetCRDConversionToNone(ctx context.Context, crdName string) error { + if op.KubeClient == nil { + return fmt.Errorf("kubernetes client is not initialized") + } + + crd, err := op.KubeClient.ApiExt().CustomResourceDefinitions().Get(ctx, crdName, metav1.GetOptions{}) + if err != nil { + if apierrors.IsNotFound(err) { + return nil + } + return fmt.Errorf("get CRD %q: %w", crdName, err) + } + + if crd.Spec.Conversion == nil || crd.Spec.Conversion.Strategy != apiextensionsv1.WebhookConverter { + return nil + } + + crd.Spec.Conversion = &apiextensionsv1.CustomResourceConversion{ + Strategy: apiextensionsv1.NoneConverter, + } + + if _, err := op.KubeClient.ApiExt().CustomResourceDefinitions().Update(ctx, crd, pkg.DefaultUpdateOptions()); err != nil { + return fmt.Errorf("update CRD %q conversion strategy: %w", crdName, err) + } + + return nil +} diff --git a/pkg/shell-operator/reload_test.go b/pkg/shell-operator/reload_test.go new file mode 100644 index 00000000..0144a81b --- /dev/null +++ b/pkg/shell-operator/reload_test.go @@ -0,0 +1,83 @@ +package shell_operator + +import ( + "context" + "fmt" + "testing" + + "github.com/deckhouse/deckhouse/pkg/log" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + admissionregv1 "k8s.io/api/admissionregistration/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + klient "github.com/flant/kube-client/client" + "github.com/flant/shell-operator/pkg/webhook/admission" +) + +func newReloadTestOperator(t *testing.T, hm *stubHookManager) (*ShellOperator, *klient.Client) { + t.Helper() + + kubeClient := klient.NewFake(nil) + op := NewBareShellOperator(context.Background(), WithLogger(log.NewNop())) + op.HookManager = hm + op.KubeClient = kubeClient + op.AdmissionWebhookManager = admission.NewWebhookManager(kubeClient, admission.WithLogger(log.NewNop())) + + return op, kubeClient +} + +func addValidatingResource(t *testing.T, op *ShellOperator, kubeClient *klient.Client, confID, name string) { + t.Helper() + + _, err := kubeClient.AdmissionregistrationV1().ValidatingWebhookConfigurations(). + Create(context.Background(), &admissionregv1.ValidatingWebhookConfiguration{ + ObjectMeta: metav1.ObjectMeta{Name: name}, + }, metav1.CreateOptions{}) + require.NoError(t, err) + + op.AdmissionWebhookManager.ValidatingResources[confID] = admission.NewValidatingWebhookResource( + admission.WebhookResourceOptions{ + KubeClient: kubeClient, + ConfigurationName: name, + }, + log.NewNop(), + ) +} + +// A failing HookManager.Init must leave the current registrations intact: +// clearing them first would strand the live webhook configurations with no +// way to unregister them. +func TestReloadHooksKeepsRegistrationsWhenInitFails(t *testing.T) { + op, kubeClient := newReloadTestOperator(t, &stubHookManager{ + initFunc: func() error { return fmt.Errorf("malformed hook") }, + }) + addValidatingResource(t, op, kubeClient, "hooks", "test-hooks") + + err := op.ReloadHooks(context.Background()) + + require.ErrorContains(t, err, "re-init hook manager") + assert.Len(t, op.AdmissionWebhookManager.ValidatingResources, 1) +} + +// After a successful reload that discovers no validating hooks, the previously +// registered ValidatingWebhookConfiguration must be deleted from the cluster. +func TestReloadHooksUnregistersStaleValidatingWebhook(t *testing.T) { + op, kubeClient := newReloadTestOperator(t, &stubHookManager{}) + addValidatingResource(t, op, kubeClient, "hooks", "test-hooks") + + require.NoError(t, op.ReloadHooks(context.Background())) + + assert.Empty(t, op.AdmissionWebhookManager.ValidatingResources) + + _, err := kubeClient.AdmissionregistrationV1().ValidatingWebhookConfigurations(). + Get(context.Background(), "test-hooks", metav1.GetOptions{}) + assert.True(t, apierrors.IsNotFound(err), "stale configuration must be deleted, got %v", err) +} + +func TestReloadHooksWithoutHookManager(t *testing.T) { + op := NewBareShellOperator(context.Background(), WithLogger(log.NewNop())) + + require.ErrorContains(t, op.ReloadHooks(context.Background()), "hook manager is not initialized") +} diff --git a/pkg/shell-operator/webhook_handlers_test.go b/pkg/shell-operator/webhook_handlers_test.go index b32ba330..214b94c5 100644 --- a/pkg/shell-operator/webhook_handlers_test.go +++ b/pkg/shell-operator/webhook_handlers_test.go @@ -29,9 +29,15 @@ type stubHookManager struct { handleAdmissionEventFunc func(ctx context.Context, event admission.Event, createTaskFn func(*hook.Hook, controller.BindingExecutionInfo)) findConversionChainFunc func(crdName string, rule conversion.Rule) []conversion.Rule handleConversionEventFunc func(ctx context.Context, crdName string, request *v1.ConversionRequest, rule conversion.Rule, createTaskFn func(*hook.Hook, controller.BindingExecutionInfo)) + initFunc func() error } -func (s *stubHookManager) Init() error { return nil } +func (s *stubHookManager) Init() error { + if s.initFunc != nil { + return s.initFunc() + } + return nil +} func (s *stubHookManager) GetHook(name string) *hook.Hook { return nil } func (s *stubHookManager) GetHookNames() []string { return nil } func (s *stubHookManager) GetHooksInOrder(htypes.BindingType) ([]string, error) {