Decouple environment from DiscoveryNode (#54373)

Today Environment is coupled to DiscoveryNode via the node.local_storage
setting. This commit decouples Environment from this setting.

server/src/main/java/org/elasticsearch/env/Environment.java

@@ -20,7 +20,6 @@
 package org.elasticsearch.env;
 
 import org.elasticsearch.cluster.ClusterName;
-import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.SuppressForbidden;
 import org.elasticsearch.common.io.PathUtils;
 import org.elasticsearch.common.settings.Setting;
@@ -91,11 +90,15 @@ public class Environment {
     private final Path tmpFile;
 
     public Environment(final Settings settings, final Path configPath) {
-        this(settings, configPath, PathUtils.get(System.getProperty("java.io.tmpdir")));
+        this(settings, configPath, true);
+    }
+
+    public Environment(final Settings settings, final Path configPath, final boolean nodeLocalStorage) {
+        this(settings, configPath, nodeLocalStorage, PathUtils.get(System.getProperty("java.io.tmpdir")));
     }
 
     // Should only be called directly by this class's unit tests
-    Environment(final Settings settings, final Path configPath, final Path tmpPath) {
+    Environment(final Settings settings, final Path configPath, final boolean nodeLocalStorage, final Path tmpPath) {
         final Path homeFile;
         if (PATH_HOME_SETTING.exists(settings)) {
             homeFile = PathUtils.get(PATH_HOME_SETTING.get(settings)).toAbsolutePath().normalize();
@@ -115,7 +118,7 @@ public class Environment {
 
         List<String> dataPaths = PATH_DATA_SETTING.get(settings);
         final ClusterName clusterName = ClusterName.CLUSTER_NAME_SETTING.get(settings);
-        if (DiscoveryNode.nodeRequiresLocalStorage(settings)) {
+        if (nodeLocalStorage) {
             if (dataPaths.isEmpty() == false) {
                 dataFiles = new Path[dataPaths.size()];
                 for (int i = 0; i < dataPaths.size(); i++) {

server/src/main/java/org/elasticsearch/node/Node.java

@@ -332,7 +332,7 @@ public class Node implements Closeable {
 
             // create the environment based on the finalized (processed) view of the settings
             // this is just to makes sure that people get the same settings, no matter where they ask them from
-            this.environment = new Environment(settings, initialEnvironment.configFile());
+            this.environment = new Environment(settings, initialEnvironment.configFile(), Node.NODE_LOCAL_STORAGE_SETTING.get(settings));
             Environment.assertEquivalent(initialEnvironment, this.environment);
 
             final List<ExecutorBuilder<?>> executorBuilders = pluginsService.getExecutorBuilders(settings);

server/src/test/java/org/elasticsearch/env/EnvironmentTests.java

@@ -129,11 +129,10 @@ public class EnvironmentTests extends ESTestCase {
         final Settings settings =
                 Settings.builder()
                         .put("path.home", pathHome)
-                        .put("node.local_storage", false)
                         .put("node.master", false)
                         .put("node.data", false)
                         .build();
-        final Environment environment = new Environment(settings, null);
+        final Environment environment = new Environment(settings, null, false);
         assertThat(environment.dataFiles(), arrayWithSize(0));
     }
 
@@ -144,11 +143,10 @@ public class EnvironmentTests extends ESTestCase {
                 Settings.builder()
                         .put("path.home", pathHome)
                         .put("path.data", pathData)
-                        .put("node.local_storage", false)
                         .put("node.master", false)
                         .put("node.data", false)
                         .build();
-        final IllegalStateException e = expectThrows(IllegalStateException.class, () -> new Environment(settings, null));
+        final IllegalStateException e = expectThrows(IllegalStateException.class, () -> new Environment(settings, null, false));
         assertThat(e, hasToString(containsString("node does not require local storage yet path.data is set to [" + pathData + "]")));
     }
 
@@ -156,7 +154,7 @@ public class EnvironmentTests extends ESTestCase {
         Settings build = Settings.builder()
             .put(Environment.PATH_HOME_SETTING.getKey(), createTempDir())
             .build();
-        Environment environment = new Environment(build, null, createTempDir().resolve("this_does_not_exist"));
+        Environment environment = new Environment(build, null, true, createTempDir().resolve("this_does_not_exist"));
         FileNotFoundException e = expectThrows(FileNotFoundException.class, environment::validateTmpFile);
         assertThat(e.getMessage(), startsWith("Temporary file directory ["));
         assertThat(e.getMessage(), endsWith("this_does_not_exist] does not exist or is not accessible"));
@@ -166,7 +164,7 @@ public class EnvironmentTests extends ESTestCase {
         Settings build = Settings.builder()
             .put(Environment.PATH_HOME_SETTING.getKey(), createTempDir())
             .build();
-        Environment environment = new Environment(build, null, createTempFile("something", ".test"));
+        Environment environment = new Environment(build, null, true, createTempFile("something", ".test"));
         IOException e = expectThrows(IOException.class, environment::validateTmpFile);
         assertThat(e.getMessage(), startsWith("Configured temporary file directory ["));
         assertThat(e.getMessage(), endsWith(".test] is not a directory"));
@@ -186,7 +184,7 @@ public class EnvironmentTests extends ESTestCase {
         // the above paths will be treated as relative to the working directory
         final Path workingDirectory = PathUtils.get(System.getProperty("user.dir"));
 
-        final Environment environment = new Environment(settings, null, createTempDir());
+        final Environment environment = new Environment(settings, null, true, createTempDir());
         final String homePath = Environment.PATH_HOME_SETTING.get(environment.settings());
         assertPath(homePath, workingDirectory.resolve("home"));