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
|
// Copyright Vespa.ai. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package ai.vespa.metricsproxy.service;
import com.yahoo.component.annotation.Inject;
import com.yahoo.component.AbstractComponent;
import java.time.Duration;
import java.util.logging.Level;
import com.yahoo.jrt.ErrorCode;
import com.yahoo.jrt.Request;
import com.yahoo.jrt.Spec;
import com.yahoo.jrt.Supervisor;
import com.yahoo.jrt.Target;
import com.yahoo.jrt.Transport;
import java.io.BufferedReader;
import java.io.StringReader;
import java.util.ArrayList;
import java.util.List;
import java.util.logging.Logger;
/**
* Connects to the config sentinel and gets information like pid for the services on the node
*/
public class ConfigSentinelClient extends AbstractComponent {
private final static Logger log = Logger.getLogger(ConfigSentinelClient.class.getName());
private static final Spec SPEC = new Spec("localhost", 19097);
private final Supervisor supervisor;
private Target connection = null;
@Inject
public ConfigSentinelClient() {
supervisor = new Supervisor(new Transport("sentinel-client")).setDropEmptyBuffers(true);
}
@Override
public void deconstruct() {
synchronized (this) {
if (connection != null) {
connection.close();
connection = null;
}
}
supervisor.transport().shutdown().join();
super.deconstruct();
}
/**
* Update all services reading from config sentinel
*
* @param services The list of services
*/
synchronized void updateServiceStatuses(List<VespaService> services) {
try {
setStatus(services);
} catch (Exception e) {
log.log(Level.SEVERE, "Unable to update service pids from sentinel", e);
}
}
/**
* Update status
*
* @param s The service to update the status for
*/
synchronized void ping(VespaService s) {
List<VespaService> services = new ArrayList<>();
services.add(s);
log.log(Level.FINE, () -> "Ping for service " + s);
try {
setStatus(services);
} catch (Exception e) {
log.log(Level.SEVERE, "Unable to update service pids from sentinel", e);
}
}
/**
* Update the status (pid check etc)
*
* @param services list of services
* @throws Exception if something went wrong
*/
protected synchronized void setStatus(List<VespaService> services) throws Exception {
String in = sentinelLs();
BufferedReader reader = new BufferedReader(new StringReader(in));
String line;
List<VespaService> updatedServices = new ArrayList<>();
while ((line = reader.readLine()) != null) {
if (line.equals("")) {
break;
}
VespaService s = parseServiceString(line, services);
if (s != null) {
updatedServices.add(s);
}
}
//Check if there are services that were not found in output
//from the sentinel
for (VespaService s : services) {
if ((!s.getServiceName().equals("configserver")) && !updatedServices.contains(s)) {
log.log(Level.FINE, () -> "Service " + s + " is no longer found with sentinel - setting alive = false");
s.setAlive(false);
}
}
//Close streams
reader.close();
}
static VespaService parseServiceString(String line, List<VespaService> services) {
String[] parts = line.split(" ");
if (parts.length < 3)
return null;
String name = parts[0];
int pid = -1;
String state = null;
VespaService service = null;
for (VespaService s : services) {
if (s.getInstanceName().compareToIgnoreCase(name) == 0) {
service = s;
break;
}
}
//Could not find this service
//nothing wrong with that as the check is invoked per line from sentinel
if (service == null) {
return service;
}
for (int i = 1; i < parts.length; i++) {
String [] keyValue = parts[i].split("=");
String key = keyValue[0];
String value = keyValue[1];
if (key.equals("state")) {
state = value;
} else if (key.equals("pid")) {
pid = Integer.parseInt(value);
}
}
if (state != null) {
service.setState(state);
if (pid >= 0 && "RUNNING".equals(state)) {
service.setAlive(true);
service.setPid(pid);
} else {
service.setAlive(false);
}
} else {
service.setAlive(false);
}
return service;
}
String sentinelLs() {
String servicelist = "";
synchronized (this) {
if (connection == null || ! connection.isValid()) {
connection = supervisor.connect(SPEC);
}
}
if (connection.isValid()) {
Request req = new Request("sentinel.ls");
connection.invokeSync(req, Duration.ofSeconds(5));
if (req.errorCode() == ErrorCode.NONE &&
req.checkReturnTypes("s"))
{
servicelist = req.returnValues().get(0).asString();
} else {
log.log(Level.WARNING, "Bad answer to RPC request: " + req.errorMessage());
}
} else {
log.log(Level.WARNING, "Could not connect to sentinel at: " + SPEC);
}
return servicelist;
}
}
|
