src/journal-remote/journal-upload.c

   1 /*-*- Mode: C; c-basic-offset: 8; indent-tabs-mode: nil -*-*/
   2
   3 /***
   4   This file is part of systemd.
   5
   6   Copyright 2014 Zbigniew Jędrzejewski-Szmek
   7
   8   systemd is free software; you can redistribute it and/or modify it
   9   under the terms of the GNU Lesser General Public License as published by
  10   the Free Software Foundation; either version 2.1 of the License, or
  11   (at your option) any later version.
  12
  13   systemd is distributed in the hope that it will be useful, but
  14   WITHOUT ANY WARRANTY; without even the implied warranty of
  15   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  16   Lesser General Public License for more details.
  17
  18   You should have received a copy of the GNU Lesser General Public License
  19   along with systemd; If not, see <http://www.gnu.org/licenses/>.
  20 ***/
  21
  22 #include <stdio.h>
  23 #include <curl/curl.h>
  24 #include <sys/stat.h>
  25 #include <fcntl.h>
  26 #include <getopt.h>
  27
  28 #include "sd-daemon.h"
  29
  30 #include "log.h"
  31 #include "util.h"
  32 #include "build.h"
  33 #include "fileio.h"
  34 #include "journal-upload.h"
  35
  36 static const char* arg_url;
  37
  38 static void close_fd_input(Uploader *u);
  39
  40 static const char *arg_key = NULL;
  41 static const char *arg_cert = NULL;
  42 static const char *arg_trust = NULL;
  43
  44 static const char *arg_directory = NULL;
  45 static char **arg_file = NULL;
  46 static const char *arg_cursor = NULL;
  47 static bool arg_after_cursor = false;
  48 static int arg_journal_type = 0;
  49 static const char *arg_machine = NULL;
  50 static bool arg_merge = false;
  51 static int arg_follow = -1;
  52 static const char *arg_save_state = NULL;
  53
  54 #define SERVER_ANSWER_KEEP 2048
  55
  56 #define STATE_FILE "/var/lib/systemd/journal-upload/state"
  57
  58 #define easy_setopt(curl, opt, value, level, cmd)                       \
  59         {                                                               \
  60                 code = curl_easy_setopt(curl, opt, value);              \
  61                 if (code) {                                             \
  62                         log_full(level,                                 \
  63                                  "curl_easy_setopt " #opt " failed: %s", \
  64                                   curl_easy_strerror(code));            \
  65                         cmd;                                            \
  66                 }                                                       \
  67         }
  68
  69 static size_t output_callback(char *buf,
  70                               size_t size,
  71                               size_t nmemb,
  72                               void *userp) {
  73         Uploader *u = userp;
  74
  75         assert(u);
  76
  77         log_debug("The server answers (%zu bytes): %.*s",
  78                   size*nmemb, (int)(size*nmemb), buf);
  79
  80         if (nmemb && !u->answer) {
  81                 u->answer = strndup(buf, size*nmemb);
  82                 if (!u->answer)
  83                         log_warning("Failed to store server answer (%zu bytes): %s",
  84                                     size*nmemb, strerror(ENOMEM));
  85         }
  86
  87         return size * nmemb;
  88 }
  89
  90 static int update_cursor_state(Uploader *u) {
  91         _cleanup_free_ char *temp_path = NULL;
  92         _cleanup_fclose_ FILE *f = NULL;
  93         int r;
  94
  95         if (!u->state_file || !u->last_cursor)
  96                 return 0;
  97
  98         r = fopen_temporary(u->state_file, &f, &temp_path);
  99         if (r < 0)
 100                 goto finish;
 101
 102         fprintf(f,
 103                 "# This is private data. Do not parse.\n"
 104                 "LAST_CURSOR=%s\n",
 105                 u->last_cursor);
 106
 107         fflush(f);
 108
 109         if (ferror(f) || rename(temp_path, u->state_file) < 0) {
 110                 r = -errno;
 111                 unlink(u->state_file);
 112                 unlink(temp_path);
 113         }
 114
 115 finish:
 116         if (r < 0)
 117                 log_error("Failed to save state %s: %s", u->state_file, strerror(-r));
 118
 119         return r;
 120 }
 121
 122 static int load_cursor_state(Uploader *u) {
 123         int r;
 124
 125         if (!u->state_file)
 126                 return 0;
 127
 128         r = parse_env_file(u->state_file, NEWLINE,
 129                            "LAST_CURSOR",  &u->last_cursor,
 130                            NULL);
 131
 132         if (r < 0 && r != -ENOENT) {
 133                 log_error("Failed to read state file %s: %s",
 134                           u->state_file, strerror(-r));
 135                 return r;
 136         }
 137
 138         return 0;
 139 }
 140
 141
 142
 143 int start_upload(Uploader *u,
 144                  size_t (*input_callback)(void *ptr,
 145                                           size_t size,
 146                                           size_t nmemb,
 147                                           void *userdata),
 148                  void *data) {
 149         CURLcode code;
 150
 151         assert(u);
 152         assert(input_callback);
 153
 154         if (!u->header) {
 155                 struct curl_slist *h;
 156
 157                 h = curl_slist_append(NULL, "Content-Type: application/vnd.fdo.journal");
 158                 if (!h)
 159                         return log_oom();
 160
 161                 h = curl_slist_append(h, "Transfer-Encoding: chunked");
 162                 if (!h) {
 163                         curl_slist_free_all(h);
 164                         return log_oom();
 165                 }
 166
 167                 h = curl_slist_append(h, "Accept: text/plain");
 168                 if (!h) {
 169                         curl_slist_free_all(h);
 170                         return log_oom();
 171                 }
 172
 173                 u->header = h;
 174         }
 175
 176         if (!u->easy) {
 177                 CURL *curl;
 178
 179                 curl = curl_easy_init();
 180                 if (!curl) {
 181                         log_error("Call to curl_easy_init failed.");
 182                         return -ENOSR;
 183                 }
 184
 185                 /* tell it to POST to the URL */
 186                 easy_setopt(curl, CURLOPT_POST, 1L,
 187                             LOG_ERR, return -EXFULL);
 188
 189                 easy_setopt(curl, CURLOPT_ERRORBUFFER, &u->error,
 190                             LOG_ERR, return -EXFULL);
 191
 192                 /* set where to write to */
 193                 easy_setopt(curl, CURLOPT_WRITEFUNCTION, output_callback,
 194                             LOG_ERR, return -EXFULL);
 195
 196                 easy_setopt(curl, CURLOPT_WRITEDATA, data,
 197                             LOG_ERR, return -EXFULL);
 198
 199                 /* set where to read from */
 200                 easy_setopt(curl, CURLOPT_READFUNCTION, input_callback,
 201                             LOG_ERR, return -EXFULL);
 202
 203                 easy_setopt(curl, CURLOPT_READDATA, data,
 204                             LOG_ERR, return -EXFULL);
 205
 206                 /* use our special own mime type and chunked transfer */
 207                 easy_setopt(curl, CURLOPT_HTTPHEADER, u->header,
 208                             LOG_ERR, return -EXFULL);
 209
 210                 /* enable verbose for easier tracing */
 211                 easy_setopt(curl, CURLOPT_VERBOSE, 1L, LOG_WARNING, );
 212
 213                 easy_setopt(curl, CURLOPT_USERAGENT,
 214                             "systemd-journal-upload " PACKAGE_STRING,
 215                             LOG_WARNING, );
 216
 217                 if (arg_key) {
 218                         assert(arg_cert);
 219
 220                         easy_setopt(curl, CURLOPT_SSLKEY, arg_key,
 221                                     LOG_ERR, return -EXFULL);
 222                         easy_setopt(curl, CURLOPT_SSLCERT, arg_cert,
 223                                     LOG_ERR, return -EXFULL);
 224                 }
 225
 226                 if (arg_trust)
 227                         easy_setopt(curl, CURLOPT_CAINFO, arg_trust,
 228                                     LOG_ERR, return -EXFULL);
 229
 230                 if (arg_key || arg_trust)
 231                         easy_setopt(curl, CURLOPT_SSLVERSION, CURL_SSLVERSION_TLSv1,
 232                                     LOG_WARNING, );
 233
 234                 u->easy = curl;
 235         } else {
 236                 /* truncate the potential old error message */
 237                 u->error[0] = '\0';
 238
 239                 free(u->answer);
 240                 u->answer = 0;
 241         }
 242
 243         /* upload to this place */
 244         code = curl_easy_setopt(u->easy, CURLOPT_URL, u->url);
 245         if (code) {
 246                 log_error("curl_easy_setopt CURLOPT_URL failed: %s",
 247                           curl_easy_strerror(code));
 248                 return -EXFULL;
 249         }
 250
 251         u->uploading = true;
 252
 253         return 0;
 254 }
 255
 256 static size_t fd_input_callback(void *buf, size_t size, size_t nmemb, void *userp) {
 257         Uploader *u = userp;
 258
 259         ssize_t r;
 260
 261         assert(u);
 262         assert(nmemb <= SSIZE_MAX / size);
 263
 264         if (u->input < 0)
 265                 return 0;
 266
 267         r = read(u->input, buf, size * nmemb);
 268         log_debug("%s: allowed %zu, read %zu", __func__, size*nmemb, r);
 269
 270         if (r > 0)
 271                 return r;
 272
 273         u->uploading = false;
 274         if (r == 0) {
 275                 log_debug("Reached EOF");
 276                 close_fd_input(u);
 277                 return 0;
 278         } else {
 279                 log_error("Aborting transfer after read error on input: %m.");
 280                 return CURL_READFUNC_ABORT;
 281         }
 282 }
 283
 284 static void close_fd_input(Uploader *u) {
 285         assert(u);
 286
 287         if (u->input >= 0)
 288                 close_nointr(u->input);
 289         u->input = -1;
 290         u->timeout = 0;
 291 }
 292
 293 static int dispatch_fd_input(sd_event_source *event,
 294                              int fd,
 295                              uint32_t revents,
 296                              void *userp) {
 297         Uploader *u = userp;
 298
 299         assert(u);
 300         assert(fd >= 0);
 301
 302         if (revents & EPOLLHUP) {
 303                 log_debug("Received HUP");
 304                 close_fd_input(u);
 305                 return 0;
 306         }
 307
 308         if (!(revents & EPOLLIN)) {
 309                 log_warning("Unexpected poll event %"PRIu32".", revents);
 310                 return -EINVAL;
 311         }
 312
 313         if (u->uploading) {
 314                 log_warning("dispatch_fd_input called when uploading, ignoring.");
 315                 return 0;
 316         }
 317
 318         return start_upload(u, fd_input_callback, u);
 319 }
 320
 321 static int open_file_for_upload(Uploader *u, const char *filename) {
 322         int fd, r;
 323
 324         if (streq(filename, "-"))
 325                 fd = STDIN_FILENO;
 326         else {
 327                 fd = open(filename, O_RDONLY|O_CLOEXEC|O_NOCTTY);
 328                 if (fd < 0) {
 329                         log_error("Failed to open %s: %m", filename);
 330                         return -errno;
 331                 }
 332         }
 333
 334         u->input = fd;
 335
 336         if (arg_follow) {
 337                 r = sd_event_add_io(u->events, &u->input_event,
 338                                     fd, EPOLLIN, dispatch_fd_input, u);
 339                 if (r < 0) {
 340                         if (r != -EPERM || arg_follow > 0) {
 341                                 log_error("Failed to register input event: %s", strerror(-r));
 342                                 return r;
 343                         }
 344
 345                         /* Normal files should just be consumed without polling. */
 346                         r = start_upload(u, fd_input_callback, u);
 347                 }
 348         }
 349
 350         return r;
 351 }
 352
 353 static int dispatch_sigterm(sd_event_source *event,
 354                             const struct signalfd_siginfo *si,
 355                             void *userdata) {
 356         Uploader *u = userdata;
 357
 358         assert(u);
 359
 360         log_received_signal(LOG_INFO, si);
 361
 362         close_fd_input(u);
 363         close_journal_input(u);
 364
 365         sd_event_exit(u->events, 0);
 366         return 0;
 367 }
 368
 369 static int setup_signals(Uploader *u) {
 370         sigset_t mask;
 371         int r;
 372
 373         assert(u);
 374
 375         assert_se(sigemptyset(&mask) == 0);
 376         sigset_add_many(&mask, SIGINT, SIGTERM, -1);
 377         assert_se(sigprocmask(SIG_SETMASK, &mask, NULL) == 0);
 378
 379         r = sd_event_add_signal(u->events, &u->sigterm_event, SIGTERM, dispatch_sigterm, u);
 380         if (r < 0)
 381                 return r;
 382
 383         r = sd_event_add_signal(u->events, &u->sigint_event, SIGINT, dispatch_sigterm, u);
 384         if (r < 0)
 385                 return r;
 386
 387         return 0;
 388 }
 389
 390 static int setup_uploader(Uploader *u, const char *url, const char *state_file) {
 391         int r;
 392
 393         assert(u);
 394         assert(url);
 395
 396         memzero(u, sizeof(Uploader));
 397         u->input = -1;
 398
 399         if (!startswith(url, "http://") && !startswith(url, "https://"))
 400                 url = strappenda("https://", url);
 401
 402         u->url = strappend(url, "/upload");
 403         if (!u->url)
 404                 return log_oom();
 405
 406         u->state_file = state_file;
 407
 408         r = sd_event_default(&u->events);
 409         if (r < 0) {
 410                 log_error("sd_event_default failed: %s", strerror(-r));
 411                 return r;
 412         }
 413
 414         r = setup_signals(u);
 415         if (r < 0) {
 416                 log_error("Failed to set up signals: %s", strerror(-r));
 417                 return r;
 418         }
 419
 420         return load_cursor_state(u);
 421 }
 422
 423 static void destroy_uploader(Uploader *u) {
 424         assert(u);
 425
 426         curl_easy_cleanup(u->easy);
 427         curl_slist_free_all(u->header);
 428         free(u->answer);
 429
 430         free(u->last_cursor);
 431         free(u->current_cursor);
 432
 433         free(u->url);
 434
 435         u->input_event = sd_event_source_unref(u->input_event);
 436
 437         close_fd_input(u);
 438         close_journal_input(u);
 439
 440         sd_event_source_unref(u->sigterm_event);
 441         sd_event_source_unref(u->sigint_event);
 442         sd_event_unref(u->events);
 443 }
 444
 445 static int perform_upload(Uploader *u) {
 446         CURLcode code;
 447         long status;
 448
 449         assert(u);
 450
 451         code = curl_easy_perform(u->easy);
 452         if (code) {
 453                 log_error("Upload to %s failed: %.*s",
 454                           u->url,
 455                           u->error[0] ? (int) sizeof(u->error) : INT_MAX,
 456                           u->error[0] ? u->error : curl_easy_strerror(code));
 457                 return -EIO;
 458         }
 459
 460         code = curl_easy_getinfo(u->easy, CURLINFO_RESPONSE_CODE, &status);
 461         if (code) {
 462                 log_error("Failed to retrieve response code: %s",
 463                           curl_easy_strerror(code));
 464                 return -EUCLEAN;
 465         }
 466
 467         if (status >= 300) {
 468                 log_error("Upload to %s failed with code %lu: %s",
 469                           u->url, status, strna(u->answer));
 470                 return -EIO;
 471         } else if (status < 200) {
 472                 log_error("Upload to %s finished with unexpected code %lu: %s",
 473                           u->url, status, strna(u->answer));
 474                 return -EIO;
 475         } else
 476                 log_debug("Upload finished successfully with code %lu: %s",
 477                           status, strna(u->answer));
 478
 479         free(u->last_cursor);
 480         u->last_cursor = u->current_cursor;
 481         u->current_cursor = NULL;
 482
 483         return update_cursor_state(u);
 484 }
 485
 486 static void help(void) {
 487         printf("%s -u URL {FILE|-}...\n\n"
 488                "Upload journal events to a remote server.\n\n"
 489                "Options:\n"
 490                "  -u --url=URL             Upload to this address\n"
 491                "  --key=FILENAME           Specify key in PEM format\n"
 492                "  --cert=FILENAME          Specify certificate in PEM format\n"
 493                "  --trust=FILENAME         Specify CA certificate in PEM format\n"
 494                "     --system              Use the system journal\n"
 495                "     --user                Use the user journal for the current user\n"
 496                "  -m --merge               Use  all available journals\n"
 497                "  -M --machine=CONTAINER   Operate on local container\n"
 498                "  -D --directory=PATH      Use journal files from directory\n"
 499                "     --file=PATH           Use this journal file\n"
 500                "  --cursor=CURSOR          Start at the specified cursor\n"
 501                "  --after-cursor=CURSOR    Start after the specified cursor\n"
 502                "  --[no-]follow            Do [not] wait for input\n"
 503                "  --save-state[=FILE]      Save uploaded cursors (default \n"
 504                "                           " STATE_FILE ")\n"
 505                "  -h --help                Show this help and exit\n"
 506                "  --version                Print version string and exit\n"
 507                , program_invocation_short_name);
 508 }
 509
 510 static int parse_argv(int argc, char *argv[]) {
 511         enum {
 512                 ARG_VERSION = 0x100,
 513                 ARG_KEY,
 514                 ARG_CERT,
 515                 ARG_TRUST,
 516                 ARG_USER,
 517                 ARG_SYSTEM,
 518                 ARG_FILE,
 519                 ARG_CURSOR,
 520                 ARG_AFTER_CURSOR,
 521                 ARG_FOLLOW,
 522                 ARG_NO_FOLLOW,
 523                 ARG_SAVE_STATE,
 524         };
 525
 526         static const struct option options[] = {
 527                 { "help",         no_argument,       NULL, 'h'                },
 528                 { "version",      no_argument,       NULL, ARG_VERSION        },
 529                 { "url",          required_argument, NULL, 'u'                },
 530                 { "key",          required_argument, NULL, ARG_KEY            },
 531                 { "cert",         required_argument, NULL, ARG_CERT           },
 532                 { "trust",        required_argument, NULL, ARG_TRUST          },
 533                 { "system",       no_argument,       NULL, ARG_SYSTEM         },
 534                 { "user",         no_argument,       NULL, ARG_USER           },
 535                 { "merge",        no_argument,       NULL, 'm'                },
 536                 { "machine",      required_argument, NULL, 'M'                },
 537                 { "directory",    required_argument, NULL, 'D'                },
 538                 { "file",         required_argument, NULL, ARG_FILE           },
 539                 { "cursor",       required_argument, NULL, ARG_CURSOR         },
 540                 { "after-cursor", required_argument, NULL, ARG_AFTER_CURSOR   },
 541                 { "follow",       no_argument,       NULL, ARG_FOLLOW         },
 542                 { "no-follow",    no_argument,       NULL, ARG_NO_FOLLOW      },
 543                 { "save-state",   optional_argument, NULL, ARG_SAVE_STATE     },
 544                 {}
 545         };
 546
 547         int c, r;
 548
 549         assert(argc >= 0);
 550         assert(argv);
 551
 552         opterr = 0;
 553
 554         while ((c = getopt_long(argc, argv, "hu:mM:D:", options, NULL)) >= 0)
 555                 switch(c) {
 556                 case 'h':
 557                         help();
 558                         return 0 /* done */;
 559
 560                 case ARG_VERSION:
 561                         puts(PACKAGE_STRING);
 562                         puts(SYSTEMD_FEATURES);
 563                         return 0 /* done */;
 564
 565                 case 'u':
 566                         if (arg_url) {
 567                                 log_error("cannot use more than one --url");
 568                                 return -EINVAL;
 569                         }
 570
 571                         arg_url = optarg;
 572                         break;
 573
 574                 case ARG_KEY:
 575                         if (arg_key) {
 576                                 log_error("cannot use more than one --key");
 577                                 return -EINVAL;
 578                         }
 579
 580                         arg_key = optarg;
 581                         break;
 582
 583                 case ARG_CERT:
 584                         if (arg_cert) {
 585                                 log_error("cannot use more than one --cert");
 586                                 return -EINVAL;
 587                         }
 588
 589                         arg_cert = optarg;
 590                         break;
 591
 592                 case ARG_TRUST:
 593                         if (arg_trust) {
 594                                 log_error("cannot use more than one --trust");
 595                                 return -EINVAL;
 596                         }
 597
 598                         arg_trust = optarg;
 599                         break;
 600
 601                 case ARG_SYSTEM:
 602                         arg_journal_type |= SD_JOURNAL_SYSTEM;
 603                         break;
 604
 605                 case ARG_USER:
 606                         arg_journal_type |= SD_JOURNAL_CURRENT_USER;
 607                         break;
 608
 609                 case 'm':
 610                         arg_merge = true;
 611                         break;
 612
 613                 case 'M':
 614                         if (arg_machine) {
 615                                 log_error("cannot use more than one --machine/-M");
 616                                 return -EINVAL;
 617                         }
 618
 619                         arg_machine = optarg;
 620                         break;
 621
 622                 case 'D':
 623                         if (arg_directory) {
 624                                 log_error("cannot use more than one --directory/-D");
 625                                 return -EINVAL;
 626                         }
 627
 628                         arg_directory = optarg;
 629                         break;
 630
 631                 case ARG_FILE:
 632                         r = glob_extend(&arg_file, optarg);
 633                         if (r < 0) {
 634                                 log_error("Failed to add paths: %s", strerror(-r));
 635                                 return r;
 636                         };
 637                         break;
 638
 639                 case ARG_CURSOR:
 640                         if (arg_cursor) {
 641                                 log_error("cannot use more than one --cursor/--after-cursor");
 642                                 return -EINVAL;
 643                         }
 644
 645                         arg_cursor = optarg;
 646                         break;
 647
 648                 case ARG_AFTER_CURSOR:
 649                         if (arg_cursor) {
 650                                 log_error("cannot use more than one --cursor/--after-cursor");
 651                                 return -EINVAL;
 652                         }
 653
 654                         arg_cursor = optarg;
 655                         arg_after_cursor = true;
 656                         break;
 657
 658                 case ARG_FOLLOW:
 659                         arg_follow = true;
 660                         break;
 661
 662                 case ARG_NO_FOLLOW:
 663                         arg_follow = false;
 664                         break;
 665
 666                 case ARG_SAVE_STATE:
 667                         arg_save_state = optarg ?: STATE_FILE;
 668                         break;
 669
 670                 case '?':
 671                         log_error("Unknown option %s.", argv[optind-1]);
 672                         return -EINVAL;
 673
 674                 case ':':
 675                         log_error("Missing argument to %s.", argv[optind-1]);
 676                         return -EINVAL;
 677
 678                 default:
 679                         assert_not_reached("Unhandled option code.");
 680                 }
 681
 682         if (!arg_url) {
 683                 log_error("Required --url/-u option missing.");
 684                 return -EINVAL;
 685         }
 686
 687         if (!!arg_key != !!arg_cert) {
 688                 log_error("Options --key and --cert must be used together.");
 689                 return -EINVAL;
 690         }
 691
 692         if (optind < argc && (arg_directory || arg_file || arg_machine || arg_journal_type)) {
 693                 log_error("Input arguments make no sense with journal input.");
 694                 return -EINVAL;
 695         }
 696
 697         return 1;
 698 }
 699
 700 static int open_journal(sd_journal **j) {
 701         int r;
 702
 703         if (arg_directory)
 704                 r = sd_journal_open_directory(j, arg_directory, arg_journal_type);
 705         else if (arg_file)
 706                 r = sd_journal_open_files(j, (const char**) arg_file, 0);
 707         else if (arg_machine)
 708                 r = sd_journal_open_container(j, arg_machine, 0);
 709         else
 710                 r = sd_journal_open(j, !arg_merge*SD_JOURNAL_LOCAL_ONLY + arg_journal_type);
 711         if (r < 0)
 712                 log_error("Failed to open %s: %s",
 713                           arg_directory ? arg_directory : arg_file ? "files" : "journal",
 714                           strerror(-r));
 715         return r;
 716 }
 717
 718 int main(int argc, char **argv) {
 719         Uploader u;
 720         int r;
 721         bool use_journal;
 722
 723         log_show_color(true);
 724         log_parse_environment();
 725
 726         r = parse_argv(argc, argv);
 727         if (r <= 0)
 728                 goto finish;
 729
 730         r = setup_uploader(&u, arg_url, arg_save_state);
 731         if (r < 0)
 732                 goto cleanup;
 733
 734         sd_event_set_watchdog(u.events, true);
 735
 736         log_debug("%s running as pid "PID_FMT,
 737                   program_invocation_short_name, getpid());
 738
 739         use_journal = optind >= argc;
 740         if (use_journal) {
 741                 sd_journal *j;
 742                 r = open_journal(&j);
 743                 if (r < 0)
 744                         goto finish;
 745                 r = open_journal_for_upload(&u, j,
 746                                             arg_cursor ?: u.last_cursor,
 747                                             arg_cursor ? arg_after_cursor : true,
 748                                             !!arg_follow);
 749                 if (r < 0)
 750                         goto finish;
 751         }
 752
 753         sd_notify(false,
 754                   "READY=1\n"
 755                   "STATUS=Processing input...");
 756
 757         while (true) {
 758                 if (use_journal) {
 759                         if (!u.journal)
 760                                 break;
 761
 762                         r = check_journal_input(&u);
 763                 } else if (u.input < 0 && !use_journal) {
 764                         if (optind >= argc)
 765                                 break;
 766
 767                         log_debug("Using %s as input.", argv[optind]);
 768                         r = open_file_for_upload(&u, argv[optind++]);
 769                 }
 770                 if (r < 0)
 771                         goto cleanup;
 772
 773                 r = sd_event_get_state(u.events);
 774                 if (r < 0)
 775                         break;
 776                 if (r == SD_EVENT_FINISHED)
 777                         break;
 778
 779                 if (u.uploading) {
 780                         r = perform_upload(&u);
 781                         if (r < 0)
 782                                 break;
 783                 }
 784
 785                 r = sd_event_run(u.events, u.timeout);
 786                 if (r < 0) {
 787                         log_error("Failed to run event loop: %s", strerror(-r));
 788                         break;
 789                 }
 790         }
 791
 792 cleanup:
 793         sd_notify(false, "STATUS=Shutting down...");
 794         destroy_uploader(&u);
 795
 796 finish:
 797         return r == 0 ? EXIT_SUCCESS : EXIT_FAILURE;
 798 }