Merge pull request #34157 from vespa-engine/vekterli/content-node-message-stats

Track, aggregate and report content node response (network) errors

Author: vekterli

clustercontroller-apps/src/main/java/com/yahoo/vespa/clustercontroller/apps/clustercontroller/ClusterControllerClusterConfigurer.java

@@ -99,6 +99,7 @@ private static void configure(FleetControllerOptions.Builder builder, Fleetcontr
         builder.setClusterFeedBlockNoiseLevel(config.cluster_feed_block_noise_level());
         builder.setMaxNumberOfGroupsAllowedToBeDown(config.max_number_of_groups_allowed_to_be_down());
         builder.setIncludeDistributionConfigInClusterStateBundles(config.include_distribution_config_in_cluster_state_bundle());
+        builder.setAggregateContentNodeErrorReportsFromDistributors(config.aggregate_content_node_error_reports_from_distributors());
     }
 
     private static void configure(FleetControllerOptions.Builder builder, SlobroksConfig config) {

clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/AggregatedClusterStats.java

@@ -10,6 +10,8 @@ public interface AggregatedClusterStats {
 
     ContentClusterStats getStats();
 
+    ContentClusterErrorStats getErrorStats();
+
     ContentNodeStats getGlobalStats();
 
 }

clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStateView.java

@@ -71,7 +71,7 @@ static Set<Integer> getIndicesOfUpNodes(ClusterState clusterState, NodeType type
 
     public ClusterState getClusterState() { return clusterState; }
 
-    public void handleUpdatedHostInfo(NodeInfo node, HostInfo hostInfo) {
+    public void handleUpdatedHostInfo(NodeInfo node, HostInfo hostInfo, boolean aggregateErrorReports) {
         if ( ! node.isDistributor()) return;
 
         final int hostVersion;
@@ -85,6 +85,17 @@ public void handleUpdatedHostInfo(NodeInfo node, HostInfo hostInfo) {
         }
         int currentStateVersion = clusterState.getVersion();
 
+        if (aggregateErrorReports) {
+            var errorStats = StorageNodeStatsBridge.generateErrors(node.getNodeIndex(), hostInfo.getDistributor());
+            // Error statistics are always updated, even if the node has not ACKed the latest cluster
+            // state version. Otherwise, we will never be able to observe errors that prevent convergence!
+            statsAggregator.updateErrorStatsFromDistributor(node.getNodeIndex(), errorStats);
+        } else {
+            // If there's an edge for disabling aggregation, clear out any existing statistics.
+            // No-op if already disabled.
+            statsAggregator.clearAllErrorStatsFromDistributors();
+        }
+
         if (hostVersion != currentStateVersion) {
             // The distributor may be old (null), or the distributor may not have updated
             // to the latest state version just yet. We log here with fine, because it may
@@ -95,8 +106,12 @@ public void handleUpdatedHostInfo(NodeInfo node, HostInfo hostInfo) {
             return;
         }
 
-        statsAggregator.updateForDistributor(node.getNodeIndex(),
-                StorageNodeStatsBridge.generate(hostInfo.getDistributor()));
+        var stats = StorageNodeStatsBridge.generate(hostInfo.getDistributor());
+        statsAggregator.updateForDistributor(node.getNodeIndex(), stats);
+    }
+
+    public void handleUpdatedHostInfo(NodeInfo node, HostInfo hostInfo) {
+        handleUpdatedHostInfo(node, hostInfo, false);
     }
 
     public ClusterStatsAggregator getStatsAggregator() {

clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsAggregator.java

@@ -33,11 +33,18 @@ public class ClusterStatsAggregator {
     // Maps the distributor node index to a map of content node index to the
     // content node's stats.
     private final Map<Integer, ContentClusterStats> distributorToStats = new HashMap<>();
+    private final Map<Integer, ContentClusterErrorStats> distributorToErrorStats = new HashMap<>();
 
     // This is only needed as an optimization. Is just the sum of distributorToStats' ContentClusterStats.
     // Maps the content node index to the content node stats for that node.
     // This MUST be kept up-to-date with distributorToStats;
     private final ContentClusterStats aggregatedStats;
+    // Aggregate of all error reports received from all distributors for all content nodes.
+    // Makes it cheap to determine which distributors, if any, have observed errors towards
+    // any given content node in the current cluster state. This conceptually mirrors
+    // `aggregatedStats`, but with an extra internal indirection that is expected to be
+    // very sparse in a healthy system.
+    private final ContentClusterErrorStats aggregatedErrorStats;
     // This is the aggregate of aggregates across content nodes, allowing a reader to
     // get a O(1) view of all merges pending in the cluster.
     private final ContentNodeStats globallyAggregatedNodeStats = new ContentNodeStats(-1);
@@ -49,6 +56,7 @@ public class ClusterStatsAggregator {
         this.distributors = distributors;
         nonUpdatedDistributors = new HashSet<>(distributors);
         aggregatedStats = new ContentClusterStats(storageNodes);
+        aggregatedErrorStats = new ContentClusterErrorStats(storageNodes);
     }
 
     public AggregatedClusterStats getAggregatedStats() {
@@ -64,6 +72,11 @@ public ContentClusterStats getStats() {
                 return aggregatedStats;
             }
 
+            @Override
+            public ContentClusterErrorStats getErrorStats() {
+                return aggregatedErrorStats;
+            }
+
             @Override
             public ContentNodeStats getGlobalStats() {
                 return globallyAggregatedNodeStats;
@@ -105,6 +118,37 @@ void updateForDistributor(int distributorIndex, ContentClusterStats clusterStats
         addStatsFromDistributor(distributorIndex, clusterStats);
     }
 
+    void updateErrorStatsFromDistributor(int distributorIndex, ContentClusterErrorStats updatedErrorStats) {
+        if (!distributors.contains(distributorIndex)) {
+            return;
+        }
+        // This does _not_ remove the distributor from `nonUpdatedDistributors` since this
+        // method will be called even if the distributor has not yet converged to the newest
+        // cluster state.
+        // It also does not implicitly put an entry into the `distributorToStats` mapping; error
+        // stats are kept entirely separate.
+        ContentClusterErrorStats prevErrorStats = distributorToErrorStats.put(distributorIndex, updatedErrorStats);
+        for (var contentNodeEntry : aggregatedErrorStats.contentNodeErrorStats().entrySet()) {
+            var nodeIndex   = contentNodeEntry.getKey();
+            var contentNode = contentNodeEntry.getValue();
+            ContentNodeErrorStats statsToAdd = updatedErrorStats.getNodeErrorStats(nodeIndex);
+            if (statsToAdd != null) {
+                contentNode.addErrorStatsFrom(distributorIndex, statsToAdd);
+            }
+            if (prevErrorStats != null) {
+                ContentNodeErrorStats statsToSubtract = prevErrorStats.getNodeErrorStats(nodeIndex);
+                if (statsToSubtract != null) {
+                    contentNode.subtractErrorStatsFrom(distributorIndex, statsToSubtract);
+                }
+            }
+        }
+    }
+
+    void clearAllErrorStatsFromDistributors() {
+        distributorToErrorStats.clear();
+        aggregatedErrorStats.clearAllStats();
+    }
+
     private void addStatsFromDistributor(int distributorIndex, ContentClusterStats clusterStats) {
         ContentClusterStats prevClusterStats = distributorToStats.put(distributorIndex, clusterStats);
 

clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ContentClusterErrorStats.java

@@ -0,0 +1,41 @@
+// Copyright Vespa.ai. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.vespa.clustercontroller.core;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * Stores, for each content node, the aggregate error reports from all distributors in
+ * a content cluster, for a particular cluster state version (or attempt at reaching
+ * a state version in the case the cluster does not converge). The per-node statistics
+ * are sparse, so in a healthy cluster this is expected to be a lightweight mapping
+ * from all nodes -> empty ContentNodeErrorStats instances.
+ *
+ * @author vekterli
+ */
+public record ContentClusterErrorStats(Map<Integer, ContentNodeErrorStats> contentNodeErrorStats) {
+
+    private static Map<Integer, ContentNodeErrorStats> toEmptyErrorStats(Set<Integer> storageNodes) {
+        Map<Integer, ContentNodeErrorStats> contentNodeErrorStats = new HashMap<>(storageNodes.size());
+        for (Integer index : storageNodes) {
+            contentNodeErrorStats.put(index, new ContentNodeErrorStats(index));
+        }
+        return contentNodeErrorStats;
+    }
+
+    public ContentClusterErrorStats(Set<Integer> storageNodes) {
+        this(toEmptyErrorStats(storageNodes));
+    }
+
+    public ContentNodeErrorStats getNodeErrorStats(Integer index) {
+        return contentNodeErrorStats.get(index);
+    }
+
+    public void clearAllStats() {
+        contentNodeErrorStats.clear();
+    }
+
+}

clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ContentClusterStats.java

@@ -13,30 +13,26 @@
  *
  * @author hakonhall
  */
-public class ContentClusterStats implements Iterable<ContentNodeStats> {
-
-    private final long documentCountTotal;
-    private final long bytesTotal;
-    // Maps a content node index to the content node's stats.
-    private final Map<Integer, ContentNodeStats> mapToNodeStats;
-
-    public ContentClusterStats(long documentCountTotal, long bytesTotal, Set<Integer> storageNodes) {
-        this.documentCountTotal = documentCountTotal;
-        this.bytesTotal = bytesTotal;
-        mapToNodeStats = new HashMap<>(storageNodes.size());
+public record ContentClusterStats(long documentCountTotal,
+                                  long bytesTotal,
+                                  // Maps a content node index to the content node's stats.
+                                  Map<Integer, ContentNodeStats> mapToNodeStats)
+        implements Iterable<ContentNodeStats> {
+
+    private static Map<Integer, ContentNodeStats> toEmptyNodeStats(Set<Integer> storageNodes) {
+        Map<Integer, ContentNodeStats> mapToNodeStats = new HashMap<>(storageNodes.size());
         for (Integer index : storageNodes) {
             mapToNodeStats.put(index, new ContentNodeStats(index));
         }
+        return mapToNodeStats;
     }
 
-    public ContentClusterStats(Set<Integer> storageNodes) {
-        this(0, 0, storageNodes);
+    public ContentClusterStats(long documentCountTotal, long bytesTotal, Set<Integer> storageNodes) {
+        this(documentCountTotal, bytesTotal, toEmptyNodeStats(storageNodes));
     }
 
-    public ContentClusterStats(long documentCountTotal, long bytesTotal, Map<Integer, ContentNodeStats> mapToNodeStats) {
-        this.documentCountTotal = documentCountTotal;
-        this.bytesTotal = bytesTotal;
-        this.mapToNodeStats = mapToNodeStats;
+    public ContentClusterStats(Set<Integer> storageNodes) {
+        this(0, 0, storageNodes);
     }
 
     public ContentClusterStats(Map<Integer, ContentNodeStats> mapToNodeStats) {
@@ -57,21 +53,4 @@ public int size() {
         return mapToNodeStats.size();
     }
 
-    @Override
-    public boolean equals(Object o) {
-        if (this == o) return true;
-        if (o == null || getClass() != o.getClass()) return false;
-        ContentClusterStats that = (ContentClusterStats) o;
-        return Objects.equals(mapToNodeStats, that.mapToNodeStats);
-    }
-
-    @Override
-    public int hashCode() {
-        return Objects.hash(mapToNodeStats);
-    }
-
-    @Override
-    public String toString() {
-        return String.format("{mapToNodeStats=[%s]}", Arrays.toString(mapToNodeStats.entrySet().toArray()));
-    }
 }