cda-37 Store blobs in object-store

cwms-data-api/build.gradle

@@ -151,6 +151,8 @@ dependencies {
     implementation(libs.bundles.overrides)
 
     testImplementation(libs.bundles.java.parser)
+    implementation(libs.togglz.core)
+    implementation(libs.minio)
 }
 
 task extractWebJars(type: Copy) {
@@ -245,7 +247,7 @@ task run(type: JavaExec) {
 }
 
 task integrationTests(type: Test) {
-    dependsOn test
+//    dependsOn test
     dependsOn generateConfig
     dependsOn war
 

cwms-data-api/src/main/java/cwms/cda/api/BinaryTimeSeriesValueController.java

@@ -29,16 +29,18 @@
 import com.codahale.metrics.Timer;
 import cwms.cda.api.errors.CdaError;
 import cwms.cda.data.dao.BlobDao;
+import cwms.cda.data.dao.StreamConsumer;
+import io.javalin.core.util.Header;
 import io.javalin.http.Context;
 import io.javalin.http.Handler;
 import io.javalin.plugin.openapi.annotations.OpenApi;
 import io.javalin.plugin.openapi.annotations.OpenApiContent;
 import io.javalin.plugin.openapi.annotations.OpenApiParam;
 import io.javalin.plugin.openapi.annotations.OpenApiResponse;
+import org.jetbrains.annotations.NotNull;
 import org.jooq.DSLContext;
 
 import javax.servlet.http.HttpServletResponse;
-import java.io.InputStream;
 
 import static com.codahale.metrics.MetricRegistry.name;
 import static cwms.cda.api.Controllers.*;
@@ -84,26 +86,39 @@ private Timer.Context markAndTime(String subject) {
                     )},
             tags = {BinaryTimeSeriesController.TAG}
     )
-    public void handle(Context ctx) {
+    public void handle(@NotNull Context ctx) {
         //Implementation will change with new CWMS schema
         //https://www.hec.usace.army.mil/confluence/display/CWMS/2024-02-29+Task2A+Text-ts+and+Binary-ts+Design
         try (Timer.Context ignored = markAndTime(GET_ALL)) {
             String binaryId = requiredParam(ctx, BLOB_ID);
             String officeId = requiredParam(ctx, OFFICE);
             DSLContext dsl = getDslContext(ctx);
+
+            ctx.header(Header.ACCEPT_RANGES, "bytes");
+
+            final Long offset;
+            final Long end ;
+            long[] ranges = RangeParser.parseFirstRange(ctx.header(io.javalin.core.util.Header.RANGE));
+            if (ranges != null) {
+                offset = ranges[0];
+                end = ranges[1];
+            } else {
+                offset = null;
+                end = null;
+            }
+
             BlobDao blobDao = new BlobDao(dsl);
-            blobDao.getBlob(binaryId, officeId, (blob, mediaType) -> {
-                if (blob == null) {
+            StreamConsumer streamConsumer = (is, isPosition, mediaType, totalLength) -> {
+                if (is == null) {
                     ctx.status(HttpServletResponse.SC_NOT_FOUND).json(new CdaError("Unable to find "
                             + "blob based on given parameters"));
                 } else {
-                    long size = blob.length();
-                    requestResultSize.update(size);
-                    try (InputStream is = blob.getBinaryStream()) {
-                        RangeRequestUtil.seekableStream(ctx, is, mediaType, size);
-                    }
+                    requestResultSize.update(totalLength);
+                    RangeRequestUtil.seekableStream(ctx, is, isPosition, mediaType, totalLength);
                 }
-            });
+            };
+
+            blobDao.getBlob(binaryId, officeId, streamConsumer, offset, end);
         }
     }
 }

cwms-data-api/src/main/java/cwms/cda/api/ClobController.java

@@ -27,6 +27,7 @@
 import cwms.cda.api.errors.CdaError;
 import cwms.cda.data.dao.ClobDao;
 import cwms.cda.data.dao.JooqDao;
+import cwms.cda.data.dao.StreamConsumer;
 import cwms.cda.data.dto.Clob;
 import cwms.cda.data.dto.Clobs;
 import cwms.cda.data.dto.CwmsDTOPaginated;
@@ -43,7 +44,7 @@
 import io.javalin.plugin.openapi.annotations.OpenApiParam;
 import io.javalin.plugin.openapi.annotations.OpenApiRequestBody;
 import io.javalin.plugin.openapi.annotations.OpenApiResponse;
-import java.io.InputStream;
+
 import java.util.Objects;
 import java.util.Optional;
 import javax.servlet.http.HttpServletResponse;
@@ -187,16 +188,16 @@ public void getOne(@NotNull Context ctx, @NotNull String clobId) {
             if (TEXT_PLAIN.equals(formatHeader)) {
                 // useful cmd:  curl -X 'GET' 'http://localhost:7000/cwms-data/clobs/encoded?office=SPK&id=%2FTIME%20SERIES%20TEXT%2F6261044'
                 // -H 'accept: text/plain' --header "Range: bytes=20000-40000"
-                dao.getClob(clobId, office, c -> {
-                    if (c == null) {
-                        ctx.status(HttpServletResponse.SC_NOT_FOUND).json(new CdaError("Unable to find "
-                                + "clob based on given parameters"));
-                    } else {
-                        try (InputStream is = c.getAsciiStream()) {
-                            RangeRequestUtil.seekableStream(ctx, is, TEXT_PLAIN, c.length());
-                        }
-                    }
-                });
+
+                ctx.header(Header.ACCEPT_RANGES, "bytes");
+
+                StreamConsumer consumer = (is, isPosition, mediaType, totalLength) -> {
+                        requestResultSize.update(totalLength);
+                        RangeRequestUtil.seekableStream(ctx, is, isPosition, mediaType, totalLength);
+                };
+
+                dao.getClob(clobId, office, consumer);
+
             } else {
                 Optional<Clob> optAc = dao.getByUniqueName(clobId, office);
 

cwms-data-api/src/main/java/cwms/cda/api/ForecastFileController.java

@@ -29,16 +29,18 @@
 import com.codahale.metrics.Timer;
 import cwms.cda.api.errors.CdaError;
 import cwms.cda.data.dao.ForecastInstanceDao;
+import cwms.cda.data.dao.StreamConsumer;
 import cwms.cda.helpers.DateUtils;
+import io.javalin.core.util.Header;
 import io.javalin.http.Context;
 import io.javalin.http.Handler;
 import io.javalin.plugin.openapi.annotations.OpenApi;
 import io.javalin.plugin.openapi.annotations.OpenApiContent;
 import io.javalin.plugin.openapi.annotations.OpenApiParam;
 import io.javalin.plugin.openapi.annotations.OpenApiResponse;
+import org.jetbrains.annotations.NotNull;
 
 import javax.servlet.http.HttpServletResponse;
-import java.io.InputStream;
 import java.time.Instant;
 
 import static com.codahale.metrics.MetricRegistry.name;
@@ -92,7 +94,7 @@ private Timer.Context markAndTime(String subject) {
             },
             tags = {ForecastSpecController.TAG}
     )
-    public void handle(Context ctx) {
+    public void handle(@NotNull Context ctx) {
         String specId = requiredParam(ctx, NAME);
         String office = requiredParam(ctx, OFFICE);
         String designator = ctx.queryParamAsClass(DESIGNATOR, String.class).allowNullable().get();
@@ -102,18 +104,17 @@ public void handle(Context ctx) {
         Instant issueInstant = DateUtils.parseUserDate(issueDate, "UTC").toInstant();
         try (Timer.Context ignored = markAndTime(GET_ALL)) {
             ForecastInstanceDao dao = new ForecastInstanceDao(getDslContext(ctx));
-            dao.getFileBlob(office, specId, designator, forecastInstant, issueInstant, (blob, mediaType) -> {
-                if (blob == null) {
+            StreamConsumer streamConsumer = (is, isPosition, mediaType, totalLength) -> {
+                if (is == null) {
                     ctx.status(HttpServletResponse.SC_NOT_FOUND).json(new CdaError("Unable to find "
                             + "blob based on given parameters"));
                 } else {
-                    long size = blob.length();
-                    requestResultSize.update(size);
-                    try (InputStream is = blob.getBinaryStream()) {
-                        RangeRequestUtil.seekableStream(ctx, is, mediaType, size);
-                    }
+                    requestResultSize.update(totalLength);
+                    ctx.header(Header.ACCEPT_RANGES, "bytes");
+                    RangeRequestUtil.seekableStream(ctx, is, isPosition, mediaType, totalLength);
                 }
-            });
+            };
+            dao.getFileBlob(office, specId, designator, forecastInstant, issueInstant, streamConsumer);
         }
     }
 }

cwms-data-api/src/main/java/cwms/cda/api/RangeParser.java

@@ -0,0 +1,128 @@
+package cwms.cda.api;
+
+import org.jetbrains.annotations.NotNull;
+
+import java.util.*;
+import java.util.regex.*;
+
+/**
+ * Utility class for parsing HTTP Range headers.
+ *  These typically look like: bytes=100-1234
+ *              or: bytes=100-  this is common to resume a download
+ *              or: bytes=0- equivalent to a regular request for the whole file
+ *              but by returning 206 we show that we support range requests
+ *              Note that multiple ranges can be requested at once such
+ *              as: bytes=500-600,700-999 Server responds identifies separator and then puts separator between chunks
+ *              bytes=0-0,-1 also legal its just the first and the last byte
+ *              or: bytes=500-600,601-999 legal but what is the point?
+ *              or: bytes=500-700,601-999 legal, notice they overlap.
+ *
+ *
+ */
+public class RangeParser {
+
+    private static final Pattern RANGE_PATTERN = Pattern.compile("(\\d*)-(\\d*)");
+
+    /**
+     * Return a list of two element long[] containing byte ranges parsed from the HTTP Range header.
+     * If the end of a range is not specified ( e.g. bytes=100- ) then a -1 is returned in the second position
+     * If the range only includes a negative byte (e.g bytes=-50) then -1 is returned as the start of the range
+     * and -1*end is returned as the end of the range.    bytes=-50 will result in [-1,50]
+     *
+     * @param header the HTTP Range header this should start with "bytes=" if it is null or empty an empty list is returned
+     * @return a list of long[2] holding the ranges
+     */
+    public static List<long[]> parse(String header) {
+        if (header == null || header.isEmpty() ) {
+            return Collections.emptyList();
+        } else if ( !header.startsWith("bytes=")){
+            throw new IllegalArgumentException("Invalid Range header: " + header);
+        }
+
+        String rangePart = header.substring(6);
+        List<long[]> retval = parseRanges(rangePart);
+        if( retval.isEmpty() ){
+            throw new IllegalArgumentException("Invalid Range header: " + header);
+        }
+        return retval;
+    }
+
+    public static long[] parseFirstRange(String header) {
+        if(header != null) {
+            List<long[]> ranges = RangeParser.parse(header);
+            if (!ranges.isEmpty()) {
+                return ranges.get(0);
+            }
+        }
+        return null;
+    }
+
+    public static @NotNull List<long[]> parseRanges(String rangePart) {
+        if( rangePart == null || rangePart.isEmpty() ){
+            throw new IllegalArgumentException("Invalid range specified: " + rangePart);
+        }
+        String[] parts = rangePart.split(",");
+        List<long[]> ranges = new ArrayList<>();
+
+        for (String part : parts) {
+            Matcher m = RANGE_PATTERN.matcher(part.trim());
+            if (m.matches()) {
+                String start = m.group(1);
+                String end = m.group(2);
+
+                long s = start.isEmpty() ? -1 : Long.parseLong(start);
+                long e = end.isEmpty() ? -1 : Long.parseLong(end);
+
+                ranges.add(new long[]{s, e});
+            }
+        }
+        return ranges;
+    }
+
+    /**
+     * The parse() method in this class can return -1 for unspecified values or when suffix ranges are supplied.
+     * This method interprets the negative values in regard to the totalSize and returns inclusive indices of the
+     * requested range.
+     * @param inputs the array of start and end byte positions
+     * @param totalBytes the total number of bytes in the file
+     * @return a long array with the start and end byte positions, these are inclusive.  [0,0] means return the first byte
+     */
+    public static long[] interpret(long[] inputs, long totalBytes){
+        if(inputs == null){
+            throw new IllegalArgumentException("null range array provided");
+        } else if( inputs.length != 2 ){
+            throw new IllegalArgumentException("Invalid number of inputs: " + Arrays.toString(inputs));
+        }
+
+        long start = inputs[0];
+        long end = inputs[1];
+
+        if(start == -1L){
+            // its a suffix request.
+            start = totalBytes - end;
+            end = totalBytes - 1;
+        } else {
+            if (start < 0 ) {
+                throw new IllegalArgumentException("Invalid range specified: " + Arrays.toString(inputs));
+            }
+
+            if(end == -1L){
+                end = totalBytes - 1;
+            }
+
+            if(end < start){
+                throw new IllegalArgumentException("Invalid range specified: " + Arrays.toString(inputs));
+            }
+
+            if(start > totalBytes - 1){
+                throw new IllegalArgumentException("Can't satisfy range request: " + Arrays.toString(inputs) + " Range starts beyond end of file.");
+            }
+
+            end = Math.min(end, totalBytes - 1);
+        }
+
+        return new long[]{start, end};
+    }
+
+
+}