<commit_before>/*
 * Licensed to Elasticsearch under one or more contributor
 * license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright
 * ownership. Elasticsearch licenses this file to you 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
 *
 * 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.
 */

package org.elasticsearch.env;

import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Sets;
import com.google.common.primitives.Ints;
import org.apache.lucene.store.*;
import org.apache.lucene.util.IOUtils;
import org.elasticsearch.ElasticsearchIllegalArgumentException;
import org.elasticsearch.ElasticsearchIllegalStateException;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.common.Nullable;
import org.elasticsearch.common.component.AbstractComponent;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.io.FileSystemUtils;
import org.elasticsearch.common.settings.ImmutableSettings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.index.Index;
import org.elasticsearch.index.settings.IndexSettings;
import org.elasticsearch.index.shard.ShardId;

import java.io.Closeable;
import java.io.IOException;
import java.nio.file.*;
import java.util.*;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * A component that holds all data paths for a single node.
 */
public class NodeEnvironment extends AbstractComponent implements Closeable{

    /* ${data.paths}/nodes/{node.id} */
    private final Path[] nodePaths;
    /* ${data.paths}/nodes/{node.id}/indices */
    private final Path[] nodeIndicesPaths;
    private final Lock[] locks;
    private final boolean addNodeId;

    private final int localNodeId;
    private final AtomicBoolean closed = new AtomicBoolean(false);
    private final Map<ShardId, InternalShardLock> shardLocks = new HashMap<>();

    private final boolean customPathsEnabled;

    // Setting to automatically append node id to custom data paths
    public static final String ADD_NODE_ID_TO_CUSTOM_PATH = "node.add_id_to_custom_path";
    // Setting to enable custom index.data_path setting for new indices
    public static final String SETTING_CUSTOM_DATA_PATH_ENABLED = "node.enable_custom_paths";

    public static final String NODES_FOLDER = "nodes";
    public static final String INDICES_FOLDER = "indices";
    public static final String NODE_LOCK_FILENAME = "node.lock";

    @Inject
    public NodeEnvironment(Settings settings, Environment environment) throws IOException {
        super(settings);

        this.addNodeId = settings.getAsBoolean(ADD_NODE_ID_TO_CUSTOM_PATH, true);
        this.customPathsEnabled = settings.getAsBoolean(SETTING_CUSTOM_DATA_PATH_ENABLED, false);

        if (!DiscoveryNode.nodeRequiresLocalStorage(settings)) {
            nodePaths = null;
            nodeIndicesPaths = null;
            locks = null;
            localNodeId = -1;
            return;
        }

        final Path[] nodePaths = new Path[environment.dataWithClusterFiles().length];
        final Lock[] locks = new Lock[environment.dataWithClusterFiles().length];
        int localNodeId = -1;
        IOException lastException = null;
        int maxLocalStorageNodes = settings.getAsInt("node.max_local_storage_nodes", 50);
        for (int possibleLockId = 0; possibleLockId < maxLocalStorageNodes; possibleLockId++) {
            for (int dirIndex = 0; dirIndex < environment.dataWithClusterFiles().length; dirIndex++) {
                Path dir = environment.dataWithClusterFiles()[dirIndex].resolve(Paths.get(NODES_FOLDER, Integer.toString(possibleLockId)));
                Files.createDirectories(dir);
                
                try (Directory luceneDir = FSDirectory.open(dir, NativeFSLockFactory.INSTANCE)) {
                    logger.trace("obtaining node lock on {} ...", dir.toAbsolutePath());
                    Lock tmpLock = luceneDir.makeLock(NODE_LOCK_FILENAME);
                    boolean obtained = tmpLock.obtain();
                    if (obtained) {
                        locks[dirIndex] = tmpLock;
                        nodePaths[dirIndex] = dir;
                        localNodeId = possibleLockId;
                    } else {
                        logger.trace("failed to obtain node lock on {}", dir.toAbsolutePath());
                        // release all the ones that were obtained up until now
                        for (int i = 0; i < locks.length; i++) {
                            if (locks[i] != null) {
                                IOUtils.closeWhileHandlingException(locks[i]);
                            }
                            locks[i] = null;
                        }
                        break;
                    }
                } catch (IOException e) {
                    logger.trace("failed to obtain node lock on {}", e, dir.toAbsolutePath());
                    lastException = new IOException("failed to obtain lock on " + dir.toAbsolutePath(), e);
                    // release all the ones that were obtained up until now
                    for (int i = 0; i < locks.length; i++) {
                        IOUtils.closeWhileHandlingException(locks[i]);
                        locks[i] = null;
                    }
                    break;
                }
            }
            if (locks[0] != null) {
                // we found a lock, break
                break;
            }
        }
        if (locks[0] == null) {
            throw new ElasticsearchIllegalStateException("Failed to obtain node lock, is the following location writable?: "
                    + Arrays.toString(environment.dataWithClusterFiles()), lastException);
        }

        this.localNodeId = localNodeId;
        this.locks = locks;
        this.nodePaths = nodePaths;


        if (logger.isDebugEnabled()) {
            logger.debug("using node location [{}], local_node_id [{}]", nodePaths, localNodeId);
        }
        if (logger.isTraceEnabled()) {
            StringBuilder sb = new StringBuilder("node data locations details:\n");
            for (Path file : nodePaths) {
                sb.append(" -> ")
                        .append(file.toAbsolutePath())
                        .append(", free_space [")
                        .append(new ByteSizeValue(Files.getFileStore(file).getUnallocatedSpace()))
                        .append("], usable_space [")
                        .append(new ByteSizeValue(Files.getFileStore(file).getUsableSpace()))
                        .append("]\n");
            }
            logger.trace(sb.toString());
        }

        this.nodeIndicesPaths = new Path[nodePaths.length];
        for (int i = 0; i < nodePaths.length; i++) {
            nodeIndicesPaths[i] = nodePaths[i].resolve(INDICES_FOLDER);
        }
    }



    /**
     * Deletes a shard data directory iff the shards locks were successfully acquired.
     *
     * @param shardId the id of the shard to delete to delete
     * @throws IOException if an IOException occurs
     */
    public void deleteShardDirectorySafe(ShardId shardId, @IndexSettings Settings indexSettings) throws IOException {
        // This is to ensure someone doesn't use ImmutableSettings.EMPTY
        assert indexSettings != ImmutableSettings.EMPTY;
        final Path[] paths = shardPaths(shardId);
        logger.trace("deleting shard {} directory, paths: [{}]", shardId, paths);
        try (ShardLock lock = shardLock(shardId)) {
            deleteShardDirectoryUnderLock(lock, indexSettings);
        }
    }<commit_msg>Core: log path.data filesystem details to INFO level<commit_after>/*
 * Licensed to Elasticsearch under one or more contributor
 * license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright
 * ownership. Elasticsearch licenses this file to you 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
 *
 * 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.
 */

package org.elasticsearch.env;

import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Sets;
import com.google.common.primitives.Ints;
import org.apache.lucene.store.*;
import org.apache.lucene.util.IOUtils;
import org.elasticsearch.ElasticsearchIllegalArgumentException;
import org.elasticsearch.ElasticsearchIllegalStateException;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.common.Nullable;
import org.elasticsearch.common.component.AbstractComponent;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.io.FileSystemUtils;
import org.elasticsearch.common.settings.ImmutableSettings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.index.Index;
import org.elasticsearch.index.settings.IndexSettings;
import org.elasticsearch.index.shard.ShardId;

import java.io.Closeable;
import java.io.IOException;
import java.nio.file.*;
import java.util.*;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * A component that holds all data paths for a single node.
 */
public class NodeEnvironment extends AbstractComponent implements Closeable{

    /* ${data.paths}/nodes/{node.id} */
    private final Path[] nodePaths;
    /* ${data.paths}/nodes/{node.id}/indices */
    private final Path[] nodeIndicesPaths;
    private final Lock[] locks;
    private final boolean addNodeId;

    private final int localNodeId;
    private final AtomicBoolean closed = new AtomicBoolean(false);
    private final Map<ShardId, InternalShardLock> shardLocks = new HashMap<>();

    private final boolean customPathsEnabled;

    // Setting to automatically append node id to custom data paths
    public static final String ADD_NODE_ID_TO_CUSTOM_PATH = "node.add_id_to_custom_path";
    // Setting to enable custom index.data_path setting for new indices
    public static final String SETTING_CUSTOM_DATA_PATH_ENABLED = "node.enable_custom_paths";

    public static final String NODES_FOLDER = "nodes";
    public static final String INDICES_FOLDER = "indices";
    public static final String NODE_LOCK_FILENAME = "node.lock";

    @Inject
    public NodeEnvironment(Settings settings, Environment environment) throws IOException {
        super(settings);

        this.addNodeId = settings.getAsBoolean(ADD_NODE_ID_TO_CUSTOM_PATH, true);
        this.customPathsEnabled = settings.getAsBoolean(SETTING_CUSTOM_DATA_PATH_ENABLED, false);

        if (!DiscoveryNode.nodeRequiresLocalStorage(settings)) {
            nodePaths = null;
            nodeIndicesPaths = null;
            locks = null;
            localNodeId = -1;
            return;
        }

        final Path[] nodePaths = new Path[environment.dataWithClusterFiles().length];
        final Lock[] locks = new Lock[environment.dataWithClusterFiles().length];
        int localNodeId = -1;
        IOException lastException = null;
        int maxLocalStorageNodes = settings.getAsInt("node.max_local_storage_nodes", 50);
        for (int possibleLockId = 0; possibleLockId < maxLocalStorageNodes; possibleLockId++) {
            for (int dirIndex = 0; dirIndex < environment.dataWithClusterFiles().length; dirIndex++) {
                Path dir = environment.dataWithClusterFiles()[dirIndex].resolve(Paths.get(NODES_FOLDER, Integer.toString(possibleLockId)));
                Files.createDirectories(dir);
                
                try (Directory luceneDir = FSDirectory.open(dir, NativeFSLockFactory.INSTANCE)) {
                    logger.trace("obtaining node lock on {} ...", dir.toAbsolutePath());
                    Lock tmpLock = luceneDir.makeLock(NODE_LOCK_FILENAME);
                    boolean obtained = tmpLock.obtain();
                    if (obtained) {
                        locks[dirIndex] = tmpLock;
                        nodePaths[dirIndex] = dir;
                        localNodeId = possibleLockId;
                    } else {
                        logger.trace("failed to obtain node lock on {}", dir.toAbsolutePath());
                        // release all the ones that were obtained up until now
                        for (int i = 0; i < locks.length; i++) {
                            if (locks[i] != null) {
                                IOUtils.closeWhileHandlingException(locks[i]);
                            }
                            locks[i] = null;
                        }
                        break;
                    }
                } catch (IOException e) {
                    logger.trace("failed to obtain node lock on {}", e, dir.toAbsolutePath());
                    lastException = new IOException("failed to obtain lock on " + dir.toAbsolutePath(), e);
                    // release all the ones that were obtained up until now
                    for (int i = 0; i < locks.length; i++) {
                        IOUtils.closeWhileHandlingException(locks[i]);
                        locks[i] = null;
                    }
                    break;
                }
            }
            if (locks[0] != null) {
                // we found a lock, break
                break;
            }
        }
        if (locks[0] == null) {
            throw new ElasticsearchIllegalStateException("Failed to obtain node lock, is the following location writable?: "
                    + Arrays.toString(environment.dataWithClusterFiles()), lastException);
        }

        this.localNodeId = localNodeId;
        this.locks = locks;
        this.nodePaths = nodePaths;


        if (logger.isDebugEnabled()) {
            logger.debug("using node location [{}], local_node_id [{}]", nodePaths, localNodeId);
        }

        // We do some I/O in here, so skip it if INFO is not enabled:
        if (logger.isInfoEnabled()) {
            StringBuilder sb = new StringBuilder("node data locations details:\n");
            for (Path file : nodePaths) {
                // NOTE: FSDirectory.open creates the directory up above so it will exist here:
                sb.append(" -> ").append(file.toAbsolutePath());
                try {
                    FileStore fileStore = getFileStore(file);
                    boolean spins = IOUtils.spins(file);
                        sb.append(", free_space [")
                          .append(new ByteSizeValue(fileStore.getUnallocatedSpace()))
                          .append("], usable_space [")
                          .append(new ByteSizeValue(fileStore.getUsableSpace()))
                          .append("], total_space [")
                          .append(new ByteSizeValue(fileStore.getTotalSpace()))
                          .append("], spins? [")
                          .append(spins ? "possibly" : "no")
                          .append("], mount [")
                          .append(fileStore)
                          .append("], type [")
                          .append(fileStore.type())
                          .append(']');
                } catch (Exception e) {
                    sb.append(", ignoring exception gathering filesystem details: " + e);
                }
                sb.append('\n');
            }
            logger.info(sb.toString());
        }

        this.nodeIndicesPaths = new Path[nodePaths.length];
        for (int i = 0; i < nodePaths.length; i++) {
            nodeIndicesPaths[i] = nodePaths[i].resolve(INDICES_FOLDER);
        }
    }

    // NOTE: poached from Lucene's IOUtils:

    // Files.getFileStore(Path) useless here!
    // don't complain, just try it yourself
    static FileStore getFileStore(Path path) throws IOException {
        FileStore store = Files.getFileStore(path);
        String mount = getMountPoint(store);

        // find the "matching" FileStore from system list, it's the one we want.
        for (FileStore fs : path.getFileSystem().getFileStores()) {
            if (mount.equals(getMountPoint(fs))) {
                return fs;
            }
        }

        // fall back to crappy one we got from Files.getFileStore
        return store;    
    }

    // NOTE: poached from Lucene's IOUtils:

    // these are hacks that are not guaranteed
    static String getMountPoint(FileStore store) {
        String desc = store.toString();
        return desc.substring(0, desc.lastIndexOf('(') - 1);
    }

    /**
     * Deletes a shard data directory iff the shards locks were successfully acquired.
     *
     * @param shardId the id of the shard to delete to delete
     * @throws IOException if an IOException occurs
     */
    public void deleteShardDirectorySafe(ShardId shardId, @IndexSettings Settings indexSettings) throws IOException {
        // This is to ensure someone doesn't use ImmutableSettings.EMPTY
        assert indexSettings != ImmutableSettings.EMPTY;
        final Path[] paths = shardPaths(shardId);
        logger.trace("deleting shard {} directory, paths: [{}]", shardId, paths);
        try (ShardLock lock = shardLock(shardId)) {
            deleteShardDirectoryUnderLock(lock, indexSettings);
        }
    }
{'fraction_non_alphanumeric': 0.09328175230207596, 'fraction_numerical': 0.0017435841551789897, 'mean_word_length': 2.28453829634932, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6957570', 'n_tokens_mistral': 4696, 'n_tokens_neox': 4602, 'n_words': 1290}