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
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
|
// Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package com.yahoo.vespa.config.server.application;
import com.fasterxml.jackson.databind.JsonNode;
import com.google.inject.Inject;
import com.yahoo.component.AbstractComponent;
import com.yahoo.config.model.api.HostInfo;
import com.yahoo.config.model.api.PortInfo;
import com.yahoo.config.model.api.ServiceInfo;
import com.yahoo.config.provision.ApplicationId;
import com.yahoo.slime.Cursor;
import com.yahoo.vespa.config.server.http.JSONResponse;
import org.glassfish.jersey.client.ClientProperties;
import org.glassfish.jersey.client.proxy.WebResourceFactory;
import javax.ws.rs.GET;
import javax.ws.rs.Path;
import javax.ws.rs.ProcessingException;
import javax.ws.rs.client.Client;
import javax.ws.rs.client.ClientBuilder;
import javax.ws.rs.client.WebTarget;
import java.net.URI;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
/**
* Checks for convergence of config generation for a given application.
*
* @author Ulf Lilleengen
* @author hmusum
*/
public class ConfigConvergenceChecker extends AbstractComponent {
private static final ApplicationId routingApplicationId = ApplicationId.from("hosted-vespa", "routing", "default");
private static final String nodeAdminName = "node-admin";
private static final String statePath = "/state/v1/";
private static final String configSubPath = "config";
private final static Set<String> serviceTypesToCheck = new HashSet<>(Arrays.asList(
"container",
"qrserver",
"searchnode",
"storagenode",
"distributor"
));
private final StateApiFactory stateApiFactory;
@Inject
public ConfigConvergenceChecker() {
this(ConfigConvergenceChecker::createStateApi);
}
public ConfigConvergenceChecker(StateApiFactory stateApiFactory) {
this.stateApiFactory = stateApiFactory;
}
/** Check all services in given application. Returns the minimum current generation of all services */
public ServiceListResponse servicesToCheck(Application application, URI requestUrl, Duration timeoutPerService) {
List<ServiceInfo> servicesToCheck = new ArrayList<>();
application.getModel().getHosts()
.forEach(host -> host.getServices().stream()
.filter(service -> serviceTypesToCheck.contains(service.getServiceType()))
.filter(service -> ! isHostAdminService(application.getId(), service))
.forEach(service -> getStatePort(service).ifPresent(port -> servicesToCheck.add(service))));
Map<ServiceInfo, Long> currentGenerations = getServiceGenerations(servicesToCheck, timeoutPerService);
long currentGeneration = currentGenerations.values().stream().mapToLong(Long::longValue).min().orElse(-1);
return new ServiceListResponse(200, currentGenerations, requestUrl, application.getApplicationGeneration(),
currentGeneration);
}
/** Check service identified by host and port in given application */
public ServiceResponse checkService(Application application, String hostAndPortToCheck, URI requestUrl, Duration timeout) {
Long wantedGeneration = application.getApplicationGeneration();
try {
if (! hostInApplication(application, hostAndPortToCheck))
return ServiceResponse.createHostNotFoundInAppResponse(requestUrl, hostAndPortToCheck, wantedGeneration);
long currentGeneration = getServiceGeneration(URI.create("http://" + hostAndPortToCheck), timeout);
boolean converged = currentGeneration >= wantedGeneration;
return ServiceResponse.createOkResponse(requestUrl, hostAndPortToCheck, wantedGeneration, currentGeneration, converged);
} catch (ProcessingException e) { // e.g. if we cannot connect to the service to find generation
return ServiceResponse.createNotFoundResponse(requestUrl, hostAndPortToCheck, wantedGeneration, e.getMessage());
} catch (Exception e) {
return ServiceResponse.createErrorResponse(requestUrl, hostAndPortToCheck, wantedGeneration, e.getMessage());
}
}
@Path(statePath)
public interface StateApi {
@Path(configSubPath)
@GET
JsonNode config();
}
public interface StateApiFactory {
StateApi createStateApi(Client client, URI serviceUri);
}
/** Gets service generation for a list of services (in parallel). */
private Map<ServiceInfo, Long> getServiceGenerations(List<ServiceInfo> services, Duration timeout) {
return services.parallelStream()
.collect(Collectors.toMap(service -> service,
service -> {
try {
return getServiceGeneration(URI.create("http://" + service.getHostName()
+ ":" + getStatePort(service).get()), timeout);
}
catch (ProcessingException e) { // Cannot connect to service to determine service generation
return -1L;
}
},
(v1, v2) -> { throw new IllegalStateException("Duplicate keys for values '" + v1 + "' and '" + v2 + "'."); },
LinkedHashMap::new
));
}
/** Get service generation of service at given URL */
private long getServiceGeneration(URI serviceUrl, Duration timeout) {
Client client = createClient(timeout);
try {
StateApi state = stateApiFactory.createStateApi(client, serviceUrl);
return generationFromContainerState(state.config());
} finally {
client.close();
}
}
private boolean hostInApplication(Application application, String hostPort) {
for (HostInfo host : application.getModel().getHosts()) {
if (hostPort.startsWith(host.getHostname())) {
for (ServiceInfo service : host.getServices()) {
for (PortInfo port : service.getPorts()) {
if (hostPort.equals(host.getHostname() + ":" + port.getPort())) {
return true;
}
}
}
}
}
return false;
}
private static Client createClient(Duration timeout) {
return ClientBuilder.newBuilder()
.property(ClientProperties.CONNECT_TIMEOUT, (int) timeout.toMillis())
.property(ClientProperties.READ_TIMEOUT, (int) timeout.toMillis())
.build();
}
private static Optional<Integer> getStatePort(ServiceInfo service) {
return service.getPorts().stream()
.filter(port -> port.getTags().contains("state"))
.map(PortInfo::getPort)
.findFirst();
}
private static long generationFromContainerState(JsonNode state) {
return state.get("config").get("generation").asLong(-1);
}
private static StateApi createStateApi(Client client, URI uri) {
WebTarget target = client.target(uri);
return WebResourceFactory.newResource(StateApi.class, target);
}
private static boolean isHostAdminService(ApplicationId id, ServiceInfo service) {
return routingApplicationId.equals(id)
&& service.getProperty("clustername")
.map("node-admin"::equals)
.orElse(false);
}
private static class ServiceListResponse extends JSONResponse {
// Pre-condition: servicesToCheck has a state port
private ServiceListResponse(int status, Map<ServiceInfo, Long> servicesToCheck, URI uri, long wantedGeneration,
long currentGeneration) {
super(status);
Cursor serviceArray = object.setArray("services");
servicesToCheck.forEach((service, generation) -> {
Cursor serviceObject = serviceArray.addObject();
String hostName = service.getHostName();
int statePort = getStatePort(service).get();
serviceObject.setString("host", hostName);
serviceObject.setLong("port", statePort);
serviceObject.setString("type", service.getServiceType());
serviceObject.setString("url", uri.toString() + "/" + hostName + ":" + statePort);
serviceObject.setLong("currentGeneration", generation);
});
object.setString("url", uri.toString());
object.setLong("currentGeneration", currentGeneration);
object.setLong("wantedGeneration", wantedGeneration);
object.setBool("converged", currentGeneration >= wantedGeneration);
}
}
private static class ServiceResponse extends JSONResponse {
private ServiceResponse(int status, URI uri, String hostname, Long wantedGeneration) {
super(status);
object.setString("url", uri.toString());
object.setString("host", hostname);
object.setLong("wantedGeneration", wantedGeneration);
}
static ServiceResponse createOkResponse(URI uri, String hostname, Long wantedGeneration, Long currentGeneration, boolean converged) {
ServiceResponse serviceResponse = new ServiceResponse(200, uri, hostname, wantedGeneration);
serviceResponse.object.setBool("converged", converged);
serviceResponse.object.setLong("currentGeneration", currentGeneration);
return serviceResponse;
}
static ServiceResponse createHostNotFoundInAppResponse(URI uri, String hostname, Long wantedGeneration) {
ServiceResponse serviceResponse = new ServiceResponse(410, uri, hostname, wantedGeneration);
serviceResponse.object.setString("problem", "Host:port (service) no longer part of application, refetch list of services.");
return serviceResponse;
}
static ServiceResponse createErrorResponse(URI uri, String hostname, Long wantedGeneration, String error) {
ServiceResponse serviceResponse = new ServiceResponse(500, uri, hostname, wantedGeneration);
serviceResponse.object.setString("error", error);
return serviceResponse;
}
static ServiceResponse createNotFoundResponse(URI uri, String hostname, Long wantedGeneration, String error) {
ServiceResponse serviceResponse = new ServiceResponse(404, uri, hostname, wantedGeneration);
serviceResponse.object.setString("error", error);
return serviceResponse;
}
}
}
|
