Allow configuring exact datanode directories without automatic nodeid subdirs (#17255)

* Allow configuring exact datanode directories without automatic nodeid subdirs

* added changelog

* use proper locking mechanism to create datanode.lock files

* code cleanup

* code cleanup

* removed unused imports

* fixed test

---------

Co-authored-by: Matthias Oesterheld <33032967+moesterheld@users.noreply.github.com>

data-node/src/main/java/org/graylog/datanode/Configuration.java

@@ -24,8 +24,6 @@ import com.github.joschi.jadconfig.ValidatorMethod;
 import com.github.joschi.jadconfig.converters.IntegerConverter;
 import com.github.joschi.jadconfig.converters.StringListConverter;
 import com.github.joschi.jadconfig.util.Duration;
-import com.github.joschi.jadconfig.validators.DirectoryPathReadableValidator;
-import com.github.joschi.jadconfig.validators.DirectoryPathWritableValidator;
 import com.github.joschi.jadconfig.validators.PositiveIntegerValidator;
 import com.github.joschi.jadconfig.validators.StringNotBlankValidator;
 import com.github.joschi.jadconfig.validators.URIAbsoluteValidator;
@@ -93,7 +91,7 @@ public class Configuration extends BaseConfiguration {
     @Parameter(value = "opensearch_config_location", required = true, validators = DirectoryWritableValidator.class)
     private Path opensearchConfigLocation = Path.of("datanode/config");
 
-    @Parameter(value = "config_location", validators = DirectoryPathReadableValidator.class)
+    @Parameter(value = "config_location", validators = DirectoryReadableValidator.class)
     private Path configLocation = null;
 
     @Parameter(value = "process_logs_buffer_size")

data-node/src/main/java/org/graylog/datanode/DirectoryReadableValidator.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright (C) 2020 Graylog, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the Server Side Public License, version 1,
+ * as published by MongoDB, Inc.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * Server Side Public License for more details.
+ *
+ * You should have received a copy of the Server Side Public License
+ * along with this program. If not, see
+ * <http://www.mongodb.com/licensing/server-side-public-license>.
+ */
+package org.graylog.datanode;
+
+import com.github.joschi.jadconfig.ValidationException;
+import com.github.joschi.jadconfig.validators.DirectoryPathReadableValidator;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+/**
+ * The original validator doesn't tell if the problem is nonexistence of the directory, file instead of dir
+ * or permissions problem. This validator provides more specific exception with detailed message.
+ */
+public class DirectoryReadableValidator extends DirectoryPathReadableValidator {
+    @Override
+    public void validate(String name, Path value) throws ValidationException {
+        if (value == null) {
+            return;
+        }
+
+        if (!Files.exists(value)) {
+            throw new ValidationException("Cannot read from directory " + name + " at path " + value + ". Directory doesn't exist. Please create the directory.");
+        }
+
+        if (!Files.isDirectory(value)) {
+            throw new ValidationException("Cannot read from directory " + name + " at path " + value + ". Referenced path is not a directory.");
+        }
+
+        if (!Files.isReadable(value)) {
+            throw new ValidationException("Cannot read from directory " + name + " at path " + value + ". Please set read permissions.");
+        }
+    }
+}

data-node/src/main/java/org/graylog/datanode/bindings/PreflightChecksBindings.java

@@ -18,6 +18,7 @@ package org.graylog.datanode.bindings;
 
 import com.google.inject.AbstractModule;
 import com.google.inject.multibindings.MapBinder;
+import org.graylog.datanode.bootstrap.preflight.DatanodeDirectoriesLockfileCheck;
 import org.graylog.datanode.bootstrap.preflight.OpensearchBinPreflightCheck;
 import org.graylog.datanode.bootstrap.preflight.OpensearchConfigSync;
 import org.graylog2.bootstrap.preflight.PreflightCheck;
@@ -29,6 +30,7 @@ public class PreflightChecksBindings extends AbstractModule {
     protected void configure() {
         addPreflightCheck(OpensearchConfigSync.class);
         addPreflightCheck(OpensearchBinPreflightCheck.class);
+        addPreflightCheck(DatanodeDirectoriesLockfileCheck.class);
     }
 
 

data-node/src/main/java/org/graylog/datanode/bootstrap/preflight/DatanodeDirectoriesLockfileCheck.java

@@ -0,0 +1,132 @@
+/*
+ * Copyright (C) 2020 Graylog, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the Server Side Public License, version 1,
+ * as published by MongoDB, Inc.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * Server Side Public License for more details.
+ *
+ * You should have received a copy of the Server Side Public License
+ * along with this program. If not, see
+ * <http://www.mongodb.com/licensing/server-side-public-license>.
+ */
+package org.graylog.datanode.bootstrap.preflight;
+
+import com.google.common.io.CharStreams;
+import org.graylog.datanode.configuration.DatanodeConfiguration;
+import org.graylog.datanode.configuration.DatanodeDirectories;
+import org.graylog2.bootstrap.preflight.PreflightCheck;
+import org.graylog2.bootstrap.preflight.PreflightCheckException;
+import org.graylog2.plugin.system.NodeId;
+
+import javax.inject.Inject;
+import java.io.IOException;
+import java.io.Reader;
+import java.io.Writer;
+import java.nio.channels.Channels;
+import java.nio.channels.FileChannel;
+import java.nio.channels.FileLock;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * To prevent two or more datanodes using the same directories, we are writing a datanode.lock files into the dirs.
+ * The content of the lock file is the nodeid of the datanode. During startup we are checking that the directory
+ * has either no lock (and then we create one) or if it has a lock, then we verify its content against current nodeid.
+ */
+public class DatanodeDirectoriesLockfileCheck implements PreflightCheck {
+
+    protected static final Path DATANODE_LOCKFILE = Path.of("datanode.lock");
+    private final DatanodeDirectories directories;
+    private final String nodeId;
+
+    @Inject
+    public DatanodeDirectoriesLockfileCheck(NodeId nodeId, DatanodeConfiguration datanodeConfiguration) {
+        this(nodeId.getNodeId(), datanodeConfiguration.datanodeDirectories());
+    }
+
+    public DatanodeDirectoriesLockfileCheck(String nodeId, DatanodeDirectories directories) {
+        this.directories = directories;
+        this.nodeId = nodeId;
+    }
+
+    @Override
+    public void runCheck() throws PreflightCheckException {
+        checkDatanodeLock(directories.getConfigurationTargetDir());
+        checkDatanodeLock(directories.getDataTargetDir());
+        checkDatanodeLock(directories.getLogsTargetDir());
+    }
+
+    private void checkDatanodeLock(Path dir) {
+        final Path lockfile = dir.resolve(DATANODE_LOCKFILE);
+        try (FileChannel channel = FileChannel.open(lockfile, StandardOpenOption.READ, StandardOpenOption.CREATE, StandardOpenOption.WRITE)) {
+            doCheckLockFile(channel, dir);
+        } catch (IOException e) {
+            throw new DatanodeLockFileException("Failed to open channel to lock file " + lockfile.toAbsolutePath(), e);
+        }
+    }
+
+    private void doCheckLockFile(FileChannel channel, Path dir) {
+        FileLock lock = null;
+        try {
+            lock = channel.lock();
+            // now we are the only process that can access the lock file.
+            verifyOrCreateLockFile(channel, dir);
+        } catch (IOException e) {
+            throw new DatanodeLockFileException("Failed to obtain lock", e);
+        } finally {
+            releaseLock(lock);
+        }
+    }
+
+    private void verifyOrCreateLockFile(FileChannel channel, Path dir) {
+        readChannel(channel).ifPresentOrElse(
+                lockedForID -> verifyLockFileContent(lockedForID, dir),
+                () -> writeLockFile(channel)
+        );
+    }
+
+    private void releaseLock(FileLock lock) {
+        try {
+            if (lock != null) {
+                lock.release();
+            }
+        } catch (IOException e) {
+            throw new DatanodeLockFileException("Failed to release lock file", e);
+        }
+    }
+
+    private void verifyLockFileContent(String lockedForID, Path dir) {
+        if (!Objects.equals(lockedForID, nodeId)) {
+            throw new DatanodeLockFileException("Directory " + dir + " locked for datanode " + lockedForID + ", access with datanode " + nodeId + " rejected. Please check your configuration and make sure that there is only one datanode instance using this directory.");
+        }
+    }
+
+    private void writeLockFile(FileChannel channel) {
+        try {
+            // do not close the writer, otherwise it will close the underlying channel. We do that explicitly elsewhere.
+            final Writer writer = Channels.newWriter(channel, StandardCharsets.UTF_8);
+            writer.write(nodeId);
+            writer.flush();
+        } catch (IOException e) {
+            throw new DatanodeLockFileException("Failed to write node ID to the lock file", e);
+        }
+    }
+
+    private Optional<String> readChannel(FileChannel channel) {
+        try {
+            // do not close the reader, otherwise it will close the underlying channel. We do that explicitly elsewhere.
+            final Reader reader = Channels.newReader(channel, StandardCharsets.UTF_8);
+            return Optional.of(CharStreams.toString(reader)).filter(v -> !v.isBlank()).map(String::trim);
+        } catch (IOException e) {
+            throw new DatanodeLockFileException("Failed to read content of lock file", e);
+        }
+    }
+}

data-node/src/main/java/org/graylog/datanode/bootstrap/preflight/DatanodeLockFileException.java

@@ -0,0 +1,27 @@
+ /*
+ * Copyright (C) 2020 Graylog, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the Server Side Public License, version 1,
+ * as published by MongoDB, Inc.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * Server Side Public License for more details.
+ *
+ * You should have received a copy of the Server Side Public License
+ * along with this program. If not, see
+ * <http://www.mongodb.com/licensing/server-side-public-license>.
+ */
+package org.graylog.datanode.bootstrap.preflight;
+
+public class DatanodeLockFileException extends RuntimeException {
+    public DatanodeLockFileException(String message, Exception cause) {
+        super(message, cause);
+    }
+
+    public DatanodeLockFileException(String message) {
+        super(message);
+    }
+}

data-node/src/main/java/org/graylog/datanode/configuration/DatanodeDirectories.java

@@ -41,14 +41,12 @@ public class DatanodeDirectories {
 
     private static final Logger LOG = LoggerFactory.getLogger(DatanodeDirectories.class);
 
-    private final String nodeId;
     private final Path dataTargetDir;
     private final Path logsTargetDir;
     private final Path configurationSourceDir;
     private final Path configurationTargetDir;
 
-    public DatanodeDirectories(String nodeId, Path dataTargetDir, Path logsTargetDir, @Nullable Path configurationSourceDir, Path configurationTargetDir) {
-        this.nodeId = nodeId;
+    public DatanodeDirectories(Path dataTargetDir, Path logsTargetDir, @Nullable Path configurationSourceDir, Path configurationTargetDir) {
         this.dataTargetDir = dataTargetDir;
         this.logsTargetDir = logsTargetDir;
         this.configurationSourceDir = configurationSourceDir;
@@ -57,23 +55,41 @@ public class DatanodeDirectories {
 
     public static DatanodeDirectories fromConfiguration(Configuration configuration, NodeId nodeId) {
         final DatanodeDirectories directories = new DatanodeDirectories(
-                nodeId.getNodeId(),
-                configuration.getOpensearchDataLocation(),
-                configuration.getOpensearchLogsLocation(),
+                backwardsCompatible(configuration.getOpensearchDataLocation(), nodeId, "opensearch_data_location"),
+                backwardsCompatible(configuration.getOpensearchLogsLocation(), nodeId, "opensearch_logs_location"),
                 configuration.getDatanodeConfigurationLocation(),
-                configuration.getOpensearchConfigLocation()
+                backwardsCompatible(configuration.getOpensearchConfigLocation(), nodeId, "opensearch_config_location")
         );
 
         LOG.info("Opensearch of the node {} uses following directories as its storage: {}", nodeId.getNodeId(), directories);
         return directories;
     }
 
+    /**
+     * Originally we created a subdir named by the node ID for each of the data/config/logs directories and automatically
+     * used that subdir. Later we discovered that this won't allow us to run rolling upgrades for opensearch, as we
+     * are unable to point the configuration to an exact directory. This method works as a backwards compatible
+     * fallback, detecting the presence of the node ID subdir and using it, if available. It also logs a warning with
+     * configuration change suggestion.
+     * TODO: Remove in 6.0 release
+     */
+    @Deprecated(forRemoval = true)
+    @NotNull
+    protected static Path backwardsCompatible(@NotNull Path path, NodeId nodeId, String configProperty) {
+        final Path nodeIdSubdir = path.resolve(nodeId.getNodeId());
+        if(Files.exists(nodeIdSubdir) && Files.isDirectory(nodeIdSubdir)) {
+            LOG.warn("Caution, this datanode instance uses old format of directories. Please configure {} to point directly to {}", configProperty, nodeIdSubdir.toAbsolutePath());
+            return nodeIdSubdir;
+        }
+        return path;
+    }
+
     /**
      * This directory is used by the managed opensearch to store its data in it.
      * Read-write permissions required.
      */
     public Path getDataTargetDir() {
-        return resolveNodeSubdir(dataTargetDir);
+        return dataTargetDir.toAbsolutePath();
     }
 
     /**
@@ -81,7 +97,7 @@ public class DatanodeDirectories {
      * Read-write permissions required.
      */
     public Path getLogsTargetDir() {
-        return resolveNodeSubdir(logsTargetDir);
+        return logsTargetDir.toAbsolutePath();
     }
 
 
@@ -110,7 +126,7 @@ public class DatanodeDirectories {
      * Read-write permissions required.
      */
     public Path getConfigurationTargetDir() {
-        return resolveNodeSubdir(configurationTargetDir);
+        return configurationTargetDir.toAbsolutePath();
     }
 
     public Path createConfigurationFile(Path relativePath) throws IOException {
@@ -132,12 +148,12 @@ public class DatanodeDirectories {
      * @see org.graylog.datanode.bootstrap.preflight.OpensearchConfigSync
      */
     public Path getOpensearchProcessConfigurationDir() {
-        return resolveNodeSubdir(configurationTargetDir).resolve("opensearch");
+        return getConfigurationTargetDir().resolve("opensearch");
     }
 
     public Path createOpensearchProcessConfigurationDir() throws IOException {
         final Path dir = getOpensearchProcessConfigurationDir();
-        // TODO: should we always delete existing process configuration dir and recreate if here? IMHO yes
+        // TODO: should we always delete existing process configuration dir and recreate it here? IMHO yes
         final Set<PosixFilePermission> permissions = Set.of(PosixFilePermission.OWNER_EXECUTE, PosixFilePermission.OWNER_WRITE, PosixFilePermission.OWNER_READ);
         final FileAttribute<Set<PosixFilePermission>> fileAttributes = PosixFilePermissions.asFileAttribute(permissions);
         Files.createDirectories(dir, fileAttributes);
@@ -149,11 +165,6 @@ public class DatanodeDirectories {
         return createRestrictedAccessFile(resolvedPath);
     }
 
-
-    private Path resolveNodeSubdir(Path path) {
-        return path.resolve(nodeId).toAbsolutePath();
-    }
-
     @Override
     public String toString() {
         return "DatanodeDirectories{" +