fix(collections): default page_size, load batches up to cap, survive 1 cpu

Dockerfile

@@ -6,6 +6,9 @@ ENV MAX_HEAP_PERCENTAGE=70
 COPY ./server/target/ogcapi-java-server-*-exec.jar app.jar
 ENTRYPOINT ["/bin/sh", "-c", "java \
     -XX:MaxRAMPercentage=${MAX_HEAP_PERCENTAGE} \
+    -XX:+ExitOnOutOfMemoryError \
+    -XX:+HeapDumpOnOutOfMemoryError \
+    -XX:HeapDumpPath=/tmp \
     -Duser.timezone=UTC \
     -Delasticsearch.index.name=${INDEX_NAME} \
     -Delasticsearch.cloud_optimized_index.name=${CO_INDEX_NAME} \

server/src/main/java/au/org/aodn/ogcapi/server/common/RestApi.java

@@ -26,6 +26,7 @@
 import org.apache.commons.lang3.NotImplementedException;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
 import org.springframework.http.ResponseEntity;
@@ -52,6 +53,9 @@ public class RestApi implements ApiApi, DefaultApi, ConformanceApi {
     @Qualifier("FeaturesRestService")
     protected OGCApiService featuresService;
 
+    @Value("${ogcapi.collections.default-page-size:800}")
+    protected int defaultPageSize;
+
     @Autowired
     protected StacToCollections stacToCollection;
 
@@ -162,6 +166,7 @@ public ResponseEntity<?> getCollections(
                 // the same, we append the bbox parameter to the filter in case user use this parameter
                 filter = OGCApiService.processBBoxParameter(CQLFields.geometry.name(), bbox, filter);
             }
+            filter = OGCApiService.addDefaultPageSize(filter, defaultPageSize);
             return commonService.getCollectionList(
                     q,
                     filter,

server/src/main/java/au/org/aodn/ogcapi/server/core/configuration/ElasticSearchConfig.java

@@ -64,7 +64,7 @@ public Search createElasticSearch(ElasticsearchClient client,
                                       CacheNoLandGeometry cacheNoLandGeometry,
                                       ObjectMapper mapper,
                                       @Value("${elasticsearch.index.name}") String indexName,
-                                      @Value("${elasticsearch.index.pageSize:2200}") Integer pageSize,
+                                      @Value("${elasticsearch.index.pageSize:800}") Integer pageSize,
                                       @Value("${elasticsearch.index.lightweightPageSize:10000}") Integer lightweightPageSize,
                                       @Value("${elasticsearch.search_as_you_type.size:10}") Integer searchAsYouTypeSize) {
 

server/src/main/java/au/org/aodn/ogcapi/server/core/service/ElasticSearchBase.java

@@ -354,12 +354,12 @@ protected SearchResult<StacCollectionModel> searchCollectionBy(final List<Query>
                                                                    final Double score,
                                                                    final Long maxSize) {
         Supplier<SearchRequest.Builder> builderSupplier = buildCollectionSearchRequestSupplier(
-                queries, should, filters, properties, searchAfter, sortOptions, score, maxSize
+                queries, should, filters, properties, null, sortOptions, score, maxSize
         );
 
         try {
             log.info("Start search {} {}", ZonedDateTime.now(), Thread.currentThread().getName());
-            Iterable<Hit<ObjectNode>> response = pageableSearch(builderSupplier, ObjectNode.class, maxSize);
+            Iterable<Hit<ObjectNode>> response = pageableSearch(builderSupplier, ObjectNode.class, maxSize, searchAfter);
 
             SearchResult<StacCollectionModel> result = new SearchResult<>();
             result.collections = new ArrayList<>();
@@ -515,12 +515,19 @@ protected Long countRecordsHit(Supplier<SearchRequest.Builder> requestBuilder) {
      *
      * @param requestBuilder, assume it is sorted with order, what order isn't important, as long as it is sorted
      * @param clazz - The type
+     * @param maxSize - Stop after this many hits, null means every hit
+     * @param searchAfter - Cursor for the first batch, later batches use the last hit's sort values. The supplier
+     *                    must not set it because the builder appends search_after values instead of replacing them
      * @return - The items that matches the query mentioned in the requestBuilder
      * @param <T> A generic type for Elastic query
      */
-    protected <T> Iterable<Hit<T>> pageableSearch(Supplier<SearchRequest.Builder> requestBuilder, Class<T> clazz, Long maxSize) {
+    protected <T> Iterable<Hit<T>> pageableSearch(Supplier<SearchRequest.Builder> requestBuilder, Class<T> clazz, Long maxSize, List<FieldValue> searchAfter) {
         try {
-            SearchRequest sr = requestBuilder.get().build();
+            SearchRequest.Builder first = requestBuilder.get();
+            if(searchAfter != null) {
+                first.searchAfter(searchAfter);
+            }
+            SearchRequest sr = first.build();
             log.debug("Final elastic search payload {}", sr);
 
             final AtomicLong count = new AtomicLong(0);
@@ -533,9 +540,9 @@ protected <T> Iterable<Hit<T>> pageableSearch(Supplier<SearchRequest.Builder> re
 
                 @Override
                 public boolean hasNext() {
-                    // No need continue if we already hit the end
-                    if(maxSize != null) {
-                        return count.get() < maxSize;
+                    // Stop once the caller's cap is reached, otherwise keep loading batches below
+                    if(maxSize != null && count.get() >= maxSize) {
+                        return false;
                     }
                     // If we hit the end, that means we have iterated to end of page.
                     if (index < response.get().hits().hits().size()) {

server/src/main/java/au/org/aodn/ogcapi/server/core/service/OGCApiService.java

@@ -20,6 +20,7 @@
 import java.math.BigDecimal;
 import java.util.List;
 import java.util.function.BiFunction;
+import java.util.regex.Pattern;
 
 /**
  *
@@ -28,6 +29,8 @@ public abstract class OGCApiService {
 
     protected Logger logger = LoggerFactory.getLogger(RestApi.class);
 
+    protected static final Pattern PAGE_SIZE_IN_FILTER = Pattern.compile("\\bpage_size\\s*=", Pattern.CASE_INSENSITIVE);
+
     @Autowired
     protected Search search;
 
@@ -174,4 +177,15 @@ else if(bbox.size() == 6) {
             return String.join(" AND ", filter, f);
         }
     }
+
+    /**
+     * Append the default page_size when the CQL filter has none, so a bare /collections never loads the whole index.
+     */
+    public static String addDefaultPageSize(String filter, int defaultPageSize) {
+        if (filter != null && PAGE_SIZE_IN_FILTER.matcher(filter).find()) {
+            return filter;
+        }
+        String f = "page_size=" + defaultPageSize;
+        return (filter == null || filter.isBlank()) ? f : String.join(" AND ", filter, f);
+    }
 }

server/src/main/java/au/org/aodn/ogcapi/server/core/util/GeometryUtils.java

@@ -54,10 +54,10 @@ public class GeometryUtils {
     @Setter
     protected static int centroidScale = 5;
 
-    // Create an ExecutorService with a fixed thread pool size
+    // Create an ExecutorService with a fixed thread pool size, at least 1 so a 1 cpu container can still start
     @Getter
     @Setter
-    protected static ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() - 1);
+    protected static ExecutorService executorService = Executors.newFixedThreadPool(Math.max(1, Runtime.getRuntime().availableProcessors() - 1));
 
     protected static Logger logger = LoggerFactory.getLogger(GeometryUtils.class);
 

server/src/main/resources/application.yaml

@@ -6,6 +6,9 @@ server:
     min-response-size: 1024 # Minimum response size in bytes to compress
 
 ogcapi:
+  collections:
+    # page_size applied when the CQL filter has none, the portal's largest page
+    default-page-size: 800
   debug:
     elasticsearch-explain-enabled: false
   http-cache:
@@ -21,8 +24,8 @@ ogcapi:
 elasticsearch:
   index:
     name: dev_portal_records
-    # search_after batch for full-document / geometry / links queries
-    pageSize: 2200
+    # search_after batch for full-document / geometry / links queries, the portal's largest page
+    pageSize: 800
     # search_after batch when properties is a small field list (id, temporal, title, ...)
     lightweightPageSize: 7000
   vocabs_index:

server/src/test/java/au/org/aodn/ogcapi/server/BaseTestClass.java

@@ -232,6 +232,7 @@ protected void insertJsonToElasticIndex(String index, String[] filenames) throws
         // Check the number of doc store inside the ES instance is correct
         SearchRequest.Builder b = new SearchRequest.Builder()
                 .index(index)
+                .size(filenames.length)   // elastic returns 10 hits by default, tests may insert more files
                 .query(QueryBuilders.matchAll().build()._toQuery());
 
         SearchRequest request = b.build();

server/src/test/java/au/org/aodn/ogcapi/server/core/service/OGCApiServiceTest.java

@@ -31,4 +31,19 @@ public void verifyProcessDatetimeParameter() {
         o = OGCApiService.processDatetimeParameter(CQLFields.temporal.name(),"/2021-10-10", "geometry is null");
         assertEquals( "geometry is null AND temporal before 2021-10-10", o, "Before plus filter incorrect1");
     }
+
+    /**
+     * A default page_size is appended only when the CQL filter has none
+     */
+    @Test
+    public void verifyAddDefaultPageSize() {
+        String o = OGCApiService.addDefaultPageSize(null, 10);
+        assertEquals("page_size=10", o, "Default when no filter");
+
+        o = OGCApiService.addDefaultPageSize("page_size=3", 10);
+        assertEquals("page_size=3", o, "Filter page_size kept");
+
+        o = OGCApiService.addDefaultPageSize("temporal after 2021-10-10", 10);
+        assertEquals("temporal after 2021-10-10 AND page_size=10", o, "Default appended to filter");
+    }
 }

server/src/test/java/au/org/aodn/ogcapi/server/features/RestApiTest.java

@@ -498,4 +498,64 @@ public void verifyBBoxCorrect() throws IOException {
         assertEquals(154.0, bbox.get(0).get(2).doubleValue(), "Overall bounding box coor 3");
         assertEquals(-9.0, bbox.get(0).get(3).doubleValue(), "Overall bounding box coor 4");
     }
+
+    /**
+     * Without page_size in the filter the server applies the configured default, 10 in the test profile,
+     * so a bare call never loads the whole index
+     */
+    @Test
+    public void verifyDefaultPageSizeCapsCollections() throws IOException {
+        super.insertJsonToElasticRecordIndex(
+                "5c418118-2581-4936-b6fd-d6bedfe74f62.json",
+                "19da2ce7-138f-4427-89de-a50c724f5f54.json",
+                "516811d7-cd1e-207a-e0440003ba8c79dd.json",
+                "7709f541-fc0c-4318-b5b9-9053aa474e0e.json",
+                "bc55eff4-7596-3565-e044-00144fdd4fa6.json",
+                "bf287dfe-9ce4-4969-9c59-51c39ea4d011.json",
+                "073fde5a-bff3-1c1f-e053-08114f8c5588.json",
+                "ae86e2f5-eaaf-459e-a405-e654d85adb9c.json",
+                "caf7220a-19e0-4a7f-9af6-eade6c79a47a.json",
+                "b299cdcd-3dee-48aa-abdd-e0fcdbb9cadc.json",
+                "35234913-aa3c-48ec-b9a4-77f822f66ef8.json",
+                "e26d0a56-5603-4413-911d-7b359a533a75.json");
+
+        ResponseEntity<ExtendedCollections> collections = testRestTemplate.exchange(
+                getBasePath() + "/collections",
+                HttpMethod.GET,
+                null,
+                new ParameterizedTypeReference<>() {
+                });
+
+        assertEquals(HttpStatus.OK, collections.getStatusCode(), "Get status OK");
+        assertEquals(10, Objects.requireNonNull(collections.getBody()).getCollections().size(), "Test profile default page_size is 10");
+        assertEquals(12, collections.getBody().getTotal(), "Total still counts every record");
+        assertEquals(3, collections.getBody().getSearchAfter().size(), "search_after given for the next page");
+    }
+
+    /**
+     * page_size larger than one elastic batch keeps loading batches until page_size, test batch size is 4
+     */
+    @Test
+    public void verifyPageSizeSpansElasticBatches() throws IOException {
+        assertEquals(4, pageSize, "This test only works with small page");
+
+        super.insertJsonToElasticRecordIndex(
+                "5c418118-2581-4936-b6fd-d6bedfe74f62.json",
+                "19da2ce7-138f-4427-89de-a50c724f5f54.json",
+                "516811d7-cd1e-207a-e0440003ba8c79dd.json",
+                "7709f541-fc0c-4318-b5b9-9053aa474e0e.json",
+                "bc55eff4-7596-3565-e044-00144fdd4fa6.json",
+                "bf287dfe-9ce4-4969-9c59-51c39ea4d011.json");
+
+        ResponseEntity<ExtendedCollections> collections = testRestTemplate.exchange(
+                getBasePath() + "/collections?filter=page_size=5",
+                HttpMethod.GET,
+                null,
+                new ParameterizedTypeReference<>() {
+                });
+
+        assertEquals(HttpStatus.OK, collections.getStatusCode(), "Get status OK");
+        assertEquals(5, Objects.requireNonNull(collections.getBody()).getCollections().size(), "page_size=5 spans two batches of 4");
+        assertEquals(6, collections.getBody().getTotal(), "Total still counts every record");
+    }
 }

server/src/test/resources/application-test.yaml

@@ -1,4 +1,6 @@
 ogcapi:
+  collections:
+    default-page-size: 10
   docker:
     elasticVersion: "8.19.10"
   debug: