chiark / gitweb /
treewide: simplify log_*_errno(r,...) immediately followed by "return r"
[elogind.git] / src / journal-remote / journal-upload-journal.c
1 #include <stdbool.h>
2
3 #include <curl/curl.h>
4
5 #include "util.h"
6 #include "log.h"
7 #include "utf8.h"
8 #include "journal-upload.h"
9
10 /**
11  * Write up to size bytes to buf. Return negative on error, and number of
12  * bytes written otherwise. The last case is a kind of an error too.
13  */
14 static ssize_t write_entry(char *buf, size_t size, Uploader *u) {
15         int r;
16         size_t pos = 0;
17
18         assert(size <= SSIZE_MAX);
19
20         while (true) {
21
22                 switch(u->entry_state) {
23                 case ENTRY_CURSOR: {
24                         free(u->current_cursor);
25                         u->current_cursor = NULL;
26
27                         r = sd_journal_get_cursor(u->journal, &u->current_cursor);
28                         if (r < 0)
29                                 return log_error_errno(r, "Failed to get cursor: %m");
30
31                         r = snprintf(buf + pos, size - pos,
32                                      "__CURSOR=%s\n", u->current_cursor);
33                         if (pos + r > size)
34                                 /* not enough space */
35                                 return pos;
36
37                         u->entry_state ++;
38
39                         if (pos + r == size) {
40                                 /* exactly one character short, but we don't need it */
41                                 buf[size - 1] = '\n';
42                                 return size;
43                         }
44
45                         pos += r;
46                 }       /* fall through */
47
48                 case ENTRY_REALTIME: {
49                         usec_t realtime;
50
51                         r = sd_journal_get_realtime_usec(u->journal, &realtime);
52                         if (r < 0)
53                                 return log_error_errno(r, "Failed to get realtime timestamp: %m");
54
55                         r = snprintf(buf + pos, size - pos,
56                                      "__REALTIME_TIMESTAMP="USEC_FMT"\n", realtime);
57                         if (r + pos > size)
58                                 /* not enough space */
59                                 return pos;
60
61                         u->entry_state ++;
62
63                         if (r + pos == size) {
64                                 /* exactly one character short, but we don't need it */
65                                 buf[size - 1] = '\n';
66                                 return size;
67                         }
68
69                         pos += r;
70                 }       /* fall through */
71
72                 case ENTRY_MONOTONIC: {
73                         usec_t monotonic;
74                         sd_id128_t boot_id;
75
76                         r = sd_journal_get_monotonic_usec(u->journal, &monotonic, &boot_id);
77                         if (r < 0)
78                                 return log_error_errno(r, "Failed to get monotonic timestamp: %m");
79
80                         r = snprintf(buf + pos, size - pos,
81                                      "__MONOTONIC_TIMESTAMP="USEC_FMT"\n", monotonic);
82                         if (r + pos > size)
83                                 /* not enough space */
84                                 return pos;
85
86                         u->entry_state ++;
87
88                         if (r + pos == size) {
89                                 /* exactly one character short, but we don't need it */
90                                 buf[size - 1] = '\n';
91                                 return size;
92                         }
93
94                         pos += r;
95                 }       /* fall through */
96
97                 case ENTRY_BOOT_ID: {
98                         sd_id128_t boot_id;
99                         char sid[33];
100
101                         r = sd_journal_get_monotonic_usec(u->journal, NULL, &boot_id);
102                         if (r < 0)
103                                 return log_error_errno(r, "Failed to get monotonic timestamp: %m");
104
105                         r = snprintf(buf + pos, size - pos,
106                                      "_BOOT_ID=%s\n", sd_id128_to_string(boot_id, sid));
107                         if (r + pos> size)
108                                 /* not enough space */
109                                 return pos;
110
111                         u->entry_state ++;
112
113                         if (r + pos == size) {
114                                 /* exactly one character short, but we don't need it */
115                                 buf[size - 1] = '\n';
116                                 return size;
117                         }
118
119                         pos += r;
120                 }       /* fall through */
121
122                 case ENTRY_NEW_FIELD: {
123                         u->field_pos = 0;
124
125                         r = sd_journal_enumerate_data(u->journal,
126                                                       &u->field_data,
127                                                       &u->field_length);
128                         if (r < 0)
129                                 return log_error_errno(r, "Failed to move to next field in entry: %m");
130                         else if (r == 0) {
131                                 u->entry_state = ENTRY_OUTRO;
132                                 continue;
133                         }
134
135                         if (!utf8_is_printable_newline(u->field_data,
136                                                        u->field_length, false)) {
137                                 u->entry_state = ENTRY_BINARY_FIELD_START;
138                                 continue;
139                         }
140
141                         u->entry_state ++;
142                 }       /* fall through */
143
144                 case ENTRY_TEXT_FIELD:
145                 case ENTRY_BINARY_FIELD: {
146                         bool done;
147                         size_t tocopy;
148
149                         done = size - pos > u->field_length - u->field_pos;
150                         if (done)
151                                 tocopy = u->field_length - u->field_pos;
152                         else
153                                 tocopy = size - pos;
154
155                         memcpy(buf + pos,
156                                (char*) u->field_data + u->field_pos,
157                                tocopy);
158
159                         if (done) {
160                                 buf[pos + tocopy] = '\n';
161                                 pos += tocopy + 1;
162                                 u->entry_state = ENTRY_NEW_FIELD;
163                                 continue;
164                         } else {
165                                 u->field_pos += tocopy;
166                                 return size;
167                         }
168                 }
169
170                 case ENTRY_BINARY_FIELD_START: {
171                         const char *c;
172                         size_t len;
173
174                         c = memchr(u->field_data, '=', u->field_length);
175                         if (!c || c == u->field_data) {
176                                 log_error("Invalid field.");
177                                 return -EINVAL;
178                         }
179
180                         len = c - (const char*)u->field_data;
181
182                         /* need space for label + '\n' */
183                         if (size - pos < len + 1)
184                                 return pos;
185
186                         memcpy(buf + pos, u->field_data, len);
187                         buf[pos + len] = '\n';
188                         pos += len + 1;
189
190                         u->field_pos = len + 1;
191                         u->entry_state ++;
192                 }       /* fall through */
193
194                 case ENTRY_BINARY_FIELD_SIZE: {
195                         uint64_t le64;
196
197                         /* need space for uint64_t */
198                         if (size - pos < 8)
199                                 return pos;
200
201                         le64 = htole64(u->field_length - u->field_pos);
202                         memcpy(buf + pos, &le64, 8);
203                         pos += 8;
204
205                         u->entry_state ++;
206                         continue;
207                 }
208
209                 case ENTRY_OUTRO:
210                         /* need space for '\n' */
211                         if (size - pos < 1)
212                                 return pos;
213
214                         buf[pos++] = '\n';
215                         u->entry_state ++;
216                         u->entries_sent ++;
217
218                         return pos;
219
220                 default:
221                         assert_not_reached("WTF?");
222                 }
223         }
224         assert_not_reached("WTF?");
225 }
226
227 static size_t journal_input_callback(void *buf, size_t size, size_t nmemb, void *userp) {
228         Uploader *u = userp;
229         int r;
230         sd_journal *j;
231         size_t filled = 0;
232         ssize_t w;
233
234         assert(u);
235         assert(nmemb <= SSIZE_MAX / size);
236
237         j = u->journal;
238
239         while (j && filled < size * nmemb) {
240                 if (u->entry_state == ENTRY_DONE) {
241                         r = sd_journal_next(j);
242                         if (r < 0) {
243                                 log_error_errno(r, "Failed to move to next entry in journal: %m");
244                                 return CURL_READFUNC_ABORT;
245                         } else if (r == 0) {
246                                 if (u->input_event)
247                                         log_debug("No more entries, waiting for journal.");
248                                 else {
249                                         log_info("No more entries, closing journal.");
250                                         close_journal_input(u);
251                                 }
252
253                                 u->uploading = false;
254
255                                 break;
256                         }
257
258                         u->entry_state = ENTRY_CURSOR;
259                 }
260
261                 w = write_entry((char*)buf + filled, size * nmemb - filled, u);
262                 if (w < 0)
263                         return CURL_READFUNC_ABORT;
264                 filled += w;
265
266                 if (filled == 0) {
267                         log_error("Buffer space is too small to write entry.");
268                         return CURL_READFUNC_ABORT;
269                 } else if (u->entry_state != ENTRY_DONE)
270                         /* This means that all available space was used up */
271                         break;
272
273                 log_debug("Entry %zu (%s) has been uploaded.",
274                           u->entries_sent, u->current_cursor);
275         }
276
277         return filled;
278 }
279
280 void close_journal_input(Uploader *u) {
281         assert(u);
282
283         if (u->journal) {
284                 log_debug("Closing journal input.");
285
286                 sd_journal_close(u->journal);
287                 u->journal = NULL;
288         }
289         u->timeout = 0;
290 }
291
292 static int process_journal_input(Uploader *u, int skip) {
293         int r;
294
295         r = sd_journal_next_skip(u->journal, skip);
296         if (r < 0)
297                 return log_error_errno(r, "Failed to skip to next entry: %m");
298         else if (r < skip)
299                 return 0;
300
301         /* have data */
302         u->entry_state = ENTRY_CURSOR;
303         return start_upload(u, journal_input_callback, u);
304 }
305
306 int check_journal_input(Uploader *u) {
307         if (u->input_event) {
308                 int r;
309
310                 r = sd_journal_process(u->journal);
311                 if (r < 0) {
312                         log_error_errno(r, "Failed to process journal: %m");
313                         close_journal_input(u);
314                         return r;
315                 }
316
317                 if (r == SD_JOURNAL_NOP)
318                         return 0;
319         }
320
321         return process_journal_input(u, 1);
322 }
323
324 static int dispatch_journal_input(sd_event_source *event,
325                                   int fd,
326                                   uint32_t revents,
327                                   void *userp) {
328         Uploader *u = userp;
329
330         assert(u);
331
332         if (u->uploading) {
333                 log_warning("dispatch_journal_input called when uploading, ignoring.");
334                 return 0;
335         }
336
337         log_debug("Detected journal input, checking for new data.");
338         return check_journal_input(u);
339 }
340
341 int open_journal_for_upload(Uploader *u,
342                             sd_journal *j,
343                             const char *cursor,
344                             bool after_cursor,
345                             bool follow) {
346         int fd, r, events;
347
348         u->journal = j;
349
350         sd_journal_set_data_threshold(j, 0);
351
352         if (follow) {
353                 fd = sd_journal_get_fd(j);
354                 if (fd < 0)
355                         return log_error_errno(fd, "sd_journal_get_fd failed: %m");
356
357                 events = sd_journal_get_events(j);
358
359                 r = sd_journal_reliable_fd(j);
360                 assert(r >= 0);
361                 if (r > 0)
362                         u->timeout = -1;
363                 else
364                         u->timeout = JOURNAL_UPLOAD_POLL_TIMEOUT;
365
366                 r = sd_event_add_io(u->events, &u->input_event,
367                                     fd, events, dispatch_journal_input, u);
368                 if (r < 0)
369                         return log_error_errno(r, "Failed to register input event: %m");
370
371                 log_debug("Listening for journal events on fd:%d, timeout %d",
372                           fd, u->timeout == (uint64_t) -1 ? -1 : (int) u->timeout);
373         } else
374                 log_debug("Not listening for journal events.");
375
376         if (cursor) {
377                 r = sd_journal_seek_cursor(j, cursor);
378                 if (r < 0) {
379                         return log_error_errno(r, "Failed to seek to cursor %s: %m",
380                                                cursor);
381                 }
382         }
383
384         return process_journal_input(u, 1 + !!after_cursor);
385 }