Merge branch 'main' into adding-settings-to-data-streams

Author: masseyke

distribution/tools/server-cli/src/main/java/org/elasticsearch/server/cli/ServerCli.java

@@ -271,8 +271,7 @@ protected ServerProcess startServer(Terminal terminal, ProcessInfo processInfo,
             .withProcessInfo(processInfo)
             .withServerArgs(args)
             .withTempDir(tempDir)
-            .withJvmOptions(jvmOptions)
-            .withWorkingDir(args.logsDir());
+            .withJvmOptions(jvmOptions);
         return serverProcessBuilder.start();
     }
 

distribution/tools/server-cli/src/main/java/org/elasticsearch/server/cli/ServerProcessBuilder.java

@@ -10,6 +10,7 @@
 package org.elasticsearch.server.cli;
 
 import org.elasticsearch.bootstrap.ServerArgs;
+import org.elasticsearch.cli.ExitCodes;
 import org.elasticsearch.cli.ProcessInfo;
 import org.elasticsearch.cli.Terminal;
 import org.elasticsearch.cli.UserException;
@@ -21,6 +22,8 @@
 import java.io.IOException;
 import java.io.OutputStream;
 import java.io.UncheckedIOException;
+import java.nio.file.FileAlreadyExistsException;
+import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.HashMap;
 import java.util.List;
@@ -44,7 +47,6 @@ public class ServerProcessBuilder {
     private ServerArgs serverArgs;
     private ProcessInfo processInfo;
     private List<String> jvmOptions;
-    private Path workingDir;
     private Terminal terminal;
 
     // this allows mocking the process building by tests
@@ -84,11 +86,6 @@ public ServerProcessBuilder withJvmOptions(List<String> jvmOptions) {
         return this;
     }
 
-    public ServerProcessBuilder withWorkingDir(Path workingDir) {
-        this.workingDir = workingDir;
-        return this;
-    }
-
     /**
      * Specifies the {@link Terminal} to use for reading input and writing output from/to the cli console
      */
@@ -141,6 +138,17 @@ public ServerProcess start() throws UserException {
         return start(ProcessBuilder::start);
     }
 
+    private void ensureWorkingDirExists() throws UserException {
+        Path workingDir = serverArgs.logsDir();
+        try {
+            Files.createDirectories(workingDir);
+        } catch (FileAlreadyExistsException e) {
+            throw new UserException(ExitCodes.CONFIG, "Logs dir [" + workingDir + "] exists but is not a directory", e);
+        } catch (IOException e) {
+            throw new UserException(ExitCodes.CONFIG, "Unable to create logs dir [" + workingDir + "]", e);
+        }
+    }
+
     private static void checkRequiredArgument(Object argument, String argumentName) {
         if (argument == null) {
             throw new IllegalStateException(
@@ -157,12 +165,14 @@ ServerProcess start(ProcessStarter processStarter) throws UserException {
         checkRequiredArgument(jvmOptions, "jvmOptions");
         checkRequiredArgument(terminal, "terminal");
 
+        ensureWorkingDirExists();
+
         Process jvmProcess = null;
         ErrorPumpThread errorPump;
 
         boolean success = false;
         try {
-            jvmProcess = createProcess(getCommand(), getJvmArgs(), jvmOptions, getEnvironment(), workingDir, processStarter);
+            jvmProcess = createProcess(getCommand(), getJvmArgs(), jvmOptions, getEnvironment(), serverArgs.logsDir(), processStarter);
             errorPump = new ErrorPumpThread(terminal, jvmProcess.getErrorStream());
             errorPump.start();
             sendArgs(serverArgs, jvmProcess.getOutputStream());

distribution/tools/server-cli/src/test/java/org/elasticsearch/server/cli/ServerProcessTests.java

@@ -65,7 +65,7 @@ public class ServerProcessTests extends ESTestCase {
     protected final Map<String, String> sysprops = new HashMap<>();
     protected final Map<String, String> envVars = new HashMap<>();
     Path esHomeDir;
-    Path workingDir;
+    Path logsDir;
     Settings.Builder nodeSettings;
     ProcessValidator processValidator;
     MainMethod mainCallback;
@@ -94,8 +94,8 @@ public void resetEnv() {
         sysprops.put("os.name", "Linux");
         sysprops.put("java.home", "javahome");
         sysprops.put("es.path.home", esHomeDir.toString());
+        logsDir = esHomeDir.resolve("logs");
         envVars.clear();
-        workingDir = createTempDir();
         nodeSettings = Settings.builder();
         processValidator = null;
         mainCallback = null;
@@ -207,15 +207,7 @@ ProcessInfo createProcessInfo() {
     }
 
     ServerArgs createServerArgs(boolean daemonize, boolean quiet) {
-        return new ServerArgs(
-            daemonize,
-            quiet,
-            null,
-            secrets,
-            nodeSettings.build(),
-            esHomeDir.resolve("config"),
-            esHomeDir.resolve("logs")
-        );
+        return new ServerArgs(daemonize, quiet, null, secrets, nodeSettings.build(), esHomeDir.resolve("config"), logsDir);
     }
 
     ServerProcess startProcess(boolean daemonize, boolean quiet) throws Exception {
@@ -231,8 +223,7 @@ ServerProcess startProcess(boolean daemonize, boolean quiet) throws Exception {
             .withProcessInfo(pinfo)
             .withServerArgs(createServerArgs(daemonize, quiet))
             .withJvmOptions(List.of())
-            .withTempDir(ServerProcessUtils.setupTempDir(pinfo))
-            .withWorkingDir(workingDir);
+            .withTempDir(ServerProcessUtils.setupTempDir(pinfo));
         return serverProcessBuilder.start(starter);
     }
 
@@ -241,7 +232,7 @@ public void testProcessBuilder() throws Exception {
             assertThat(pb.redirectInput(), equalTo(ProcessBuilder.Redirect.PIPE));
             assertThat(pb.redirectOutput(), equalTo(ProcessBuilder.Redirect.INHERIT));
             assertThat(pb.redirectError(), equalTo(ProcessBuilder.Redirect.PIPE));
-            assertThat(String.valueOf(pb.directory()), equalTo(workingDir.toString())); // leave default, which is working directory
+            assertThat(String.valueOf(pb.directory()), equalTo(esHomeDir.resolve("logs").toString()));
         };
         mainCallback = (args, stdin, stderr, exitCode) -> {
             try (PrintStream err = new PrintStream(stderr, true, StandardCharsets.UTF_8)) {
@@ -315,8 +306,7 @@ public void testCommandLineSysprops() throws Exception {
             .withProcessInfo(createProcessInfo())
             .withServerArgs(createServerArgs(false, false))
             .withJvmOptions(List.of("-Dfoo1=bar", "-Dfoo2=baz"))
-            .withTempDir(Path.of("."))
-            .withWorkingDir(workingDir);
+            .withTempDir(Path.of("."));
         serverProcessBuilder.start(starter).waitFor();
     }
 
@@ -433,4 +423,26 @@ public void testProcessDies() throws Exception {
         int exitCode = server.waitFor();
         assertThat(exitCode, equalTo(-9));
     }
+
+    public void testLogsDirIsFile() throws Exception {
+        Files.createFile(logsDir);
+        var e = expectThrows(UserException.class, this::runForeground);
+        assertThat(e.getMessage(), containsString("exists but is not a directory"));
+    }
+
+    public void testLogsDirCreateParents() throws Exception {
+        Path testDir = createTempDir();
+        logsDir = testDir.resolve("subdir/logs");
+        processValidator = pb -> assertThat(String.valueOf(pb.directory()), equalTo(logsDir.toString()));
+        runForeground();
+    }
+
+    public void testLogsCreateFailure() throws Exception {
+        Path testDir = createTempDir();
+        Path parentFile = testDir.resolve("exists");
+        Files.createFile(parentFile);
+        logsDir = parentFile.resolve("logs");
+        var e = expectThrows(UserException.class, this::runForeground);
+        assertThat(e.getMessage(), containsString("Unable to create logs dir"));
+    }
 }

docs/reference/elasticsearch/configuration-reference/thread-pool-settings.md

@@ -94,11 +94,7 @@ The following are the types of thread pools and their respective parameters:
 
 ### `fixed` [fixed-thread-pool]
 
-The `fixed` thread pool holds a fixed size of threads to handle the requests with a queue (optionally bounded) for pending requests that have no threads to service them.
-
-The `size` parameter controls the number of threads.
-
-The `queue_size` allows to control the size of the queue of pending requests that have no threads to execute them. By default, it is set to `-1` which means its unbounded. When a request comes in and the queue is full, it will abort the request.
+A `fixed` thread pool holds a fixed number of threads as determined by the `size` parameter. If a task is submitted to a `fixed` thread pool and there are fewer than `size` busy threads in the pool then the task will execute immediately. If all the threads are busy when a task is submitted then it will be held in a queue for later execution. The `queue_size` parameter controls the maximum size of this queue. A `queue_size` of `-1` means that the queue is unbounded, but most `fixed` thread pools specify a bound on their queue size by default. If a bounded queue is full then it will reject further work, which typically causes the corresponding requests to fail.
 
 ```yaml
 thread_pool:
@@ -114,6 +110,8 @@ The `scaling` thread pool holds a dynamic number of threads. This number is prop
 
 The `keep_alive` parameter determines how long a thread should be kept around in the thread pool without it doing any work.
 
+If a task is submitted to a `scaling` thread pool when its maximum number of threads are already busy with other tasks, the new task will be held in a queue for later execution. The queue in a `scaling` thread pool is always unbounded.
+
 ```yaml
 thread_pool:
     warmer: