Add LockedNodeList

A LockedNodeList indicates that the entire node repository is locked, and that
no nodes in the list will change while the lock held.

7 files changed, 110 insertions, 75 deletions

diff --git a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/LockedNodeList.java b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/LockedNodeList.java
new file mode 100644
index 00000000000..b90906f8974
--- /dev/null
+++ b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/LockedNodeList.java
@@ -0,0 +1,39 @@
+// Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.vespa.hosted.provision;
+
+import com.yahoo.transaction.Mutex;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+
+import static java.util.stream.Collectors.collectingAndThen;
+
+/**
+ * A type-safe wrapper for {@link NodeList}. Callers that have a reference to this can safely be assumed to holding the
+ * write lock for the node repository.
+ *
+ * This is typically used in situations where modifying a node object depends on inspecting a consistent state of other
+ * nodes in the repository.
+ *
+ * @author mpolden
+ */
+public class LockedNodeList extends NodeList {
+
+    private final Mutex lock;
+
+    public LockedNodeList(List<Node> nodes, Mutex lock) {
+        super(nodes);
+        this.lock = Objects.requireNonNull(lock, "lock must be non-null");
+    }
+
+    @Override
+    public LockedNodeList filter(Predicate<Node> predicate) {
+        return asList().stream()
+                       .filter(predicate)
+                       .collect(collectingAndThen(Collectors.toList(),
+                                                  (nodes) -> new LockedNodeList(nodes, lock)));
+    }
+
+}
diff --git a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/NodeRepository.java b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/NodeRepository.java
index fd77f3fcb0e..16105181ab3 100644
--- a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/NodeRepository.java
+++ b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/NodeRepository.java
@@ -184,6 +184,11 @@ public class NodeRepository extends AbstractComponent {
         return new NodeList(getNodes());
     }
 
+    /** Returns a locked list of all nodes in this repository */
+    public LockedNodeList list(Mutex lock) {
+        return new LockedNodeList(getNodes(), lock);
+    }
+
     /** Returns a filterable list of all load balancers in this repository */
     public LoadBalancerList loadBalancers() {
         return new LoadBalancerList(database().readLoadBalancers().values());
@@ -313,9 +318,7 @@ public class NodeRepository extends AbstractComponent {
     }
 
     /** Adds a list of newly created docker container nodes to the node repository as <i>reserved</i> nodes */
-    // NOTE: This can only be called while holding the allocation lock, and that lock must have been held since
-    //       the nodes list was computed
-    public List<Node> addDockerNodes(List<Node> nodes, Mutex allocationLock) {
+    public List<Node> addDockerNodes(LockedNodeList nodes) {
         for (Node node : nodes) {
             if (!node.flavor().getType().equals(Flavor.Type.DOCKER_CONTAINER)) {
                 throw new IllegalArgumentException("Cannot add " + node.hostname() + ": This is not a docker node");
@@ -329,7 +332,7 @@ public class NodeRepository extends AbstractComponent {
                                                    existing.get() + ", " + existing.get().history() + "). Node to be added: " +
                                                    node + ", " + node.history());
         }
-        return db.addNodesInState(nodes, Node.State.reserved);
+        return db.addNodesInState(nodes.asList(), Node.State.reserved);
     }
 
     /** Adds a list of (newly created) nodes to the node repository as <i>provisioned</i> nodes */
diff --git a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/maintenance/MetricsReporter.java b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/maintenance/MetricsReporter.java
index 87c69d6a124..d8e6b832d5a 100644
--- a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/maintenance/MetricsReporter.java
+++ b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/maintenance/MetricsReporter.java
@@ -9,6 +9,7 @@ import com.yahoo.jdisc.Metric;
 import com.yahoo.vespa.applicationmodel.HostName;
 import com.yahoo.vespa.applicationmodel.ServiceInstance;
 import com.yahoo.vespa.applicationmodel.ServiceStatus;
+import com.yahoo.vespa.hosted.provision.LockedNodeList;
 import com.yahoo.vespa.hosted.provision.Node;
 import com.yahoo.vespa.hosted.provision.NodeRepository;
 import com.yahoo.vespa.hosted.provision.node.Allocation;
@@ -54,7 +55,7 @@ public class MetricsReporter extends Maintainer {
 
     @Override
     public void maintain() {
-        List<Node> nodes = nodeRepository().getNodes();
+        LockedNodeList nodes = nodeRepository().list(() -> {}); // Ignore locking for the purposes of reporting metrics
 
         Map<HostName, List<ServiceInstance>> servicesByHost =
                 serviceMonitor.getServiceModelSnapshot().getServiceInstancesByHostName();
@@ -200,8 +201,8 @@ public class MetricsReporter extends Maintainer {
         return context;
     }
 
-    private void updateStateMetrics(List<Node> nodes) {
-        Map<Node.State, List<Node>> nodesByState = nodes.stream()
+    private void updateStateMetrics(LockedNodeList nodes) {
+        Map<Node.State, List<Node>> nodesByState = nodes.asList().stream()
                 .collect(Collectors.groupingBy(Node::state));
 
         // Metrics pr state
@@ -212,7 +213,7 @@ public class MetricsReporter extends Maintainer {
         }
     }
 
-    private void updateDockerMetrics(List<Node> nodes) {
+    private void updateDockerMetrics(LockedNodeList nodes) {
         // Capacity flavors for docker
         DockerHostCapacity capacity = new DockerHostCapacity(nodes);
         metric.set("hostedVespa.docker.totalCapacityCpu", capacity.getCapacityTotal().vcpu(), null);
@@ -230,7 +231,6 @@ public class MetricsReporter extends Maintainer {
             metric.set("hostedVespa.docker.freeCapacityFlavor", capacity.freeCapacityInFlavorEquivalence(flavor), context);
             metric.set("hostedVespa.docker.hostsAvailableFlavor", capacity.getNofHostsAvailableFor(flavor), context);
         }
-
-
     }
+
 }
diff --git a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/node/IP.java b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/node/IP.java
index 221f40239eb..544c3743d01 100644
--- a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/node/IP.java
+++ b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/node/IP.java
@@ -4,7 +4,7 @@ package com.yahoo.vespa.hosted.provision.node;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.net.InetAddresses;
 import com.google.common.primitives.UnsignedBytes;
-import com.yahoo.vespa.hosted.provision.NodeList;
+import com.yahoo.vespa.hosted.provision.LockedNodeList;
 import com.yahoo.vespa.hosted.provision.persistence.NameResolver;
 
 import java.net.Inet4Address;
@@ -127,10 +127,10 @@ public class IP {
         /**
          * Find a free allocation in this pool. Note that the allocation is not final until it is assigned to a node
          *
-         * @param nodes All nodes in the repository
+         * @param nodes A locked list of all nodes in the repository
          * @return An allocation from the pool, if any can be made
          */
-        public Optional<Allocation> findAllocation(NodeList nodes, NameResolver resolver) {
+        public Optional<Allocation> findAllocation(LockedNodeList nodes, NameResolver resolver) {
             var unusedAddresses = findUnused(nodes);
             var allocation = unusedAddresses.stream()
                                             .filter(IP::isV6)
@@ -149,9 +149,9 @@ public class IP {
         /**
          * Finds all unused addresses in this pool
          *
-         * @param nodes All nodes in the repository
+         * @param nodes Locked list of all nodes in the repository
          */
-        public Set<String> findUnused(NodeList nodes) {
+        public Set<String> findUnused(LockedNodeList nodes) {
             var unusedAddresses = new LinkedHashSet<>(addresses);
             nodes.filter(node -> node.ipConfig().primary().stream().anyMatch(addresses::contains))
                  .forEach(node -> unusedAddresses.removeAll(node.ipConfig().primary()));
diff --git a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/provisioning/DockerHostCapacity.java b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/provisioning/DockerHostCapacity.java
index 4c557c60802..5b33f8261a4 100644
--- a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/provisioning/DockerHostCapacity.java
+++ b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/provisioning/DockerHostCapacity.java
@@ -3,10 +3,10 @@ package com.yahoo.vespa.hosted.provision.provisioning;
 
 import com.yahoo.config.provision.Flavor;
 import com.yahoo.config.provision.NodeType;
+import com.yahoo.vespa.hosted.provision.LockedNodeList;
 import com.yahoo.vespa.hosted.provision.Node;
-import com.yahoo.vespa.hosted.provision.NodeList;
 
-import java.util.List;
+import java.util.Objects;
 
 /**
  * Capacity calculation for docker hosts.
@@ -18,17 +18,10 @@ import java.util.List;
  */
 public class DockerHostCapacity {
 
-    /**
-     * An immutable list of nodes
-     */
-    private final NodeList allNodes;
-
-    public DockerHostCapacity(List<Node> allNodes) {
-        this(new NodeList(allNodes));
-    }
+    private final LockedNodeList allNodes;
 
-    public DockerHostCapacity(NodeList allNodes) {
-        this.allNodes = allNodes;
+    public DockerHostCapacity(LockedNodeList allNodes) {
+        this.allNodes = Objects.requireNonNull(allNodes, "allNodes must be non-null");
     }
 
     /**
diff --git a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/provisioning/GroupPreparer.java b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/provisioning/GroupPreparer.java
index d8862063491..4a1f4f86711 100644
--- a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/provisioning/GroupPreparer.java
+++ b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/provisioning/GroupPreparer.java
@@ -8,8 +8,8 @@ import com.yahoo.lang.MutableInteger;
 import com.yahoo.transaction.Mutex;
 import com.yahoo.vespa.flags.BooleanFlag;
 import com.yahoo.vespa.flags.FetchVector;
+import com.yahoo.vespa.hosted.provision.LockedNodeList;
 import com.yahoo.vespa.hosted.provision.Node;
-import com.yahoo.vespa.hosted.provision.NodeList;
 import com.yahoo.vespa.hosted.provision.NodeRepository;
 
 import java.util.List;
@@ -62,7 +62,7 @@ public class GroupPreparer {
             try (Mutex allocationLock = nodeRepository.lockAllocation()) {
 
                 // Create a prioritized set of nodes
-                NodeList nodeList = nodeRepository.list();
+                LockedNodeList nodeList = nodeRepository.list(allocationLock);
                 NodePrioritizer prioritizer = new NodePrioritizer(
                         nodeList, application, cluster, requestedNodes, spareCount, nodeRepository.nameResolver(),
                         nodeRepository.getAvailableFlavors());
@@ -70,7 +70,7 @@ public class GroupPreparer {
                 prioritizer.addApplicationNodes();
                 prioritizer.addSurplusNodes(surplusActiveNodes);
                 prioritizer.addReadyNodes();
-                prioritizer.addNewDockerNodes(allocationLock, dynamicProvisioningEnabled);
+                prioritizer.addNewDockerNodes(dynamicProvisioningEnabled);
 
                 // Allocate from the prioritized list
                 NodeAllocation allocation = new NodeAllocation(nodeList, application, cluster, requestedNodes,
@@ -112,7 +112,7 @@ public class GroupPreparer {
 
                 // Carry out and return allocation
                 nodeRepository.reserve(allocation.reservableNodes());
-                nodeRepository.addDockerNodes(allocation.newNodes(), allocationLock);
+                nodeRepository.addDockerNodes(new LockedNodeList(allocation.newNodes(), allocationLock));
                 surplusActiveNodes.removeAll(allocation.surplusNodes());
                 return allocation.finalNodes(surplusActiveNodes);
             }
@@ -129,4 +129,5 @@ public class GroupPreparer {
         else
             return ".";
     }
+
 }
diff --git a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/provisioning/NodePrioritizer.java b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/provisioning/NodePrioritizer.java
index 288a24a0d89..818d3979216 100644
--- a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/provisioning/NodePrioritizer.java
+++ b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/provisioning/NodePrioritizer.java
@@ -4,14 +4,13 @@ package com.yahoo.vespa.hosted.provision.provisioning;
 import com.yahoo.config.provision.ApplicationId;
 import com.yahoo.config.provision.ClusterSpec;
 import com.yahoo.config.provision.Flavor;
-import com.yahoo.config.provision.NodeResources;
 import com.yahoo.config.provision.NodeFlavors;
+import com.yahoo.config.provision.NodeResources;
 import com.yahoo.config.provision.NodeType;
 import com.yahoo.log.LogLevel;
-import com.yahoo.transaction.Mutex;
+import com.yahoo.vespa.hosted.provision.LockedNodeList;
 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.IP;
 import com.yahoo.vespa.hosted.provision.persistence.NameResolver;
 
@@ -40,7 +39,7 @@ class NodePrioritizer {
     private final static Logger log = Logger.getLogger(NodePrioritizer.class.getName());
 
     private final Map<Node, PrioritizableNode> nodes = new HashMap<>();
-    private final NodeList allNodes;
+    private final LockedNodeList allNodes;
     private final DockerHostCapacity capacity;
     private final NodeSpec requestedNodes;
     private final ApplicationId appId;
@@ -51,7 +50,7 @@ class NodePrioritizer {
     private final boolean isAllocatingForReplacement;
     private final Set<Node> spareHosts;
 
-    NodePrioritizer(NodeList allNodes, ApplicationId appId, ClusterSpec clusterSpec, NodeSpec nodeSpec,
+    NodePrioritizer(LockedNodeList allNodes, ApplicationId appId, ClusterSpec clusterSpec, NodeSpec nodeSpec,
                     int spares, NameResolver nameResolver, NodeFlavors flavors) {
         this.allNodes = allNodes;
         this.capacity = new DockerHostCapacity(allNodes);
@@ -63,17 +62,17 @@ class NodePrioritizer {
         this.spareHosts = findSpareHosts(allNodes, capacity, spares);
 
         int nofFailedNodes = (int) allNodes.asList().stream()
-                .filter(node -> node.state().equals(Node.State.failed))
-                .filter(node -> node.allocation().isPresent())
-                .filter(node -> node.allocation().get().owner().equals(appId))
-                .filter(node -> node.allocation().get().membership().cluster().id().equals(clusterSpec.id()))
-                .count();
+                                           .filter(node -> node.state().equals(Node.State.failed))
+                                           .filter(node -> node.allocation().isPresent())
+                                           .filter(node -> node.allocation().get().owner().equals(appId))
+                                           .filter(node -> node.allocation().get().membership().cluster().id().equals(clusterSpec.id()))
+                                           .count();
 
         int nofNodesInCluster = (int) allNodes.asList().stream()
-                .filter(node -> node.allocation().isPresent())
-                .filter(node -> node.allocation().get().owner().equals(appId))
-                .filter(node -> node.allocation().get().membership().cluster().id().equals(clusterSpec.id()))
-                .count();
+                                              .filter(node -> node.allocation().isPresent())
+                                              .filter(node -> node.allocation().get().owner().equals(appId))
+                                              .filter(node -> node.allocation().get().membership().cluster().id().equals(clusterSpec.id()))
+                                              .count();
 
         this.isAllocatingForReplacement = isReplacement(nofNodesInCluster, nofFailedNodes);
         this.isDocker = isDocker();
@@ -85,14 +84,14 @@ class NodePrioritizer {
      * We do not count retired or inactive nodes as used capacity (as they could have been
      * moved to create space for the spare node in the first place).
      */
-    private static Set<Node> findSpareHosts(NodeList nodes, DockerHostCapacity capacity, int spares) {
+    private static Set<Node> findSpareHosts(LockedNodeList nodes, DockerHostCapacity capacity, int spares) {
         return nodes.asList().stream()
-                .filter(node -> node.type().equals(NodeType.host))
-                .filter(dockerHost -> dockerHost.state().equals(Node.State.active))
-                .filter(dockerHost -> capacity.freeIPs(dockerHost) > 0)
-                .sorted(capacity::compareWithoutInactive)
-                .limit(spares)
-                .collect(Collectors.toSet());
+                    .filter(node -> node.type().equals(NodeType.host))
+                    .filter(dockerHost -> dockerHost.state().equals(Node.State.active))
+                    .filter(dockerHost -> capacity.freeIPs(dockerHost) > 0)
+                    .sorted(capacity::compareWithoutInactive)
+                    .limit(spares)
+                    .collect(Collectors.toSet());
     }
 
     /**
@@ -120,32 +119,32 @@ class NodePrioritizer {
     /**
      * Add a node on each docker host with enough capacity for the requested flavor
      *
-     * @param allocationLock allocation lock from {@link NodeRepository#lockAllocation()}
      * @param exclusively Whether the ready docker nodes should only be added on hosts that
      *                    already have nodes allocated to this tenant
      */
-    void addNewDockerNodes(Mutex allocationLock, boolean exclusively) {
-        NodeList candidates;
+    void addNewDockerNodes(boolean exclusively) {
+        LockedNodeList candidates = allNodes;
 
         if (exclusively) {
-            Set<String> candidateHostnames = allNodes.asList().stream()
-                    .filter(node -> node.type() == NodeType.tenant)
-                    .filter(node -> node.allocation().map(a -> a.owner().tenant().equals(appId.tenant())).orElse(false))
-                    .flatMap(node -> node.parentHostname().stream())
-                    .collect(Collectors.toSet());
-
-            candidates = allNodes
-                    .filter(node -> candidateHostnames.contains(node.hostname()))
-                    .filter(node -> EnumSet.of(Node.State.provisioned, Node.State.ready, Node.State.active)
-                            .contains(node.state()));
+            Set<String> candidateHostnames = candidates.asList().stream()
+                                                       .filter(node -> node.type() == NodeType.tenant)
+                                                       .filter(node -> node.allocation()
+                                                                           .map(a -> a.owner().tenant().equals(appId.tenant()))
+                                                                           .orElse(false))
+                                                       .flatMap(node -> node.parentHostname().stream())
+                                                       .collect(Collectors.toSet());
+
+            candidates = candidates.filter(node -> candidateHostnames.contains(node.hostname()))
+                                   .filter(node -> EnumSet.of(Node.State.provisioned, Node.State.ready, Node.State.active)
+                                                          .contains(node.state()));
         } else {
-            candidates = allNodes.state(Node.State.active);
+            candidates = candidates.filter(node -> node.state() == Node.State.active);
         }
 
-        addNewDockerNodesOn(allocationLock, candidates);
+        addNewDockerNodesOn(candidates);
     }
 
-    void addNewDockerNodesOn(Mutex allocationLock, NodeList candidates) {
+    private void addNewDockerNodesOn(LockedNodeList candidates) {
         if ( ! isDocker) return;
         ResourceCapacity wantedResourceCapacity = ResourceCapacity.of(resources(requestedNodes));
 
@@ -155,7 +154,7 @@ class NodePrioritizer {
 
             boolean hostHasCapacityForWantedFlavor = capacity.hasCapacity(node, wantedResourceCapacity);
             boolean conflictingCluster = allNodes.childrenOf(node).owner(appId).asList().stream()
-                    .anyMatch(child -> child.allocation().get().membership().cluster().id().equals(clusterSpec.id()));
+                                                 .anyMatch(child -> child.allocation().get().membership().cluster().id().equals(clusterSpec.id()));
 
             if (!hostHasCapacityForWantedFlavor || conflictingCluster) continue;
 
@@ -163,15 +162,15 @@ class NodePrioritizer {
 
             Optional<IP.Allocation> allocation;
             try {
-                allocation = node.ipAddressPool().findAllocation(allNodes, nameResolver);
-                if (!allocation.isPresent()) continue; // No free addresses in this pool
+                allocation = node.ipConfig().pool().findAllocation(allNodes, nameResolver);
+                if (allocation.isEmpty()) continue; // No free addresses in this pool
             } catch (Exception e) {
-                log.log(LogLevel.WARNING, "Failed to resolve hostname for allocation, skipping", e);
+                log.log(LogLevel.WARNING, "Failed allocating IP address on " + node.hostname(), e);
                 continue;
             }
 
             Node newNode = Node.createDockerNode(allocation.get().addresses(),
-                                                 Collections.emptySet(),
+                                                 Set.of(),
                                                  allocation.get().hostname(),
                                                  Optional.of(node.hostname()),
                                                  resources(requestedNodes),
@@ -283,10 +282,10 @@ class NodePrioritizer {
         // Choose container over content nodes
         if (a.allocation().isPresent() && b.allocation().isPresent()) {
             if (a.allocation().get().membership().cluster().type().equals(ClusterSpec.Type.container) &&
-                    !b.allocation().get().membership().cluster().type().equals(ClusterSpec.Type.container))
+                !b.allocation().get().membership().cluster().type().equals(ClusterSpec.Type.container))
                 return -1;
             if (!a.allocation().get().membership().cluster().type().equals(ClusterSpec.Type.container) &&
-                    b.allocation().get().membership().cluster().type().equals(ClusterSpec.Type.container))
+                b.allocation().get().membership().cluster().type().equals(ClusterSpec.Type.container))
                 return 1;
         }