]> git.ipfire.org Git - thirdparty/systemd.git/blame - src/journal-remote/journal-upload-journal.c
Merge pull request #5046 from stefanha/vsock
[thirdparty/systemd.git] / src / journal-remote / journal-upload-journal.c
CommitLineData
b5efdb8a
LP
1/***
2 This file is part of systemd.
3
4 Copyright 2014 Zbigniew Jędrzejewski-Szmek
5
6 systemd is free software; you can redistribute it and/or modify it
7 under the terms of the GNU Lesser General Public License as published by
8 the Free Software Foundation; either version 2.1 of the License, or
9 (at your option) any later version.
10
11 systemd is distributed in the hope that it will be useful, but
12 WITHOUT ANY WARRANTY; without even the implied warranty of
13 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
14 Lesser General Public License for more details.
15
16 You should have received a copy of the GNU Lesser General Public License
17 along with systemd; If not, see <http://www.gnu.org/licenses/>.
18***/
19
eacbb4d3 20#include <curl/curl.h>
cf0fbc49 21#include <stdbool.h>
eacbb4d3 22
b5efdb8a
LP
23#include "alloc-util.h"
24#include "journal-upload.h"
eacbb4d3
ZJS
25#include "log.h"
26#include "utf8.h"
b5efdb8a 27#include "util.h"
d79ca7a6 28#include "sd-daemon.h"
eacbb4d3
ZJS
29
30/**
31 * Write up to size bytes to buf. Return negative on error, and number of
32 * bytes written otherwise. The last case is a kind of an error too.
33 */
34static ssize_t write_entry(char *buf, size_t size, Uploader *u) {
35 int r;
36 size_t pos = 0;
37
38 assert(size <= SSIZE_MAX);
39
57255510 40 for (;;) {
eacbb4d3
ZJS
41
42 switch(u->entry_state) {
43 case ENTRY_CURSOR: {
a1e58e8e 44 u->current_cursor = mfree(u->current_cursor);
eacbb4d3 45
722b6795 46 r = sd_journal_get_cursor(u->journal, &u->current_cursor);
eb56eb9b
MS
47 if (r < 0)
48 return log_error_errno(r, "Failed to get cursor: %m");
eacbb4d3
ZJS
49
50 r = snprintf(buf + pos, size - pos,
722b6795 51 "__CURSOR=%s\n", u->current_cursor);
eacbb4d3
ZJS
52 if (pos + r > size)
53 /* not enough space */
54 return pos;
55
313cefa1 56 u->entry_state++;
eacbb4d3
ZJS
57
58 if (pos + r == size) {
59 /* exactly one character short, but we don't need it */
60 buf[size - 1] = '\n';
61 return size;
62 }
63
64 pos += r;
65 } /* fall through */
66
67 case ENTRY_REALTIME: {
68 usec_t realtime;
69
70 r = sd_journal_get_realtime_usec(u->journal, &realtime);
eb56eb9b
MS
71 if (r < 0)
72 return log_error_errno(r, "Failed to get realtime timestamp: %m");
eacbb4d3
ZJS
73
74 r = snprintf(buf + pos, size - pos,
75 "__REALTIME_TIMESTAMP="USEC_FMT"\n", realtime);
76 if (r + pos > size)
77 /* not enough space */
78 return pos;
79
313cefa1 80 u->entry_state++;
eacbb4d3
ZJS
81
82 if (r + pos == size) {
83 /* exactly one character short, but we don't need it */
84 buf[size - 1] = '\n';
85 return size;
86 }
87
88 pos += r;
89 } /* fall through */
90
91 case ENTRY_MONOTONIC: {
92 usec_t monotonic;
93 sd_id128_t boot_id;
94
95 r = sd_journal_get_monotonic_usec(u->journal, &monotonic, &boot_id);
eb56eb9b
MS
96 if (r < 0)
97 return log_error_errno(r, "Failed to get monotonic timestamp: %m");
eacbb4d3
ZJS
98
99 r = snprintf(buf + pos, size - pos,
100 "__MONOTONIC_TIMESTAMP="USEC_FMT"\n", monotonic);
101 if (r + pos > size)
102 /* not enough space */
103 return pos;
104
313cefa1 105 u->entry_state++;
eacbb4d3
ZJS
106
107 if (r + pos == size) {
108 /* exactly one character short, but we don't need it */
109 buf[size - 1] = '\n';
110 return size;
111 }
112
113 pos += r;
114 } /* fall through */
115
116 case ENTRY_BOOT_ID: {
117 sd_id128_t boot_id;
118 char sid[33];
119
120 r = sd_journal_get_monotonic_usec(u->journal, NULL, &boot_id);
eb56eb9b
MS
121 if (r < 0)
122 return log_error_errno(r, "Failed to get monotonic timestamp: %m");
eacbb4d3
ZJS
123
124 r = snprintf(buf + pos, size - pos,
125 "_BOOT_ID=%s\n", sd_id128_to_string(boot_id, sid));
5ffa8c81 126 if (r + pos > size)
eacbb4d3
ZJS
127 /* not enough space */
128 return pos;
129
313cefa1 130 u->entry_state++;
eacbb4d3
ZJS
131
132 if (r + pos == size) {
133 /* exactly one character short, but we don't need it */
134 buf[size - 1] = '\n';
135 return size;
136 }
137
138 pos += r;
139 } /* fall through */
140
141 case ENTRY_NEW_FIELD: {
142 u->field_pos = 0;
143
144 r = sd_journal_enumerate_data(u->journal,
145 &u->field_data,
146 &u->field_length);
eb56eb9b
MS
147 if (r < 0)
148 return log_error_errno(r, "Failed to move to next field in entry: %m");
149 else if (r == 0) {
eacbb4d3
ZJS
150 u->entry_state = ENTRY_OUTRO;
151 continue;
152 }
153
154 if (!utf8_is_printable_newline(u->field_data,
155 u->field_length, false)) {
156 u->entry_state = ENTRY_BINARY_FIELD_START;
157 continue;
158 }
159
313cefa1 160 u->entry_state++;
eacbb4d3
ZJS
161 } /* fall through */
162
163 case ENTRY_TEXT_FIELD:
164 case ENTRY_BINARY_FIELD: {
165 bool done;
166 size_t tocopy;
167
168 done = size - pos > u->field_length - u->field_pos;
169 if (done)
170 tocopy = u->field_length - u->field_pos;
171 else
172 tocopy = size - pos;
173
174 memcpy(buf + pos,
175 (char*) u->field_data + u->field_pos,
176 tocopy);
177
178 if (done) {
179 buf[pos + tocopy] = '\n';
180 pos += tocopy + 1;
181 u->entry_state = ENTRY_NEW_FIELD;
182 continue;
183 } else {
184 u->field_pos += tocopy;
185 return size;
186 }
187 }
188
189 case ENTRY_BINARY_FIELD_START: {
190 const char *c;
191 size_t len;
192
193 c = memchr(u->field_data, '=', u->field_length);
194 if (!c || c == u->field_data) {
195 log_error("Invalid field.");
196 return -EINVAL;
197 }
198
199 len = c - (const char*)u->field_data;
200
201 /* need space for label + '\n' */
202 if (size - pos < len + 1)
203 return pos;
204
205 memcpy(buf + pos, u->field_data, len);
206 buf[pos + len] = '\n';
207 pos += len + 1;
208
209 u->field_pos = len + 1;
313cefa1 210 u->entry_state++;
eacbb4d3
ZJS
211 } /* fall through */
212
213 case ENTRY_BINARY_FIELD_SIZE: {
214 uint64_t le64;
215
216 /* need space for uint64_t */
217 if (size - pos < 8)
218 return pos;
219
220 le64 = htole64(u->field_length - u->field_pos);
221 memcpy(buf + pos, &le64, 8);
222 pos += 8;
223
313cefa1 224 u->entry_state++;
eacbb4d3
ZJS
225 continue;
226 }
227
228 case ENTRY_OUTRO:
229 /* need space for '\n' */
230 if (size - pos < 1)
231 return pos;
232
233 buf[pos++] = '\n';
313cefa1
VC
234 u->entry_state++;
235 u->entries_sent++;
eacbb4d3
ZJS
236
237 return pos;
238
239 default:
240 assert_not_reached("WTF?");
241 }
242 }
243 assert_not_reached("WTF?");
244}
245
d79ca7a6 246static inline void check_update_watchdog(Uploader *u) {
d79ca7a6
KC
247 usec_t after;
248 usec_t elapsed_time;
249
0aa176a7 250 if (u->watchdog_usec <= 0)
d79ca7a6 251 return;
0aa176a7
ZJS
252
253 after = now(CLOCK_MONOTONIC);
254 elapsed_time = usec_sub(after, u->watchdog_timestamp);
255 if (elapsed_time > u->watchdog_usec / 2) {
256 log_debug("Update watchdog timer");
257 sd_notify(false, "WATCHDOG=1");
258 u->watchdog_timestamp = after;
d79ca7a6
KC
259 }
260}
261
eacbb4d3
ZJS
262static size_t journal_input_callback(void *buf, size_t size, size_t nmemb, void *userp) {
263 Uploader *u = userp;
264 int r;
265 sd_journal *j;
266 size_t filled = 0;
267 ssize_t w;
268
269 assert(u);
270 assert(nmemb <= SSIZE_MAX / size);
271
d79ca7a6
KC
272 check_update_watchdog(u);
273
eacbb4d3
ZJS
274 j = u->journal;
275
276 while (j && filled < size * nmemb) {
277 if (u->entry_state == ENTRY_DONE) {
278 r = sd_journal_next(j);
279 if (r < 0) {
c33b3297 280 log_error_errno(r, "Failed to move to next entry in journal: %m");
eacbb4d3
ZJS
281 return CURL_READFUNC_ABORT;
282 } else if (r == 0) {
283 if (u->input_event)
284 log_debug("No more entries, waiting for journal.");
285 else {
286 log_info("No more entries, closing journal.");
287 close_journal_input(u);
288 }
289
290 u->uploading = false;
291
292 break;
293 }
294
295 u->entry_state = ENTRY_CURSOR;
296 }
297
298 w = write_entry((char*)buf + filled, size * nmemb - filled, u);
299 if (w < 0)
300 return CURL_READFUNC_ABORT;
301 filled += w;
302
303 if (filled == 0) {
304 log_error("Buffer space is too small to write entry.");
305 return CURL_READFUNC_ABORT;
306 } else if (u->entry_state != ENTRY_DONE)
307 /* This means that all available space was used up */
308 break;
309
310 log_debug("Entry %zu (%s) has been uploaded.",
722b6795 311 u->entries_sent, u->current_cursor);
eacbb4d3
ZJS
312 }
313
314 return filled;
315}
316
317void close_journal_input(Uploader *u) {
318 assert(u);
319
320 if (u->journal) {
321 log_debug("Closing journal input.");
322
323 sd_journal_close(u->journal);
324 u->journal = NULL;
325 }
326 u->timeout = 0;
327}
328
329static int process_journal_input(Uploader *u, int skip) {
330 int r;
331
8a3db16d
KC
332 if (u->uploading)
333 return 0;
334
eacbb4d3 335 r = sd_journal_next_skip(u->journal, skip);
eb56eb9b
MS
336 if (r < 0)
337 return log_error_errno(r, "Failed to skip to next entry: %m");
338 else if (r < skip)
eacbb4d3
ZJS
339 return 0;
340
341 /* have data */
342 u->entry_state = ENTRY_CURSOR;
343 return start_upload(u, journal_input_callback, u);
344}
345
346int check_journal_input(Uploader *u) {
347 if (u->input_event) {
348 int r;
349
350 r = sd_journal_process(u->journal);
351 if (r < 0) {
da927ba9 352 log_error_errno(r, "Failed to process journal: %m");
eacbb4d3
ZJS
353 close_journal_input(u);
354 return r;
355 }
356
357 if (r == SD_JOURNAL_NOP)
358 return 0;
359 }
360
361 return process_journal_input(u, 1);
362}
363
364static int dispatch_journal_input(sd_event_source *event,
365 int fd,
366 uint32_t revents,
367 void *userp) {
368 Uploader *u = userp;
369
370 assert(u);
371
8a3db16d 372 if (u->uploading)
eacbb4d3 373 return 0;
eacbb4d3
ZJS
374
375 log_debug("Detected journal input, checking for new data.");
376 return check_journal_input(u);
377}
378
379int open_journal_for_upload(Uploader *u,
380 sd_journal *j,
381 const char *cursor,
382 bool after_cursor,
383 bool follow) {
384 int fd, r, events;
385
386 u->journal = j;
387
388 sd_journal_set_data_threshold(j, 0);
389
390 if (follow) {
391 fd = sd_journal_get_fd(j);
eb56eb9b
MS
392 if (fd < 0)
393 return log_error_errno(fd, "sd_journal_get_fd failed: %m");
eacbb4d3
ZJS
394
395 events = sd_journal_get_events(j);
396
397 r = sd_journal_reliable_fd(j);
398 assert(r >= 0);
399 if (r > 0)
400 u->timeout = -1;
401 else
402 u->timeout = JOURNAL_UPLOAD_POLL_TIMEOUT;
403
404 r = sd_event_add_io(u->events, &u->input_event,
405 fd, events, dispatch_journal_input, u);
eb56eb9b
MS
406 if (r < 0)
407 return log_error_errno(r, "Failed to register input event: %m");
eacbb4d3
ZJS
408
409 log_debug("Listening for journal events on fd:%d, timeout %d",
410 fd, u->timeout == (uint64_t) -1 ? -1 : (int) u->timeout);
411 } else
412 log_debug("Not listening for journal events.");
413
414 if (cursor) {
415 r = sd_journal_seek_cursor(j, cursor);
ece174c5 416 if (r < 0)
eb56eb9b
MS
417 return log_error_errno(r, "Failed to seek to cursor %s: %m",
418 cursor);
eacbb4d3
ZJS
419 }
420
421 return process_journal_input(u, 1 + !!after_cursor);
422}