1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
|
// Copyright Vespa.ai. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package com.yahoo.vespa.hosted.provision.provisioning;
import com.yahoo.config.provision.ApplicationId;
import com.yahoo.config.provision.ClusterSpec;
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 com.yahoo.vespa.hosted.provision.node.IP;
import com.yahoo.vespa.hosted.provision.node.Nodes;
import java.util.ArrayList;
import java.util.Collections;
import java.util.EnumSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
/**
* Builds up data structures necessary for node prioritization. It wraps each node
* up in a {@link NodeCandidate} object with attributes used in sorting.
*
* The prioritization logic is implemented by {@link NodeCandidate}.
*
* @author smorgrav
*/
public class NodePrioritizer {
private final List<NodeCandidate> candidates = new ArrayList<>();
private final LockedNodeList allNodes;
private final HostCapacity capacity;
private final HostResourcesCalculator calculator;
private final NodeSpec requested;
private final ApplicationId application;
private final ClusterSpec clusterSpec;
private final IP.Allocation.Context ipAllocationContext;
private final Nodes nodes;
private final boolean dynamicProvisioning;
private final boolean allowHostSharing;
private final boolean exclusiveAllocation;
private final boolean makeExclusive;
private final boolean canAllocateToSpareHosts;
private final boolean topologyChange;
private final int currentClusterSize;
private final Set<Node> spareHosts;
public NodePrioritizer(LockedNodeList allNodes, ApplicationId application, ClusterSpec clusterSpec, NodeSpec nodeSpec,
boolean dynamicProvisioning, boolean allowHostSharing, IP.Allocation.Context ipAllocationContext, Nodes nodes,
HostResourcesCalculator hostResourcesCalculator, int spareCount, boolean exclusiveAllocation, boolean makeExclusive) {
this.allNodes = allNodes;
this.calculator = hostResourcesCalculator;
this.capacity = new HostCapacity(this.allNodes, hostResourcesCalculator);
this.requested = nodeSpec;
this.clusterSpec = clusterSpec;
this.application = application;
this.dynamicProvisioning = dynamicProvisioning;
this.allowHostSharing = allowHostSharing;
this.exclusiveAllocation = exclusiveAllocation;
this.makeExclusive = makeExclusive;
this.spareHosts = dynamicProvisioning ?
capacity.findSpareHostsInDynamicallyProvisionedZones(this.allNodes.asList()) :
capacity.findSpareHosts(this.allNodes.asList(), spareCount);
this.ipAllocationContext = ipAllocationContext;
this.nodes = nodes;
NodeList nodesInCluster = this.allNodes.owner(application).type(clusterSpec.type()).cluster(clusterSpec.id());
NodeList nonRetiredNodesInCluster = nodesInCluster.not().retired();
long currentGroups = nonRetiredNodesInCluster.state(Node.State.active).stream()
.flatMap(node -> node.allocation()
.flatMap(alloc -> alloc.membership().cluster().group().map(ClusterSpec.Group::index))
.stream())
.distinct()
.count();
this.topologyChange = currentGroups != requested.groups();
this.currentClusterSize = (int) nonRetiredNodesInCluster.state(Node.State.active).stream().count();
// In dynamically provisioned zones, we can always take spare hosts since we can provision new on-demand,
// NodeCandidate::compareTo will ensure that they will not be used until there is no room elsewhere.
// In non-dynamically provisioned zones, we only allow allocating to spare hosts to replace failed nodes.
this.canAllocateToSpareHosts = dynamicProvisioning || isReplacement(nodesInCluster, clusterSpec.group());
}
/** Collects all node candidates for this application and returns them in the most-to-least preferred order */
public List<NodeCandidate> collect() {
addApplicationNodes();
addReadyNodes();
addCandidatesOnExistingHosts();
return prioritize();
}
/** Returns the list of nodes sorted by {@link NodeCandidate#compareTo(NodeCandidate)} */
private List<NodeCandidate> prioritize() {
// Group candidates by their switch hostname
Map<String, List<NodeCandidate>> candidatesBySwitch = this.candidates.stream()
.collect(Collectors.groupingBy(candidate -> candidate.parent.orElseGet(candidate::toNode)
.switchHostname()
.orElse("")));
// Mark lower priority nodes on shared switch as non-exclusive
List<NodeCandidate> nodes = new ArrayList<>(this.candidates.size());
for (var clusterSwitch : candidatesBySwitch.keySet()) {
List<NodeCandidate> switchCandidates = candidatesBySwitch.get(clusterSwitch);
if (clusterSwitch.isEmpty()) {
nodes.addAll(switchCandidates); // Nodes are on exclusive switch by default
} else {
Collections.sort(switchCandidates);
NodeCandidate bestNode = switchCandidates.get(0);
nodes.add(bestNode);
for (var node : switchCandidates.subList(1, switchCandidates.size())) {
nodes.add(node.withExclusiveSwitch(false));
}
}
}
Collections.sort(nodes);
return nodes;
}
/** Add a node on each host with enough capacity for the requested flavor */
private void addCandidatesOnExistingHosts() {
if (requested.resources().isEmpty()) return;
for (Node host : allNodes) {
if ( ! nodes.canAllocateTenantNodeTo(host, dynamicProvisioning)) continue;
if (nodes.suspended(host)) continue; // Hosts that are suspended may be down for some time, e.g. for OS upgrade
if (host.reservedTo().isPresent() && !host.reservedTo().get().equals(application.tenant())) continue;
if (host.reservedTo().isPresent() && application.instance().isTester()) continue;
if (makeExclusive) {
if ( ! allowHostSharing && exclusiveAllocation && ! fitsPerfectly(host)) continue;
} else {
if (host.exclusiveToApplicationId().isPresent() && ! fitsPerfectly(host)) continue;
}
if ( ! host.provisionedForApplicationId().map(application::equals).orElse(true)) continue;
if ( ! host.exclusiveToApplicationId().map(application::equals).orElse(true)) continue;
if ( ! host.exclusiveToClusterType().map(clusterSpec.type()::equals).orElse(true)) continue;
if (spareHosts.contains(host) && !canAllocateToSpareHosts) continue;
if ( ! capacity.hasCapacity(host, requested.resources().get())) continue;
if ( ! allNodes.childrenOf(host).owner(application).cluster(clusterSpec.id()).isEmpty()) continue;
if ( ! requested.cloudAccount().isUnspecified() && ! requested.cloudAccount().equals(host.cloudAccount())) continue;
candidates.add(NodeCandidate.createNewChild(requested.resources().get(),
capacity.availableCapacityOf(host),
host,
spareHosts.contains(host),
allNodes,
ipAllocationContext));
}
}
private boolean fitsPerfectly(Node host) {
return calculator.advertisedResourcesOf(host.flavor()).compatibleWith(requested.resources().get());
}
/** Add existing nodes allocated to the application */
private void addApplicationNodes() {
EnumSet<Node.State> legalStates = EnumSet.of(Node.State.active, Node.State.inactive, Node.State.reserved);
allNodes.stream()
.filter(node -> node.type() == requested.type())
.filter(node -> legalStates.contains(node.state()))
.filter(node -> node.allocation().isPresent())
.filter(node -> node.allocation().get().owner().equals(application))
.filter(node -> node.allocation().get().membership().cluster().id().equals(clusterSpec.id()))
.filter(node -> node.state() == Node.State.active || canStillAllocate(node))
.map(node -> candidateFrom(node, false))
.forEach(candidates::add);
}
/** Add nodes already provisioned, but not allocated to any application */
private void addReadyNodes() {
allNodes.stream()
.filter(node -> node.type() == requested.type())
.filter(node -> node.state() == Node.State.ready)
.map(node -> candidateFrom(node, false))
.filter(n -> !n.violatesSpares || canAllocateToSpareHosts)
.forEach(candidates::add);
}
/** Create a candidate from given pre-existing node */
private NodeCandidate candidateFrom(Node node, boolean isSurplus) {
Optional<Node> optionalParent = allNodes.parentOf(node);
if (optionalParent.isPresent()) {
Node parent = optionalParent.get();
return NodeCandidate.createChild(node,
capacity.availableCapacityOf(parent),
parent,
spareHosts.contains(parent),
isSurplus,
false,
parent.exclusiveToApplicationId().isEmpty()
&& requested.canResize(node.resources(),
capacity.unusedCapacityOf(parent),
clusterSpec.type(),
topologyChange,
currentClusterSize));
} else {
return NodeCandidate.createStandalone(node, isSurplus, false);
}
}
/** Returns whether we are allocating to replace a failed node */
private boolean isReplacement(NodeList nodesInCluster, Optional<ClusterSpec.Group> group) {
NodeList nodesInGroup = group.map(ClusterSpec.Group::index)
.map(nodesInCluster::group)
.orElse(nodesInCluster);
int failedNodesInGroup = nodesInGroup.failing().size() + nodesInGroup.state(Node.State.failed).size();
if (failedNodesInGroup == 0) return false;
return ! requested.fulfilledBy(nodesInGroup.size() - failedNodesInGroup);
}
/**
* We may regret that a non-active node is allocated to a host and not offer it to the application
* now, e.g if we want to retire the host.
*
* @return true if we still want to allocate the given node to its parent
*/
private boolean canStillAllocate(Node node) {
if (node.type() != NodeType.tenant || node.parentHostname().isEmpty()) return true;
Optional<Node> parent = allNodes.parentOf(node);
return parent.isPresent() && nodes.canAllocateTenantNodeTo(parent.get(), dynamicProvisioning);
}
}
|
