Enable advanced monitoring for SQLServer & Postgres.

.github/workflows/maven.yml

@@ -468,14 +468,14 @@ jobs:
           # In this case, we load the tpcc data.
           if [[ ${{matrix.benchmark}} == templated ]]; then
             java -jar benchbase.jar -b tpcc -c config/postgres/sample_tpcc_config.xml --create=true --load=true --execute=false --json-histograms results/histograms.json
-            java -jar benchbase.jar -b ${{matrix.benchmark}} -c config/postgres/sample_${{matrix.benchmark}}_config.xml --create=false --load=false --execute=true --json-histograms results/histograms.json
+            java -jar benchbase.jar -b ${{matrix.benchmark}} -c config/postgres/sample_${{matrix.benchmark}}_config.xml -im 1000 -mt advanced --create=false --load=false --execute=true --json-histograms results/histograms.json
           elif [[ ${{matrix.benchmark}} == tpcc-with-reconnects ]]; then
             # See Also: WITH_SERVICE_INTERRUPTIONS=true docker/build-run-benchmark-with-docker.sh
             java -jar benchbase.jar -b tpcc -c config/postgres/sample_tpcc_config.xml --create=true --load=true
             (sleep 10 && ./scripts/interrupt-docker-db-service.sh postgres) &
-            java -jar benchbase.jar -b tpcc -c config/postgres/sample_tpcc_config.xml --execute=true --json-histograms results/histograms.json
+            java -jar benchbase.jar -b tpcc -c config/postgres/sample_tpcc_config.xml -im 1000 -mt advanced --execute=true --json-histograms results/histograms.json
           else
-            java -jar benchbase.jar -b ${{matrix.benchmark}} -c config/postgres/sample_${{matrix.benchmark}}_config.xml --create=true --load=true --execute=true --json-histograms results/histograms.json
+            java -jar benchbase.jar -b ${{matrix.benchmark}} -c config/postgres/sample_${{matrix.benchmark}}_config.xml -im 1000 -mt advanced --create=true --load=true --execute=true --json-histograms results/histograms.json
           fi
 
           # FIXME: Reduce the error rate so we don't need these overrides.
@@ -491,6 +491,12 @@ jobs:
           ./scripts/check_latest_benchmark_results.sh $results_benchmark
           ./scripts/check_histogram_results.sh results/histograms.json $ERRORS_THRESHOLD
 
+          # Running the monitor should create at least three files in the 'monitor' directory.
+          if ![ $(find "./results/monitor" -maxdepth 1 -mindepth 1 | wc -l) -gt 2]; then
+            echo "ERROR: Advanced monitoring unsuccessful, file directory and/or appropriate files not created." >&2
+            exit 1
+          fi
+
       - name: Stop custom postgres service
         run: |
           ./docker/postgres-latest/down.sh
@@ -638,14 +644,14 @@ jobs:
           # In this case, we load the tpcc data.
           if [[ ${{matrix.benchmark}} == templated ]]; then
             java -jar benchbase.jar -b tpcc -c config/sqlserver/sample_tpcc_config.xml --create=true --load=true --execute=false --json-histograms results/histograms.json
-            java -jar benchbase.jar -b ${{matrix.benchmark}} -c config/sqlserver/sample_${{matrix.benchmark}}_config.xml --create=false --load=false --execute=true --json-histograms results/histograms.json
+            java -jar benchbase.jar -b ${{matrix.benchmark}} -c config/sqlserver/sample_${{matrix.benchmark}}_config.xml -im 1000 -mt advanced  --create=false --load=false --execute=true --json-histograms results/histograms.json
           elif [[ ${{matrix.benchmark}} == tpcc-with-reconnects ]]; then
             # See Also: WITH_SERVICE_INTERRUPTIONS=true docker/build-run-benchmark-with-docker.sh
             java -jar benchbase.jar -b tpcc -c config/sqlserver/sample_tpcc_config.xml --create=true --load=true
             (sleep 10 && ./scripts/interrupt-docker-db-service.sh sqlserver) &
-            java -jar benchbase.jar -b tpcc -c config/sqlserver/sample_tpcc_config.xml --execute=true --json-histograms results/histograms.json
+            java -jar benchbase.jar -b tpcc -c config/sqlserver/sample_tpcc_config.xml -im 1000 -mt advanced --execute=true --json-histograms results/histograms.json
           else
-            java -jar benchbase.jar -b ${{matrix.benchmark}} -c config/sqlserver/sample_${{matrix.benchmark}}_config.xml --create=true --load=true --execute=true --json-histograms results/histograms.json
+            java -jar benchbase.jar -b ${{matrix.benchmark}} -c config/sqlserver/sample_${{matrix.benchmark}}_config.xml -im 1000 -mt advanced --create=true --load=true --execute=true --json-histograms results/histograms.json
           fi
 
           # FIXME: Reduce the error rate so we don't need these overrides.
@@ -659,6 +665,12 @@ jobs:
           ./scripts/check_latest_benchmark_results.sh $results_benchmark
           ./scripts/check_histogram_results.sh results/histograms.json $ERRORS_THRESHOLD
 
+          # Running the monitor should create at least three files in the 'monitor' directory.
+          if ![ $(find "./results/monitor" -maxdepth 1 -mindepth 1 | wc -l) -gt 2]; then
+            echo "ERROR: Advanced monitoring unsuccessful, file directory and/or appropriate files not created." >&2
+            exit 1
+          fi
+
   ## ----------------------------------------------------------------------------------
   ## Docker Build Test Publish
   ## ----------------------------------------------------------------------------------

pom.xml

@@ -349,6 +349,13 @@
             <artifactId>janino</artifactId>
             <version>3.1.12</version>
         </dependency>
+        <dependency>
+            <groupId>org.immutables</groupId>
+            <artifactId>value</artifactId>
+            <version>2.9.0</version>
+            <scope>provided</scope>
+        </dependency>
+
     </dependencies>
 
     <build>

src/main/java/com/oltpbenchmark/DBWorkload.java

@@ -79,11 +79,27 @@ public static void main(String[] args) throws Exception {
       return;
     }
 
-    // Seconds
-    int intervalMonitor = 0;
+    // Monitoring setup.
+    ImmutableMonitorInfo.Builder builder = ImmutableMonitorInfo.builder();
     if (argsLine.hasOption("im")) {
-      intervalMonitor = Integer.parseInt(argsLine.getOptionValue("im"));
+      builder.monitoringInterval(Integer.parseInt(argsLine.getOptionValue("im")));
     }
+    if (argsLine.hasOption("mt")) {
+      switch (argsLine.getOptionValue("mt")) {
+        case "advanced":
+          builder.monitoringType(MonitorInfo.MonitoringType.ADVANCED);
+          break;
+        case "throughput":
+          builder.monitoringType(MonitorInfo.MonitoringType.THROUGHPUT);
+          break;
+        default:
+          throw new ParseException(
+              "Monitoring type '"
+                  + argsLine.getOptionValue("mt")
+                  + "' is undefined, allowed values are: advanced/throughput");
+      }
+    }
+    MonitorInfo monitorInfo = builder.build();
 
     // -------------------------------------------------------------------
     // GET PLUGIN LIST
@@ -151,6 +167,14 @@ public static void main(String[] args) throws Exception {
         // Nothing to do here !
       }
 
+      // Set monitoring enabled, if all requirements are met.
+      if (monitorInfo.getMonitoringInterval() > 0
+          && monitorInfo.getMonitoringType() == MonitorInfo.MonitoringType.ADVANCED
+          && DatabaseType.get(xmlConfig.getString("type")).shouldCreateMonitoringPrefix()) {
+        LOG.info("Advanced monitoring enabled, prefix will be added to queries.");
+        wrkld.setAdvancedMonitoringEnabled(true);
+      }
+
       // ----------------------------------------------------------------
       // CREATE BENCHMARK MODULE
       // ----------------------------------------------------------------
@@ -518,7 +542,7 @@ public static void main(String[] args) throws Exception {
     if (isBooleanOptionSet(argsLine, "execute")) {
       // Bombs away!
       try {
-        Results r = runWorkload(benchList, intervalMonitor);
+        Results r = runWorkload(benchList, monitorInfo);
         writeOutputs(r, activeTXTypes, argsLine, xmlConfig);
         writeHistograms(r);
 
@@ -558,8 +582,8 @@ private static Options buildOptions(XMLConfiguration pluginConfig) {
     options.addOption(null, "execute", true, "Execute the benchmark workload");
     options.addOption("h", "help", false, "Print this help");
     options.addOption("s", "sample", true, "Sampling window");
-    options.addOption(
-        "im", "interval-monitor", true, "Throughput Monitoring Interval in milliseconds");
+    options.addOption("im", "interval-monitor", true, "Monitoring Interval in milliseconds");
+    options.addOption("mt", "monitor-type", true, "Type of Monitoring (throughput/advanced)");
     options.addOption(
         "d",
         "directory",
@@ -733,7 +757,7 @@ private static void runLoader(BenchmarkModule bench)
     bench.loadDatabase();
   }
 
-  private static Results runWorkload(List<BenchmarkModule> benchList, int intervalMonitor)
+  private static Results runWorkload(List<BenchmarkModule> benchList, MonitorInfo monitorInfo)
       throws IOException {
     List<Worker<?>> workers = new ArrayList<>();
     List<WorkloadConfiguration> workConfs = new ArrayList<>();
@@ -748,7 +772,7 @@ private static Results runWorkload(List<BenchmarkModule> benchList, int interval
               bench.getBenchmarkName().toUpperCase(), num_phases, (num_phases > 1 ? "s" : "")));
       workConfs.add(bench.getWorkloadConfiguration());
     }
-    Results r = ThreadBench.runRateLimitedBenchmark(workers, workConfs, intervalMonitor);
+    Results r = ThreadBench.runRateLimitedBenchmark(workers, workConfs, monitorInfo);
     LOG.info(SINGLE_LINE);
     LOG.info("Rate limited reqs/s: {}", r);
     return r;

src/main/java/com/oltpbenchmark/ThreadBench.java

@@ -1,17 +1,15 @@
 /*
  * Copyright 2020 by OLTPBenchmark Project
  *
- * 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
+ * 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
+ * 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.
+ * 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.
  *
  */
 
@@ -21,7 +19,10 @@
 import com.oltpbenchmark.api.BenchmarkModule;
 import com.oltpbenchmark.api.TransactionType;
 import com.oltpbenchmark.api.Worker;
+import com.oltpbenchmark.api.collectors.monitoring.Monitor;
+import com.oltpbenchmark.api.collectors.monitoring.MonitorGen;
 import com.oltpbenchmark.types.State;
+import com.oltpbenchmark.util.MonitorInfo;
 import com.oltpbenchmark.util.StringUtil;
 import java.util.*;
 import org.apache.commons.collections4.map.ListOrderedMap;
@@ -30,30 +31,35 @@
 
 public class ThreadBench implements Thread.UncaughtExceptionHandler {
   private static final Logger LOG = LoggerFactory.getLogger(ThreadBench.class);
+  // Determines how long (in ms) to wait until monitoring thread rejoins the
+  // main thread.
+  private static final int MONITOR_REJOIN_TIME = 60000;
 
   private final BenchmarkState testState;
   private final List<? extends Worker<? extends BenchmarkModule>> workers;
   private final ArrayList<Thread> workerThreads;
   private final List<WorkloadConfiguration> workConfs;
   private final ArrayList<LatencyRecord.Sample> samples = new ArrayList<>();
-  private final int intervalMonitor;
+  private final MonitorInfo monitorInfo;
+
+  private Monitor monitor = null;
 
   private ThreadBench(
       List<? extends Worker<? extends BenchmarkModule>> workers,
       List<WorkloadConfiguration> workConfs,
-      int intervalMonitoring) {
+      MonitorInfo monitorInfo) {
     this.workers = workers;
     this.workConfs = workConfs;
     this.workerThreads = new ArrayList<>(workers.size());
-    this.intervalMonitor = intervalMonitoring;
+    this.monitorInfo = monitorInfo;
     this.testState = new BenchmarkState(workers.size() + 1);
   }
 
   public static Results runRateLimitedBenchmark(
       List<Worker<? extends BenchmarkModule>> workers,
       List<WorkloadConfiguration> workConfs,
-      int intervalMonitoring) {
-    ThreadBench bench = new ThreadBench(workers, workConfs, intervalMonitoring);
+      MonitorInfo monitorInfo) {
+    ThreadBench bench = new ThreadBench(workers, workConfs, monitorInfo);
     return bench.runRateLimitedMultiPhase();
   }
 
@@ -88,10 +94,9 @@ private int finalizeWorkers(ArrayList<Thread> workerThreads) throws InterruptedE
       // to terminate... hands otherwise
 
       /*
-       * // CARLO: Maybe we might want to do this to kill threads that are
-       * hanging... if (workerThreads.get(i).isAlive()) {
-       * workerThreads.get(i).kill(); try { workerThreads.get(i).join(); }
-       * catch (InterruptedException e) { } }
+       * // CARLO: Maybe we might want to do this to kill threads that are hanging... if
+       * (workerThreads.get(i).isAlive()) { workerThreads.get(i).kill(); try {
+       * workerThreads.get(i).join(); } catch (InterruptedException e) { } }
        */
 
       requests += workers.get(i).getRequests();
@@ -116,17 +121,11 @@ private Results runRateLimitedMultiPhase() {
     this.createWorkerThreads();
 
     // long measureStart = start;
+    Phase phase = null;
 
-    long startTs = System.currentTimeMillis();
-    long start = System.nanoTime();
-    long warmupStart = System.nanoTime();
-    long warmup = warmupStart;
-    long measureEnd = -1;
     // used to determine the longest sleep interval
     double lowestRate = Double.MAX_VALUE;
 
-    Phase phase = null;
-
     for (WorkloadState workState : workStates) {
       workState.switchToNextPhase();
       phase = workState.getCurrentPhase();
@@ -145,6 +144,12 @@ private Results runRateLimitedMultiPhase() {
       }
     }
 
+    long startTs = System.currentTimeMillis();
+    long start = System.nanoTime();
+    long warmupStart = System.nanoTime();
+    long warmup = warmupStart;
+    long measureEnd = -1;
+
     long intervalNs = getInterval(lowestRate, phase.getArrival());
 
     long nextInterval = start + intervalNs;
@@ -157,8 +162,11 @@ private Results runRateLimitedMultiPhase() {
     boolean lastEntry = false;
 
     // Initialize the Monitor
-    if (this.intervalMonitor > 0) {
-      new MonitorThread(this.intervalMonitor).start();
+    if (this.monitorInfo.getMonitoringInterval() > 0) {
+      this.monitor =
+          MonitorGen.getMonitor(
+              this.monitorInfo, this.testState, this.workers, this.workConfs.get(0));
+      this.monitor.start();
     }
 
     // Allow workers to start work.
@@ -301,6 +309,18 @@ private Results runRateLimitedMultiPhase() {
       }
     }
 
+    // Stop the monitoring thread separately from cleanup all the workers so we can ignore errors
+    // from these threads (including possible SQLExceptions), but not the others.
+    try {
+      if (this.monitor != null) {
+        this.monitor.interrupt();
+        this.monitor.join(MONITOR_REJOIN_TIME);
+        this.monitor.tearDown();
+      }
+    } catch (Exception e) {
+      LOG.error(e.getMessage(), e);
+    }
+
     try {
       int requests = finalizeWorkers(this.workerThreads);
 
@@ -528,42 +548,4 @@ public void run() {
       }
     }
   }
-
-  private class MonitorThread extends Thread {
-    private final int intervalMonitor;
-
-    {
-      this.setDaemon(true);
-    }
-
-    /**
-     * @param interval How long to wait between polling in milliseconds
-     */
-    MonitorThread(int interval) {
-      this.intervalMonitor = interval;
-    }
-
-    @Override
-    public void run() {
-      LOG.info("Starting MonitorThread Interval [{}ms]", this.intervalMonitor);
-      while (true) {
-        try {
-          Thread.sleep(this.intervalMonitor);
-        } catch (InterruptedException ex) {
-          return;
-        }
-
-        // Compute the last throughput
-        long measuredRequests = 0;
-        synchronized (testState) {
-          for (Worker<?> w : workers) {
-            measuredRequests += w.getAndResetIntervalRequests();
-          }
-        }
-        double seconds = this.intervalMonitor / 1000d;
-        double tps = (double) measuredRequests / seconds;
-        LOG.info("Throughput: {} txn/sec", tps);
-      }
-    }
-  }
 }