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
|
// Copyright 2020 Oath Inc. 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.HostName;
import com.yahoo.config.provision.NodeType;
import com.yahoo.config.provision.exception.LoadBalancerServiceException;
import com.yahoo.log.LogLevel;
import com.yahoo.transaction.Mutex;
import com.yahoo.transaction.NestedTransaction;
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.lb.LoadBalancer;
import com.yahoo.vespa.hosted.provision.lb.LoadBalancerId;
import com.yahoo.vespa.hosted.provision.lb.LoadBalancerInstance;
import com.yahoo.vespa.hosted.provision.lb.LoadBalancerService;
import com.yahoo.vespa.hosted.provision.lb.Real;
import com.yahoo.vespa.hosted.provision.node.IP;
import com.yahoo.vespa.hosted.provision.persistence.CuratorDatabaseClient;
import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
import java.util.function.Function;
import java.util.logging.Logger;
import java.util.stream.Collectors;
/**
* Provisions and configures application load balancers.
*
* @author mpolden
*/
// Load balancer state transitions:
// 1) (new) -> reserved -> active
// 2) active | reserved -> inactive
// 3) inactive -> active | (removed)
public class LoadBalancerProvisioner {
private static final Logger log = Logger.getLogger(LoadBalancerProvisioner.class.getName());
private final NodeRepository nodeRepository;
private final CuratorDatabaseClient db;
private final LoadBalancerService service;
public LoadBalancerProvisioner(NodeRepository nodeRepository, LoadBalancerService service) {
this.nodeRepository = nodeRepository;
this.db = nodeRepository.database();
this.service = service;
// Read and write all load balancers to make sure they are stored in the latest version of the serialization format
for (var id : db.readLoadBalancerIds()) {
try (var lock = db.lockLoadBalancers(id.application())) {
var loadBalancer = db.readLoadBalancer(id);
loadBalancer.ifPresent(db::writeLoadBalancer);
}
}
}
/**
* Prepare a load balancer for given application and cluster.
*
* If a load balancer for the cluster already exists, it will be reconfigured based on the currently allocated
* nodes. It's state will remain unchanged.
*
* If no load balancer exists, a new one will be provisioned in {@link LoadBalancer.State#reserved}.
*
* Calling this for irrelevant node or cluster types is a no-op.
*/
public void prepare(ApplicationId application, ClusterSpec cluster, NodeSpec requestedNodes) {
if (requestedNodes.type() != NodeType.tenant) return; // Nothing to provision for this node type
if (!cluster.type().isContainer()) return; // Nothing to provision for this cluster type
if (application.instance().isTester()) return; // Do not provision for tester instances
try (var lock = db.lockLoadBalancers(application)) {
provision(application, cluster.id(), false, lock);
}
}
/**
* Activate load balancer for given application and cluster.
*
* If a load balancer for the cluster already exists, it will be reconfigured based on the currently allocated
* nodes and the load balancer itself will be moved to {@link LoadBalancer.State#active}.
*
* Load balancers for clusters that are no longer in given clusters are deactivated.
*
* Calling this when no load balancer has been prepared for given cluster is a no-op.
*/
public void activate(ApplicationId application, Set<ClusterSpec> clusters,
@SuppressWarnings("unused") Mutex applicationLock, NestedTransaction transaction) {
try (var lock = db.lockLoadBalancers(application)) {
var containerClusters = containerClusterOf(clusters);
for (var clusterId : containerClusters) {
// Provision again to ensure that load balancer instance is re-configured with correct nodes
provision(application, clusterId, true, lock);
}
// Deactivate any surplus load balancers, i.e. load balancers for clusters that have been removed
var surplusLoadBalancers = surplusLoadBalancersOf(application, containerClusters);
deactivate(surplusLoadBalancers, transaction);
}
}
/**
* Deactivate all load balancers assigned to given application. This is a no-op if an application does not have any
* load balancer(s).
*/
public void deactivate(ApplicationId application, NestedTransaction transaction) {
try (var applicationLock = nodeRepository.lock(application)) {
try (var lock = db.lockLoadBalancers(application)) {
deactivate(nodeRepository.loadBalancers(application).asList(), transaction);
}
}
}
/** Returns load balancers of given application that are no longer referenced by given clusters */
private List<LoadBalancer> surplusLoadBalancersOf(ApplicationId application, Set<ClusterSpec.Id> activeClusters) {
var activeLoadBalancersByCluster = nodeRepository.loadBalancers(application)
.in(LoadBalancer.State.active)
.asList()
.stream()
.collect(Collectors.toMap(lb -> lb.id().cluster(),
Function.identity()));
var surplus = new ArrayList<LoadBalancer>();
for (var kv : activeLoadBalancersByCluster.entrySet()) {
if (activeClusters.contains(kv.getKey())) continue;
surplus.add(kv.getValue());
}
return Collections.unmodifiableList(surplus);
}
private void deactivate(List<LoadBalancer> loadBalancers, NestedTransaction transaction) {
var now = nodeRepository.clock().instant();
var deactivatedLoadBalancers = loadBalancers.stream()
.map(lb -> lb.with(LoadBalancer.State.inactive, now))
.collect(Collectors.toList());
db.writeLoadBalancers(deactivatedLoadBalancers, transaction);
}
/** Idempotently provision a load balancer for given application and cluster */
private void provision(ApplicationId application, ClusterSpec.Id clusterId, boolean activate,
@SuppressWarnings("unused") Mutex loadBalancersLock) {
var id = new LoadBalancerId(application, clusterId);
var now = nodeRepository.clock().instant();
var loadBalancer = db.readLoadBalancer(id);
if (loadBalancer.isEmpty() && activate) return; // Nothing to activate as this load balancer was never prepared
var force = loadBalancer.isPresent() && loadBalancer.get().state() != LoadBalancer.State.active;
var instance = create(application, clusterId, allocatedContainers(application, clusterId), force);
LoadBalancer newLoadBalancer;
if (loadBalancer.isEmpty()) {
newLoadBalancer = new LoadBalancer(id, instance, LoadBalancer.State.reserved, now);
} else {
var newState = activate ? LoadBalancer.State.active : loadBalancer.get().state();
newLoadBalancer = loadBalancer.get().with(instance).with(newState, now);
}
db.writeLoadBalancer(newLoadBalancer);
}
private LoadBalancerInstance create(ApplicationId application, ClusterSpec.Id cluster, List<Node> nodes, boolean force) {
var reals = new LinkedHashSet<Real>();
for (var node : nodes) {
for (var ip : reachableIpAddresses(node)) {
reals.add(new Real(HostName.from(node.hostname()), ip));
}
}
log.log(LogLevel.DEBUG, "Creating load balancer for " + cluster + " in " + application.toShortString() +
", targeting: " + reals);
try {
return service.create(application, cluster, reals, force);
} catch (Exception e) {
throw new LoadBalancerServiceException("Failed to (re)configure load balancer for " + cluster + " in " +
application + ", targeting: " + reals + ". The operation will be " +
"retried on next deployment", e);
}
}
/** Returns a list of active and reserved nodes of type container in given cluster */
private List<Node> allocatedContainers(ApplicationId application, ClusterSpec.Id clusterId) {
return NodeList.copyOf(nodeRepository.getNodes(NodeType.tenant, Node.State.reserved, Node.State.active))
.owner(application)
.filter(node -> node.state().isAllocated())
.container()
.filter(node -> node.allocation().get().membership().cluster().id().equals(clusterId))
.asList();
}
/** Find IP addresses reachable by the load balancer service */
private Set<String> reachableIpAddresses(Node node) {
Set<String> reachable = new LinkedHashSet<>(node.ipAddresses());
// Remove addresses unreachable by the load balancer service
switch (service.protocol()) {
case ipv4:
reachable.removeIf(IP::isV6);
break;
case ipv6:
reachable.removeIf(IP::isV4);
break;
}
return reachable;
}
private static Set<ClusterSpec.Id> containerClusterOf(Set<ClusterSpec> clusters) {
return clusters.stream()
.filter(c -> c.type().isContainer())
.map(ClusterSpec::id)
.collect(Collectors.toUnmodifiableSet());
}
}
|
