[7.17] Fix _cluster/stats .nodes.fs deduplication (#94843)

joegallo · web-flow · commit ad68908f4475 · 2023-03-28T18:01:05.000-04:00
* Refactor _cluster/stats .nodes.fs deduplication (#94780) * Fix FsInfo device deduplication (#94744) * Fix _cluster/stats .nodes.fs deduplication (#94798)
diff --git a/docs/changelog/94744.yaml b/docs/changelog/94744.yaml
@@ -0,0 +1,5 @@
+pr: 94744
+summary: Fix `FsInfo` device deduplication
+area: Stats
+type: bug
+issues: []
diff --git a/docs/changelog/94798.yaml b/docs/changelog/94798.yaml
@@ -0,0 +1,6 @@
+pr: 94798
+summary: Fix _cluster/stats `.nodes.fs` deduplication
+area: Stats
+type: bug
+issues:
+ - 24472
diff --git a/server/src/main/java/org/elasticsearch/action/admin/cluster/stats/ClusterStatsNodes.java b/server/src/main/java/org/elasticsearch/action/admin/cluster/stats/ClusterStatsNodes.java
@@ -41,6 +41,7 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 import java.util.SortedMap;
 import java.util.TreeMap;
@@ -63,10 +64,10 @@ public class ClusterStatsNodes implements ToXContentFragment {
 
     ClusterStatsNodes(List<ClusterStatsNodeResponse> nodeResponses) {
         this.versions = new HashSet<>();
-        this.fs = new FsInfo.Path();
         this.plugins = new HashSet<>();
 
-        Set<InetAddress> seenAddresses = new HashSet<>(nodeResponses.size());
+        ClusterFsStatsDeduplicator deduplicator = new ClusterFsStatsDeduplicator(nodeResponses.size());
+
         List<NodeInfo> nodeInfos = new ArrayList<>(nodeResponses.size());
         List<NodeStats> nodeStats = new ArrayList<>(nodeResponses.size());
         for (ClusterStatsNodeResponse nodeResponse : nodeResponses) {
@@ -75,16 +76,12 @@ public class ClusterStatsNodes implements ToXContentFragment {
             this.versions.add(nodeResponse.nodeInfo().getVersion());
             this.plugins.addAll(nodeResponse.nodeInfo().getInfo(PluginsAndModules.class).getPluginInfos());
 
-            // now do the stats that should be deduped by hardware (implemented by ip deduping)
             TransportAddress publishAddress = nodeResponse.nodeInfo().getInfo(TransportInfo.class).address().publishAddress();
             final InetAddress inetAddress = publishAddress.address().getAddress();
-            if (seenAddresses.add(inetAddress) == false) {
-                continue;
-            }
-            if (nodeResponse.nodeStats().getFs() != null) {
-                this.fs.add(nodeResponse.nodeStats().getFs().getTotal());
-            }
+            deduplicator.add(inetAddress, nodeResponse.nodeStats().getFs());
         }
+        this.fs = deduplicator.getTotal();
+
         this.counts = new Counts(nodeInfos);
         this.os = new OsStats(nodeInfos, nodeStats);
         this.process = new ProcessStats(nodeStats);
@@ -764,4 +761,75 @@ public XContentBuilder toXContent(final XContentBuilder builder, final Params pa
 
     }
 
+    static class ClusterFsStatsDeduplicator {
+
+        private static class DedupEntry {
+
+            private final InetAddress inetAddress;
+            private final String mount;
+            private final String path;
+
+            DedupEntry(InetAddress inetAddress, String mount, String path) {
+                this.inetAddress = inetAddress;
+                this.mount = mount;
+                this.path = path;
+            }
+
+            @Override
+            public boolean equals(Object o) {
+                if (this == o) return true;
+                if (o == null || getClass() != o.getClass()) return false;
+                DedupEntry that = (DedupEntry) o;
+                return Objects.equals(inetAddress, that.inetAddress)
+                    && Objects.equals(mount, that.mount)
+                    && Objects.equals(path, that.path);
+            }
+
+            @Override
+            public int hashCode() {
+                return Objects.hash(inetAddress, mount, path);
+            }
+        }
+
+        private final Set<DedupEntry> seenAddressesMountsPaths;
+
+        private final FsInfo.Path total = new FsInfo.Path();
+
+        ClusterFsStatsDeduplicator(int expectedSize) {
+            // each address+mount is stored twice (once without a path, and once with a path), thus 2x
+            seenAddressesMountsPaths = new HashSet<>(2 * expectedSize);
+        }
+
+        public void add(InetAddress inetAddress, FsInfo fsInfo) {
+            if (fsInfo != null) {
+                for (FsInfo.Path p : fsInfo) {
+                    final String mount = p.getMount();
+                    final String path = p.getPath();
+
+                    // this deduplication logic is hard to get right. it might be impossible to make it work correctly in
+                    // *all* circumstances. this is best-effort only, but it's aimed at trying to solve 90%+ of cases.
+
+                    // rule 0: we want to sum the unique mounts for each ip address, so if we *haven't* seen a particular
+                    // address and mount, then definitely add that to the total.
+
+                    // rule 1: however, as a special case, if we see the same address+mount+path triple more than once, then we
+                    // override the ip+mount de-duplication logic -- using that as indicator that we're seeing a special
+                    // containerization situation, in which case we assume the operator is maintaining different disks for each node.
+
+                    boolean seenAddressMount = seenAddressesMountsPaths.add(new DedupEntry(inetAddress, mount, null)) == false;
+                    boolean seenAddressMountPath = seenAddressesMountsPaths.add(new DedupEntry(inetAddress, mount, path)) == false;
+
+                    if ((seenAddressMount == false) || seenAddressMountPath) {
+                        total.add(p);
+                    }
+                }
+            }
+        }
+
+        public FsInfo.Path getTotal() {
+            FsInfo.Path result = new FsInfo.Path();
+            result.add(total);
+            return result;
+        }
+    }
 }
diff --git a/server/src/main/java/org/elasticsearch/monitor/fs/FsInfo.java b/server/src/main/java/org/elasticsearch/monitor/fs/FsInfo.java
@@ -14,12 +14,12 @@
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.unit.ByteSizeValue;
-import org.elasticsearch.core.Nullable;
 import org.elasticsearch.xcontent.ToXContentFragment;
 import org.elasticsearch.xcontent.ToXContentObject;
 import org.elasticsearch.xcontent.XContentBuilder;
 
 import java.io.IOException;
+import java.nio.file.FileStore;
 import java.util.Arrays;
 import java.util.HashSet;
 import java.util.Iterator;
@@ -30,18 +30,17 @@ public class FsInfo implements Iterable<FsInfo.Path>, Writeable, ToXContentFragm
     public static class Path implements Writeable, ToXContentObject {
 
         String path;
-        @Nullable
-        String mount;
-        /** File system type from {@code java.nio.file.FileStore type()}, if available. */
-        @Nullable
-        String type;
+        /** File system string from {@link FileStore#toString()}. The concrete subclasses of FileStore have meaningful toString methods. */
+        String mount; // e.g. "/app (/dev/mapper/lxc-data)", "/System/Volumes/Data (/dev/disk1s2)", "Local Disk (C:)", etc.
+        /** File system type from {@link FileStore#type()}. */
+        String type; // e.g. "xfs", "apfs", "NTFS", etc.
         long total = -1;
         long free = -1;
         long available = -1;
 
         public Path() {}
 
-        public Path(String path, @Nullable String mount, long total, long free, long available) {
+        public Path(String path, String mount, long total, long free, long available) {
             this.path = path;
             this.mount = mount;
             this.total = total;
@@ -53,7 +52,7 @@ public Path(String path, @Nullable String mount, long total, long free, long ava
          * Read from a stream.
          */
         public Path(StreamInput in) throws IOException {
-            path = in.readOptionalString();
+            path = in.readOptionalString(); // total aggregates do not have a path, mount, or type
             mount = in.readOptionalString();
             type = in.readOptionalString();
             total = in.readLong();
@@ -66,7 +65,7 @@ public Path(StreamInput in) throws IOException {
 
         @Override
         public void writeTo(StreamOutput out) throws IOException {
-            out.writeOptionalString(path); // total aggregates do not have a path
+            out.writeOptionalString(path); // total aggregates do not have a path, mount, or type
             out.writeOptionalString(mount);
             out.writeOptionalString(type);
             out.writeLong(total);
@@ -504,8 +503,8 @@ private Path total() {
         Path res = new Path();
         Set<String> seenDevices = new HashSet<>(paths.length);
         for (Path subPath : paths) {
-            if (subPath.path != null) {
-                if (seenDevices.add(subPath.path) == false) {
+            if (subPath.mount != null) {
+                if (seenDevices.add(subPath.mount) == false) {
                     continue; // already added numbers for this device;
                 }
             }
diff --git a/server/src/main/java/org/elasticsearch/monitor/fs/FsProbe.java b/server/src/main/java/org/elasticsearch/monitor/fs/FsProbe.java
@@ -135,7 +135,7 @@ public static FsInfo.Path getFSInfo(DataPath dataPath) throws IOException {
         FsInfo.Path fsPath = new FsInfo.Path();
         fsPath.path = dataPath.path.toString();
 
-        // NOTE: we use already cached (on node startup) FileStore and spins
+        // NOTE: we use already cached (on node startup) FileStore
         // since recomputing these once per second (default) could be costly,
         // and they should not change:
         fsPath.total = getTotal(dataPath.fileStore);
diff --git a/server/src/test/java/org/elasticsearch/action/admin/cluster/stats/ClusterStatsNodesTests.java b/server/src/test/java/org/elasticsearch/action/admin/cluster/stats/ClusterStatsNodesTests.java
@@ -12,11 +12,14 @@
 import org.elasticsearch.action.admin.cluster.node.stats.NodeStats;
 import org.elasticsearch.action.admin.cluster.node.stats.NodeStatsTests;
 import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.common.network.InetAddresses;
 import org.elasticsearch.common.network.NetworkModule;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.monitor.fs.FsInfo;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xcontent.XContentType;
 
+import java.net.InetAddress;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.Iterator;
@@ -121,6 +124,128 @@ public void testIngestStats() throws Exception {
         );
     }
 
+    public void testClusterFsStatsDeduplicator() {
+        {
+            // single node, multiple data paths, different devices
+            InetAddress address1 = InetAddresses.forString("192.168.0.1");
+            FsInfo.Path path1 = new FsInfo.Path("/a", "/dev/sda", 3, 2, 1);
+            FsInfo.Path path2 = new FsInfo.Path("/b", "/dev/sdb", 3, 2, 1);
+            ClusterStatsNodes.ClusterFsStatsDeduplicator deduplicator = new ClusterStatsNodes.ClusterFsStatsDeduplicator(1);
+            deduplicator.add(address1, newFsInfo(path1, path2));
+            FsInfo.Path total = deduplicator.getTotal();
+
+            // since they're different devices, they sum
+            assertThat(total.getTotal().getBytes(), equalTo(6L));
+            assertThat(total.getFree().getBytes(), equalTo(4L));
+            assertThat(total.getAvailable().getBytes(), equalTo(2L));
+        }
+
+        {
+            // single node, multiple data paths, same device
+            InetAddress address1 = InetAddresses.forString("192.168.0.1");
+            FsInfo.Path path1 = new FsInfo.Path("/data/a", "/dev/sda", 3, 2, 1);
+            FsInfo.Path path2 = new FsInfo.Path("/data/b", "/dev/sda", 3, 2, 1);
+            ClusterStatsNodes.ClusterFsStatsDeduplicator deduplicator = new ClusterStatsNodes.ClusterFsStatsDeduplicator(1);
+            deduplicator.add(address1, newFsInfo(path1, path2));
+            FsInfo.Path total = deduplicator.getTotal();
+
+            // since it's the same device, they don't sum, we just see the one
+            assertThat(total.getTotal().getBytes(), equalTo(3L));
+            assertThat(total.getFree().getBytes(), equalTo(2L));
+            assertThat(total.getAvailable().getBytes(), equalTo(1L));
+        }
+
+        {
+            // two nodes, same ip address, but different data paths on different devices
+            InetAddress address1 = InetAddresses.forString("192.168.0.1");
+            FsInfo.Path path1 = new FsInfo.Path("/data/a", "/dev/sda", 3, 2, 1);
+            InetAddress address2 = InetAddresses.forString("192.168.0.1");
+            FsInfo.Path path2 = new FsInfo.Path("/data/b", "/dev/sdb", 3, 2, 1);
+            ClusterStatsNodes.ClusterFsStatsDeduplicator deduplicator = new ClusterStatsNodes.ClusterFsStatsDeduplicator(1);
+            deduplicator.add(address1, newFsInfo(path1));
+            deduplicator.add(address2, newFsInfo(path2));
+            FsInfo.Path total = deduplicator.getTotal();
+
+            // since they're different devices, they sum
+            assertThat(total.getTotal().getBytes(), equalTo(6L));
+            assertThat(total.getFree().getBytes(), equalTo(4L));
+            assertThat(total.getAvailable().getBytes(), equalTo(2L));
+        }
+
+        {
+            // two nodes, different ip addresses, different data paths, same device
+            InetAddress address1 = InetAddresses.forString("192.168.0.1");
+            FsInfo.Path path1 = new FsInfo.Path("/data/a", "/dev/sda", 3, 2, 1);
+            InetAddress address2 = InetAddresses.forString("192.168.0.2");
+            FsInfo.Path path2 = new FsInfo.Path("/data/b", "/dev/sda", 3, 2, 1);
+            ClusterStatsNodes.ClusterFsStatsDeduplicator deduplicator = new ClusterStatsNodes.ClusterFsStatsDeduplicator(1);
+            deduplicator.add(address1, newFsInfo(path1));
+            deduplicator.add(address2, newFsInfo(path2));
+            FsInfo.Path total = deduplicator.getTotal();
+
+            // it's the same device, yeah, but on entirely different machines, so they sum
+            assertThat(total.getTotal().getBytes(), equalTo(6L));
+            assertThat(total.getFree().getBytes(), equalTo(4L));
+            assertThat(total.getAvailable().getBytes(), equalTo(2L));
+        }
+
+        {
+            // two nodes, same ip address, same data path, same device
+            InetAddress address1 = InetAddresses.forString("192.168.0.1");
+            FsInfo.Path path1 = new FsInfo.Path("/app/data", "/app (/dev/mapper/lxc-data)", 3, 2, 1);
+            InetAddress address2 = InetAddresses.forString("192.168.0.1");
+            FsInfo.Path path2 = new FsInfo.Path("/app/data", "/app (/dev/mapper/lxc-data)", 3, 2, 1);
+            ClusterStatsNodes.ClusterFsStatsDeduplicator deduplicator = new ClusterStatsNodes.ClusterFsStatsDeduplicator(1);
+            deduplicator.add(address1, newFsInfo(path1));
+            deduplicator.add(address2, newFsInfo(path2));
+            FsInfo.Path total = deduplicator.getTotal();
+
+            // wait a second, this is the super-special case -- you can't actually have two nodes doing this unless something
+            // very interesting is happening, so they sum (i.e. we assume the operator is doing smart things)
+            assertThat(total.getTotal().getBytes(), equalTo(6L));
+            assertThat(total.getFree().getBytes(), equalTo(4L));
+            assertThat(total.getAvailable().getBytes(), equalTo(2L));
+        }
+
+        {
+            // two nodes, same ip address, different data paths, same device
+            InetAddress address1 = InetAddresses.forString("192.168.0.1");
+            FsInfo.Path path1 = new FsInfo.Path("/app/data1", "/dev/sda", 3, 2, 1);
+            InetAddress address2 = InetAddresses.forString("192.168.0.1");
+            FsInfo.Path path2 = new FsInfo.Path("/app/data2", "/dev/sda", 3, 2, 1);
+            ClusterStatsNodes.ClusterFsStatsDeduplicator deduplicator = new ClusterStatsNodes.ClusterFsStatsDeduplicator(1);
+            deduplicator.add(address1, newFsInfo(path1));
+            deduplicator.add(address2, newFsInfo(path2));
+            FsInfo.Path total = deduplicator.getTotal();
+
+            // since the paths aren't the same, it doesn't trigger the special case -- it's just the same device and doesn't sum
+            assertThat(total.getTotal().getBytes(), equalTo(3L));
+            assertThat(total.getFree().getBytes(), equalTo(2L));
+            assertThat(total.getAvailable().getBytes(), equalTo(1L));
+        }
+
+        {
+            // two nodes, same ip address, same data path, different devices
+            InetAddress address1 = InetAddresses.forString("192.168.0.1");
+            FsInfo.Path path1 = new FsInfo.Path("/app/data", "/dev/sda", 3, 2, 1);
+            InetAddress address2 = InetAddresses.forString("192.168.0.1");
+            FsInfo.Path path2 = new FsInfo.Path("/app/data", "/dev/sdb", 3, 2, 1);
+            ClusterStatsNodes.ClusterFsStatsDeduplicator deduplicator = new ClusterStatsNodes.ClusterFsStatsDeduplicator(1);
+            deduplicator.add(address1, newFsInfo(path1));
+            deduplicator.add(address2, newFsInfo(path2));
+            FsInfo.Path total = deduplicator.getTotal();
+
+            // having the same path isn't special in this case, it's just unique ip/mount pairs, so they sum
+            assertThat(total.getTotal().getBytes(), equalTo(6L));
+            assertThat(total.getFree().getBytes(), equalTo(4L));
+            assertThat(total.getAvailable().getBytes(), equalTo(2L));
+        }
+    }
+
+    private static FsInfo newFsInfo(FsInfo.Path... paths) {
+        return new FsInfo(-1, null, paths);
+    }
+
     private static NodeInfo createNodeInfo(String nodeId, String transportType, String httpType) {
         Settings.Builder settings = Settings.builder();
         if (transportType != null) {
diff --git a/server/src/test/java/org/elasticsearch/monitor/fs/FsInfoTests.java b/server/src/test/java/org/elasticsearch/monitor/fs/FsInfoTests.java
@@ -0,0 +1,40 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.monitor.fs;
+
+import org.elasticsearch.monitor.fs.FsInfo.Path;
+import org.elasticsearch.test.ESTestCase;
+
+import static org.hamcrest.Matchers.is;
+
+public class FsInfoTests extends ESTestCase {
+
+    public void testTotalSummation() {
+        Path p1 = new Path("J:\\data\\nodes\\0", "Disk10 (J:)", 4, 3, 2);
+        Path p2 = new Path("K:\\data\\nodes\\0", "Disk11 (K:)", 8, 6, 4);
+        FsInfo info = new FsInfo(-1, null, new Path[] { p1, p2 });
+
+        Path total = info.getTotal();
+        assertThat(total.getTotal().getBytes(), is(12L));
+        assertThat(total.getFree().getBytes(), is(9L));
+        assertThat(total.getAvailable().getBytes(), is(6L));
+    }
+
+    public void testTotalDeduplication() {
+        Path p1 = new Path("/app/data/es-1", "/app (/dev/sda1)", 8, 6, 4);
+        Path p2 = new Path("/app/data/es-2", "/app (/dev/sda1)", 8, 6, 4);
+        FsInfo info = new FsInfo(-1, null, new Path[] { p1, p2 });
+
+        Path total = info.getTotal();
+        assertThat(total.getTotal().getBytes(), is(8L));
+        assertThat(total.getFree().getBytes(), is(6L));
+        assertThat(total.getAvailable().getBytes(), is(4L));
+    }
+
+}
diff --git a/x-pack/plugin/monitoring/src/test/java/org/elasticsearch/xpack/monitoring/collector/cluster/ClusterStatsMonitoringDocTests.java b/x-pack/plugin/monitoring/src/test/java/org/elasticsearch/xpack/monitoring/collector/cluster/ClusterStatsMonitoringDocTests.java
