chore: Update cache library

Makefile

@@ -1,4 +1,3 @@
-include .env.sample
 export
 
 BIN_NAME=merlin
@@ -129,7 +128,7 @@ build-inference-logger:
 .PHONY: run
 run:
  @echo "> Running application ..."
- @./bin/${BIN_NAME}
+ @./bin/${BIN_NAME} -config ./env-config.yaml
 
 .PHONY: run-ui
 run-ui:

api/README.md

@@ -27,7 +27,7 @@ make setup
 
 To start Merlin locally you have to:
 
-- Replace `VAULT_TOKEN` value in .env.sample
+- Generate the local / test cluster credentials and add to `./environment.yaml` and `./env-config.yaml`
 
 Then execute:
 

api/cmd/transformer/Makefile

@@ -1,4 +1,3 @@
-include .env.sample
 export
 
 BIN_NAME=feast-transformer

api/go.mod

@@ -17,7 +17,6 @@ require (
  github.com/cespare/xxhash v1.1.0
  github.com/confluentinc/confluent-kafka-go v1.9.2
  github.com/confluentinc/confluent-kafka-go/v2 v2.0.2
- github.com/coocood/freecache v1.1.1
  github.com/fatih/color v1.13.0
  github.com/feast-dev/feast/sdk/go v0.9.4
  github.com/fraugster/parquet-go v0.10.0
@@ -52,6 +51,7 @@ require (
  github.com/opentracing-contrib/go-stdlib v1.0.0
  github.com/opentracing/opentracing-go v1.2.0
  github.com/ory/viper v1.7.5
+ github.com/patrickmn/go-cache v2.1.0+incompatible
  github.com/pilagod/gorm-cursor-paginator v1.3.0
  github.com/pkg/errors v0.9.1
  github.com/prometheus/client_golang v1.12.1
@@ -184,7 +184,6 @@ require (
  github.com/opencontainers/go-digest v1.0.0 // indirect
  github.com/opencontainers/image-spec v1.0.3-0.20211202183452-c5a74bcca799 // indirect
  github.com/ory/keto-client-go v0.4.4-alpha.1 // indirect
- github.com/patrickmn/go-cache v2.1.0+incompatible // indirect
  github.com/pelletier/go-toml v1.9.5 // indirect
  github.com/pmezard/go-difflib v1.0.0 // indirect
  github.com/princjef/gomarkdoc v0.4.1 // indirect

api/go.sum

@@ -405,8 +405,6 @@ github.com/containernetworking/plugins v0.9.1/go.mod h1:xP/idU2ldlzN6m4p5LmGiwRD
 github.com/containers/ocicrypt v1.0.1/go.mod h1:MeJDzk1RJHv89LjsH0Sp5KTY3ZYkjXO/C+bKAeWFIrc=
 github.com/containers/ocicrypt v1.1.0/go.mod h1:b8AOe0YR67uU8OqfVNcznfFpAzu3rdgUV4GP9qXPfu4=
 github.com/containers/ocicrypt v1.1.1/go.mod h1:Dm55fwWm1YZAjYRaJ94z2mfZikIyIN4B0oB3dj3jFxY=
-github.com/coocood/freecache v1.1.1 h1:uukNF7QKCZEdZ9gAV7WQzvh0SbjwdMF6m3x3rxEkaPc=
-github.com/coocood/freecache v1.1.1/go.mod h1:OKrEjkGVoxZhyWAJoeFi5BMLUJm2Tit0kpGkIr7NGYY=
 github.com/coredns/corefile-migration v1.0.2/go.mod h1:OFwBp/Wc9dJt5cAZzHWMNhK1r5L0p0jDwIBc6j8NC8E=
 github.com/coreos/bbolt v1.3.2/go.mod h1:iRUV2dpdMOn7Bo10OQBFzIJO9kkE559Wcmn+qkEiiKk=
 github.com/coreos/etcd v3.3.10+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=

api/pkg/transformer/cache/cache.go

@@ -3,31 +3,31 @@ package cache
 import (
  "time"
 
- "github.com/coocood/freecache"
+ cache "github.com/patrickmn/go-cache"
 )
 
 type Cache interface {
- Insert(key []byte, value []byte, ttl time.Duration) error
- Fetch(key []byte) ([]byte, error)
+ Insert(key string, value interface{}, ttl time.Duration)
+ Fetch(key string) (interface{}, bool)
 }
 
 type inMemoryCache struct {
- cache *freecache.Cache
+ cache *cache.Cache
 }
 
 const (
- MB = 1024 * 1024
+ cacheCleanUpSeconds = 300
 )
 
-func NewInMemoryCache(sizeInMB int) *inMemoryCache {
- executor := freecache.NewCache(sizeInMB * MB)
+func NewInMemoryCache() *inMemoryCache {
+ executor := cache.New(0, cacheCleanUpSeconds*time.Second)
  return &inMemoryCache{cache: executor}
 }
 
-func (c *inMemoryCache) Insert(key []byte, value []byte, ttl time.Duration) error {
- return c.cache.Set(key, value, int(ttl/time.Second))
+func (c *inMemoryCache) Insert(key string, value interface{}, ttl time.Duration) {
+ c.cache.Set(key, value, ttl)
 }
 
-func (c *inMemoryCache) Fetch(key []byte) ([]byte, error) {
+func (c *inMemoryCache) Fetch(key string) (interface{}, bool) {
  return c.cache.Get(key)
 }

api/pkg/transformer/cache/cache_test.go

@@ -1,12 +1,9 @@
 package cache
 
 import (
- "encoding/json"
- "reflect"
  "testing"
  "time"
 
- "github.com/coocood/freecache"
  "github.com/stretchr/testify/assert"
  "github.com/stretchr/testify/require"
 )
@@ -15,40 +12,33 @@ func TestCache(t *testing.T) {
  testCases := []struct {
  desc string
  data interface{}
- key []byte
+ key string
  }{
  {
  desc: "Success - 1",
  data: map[string]string{"key": "value"},
- key: []byte("key1"),
+ key: "key1",
  },
  {
  desc: "Success - 2",
  data: map[string]string{"key": "value"},
- key: []byte("key2"),
+ key: "key2",
  },
  {
  desc: "Success - 3",
  data: []int{1, 2, 3},
- key: []byte("key3"),
+ key: "key3",
  },
  }
  for _, tC := range testCases {
  t.Run(tC.desc, func(t *testing.T) {
- dataByte, err := json.Marshal(tC.data)
- require.NoError(t, err)
- cache := NewInMemoryCache(1)
- err = cache.Insert(tC.key, dataByte, 1)
- assert.NoError(t, err)
+ cache := NewInMemoryCache()
+ cache.Insert(tC.key, tC.data, 2*time.Second)
 
- cachedValue, err := cache.Fetch(tC.key)
- require.NoError(t, err)
+ cachedValue, ok := cache.Fetch(tC.key)
+ require.True(t, ok)
 
- var val interface{}
- err = json.Unmarshal(cachedValue, &val)
- require.NoError(t, err)
-
- reflect.DeepEqual(tC.data, val)
+ assert.Equal(t, tC.data, cachedValue)
  })
  }
 }
@@ -57,45 +47,38 @@ func TestCache_Expiry(t *testing.T) {
  testCases := []struct {
  desc string
  data interface{}
- key []byte
+ key string
  delayOfFetching int
- err  error
+ foundInCache bool
  }{
  {
  desc: "Success - Not expired",
  data: map[string]string{"key": "value"},
- key: []byte("key1"),
+ key: "key1",
  delayOfFetching: 0,
+ foundInCache: true,
  },
  {
  desc: "Success - Expired",
  data: map[string]string{"key": "value"},
- key: []byte("key1"),
- delayOfFetching: 2,
- err: freecache.ErrNotFound,
+ key: "key1",
+ delayOfFetching: 3,
  },
  }
  for _, tC := range testCases {
  t.Run(tC.desc, func(t *testing.T) {
- dataByte, err := json.Marshal(tC.data)
- require.NoError(t, err)
- cache := NewInMemoryCache(1)
- err = cache.Insert(tC.key, dataByte, 1*time.Second)
- assert.NoError(t, err)
+ cache := NewInMemoryCache()
+ cache.Insert(tC.key, tC.data, 2*time.Second)
 
  if tC.delayOfFetching > 0 {
  time.Sleep(time.Duration(tC.delayOfFetching) * time.Second)
  }
 
- cachedValue, err := cache.Fetch(tC.key)
- assert.Equal(t, tC.err, err)
-
- if err == nil {
- var val interface{}
- err = json.Unmarshal(cachedValue, &val)
- require.NoError(t, err)
+ cachedValue, ok := cache.Fetch(tC.key)
+ require.Equal(t, tC.foundInCache, ok)
 
- reflect.DeepEqual(tC.data, val)
+ if ok {
+ assert.Equal(t, tC.data, cachedValue)
  }
 
  })

api/pkg/transformer/executor/transformer_test.go

@@ -445,11 +445,10 @@ func TestStandardTransformer_Execute(t *testing.T) {
  require.NoError(t, err)
 
  compiler := pipeline.NewCompiler(symbol.NewRegistry(), feastClients, &feast.Options{
- CacheEnabled: true,
- CacheSizeInMB: 100,
- CacheTTL: 60 * time.Second,
- BatchSize: 100,
- FeastTimeout: 1 * time.Second,
+ CacheEnabled: true,
+ CacheTTL: 60 * time.Second,
+ BatchSize: 100,
+ FeastTimeout: 1 * time.Second,
 
  DefaultFeastSource: spec.ServingSource_BIGTABLE,
  StorageConfigs: feast.FeastStorageConfig{

api/pkg/transformer/feast/feature_cache.go

@@ -23,9 +23,9 @@ type featureCache struct {
  ttl time.Duration
 }
 
-func newFeatureCache(ttl time.Duration, sizeInMB int) *featureCache {
+func newFeatureCache(ttl time.Duration) *featureCache {
  return &featureCache{
- cache: cache.NewInMemoryCache(sizeInMB),
+ cache: cache.NewInMemoryCache(),
  ttl: ttl,
  }
 }
@@ -74,15 +74,15 @@ func (fc *featureCache) fetchFeatureTable(entities []feast.Row, columnNames []st
  }
 
  feastCacheRetrievalCount.Inc()
- val, err := fc.cache.Fetch(keyByte)
- if err != nil {
+ val, ok := fc.cache.Fetch(string(keyByte))
+ if !ok {
  entityNotInCache = append(entityNotInCache, entity)
  continue
  }
 
  feastCacheHitCount.Inc()
- var cacheValue CacheValue
- if err := json.Unmarshal(val, &cacheValue); err != nil {
+ cacheValue, ok := val.(CacheValue)
+ if !ok {
  entityNotInCache = append(entityNotInCache, entity)
  continue
  }
@@ -136,11 +136,8 @@ func (fc *featureCache) insertFeaturesOfEntity(entity feast.Row, columnNames []s
  ValueRow: value,
  ValueTypes: valueTypes,
  }
- dataByte, err := json.Marshal(cacheValue)
- if err != nil {
- return err
- }
- return fc.cache.Insert(keyByte, dataByte, fc.ttl)
+ fc.cache.Insert(string(keyByte), cacheValue, fc.ttl)
+ return nil
 }
 
 func castValueRow(row types.ValueRow, columnTypes []feastTypes.ValueType_Enum) (types.ValueRow, error) {