include space usage in diag for testing

adapter-diag/src/main/java/io/nosqlbench/adapter/diag/DiagDriverAdapter.java

@@ -51,7 +51,7 @@ public class DiagDriverAdapter extends BaseDriverAdapter<DiagOp, DiagSpace> impl
     @Override
     public synchronized OpMapper<DiagOp> getOpMapper() {
         if (this.mapper == null) {
-            this.mapper = new DiagOpMapper(this, getSpaceCache());
+            this.mapper = new DiagOpMapper(this);
         }
         return this.mapper;
     }

adapter-diag/src/main/java/io/nosqlbench/adapter/diag/DiagOp.java

@@ -28,9 +28,11 @@ public class DiagOp implements CycleOp<Integer> {
 
     private final static Logger logger = LogManager.getLogger(DiagOp.class);
     private final List<DiagTask> mutators;
+    private final DiagSpace space;
 
-    public DiagOp(List<DiagTask> mutators) {
+    public DiagOp(DiagSpace space, List<DiagTask> mutators) {
         this.mutators = mutators;
+        this.space = space;
     }
 
     @Override

adapter-diag/src/main/java/io/nosqlbench/adapter/diag/DiagOpDispenser.java

@@ -17,14 +17,13 @@
 package io.nosqlbench.adapter.diag;
 
 import io.nosqlbench.adapter.diag.optasks.DiagTask;
-import io.nosqlbench.engine.api.activityapi.ratelimits.RateLimiter;
-import io.nosqlbench.engine.api.activityimpl.BaseOpDispenser;
-import io.nosqlbench.engine.api.activityimpl.uniform.DriverAdapter;
-import io.nosqlbench.engine.api.templating.ParsedOp;
-import io.nosqlbench.nb.annotations.ServiceSelector;
 import io.nosqlbench.api.config.standard.NBConfigModel;
 import io.nosqlbench.api.config.standard.NBConfiguration;
 import io.nosqlbench.api.config.standard.NBReconfigurable;
+import io.nosqlbench.engine.api.activityapi.ratelimits.RateLimiter;
+import io.nosqlbench.engine.api.activityimpl.BaseOpDispenser;
+import io.nosqlbench.engine.api.templating.ParsedOp;
+import io.nosqlbench.nb.annotations.ServiceSelector;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
@@ -39,12 +38,12 @@ public class DiagOpDispenser extends BaseOpDispenser<DiagOp,DiagSpace> implement
     private LongFunction<DiagSpace> spaceF;
     private OpFunc opFuncs;
 
-    public DiagOpDispenser(DriverAdapter adapter, ParsedOp op) {
+    public DiagOpDispenser(DiagDriverAdapter adapter, LongFunction<DiagSpace> spaceF, ParsedOp op) {
         super(adapter,op);
-        this.opFunc = resolveOpFunc(op);
+        this.opFunc = resolveOpFunc(spaceF, op);
     }
 
-    private OpFunc resolveOpFunc(ParsedOp op) {
+    private OpFunc resolveOpFunc(LongFunction<DiagSpace> spaceF, ParsedOp op) {
         List<DiagTask> tasks = new ArrayList<>();
         Set<String> tasknames = op.getDefinedNames();
 
@@ -82,7 +81,7 @@ public class DiagOpDispenser extends BaseOpDispenser<DiagOp,DiagSpace> implement
             // Store the task into the diag op's list of things to do when it runs
             tasks.add(task);
         }
-        this.opFunc = new OpFunc(tasks);
+        this.opFunc = new OpFunc(spaceF,tasks);
         return opFunc;
     }
 
@@ -98,13 +97,17 @@ public class DiagOpDispenser extends BaseOpDispenser<DiagOp,DiagSpace> implement
 
     private final static class OpFunc implements LongFunction<DiagOp>, NBReconfigurable {
         private final List<DiagTask> tasks;
-        public OpFunc(List<DiagTask> tasks) {
+        private final LongFunction<DiagSpace> spaceF;
+
+        public OpFunc(LongFunction<DiagSpace> spaceF, List<DiagTask> tasks) {
             this.tasks = tasks;
+            this.spaceF = spaceF;
         }
 
         @Override
         public DiagOp apply(long value) {
-            return new DiagOp(tasks);
+            DiagSpace space = spaceF.apply(value);
+            return new DiagOp(space, tasks);
         }
 
         @Override

adapter-diag/src/main/java/io/nosqlbench/adapter/diag/DiagOpMapper.java

@@ -16,14 +16,12 @@
 
 package io.nosqlbench.adapter.diag;
 
-import io.nosqlbench.engine.api.activityimpl.OpDispenser;
-import io.nosqlbench.engine.api.activityimpl.OpMapper;
-import io.nosqlbench.engine.api.activityimpl.uniform.DriverAdapter;
-import io.nosqlbench.engine.api.activityimpl.uniform.DriverSpaceCache;
-import io.nosqlbench.engine.api.templating.ParsedOp;
 import io.nosqlbench.api.config.standard.NBConfigModel;
 import io.nosqlbench.api.config.standard.NBConfiguration;
 import io.nosqlbench.api.config.standard.NBReconfigurable;
+import io.nosqlbench.engine.api.activityimpl.OpDispenser;
+import io.nosqlbench.engine.api.activityimpl.OpMapper;
+import io.nosqlbench.engine.api.templating.ParsedOp;
 
 import java.util.ArrayList;
 import java.util.LinkedHashMap;
@@ -31,20 +29,17 @@ import java.util.Map;
 import java.util.function.LongFunction;
 
 public class DiagOpMapper implements OpMapper<DiagOp>, NBReconfigurable {
-    private final DriverSpaceCache<? extends DiagSpace> spaceCache;
     private final Map<String,DiagOpDispenser> dispensers = new LinkedHashMap<>();
-    private final DriverAdapter adapter;
+    private final DiagDriverAdapter adapter;
 
-    public DiagOpMapper(DriverAdapter adapter, DriverSpaceCache<? extends DiagSpace> spaceCache) {
+    public DiagOpMapper(DiagDriverAdapter adapter) {
-        this.spaceCache = spaceCache;
         this.adapter = adapter;
     }
 
     @Override
     public OpDispenser<? extends DiagOp> apply(ParsedOp op) {
-        DiagOpDispenser dispenser = new DiagOpDispenser(adapter,op);
+        LongFunction<DiagSpace> spaceF = adapter.getSpaceFunc(op);
-        LongFunction<String> spaceName = op.getAsFunctionOr("space", "default");
+        DiagOpDispenser dispenser = new DiagOpDispenser(adapter,spaceF,op);
-        LongFunction<DiagSpace> spacef = l -> spaceCache.get(spaceName.apply(l));
         dispensers.put(op.getName(),dispenser);
         return dispenser;
     }

adapter-diag/src/main/java/io/nosqlbench/adapter/diag/DiagSpace.java

@@ -26,13 +26,14 @@ import io.nosqlbench.api.config.standard.Param;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
-public class DiagSpace implements ActivityDefObserver {
+public class DiagSpace implements ActivityDefObserver, AutoCloseable {
     private final Logger logger = LogManager.getLogger(DiagSpace.class);
 
     private final NBConfiguration cfg;
     private final String name;
     private RateLimiter diagRateLimiter;
     private long interval;
+    private boolean errorOnClose;
 
     public DiagSpace(String name, NBConfiguration cfg) {
         this.cfg = cfg;
@@ -42,11 +43,13 @@ public class DiagSpace implements ActivityDefObserver {
 
     public void applyConfig(NBConfiguration cfg) {
         this.interval = cfg.get("interval",long.class);
+        this.errorOnClose = cfg.get("erroronclose",boolean.class);
     }
 
     public static NBConfigModel getConfigModel() {
         return ConfigModel.of(DiagSpace.class)
             .add(Param.defaultTo("interval",1000))
+            .add(Param.defaultTo("erroronclose", false))
             .asReadOnly();
     }
 
@@ -61,4 +64,12 @@ public class DiagSpace implements ActivityDefObserver {
         NBConfiguration cfg = getConfigModel().apply(activityDef.getParams().getStringStringMap());
         this.applyConfig(cfg);
     }
+
+    @Override
+    public void close() throws Exception {
+        logger.debug("closing diag space '" + this.name + "'");
+        if (errorOnClose) {
+            throw new RuntimeException("diag space was configured to throw this error when it was configured.");
+        }
+    }
 }

nbr/src/test/java/io/nosqlbench/cli/testing/ExitStatusIntegrationTests.java

@@ -19,7 +19,6 @@ package io.nosqlbench.cli.testing;
 import org.junit.jupiter.api.Test;
 
 import java.util.Optional;
-import java.util.stream.Collectors;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -38,7 +37,7 @@ public class ExitStatusIntegrationTests {
             "badparam"
         );
         assertThat(result.exception).isNull();
-        String stderr = result.getStderrData().stream().collect(Collectors.joining("\n"));
+        String stderr = String.join("\n", result.getStderrData());
         assertThat(stderr).contains("Scenario stopped due to error");
         assertThat(result.exitStatus).isEqualTo(2);
     }
@@ -52,7 +51,7 @@ public class ExitStatusIntegrationTests {
             "driver=diag", "op=initdelay:initdelay=notanumber"
         );
         assertThat(result.exception).isNull();
-        String stderr = result.getStdoutData().stream().collect(Collectors.joining("\n"));
+        String stderr = String.join("\n", result.getStdoutData());
         assertThat(stderr).contains("For input string: \"notanumber\"");
         assertThat(result.exitStatus).isEqualTo(2);
     }
@@ -77,15 +76,26 @@ public class ExitStatusIntegrationTests {
         ProcessInvoker invoker = new ProcessInvoker();
         invoker.setLogDir("logs/test");
         ProcessResult result = invoker.run("exitstatus_asyncstoprequest", 30,
-                java, "-jar", JARNAME, "--logs-dir", "logs/test/asyncstop", "run",
+                java, "-jar", JARNAME, "--logs-dir", "logs/test/op_exception", "run",
-            "driver=diag", "cyclerate=5", "op=erroroncycle:erroroncycle=10", "cycles=2000", "-vvv"
+            "driver=diag", "rate=5", "op=erroroncycle:erroroncycle=10", "cycles=2000", "-vvv"
         );
         assertThat(result.exception).isNull();
-        String stdout = result.getStdoutData().stream().collect(Collectors.joining("\n"));
+        String stdout = String.join("\n", result.getStdoutData());
         assertThat(stdout).contains("Diag was requested to stop on cycle 10");
         assertThat(result.exitStatus).isEqualTo(2);
     }
 
-
+    @Test
-
+    public void testCloseErrorHandlerOnSpace() {
+        ProcessInvoker invoker = new ProcessInvoker();
+        invoker.setLogDir("logs/test");
+        ProcessResult result = invoker.run("exitstatus_erroronclose", 30,
+            java, "-jar", JARNAME, "--logs-dir", "logs/test/error_on_close", "run",
+            "driver=diag", "rate=5", "op=noop", "cycles=10", "erroronclose=true", "-vvv"
+        );
+        String stdout = String.join("\n", result.getStdoutData());
+        String stderr = String.join("\n", result.getStderrData());
+        assertThat(result.exception).isNotNull();
+        assertThat(result.exception.getMessage()).contains("diag space was configured to throw");
+    }
 }