src/basic/journal-importer.c

   1 /* SPDX-License-Identifier: LGPL-2.1+ */
   2
   3 #include <errno.h>
   4 #include <unistd.h>
   5
   6 #include "alloc-util.h"
   7 #include "escape.h"
   8 #include "fd-util.h"
   9 #include "io-util.h"
  10 #include "journal-file.h"
  11 #include "journal-importer.h"
  12 #include "journal-util.h"
  13 #include "parse-util.h"
  14 #include "string-util.h"
  15 #include "unaligned.h"
  16
  17 enum {
  18         IMPORTER_STATE_LINE = 0,    /* waiting to read, or reading line */
  19         IMPORTER_STATE_DATA_START,  /* reading binary data header */
  20         IMPORTER_STATE_DATA,        /* reading binary data */
  21         IMPORTER_STATE_DATA_FINISH, /* expecting newline */
  22         IMPORTER_STATE_EOF,         /* done */
  23 };
  24
  25 static int iovw_put(struct iovec_wrapper *iovw, void* data, size_t len) {
  26         if (!GREEDY_REALLOC(iovw->iovec, iovw->size_bytes, iovw->count + 1))
  27                 return log_oom();
  28
  29         iovw->iovec[iovw->count++] = IOVEC_MAKE(data, len);
  30         return 0;
  31 }
  32
  33 static void iovw_free_contents(struct iovec_wrapper *iovw) {
  34         iovw->iovec = mfree(iovw->iovec);
  35         iovw->size_bytes = iovw->count = 0;
  36 }
  37
  38 static void iovw_rebase(struct iovec_wrapper *iovw, char *old, char *new) {
  39         size_t i;
  40
  41         for (i = 0; i < iovw->count; i++)
  42                 iovw->iovec[i].iov_base = (char*) iovw->iovec[i].iov_base - old + new;
  43 }
  44
  45 size_t iovw_size(struct iovec_wrapper *iovw) {
  46         size_t n = 0, i;
  47
  48         for (i = 0; i < iovw->count; i++)
  49                 n += iovw->iovec[i].iov_len;
  50
  51         return n;
  52 }
  53
  54 void journal_importer_cleanup(JournalImporter *imp) {
  55         if (imp->fd >= 0 && !imp->passive_fd) {
  56                 log_debug("Closing %s (fd=%d)", imp->name ?: "importer", imp->fd);
  57                 safe_close(imp->fd);
  58         }
  59
  60         free(imp->name);
  61         free(imp->buf);
  62         iovw_free_contents(&imp->iovw);
  63 }
  64
  65 static char* realloc_buffer(JournalImporter *imp, size_t size) {
  66         char *b, *old = imp->buf;
  67
  68         b = GREEDY_REALLOC(imp->buf, imp->size, size);
  69         if (!b)
  70                 return NULL;
  71
  72         iovw_rebase(&imp->iovw, old, imp->buf);
  73
  74         return b;
  75 }
  76
  77 static int get_line(JournalImporter *imp, char **line, size_t *size) {
  78         ssize_t n;
  79         char *c = NULL;
  80
  81         assert(imp);
  82         assert(imp->state == IMPORTER_STATE_LINE);
  83         assert(imp->offset <= imp->filled);
  84         assert(imp->filled <= imp->size);
  85         assert(!imp->buf || imp->size > 0);
  86         assert(imp->fd >= 0);
  87
  88         for (;;) {
  89                 if (imp->buf) {
  90                         size_t start = MAX(imp->scanned, imp->offset);
  91
  92                         c = memchr(imp->buf + start, '\n',
  93                                    imp->filled - start);
  94                         if (c != NULL)
  95                                 break;
  96                 }
  97
  98                 imp->scanned = imp->filled;
  99                 if (imp->scanned >= DATA_SIZE_MAX) {
 100                         log_error("Entry is bigger than %u bytes.", DATA_SIZE_MAX);
 101                         return -E2BIG;
 102                 }
 103
 104                 if (imp->passive_fd)
 105                         /* we have to wait for some data to come to us */
 106                         return -EAGAIN;
 107
 108                 /* We know that imp->filled is at most DATA_SIZE_MAX, so if
 109                    we reallocate it, we'll increase the size at least a bit. */
 110                 assert_cc(DATA_SIZE_MAX < ENTRY_SIZE_MAX);
 111                 if (imp->size - imp->filled < LINE_CHUNK &&
 112                     !realloc_buffer(imp, MIN(imp->filled + LINE_CHUNK, ENTRY_SIZE_MAX)))
 113                                 return log_oom();
 114
 115                 assert(imp->buf);
 116                 assert(imp->size - imp->filled >= LINE_CHUNK ||
 117                        imp->size == ENTRY_SIZE_MAX);
 118
 119                 n = read(imp->fd,
 120                          imp->buf + imp->filled,
 121                          imp->size - imp->filled);
 122                 if (n < 0) {
 123                         if (errno != EAGAIN)
 124                                 log_error_errno(errno, "read(%d, ..., %zu): %m",
 125                                                 imp->fd,
 126                                                 imp->size - imp->filled);
 127                         return -errno;
 128                 } else if (n == 0)
 129                         return 0;
 130
 131                 imp->filled += n;
 132         }
 133
 134         *line = imp->buf + imp->offset;
 135         *size = c + 1 - imp->buf - imp->offset;
 136         imp->offset += *size;
 137
 138         return 1;
 139 }
 140
 141 static int fill_fixed_size(JournalImporter *imp, void **data, size_t size) {
 142
 143         assert(imp);
 144         assert(IN_SET(imp->state, IMPORTER_STATE_DATA_START, IMPORTER_STATE_DATA, IMPORTER_STATE_DATA_FINISH));
 145         assert(size <= DATA_SIZE_MAX);
 146         assert(imp->offset <= imp->filled);
 147         assert(imp->filled <= imp->size);
 148         assert(imp->buf || imp->size == 0);
 149         assert(!imp->buf || imp->size > 0);
 150         assert(imp->fd >= 0);
 151         assert(data);
 152
 153         while (imp->filled - imp->offset < size) {
 154                 int n;
 155
 156                 if (imp->passive_fd)
 157                         /* we have to wait for some data to come to us */
 158                         return -EAGAIN;
 159
 160                 if (!realloc_buffer(imp, imp->offset + size))
 161                         return log_oom();
 162
 163                 n = read(imp->fd, imp->buf + imp->filled,
 164                          imp->size - imp->filled);
 165                 if (n < 0) {
 166                         if (errno != EAGAIN)
 167                                 log_error_errno(errno, "read(%d, ..., %zu): %m", imp->fd,
 168                                                 imp->size - imp->filled);
 169                         return -errno;
 170                 } else if (n == 0)
 171                         return 0;
 172
 173                 imp->filled += n;
 174         }
 175
 176         *data = imp->buf + imp->offset;
 177         imp->offset += size;
 178
 179         return 1;
 180 }
 181
 182 static int get_data_size(JournalImporter *imp) {
 183         int r;
 184         void *data;
 185
 186         assert(imp);
 187         assert(imp->state == IMPORTER_STATE_DATA_START);
 188         assert(imp->data_size == 0);
 189
 190         r = fill_fixed_size(imp, &data, sizeof(uint64_t));
 191         if (r <= 0)
 192                 return r;
 193
 194         imp->data_size = unaligned_read_le64(data);
 195         if (imp->data_size > DATA_SIZE_MAX) {
 196                 log_error("Stream declares field with size %zu > DATA_SIZE_MAX = %u",
 197                           imp->data_size, DATA_SIZE_MAX);
 198                 return -EINVAL;
 199         }
 200         if (imp->data_size == 0)
 201                 log_warning("Binary field with zero length");
 202
 203         return 1;
 204 }
 205
 206 static int get_data_data(JournalImporter *imp, void **data) {
 207         int r;
 208
 209         assert(imp);
 210         assert(data);
 211         assert(imp->state == IMPORTER_STATE_DATA);
 212
 213         r = fill_fixed_size(imp, data, imp->data_size);
 214         if (r <= 0)
 215                 return r;
 216
 217         return 1;
 218 }
 219
 220 static int get_data_newline(JournalImporter *imp) {
 221         int r;
 222         char *data;
 223
 224         assert(imp);
 225         assert(imp->state == IMPORTER_STATE_DATA_FINISH);
 226
 227         r = fill_fixed_size(imp, (void**) &data, 1);
 228         if (r <= 0)
 229                 return r;
 230
 231         assert(data);
 232         if (*data != '\n') {
 233                 char buf[4];
 234                 int l;
 235
 236                 l = cescape_char(*data, buf);
 237                 log_error("Expected newline, got '%.*s'", l, buf);
 238                 return -EINVAL;
 239         }
 240
 241         return 1;
 242 }
 243
 244 static int process_special_field(JournalImporter *imp, char *line) {
 245         const char *value;
 246         char buf[CELLESCAPE_DEFAULT_LENGTH];
 247         int r;
 248
 249         assert(line);
 250
 251         value = startswith(line, "__CURSOR=");
 252         if (value)
 253                 /* ignore __CURSOR */
 254                 return 1;
 255
 256         value = startswith(line, "__REALTIME_TIMESTAMP=");
 257         if (value) {
 258                 uint64_t x;
 259
 260                 r = safe_atou64(value, &x);
 261                 if (r < 0)
 262                         return log_warning_errno(r, "Failed to parse __REALTIME_TIMESTAMP '%s': %m",
 263                                                  cellescape(buf, sizeof buf, value));
 264                 else if (!VALID_REALTIME(x)) {
 265                         log_warning("__REALTIME_TIMESTAMP out of range, ignoring: %"PRIu64, x);
 266                         return -ERANGE;
 267                 }
 268
 269                 imp->ts.realtime = x;
 270                 return 1;
 271         }
 272
 273         value = startswith(line, "__MONOTONIC_TIMESTAMP=");
 274         if (value) {
 275                 uint64_t x;
 276
 277                 r = safe_atou64(value, &x);
 278                 if (r < 0)
 279                         return log_warning_errno(r, "Failed to parse __MONOTONIC_TIMESTAMP '%s': %m",
 280                                                  cellescape(buf, sizeof buf, value));
 281                 else if (!VALID_MONOTONIC(x)) {
 282                         log_warning("__MONOTONIC_TIMESTAMP out of range, ignoring: %"PRIu64, x);
 283                         return -ERANGE;
 284                 }
 285
 286                 imp->ts.monotonic = x;
 287                 return 1;
 288         }
 289
 290         /* Just a single underline, but it needs special treatment too. */
 291         value = startswith(line, "_BOOT_ID=");
 292         if (value) {
 293                 r = sd_id128_from_string(value, &imp->boot_id);
 294                 if (r < 0)
 295                         return log_warning_errno(r, "Failed to parse _BOOT_ID '%s': %m",
 296                                                  cellescape(buf, sizeof buf, value));
 297
 298                 /* store the field in the usual fashion too */
 299                 return 0;
 300         }
 301
 302         value = startswith(line, "__");
 303         if (value) {
 304                 log_notice("Unknown dunder line __%s, ignoring.", cellescape(buf, sizeof buf, value));
 305                 return 1;
 306         }
 307
 308         /* no dunder */
 309         return 0;
 310 }
 311
 312 int journal_importer_process_data(JournalImporter *imp) {
 313         int r;
 314
 315         switch(imp->state) {
 316         case IMPORTER_STATE_LINE: {
 317                 char *line, *sep;
 318                 size_t n = 0;
 319
 320                 assert(imp->data_size == 0);
 321
 322                 r = get_line(imp, &line, &n);
 323                 if (r < 0)
 324                         return r;
 325                 if (r == 0) {
 326                         imp->state = IMPORTER_STATE_EOF;
 327                         return 0;
 328                 }
 329                 assert(n > 0);
 330                 assert(line[n-1] == '\n');
 331
 332                 if (n == 1) {
 333                         log_trace("Received empty line, event is ready");
 334                         return 1;
 335                 }
 336
 337                 /* MESSAGE=xxx\n
 338                    or
 339                    COREDUMP\n
 340                    LLLLLLLL0011223344...\n
 341                 */
 342                 sep = memchr(line, '=', n);
 343                 if (sep) {
 344                         /* chomp newline */
 345                         n--;
 346
 347                         if (!journal_field_valid(line, sep - line, true)) {
 348                                 char buf[64], *t;
 349
 350                                 t = strndupa(line, sep - line);
 351                                 log_debug("Ignoring invalid field: \"%s\"",
 352                                           cellescape(buf, sizeof buf, t));
 353
 354                                 return 0;
 355                         }
 356
 357                         line[n] = '\0';
 358                         r = process_special_field(imp, line);
 359                         if (r != 0)
 360                                 return r < 0 ? r : 0;
 361
 362                         r = iovw_put(&imp->iovw, line, n);
 363                         if (r < 0)
 364                                 return r;
 365                 } else {
 366                         /* replace \n with = */
 367                         line[n-1] = '=';
 368
 369                         imp->field_len = n;
 370                         imp->state = IMPORTER_STATE_DATA_START;
 371
 372                         /* we cannot put the field in iovec until we have all data */
 373                 }
 374
 375                 log_trace("Received: %.*s (%s)", (int) n, line, sep ? "text" : "binary");
 376
 377                 return 0; /* continue */
 378         }
 379
 380         case IMPORTER_STATE_DATA_START:
 381                 assert(imp->data_size == 0);
 382
 383                 r = get_data_size(imp);
 384                 // log_debug("get_data_size() -> %d", r);
 385                 if (r < 0)
 386                         return r;
 387                 if (r == 0) {
 388                         imp->state = IMPORTER_STATE_EOF;
 389                         return 0;
 390                 }
 391
 392                 imp->state = imp->data_size > 0 ?
 393                         IMPORTER_STATE_DATA : IMPORTER_STATE_DATA_FINISH;
 394
 395                 return 0; /* continue */
 396
 397         case IMPORTER_STATE_DATA: {
 398                 void *data;
 399                 char *field;
 400
 401                 assert(imp->data_size > 0);
 402
 403                 r = get_data_data(imp, &data);
 404                 // log_debug("get_data_data() -> %d", r);
 405                 if (r < 0)
 406                         return r;
 407                 if (r == 0) {
 408                         imp->state = IMPORTER_STATE_EOF;
 409                         return 0;
 410                 }
 411
 412                 assert(data);
 413
 414                 field = (char*) data - sizeof(uint64_t) - imp->field_len;
 415                 memmove(field + sizeof(uint64_t), field, imp->field_len);
 416
 417                 r = iovw_put(&imp->iovw, field + sizeof(uint64_t), imp->field_len + imp->data_size);
 418                 if (r < 0)
 419                         return r;
 420
 421                 imp->state = IMPORTER_STATE_DATA_FINISH;
 422
 423                 return 0; /* continue */
 424         }
 425
 426         case IMPORTER_STATE_DATA_FINISH:
 427                 r = get_data_newline(imp);
 428                 // log_debug("get_data_newline() -> %d", r);
 429                 if (r < 0)
 430                         return r;
 431                 if (r == 0) {
 432                         imp->state = IMPORTER_STATE_EOF;
 433                         return 0;
 434                 }
 435
 436                 imp->data_size = 0;
 437                 imp->state = IMPORTER_STATE_LINE;
 438
 439                 return 0; /* continue */
 440         default:
 441                 assert_not_reached("wtf?");
 442         }
 443 }
 444
 445 int journal_importer_push_data(JournalImporter *imp, const char *data, size_t size) {
 446         assert(imp);
 447         assert(imp->state != IMPORTER_STATE_EOF);
 448
 449         if (!realloc_buffer(imp, imp->filled + size)) {
 450                 log_error("Failed to store received data of size %zu "
 451                           "(in addition to existing %zu bytes with %zu filled): %s",
 452                           size, imp->size, imp->filled, strerror(ENOMEM));
 453                 return -ENOMEM;
 454         }
 455
 456         memcpy(imp->buf + imp->filled, data, size);
 457         imp->filled += size;
 458
 459         return 0;
 460 }
 461
 462 void journal_importer_drop_iovw(JournalImporter *imp) {
 463         size_t remain, target;
 464
 465         /* This function drops processed data that along with the iovw that points at it */
 466
 467         iovw_free_contents(&imp->iovw);
 468
 469         /* possibly reset buffer position */
 470         remain = imp->filled - imp->offset;
 471
 472         if (remain == 0) /* no brainer */
 473                 imp->offset = imp->scanned = imp->filled = 0;
 474         else if (imp->offset > imp->size - imp->filled &&
 475                  imp->offset > remain) {
 476                 memcpy(imp->buf, imp->buf + imp->offset, remain);
 477                 imp->offset = imp->scanned = 0;
 478                 imp->filled = remain;
 479         }
 480
 481         target = imp->size;
 482         while (target > 16 * LINE_CHUNK && imp->filled < target / 2)
 483                 target /= 2;
 484         if (target < imp->size) {
 485                 char *tmp;
 486
 487                 tmp = realloc(imp->buf, target);
 488                 if (!tmp)
 489                         log_warning("Failed to reallocate buffer to (smaller) size %zu",
 490                                     target);
 491                 else {
 492                         log_debug("Reallocated buffer from %zu to %zu bytes",
 493                                   imp->size, target);
 494                         imp->buf = tmp;
 495                         imp->size = target;
 496                 }
 497         }
 498 }
 499
 500 bool journal_importer_eof(const JournalImporter *imp) {
 501         return imp->state == IMPORTER_STATE_EOF;
 502 }