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
|
// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package com.yahoo.vespa.hosted.node.admin.nodeadmin;
import com.yahoo.vespa.hosted.node.admin.container.metrics.Counter;
import com.yahoo.vespa.hosted.node.admin.container.metrics.Dimensions;
import com.yahoo.vespa.hosted.node.admin.container.metrics.Gauge;
import com.yahoo.vespa.hosted.node.admin.container.metrics.Metrics;
import com.yahoo.vespa.hosted.node.admin.nodeagent.NodeAgent;
import com.yahoo.vespa.hosted.node.admin.nodeagent.NodeAgentContext;
import com.yahoo.vespa.hosted.node.admin.nodeagent.NodeAgentContextManager;
import com.yahoo.vespa.hosted.node.admin.nodeagent.NodeAgentFactory;
import com.yahoo.vespa.hosted.node.admin.nodeagent.NodeAgentScheduler;
import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;
/**
* Administers a host (for now only docker hosts) and its nodes (docker containers nodes).
*
* @author stiankri
*/
public class NodeAdminImpl implements NodeAdmin {
private static final Duration NODE_AGENT_FREEZE_TIMEOUT = Duration.ofSeconds(5);
private static final Duration NODE_AGENT_SPREAD = Duration.ofSeconds(3);
private final NodeAgentWithSchedulerFactory nodeAgentWithSchedulerFactory;
private final Clock clock;
private final Duration freezeTimeout;
private final Duration spread;
private boolean previousWantFrozen;
private boolean isFrozen;
private Instant startOfFreezeConvergence;
private final Map<String, NodeAgentWithScheduler> nodeAgentWithSchedulerByHostname = new ConcurrentHashMap<>();
private final Gauge jvmHeapUsed;
private final Gauge jvmHeapFree;
private final Gauge jvmHeapTotal;
private final Counter numberOfUnhandledExceptions;
public NodeAdminImpl(NodeAgentFactory nodeAgentFactory, Metrics metrics, Clock clock) {
this(nodeAgentContext -> create(clock, nodeAgentFactory, nodeAgentContext),
metrics, clock, NODE_AGENT_FREEZE_TIMEOUT, NODE_AGENT_SPREAD);
}
public NodeAdminImpl(NodeAgentFactory nodeAgentFactory, Metrics metrics,
Clock clock, Duration freezeTimeout, Duration spread) {
this(nodeAgentContext -> create(clock, nodeAgentFactory, nodeAgentContext),
metrics, clock, freezeTimeout, spread);
}
NodeAdminImpl(NodeAgentWithSchedulerFactory nodeAgentWithSchedulerFactory,
Metrics metrics, Clock clock, Duration freezeTimeout, Duration spread) {
this.nodeAgentWithSchedulerFactory = nodeAgentWithSchedulerFactory;
this.clock = clock;
this.freezeTimeout = freezeTimeout;
this.spread = spread;
this.previousWantFrozen = true;
this.isFrozen = true;
this.startOfFreezeConvergence = clock.instant();
this.numberOfUnhandledExceptions = metrics.declareCounter("unhandled_exceptions",
new Dimensions(Map.of("src", "node-agents")));
this.jvmHeapUsed = metrics.declareGauge("mem.heap.used");
this.jvmHeapFree = metrics.declareGauge("mem.heap.free");
this.jvmHeapTotal = metrics.declareGauge("mem.heap.total");
}
@Override
public void refreshContainersToRun(Set<NodeAgentContext> nodeAgentContexts) {
Map<String, NodeAgentContext> nodeAgentContextsByHostname = nodeAgentContexts.stream()
.collect(Collectors.toMap(ctx -> ctx.node().id(), Function.identity()));
// Stop and remove NodeAgents that should no longer be running
diff(nodeAgentWithSchedulerByHostname.keySet(), nodeAgentContextsByHostname.keySet())
.forEach(hostname -> nodeAgentWithSchedulerByHostname.remove(hostname).stopForRemoval());
// Start NodeAgent for hostnames that should be running, but aren't yet
diff(nodeAgentContextsByHostname.keySet(), nodeAgentWithSchedulerByHostname.keySet()).forEach(hostname -> {
NodeAgentWithScheduler naws = nodeAgentWithSchedulerFactory.create(nodeAgentContextsByHostname.get(hostname));
naws.start();
nodeAgentWithSchedulerByHostname.put(hostname, naws);
});
Duration timeBetweenNodeAgents = spread.dividedBy(Math.max(nodeAgentContextsByHostname.size() - 1, 1));
Instant nextAgentStart = clock.instant();
// At this point, nodeAgentContextsByHostname and nodeAgentWithSchedulerByHostname should have the same keys
for (Map.Entry<String, NodeAgentContext> entry : nodeAgentContextsByHostname.entrySet()) {
nodeAgentWithSchedulerByHostname.get(entry.getKey()).scheduleTickWith(entry.getValue(), nextAgentStart);
nextAgentStart = nextAgentStart.plus(timeBetweenNodeAgents);
}
}
@Override
public void updateMetrics(boolean isSuspended) {
for (NodeAgentWithScheduler nodeAgentWithScheduler : nodeAgentWithSchedulerByHostname.values()) {
if (!isSuspended) numberOfUnhandledExceptions.add(nodeAgentWithScheduler.getAndResetNumberOfUnhandledExceptions());
nodeAgentWithScheduler.updateContainerNodeMetrics(isSuspended);
}
if (!isSuspended) {
Runtime runtime = Runtime.getRuntime();
long freeMemory = runtime.freeMemory();
long totalMemory = runtime.totalMemory();
long usedMemory = totalMemory - freeMemory;
jvmHeapFree.sample(freeMemory);
jvmHeapUsed.sample(usedMemory);
jvmHeapTotal.sample(totalMemory);
}
}
@Override
public boolean setFrozen(boolean wantFrozen) {
if (wantFrozen != previousWantFrozen) {
if (wantFrozen) {
this.startOfFreezeConvergence = clock.instant();
} else {
this.startOfFreezeConvergence = null;
}
previousWantFrozen = wantFrozen;
}
// Use filter with count instead of allMatch() because allMatch() will short circuit on first non-match
boolean allNodeAgentsConverged = parallelStreamOfNodeAgentWithScheduler()
.filter(nodeAgentScheduler -> !nodeAgentScheduler.setFrozen(wantFrozen, freezeTimeout))
.count() == 0;
if (wantFrozen) {
if (allNodeAgentsConverged) isFrozen = true;
} else isFrozen = false;
return allNodeAgentsConverged;
}
@Override
public boolean isFrozen() {
return isFrozen;
}
@Override
public Duration subsystemFreezeDuration() {
if (startOfFreezeConvergence == null) {
return Duration.ZERO;
} else {
return Duration.between(startOfFreezeConvergence, clock.instant());
}
}
@Override
public void stopNodeAgentServices() {
// Each container may spend 1-1:30 minutes stopping
parallelStreamOfNodeAgentWithScheduler().forEach(NodeAgentWithScheduler::stopForHostSuspension);
}
@Override
public void start() {
}
@Override
public void stop() {
// Stop all node-agents in parallel, will block until the last NodeAgent is stopped
parallelStreamOfNodeAgentWithScheduler().forEach(NodeAgentWithScheduler::stopForRemoval);
}
/**
* Returns a parallel stream of NodeAgentWithScheduler.
*
* <p>Why not just call nodeAgentWithSchedulerByHostname.values().parallelStream()? Experiments
* with Java 11 have shown that with 10 nodes and forEach(), there are a maximum of 3 concurrent
* threads. With HashMap it produces 5. With List it produces 10 concurrent threads.</p>
*/
private Stream<NodeAgentWithScheduler> parallelStreamOfNodeAgentWithScheduler() {
return List.copyOf(nodeAgentWithSchedulerByHostname.values()).parallelStream();
}
// Set-difference. Returns minuend minus subtrahend.
private static <T> Set<T> diff(Set<T> minuend, Set<T> subtrahend) {
var result = new HashSet<>(minuend);
result.removeAll(subtrahend);
return result;
}
static class NodeAgentWithScheduler implements NodeAgentScheduler {
private final NodeAgent nodeAgent;
private final NodeAgentScheduler nodeAgentScheduler;
private NodeAgentWithScheduler(NodeAgent nodeAgent, NodeAgentScheduler nodeAgentScheduler) {
this.nodeAgent = nodeAgent;
this.nodeAgentScheduler = nodeAgentScheduler;
}
void start() { nodeAgent.start(currentContext()); }
void stopForHostSuspension() { nodeAgent.stopForHostSuspension(currentContext()); }
void stopForRemoval() { nodeAgent.stopForRemoval(currentContext()); }
void updateContainerNodeMetrics(boolean isSuspended) { nodeAgent.updateContainerNodeMetrics(currentContext(), isSuspended); }
int getAndResetNumberOfUnhandledExceptions() { return nodeAgent.getAndResetNumberOfUnhandledExceptions(); }
@Override public void scheduleTickWith(NodeAgentContext context, Instant at) { nodeAgentScheduler.scheduleTickWith(context, at); }
@Override public boolean setFrozen(boolean frozen, Duration timeout) { return nodeAgentScheduler.setFrozen(frozen, timeout); }
@Override public NodeAgentContext currentContext() { return nodeAgentScheduler.currentContext(); }
}
@FunctionalInterface
interface NodeAgentWithSchedulerFactory {
NodeAgentWithScheduler create(NodeAgentContext context);
}
private static NodeAgentWithScheduler create(Clock clock, NodeAgentFactory nodeAgentFactory, NodeAgentContext context) {
NodeAgentContextManager contextManager = new NodeAgentContextManager(clock, context);
NodeAgent nodeAgent = nodeAgentFactory.create(contextManager, context);
return new NodeAgentWithScheduler(nodeAgent, contextManager);
}
}
|
