feat: durable deployment-operator caches (#4021)

Co-authored-by: Sebastian Florek <s.florek91@gmail.com>

Author: zreigz

.vscode/settings.json

@@ -14,5 +14,6 @@
     "editor.formatOnSave": false,
     "editor.formatOnPaste": false
   },
-  "editor.formatOnSave": true
+  "editor.formatOnSave": true,
+  "git.ignoreLimitWarning": true
 }

go/deployment-operator/.gitignore

@@ -9,6 +9,7 @@ go.work.sum
 
 # Binaries for programs and plugins
 bin/
+/agent
 *.exe
 *.exe~
 *.dll

go/deployment-operator/cmd/agent/args/args.go

@@ -8,24 +8,27 @@ import (
 	"strings"
 	"time"
 
-	"github.com/pluralsh/console/go/deployment-operator/api/v1alpha1"
-	"github.com/pluralsh/console/go/polly/containers"
 	"github.com/spf13/pflag"
 	"k8s.io/klog/v2"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 
+	"github.com/pluralsh/console/go/deployment-operator/api/v1alpha1"
+	"github.com/pluralsh/console/go/polly/containers"
+
 	"github.com/pluralsh/console/go/deployment-operator/internal/helpers"
 	"github.com/pluralsh/console/go/deployment-operator/pkg/log"
 	"github.com/pluralsh/console/go/deployment-operator/pkg/streamline/api"
 )
 
 const (
-	EnvDeployToken      = "DEPLOY_TOKEN"
-	EnvDatadogEnabled   = "DATADOG_ENABLED"
-	EnvPyroscopeEnabled = "PYROSCOPE_ENABLED"
-	EnvProfilerEnabled  = "PROFILER_ENABLED"
-	EnvLocal            = "LOCAL"
+	EnvDeployToken        = "DEPLOY_TOKEN"
+	EnvDatadogEnabled     = "DATADOG_ENABLED"
+	EnvDatadogHost        = "DATADOG_HOST"
+	EnvDatadogEnvironment = "DATADOG_ENV"
+	EnvPyroscopeEnabled   = "PYROSCOPE_ENABLED"
+	EnvProfilerEnabled    = "PROFILER_ENABLED"
+	EnvLocal              = "LOCAL"
 
 	defaultProbeAddress   = ":9001"
 	defaultMetricsAddress = ":8000"
@@ -64,6 +67,9 @@ const (
 	defaultManifestCacheTTL         = "3h"
 	defaultManifestCacheTTLDuration = 3 * time.Hour
 
+	defaultCachePersistInterval         = "10s"
+	defaultCachePersistIntervalDuration = 10 * time.Second
+
 	defaultComponentShaCacheTTL         = "6h"
 	defaultComponentShaCacheTTLDuration = 6 * time.Hour
 
@@ -153,14 +159,16 @@ var (
 	argResourceCacheTTL                     = flag.String("resource-cache-ttl", defaultResourceCacheTTL, "The time to live of each resource cache entry.")
 	argManifestCacheTTL                     = flag.String("manifest-cache-ttl", defaultManifestCacheTTL, "The time to live of service manifests in cache entry.")
 	argManifestCacheJitter                  = flag.String("manifest-cache-jitter", defaultManifestCacheJitter, "Deprecated: ignored; manifest cache jitter is fixed at 50% of its TTL.")
+	argCacheDir                             = flag.String("cache-dir", "", "Directory used to persist operator caches across restarts. Empty disables persistence.")
+	argCachePersistInterval                 = flag.String("cache-persist-interval", defaultCachePersistInterval, "Interval to flush in-memory caches to cache-dir.")
 	argComponentShaCacheTTL                 = flag.String("component-sha-cache-ttl", defaultComponentShaCacheTTL, "The time to live of the component sha cache entries.")
 	argComponentShaCacheJitter              = flag.String("component-sha-cache-jitter", defaultComponentShaCacheJitter, "Deprecated: ignored; component SHA cache jitter is fixed at 50% of its TTL.")
 	argControllerCacheTTL                   = flag.String("controller-cache-ttl", defaultControllerCacheTTL, "The time to live of console controller cache entries.")
 	argRestoreNamespace                     = flag.String("restore-namespace", defaultRestoreNamespace, "The namespace where Velero restores are located.")
 	argServices                             = flag.String("services", "", "A comma separated list of service ids to reconcile. Leave empty to reconcile all.")
 	argPyroscopeAddress                     = flag.String("pyroscope-address", defaultPyroscopeAddress, "The address of the Pyroscope server.")
-	argDatadogHost                          = flag.String("datadog-host", defaultDatadogHost, "The address of the Datadog server.")
-	argDatadogEnv                           = flag.String("datadog-env", defaultDatadogEnv, "The environment of the Datadog server.")
+	argDatadogHost                          = flag.String("datadog-host", helpers.GetEnv(EnvDatadogHost, defaultDatadogHost), "The address of the Datadog server.")
+	argDatadogEnv                           = flag.String("datadog-env", helpers.GetEnv(EnvDatadogEnvironment, defaultDatadogEnv), "The environment of the Datadog server.")
 	argWorkqueueBaseDelay                   = flag.String("workqueue-base-delay", defaultWorkqueueBaseDelay, "The base delay for the workqueue.")
 	argWorkqueueMaxDelay                    = flag.String("workqueue-max-delay", defaultWorkqueueMaxDelay, "The maximum delay for the workqueue.")
 	argWorkqueueQPS                         = flag.Int("workqueue-qps", 10, "The maximum number of items to process per second.")
@@ -377,6 +385,20 @@ func ManifestCacheTTL() time.Duration {
 	return duration
 }
 
+func CacheDir() string {
+	return *argCacheDir
+}
+
+func CachePersistInterval() time.Duration {
+	duration, err := time.ParseDuration(*argCachePersistInterval)
+	if err != nil {
+		klog.ErrorS(err, "Could not parse cache-persist-interval", "value", *argCachePersistInterval, "default", defaultCachePersistIntervalDuration)
+		return defaultCachePersistIntervalDuration
+	}
+
+	return duration
+}
+
 func ComponentShaCacheTTL() time.Duration {
 	duration, err := time.ParseDuration(*argComponentShaCacheTTL)
 	if err != nil {

go/deployment-operator/cmd/agent/console.go

@@ -60,6 +60,7 @@ func registerConsoleReconcilersOrDie(
 	discoveryCache discoverycache.Cache,
 	namespaceCache streamline.NamespaceCache,
 	svcCache cache.Store[console.ServiceDeploymentForAgent],
+	cacheDir string,
 ) {
 	mgr.AddReconcilerOrDie(service.Identifier, func() (v1.Reconciler, error) {
 		r, err := service.NewServiceReconciler(consoleClient,
@@ -72,6 +73,7 @@ func registerConsoleReconcilersOrDie(
 			svcCache,
 			store,
 			service.WithManifestTTL(args.ManifestCacheTTL()),
+			service.WithCacheDir(cacheDir),
 			service.WithWorkqueueBaseDelay(args.WorkqueueBaseDelay()),
 			service.WithWorkqueueMaxDelay(args.WorkqueueMaxDelay()),
 			service.WithWorkqueueQPS(args.WorkqueueQPS()),

go/deployment-operator/cmd/agent/kubernetes.go

@@ -131,6 +131,7 @@ func registerKubeReconcilersOrDie(
 	discoveryCache discoverycache.Cache,
 	enableKubecostProxy bool,
 	consoleURL, deployToken string,
+	userGroupCache cache.UserGroupCache,
 ) {
 	rolloutsClient, dynamicClient, kubeClient := initKubeClientsOrDie(config)
 
@@ -251,6 +252,7 @@ func registerKubeReconcilersOrDie(
 		Scheme:           manager.GetScheme(),
 		ExtConsoleClient: extConsoleClient,
 		ConsoleUrl:       rawConsoleUrl,
+		UserGroupCache:   userGroupCache,
 	}).SetupWithManager(manager); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "VirtualCluster")
 	}
@@ -330,9 +332,10 @@ func registerKubeReconcilersOrDie(
 		setupLog.Error(err, "unable to create controller", "controller", "AgentRun")
 	}
 	if err := (&controller.PluralCAPIClusterController{
-		Client:     manager.GetClient(),
-		Scheme:     manager.GetScheme(),
-		ConsoleUrl: consoleURL,
+		Client:         manager.GetClient(),
+		Scheme:         manager.GetScheme(),
+		ConsoleUrl:     consoleURL,
+		UserGroupCache: userGroupCache,
 	}).SetupWithManager(manager); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "PluralCAPIClusterController")
 	}

go/deployment-operator/cmd/agent/main.go

@@ -40,6 +40,7 @@ import (
 	"github.com/pluralsh/console/go/deployment-operator/internal/utils"
 	"github.com/pluralsh/console/go/deployment-operator/pkg/cache"
 	discoverycache "github.com/pluralsh/console/go/deployment-operator/pkg/cache/discovery"
+	"github.com/pluralsh/console/go/deployment-operator/pkg/cache/persist"
 	"github.com/pluralsh/console/go/deployment-operator/pkg/client"
 	"github.com/pluralsh/console/go/deployment-operator/pkg/common"
 	"github.com/pluralsh/console/go/deployment-operator/pkg/ping"
@@ -51,6 +52,7 @@ import (
 	deploymentsv1alpha1 "github.com/pluralsh/console/go/deployment-operator/api/v1alpha1"
 	"github.com/pluralsh/console/go/deployment-operator/cmd/agent/args"
 	consolectrl "github.com/pluralsh/console/go/deployment-operator/pkg/controller"
+	"github.com/pluralsh/console/go/deployment-operator/pkg/controller/namespaces"
 	"github.com/pluralsh/console/go/deployment-operator/pkg/controller/service"
 )
 
@@ -161,12 +163,47 @@ func main() {
 			return extConsoleClient.GetService(id)
 		})
 
+	cacheStore := openCacheStoreOrDie()
+	defer func() {
+		if err := cacheStore.Close(); err != nil {
+			setupLog.Error(err, "unable to release cache dir lock")
+		}
+	}()
+
 	// Start synchronizer supervisor
 	supervisor := runSynchronizerSupervisorOrDie(ctx, dynamicClient, dbStore, statusSynchronizer, discoveryCache, namespaceCache, svcCache)
 	defer supervisor.Stop()
 
-	registerConsoleReconcilersOrDie(consoleManager, mapper, clientSet, kubeManager.GetClient(), dynamicClient, dbStore, kubeManager.GetScheme(), extConsoleClient, supervisor, discoveryCache, namespaceCache, svcCache)
-	registerKubeReconcilersOrDie(ctx, clientSet, kubeManager, consoleManager, config, extConsoleClient, discoveryCache, args.EnableKubecostProxy(), args.ConsoleUrl(), args.DeployToken())
+	userGroupCache := cache.NewUserGroupCache(extConsoleClient)
+
+	registerConsoleReconcilersOrDie(consoleManager, mapper, clientSet, kubeManager.GetClient(), dynamicClient, dbStore, kubeManager.GetScheme(), extConsoleClient, supervisor, discoveryCache, namespaceCache, svcCache, cacheStore.Dir())
+	registerKubeReconcilersOrDie(ctx, clientSet, kubeManager, consoleManager, config, extConsoleClient, discoveryCache, args.EnableKubecostProxy(), args.ConsoleUrl(), args.DeployToken(), userGroupCache)
+
+	svcReconciler := consoleManager.GetReconcilerOrDie(service.Identifier).(*service.ServiceReconciler)
+	nsReconciler := consoleManager.GetReconcilerOrDie(namespaces.Identifier).(*namespaces.NamespaceReconciler)
+	saveCaches := func() error {
+		userIDs, groupIDs := persist.IdentityRecordsFrom(userGroupCache)
+		return cacheStore.Save(persist.Snapshot{
+			Manifests:         svcReconciler.ManifestCache().Export(),
+			ComponentSHAs:     persist.SHARecordsFrom(cache.ComponentShaCache()),
+			StatusSHAs:        persist.SHARecordsFrom(statusSynchronizer.SHACache()),
+			UserIDs:           userIDs,
+			GroupIDs:          groupIDs,
+			ManagedNamespaces: persist.PollyRecordsFrom(nsReconciler.NamespaceCache()),
+		})
+	}
+	if snap, err := cacheStore.Load(); err != nil {
+		setupLog.Error(err, "unable to load durable cache, starting cold")
+	} else {
+		setupLog.Info("importing durable cache snapshot")
+		svcReconciler.ManifestCache().Import(snap.Manifests)
+		persist.ApplySHARecords(cache.ComponentShaCache(), snap.ComponentSHAs)
+		persist.ApplySHARecords(statusSynchronizer.SHACache(), snap.StatusSHAs)
+		persist.ApplyIdentityRecords(userGroupCache, snap.UserIDs, snap.GroupIDs)
+		persist.ApplyPollyRecords(nsReconciler.NamespaceCache(), snap.ManagedNamespaces)
+		setupLog.Info("durable cache import finished")
+	}
+	cacheStore.StartPeriodic(ctx, args.CachePersistInterval(), saveCaches)
 
 	//+kubebuilder:scaffold:builder
 
@@ -189,6 +226,11 @@ func main() {
 	// Block the main thread until context cancel.
 	<-ctx.Done()
 	setupLog.Info("shutting down")
+	cacheStore.WaitPeriodic()
+	setupLog.Info("exporting durable cache snapshot")
+	if err := saveCaches(); err != nil {
+		setupLog.Error(err, "unable to persist cache snapshot")
+	}
 }
 
 func loadAgentConfigurationOrDie(ctx context.Context, reader ctrlclient.Reader) {
@@ -338,6 +380,15 @@ func runSynchronizerSupervisorOrDie(ctx context.Context, dynamicClient dynamic.I
 	return supervisor
 }
 
+func openCacheStoreOrDie() *persist.Store {
+	cacheStore, err := persist.Open(args.CacheDir())
+	if err != nil {
+		setupLog.Error(err, "unable to open cache dir")
+		os.Exit(1)
+	}
+	return cacheStore
+}
+
 func initDatabaseStoreOrDie(ctx context.Context) store.Store {
 	dbStore, err := store.NewDatabaseStore(ctx, store.WithStorage(args.StoreStorage()), store.WithFilePath(args.StoreFilePath()))
 	if err != nil {

go/deployment-operator/go.mod

@@ -69,6 +69,7 @@ require (
 	github.com/yuin/gopher-lua v1.1.2
 	gitlab.com/gitlab-org/api/client-go v1.46.0
 	golang.org/x/oauth2 v0.36.0
+	golang.org/x/sys v0.46.0
 	golang.org/x/time v0.15.0
 	gopkg.in/yaml.v3 v3.0.1
 	gotest.tools/gotestsum v1.13.0
@@ -342,7 +343,6 @@ require (
 	golang.org/x/mod v0.37.0 // indirect
 	golang.org/x/net v0.56.0 // indirect
 	golang.org/x/sync v0.22.0 // indirect
-	golang.org/x/sys v0.46.0 // indirect
 	golang.org/x/term v0.44.0 // indirect
 	golang.org/x/text v0.39.0 // indirect
 	golang.org/x/tools v0.47.0 // indirect

go/deployment-operator/internal/controller/pluralcapicluster_controller.go

@@ -32,11 +32,11 @@ const PluralCAPIClusterFinalizer = "deployments.plural.sh/plural-capi-cluster-pr
 
 type PluralCAPIClusterController struct {
 	k8sClient.Client
-	Scheme     *runtime.Scheme
-	ConsoleUrl string
+	Scheme         *runtime.Scheme
+	ConsoleUrl     string
+	UserGroupCache cache.UserGroupCache
 
-	userGroupCache cache.UserGroupCache
-	consoleClient  client.Client
+	consoleClient client.Client
 }
 
 func (in *PluralCAPIClusterController) Reconcile(ctx context.Context, req ctrl.Request) (_ reconcile.Result, reterr error) {
@@ -191,7 +191,9 @@ func (in *PluralCAPIClusterController) initConsoleClient(consoleToken string) er
 			return err
 		}
 		in.consoleClient = client.New(fmt.Sprintf("%s/gql", url), consoleToken)
-		in.userGroupCache = cache.NewUserGroupCache(in.consoleClient)
+		if in.UserGroupCache == nil {
+			in.UserGroupCache = cache.NewUserGroupCache(in.consoleClient)
+		}
 	}
 	return nil
 }
@@ -238,14 +240,14 @@ func (in *PluralCAPIClusterController) ensureCluster(cluster *v1alpha1.PluralCAP
 		return nil
 	}
 
-	bindings, req, err := ensureBindings(cluster.Spec.Cluster.Bindings.Read, in.userGroupCache)
+	bindings, req, err := ensureBindings(cluster.Spec.Cluster.Bindings.Read, in.UserGroupCache)
 	if err != nil {
 		return err
 	}
 
 	cluster.Spec.Cluster.Bindings.Read = bindings
 
-	bindings, req2, err := ensureBindings(cluster.Spec.Cluster.Bindings.Write, in.userGroupCache)
+	bindings, req2, err := ensureBindings(cluster.Spec.Cluster.Bindings.Write, in.UserGroupCache)
 	if err != nil {
 		return err
 	}

go/deployment-operator/internal/controller/virtualcluster_controller.go

@@ -35,10 +35,10 @@ type VirtualClusterController struct {
 	Scheme           *runtime.Scheme
 	ExtConsoleClient client.Client
 	ConsoleUrl       string
+	UserGroupCache   cache.UserGroupCache
 
-	userGroupCache cache.UserGroupCache
-	consoleClient  client.Client
-	myCluster      *console.MyCluster_MyCluster_
+	consoleClient client.Client
+	myCluster     *console.MyCluster_MyCluster_
 }
 
 func (in *VirtualClusterController) Reconcile(ctx context.Context, req reconcile.Request) (_ reconcile.Result, reterr error) {
@@ -245,14 +245,14 @@ func (in *VirtualClusterController) ensureCluster(cluster *v1alpha1.VirtualClust
 		return nil
 	}
 
-	bindings, req, err := ensureBindings(cluster.Spec.Cluster.Bindings.Read, in.userGroupCache)
+	bindings, req, err := ensureBindings(cluster.Spec.Cluster.Bindings.Read, in.UserGroupCache)
 	if err != nil {
 		return err
 	}
 
 	cluster.Spec.Cluster.Bindings.Read = bindings
 
-	bindings, req2, err := ensureBindings(cluster.Spec.Cluster.Bindings.Write, in.userGroupCache)
+	bindings, req2, err := ensureBindings(cluster.Spec.Cluster.Bindings.Write, in.UserGroupCache)
 	if err != nil {
 		return err
 	}
@@ -277,7 +277,9 @@ func (in *VirtualClusterController) initConsoleClient(ctx context.Context, vClus
 	}
 
 	in.consoleClient = client.New(fmt.Sprintf("%s/gql", in.ConsoleUrl), token)
-	in.userGroupCache = cache.NewUserGroupCache(in.consoleClient)
+	if in.UserGroupCache == nil {
+		in.UserGroupCache = cache.NewUserGroupCache(in.consoleClient)
+	}
 
 	return nil
 }