provide typed update interface for dynamic params

adapters-api/src/main/java/io/nosqlbench/engine/api/activityimpl/uniform/BaseDriverAdapter.java

@@ -27,14 +27,11 @@ import java.util.Optional;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
-public abstract class BaseDriverAdapter<R extends Op,S> implements DriverAdapter<R,S>, NBConfigurable {
+public abstract class BaseDriverAdapter<R extends Op,S> implements DriverAdapter<R,S>, NBConfigurable, NBReconfigurable {
 
     private DriverSpaceCache<? extends S> spaceCache;
     private NBConfiguration cfg;
 
-    protected BaseDriverAdapter() {
-    }
-
     /**
      * BaseDriverAdapter will take any provided functions from {@link #getOpStmtRemappers()}
      * and {@link #getOpFieldRemappers()} and construct a preprocessor list. These are applied
@@ -72,12 +69,34 @@ public abstract class BaseDriverAdapter<R extends Op,S> implements DriverAdapter
      * in the op template. These are useful, for example, for taking the 'stmt' field
      * and parsing it into component fields which might otherwise be specified by the user.
      * This allows users to specify String-form op templates which are automatically
-     * destructured into the canonical field-wise form for a given type of operation.</p>
+     * parsed and destructured into the canonical field-wise form for a given type of
+     * operation.</p>
+     *
      * <br/>
+     *
      * <p>Each function in this list is applied in order. If the function returns a value,
      * then the 'stmt' field is removed and the resulting map is added to the other
      * fields in the op template.</p>
      *
+     * <br/>
+     *
+     * <p>If a driver adapter is meant to support the {@code stmt} field, then this
+     * <em>must</em> be provided. The use of the stmt field should be documented in
+     * the driver adapter user docs with examples for any supported formats. A default
+     * implementation does nothing, as it must be decided per-driver whether or not
+     * the stmt field will be used directly or whether it is short-hand for a more
+     * canonical form.
+     *
+     * <br/>
+     *
+     * <p>If you want to automatically destructure stmt values into a map and inject
+     * its entries into the op template fields, simply provide an implementation
+     * like this:<pre>
+     * {@code
+     *     return List.of(stmt -> Optional.of(NBParams.one(stmt).getMap()));
+     * }
+     * </pre></p>
+     *
      * @return A list of optionally applied remapping functions.
      */
     public List<Function<String, Optional<Map<String,Object>>>> getOpStmtRemappers() {
@@ -112,6 +131,12 @@ public abstract class BaseDriverAdapter<R extends Op,S> implements DriverAdapter
         this.cfg = cfg;
     }
 
+    @Override
+    public void applyReconfig(NBConfiguration reconf) {
+        this.cfg = getReconfigModel().apply(reconf.getMap());
+    }
+
+
     /**
      * In order to be provided with config information, it is required
      * that the driver adapter specify the valid configuration options,
@@ -123,7 +148,6 @@ public abstract class BaseDriverAdapter<R extends Op,S> implements DriverAdapter
             .add(Param.optional("alias"))
             .add(Param.defaultTo("strict",true,"strict op field mode, which requires that provided op fields are recognized and used"))
             .add(Param.optional(List.of("op", "stmt", "statement"), String.class, "op template in statement form"))
-            .add(Param.optional(List.of("workload", "yaml"), String.class, "location of workload yaml file"))
             .add(Param.optional("tags", String.class, "tags to be used to filter operations"))
             .add(Param.defaultTo("errors", "stop", "error handler configuration"))
             .add(Param.optional("threads").setRegex("\\d+|\\d+x|auto").setDescription("number of concurrent operations, controlled by threadpool"))
@@ -134,6 +158,16 @@ public abstract class BaseDriverAdapter<R extends Op,S> implements DriverAdapter
             .add(Param.optional("phaserate", String.class, "rate limit for phases per second"))
             .add(Param.optional("seq", String.class, "sequencing algorithm"))
             .add(Param.optional("instrument", Boolean.class))
+            .add(Param.optional(List.of("workload", "yaml"), String.class, "location of workload yaml file"))
+            .asReadOnly();
+    }
+
+    @Override
+    public NBConfigModel getReconfigModel() {
+        return ConfigModel.of(BaseDriverAdapter.class)
+            .add(Param.optional("threads").setRegex("\\d+|\\d+x|auto").setDescription("number of concurrent operations, controlled by threadpool"))
+            .add(Param.optional("striderate", String.class, "rate limit for strides per second"))
+            .add(Param.optional(List.of("cyclerate", "targetrate", "rate"), String.class, "rate limit for cycles per second"))
             .asReadOnly();
     }
 

engine-api/src/main/java/io/nosqlbench/engine/api/activityimpl/SimpleActivity.java

@@ -19,13 +19,11 @@ package io.nosqlbench.engine.api.activityimpl;
 import com.codahale.metrics.Timer;
 import io.nosqlbench.engine.api.activityapi.core.*;
 import io.nosqlbench.engine.api.activityapi.core.progress.ActivityMetricProgressMeter;
-import io.nosqlbench.engine.api.activityapi.core.progress.InputProgressMeter;
 import io.nosqlbench.engine.api.activityapi.core.progress.ProgressCapable;
 import io.nosqlbench.engine.api.activityapi.core.progress.ProgressMeter;
 import io.nosqlbench.engine.api.activityapi.cyclelog.filters.IntPredicateDispenser;
 import io.nosqlbench.engine.api.activityapi.errorhandling.ErrorMetrics;
 import io.nosqlbench.engine.api.activityapi.errorhandling.modular.NBErrorHandler;
-import io.nosqlbench.engine.api.activityapi.input.Input;
 import io.nosqlbench.engine.api.activityapi.input.InputDispenser;
 import io.nosqlbench.engine.api.activityapi.output.OutputDispenser;
 import io.nosqlbench.engine.api.activityapi.planning.OpSequence;
@@ -102,7 +100,7 @@ public class SimpleActivity implements Activity, ProgressCapable {
 
     @Override
     public void initActivity() {
-        onActivityDefUpdate(this.activityDef);
+        initOrUpdateRateLimiters(this.activityDef);
     }
 
     public synchronized NBErrorHandler getErrorHandler() {
@@ -268,7 +266,7 @@ public class SimpleActivity implements Activity, ProgressCapable {
 
     @Override
     public Timer getResultTimer() {
-        return ActivityMetrics.timer(getActivityDef(), "result");
+        return ActivityMetrics.timer(getActivityDef(), "result", getParams().getOptionalInteger("hdr_digits").orElse(4));
     }
 
     @Override
@@ -320,6 +318,10 @@ public class SimpleActivity implements Activity, ProgressCapable {
 
     @Override
     public synchronized void onActivityDefUpdate(ActivityDef activityDef) {
+        initOrUpdateRateLimiters(activityDef);
+    }
+
+    public synchronized void initOrUpdateRateLimiters(ActivityDef activityDef) {
 
         activityDef.getParams().getOptionalNamedParameter("striderate")
             .map(RateSpec::new)
@@ -347,14 +349,16 @@ public class SimpleActivity implements Activity, ProgressCapable {
         if (strideOpt.isEmpty()) {
             String stride = String.valueOf(seq.getSequence().length);
             logger.info("defaulting stride to " + stride + " (the sequence length)");
-            getParams().set("stride", stride);
+//            getParams().set("stride", stride);
+            getParams().setSilently("stride",stride);
         }
 
         Optional<String> cyclesOpt = getParams().getOptionalString("cycles");
         if (cyclesOpt.isEmpty()) {
             String cycles = getParams().getOptionalString("stride").orElseThrow();
             logger.info("defaulting cycles to " + cycles + " (the stride length)");
-            getParams().set("cycles", getParams().getOptionalString("stride").orElseThrow());
+//            getParams().set("cycles", getParams().getOptionalString("stride").orElseThrow());
+            getParams().setSilently("cycles", getParams().getOptionalString("stride").orElseThrow());
         } else {
             if (getActivityDef().getCycleCount() == 0) {
                 throw new RuntimeException(
@@ -391,15 +395,18 @@ public class SimpleActivity implements Activity, ProgressCapable {
                 } else {
                     logger.info("setting threads to " + threads + " (auto) [10xCORES]");
                 }
-                activityDef.setThreads(threads);
+//                activityDef.setThreads(threads);
+                activityDef.getParams().setSilently("threads",threads);
             } else if (spec.toLowerCase().matches("\\d+x")) {
                 String multiplier = spec.substring(0, spec.length() - 1);
                 int threads = processors * Integer.parseInt(multiplier);
                 logger.info("setting threads to " + threads + " (" + multiplier + "x)");
-                activityDef.setThreads(threads);
+//                activityDef.setThreads(threads);
+                activityDef.getParams().setSilently("threads",threads);
             } else if (spec.toLowerCase().matches("\\d+")) {
                 logger.info("setting threads to " + spec + " (direct)");
-                activityDef.setThreads(Integer.parseInt(spec));
+//                activityDef.setThreads(Integer.parseInt(spec));
+                activityDef.getParams().setSilently("threads",Integer.parseInt(spec));
             }
 
             if (activityDef.getThreads() > activityDef.getCycleCount()) {
@@ -563,4 +570,8 @@ public class SimpleActivity implements Activity, ProgressCapable {
     }
 
 
+    @Override
+    public String getName() {
+        return this.activityDef.getAlias();
+    }
 }

engine-api/src/main/java/io/nosqlbench/engine/api/activityimpl/uniform/StandardActivity.java

@@ -17,15 +17,22 @@
 package io.nosqlbench.engine.api.activityimpl.uniform;
 
 import io.nosqlbench.engine.api.activityapi.planning.OpSequence;
+import io.nosqlbench.engine.api.activityconfig.StatementsLoader;
+import io.nosqlbench.engine.api.activityconfig.yaml.StmtsDocList;
 import io.nosqlbench.engine.api.activityimpl.ActivityDef;
 import io.nosqlbench.engine.api.activityimpl.OpDispenser;
 import io.nosqlbench.engine.api.activityimpl.OpMapper;
 import io.nosqlbench.engine.api.activityimpl.SimpleActivity;
 import io.nosqlbench.engine.api.activityimpl.uniform.flowtypes.Op;
+import io.nosqlbench.nb.api.config.standard.*;
 import io.nosqlbench.nb.api.errors.OpConfigError;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.function.Function;
 
 /**
@@ -37,14 +44,35 @@ import java.util.function.Function;
  * @param <S> The context type for the activity, AKA the 'space' for a named driver instance and its associated object graph
  */
 public class StandardActivity<R extends Op, S> extends SimpleActivity {
+    private final static Logger logger = LogManager.getLogger("ACTIVITY");
 
     private final DriverAdapter<R, S> adapter;
     private final OpSequence<OpDispenser<? extends R>> sequence;
+    private final NBConfigModel yamlmodel;
 
     public StandardActivity(DriverAdapter<R, S> adapter, ActivityDef activityDef) {
         super(activityDef);
         this.adapter = adapter;
 
+        if (adapter instanceof NBConfigurable configurable) {
+            NBConfigModel cmodel = configurable.getConfigModel();
+            Optional<String> yaml_loc = activityDef.getParams().getOptionalString("yaml", "workload");
+            if (yaml_loc.isPresent()) {
+                Map<String,Object> disposable = new LinkedHashMap<>(activityDef.getParams());
+                StmtsDocList workload = StatementsLoader.loadPath(logger, yaml_loc.get(), disposable, "activities");
+                yamlmodel = workload.getConfigModel();
+            }
+            else {
+                yamlmodel= ConfigModel.of(StandardActivity.class).asReadOnly();
+            }
+            NBConfigModel combinedModel = cmodel.add(yamlmodel);
+            NBConfiguration configuration = combinedModel.apply(activityDef.getParams());
+            configurable.applyConfig(configuration);
+        }
+        else {
+            yamlmodel= ConfigModel.of(StandardActivity.class).asReadOnly();
+        }
+
         try {
             OpMapper<R> opmapper = adapter.getOpMapper();
             Function<Map<String, Object>, Map<String, Object>> preprocessor = adapter.getPreprocessor();
@@ -81,4 +109,16 @@ public class StandardActivity<R extends Op, S> extends SimpleActivity {
         return adapter.getErrorNameMapper();
     }
 
+    @Override
+    public synchronized void onActivityDefUpdate(ActivityDef activityDef) {
+        super.onActivityDefUpdate(activityDef);
+
+        if (adapter instanceof NBReconfigurable configurable) {
+            NBConfigModel cfgModel = configurable.getReconfigModel();
+            NBConfiguration cfg = cfgModel.matchConfig(activityDef.getParams());
+            NBReconfigurable.applyMatching(cfg,List.of(configurable));
+        }
+//
+//        ActivityDefObserver.apply(activityDef, adapter, sequence);
+    }
 }

engine-api/src/main/java/io/nosqlbench/engine/api/activityimpl/uniform/StandardActivityType.java

@@ -18,11 +18,24 @@ package io.nosqlbench.engine.api.activityimpl.uniform;
 
 import io.nosqlbench.engine.api.activityapi.core.ActionDispenser;
 import io.nosqlbench.engine.api.activityapi.core.ActivityType;
+import io.nosqlbench.engine.api.activityconfig.StatementsLoader;
+import io.nosqlbench.engine.api.activityconfig.yaml.StmtsDocList;
 import io.nosqlbench.engine.api.activityimpl.ActivityDef;
 import io.nosqlbench.engine.api.activityimpl.SimpleActivity;
+import io.nosqlbench.nb.api.config.standard.NBConfigModel;
+import io.nosqlbench.nb.api.config.standard.NBConfiguration;
+import io.nosqlbench.nb.api.config.standard.NBReconfigurable;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Optional;
 
 public class StandardActivityType<A extends StandardActivity<?,?>> extends SimpleActivity implements ActivityType<A> {
 
+    private final static Logger logger = LogManager.getLogger("ACTIVITY");
+
     private final DriverAdapter<?,?> adapter;
 
     public StandardActivityType(DriverAdapter<?,?> adapter, ActivityDef activityDef) {
@@ -42,6 +55,24 @@ public class StandardActivityType<A extends StandardActivity<?,?>> extends Simpl
         return (A) new StandardActivity(adapter,activityDef);
     }
 
+    @Override
+    public synchronized void onActivityDefUpdate(ActivityDef activityDef) {
+        super.onActivityDefUpdate(activityDef);
+
+        if (adapter instanceof NBReconfigurable reconfigurable) {
+            NBConfigModel cfgModel = reconfigurable.getReconfigModel();
+            Optional<String> op_yaml_loc = activityDef.getParams().getOptionalString("yaml", "workload");
+            if (op_yaml_loc.isPresent()) {
+                Map<String,Object> disposable = new LinkedHashMap<>(activityDef.getParams());
+                StmtsDocList workload = StatementsLoader.loadPath(logger, op_yaml_loc.get(), disposable, "activities");
+                cfgModel=cfgModel.add(workload.getConfigModel());
+            }
+            NBConfiguration cfg = cfgModel.apply(activityDef.getParams());
+            reconfigurable.applyReconfig(cfg);
+        }
+
+    }
+
     @Override
     public ActionDispenser getActionDispenser(A activity) {
         return new StandardActionDispenser(activity);

engine-core/src/main/java/io/nosqlbench/engine/core/lifecycle/ActivityTypeLoader.java

@@ -17,16 +17,11 @@
 package io.nosqlbench.engine.core.lifecycle;
 
 import io.nosqlbench.engine.api.activityapi.core.ActivityType;
-import io.nosqlbench.engine.api.activityconfig.StatementsLoader;
-import io.nosqlbench.engine.api.activityconfig.yaml.StmtsDocList;
 import io.nosqlbench.engine.api.activityimpl.ActivityDef;
 import io.nosqlbench.engine.api.activityimpl.uniform.DriverAdapter;
 import io.nosqlbench.engine.api.activityimpl.uniform.StandardActivityType;
 import io.nosqlbench.nb.annotations.Maturity;
 import io.nosqlbench.nb.api.NBEnvironment;
-import io.nosqlbench.nb.api.config.standard.NBConfigModel;
-import io.nosqlbench.nb.api.config.standard.NBConfigurable;
-import io.nosqlbench.nb.api.config.standard.NBConfiguration;
 import io.nosqlbench.nb.api.content.Content;
 import io.nosqlbench.nb.api.content.NBIO;
 import io.nosqlbench.nb.api.errors.BasicError;
@@ -156,17 +151,17 @@ public class ActivityTypeLoader {
             DriverAdapter<?, ?> driverAdapter = oda.get();
 
             activityDef.getParams().remove("driver");
-            if (driverAdapter instanceof NBConfigurable) {
+//            if (driverAdapter instanceof NBConfigurable) {
-                NBConfigModel cfgModel = ((NBConfigurable) driverAdapter).getConfigModel();
+//                NBConfigModel cfgModel = ((NBConfigurable) driverAdapter).getConfigModel();
-                Optional<String> op_yaml_loc = activityDef.getParams().getOptionalString("yaml", "workload");
+//                Optional<String> op_yaml_loc = activityDef.getParams().getOptionalString("yaml", "workload");
-                if (op_yaml_loc.isPresent()) {
+//                if (op_yaml_loc.isPresent()) {
-                    Map<String,Object> disposable = new LinkedHashMap<>(activityDef.getParams());
+//                    Map<String,Object> disposable = new LinkedHashMap<>(activityDef.getParams());
-                    StmtsDocList workload = StatementsLoader.loadPath(logger, op_yaml_loc.get(), disposable, "activities");
+//                    StmtsDocList workload = StatementsLoader.loadPath(logger, op_yaml_loc.get(), disposable, "activities");
-                    cfgModel=cfgModel.add(workload.getConfigModel());
+//                    cfgModel=cfgModel.add(workload.getConfigModel());
-                }
+//                }
-                NBConfiguration cfg = cfgModel.apply(activityDef.getParams());
+//                NBConfiguration cfg = cfgModel.apply(activityDef.getParams());
-                ((NBConfigurable) driverAdapter).applyConfig(cfg);
+//                ((NBConfigurable) driverAdapter).applyConfig(cfg);
-            }
+//            }
             ActivityType activityType = new StandardActivityType<>(driverAdapter, activityDef);
             return Optional.of(activityType);
         } else {

nb-api/src/main/java/io/nosqlbench/nb/api/config/standard/NBReconfigModelProvider.java

@@ -14,15 +14,8 @@
  * limitations under the License.
  */
 
-activitydef1 = {
+package io.nosqlbench.nb.api.config.standard;
-    "alias" : "activity_to_await",
-    "type" : "diag",
-    "cycles" : "0..1500",
-    "threads" : "1",
-    "targetrate" : "500"
-};
 
-print('starting activity teststartstopdiag');
+public interface NBReconfigModelProvider {
-scenario.start(activitydef1);
+    NBConfigModel getReconfigModel();
-scenario.awaitActivity("activity_to_await");
+}
-print("awaited activity");

nb-api/src/main/java/io/nosqlbench/nb/api/config/standard/NBReconfigurable.java

@@ -0,0 +1,88 @@
+/*
+ * Copyright (c) 2022 nosqlbench
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.nosqlbench.nb.api.config.standard;
+
+import java.util.Collection;
+
+/**
+ * All implementation types which wish to have a type-marshalled configuration
+ * should implement this interface IFF they wish to support follow-on configuration
+ * after initialization. This is distinct and separate from initial configuration
+ * via {@link NBConfigurable}. A type may be NBReconfigurable without implementing
+ * the NBConfigurable interface, given that initialization for a type may happen
+ * via constructor or other means.
+ *
+ * When a type which implements this interface is instantiated, and the
+ * {@link NBConfiguration} was not injected into its constructor,
+ * the builder should call
+ * {@link NBConfigurable#applyConfig(NBConfiguration)} immediately
+ * after calling the constructor.
+ *
+ * Subsequently, when an owning instance has a configuration update to provide to
+ * the original NBConfigurable which <EM>ALSO</EM> implements NBReconfigurable, then
+ * {@link NBReconfigurable#applyReconfig(NBConfiguration)} should be called.
+ * The helper methods {@link #collectModels(Class, Collection)} and
+ * {@link #applyMatching(NBConfiguration, Object...)} can be used to apply
+ * reconfigurations to groups of elements with a shared configuration model.
+ */
+public interface NBReconfigurable extends NBCanReconfigure, NBReconfigModelProvider {
+
+    /**
+     * This applies a configuration to an element <EM>AFTER</EM> the initial
+     * configuration from {@link NBConfigurable}.
+     * @param recfg The configuration data to be applied to a new instance
+     */
+    @Override
+    void applyReconfig(NBConfiguration recfg);
+
+    /**
+     * Implement this method by returning an instance of {@link ConfigModel}.
+     * Any configuration which is provided to the {@link #applyReconfig(NBConfiguration)}
+     * method will be validated through this model. A configuration model
+     * is <em>required</em> in order to build a validated configuration
+     * from source data provided by a user.
+     * @return A valid configuration model for the implementing class
+     */
+    @Override
+    NBConfigModel getReconfigModel();
+
+    /**
+     * Convenience method to apply a configuration to any object which
+     * is expected to be be configurable.
+     * @param cfg The cfg to apply
+     * @param configurables zero or more Objects which may implement NBConfigurable
+     */
+    static void applyMatching(NBConfiguration cfg, Collection<?> configurables) {
+        for (Object configurable : configurables) {
+            if (configurable instanceof NBReconfigurable c) {
+                NBConfiguration partial = c.getReconfigModel().matchConfig(cfg);
+                c.applyReconfig(partial);
+            }
+        }
+    }
+
+    static NBConfigModel collectModels(Class<?> of, Collection<?> configurables) {
+        ConfigModel model = ConfigModel.of(of);
+        for (Object configurable : configurables) {
+            if (configurable instanceof NBReconfigurable c) {
+                model = model.add(c.getReconfigModel());
+            }
+        }
+        return model.asReadOnly();
+    }
+
+}