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