Fix: Java agent: handle partial payload read
[lttng-ust.git] / liblttng-ust-java-agent / java / lttng-ust-agent-common / org / lttng / ust / agent / client / LttngTcpSessiondClient.java
1 /*
2 * Copyright (C) 2015-2016 EfficiOS Inc., Alexandre Montplaisir <alexmonthy@efficios.com>
3 * Copyright (C) 2013 - David Goulet <dgoulet@efficios.com>
4 *
5 * This library is free software; you can redistribute it and/or modify it
6 * under the terms of the GNU Lesser General Public License, version 2.1 only,
7 * as published by the Free Software Foundation.
8 *
9 * This library is distributed in the hope that it will be useful, but WITHOUT
10 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
11 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License
12 * for more details.
13 *
14 * You should have received a copy of the GNU Lesser General Public License
15 * along with this library; if not, write to the Free Software Foundation,
16 * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17 */
18
19 package org.lttng.ust.agent.client;
20
21 import java.io.BufferedReader;
22 import java.io.DataInputStream;
23 import java.io.DataOutputStream;
24 import java.io.FileInputStream;
25 import java.io.FileNotFoundException;
26 import java.io.IOException;
27 import java.io.InputStreamReader;
28 import java.lang.management.ManagementFactory;
29 import java.net.Socket;
30 import java.net.UnknownHostException;
31 import java.nio.ByteBuffer;
32 import java.nio.ByteOrder;
33 import java.nio.charset.Charset;
34 import java.util.concurrent.CountDownLatch;
35 import java.util.concurrent.TimeUnit;
36
37 import org.lttng.ust.agent.utils.LttngUstAgentLogger;
38
39 /**
40 * Client for agents to connect to a local session daemon, using a TCP socket.
41 *
42 * @author David Goulet
43 */
44 public class LttngTcpSessiondClient implements Runnable {
45
46 private static final String SESSION_HOST = "127.0.0.1";
47 private static final String ROOT_PORT_FILE = "/var/run/lttng/agent.port";
48 private static final String USER_PORT_FILE = "/.lttng/agent.port";
49 private static final Charset PORT_FILE_ENCODING = Charset.forName("UTF-8");
50
51 private static final int PROTOCOL_MAJOR_VERSION = 2;
52 private static final int PROTOCOL_MINOR_VERSION = 0;
53
54 /** Command header from the session deamon. */
55 private final CountDownLatch registrationLatch = new CountDownLatch(1);
56
57 private Socket sessiondSock;
58 private volatile boolean quit = false;
59
60 private DataInputStream inFromSessiond;
61 private DataOutputStream outToSessiond;
62
63 private final ILttngTcpClientListener logAgent;
64 private final int domainValue;
65 private final boolean isRoot;
66
67 /**
68 * Constructor
69 *
70 * @param logAgent
71 * The listener this client will operate on, typically an LTTng
72 * agent.
73 * @param domainValue
74 * The integer to send to the session daemon representing the
75 * tracing domain to handle.
76 * @param isRoot
77 * True if this client should connect to the root session daemon,
78 * false if it should connect to the user one.
79 */
80 public LttngTcpSessiondClient(ILttngTcpClientListener logAgent, int domainValue, boolean isRoot) {
81 this.logAgent = logAgent;
82 this.domainValue = domainValue;
83 this.isRoot = isRoot;
84 }
85
86 /**
87 * Wait until this client has successfully established a connection to its
88 * target session daemon.
89 *
90 * @param seconds
91 * A timeout in seconds after which this method will return
92 * anyway.
93 * @return True if the the client actually established the connection, false
94 * if we returned because the timeout has elapsed or the thread was
95 * interrupted.
96 */
97 public boolean waitForConnection(int seconds) {
98 try {
99 return registrationLatch.await(seconds, TimeUnit.SECONDS);
100 } catch (InterruptedException e) {
101 return false;
102 }
103 }
104
105 @Override
106 public void run() {
107 for (;;) {
108 if (this.quit) {
109 break;
110 }
111
112 try {
113
114 /*
115 * Connect to the session daemon before anything else.
116 */
117 log("Connecting to sessiond");
118 connectToSessiond();
119
120 /*
121 * Register to the session daemon as the Java component of the
122 * UST application.
123 */
124 log("Registering to sessiond");
125 registerToSessiond();
126
127 /*
128 * Block on socket receive and wait for command from the
129 * session daemon. This will return if and only if there is a
130 * fatal error or the socket closes.
131 */
132 log("Waiting on sessiond commands...");
133 handleSessiondCmd();
134 } catch (UnknownHostException uhe) {
135 uhe.printStackTrace();
136 } catch (IOException ioe) {
137 try {
138 Thread.sleep(3000);
139 } catch (InterruptedException e) {
140 e.printStackTrace();
141 }
142 }
143 }
144 }
145
146 /**
147 * Dispose this client and close any socket connection it may hold.
148 */
149 public void close() {
150 log("Closing client");
151 this.quit = true;
152
153 try {
154 if (this.sessiondSock != null) {
155 this.sessiondSock.close();
156 }
157 } catch (IOException e) {
158 e.printStackTrace();
159 }
160 }
161
162 private void connectToSessiond() throws IOException {
163 int rootPort = getPortFromFile(ROOT_PORT_FILE);
164 int userPort = getPortFromFile(getHomePath() + USER_PORT_FILE);
165
166 /*
167 * Check for the edge case of both files existing but pointing to the
168 * same port. In this case, let the root client handle it.
169 */
170 if ((rootPort != 0) && (rootPort == userPort) && (!isRoot)) {
171 log("User and root config files both point to port " + rootPort +
172 ". Letting the root client handle it.");
173 throw new IOException();
174 }
175
176 int portToUse = (isRoot ? rootPort : userPort);
177
178 if (portToUse == 0) {
179 /* No session daemon available. Stop and retry later. */
180 throw new IOException();
181 }
182
183 this.sessiondSock = new Socket(SESSION_HOST, portToUse);
184 this.inFromSessiond = new DataInputStream(sessiondSock.getInputStream());
185 this.outToSessiond = new DataOutputStream(sessiondSock.getOutputStream());
186 }
187
188 private static String getHomePath() {
189 /*
190 * The environment variable LTTNG_HOME overrides HOME if
191 * defined.
192 */
193 String homePath = System.getenv("LTTNG_HOME");
194
195 if (homePath == null) {
196 homePath = System.getProperty("user.home");
197 }
198 return homePath;
199 }
200
201 /**
202 * Read port number from file created by the session daemon.
203 *
204 * @return port value if found else 0.
205 */
206 private static int getPortFromFile(String path) throws IOException {
207 BufferedReader br = null;
208
209 try {
210 br = new BufferedReader(new InputStreamReader(new FileInputStream(path), PORT_FILE_ENCODING));
211 String line = br.readLine();
212 if (line == null) {
213 /* File exists but is empty. */
214 return 0;
215 }
216
217 int port = Integer.parseInt(line, 10);
218 if (port < 0 || port > 65535) {
219 /* Invalid value. Ignore. */
220 port = 0;
221 }
222 return port;
223
224 } catch (NumberFormatException e) {
225 /* File contained something that was not a number. */
226 return 0;
227 } catch (FileNotFoundException e) {
228 /* No port available. */
229 return 0;
230 } finally {
231 if (br != null) {
232 br.close();
233 }
234 }
235 }
236
237 private void registerToSessiond() throws IOException {
238 byte data[] = new byte[16];
239 ByteBuffer buf = ByteBuffer.wrap(data);
240 String pid = ManagementFactory.getRuntimeMXBean().getName().split("@")[0];
241
242 buf.putInt(domainValue);
243 buf.putInt(Integer.parseInt(pid));
244 buf.putInt(PROTOCOL_MAJOR_VERSION);
245 buf.putInt(PROTOCOL_MINOR_VERSION);
246 this.outToSessiond.write(data, 0, data.length);
247 this.outToSessiond.flush();
248 }
249
250 /**
251 * Handle session command from the session daemon.
252 */
253 private void handleSessiondCmd() throws IOException {
254 /* Data read from the socket */
255 byte inputData[] = null;
256 /* Reply data written to the socket, sent to the sessiond */
257 LttngAgentResponse response;
258
259 while (true) {
260 /* Get header from session daemon. */
261 SessiondCommandHeader cmdHeader = recvHeader();
262
263 if (cmdHeader.getDataSize() > 0) {
264 inputData = recvPayload(cmdHeader);
265 }
266
267 switch (cmdHeader.getCommandType()) {
268 case CMD_REG_DONE:
269 {
270 /*
271 * Countdown the registration latch, meaning registration is
272 * done and we can proceed to continue tracing.
273 */
274 registrationLatch.countDown();
275 /*
276 * We don't send any reply to the registration done command.
277 * This just marks the end of the initial session setup.
278 */
279 log("Registration done");
280 continue;
281 }
282 case CMD_LIST:
283 {
284 SessiondCommand listLoggerCmd = new SessiondListLoggersCommand();
285 response = listLoggerCmd.execute(logAgent);
286 log("Received list loggers command");
287 break;
288 }
289 case CMD_EVENT_ENABLE:
290 {
291 if (inputData == null) {
292 /* Invalid command */
293 response = LttngAgentResponse.FAILURE_RESPONSE;
294 break;
295 }
296 SessiondCommand enableEventCmd = new SessiondEnableEventCommand(inputData);
297 response = enableEventCmd.execute(logAgent);
298 log("Received enable event command: " + enableEventCmd.toString());
299 break;
300 }
301 case CMD_EVENT_DISABLE:
302 {
303 if (inputData == null) {
304 /* Invalid command */
305 response = LttngAgentResponse.FAILURE_RESPONSE;
306 break;
307 }
308 SessiondCommand disableEventCmd = new SessiondDisableEventCommand(inputData);
309 response = disableEventCmd.execute(logAgent);
310 log("Received disable event command: " + disableEventCmd.toString());
311 break;
312 }
313 case CMD_APP_CTX_ENABLE:
314 {
315 if (inputData == null) {
316 /* This commands expects a payload, invalid command */
317 response = LttngAgentResponse.FAILURE_RESPONSE;
318 break;
319 }
320 SessiondCommand enableAppCtxCmd = new SessiondEnableAppContextCommand(inputData);
321 response = enableAppCtxCmd.execute(logAgent);
322 log("Received enable app-context command");
323 break;
324 }
325 case CMD_APP_CTX_DISABLE:
326 {
327 if (inputData == null) {
328 /* This commands expects a payload, invalid command */
329 response = LttngAgentResponse.FAILURE_RESPONSE;
330 break;
331 }
332 SessiondCommand disableAppCtxCmd = new SessiondDisableAppContextCommand(inputData);
333 response = disableAppCtxCmd.execute(logAgent);
334 log("Received disable app-context command");
335 break;
336 }
337 default:
338 {
339 /* Unknown command, send empty reply */
340 response = null;
341 log("Received unknown command, ignoring");
342 break;
343 }
344 }
345
346 /* Send response to the session daemon. */
347 byte[] responseData;
348 if (response == null) {
349 responseData = new byte[4];
350 ByteBuffer buf = ByteBuffer.wrap(responseData);
351 buf.order(ByteOrder.BIG_ENDIAN);
352 } else {
353 log("Sending response: " + response.toString());
354 responseData = response.getBytes();
355 }
356 this.outToSessiond.write(responseData, 0, responseData.length);
357 this.outToSessiond.flush();
358 }
359 }
360
361 /**
362 * Receive header data from the session daemon using the LTTng command
363 * static buffer of the right size.
364 */
365 private SessiondCommandHeader recvHeader() throws IOException {
366 byte data[] = new byte[SessiondCommandHeader.HEADER_SIZE];
367 int bytesLeft = data.length;
368 int bytesOffset = 0;
369
370 while (bytesLeft != 0) {
371 int bytesRead = this.inFromSessiond.read(data, bytesOffset, bytesLeft);
372
373 if (bytesRead < 0) {
374 throw new IOException();
375 }
376 bytesLeft -= bytesRead;
377 bytesOffset += bytesRead;
378 }
379 return new SessiondCommandHeader(data);
380 }
381
382 /**
383 * Receive payload from the session daemon. This MUST be done after a
384 * recvHeader() so the header value of a command are known.
385 *
386 * The caller SHOULD use isPayload() before which returns true if a payload
387 * is expected after the header.
388 */
389 private byte[] recvPayload(SessiondCommandHeader headerCmd) throws IOException {
390 byte payload[] = new byte[(int) headerCmd.getDataSize()];
391 int bytesLeft = payload.length;
392 int bytesOffset = 0;
393
394 /* Failsafe check so we don't waste our time reading 0 bytes. */
395 if (bytesLeft == 0) {
396 return null;
397 }
398
399 while (bytesLeft != 0) {
400 int bytesRead = inFromSessiond.read(payload, bytesOffset, bytesLeft);
401
402 if (bytesRead < 0) {
403 throw new IOException();
404 }
405 bytesLeft -= bytesRead;
406 bytesOffset += bytesRead;
407 }
408 return payload;
409 }
410
411 /**
412 * Wrapper for this class's logging, adds the connection's characteristics
413 * to help differentiate between multiple TCP clients.
414 */
415 private void log(String message) {
416 LttngUstAgentLogger.log(getClass(),
417 "(root=" + isRoot + ", domain=" + domainValue + ") " + message);
418 }
419 }
This page took 0.038714 seconds and 4 git commands to generate.