diff --git a/cmd/kube-mgmt/main.go b/cmd/kube-mgmt/main.go index 2621693a..08bef0a0 100644 --- a/cmd/kube-mgmt/main.go +++ b/cmd/kube-mgmt/main.go @@ -13,6 +13,7 @@ import ( "os" "path" "strings" + "time" "github.com/open-policy-agent/kube-mgmt/pkg/configmap" "github.com/open-policy-agent/kube-mgmt/pkg/data" @@ -54,6 +55,7 @@ type params struct { replicateIgnoreNs []string analysisEntrypoint string healthEndpoint string + opaResyncInterval time.Duration } func main() { @@ -104,6 +106,7 @@ func main() { rootCmd.Flags().StringVarP(¶ms.opaConfigFile, "opa-config", "", "", "set file containing OPA configuration to enable data replication based on configured bundles") rootCmd.Flags().StringVarP(¶ms.analysisEntrypoint, "analysis-entrypoint", "", "main/main", "set decision to analyze for dynamic data replication configuration (requires --opa-config)") rootCmd.Flags().StringVarP(¶ms.healthEndpoint, "health-endpoint", "", "", "set health check listening endpoint (e.g., localhost:8000)") + rootCmd.Flags().DurationVarP(¶ms.opaResyncInterval, "opa-resync-interval", "", 15*time.Second, "how often to check OPA health and resync all data if OPA restarted (0 to disable)") rootCmd.PersistentPreRunE = func(cmd *cobra.Command, args []string) error { if rootCmd.Flag("policy-label").Value.String() != "" || rootCmd.Flag("policy-value").Value.String() != "" { @@ -183,8 +186,9 @@ func run(params *params) { http.DefaultTransport.(*http.Transport).TLSClientConfig = config } + var cmSync *configmap.Sync if params.enablePolicies || params.enableData { - sync := configmap.New( + cmSync = configmap.New( kubeconfig, opa.New(params.opaURL, params.opaAuth), configmap.DefaultConfigMapMatcher( @@ -197,31 +201,34 @@ func run(params *params) { params.dataValue, ), ) - _, err = sync.Run(params.namespaces) + _, err = cmSync.Run(params.namespaces) if err != nil { logrus.Fatalf("Failed to start configmap sync: %v", err) } } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + var genericSyncs []*data.GenericSync if len(params.replicateCluster)+len(params.replicateNamespace) > 0 { client, err := dynamic.NewForConfig(kubeconfig) if err != nil { logrus.Fatalf("Failed to get dynamic client: %v", err) } - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - opts := data.WithIgnoreNamespaces(params.replicateIgnoreNs) for _, gvk := range params.replicateCluster { - sync := data.NewFromInterface(client, opa.New(params.opaURL, params.opaAuth).Prefix(params.replicatePath), getResourceType(gvk, false), opts) - go sync.RunContext(ctx) + s := data.NewFromInterface(client, opa.New(params.opaURL, params.opaAuth).Prefix(params.replicatePath), getResourceType(gvk, false), opts) + genericSyncs = append(genericSyncs, s) + go s.RunContext(ctx) } for _, gvk := range params.replicateNamespace { - sync := data.NewFromInterface(client, opa.New(params.opaURL, params.opaAuth).Prefix(params.replicatePath), getResourceType(gvk, true), opts) - go sync.RunContext(ctx) + s := data.NewFromInterface(client, opa.New(params.opaURL, params.opaAuth).Prefix(params.replicatePath), getResourceType(gvk, true), opts) + genericSyncs = append(genericSyncs, s) + go s.RunContext(ctx) } } @@ -241,7 +248,12 @@ func run(params *params) { if err != nil { logrus.Fatalf("Failed to create dynamic synchronizer: %v", err) } - go sync.Run(context.Background()) + go sync.Run(ctx) + } + + if params.opaResyncInterval > 0 { + healthURL := strings.TrimSuffix(params.opaURL, "/v1") + "/health" + go watchOPAHealth(ctx, healthURL, params.opaResyncInterval, cmSync, genericSyncs) } if params.healthEndpoint != "" { @@ -272,6 +284,43 @@ func run(params *params) { <-quit } +// watchOPAHealth polls OPA's /health endpoint and triggers a full re-sync of +// all sync subsystems whenever OPA recovers after being unreachable. This +// handles the case where the OPA container restarts (e.g. due to OOM) while +// kube-mgmt is idle and would not otherwise detect the data loss. +func watchOPAHealth(ctx context.Context, healthURL string, interval time.Duration, cmSync *configmap.Sync, genericSyncs []*data.GenericSync) { + ticker := time.NewTicker(interval) + defer ticker.Stop() + wasDown := false + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + resp, err := http.Get(healthURL) //nolint:noctx // intentional fire-and-forget health probe + up := err == nil && resp != nil && resp.StatusCode == http.StatusOK + if resp != nil { + resp.Body.Close() + } + if !up { + if !wasDown { + logrus.Warnf("OPA health check failed (%v), will resync on recovery", healthURL) + } + wasDown = true + } else if wasDown { + logrus.Infof("OPA recovered, triggering full resync") + if cmSync != nil { + cmSync.Resync() + } + for _, s := range genericSyncs { + s.ForceResync() + } + wasDown = false + } + } + } +} + func loadRESTConfig(path string) (*rest.Config, error) { if path != "" { return clientcmd.BuildConfigFromFlags("", path) diff --git a/pkg/configmap/configmap.go b/pkg/configmap/configmap.go index 7b6699bf..c80e2773 100644 --- a/pkg/configmap/configmap.go +++ b/pkg/configmap/configmap.go @@ -12,6 +12,7 @@ import ( "sort" "strconv" "strings" + "sync" "time" "github.com/open-policy-agent/kube-mgmt/pkg/opa" @@ -93,6 +94,8 @@ type Sync struct { opa opa.Client clientset *kubernetes.Clientset matcher func(*v1.ConfigMap) (bool, bool) + mu sync.Mutex + stores []cache.Store // one per namespace watcher, populated by Run() } // New returns a new Sync that can be started. @@ -145,7 +148,7 @@ func (s *Sync) Run(namespaces []string) (chan struct{}, error) { "configmaps", namespace, fields.Everything()) - _, controller := cache.NewInformerWithOptions(cache.InformerOptions{ + store, controller := cache.NewInformerWithOptions(cache.InformerOptions{ ListerWatcher: listerWatcher, ObjectType: &v1.ConfigMap{}, Handler: cache.ResourceEventHandlerFuncs{ @@ -155,11 +158,35 @@ func (s *Sync) Run(namespaces []string) (chan struct{}, error) { }, ResyncPeriod: 0, // Set to 0 as in the original code }) + func() { + s.mu.Lock() + defer s.mu.Unlock() + s.stores = append(s.stores, store) + }() go controller.Run(quit) } return quit, nil } +// Resync re-pushes all matching ConfigMaps from the local informer cache into +// OPA. Intended for use when OPA has restarted and lost its in-memory state. +func (s *Sync) Resync() { + s.mu.Lock() + stores := make([]cache.Store, len(s.stores)) + copy(stores, s.stores) + s.mu.Unlock() + + logrus.Infof("Resyncing all ConfigMaps to OPA") + for _, store := range stores { + for _, obj := range store.List() { + cm := obj.(*v1.ConfigMap) + if match, isPolicy := s.matcher(cm); match { + s.syncAdd(cm, isPolicy) + } + } + } +} + func (s *Sync) add(obj interface{}) { cm := obj.(*v1.ConfigMap) if match, isPolicy := s.matcher(cm); match { diff --git a/pkg/data/generic.go b/pkg/data/generic.go index 0c372277..9d098b7c 100644 --- a/pkg/data/generic.go +++ b/pkg/data/generic.go @@ -47,6 +47,7 @@ type GenericSync struct { ignoreNamespaces []string mu sync.Mutex ready bool + queue workqueue.TypedDelayingInterface[any] // protected by mu } // New returns a new GenericSync that can be started. @@ -142,6 +143,9 @@ func (s *GenericSync) setup(ctx context.Context) (cache.Store, workqueue.TypedDe resource := s.client.ResourceFor(s.ns, metav1.NamespaceAll) queue := workqueue.NewNamedDelayingQueue(s.ns.String()) + s.mu.Lock() + s.queue = queue + s.mu.Unlock() store, controller := cache.NewInformerWithOptions(cache.InformerOptions{ ListerWatcher: &cache.ListWatch{ ListFunc: func(options metav1.ListOptions) (runtime.Object, error) { @@ -233,7 +237,23 @@ func (q resourceEventQueue) OnDelete(obj interface{}) { q.Add(key) } -const initPath = "" +const ( + initPath = "" + resetPath = "\x00reset" // sentinel used by ForceResync to trigger a full reload +) + +// ForceResync triggers a full re-sync of all resources into OPA, as if the sync +// had just started. Safe to call concurrently. Intended for use when OPA has +// restarted and lost its in-memory state. +func (s *GenericSync) ForceResync() { + s.mu.Lock() + q := s.queue + s.mu.Unlock() + if q != nil && !q.ShuttingDown() { + q.Add(resetPath) + logrus.Infof("Queued force resync for %v", s.ns) + } +} // loop starts replicating Kubernetes resources into OPA. If an error occurs // during the replication process, this function will backoff and reload @@ -271,6 +291,13 @@ func (s *GenericSync) loop(store cache.Store, queue workqueue.TypedDelayingInter func (s *GenericSync) processNext(store cache.Store, path string, syncDone *bool) error { + // A force-resync was requested (e.g. OPA restarted). Reset state so the + // outer loop will perform a full syncAll on its next iteration. + if path == resetPath { + *syncDone = false + return fmt.Errorf("resync requested for %v", s.ns) + } + // On receiving the initPath, load a full dump of the data store if path == initPath { if *syncDone {