From 9418555d65482f75dc3b0061f1348f8a9cbdde8f Mon Sep 17 00:00:00 2001 From: Lyn Long Date: Wed, 26 Aug 2026 11:41:35 +1000 Subject: [PATCH] add caching config for co estimation --- .../core/configuration/CacheConfig.java | 7 + .../server/core/service/das/DasService.java | 11 +- .../core/http/RecordingSseConnector.java | 10 + .../core/service/das/DasServiceCacheTest.java | 191 ++++++++++++++++++ 4 files changed, 218 insertions(+), 1 deletion(-) create mode 100644 server/src/test/java/au/org/aodn/ogcapi/server/core/service/das/DasServiceCacheTest.java diff --git a/server/src/main/java/au/org/aodn/ogcapi/server/core/configuration/CacheConfig.java b/server/src/main/java/au/org/aodn/ogcapi/server/core/configuration/CacheConfig.java index a11a116d..2fb96081 100644 --- a/server/src/main/java/au/org/aodn/ogcapi/server/core/configuration/CacheConfig.java +++ b/server/src/main/java/au/org/aodn/ogcapi/server/core/configuration/CacheConfig.java @@ -35,6 +35,7 @@ public class CacheConfig { public static final String DOWNLOADABLE_FIELDS = "downloadable-fields"; public static final String DOWNLOADABLE_SIZE = "downloadable-size"; + public static final String CLOUD_OPTIMISED_ESTIMATE = "cloud-optimised-estimate"; public static final String ALL_NO_LAND_GEOMETRY = "all-noland-geometry"; public static final String ALL_PARAM_VOCABS = "parameter-vocabs"; @@ -91,6 +92,12 @@ public JCacheCacheManager cacheManager() throws IOException { ResourcePoolsBuilder.heap(100) ).withExpiry(ExpiryPolicyBuilder.timeToLiveExpiration(Duration.ofHours(24))) ) + .withCache(CLOUD_OPTIMISED_ESTIMATE, + CacheConfigurationBuilder.newCacheConfigurationBuilder( + Object.class, String.class, + ResourcePoolsBuilder.heap(200) + ).withExpiry(ExpiryPolicyBuilder.timeToLiveExpiration(Duration.ofHours(24))) + ) .withCache(ELASTIC_SEARCH_UUID_ONLY, CacheConfigurationBuilder.newCacheConfigurationBuilder( Object.class, Object.class, diff --git a/server/src/main/java/au/org/aodn/ogcapi/server/core/service/das/DasService.java b/server/src/main/java/au/org/aodn/ogcapi/server/core/service/das/DasService.java index 2477fff7..085efba3 100644 --- a/server/src/main/java/au/org/aodn/ogcapi/server/core/service/das/DasService.java +++ b/server/src/main/java/au/org/aodn/ogcapi/server/core/service/das/DasService.java @@ -1,5 +1,6 @@ package au.org.aodn.ogcapi.server.core.service.das; +import au.org.aodn.ogcapi.server.core.configuration.CacheConfig; import au.org.aodn.ogcapi.server.core.configuration.Config; import au.org.aodn.ogcapi.server.core.model.DatasetMetadata; import au.org.aodn.ogcapi.server.core.service.ApplicationInfo; @@ -7,6 +8,7 @@ import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.cache.annotation.Cacheable; import org.springframework.core.ParameterizedTypeReference; import org.springframework.http.*; import org.springframework.http.codec.ServerSentEvent; @@ -108,7 +110,7 @@ public ResponseEntity getMooringDetailsBetweenDates(String startDateTime /** * Ask DAS for a cloud-optimised size estimate and return the estimate JSON unchanged, for the * SSE layer to forward on. The parameters map is the same subset request the download job - * sends (see SubsetParametersUtils), so DAS treats both alike. Three things to know: + * sends (see SubsetParametersUtils), so DAS treats both alike. Four things to know: * 1. DAS answers over SSE: heartbeats while it computes, then the estimate as a final event. * The stream returns 200 as soon as it opens, so a failed estimate arrives as an error event * that DasSseFrames turns into an exception. Only failures before the stream opens (auth, API @@ -120,7 +122,14 @@ public ResponseEntity getMooringDetailsBetweenDates(String startDateTime * reaches the socket because of CancelPropagatingJdkConnector. * 3. sseIdleTimeout is the gap allowed between frames, not a limit on the whole call. A slow * estimate is fine while DAS keeps heartbeating; a silent DAS is given up on. + * 4. The result is cached for 24 hours under uuid plus parameters, so repeating the same + * subset answers without calling DAS at all. Only the final result is ever cached: heartbeats + * are handed to onHeartbeat as they arrive and never become a return value. onHeartbeat is + * left out of the key on purpose, because it is a new lambda per request and including it + * would make every key unique; on a hit the body never runs, so there is nothing to keep + * alive anyway. A failed estimate throws, and a throwing call caches nothing. */ + @Cacheable(cacheNames = CacheConfig.CLOUD_OPTIMISED_ESTIMATE, key = "{#uuid, #parameters}") public String estimateCloudOptimisedDownloadSize(String uuid, Map parameters, DasSseFrames.FrameCallback onHeartbeat) { diff --git a/server/src/test/java/au/org/aodn/ogcapi/server/core/http/RecordingSseConnector.java b/server/src/test/java/au/org/aodn/ogcapi/server/core/http/RecordingSseConnector.java index 2a279f5b..65026ea5 100644 --- a/server/src/test/java/au/org/aodn/ogcapi/server/core/http/RecordingSseConnector.java +++ b/server/src/test/java/au/org/aodn/ogcapi/server/core/http/RecordingSseConnector.java @@ -38,6 +38,7 @@ public final class RecordingSseConnector implements ClientHttpConnector { private String lastBody; private final AtomicBoolean cancelled = new AtomicBoolean(); private final AtomicInteger framesDelivered = new AtomicInteger(); + private final AtomicInteger requests = new AtomicInteger(); /** * Answer with 200 and these SSE frames, each one a complete frame ending in a blank line. @@ -91,12 +92,21 @@ public int framesDelivered() { return framesDelivered.get(); } + /** + * How many times a WebClient asked to connect, so a test can tell a real call from a + * cached one. + */ + public int requests() { + return requests.get(); + } + @Override public Mono connect(HttpMethod method, URI uri, Function> requestCallback) { MockClientHttpRequest request = new MockClientHttpRequest(method, uri); lastRequest = request; + requests.incrementAndGet(); return requestCallback.apply(request) // Deferred: the mock only has a body once the callback above has written one. diff --git a/server/src/test/java/au/org/aodn/ogcapi/server/core/service/das/DasServiceCacheTest.java b/server/src/test/java/au/org/aodn/ogcapi/server/core/service/das/DasServiceCacheTest.java new file mode 100644 index 00000000..5d5a9363 --- /dev/null +++ b/server/src/test/java/au/org/aodn/ogcapi/server/core/service/das/DasServiceCacheTest.java @@ -0,0 +1,191 @@ +package au.org.aodn.ogcapi.server.core.service.das; + +import au.org.aodn.ogcapi.server.core.configuration.CacheConfig; +import au.org.aodn.ogcapi.server.core.http.RecordingSseConnector; +import au.org.aodn.ogcapi.server.core.util.DasSseFrames; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.cache.CacheManager; +import org.springframework.cache.annotation.EnableCaching; +import org.springframework.cache.concurrent.ConcurrentMapCacheManager; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.web.client.RestTemplate; +import org.springframework.web.reactive.function.client.WebClient; + +import java.time.Duration; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +/** + * Covers the cache in front of the cloud-optimised size estimate: a repeat of the same subset is + * answered without calling DAS, a different subset is not, and a failure leaves nothing behind. + * The other DAS tests build the service with new, which means no proxy and no caching, so caching + * only shows up with a Spring context around the bean. What is under test here is the annotation + * and its key, not EhCache, so a plain ConcurrentMapCacheManager stands in for the real one. + */ +public class DasServiceCacheTest { + + private static final String HOST = "http://localhost:5000"; + + private static final String HEARTBEAT_FRAME = """ + event: processing + data: {"status":"processing","message":"Processing your request..."} + + """; + + private static final String RESULT_FRAME = """ + event: result + data: {"status":"completed","data":{"estimated_output_bytes":123}} + + """; + + private static final String OTHER_RESULT_FRAME = """ + event: result + data: {"status":"completed","data":{"estimated_output_bytes":456}} + + """; + + private static final String ERROR_FRAME = """ + event: error + data: {"status":"error","message":"boom"} + + """; + + /** + * proxyTargetClass matches what Spring Boot does by default. DasService implements + * ApplicationInfo, so a JDK proxy would only expose that interface and the service could not + * be looked up, or injected into RestServices, by its own type. + */ + @Configuration + @EnableCaching(proxyTargetClass = true) + static class CachingContext { + + @Bean + public CacheManager cacheManager() { + return new ConcurrentMapCacheManager(CacheConfig.CLOUD_OPTIMISED_ESTIMATE); + } + + @Bean + public RecordingSseConnector connector() { + return new RecordingSseConnector(); + } + + @Bean + public DasService dasService(RecordingSseConnector connector) { + // A null infoPath keeps the constructor's info query from making a request. + DasProperties properties = new DasProperties(HOST, null, "test-secret", null, + Duration.ofSeconds(5), Duration.ofSeconds(30), Duration.ofMinutes(2)); + WebClient webClient = WebClient.builder() + .clientConnector(connector) + .baseUrl(HOST) + .build(); + return new DasService(properties, new RestTemplate(), webClient, new ObjectMapper()); + } + } + + private AnnotationConfigApplicationContext context; + private DasService dasService; + private RecordingSseConnector connector; + + @BeforeEach + public void setUp() { + context = new AnnotationConfigApplicationContext(CachingContext.class); + dasService = context.getBean(DasService.class); + connector = context.getBean(RecordingSseConnector.class); + } + + @AfterEach + public void tearDown() { + context.close(); + } + + /** + * A new map each time, the way SubsetParametersUtils builds one per request, so the tests + * show the key matching on content rather than on the same instance coming back. + */ + private static Map parameters(String outputFormat) { + Map parameters = new HashMap<>(); + parameters.put("uuid", "test-uuid"); + parameters.put("key", "a.zarr"); + parameters.put("start_date", "2020-01-01"); + parameters.put("end_date", "2020-12-31"); + parameters.put("multi_polygon", "non-specified"); + parameters.put("output_format", outputFormat); + return parameters; + } + + /** + * A distinct callback instance each call, the way RestServices passes a new + * session::probeClient per request. Reusing one constant here would hide a key that wrongly + * included the callback, because that key would still match on the second call. + */ + private static DasSseFrames.FrameCallback freshCallback() { + return new AtomicInteger()::incrementAndGet; + } + + private String estimate(Map parameters, DasSseFrames.FrameCallback onHeartbeat) { + return dasService.estimateCloudOptimisedDownloadSize("test-uuid", parameters, onHeartbeat); + } + + @Test + public void testRepeatedEstimateIsServedFromCache() { + connector.respondWith(List.of(RESULT_FRAME)); + + String first = estimate(parameters("netcdf"), freshCallback()); + String second = estimate(parameters("netcdf"), freshCallback()); + + assertEquals("{\"estimated_output_bytes\":123}", first); + assertEquals(first, second, "The repeat returns the same estimate"); + assertEquals(1, connector.requests(), "The repeat is answered from the cache, not by data-access-service"); + } + + @Test + public void testDifferentParametersStillCallDas() { + connector.respondWith(List.of(RESULT_FRAME)); + estimate(parameters("netcdf"), freshCallback()); + + connector.respondWith(List.of(OTHER_RESULT_FRAME)); + String second = estimate(parameters("csv"), freshCallback()); + + assertEquals("{\"estimated_output_bytes\":456}", second); + assertEquals(2, connector.requests(), "A different output format is a different subset, so a different key"); + } + + @Test + public void testFailedEstimateIsNotCached() { + connector.respondWith(List.of(ERROR_FRAME)); + assertThrows(RuntimeException.class, + () -> estimate(parameters("netcdf"), freshCallback())); + + connector.respondWith(List.of(RESULT_FRAME)); + String retry = estimate(parameters("netcdf"), freshCallback()); + + assertEquals("{\"estimated_output_bytes\":123}", retry); + assertEquals(2, connector.requests(), "A failure caches nothing, so the retry reaches data-access-service"); + } + + /** + * The callback is a new lambda per request, so it has to stay out of the key. A hit that + * still ran the body, or a key that included the callback, would both show up here. + */ + @Test + public void testCacheHitSkipsTheStreamAndItsHeartbeats() { + connector.respondWith(List.of(HEARTBEAT_FRAME, RESULT_FRAME)); + + AtomicInteger heartbeats = new AtomicInteger(); + estimate(parameters("netcdf"), heartbeats::incrementAndGet); + assertEquals(1, heartbeats.get(), "The first estimate reads the DAS stream"); + + estimate(parameters("netcdf"), heartbeats::incrementAndGet); + assertEquals(1, heartbeats.get(), "A cache hit never opens a stream, so nothing heartbeats"); + } +}