Introduce quota-aware filesystem ES plugin (#64331)

Partial backport of #63620. Closes #61145.

This PR adds a quota-aware filesystem plugin to Elasticsearch. This plugin
offers a way to provide user quota limits (specifically, total quota size
and available quota size) to Elasticsearch, in an implementation-agnostic
manner.

distribution/src/bin/elasticsearch

@@ -20,6 +20,14 @@ ES_JVM_OPTIONS="$ES_PATH_CONF"/jvm.options
 JVM_OPTIONS=`"$JAVA" -cp "$ES_CLASSPATH" org.elasticsearch.tools.launchers.JvmOptionsParser "$ES_JVM_OPTIONS"`
 ES_JAVA_OPTS="${JVM_OPTIONS//\$\{ES_TMPDIR\}/$ES_TMPDIR} $ES_JAVA_OPTS"
 
+# If the quote-aware filesystem plugin is installed, then we need to pass extra
+# flags to Java in order to use the plugin.
+if [[ -d "$ES_HOME/plugins/quota-aware-fs" ]]; then
+  PLUGIN_JAR_PATH="$(find "$ES_HOME/plugins/quota-aware-fs" -name quota-aware-fs\*.jar)"
+  ES_JAVA_OPTS="$ES_JAVA_OPTS -Djava.nio.file.spi.DefaultFileSystemProvider=org.elasticsearch.fs.quotaaware.QuotaAwareFileSystemProvider"
+  ES_JAVA_OPTS="$ES_JAVA_OPTS -Xbootclasspath/a:$PLUGIN_JAR_PATH"
+fi
+
 cd "$ES_HOME"
 # manual parsing to find out, if process should be detached
 if ! echo $* | grep -E '(^-d |-d$| -d |--daemonize$|--daemonize )' > /dev/null; then

docs/build.gradle

@@ -70,8 +70,10 @@ integTestCluster {
 // build the cluster with all plugins
 project.rootProject.subprojects.findAll { it.parent.path == ':plugins' }.each { subproj ->
   /* Skip repositories. We just aren't going to be able to test them so it
-   * doesn't make sense to waste time installing them. */
-  if (subproj.path.startsWith(':plugins:repository-')) {
+   * doesn't make sense to waste time installing them.
+   * Also skip quota-aware-fs since it has to be configured in order to use
+   * it, otherwise ES will not start. */
+  if (subproj.path.startsWith(':plugins:repository-') || subproj.path.startsWith(':plugins:quota-aware-fs')) {
     return
   }
   subproj.afterEvaluate { // need to wait until the project has been configured

plugins/quota-aware-fs/build.gradle

@@ -0,0 +1,6 @@
+esplugin {
+  description 'A bootstrap plugin that adds support for interfacing with filesystem that enforce user quotas.'
+  classname '<none>'
+}
+
+integTest.enabled = false

plugins/quota-aware-fs/src/main/java/org/elasticsearch/fs/quotaaware/QuotaAwareFileStore.java

@@ -0,0 +1,98 @@
+/*
+ * 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.fs.quotaaware;
+
+import java.io.IOException;
+import java.nio.file.FileStore;
+import java.nio.file.attribute.FileAttributeView;
+import java.nio.file.attribute.FileStoreAttributeView;
+
+/**
+ * An implementation of {@link FileStore} that relies on
+ * {@link QuotaAwareFileSystemProvider} for usage reporting. Other methods are
+ * delegated to the backing instance of {@link FileStore}
+ */
+public final class QuotaAwareFileStore extends FileStore {
+
+    private final FileStore backingFS;
+    private final QuotaAwareFileSystemProvider provider;
+
+    QuotaAwareFileStore(QuotaAwareFileSystemProvider provider, FileStore backingFS) {
+        this.provider = provider;
+        this.backingFS = backingFS;
+    }
+
+    @Override
+    public String name() {
+        return backingFS.name();
+    }
+
+    @Override
+    public String type() {
+        return backingFS.type();
+    }
+
+    @Override
+    public boolean isReadOnly() {
+        return backingFS.isReadOnly();
+    }
+
+    @Override
+    public long getTotalSpace() throws IOException {
+        return Math.min(provider.getTotal(), backingFS.getTotalSpace());
+    }
+
+    @Override
+    public long getUsableSpace() throws IOException {
+        return Math.min(provider.getRemaining(), backingFS.getUsableSpace());
+    }
+
+    @Override
+    public long getUnallocatedSpace() throws IOException {
+        // There is no point in telling users that the underlying
+        // host has more capacity than what they're allowed to use so we limit
+        // this one with remaining as well.
+        return Math.min(provider.getRemaining(), backingFS.getUnallocatedSpace());
+    }
+
+    @Override
+    public boolean supportsFileAttributeView(Class<? extends FileAttributeView> type) {
+        return backingFS.supportsFileAttributeView(type);
+    }
+
+    @Override
+    public boolean supportsFileAttributeView(String name) {
+        return backingFS.supportsFileAttributeView(name);
+    }
+
+    @Override
+    public <V extends FileStoreAttributeView> V getFileStoreAttributeView(Class<V> type) {
+        return backingFS.getFileStoreAttributeView(type);
+    }
+
+    @Override
+    public Object getAttribute(String attribute) throws IOException {
+        return backingFS.getAttribute(attribute);
+    }
+
+    @Override
+    public String toString() {
+        return "QuotaAwareFileStore(" + backingFS.toString() + ")";
+    }
+}

plugins/quota-aware-fs/src/main/java/org/elasticsearch/fs/quotaaware/QuotaAwareFileSystem.java

@@ -0,0 +1,144 @@
+/*
+ * 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.fs.quotaaware;
+
+import org.elasticsearch.common.SuppressForbidden;
+
+import java.io.IOException;
+import java.nio.file.FileStore;
+import java.nio.file.FileSystem;
+import java.nio.file.FileSystems;
+import java.nio.file.Path;
+import java.nio.file.PathMatcher;
+import java.nio.file.WatchService;
+import java.nio.file.attribute.UserPrincipalLookupService;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.StreamSupport;
+
+/**
+ * An implementation of {@link FileSystem} that returns the given
+ * {@link QuotaAwareFileSystemProvider} provider {@link #provider()}.
+ *
+ * Other methods are delegated to given instance of {@link FileSystem} and
+ * wrapped where result types are either @link {@link QuotaAwareFileSystem}
+ * or @link {@link QuotaAwareFileStore}.
+ *
+ */
+public final class QuotaAwareFileSystem extends FileSystem {
+    private final FileSystem delegate;
+    private final QuotaAwareFileSystemProvider provider;
+
+    QuotaAwareFileSystem(QuotaAwareFileSystemProvider provider, FileSystem delegate) {
+        this.provider = Objects.requireNonNull(provider, "Provider is required");
+        this.delegate = Objects.requireNonNull(delegate, "FileSystem is required");
+    }
+
+    @Override
+    public QuotaAwareFileSystemProvider provider() {
+        return provider;
+    }
+
+    @Override
+    @SuppressForbidden(reason = "accesses the default filesystem by design")
+    public void close() throws IOException {
+        if (this == FileSystems.getDefault()) {
+            throw new UnsupportedOperationException("The default file system cannot be closed");
+        } else if (delegate != FileSystems.getDefault()) {
+            delegate.close();
+        }
+        provider.purge(delegate);
+    }
+
+    @Override
+    public boolean isOpen() {
+        return delegate.isOpen();
+    }
+
+    @Override
+    public boolean isReadOnly() {
+        return delegate.isReadOnly();
+    }
+
+    @Override
+    public String getSeparator() {
+        return delegate.getSeparator();
+    }
+
+    @Override
+    public Iterable<Path> getRootDirectories() {
+        return StreamSupport.stream(delegate.getRootDirectories().spliterator(), false).map((Function<Path, Path>) this::wrap)::iterator;
+    }
+
+    @Override
+    public Iterable<FileStore> getFileStores() {
+        return StreamSupport.stream(delegate.getFileStores().spliterator(), false)
+            .map((Function<FileStore, FileStore>) provider::getFileStore)::iterator;
+    }
+
+    @Override
+    public Set<String> supportedFileAttributeViews() {
+        return delegate.supportedFileAttributeViews();
+    }
+
+    @Override
+    public Path getPath(String first, String... more) {
+        return wrap(delegate.getPath(first, more));
+    }
+
+    private QuotaAwarePath wrap(Path delegatePath) {
+        if (delegatePath == null) return null;
+        else return new QuotaAwarePath(this, delegatePath);
+    }
+
+    @Override
+    public PathMatcher getPathMatcher(String syntaxAndPattern) {
+        PathMatcher matcher = delegate.getPathMatcher(syntaxAndPattern);
+        return (path) -> matcher.matches(QuotaAwarePath.unwrap(path));
+    }
+
+    @Override
+    public UserPrincipalLookupService getUserPrincipalLookupService() {
+        return delegate.getUserPrincipalLookupService();
+    }
+
+    @Override
+    public WatchService newWatchService() throws IOException {
+        return delegate.newWatchService();
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj) return true;
+        if (obj == null) return false;
+        if (getClass() != obj.getClass()) return false;
+        QuotaAwareFileSystem other = (QuotaAwareFileSystem) obj;
+        if (!delegate.equals(other.delegate)) return false;
+        if (!provider.equals(other.provider)) return false;
+        return true;
+    }
+
+    @Override
+    public int hashCode() {
+        return delegate.hashCode();
+    }
+
+}