Consider group membership when retiring host for OS upgrade

1 files changed, 64 insertions, 12 deletions

diff --git a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/os/RetiringOsUpgrader.java b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/os/RetiringOsUpgrader.java
index cb6c7683f23..a5ff7b82551 100644
--- a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/os/RetiringOsUpgrader.java
+++ b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/os/RetiringOsUpgrader.java
@@ -2,21 +2,32 @@
 package com.yahoo.vespa.hosted.provision.os;
 
 import com.yahoo.component.Version;
+import com.yahoo.config.provision.ClusterSpec;
 import com.yahoo.config.provision.NodeType;
 import com.yahoo.vespa.hosted.provision.Node;
 import com.yahoo.vespa.hosted.provision.NodeList;
 import com.yahoo.vespa.hosted.provision.NodeRepository;
 import com.yahoo.vespa.hosted.provision.node.Agent;
+import com.yahoo.vespa.hosted.provision.node.ClusterId;
 import com.yahoo.vespa.hosted.provision.node.filter.NodeListFilter;
 
 import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.logging.Logger;
+import java.util.stream.Collectors;
 
 /**
- * An upgrader that retires and deprovisions hosts on stale OS versions.
+ * An upgrader that retires and deprovisions hosts on stale OS versions. For hosts containing stateful clusters, this
+ * upgrader limits node retirement so that at most one group per cluster is affected at a time.
  *
- * Used in clouds where hosts must be re-provisioned to upgrade their OS.
+ * Used in clouds where the host configuration (e.g. local disk) requires re-provisioning to upgrade OS.
  *
  * @author mpolden
  */
@@ -35,8 +46,8 @@ public class RetiringOsUpgrader extends OsUpgrader {
     public void upgradeTo(OsVersionTarget target) {
         NodeList allNodes = nodeRepository.nodes().list();
         Instant now = nodeRepository.clock().instant();
-        for (var candidate : candidates(now, target, allNodes)) {
-            deprovision(candidate, target.version(), now);
+        for (Node host : deprovisionable(now, target, allNodes)) {
+            deprovision(host, target.version(), now);
         }
     }
 
@@ -45,18 +56,46 @@ public class RetiringOsUpgrader extends OsUpgrader {
         // No action needed in this implementation.
     }
 
-    /** Returns nodes that are candidates for upgrade */
-    private NodeList candidates(Instant instant, OsVersionTarget target, NodeList allNodes) {
+    /** Returns nodes that can be deprovisioned at given instant */
+    private List<Node> deprovisionable(Instant instant, OsVersionTarget target, NodeList allNodes) {
         NodeList nodes = allNodes.state(Node.State.active, Node.State.provisioned).nodeType(target.nodeType());
         if (softRebuild) {
-            // Retire only hosts which do not have a replaceable root disk
+            // Consider only hosts which do not have a replaceable root disk
             nodes = nodes.not().replaceableRootDisk();
         }
-        return nodes.not().deprovisioning()
-                    .not().onOsVersion(target.version())
-                    .matching(node -> canUpgradeTo(target.version(), instant, node))
-                    .byIncreasingOsVersion()
-                    .first(upgradeSlots(target, nodes.deprovisioning()));
+        // Retire hosts up to slot limit while ensuring that only one group is retired at a time
+        NodeList activeNodes = allNodes.state(Node.State.active);
+        Map<ClusterId, Set<ClusterSpec.Group>> retiringGroupsByCluster = groupsOf(activeNodes.retiring());
+        int limit = upgradeSlots(target, nodes.deprovisioning());
+        List<Node> result = new ArrayList<>();
+        NodeList candidates = nodes.not().deprovisioning()
+                                   .not().onOsVersion(target.version())
+                                   .matching(node -> canUpgradeTo(target.version(), instant, node))
+                                   .byIncreasingOsVersion();
+        for (Node host : candidates) {
+            if (result.size() == limit) break;
+            // For all clusters residing on this host: Determine if deprovisioning the host would imply retiring nodes
+            // in additional groups beyond those already having retired nodes. If true, defer deprovisioning the host
+            boolean canDeprovision = true;
+            Map<ClusterId, Set<ClusterSpec.Group>> groupsOnHost = groupsOf(activeNodes.childrenOf(host));
+            for (var clusterAndGroups : groupsOnHost.entrySet()) {
+                Set<ClusterSpec.Group> groups = clusterAndGroups.getValue();
+                Set<ClusterSpec.Group> retiringGroups = retiringGroupsByCluster.get(clusterAndGroups.getKey());
+                if (retiringGroups != null && !groups.equals(retiringGroups)) {
+                    canDeprovision = false;
+                    break;
+                }
+            }
+            // Deprovision host and count all cluster groups on the host as being retired
+            if (canDeprovision) {
+                result.add(host);
+                groupsOnHost.forEach((cluster, groups) -> retiringGroupsByCluster.merge(cluster, groups, (oldVal, newVal) -> {
+                    oldVal.addAll(newVal);
+                    return oldVal;
+                }));
+            }
+        }
+        return Collections.unmodifiableList(result);
     }
 
     /** Upgrade given host by retiring and deprovisioning it */
@@ -68,4 +107,17 @@ public class RetiringOsUpgrader extends OsUpgrader {
         nodeRepository.nodes().upgradeOs(NodeListFilter.from(host), Optional.of(target));
     }
 
+    /** Returns the stateful groups present on given nodes, grouped by their cluster ID */
+    private static Map<ClusterId, Set<ClusterSpec.Group>> groupsOf(NodeList nodes) {
+        return nodes.stream()
+                    .filter(node -> node.allocation().isPresent() &&
+                                    node.allocation().get().membership().cluster().isStateful() &&
+                                    node.allocation().get().membership().cluster().group().isPresent())
+                    .collect(Collectors.groupingBy(node -> new ClusterId(node.allocation().get().owner(),
+                                                                         node.allocation().get().membership().cluster().id()),
+                                                   HashMap::new,
+                                                   Collectors.mapping(n -> n.allocation().get().membership().cluster().group().get(),
+                                                                      Collectors.toCollection(HashSet::new))));
+    }
+
 }