RONDB-933: RDRS1. Change feature store cache to asynchronously update… (#738)

* RONDB-933: RDRS1. Change feature store cache to asynchronously update the cached entries

* RONDB-933: Fixes for review

Author: smkniazi

storage/ndb/rest-server/rest-api-server/cmd/server/main.go

@@ -27,6 +27,7 @@ import (
 
 	"hopsworks.ai/rdrs/internal/config"
 	"hopsworks.ai/rdrs/internal/dal/heap"
+	"hopsworks.ai/rdrs/internal/feature_store"
 	"hopsworks.ai/rdrs/internal/log"
 	"hopsworks.ai/rdrs/internal/metrics"
 	"hopsworks.ai/rdrs/internal/security/apikey/hopsworkscache"
@@ -71,13 +72,18 @@ func main() {
 	apiKeyCache := hopsworkscache.New()
 	defer apiKeyCache.Cleanup()
 
+	//FS metadata cache
+	featureViewMetaDataCache := feature_store.NewFeatureViewMetaDataCache()
+	defer featureViewMetaDataCache.Cleanup()
+
 	// Prometheus metrics
 	rdrsMetrics, rdrsMetricsCleanup := metrics.NewRDRSMetrics()
 	defer rdrsMetricsCleanup()
 
 	cleanupServers, err := servers.CreateAndStartDefaultServers(
 		newHeap,
 		apiKeyCache,
+		featureViewMetaDataCache,
 		rdrsMetrics,
 		quit)
 	if err != nil {

storage/ndb/rest-server/rest-api-server/go.mod

@@ -71,7 +71,6 @@ require (
 	github.com/bytedance/sonic v1.12.5
 	github.com/goccy/go-json v0.10.2 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.9 // indirect
-	github.com/patrickmn/go-cache v2.1.0+incompatible
 	github.com/pelletier/go-toml/v2 v2.1.0 // indirect
 	github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
 	golang.org/x/arch v0.12.0 // indirect

storage/ndb/rest-server/rest-api-server/go.sum

@@ -71,8 +71,6 @@ github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w
 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
 github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M=
 github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
-github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc=
-github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
 github.com/pelletier/go-toml/v2 v2.1.0 h1:FnwAJ4oYMvbT/34k9zzHuZNrhlz48GB3/s6at6/MHO4=
 github.com/pelletier/go-toml/v2 v2.1.0/go.mod h1:tJU2Z3ZkXwnxa4DPO899bsyIoywizdUvyaeZurnPPDc=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=

storage/ndb/rest-server/rest-api-server/internal/config/init_defaults.go

@@ -106,6 +106,13 @@ func newWithDefaults() AllConfigs {
 				CacheRefreshIntervalJitterMS: 1000,
 			},
 		},
+		FeatureStore: FeatureStore{
+			FeatureStoreMetadataCache: FeatureStoreMetadataCache{
+				CacheRefreshIntervalMS:       600000,  // 10 min
+				CacheUnusedEntriesEvictionMS: 1800000, // 30 min
+				CacheRefreshIntervalJitterMS: 1000,
+			},
+		},
 		Log: log.LogConfig{
 			Level:      "warn",
 			FilePath:   "",

storage/ndb/rest-server/rest-api-server/internal/config/structs.go

@@ -247,13 +247,39 @@ func (t *TestParameters) Validate() error {
 	return nil
 }
 
+type FeatureStoreMetadataCache struct {
+	CacheRefreshIntervalMS       uint32
+	CacheUnusedEntriesEvictionMS uint32
+	CacheRefreshIntervalJitterMS uint32
+}
+
 type APIKey struct {
 	UseHopsworksAPIKeys          bool
 	CacheRefreshIntervalMS       uint32
 	CacheUnusedEntriesEvictionMS uint32
 	CacheRefreshIntervalJitterMS uint32
 }
 
+func (fsmdc *FeatureStoreMetadataCache) Validate() error {
+	if fsmdc.CacheRefreshIntervalMS == 0 {
+		return errors.New("CacheRefreshIntervalMS cannot be 0")
+	}
+
+	if fsmdc.CacheUnusedEntriesEvictionMS == 0 {
+		return errors.New("CacheUnusedEntriesEvictionMS cannot be 0")
+	}
+
+	if fsmdc.CacheRefreshIntervalMS > fsmdc.CacheUnusedEntriesEvictionMS {
+		return errors.New("CacheRefreshIntervalMS can not be more that CacheUnusedEntriesEvictionMS")
+	}
+
+	if fsmdc.CacheRefreshIntervalJitterMS >= fsmdc.CacheRefreshIntervalMS {
+		return errors.New("CacheRefreshIntervalJitterMS must be smaller than CacheRefreshIntervalMS")
+	}
+
+	return nil
+}
+
 func (a *APIKey) Validate() error {
 	if a.CacheRefreshIntervalMS == 0 {
 		return errors.New("CacheRefreshIntervalMS cannot be 0")
@@ -302,6 +328,10 @@ func (t *TLS) Validate() error {
 	return t.TestParameters.Validate()
 }
 
+type FeatureStore struct {
+	FeatureStoreMetadataCache FeatureStoreMetadataCache
+}
+
 type Security struct {
 	TLS    TLS
 	APIKey APIKey
@@ -321,6 +351,15 @@ func (c *Security) Validate() error {
 	return nil
 }
 
+func (fs *FeatureStore) Validate() error {
+	err := fs.FeatureStoreMetadataCache.Validate()
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
 /*
 The RDRS is tested on a regular basis by RonDB's MTR tests. These MTR tests
 have a config file defined for the RDRS in `mysql-test/suite/rdrs/include/have_rdrs.inc`.
@@ -335,6 +374,7 @@ type AllConfigs struct {
 	RonDB                RonDB
 	RonDBMetadataCluster RonDB
 	Security             Security
+	FeatureStore         FeatureStore
 	Log                  log.LogConfig
 	Testing              Testing
 }
@@ -351,6 +391,8 @@ func (c *AllConfigs) Validate() error {
 		return err
 	} else if err = c.Security.Validate(); err != nil {
 		return err
+	} else if err = c.FeatureStore.Validate(); err != nil {
+		return err
 	}
 
 	// c.RonDBMetaCluster is optional. Copy the cluster

storage/ndb/rest-server/rest-api-server/internal/feature_store/metadata.go

@@ -18,24 +18,18 @@
 package feature_store
 
 import (
-	"encoding/json"
 	"errors"
 	"fmt"
 	"reflect"
 	"sort"
 	"strings"
-	"time"
 
 	"github.com/hamba/avro/v2"
-	"github.com/patrickmn/go-cache"
 
 	"hopsworks.ai/rdrs/internal/dal"
 	"hopsworks.ai/rdrs/internal/log"
 )
 
-var DefaultExpiration time.Duration = 15 * time.Minute
-var CleanupInterval time.Duration = 15 * time.Minute
-
 const ERROR_NOT_FOUND = "Not Found"
 
 type ComplexFeature struct {
@@ -298,39 +292,6 @@ func getFeatureIndexKey(joinIndex int, fgId int, f string) *string {
 	return &featureIndexKey
 }
 
-type FeatureViewMetaDataCache struct {
-	metadataCache cache.Cache
-}
-
-func NewFeatureViewMetaDataCache() *FeatureViewMetaDataCache {
-	var c = cache.New(DefaultExpiration, CleanupInterval)
-	return &FeatureViewMetaDataCache{*c}
-}
-
-func (fvmeta *FeatureViewMetaDataCache) Get(featureStoreName, featureViewName string, featureViewVersion int) (*FeatureViewMetadata, *RestErrorCode) {
-	var fvCacheKey = getFeatureViewCacheKey(featureStoreName, featureViewName, featureViewVersion)
-	var metadataInf, exist = fvmeta.metadataCache.Get(fvCacheKey)
-	if !exist {
-		var metadata, err = GetFeatureViewMetadata(featureStoreName, featureViewName, featureViewVersion)
-		if err != nil {
-			return nil, err
-		} else {
-			fvmeta.metadataCache.SetDefault(fvCacheKey, metadata)
-			if log.IsDebug() {
-				metadataJson, _ := json.MarshalIndent(metadata, "", "  ")
-				log.Debugf("Feature store metadata is %s", metadataJson)
-			}
-			return metadata, nil
-		}
-	} else {
-		var metadata, ok = metadataInf.(*FeatureViewMetadata)
-		if !ok {
-			return nil, FETCH_METADATA_FROM_CACHE_FAIL
-		}
-		return metadata, nil
-	}
-}
-
 func getFeatureViewCacheKey(featureStoreName, featureViewName string, featureViewVersion int) string {
 	return fmt.Sprintf("%s|%s|%d", featureStoreName, featureViewName, featureViewVersion)
 }