blob: ec2cc5107abf2156d9e858bb25e90b979f308b64 (
plain) (
blame)
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
|
package com.yahoo.search.dispatch.rpc;
import ai.vespa.searchlib.searchprotocol.protobuf.SearchProtocol;
import com.google.protobuf.InvalidProtocolBufferException;
import com.yahoo.compress.CompressionType;
import com.yahoo.compress.Compressor;
import com.yahoo.prelude.Pong;
import com.yahoo.search.cluster.ClusterMonitor;
import com.yahoo.search.dispatch.rpc.Client.ProtobufResponse;
import com.yahoo.search.dispatch.rpc.Client.ResponseOrError;
import com.yahoo.search.dispatch.searchcluster.Node;
import com.yahoo.search.result.ErrorMessage;
import com.yahoo.yolean.Exceptions;
import java.util.concurrent.Callable;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
public class RpcPing implements Callable<Pong> {
private static final String RPC_METHOD = "vespa.searchprotocol.ping";
private static final CompressionType PING_COMPRESSION = CompressionType.NONE;
private final Node node;
private final RpcResourcePool resourcePool;
private final ClusterMonitor<Node> clusterMonitor;
public RpcPing(Node node, ClusterMonitor<Node> clusterMonitor, RpcResourcePool rpcResourcePool) {
this.node = node;
this.resourcePool = rpcResourcePool;
this.clusterMonitor = clusterMonitor;
}
@Override
public Pong call() throws Exception {
try {
var queue = new LinkedBlockingQueue<ResponseOrError<ProtobufResponse>>(1);
sendPing(queue);
var responseOrError = queue.poll(clusterMonitor.getConfiguration().getRequestTimeout(), TimeUnit.MILLISECONDS);
if (responseOrError == null) {
return new Pong(ErrorMessage.createNoAnswerWhenPingingNode("Timed out waiting for pong from " + node));
} else if (responseOrError.error().isPresent()) {
return new Pong(ErrorMessage.createBackendCommunicationError(responseOrError.error().get()));
}
return decodeReply(responseOrError.response().get());
} catch (RuntimeException e) {
return new Pong(
ErrorMessage.createBackendCommunicationError("Exception when pinging " + node + ": " + Exceptions.toMessageString(e)));
}
}
private void sendPing(LinkedBlockingQueue<ResponseOrError<ProtobufResponse>> queue) {
var connection = resourcePool.getConnection(node.key());
var ping = SearchProtocol.MonitorRequest.newBuilder().build().toByteArray();
double timeoutSeconds = ((double) clusterMonitor.getConfiguration().getRequestTimeout()) / 1000.0;
Compressor.Compression compressionResult = resourcePool.compressor().compress(PING_COMPRESSION, ping);
connection.request(RPC_METHOD, compressionResult.type(), ping.length, compressionResult.data(), rsp -> queue.add(rsp),
timeoutSeconds);
}
private Pong decodeReply(ProtobufResponse response) throws InvalidProtocolBufferException {
CompressionType compression = CompressionType.valueOf(response.compression());
byte[] responseBytes = resourcePool.compressor().decompress(response.compressedPayload(), compression, response.uncompressedSize());
var reply = SearchProtocol.MonitorReply.parseFrom(responseBytes);
if (reply.getDistributionKey() != node.key()) {
return new Pong(ErrorMessage.createBackendCommunicationError(
"Expected pong from node id " + node.key() + ", response is from id " + reply.getDistributionKey()));
} else if (!reply.getOnline()) {
return new Pong(ErrorMessage.createBackendCommunicationError("Node id " + node.key() + " reports being offline"));
} else {
return new Pong(reply.getActiveDocs());
}
}
}
|