BENCHMARK: Add Workercount and Batch size as configurable settings in…

… benchmark

Fixes elastic#8217

tools/benchmark-cli/README.md

@@ -31,6 +31,10 @@ Option                           Description
 --local-path <String>            Path to the root of a local Logstash           
                                    distribution.                                
                                   E.g. `/opt/logstash`                          
+--ls-batch-size <Integer>        Logstash batch size (-b argument) to           
+                                   configure. (default: 128)                    
+--ls-workers <Integer>           Number of Logstash worker threads (-w          
+                                   argument) to configure. (default: 2)         
 --repeat-data <Integer>          Sets how often the test's dataset should be    
                                    run. (default: 1)                            
 --testcase <String>              Currently available test cases are 'baseline'  

tools/benchmark-cli/src/main/java/org/logstash/benchmark/cli/BenchmarkMeta.java

@@ -0,0 +1,69 @@
+package org.logstash.benchmark.cli;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.commons.lang3.SystemUtils;
+import org.logstash.benchmark.cli.ui.LsVersionType;
+
+public final class BenchmarkMeta {
+
+    private final String testcase;
+
+    private final String version;
+
+    private final LsVersionType vtype;
+
+    private final int workers;
+
+    private final int batchsize;
+
+    BenchmarkMeta(final String testcase, final String version, final LsVersionType vtype,
+        final int workers, final int batchsize) {
+        this.testcase = testcase;
+        this.version = version;
+        this.vtype = vtype;
+        this.workers = workers;
+        this.batchsize = batchsize;
+    }
+
+    public String getVersion() {
+        return version;
+    }
+
+    public String getTestcase() {
+        return testcase;
+    }
+
+    public LsVersionType getVtype() {
+        return vtype;
+    }
+
+    public int getWorkers() {
+        return workers;
+    }
+
+    public int getBatchsize() {
+        return batchsize;
+    }
+
+    public Map<String, Object> asMap() {
+        final Map<String, Object> result = new HashMap<>();
+        result.put("test_name", testcase);
+        result.put("os_name", SystemUtils.OS_NAME);
+        result.put("os_version", SystemUtils.OS_VERSION);
+        try {
+            result.put("host_name", InetAddress.getLocalHost().getHostName());
+        } catch (final UnknownHostException ex) {
+            throw new IllegalStateException(ex);
+        }
+        result.put("cpu_cores", Runtime.getRuntime().availableProcessors());
+        result.put("version_type", vtype);
+        result.put("version", version);
+        result.put("batch_size", batchsize);
+        result.put("worker_threads", workers);
+        return Collections.unmodifiableMap(result);
+    }
+}

tools/benchmark-cli/src/main/java/org/logstash/benchmark/cli/LogstashInstallation.java

@@ -9,6 +9,7 @@
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
 import java.nio.file.StandardOpenOption;
 import java.security.NoSuchAlgorithmException;
 import java.util.Map;
@@ -43,6 +44,13 @@ public interface LogstashInstallation {
      */
     String metrics();
 
+    /**
+     * Temporarily (will be reverted after the benchmark execution finishes) overwrites the existing
+     * logstash.yml configuration in this installation.
+     * @param meta Benchmark settings
+     */
+    void configure(BenchmarkMeta meta);
+
     final class FromRelease implements LogstashInstallation {
 
         private final LogstashInstallation base;
@@ -79,6 +87,11 @@ public String metrics() {
             return base.metrics();
         }
 
+        @Override
+        public void configure(final BenchmarkMeta meta) {
+            base.configure(meta);
+        }
+
         private static void download(final File pwd, final String version)
             throws IOException, NoSuchAlgorithmException {
             LsBenchDownloader.downloadDecompress(
@@ -105,6 +118,10 @@ final class FromLocalPath implements LogstashInstallation {
 
         private final ProcessBuilder pbuilder;
 
+        private final Path previousYml;
+
+        private final Path logstashYml;
+
         public FromLocalPath(final String path) {
             this.location = Paths.get(path);
             this.pbuilder = new ProcessBuilder().directory(location.toFile());
@@ -113,6 +130,16 @@ public FromLocalPath(final String path) {
             final Map<String, String> env = pbuilder.environment();
             env.put("JRUBY_HOME", jruby.toString());
             env.put("JAVA_OPTS", "");
+            logstashYml = location.resolve("config").resolve("logstash.yml");
+            try {
+                previousYml =
+                    Files.copy(
+                        logstashYml, logstashYml.getParent().resolve("logstash.yml.back"),
+                        StandardCopyOption.REPLACE_EXISTING
+                    );
+            } catch (final IOException ex) {
+                throw new IllegalStateException(ex);
+            }
         }
 
         @Override
@@ -134,9 +161,8 @@ public void execute(final String configuration, final File data, final int repea
             LsBenchFileUtil.ensureExecutable(lsbin.toFile());
             final File output = Files.createTempFile(null, null).toFile();
             final Process process =
-                pbuilder.command(lsbin.toString(), "-w", "2", "-f", cfg.toString()).redirectOutput(
-                    ProcessBuilder.Redirect.to(output)
-                ).start();
+                pbuilder.command(lsbin.toString(), "-f", cfg.toString()
+                ).redirectOutput(ProcessBuilder.Redirect.to(output)).start();
             if (data != null) {
                 try (final OutputStream out = process.getOutputStream()) {
                     pipeRepeatedly(data, out, repeat);
@@ -146,6 +172,7 @@ public void execute(final String configuration, final File data, final int repea
                 throw new IllegalStateException("Logstash failed to start!");
             }
             LsBenchFileUtil.ensureDeleted(cfg.toFile());
+            Files.move(previousYml, logstashYml, StandardCopyOption.REPLACE_EXISTING);
             LsBenchFileUtil.ensureDeleted(output);
         }
 
@@ -154,6 +181,22 @@ public String metrics() {
             return METRICS_URL;
         }
 
+        @Override
+        public void configure(final BenchmarkMeta meta) {
+            try {
+                Files.write(
+                    logstashYml, String.format(
+                        "pipeline.workers: %d\npipeline.batch.size: %d", meta.getWorkers()
+                        , meta.getBatchsize()
+                    ).getBytes(StandardCharsets.UTF_8),
+                    StandardOpenOption.WRITE, StandardOpenOption.TRUNCATE_EXISTING,
+                    StandardOpenOption.CREATE
+                );
+            } catch (final IOException ex) {
+                throw new IllegalStateException(ex);
+            }
+        }
+
         /**
          * Pipes the content of the given input {@link File} to the given {@link OutputStream}
          * repeatedly.
@@ -209,6 +252,11 @@ public String metrics() {
             return base.metrics();
         }
 
+        @Override
+        public void configure(final BenchmarkMeta meta) {
+            base.configure(meta);
+        }
+
         private static void download(final File pwd, final String user, final String hash)
             throws IOException, NoSuchAlgorithmException {
             LsBenchDownloader.downloadDecompress(

tools/benchmark-cli/src/main/java/org/logstash/benchmark/cli/Main.java

@@ -3,23 +3,18 @@
 import java.io.File;
 import java.io.IOException;
 import java.io.InputStream;
-import java.net.InetAddress;
 import java.net.URI;
-import java.net.UnknownHostException;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.security.NoSuchAlgorithmException;
 import java.util.AbstractMap;
-import java.util.HashMap;
-import java.util.Map;
 import java.util.Properties;
 import java.util.concurrent.TimeUnit;
 import joptsimple.OptionException;
 import joptsimple.OptionParser;
 import joptsimple.OptionSet;
 import joptsimple.OptionSpec;
 import joptsimple.OptionSpecBuilder;
-import org.apache.commons.lang3.SystemUtils;
 import org.logstash.benchmark.cli.cases.ApacheLogsComplex;
 import org.logstash.benchmark.cli.cases.Case;
 import org.logstash.benchmark.cli.cases.GeneratorToStdout;
@@ -46,7 +41,7 @@ private Main() {
      * CLI Entrypoint.
      * @param args Cli Args
      */
-    public static void main(final String... args) throws Exception {
+    public static void main(final String... args) throws IOException, NoSuchAlgorithmException {
         final OptionParser parser = new OptionParser();
         final OptionSpecBuilder gitbuilder =
             parser.accepts(UserInput.GIT_VERSION_PARAM, UserInput.GIT_VERSION_HELP);
@@ -82,6 +77,14 @@ public static void main(final String... args) throws Exception {
         final OptionSpec<Integer> repeats = parser.accepts(
             UserInput.REPEAT_PARAM, UserInput.REPEAT_PARAM_HELP
         ).withRequiredArg().ofType(Integer.class).defaultsTo(1).forHelp();
+        final OptionSpec<Integer> workers = parser.accepts(
+            UserInput.LS_WORKER_THREADS, UserInput.LS_WORKER_THREADS_HELP
+        ).withRequiredArg().ofType(Integer.class)
+            .defaultsTo(UserInput.LS_WORKER_THREADS_DEFAULT).forHelp();
+        final OptionSpec<Integer> batchsize = parser.accepts(
+            UserInput.LS_BATCH_SIZE, UserInput.LS_BATCH_SIZE_HELP
+        ).withRequiredArg().ofType(Integer.class)
+            .defaultsTo(UserInput.LS_BATCHSIZE_DEFAULT).forHelp();
         final OptionSet options;
         try {
             options = parser.parse(args);
@@ -105,30 +108,40 @@ public static void main(final String... args) throws Exception {
         settings.setProperty(
             LsBenchSettings.INPUT_DATA_REPEAT, String.valueOf(options.valueOf(repeats))
         );
+        final BenchmarkMeta runConfig = new BenchmarkMeta(
+            options.valueOf(testcase), version, type, options.valueOf(workers),
+            options.valueOf(batchsize)
+        );
         execute(
-            new UserOutput(System.out), settings, options.valueOf(testcase),
-            options.valueOf(pwd).toPath(), version, type, options.valueOf(esout)
+            new UserOutput(System.out), settings, options.valueOf(pwd).toPath(), runConfig,
+            options.valueOf(esout)
         );
     }
 
     /**
      * Programmatic Entrypoint.
      * @param output Output Printer
      * @param settings Properties
-     * @param test String identifier of the testcase to run
      * @param cwd Working Directory to run in and write cache files to
-     * @param version Version of Logstash to benchmark
-     * @param type Type of Logstash version to benchmark
+     * @param runConfig Logstash Settings
      * @param esout Elastic Search URL (empty string is interpreted as not using ES output)
-     * @throws Exception On Failure
+     * @throws IOException On Failure
+     * @throws NoSuchAlgorithmException On Failure
      */
     public static void execute(final UserOutput output, final Properties settings,
-        final String test, final Path cwd, final String version, final LsVersionType type,
-        final String esout)
-        throws Exception {
+        final Path cwd, final BenchmarkMeta runConfig, final String esout)
+        throws IOException, NoSuchAlgorithmException {
         output.printBanner();
         output.printLine();
+        final String version = runConfig.getVersion();
         output.green(String.format("Benchmarking Version: %s", version));
+        output.green(
+            String.format(
+                "Logstash Parameters: -w %d -b %d", runConfig.getWorkers(),
+                runConfig.getBatchsize()
+            )
+        );
+        final String test = runConfig.getTestcase();
         output.green(
             String.format(
                 "Running Test Case: %s (x%d)", test,
@@ -138,6 +151,7 @@ public static void execute(final UserOutput output, final Properties settings,
         output.printLine();
         Files.createDirectories(cwd);
         final LogstashInstallation logstash;
+        final LsVersionType type = runConfig.getVtype();
         if (type == LsVersionType.GIT) {
             logstash = LsBenchLsSetup.logstashFromGit(
                 cwd.toAbsolutePath().toString(), version, JRubyInstallation.bootstrapJruby(cwd)
@@ -146,8 +160,8 @@ public static void execute(final UserOutput output, final Properties settings,
             logstash =
                 LsBenchLsSetup.setupLS(cwd.toAbsolutePath().toString(), version, type, output);
         }
-        try (final DataStore store = setupDataStore(esout, test, version, type)) {
-            final Case testcase = setupTestCase(store, logstash, cwd, settings, test, output);
+        try (final DataStore store = setupDataStore(esout, runConfig)) {
+            final Case testcase = setupTestCase(store, logstash, cwd, settings, runConfig, output);
             output.printStartTime();
             final long start = System.currentTimeMillis();
             final AbstractMap<LsMetricStats, ListStatistics> stats = testcase.run();
@@ -163,45 +177,32 @@ public static void execute(final UserOutput output, final Properties settings,
     }
 
     private static Case setupTestCase(final DataStore store, final LogstashInstallation logstash,
-        final Path cwd, final Properties settings, final String test, final UserOutput output)
+        final Path cwd, final Properties settings, final BenchmarkMeta runConfig, final UserOutput output)
         throws IOException, NoSuchAlgorithmException {
         final Case testcase;
+        final String test = runConfig.getTestcase();
         if (GeneratorToStdout.IDENTIFIER.equalsIgnoreCase(test)) {
-            testcase = new GeneratorToStdout(store, logstash, settings);
+            testcase = new GeneratorToStdout(store, logstash, settings, runConfig);
         } else if (ApacheLogsComplex.IDENTIFIER.equalsIgnoreCase(test)) {
-            testcase = new ApacheLogsComplex(store, logstash, cwd, settings, output);
+            testcase = new ApacheLogsComplex(store, logstash, cwd, settings, output, runConfig);
         } else {
             throw new IllegalArgumentException(String.format("Unknown test case %s", test));
         }
         return testcase;
     }
 
-    private static DataStore setupDataStore(final String elastic, final String test,
-        final String version, final LsVersionType vtype) throws UnknownHostException {
+    private static DataStore setupDataStore(final String elastic, final BenchmarkMeta config) {
         if (elastic.isEmpty()) {
             return DataStore.NONE;
         } else {
             final URI elasticsearch = URI.create(elastic);
             return new DataStore.ElasticSearch(
                 elasticsearch.getHost(), elasticsearch.getPort(), elasticsearch.getScheme(),
-                meta(test, version, vtype)
+                config.asMap()
             );
         }
     }
 
-    private static Map<String, Object> meta(final String testcase, final String version,
-        final LsVersionType vtype) throws UnknownHostException {
-        final Map<String, Object> result = new HashMap<>();
-        result.put("test_name", testcase);
-        result.put("os_name", SystemUtils.OS_NAME);
-        result.put("os_version", SystemUtils.OS_VERSION);
-        result.put("host_name", InetAddress.getLocalHost().getHostName());
-        result.put("cpu_cores", Runtime.getRuntime().availableProcessors());
-        result.put("version_type", vtype);
-        result.put("version", version);
-        return result;
-    }
-
     private static Properties loadSettings() throws IOException {
         final Properties props = new Properties();
         try (final InputStream settings =

tools/benchmark-cli/src/main/java/org/logstash/benchmark/cli/cases/ApacheLogsComplex.java

@@ -11,6 +11,7 @@
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeoutException;
 import org.apache.commons.io.IOUtils;
+import org.logstash.benchmark.cli.BenchmarkMeta;
 import org.logstash.benchmark.cli.DataStore;
 import org.logstash.benchmark.cli.LogstashInstallation;
 import org.logstash.benchmark.cli.LsBenchSettings;
@@ -46,14 +47,15 @@ public final class ApacheLogsComplex implements Case {
     private final int repeats;
 
     public ApacheLogsComplex(final DataStore store, final LogstashInstallation logstash,
-        final Path cwd, final Properties settings, final UserOutput output)
-        throws IOException, NoSuchAlgorithmException {
+        final Path cwd, final Properties settings, final UserOutput output,
+        final BenchmarkMeta runConfig) throws IOException, NoSuchAlgorithmException {
         data = cwd.resolve("data_apache").resolve("apache_access_logs").toFile();
         ensureDatafile(
             data.toPath().getParent().toFile(),
             settings.getProperty(LsBenchSettings.APACHE_DATASET_URL), output
         );
         this.logstash = logstash;
+        logstash.configure(runConfig);
         this.store = store;
         repeats = Integer.parseInt(settings.getProperty(LsBenchSettings.INPUT_DATA_REPEAT));
     }