feat: resync OPA state after container restart. Fixes #296

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(&params.opaConfigFile, "opa-config", "", "", "set file containing OPA configuration to enable data replication based on configured bundles")
 	rootCmd.Flags().StringVarP(&params.analysisEntrypoint, "analysis-entrypoint", "", "main/main", "set decision to analyze for dynamic data replication configuration (requires --opa-config)")
 	rootCmd.Flags().StringVarP(&params.healthEndpoint, "health-endpoint", "", "", "set health check listening endpoint (e.g., localhost:8000)")
+	rootCmd.Flags().DurationVarP(&params.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)

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 {

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 {