improve scanner scanning to ensure only values within the given range are returned

Author: cwensel

clusterless-main-common/src/main/java/clusterless/cls/command/exec/ArcExecCommandOptions.java

@@ -23,6 +23,12 @@ public class ArcExecCommandOptions extends CommonCommandOptions {
     @CommandLine.Mixin
     ArcCommonOptions arcCommonOptions = new ArcCommonOptions();
 
+    @CommandLine.Option(
+            names = "--dry-run",
+            description = "Do not execute the arc."
+    )
+    private boolean dryRun = false;
+
     static class RangeOrLot {
         @CommandLine.ArgGroup(
                 exclusive = false,
@@ -69,6 +75,10 @@ public ArcExecCommandOptions setNames(List<String> names) {
         return this;
     }
 
+    public boolean dryRun() {
+        return dryRun;
+    }
+
     public RangeOptions rangeOptions() {
         return rangeOrLot.rangeOptions;
     }

clusterless-substrate-aws-kernel/src/main/java/clusterless/cls/substrate/aws/report/ArcExec.java

@@ -150,15 +150,15 @@ private void handleScanner(Printer printer, ArcStatusScanner arcStatusScanner, A
         }
     }
 
-    private static void execStepFunction(Printer printer, StepFunction stepFunction, ArcStatusRecord arcStatusRecord, SinkDataset sink, ManifestURI manifestURI, ArcMeta arcMeta, Placement placement) {
+    private void execStepFunction(Printer printer, StepFunction stepFunction, ArcStatusRecord arcStatusRecord, SinkDataset sink, ManifestURI manifestURI, ArcMeta arcMeta, Placement placement) {
         printer.print("starting arc exec for");
         printer.print(" placement: " + arcStatusRecord.arcRecord().placement().display());
         printer.print(" project: " + arcStatusRecord.arcRecord().project().display());
         printer.print(" arc: " + arcStatusRecord.arcRecord().name());
         printer.print(" dataset: " + sink.display());
         printer.print(" state: " + Strings.nullToEmpty(arcStatusRecord.state()));
         printer.print(" lot: " + arcStatusRecord.lotId());
-        printer.print(" manifest: " + manifestURI.uri());
+        printer.print(" source manifest: " + manifestURI.uri());
         printer.println();
 
         ArcNotifyEvent notifyEvent = ArcNotifyEvent.builder()
@@ -172,6 +172,11 @@ private static void execStepFunction(Printer printer, StepFunction stepFunction,
         ArcDeployment arcDeployment = arcMeta.arcDeployment();
         String stepFunctionName = arcDeployment.stepFunctionName();
 
+        if (arcExecCommandOptions.dryRun()) {
+            printer.println("dry-run, not executing arc");
+            return;
+        }
+
         stepFunction.start(placement.account(), stepFunctionName, payload)
                 .isSuccessOrThrow(e -> {
                     throw new IllegalStateException("unable to start step function: " + stepFunctionName + ", " + e.getMessage(), e);

clusterless-substrate-aws-kernel/src/main/java/clusterless/cls/substrate/aws/report/scanner/ArcStatusScanner.java

@@ -41,6 +41,11 @@ public ArcStatusScanner(String profile, ArcRecord arcRecord, Moment earliest, Mo
         this.arcStateSupplier = arcStateSupplier;
     }
 
+    @Override
+    protected String scannerType() {
+        return "ArcStatus";
+    }
+
     @Override
     protected StateURI<?, ?> createStateURIFrom(ArcRecord record) {
         return ArcStateURI.builder()
@@ -56,19 +61,24 @@ protected Stream<ArcStatusRecord> parseUriStreamIntoStatusRec(Stream<String> res
         Stream<ArcStatusRecord> arcStatusRecordStream;
 
         arcStatusRecordStream = resultStream.map(ArcStateURI::parse)
+                .filter(uri -> uri.lotId().compareTo(startLotInclusive) >= 0 && uri.lotId().compareTo(endLotExclusive) < 0)
                 .map(uri -> new ArcStatusRecord(record, uri.lotId(), uri.state()));
 
         if (fillGaps) {
             // there has to be a better way to zip together ordered streams and remove any dupes by a predicate or
             // bifunction
+            LOG.info("filling gaps for inclusive start: {}, exclusive end: {}", startLotInclusive, endLotExclusive);
             arcStatusRecordStream = StreamEx.of(LotStream.stream(startLotInclusive, endLotExclusive))
                     .map(lot -> new ArcStatusRecord(record, lot, null))
                     .append(arcStatusRecordStream)
                     .sortedBy(ArcStatusRecord::lotId)
                     .collapse(
                             (l, r) -> l.lotId().equals(r.lotId()),
                             (l, r) -> l.state() == null ? r : l
-                    );
+                    )
+                    .peek(arcStatusRecord -> {
+                        LOG.info("using arc record: {}", arcStatusRecord);
+                    });
         }
 
         Optional<Predicate<ArcState>> supplied = arcStateSupplier.get();

clusterless-substrate-aws-kernel/src/main/java/clusterless/cls/substrate/aws/report/scanner/ManifestScanner.java

@@ -25,6 +25,11 @@ public ManifestScanner(String profile, DatasetRecord datasetRecord, Moment earli
         super(profile, datasetRecord, earliest, latest, false);
     }
 
+    @Override
+    protected String scannerType() {
+        return "ManifestStatus";
+    }
+
     @Override
     protected StateURI<?, ?> createStateURIFrom(DatasetRecord record) {
         return ManifestURI.builder()
@@ -36,6 +41,7 @@ public ManifestScanner(String profile, DatasetRecord datasetRecord, Moment earli
     @NotNull
     protected Stream<DatasetStatusRecord> parseUriStreamIntoStatusRec(Stream<String> resultStream) {
         return resultStream.map(ManifestURI::parse)
+                .filter(uri -> uri.lotId().compareTo(startLotInclusive) >= 0 && uri.lotId().compareTo(endLotExclusive) < 0)
                 .map(uri -> new DatasetStatusRecord(record, uri));
     }
 

clusterless-substrate-aws-kernel/src/main/java/clusterless/cls/substrate/aws/report/scanner/Scanner.java

@@ -46,7 +46,7 @@ public Scanner(String profile, Rec record, Moment earliest, Moment latest, boole
         this.profile = profile;
         this.record = record;
         this.fillGaps = fillGaps;
-        LOG.info("creating scanner for: {}", record.display());
+        LOG.info("creating {} scanner for: {}", scannerType(), record.display());
 
         this.stateURI = createStateURIFrom(record);
         this.temporalUnit = findTemporalKeyFor(this.stateURI, earliest, latest);
@@ -67,6 +67,8 @@ public Scanner(String profile, Rec record, Moment earliest, Moment latest, boole
         LOG.info("using lot earliest: {}, latest: {}", startLotInclusive, endLotExclusive);
     }
 
+    protected abstract String scannerType();
+
     public String profile() {
         return profile;
     }