Make more decisions under locks, specifically in expirers

5 files changed, 99 insertions, 76 deletions

diff --git a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/maintenance/DirtyExpirer.java b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/maintenance/DirtyExpirer.java
index 8766dea3d61..e300591fbb2 100644
--- a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/maintenance/DirtyExpirer.java
+++ b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/maintenance/DirtyExpirer.java
@@ -3,6 +3,8 @@ package com.yahoo.vespa.hosted.provision.maintenance;
 
 import com.yahoo.jdisc.Metric;
 import com.yahoo.vespa.hosted.provision.Node;
+import com.yahoo.vespa.hosted.provision.Node.State;
+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.History;
@@ -33,8 +35,12 @@ public class DirtyExpirer extends Expirer {
 
     @Override
     protected void expire(List<Node> expired) {
-        for (Node expiredNode : expired)
-            nodeRepository().nodes().fail(expiredNode.hostname(), wantToDeprovisionOnExpiry, Agent.DirtyExpirer, "Node is stuck in dirty");
+        nodeRepository().nodes().performOn(NodeList.copyOf(expired),
+                                           node -> node.state() == State.dirty && isExpired(node),
+                                           (node, lock) -> nodeRepository().nodes().fail(node.hostname(),
+                                                                                         wantToDeprovisionOnExpiry,
+                                                                                         Agent.DirtyExpirer,
+                                                                                         "Node is stuck in dirty"));
     }
 
 }
diff --git a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/maintenance/FailedExpirer.java b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/maintenance/FailedExpirer.java
index c594f7f43ae..cb0a8005e87 100644
--- a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/maintenance/FailedExpirer.java
+++ b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/maintenance/FailedExpirer.java
@@ -6,13 +6,14 @@ import com.yahoo.config.provision.NodeType;
 import com.yahoo.config.provision.Zone;
 import com.yahoo.jdisc.Metric;
 import com.yahoo.vespa.hosted.provision.Node;
+import com.yahoo.vespa.hosted.provision.Node.State;
 import com.yahoo.vespa.hosted.provision.NodeList;
+import com.yahoo.vespa.hosted.provision.NodeMutex;
 import com.yahoo.vespa.hosted.provision.NodeRepository;
 import com.yahoo.vespa.hosted.provision.node.Agent;
-import com.yahoo.vespa.hosted.provision.node.History;
+import com.yahoo.vespa.hosted.provision.node.History.Event.Type;
 
 import java.time.Duration;
-import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
 import java.util.function.Predicate;
@@ -67,56 +68,47 @@ public class FailedExpirer extends NodeRepositoryMaintainer {
 
     @Override
     protected double maintain() {
-        NodeList allNodes = nodeRepository.nodes().list();
-        List<Node> remainingNodes = new ArrayList<>(allNodes.state(Node.State.failed)
-                                                            .nodeType(NodeType.tenant, NodeType.host)
-                                                            .asList());
+        Predicate<Node> isExpired = node ->    node.state() == State.failed
+                                            && node.history().hasEventBefore(Type.failed, clock().instant().minus(expiryFor(node)));
+        NodeList allNodes = nodeRepository.nodes().list(); // Stale snapshot, not critical.
 
-        recycleIf(node -> node.allocation().isEmpty(), remainingNodes, allNodes);
-        recycleIf(node -> !node.allocation().get().membership().cluster().isStateful() &&
-                          node.history().hasEventBefore(History.Event.Type.failed, clock().instant().minus(statelessExpiry)),
-                  remainingNodes,
-                  allNodes);
-        recycleIf(node -> node.allocation().get().membership().cluster().isStateful() &&
-                          node.history().hasEventBefore(History.Event.Type.failed, clock().instant().minus(statefulExpiry)),
-                  remainingNodes,
-                  allNodes);
+        nodeRepository.nodes().performOn(allNodes.nodeType(NodeType.tenant),
+                                         isExpired,
+                                         (node, lock) -> recycle(node, List.of(), allNodes).get());
+
+        nodeRepository.nodes().performOnRecursively(allNodes.nodeType(NodeType.host),
+                                                    nodes -> isExpired.test(nodes.parent().node()),
+                                                    nodes -> recycle(nodes.parent().node(),
+                                                                     nodes.children().stream().map(NodeMutex::node).toList(),
+                                                                     allNodes)
+                                                            .map(List::of).orElse(List.of()));
         return 1.0;
     }
 
-    /** Recycle the nodes matching condition, and remove those nodes from the nodes list. */
-    private void recycleIf(Predicate<Node> condition, List<Node> failedNodes, NodeList allNodes) {
-        List<Node> nodesToRecycle = failedNodes.stream().filter(condition).toList();
-        failedNodes.removeAll(nodesToRecycle);
-        recycle(nodesToRecycle, allNodes);
+    private Duration expiryFor(Node node) {
+        return node.allocation().isEmpty() ? Duration.ZERO
+                                           : node.allocation().get().membership().cluster().isStateful() ? statefulExpiry
+                                                                                                         : statelessExpiry;
     }
 
-    /** Move eligible nodes to dirty or parked. This may be a subset of the given nodes */
-    private void recycle(List<Node> nodes, NodeList allNodes) {
-        List<Node> nodesToRecycle = new ArrayList<>();
-        for (Node candidate : nodes) {
-            Optional<String> reason = shouldPark(candidate, allNodes);
-            if (reason.isPresent()) {
-                List<String> unparkedChildren = candidate.type().isHost() ?
-                                                allNodes.childrenOf(candidate)
-                                                        .not()
-                                                        .state(Node.State.parked)
-                                                        .mapToList(Node::hostname) :
-                                                List.of();
-
-                if (unparkedChildren.isEmpty()) {
-                    nodeRepository.nodes().park(candidate.hostname(), true, Agent.FailedExpirer,
-                                                "Parked by FailedExpirer due to " + reason.get());
-                } else {
-                    log.info(String.format("Expired failed node %s was not parked because of unparked children: %s",
-                                           candidate.hostname(), String.join(", ", unparkedChildren)));
-                }
+    private Optional<Node> recycle(Node node, List<Node> children, NodeList allNodes) {
+        Optional<String> reason = shouldPark(node, allNodes);
+        if (reason.isPresent()) {
+            List<String> unparkedChildren = children.stream()
+                                                    .filter(child -> child.state() != Node.State.parked)
+                                                    .map(Node::hostname)
+                                                    .toList();
+            if (unparkedChildren.isEmpty()) {
+                return Optional.of(nodeRepository.nodes().park(node.hostname(), true, Agent.FailedExpirer,
+                                                               "Parked by FailedExpirer due to " + reason.get()));
             } else {
-                nodesToRecycle.add(candidate);
+                log.info(String.format("Expired failed node %s was not parked because of unparked children: %s",
+                                       node.hostname(), String.join(", ", unparkedChildren)));
+                return Optional.empty();
             }
+        } else {
+            return Optional.of(nodeRepository.nodes().deallocate(node, Agent.FailedExpirer, "Expired by FailedExpirer"));
         }
-        // TODO: consider locked nodes, and perform on directly
-        nodeRepository.nodes().deallocate(nodesToRecycle, Agent.FailedExpirer, "Expired by FailedExpirer");
     }
 
     /** Returns whether the node should be parked instead of recycled */
diff --git a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/maintenance/InactiveExpirer.java b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/maintenance/InactiveExpirer.java
index aa7aac34389..503ac4be86c 100644
--- a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/maintenance/InactiveExpirer.java
+++ b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/maintenance/InactiveExpirer.java
@@ -3,6 +3,8 @@ package com.yahoo.vespa.hosted.provision.maintenance;
 
 import com.yahoo.jdisc.Metric;
 import com.yahoo.vespa.hosted.provision.Node;
+import com.yahoo.vespa.hosted.provision.Node.State;
+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.History;
@@ -40,9 +42,9 @@ public class InactiveExpirer extends Expirer {
 
     @Override
     protected void expire(List<Node> expired) {
-        expired.forEach(node -> {
-            nodeRepository.nodes().deallocate(node, Agent.InactiveExpirer, "Expired by InactiveExpirer");
-        });
+        nodeRepository.nodes().performOn(NodeList.copyOf(expired),
+                                         node -> node.state() == State.inactive && isExpired(node),
+                                         (node, lock) -> nodeRepository.nodes().deallocate(node, Agent.InactiveExpirer, "Expired by InactiveExpirer"));
     }
 
     @Override
diff --git a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/maintenance/ReservationExpirer.java b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/maintenance/ReservationExpirer.java
index 6f06a2ac22e..2484f496ece 100644
--- a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/maintenance/ReservationExpirer.java
+++ b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/maintenance/ReservationExpirer.java
@@ -25,6 +25,8 @@ public class ReservationExpirer extends Expirer {
     }
 
     @Override
-    protected void expire(List<Node> expired) { nodeRepository().nodes().deallocate(expired, Agent.ReservationExpirer, "Expired by ReservationExpirer"); }
+    protected void expire(List<Node> expired) {
+        nodeRepository().nodes().deallocate(expired, Agent.ReservationExpirer, "Expired by ReservationExpirer");
+    }
 
 }
diff --git a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/node/Nodes.java b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/node/Nodes.java
index e1f89bf1642..2ef2609e6f9 100644
--- a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/node/Nodes.java
+++ b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/node/Nodes.java
@@ -16,6 +16,7 @@ import com.yahoo.vespa.applicationmodel.InfrastructureApplication;
 import com.yahoo.vespa.hosted.provision.LockedNodeList;
 import com.yahoo.vespa.hosted.provision.NoSuchNodeException;
 import com.yahoo.vespa.hosted.provision.Node;
+import com.yahoo.vespa.hosted.provision.Node.State;
 import com.yahoo.vespa.hosted.provision.NodeList;
 import com.yahoo.vespa.hosted.provision.NodeMutex;
 import com.yahoo.vespa.hosted.provision.applications.Applications;
@@ -41,11 +42,11 @@ import java.util.Optional;
 import java.util.Set;
 import java.util.TreeSet;
 import java.util.function.BiFunction;
+import java.util.function.Function;
 import java.util.function.Predicate;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import java.util.stream.Collectors;
-import java.util.stream.Stream;
 
 import static com.yahoo.vespa.hosted.provision.restapi.NodePatcher.DROP_DOCUMENTS_REPORT;
 import static java.util.Comparator.comparing;
@@ -236,8 +237,7 @@ public class Nodes {
      * @param reusable move the node directly to {@link Node.State#dirty} after removal
      */
     public void setRemovable(NodeList nodes, boolean reusable) {
-        performOn(nodes, (node, mutex) -> write(node.with(node.allocation().get().removable(true, reusable)),
-                                                mutex));
+        performOn(nodes, (node, mutex) -> write(node.with(node.allocation().get().removable(true, reusable)), mutex));
     }
 
     /**
@@ -275,25 +275,29 @@ public class Nodes {
         return performOn(NodeList.copyOf(nodes), (node, lock) -> deallocate(node, agent, reason));
     }
 
-    // TODO: take recursive lock
     public List<Node> deallocateRecursively(String hostname, Agent agent, String reason) {
         Node nodeToDirty = node(hostname).orElseThrow(() -> new NoSuchNodeException("Could not deallocate " + hostname + ": Node not found"));
-
-        List<Node> nodesToDirty =
-                (nodeToDirty.type().isHost() ?
-                 Stream.concat(list().childrenOf(hostname).asList().stream(), Stream.of(nodeToDirty)) :
-                 Stream.of(nodeToDirty)).filter(node -> node.state() != Node.State.dirty).toList();
-        List<String> hostnamesNotAllowedToDirty = nodesToDirty.stream()
-                                                              .filter(node -> node.state() != Node.State.provisioned)
-                                                              .filter(node -> node.state() != Node.State.failed)
-                                                              .filter(node -> node.state() != Node.State.parked)
-                                                              .filter(node -> node.state() != Node.State.breakfixed)
-                                                              .map(Node::hostname).toList();
-        if ( ! hostnamesNotAllowedToDirty.isEmpty())
-            illegal("Could not deallocate " + nodeToDirty + ": " +
-                    hostnamesNotAllowedToDirty + " are not in states [provisioned, failed, parked, breakfixed]");
-
-        return nodesToDirty.stream().map(node -> deallocate(node, agent, reason)).toList();
+        List<Node> nodesToDirty = new ArrayList<>();
+        try (RecursiveNodeMutexes locked = lockAndGetRecursively(hostname, Optional.empty())) {
+            for (NodeMutex child : locked.children())
+                if (child.node().state() != Node.State.dirty)
+                    nodesToDirty.add(child.node());
+
+            if (locked.parent().node().state() != State.dirty)
+                nodesToDirty.add(locked.parent().node());
+
+            List<String> hostnamesNotAllowedToDirty = nodesToDirty.stream()
+                                                                  .filter(node -> node.state() != Node.State.provisioned)
+                                                                  .filter(node -> node.state() != Node.State.failed)
+                                                                  .filter(node -> node.state() != Node.State.parked)
+                                                                  .filter(node -> node.state() != Node.State.breakfixed)
+                                                                  .map(Node::hostname).toList();
+            if ( ! hostnamesNotAllowedToDirty.isEmpty())
+                illegal("Could not deallocate " + nodeToDirty + ": " +
+                        hostnamesNotAllowedToDirty + " are not in states [provisioned, failed, parked, breakfixed]");
+
+            return nodesToDirty.stream().map(node -> deallocate(node, agent, reason)).toList();
+        }
     }
 
     /**
@@ -352,7 +356,6 @@ public class Nodes {
      */
     public List<Node> failOrMarkRecursively(String hostname, Agent agent, String reason) {
         List<Node> changed = new ArrayList<>();
-        // TODO: consider whether we can instead fail all unfailed children until none remain, with single locks.
         try (RecursiveNodeMutexes nodes = lockAndGetRecursively(hostname, Optional.empty())) {
             for (NodeMutex child : nodes.children())
                 changed.add(failOrMark(child.node(), agent, reason, child));
@@ -678,7 +681,6 @@ public class Nodes {
         return decommission(hostname, upgrade ? HostOperation.upgradeFlavor : HostOperation.cancel, agent, instant);
     }
 
-    // TODO: use recursive lock
     private List<Node> decommission(String hostname, HostOperation op, Agent agent, Instant instant) {
         Optional<NodeMutex> nodeMutex = lockAndGet(hostname);
         if (nodeMutex.isEmpty()) return List.of();
@@ -727,8 +729,8 @@ public class Nodes {
         return db.writeTo(nodes, Agent.system, Optional.empty());
     }
 
-    private List<Node> performOn(Predicate<Node> filter, BiFunction<Node, Mutex, Node> action) {
-        return performOn(list().matching(filter), action);
+    public List<Node> performOn(Predicate<Node> filter, BiFunction<Node, Mutex, Node> action) {
+        return performOn(list(), filter, action);
     }
 
     /**
@@ -737,19 +739,38 @@ public class Nodes {
      * @param action the action to perform
      * @return the set of nodes on which the action was performed, as they became as a result of the operation
      */
-    // TODO: make public, with changed API, and apply filter after reading nodes under lock
-    private List<Node> performOn(NodeList nodes, BiFunction<Node, Mutex, Node> action) {
+    public List<Node> performOn(NodeList nodes, BiFunction<Node, Mutex, Node> action) {
+        return performOn(nodes, __ -> true, action);
+    }
+
+    public List<Node> performOn(NodeList nodes, Predicate<Node> filter, BiFunction<Node, Mutex, Node> action) {
         List<Node> resultingNodes = new ArrayList<>();
         nodes.stream().collect(groupingBy(Nodes::applicationIdForLock))
              .forEach((applicationId, nodeList) -> { // Grouped only to reduce number of lock acquire/release cycles.
                  try (NodeMutexes locked = lockAndGetAll(nodeList, Optional.empty())) {
                      for (NodeMutex node : locked.nodes())
-                         resultingNodes.add(action.apply(node.node(), node));
+                         if (filter.test(node.node()))
+                             resultingNodes.add(action.apply(node.node(), node));
                  }
              });
         return resultingNodes;
     }
 
+    public List<Node> performOnRecursively(NodeList parents, Predicate<RecursiveNodeMutexes> filter, Function<RecursiveNodeMutexes, List<Node>> action) {
+        for (Node node : parents)
+            if (node.parentHostname().isPresent())
+                throw new IllegalArgumentException(node + " is not a parent host");
+
+        List<Node> resultingNodes = new ArrayList<>();
+        for (Node parent : parents) {
+            try (RecursiveNodeMutexes locked = lockAndGetRecursively(parent.hostname(), Optional.empty())) {
+                if (filter.test(locked))
+                    resultingNodes.addAll(action.apply(locked));
+            }
+        }
+        return resultingNodes;
+    }
+
     public List<Node> dropDocuments(ApplicationId applicationId, Optional<ClusterSpec.Id> clusterId) {
         try (Mutex lock = applications.lock(applicationId)) {
             Instant now = clock.instant();