diff --git a/serving/pom.xml b/serving/pom.xml index 34cfcd777ed..c6db97ec91b 100644 --- a/serving/pom.xml +++ b/serving/pom.xml @@ -232,6 +232,13 @@ 2.23.0 test + + + com.google.guava + guava-testlib + 26.0-jre + test + diff --git a/serving/src/main/java/feast/serving/config/ServingApiConfiguration.java b/serving/src/main/java/feast/serving/config/ServingApiConfiguration.java index c8e5225a3c5..dcee2f73c65 100644 --- a/serving/src/main/java/feast/serving/config/ServingApiConfiguration.java +++ b/serving/src/main/java/feast/serving/config/ServingApiConfiguration.java @@ -30,6 +30,8 @@ import java.util.Map; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; @@ -39,15 +41,12 @@ import org.springframework.http.converter.protobuf.ProtobufJsonFormatHttpMessageConverter; import org.springframework.web.servlet.config.annotation.WebMvcConfigurer; -/** - * Global bean configuration. - */ +/** Global bean configuration. */ @Slf4j @Configuration public class ServingApiConfiguration implements WebMvcConfigurer { - @Autowired - private ProtobufJsonFormatHttpMessageConverter protobufConverter; + @Autowired private ProtobufJsonFormatHttpMessageConverter protobufConverter; @Bean public AppConfig getAppConfig( @@ -66,9 +65,24 @@ public AppConfig getAppConfig( @Bean public SpecStorage getCoreServiceSpecStorage( @Value("${feast.core.host}") String coreServiceHost, - @Value("${feast.core.grpc.port}") String coreServicePort) { - return new CachedSpecStorage( - new CoreService(coreServiceHost, Integer.parseInt(coreServicePort))); + @Value("${feast.core.grpc.port}") String coreServicePort, + @Value("${feast.cacheDurationMinute}") int cacheDurationMinute) { + ScheduledExecutorService scheduledExecutorService = + Executors.newSingleThreadScheduledExecutor(); + final CachedSpecStorage cachedSpecStorage = + new CachedSpecStorage(new CoreService(coreServiceHost, Integer.parseInt(coreServicePort))); + + // reload all specs including new ones periodically + scheduledExecutorService.schedule( + () -> cachedSpecStorage.populateCache(), cacheDurationMinute, TimeUnit.MINUTES); + + // load all specs during start up + try { + cachedSpecStorage.populateCache(); + } catch (Exception e) { + log.error("Unable to preload feast's spec"); + } + return cachedSpecStorage; } @Bean diff --git a/serving/src/main/java/feast/serving/service/CachedSpecStorage.java b/serving/src/main/java/feast/serving/service/CachedSpecStorage.java index dcbb74f6c88..bddcdbc23f8 100644 --- a/serving/src/main/java/feast/serving/service/CachedSpecStorage.java +++ b/serving/src/main/java/feast/serving/service/CachedSpecStorage.java @@ -20,19 +20,19 @@ import com.google.common.cache.CacheBuilder; import com.google.common.cache.CacheLoader; import com.google.common.cache.LoadingCache; -import lombok.extern.slf4j.Slf4j; import feast.serving.exception.SpecRetrievalException; import feast.specs.EntitySpecProto.EntitySpec; import feast.specs.FeatureSpecProto.FeatureSpec; import feast.specs.StorageSpecProto.StorageSpec; - -import java.time.Duration; import java.util.Collections; import java.util.Map; +import lombok.extern.slf4j.Slf4j; /** SpecStorage implementation with built-in in-memory cache. */ @Slf4j public class CachedSpecStorage implements SpecStorage { + private static final int MAX_SPEC_COUNT = 10000; + private final CoreService coreService; private final LoadingCache entitySpecCache; private final CacheLoader entitySpecLoader; @@ -41,70 +41,24 @@ public class CachedSpecStorage implements SpecStorage { private final LoadingCache storageSpecCache; private final CacheLoader storageSpecLoader; - private static final Duration CACHE_DURATION; - private static final int MAX_SPEC_COUNT = 1000; - - static { - CACHE_DURATION = Duration.ofMinutes(30); - } - public CachedSpecStorage(CoreService coreService) { this.coreService = coreService; entitySpecLoader = - new CacheLoader() { - @Override - public EntitySpec load(String key) throws Exception { - return coreService.getEntitySpecs(Collections.singletonList(key)).get(key); - } - - @Override - public Map loadAll(Iterable keys) throws Exception { - return coreService.getEntitySpecs((Iterable) keys); - } - }; - entitySpecCache = - CacheBuilder.newBuilder() - .maximumSize(MAX_SPEC_COUNT) - .expireAfterAccess(CACHE_DURATION) - .build(entitySpecLoader); + CacheLoader.from( + (String key) -> coreService.getEntitySpecs(Collections.singletonList(key)).get(key)); + entitySpecCache = CacheBuilder.newBuilder().maximumSize(MAX_SPEC_COUNT).build(entitySpecLoader); featureSpecLoader = - new CacheLoader() { - @Override - public FeatureSpec load(String key) throws Exception { - return coreService.getFeatureSpecs(Collections.singletonList(key)).get(key); - } - - @Override - public Map loadAll(Iterable keys) - throws Exception { - return coreService.getFeatureSpecs((Iterable) keys); - } - }; + CacheLoader.from( + (String key) -> coreService.getFeatureSpecs(Collections.singletonList(key)).get(key)); featureSpecCache = - CacheBuilder.newBuilder() - .maximumSize(MAX_SPEC_COUNT) - .expireAfterAccess(CACHE_DURATION) - .build(featureSpecLoader); + CacheBuilder.newBuilder().maximumSize(MAX_SPEC_COUNT).build(featureSpecLoader); storageSpecLoader = - new CacheLoader() { - @Override - public Map loadAll(Iterable keys) - throws Exception { - return coreService.getStorageSpecs((Iterable) keys); - } - - @Override - public StorageSpec load(String key) throws Exception { - return coreService.getStorageSpecs(Collections.singleton(key)).get(key); - } - }; + CacheLoader.from( + (String key) -> coreService.getStorageSpecs(Collections.singletonList(key)).get(key)); storageSpecCache = - CacheBuilder.newBuilder() - .maximumSize(MAX_SPEC_COUNT) - .expireAfterAccess(CACHE_DURATION) - .build(storageSpecLoader); + CacheBuilder.newBuilder().maximumSize(MAX_SPEC_COUNT).build(storageSpecLoader); } @Override @@ -177,4 +131,16 @@ public Map getAllStorageSpecs() { public boolean isConnected() { return coreService.isConnected(); } + + /** Preload all spec into cache. */ + public void populateCache() { + Map featureSpecMap = coreService.getAllFeatureSpecs(); + featureSpecCache.putAll(featureSpecMap); + + Map entitySpecMap = coreService.getAllEntitySpecs(); + entitySpecCache.putAll(entitySpecMap); + + Map storageSpecMap = coreService.getAllStorageSpecs(); + storageSpecCache.putAll(storageSpecMap); + } } diff --git a/serving/src/main/resources/application.properties b/serving/src/main/resources/application.properties index ff14ded7422..0327f187696 100644 --- a/serving/src/main/resources/application.properties +++ b/serving/src/main/resources/application.properties @@ -24,6 +24,7 @@ feast.maxentity=${FEAST_MAX_ENTITY_PER_BATCH:2000} feast.timeout=${FEAST_RETRIEVAL_TIMEOUT:5} feast.redispool.maxsize=${FEAST_REDIS_POOL_MAX_SIZE:128} feast.redispool.maxidle=${FEAST_REDIS_POOL_MAX_IDLE:16} +feast.cacheDurationMinute=${FEAST_SPEC_CACHE_DURATION_MINUTE:5} statsd.host= ${STATSD_HOST:localhost} statsd.port= ${STATSD_PORT:8125} diff --git a/serving/src/test/java/feast/serving/service/CachedSpecStorageTest.java b/serving/src/test/java/feast/serving/service/CachedSpecStorageTest.java new file mode 100644 index 00000000000..76a7b780618 --- /dev/null +++ b/serving/src/test/java/feast/serving/service/CachedSpecStorageTest.java @@ -0,0 +1,106 @@ +package feast.serving.service; + +import static org.hamcrest.Matchers.equalTo; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.google.common.testing.FakeTicker; +import feast.specs.EntitySpecProto.EntitySpec; +import feast.specs.FeatureSpecProto.FeatureSpec; +import feast.specs.StorageSpecProto.StorageSpec; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import org.junit.Before; +import org.junit.Test; + +public class CachedSpecStorageTest { + + private CoreService coreService; + private CachedSpecStorage cachedSpecStorage; + + @Before + public void setUp() throws Exception { + coreService = mock(CoreService.class); + cachedSpecStorage = new CachedSpecStorage(coreService); + } + + @Test + public void testPopulateCache() { + Map featureSpecMap = new HashMap<>(); + featureSpecMap.put("feature_1", mock(FeatureSpec.class)); + + Map storageSpecMap = new HashMap<>(); + storageSpecMap.put("storage_1", mock(StorageSpec.class)); + + Map entitySpecMap = new HashMap<>(); + entitySpecMap.put("entity_1", mock(EntitySpec.class)); + + when(coreService.getAllFeatureSpecs()).thenReturn(featureSpecMap); + when(coreService.getAllEntitySpecs()).thenReturn(entitySpecMap); + when(coreService.getAllStorageSpecs()).thenReturn(storageSpecMap); + + cachedSpecStorage.populateCache(); + Map result = + cachedSpecStorage.getFeatureSpecs(Collections.singletonList("feature_1")); + Map result1 = + cachedSpecStorage.getStorageSpecs(Collections.singletonList("storage_1")); + Map result2 = + cachedSpecStorage.getEntitySpecs(Collections.singletonList("entity_1")); + + assertThat(result.size(), equalTo(1)); + assertThat(result1.size(), equalTo(1)); + assertThat(result2.size(), equalTo(1)); + + verify(coreService, times(0)).getFeatureSpecs(any(Iterable.class)); + verify(coreService, times(0)).getStorageSpecs(any(Iterable.class)); + verify(coreService, times(0)).getEntitySpecs(any(Iterable.class)); + } + + @Test + public void reloadFailureShouldReturnOldValue() { + Map featureSpecMap = new HashMap<>(); + featureSpecMap.put("feature_1", mock(FeatureSpec.class)); + + Map storageSpecMap = new HashMap<>(); + storageSpecMap.put("storage_1", mock(StorageSpec.class)); + + Map entitySpecMap = new HashMap<>(); + entitySpecMap.put("entity_1", mock(EntitySpec.class)); + + when(coreService.getAllFeatureSpecs()).thenReturn(featureSpecMap); + when(coreService.getFeatureSpecs(any(Iterable.class))).thenThrow(new RuntimeException("error")); + when(coreService.getAllEntitySpecs()).thenReturn(entitySpecMap); + when(coreService.getEntitySpecs(any(Iterable.class))).thenThrow(new RuntimeException("error")); + when(coreService.getAllStorageSpecs()).thenReturn(storageSpecMap); + when(coreService.getStorageSpecs(any(Iterable.class))).thenThrow(new RuntimeException("error")); + + cachedSpecStorage.populateCache(); + Map result = + cachedSpecStorage.getFeatureSpecs(Collections.singletonList("feature_1")); + Map result1 = + cachedSpecStorage.getStorageSpecs(Collections.singletonList("storage_1")); + Map result2 = + cachedSpecStorage.getEntitySpecs(Collections.singletonList("entity_1")); + + assertThat(result.size(), equalTo(1)); + assertThat(result1.size(), equalTo(1)); + assertThat(result2.size(), equalTo(1)); + verify(coreService, times(0)).getFeatureSpecs(any(Iterable.class)); + verify(coreService, times(0)).getStorageSpecs(any(Iterable.class)); + verify(coreService, times(0)).getEntitySpecs(any(Iterable.class)); + + result = cachedSpecStorage.getFeatureSpecs(Collections.singletonList("feature_1")); + result1 = cachedSpecStorage.getStorageSpecs(Collections.singletonList("storage_1")); + result2 = cachedSpecStorage.getEntitySpecs(Collections.singletonList("entity_1")); + assertThat(result.size(), equalTo(1)); + assertThat(result1.size(), equalTo(1)); + assertThat(result2.size(), equalTo(1)); + } +} \ No newline at end of file