src/journal-remote/journal-remote-write.c

   1 /* SPDX-License-Identifier: LGPL-2.1-or-later */
   2
   3 #include <libgen.h>
   4
   5 #include "alloc-util.h"
   6 #include "journal-remote.h"
   7 #include "path-util.h"
   8 #include "stat-util.h"
   9
  10 static int do_rotate(JournalFile **f, MMapCache *m, JournalFileFlags file_flags) {
  11         int r;
  12
  13         r = journal_file_rotate(f, m, file_flags, UINT64_MAX, NULL);
  14         if (r < 0) {
  15                 if (*f)
  16                         log_error_errno(r, "Failed to rotate %s: %m", (*f)->path);
  17                 else
  18                         log_error_errno(r, "Failed to create rotated journal: %m");
  19         }
  20
  21         return r;
  22 }
  23
  24 int writer_new(RemoteServer *server, Writer **ret) {
  25         _cleanup_(writer_unrefp) Writer *w = NULL;
  26         int r;
  27
  28         assert(server);
  29         assert(ret);
  30
  31         w = new(Writer, 1);
  32         if (!w)
  33                 return -ENOMEM;
  34
  35         *w = (Writer) {
  36                 .n_ref = 1,
  37                 .metrics = server->metrics,
  38                 .server = server,
  39         };
  40
  41         w->mmap = mmap_cache_new();
  42         if (!w->mmap)
  43                 return -ENOMEM;
  44
  45         if (is_dir(server->output, /* follow = */ true) > 0) {
  46                 w->output = strdup(server->output);
  47                 if (!w->output)
  48                         return -ENOMEM;
  49         } else {
  50                 r = path_extract_directory(server->output, &w->output);
  51                 if (r < 0)
  52                         return r;
  53         }
  54
  55         *ret = TAKE_PTR(w);
  56         return 0;
  57 }
  58
  59 static Writer* writer_free(Writer *w) {
  60         if (!w)
  61                 return NULL;
  62
  63         if (w->journal) {
  64                 log_debug("Closing journal file %s.", w->journal->path);
  65                 journal_file_offline_close(w->journal);
  66         }
  67
  68         if (w->server && w->hashmap_key)
  69                 hashmap_remove(w->server->writers, w->hashmap_key);
  70
  71         free(w->hashmap_key);
  72
  73         if (w->mmap)
  74                 mmap_cache_unref(w->mmap);
  75
  76         free(w->output);
  77
  78         return mfree(w);
  79 }
  80
  81 DEFINE_TRIVIAL_REF_UNREF_FUNC(Writer, writer, writer_free);
  82
  83 int writer_write(Writer *w,
  84                  const struct iovec_wrapper *iovw,
  85                  const dual_timestamp *ts,
  86                  const sd_id128_t *boot_id,
  87                  JournalFileFlags file_flags) {
  88         int r;
  89
  90         assert(w);
  91         assert(!iovw_isempty(iovw));
  92
  93         if (journal_file_rotate_suggested(w->journal, 0, LOG_DEBUG)) {
  94                 log_info("%s: Journal header limits reached or header out-of-date, rotating",
  95                          w->journal->path);
  96                 r = do_rotate(&w->journal, w->mmap, file_flags);
  97                 if (r < 0)
  98                         return r;
  99                 r = journal_directory_vacuum(w->output, w->metrics.max_use, w->metrics.n_max_files, 0, NULL, /* verbose = */ true);
 100                 if (r < 0)
 101                         return r;
 102         }
 103
 104         r = journal_file_append_entry(
 105                         w->journal,
 106                         ts,
 107                         boot_id,
 108                         iovw->iovec,
 109                         iovw->count,
 110                         &w->seqnum,
 111                         /* seqnum_id= */ NULL,
 112                         /* ret_object= */ NULL,
 113                         /* ret_offset= */ NULL);
 114         if (r >= 0) {
 115                 if (w->server)
 116                         w->server->event_count += 1;
 117                 return 0;
 118         } else if (r == -EBADMSG)
 119                 return r;
 120
 121         log_debug_errno(r, "%s: Write failed, rotating: %m", w->journal->path);
 122         r = do_rotate(&w->journal, w->mmap, file_flags);
 123         if (r < 0)
 124                 return r;
 125         else
 126                 log_debug("%s: Successfully rotated journal", w->journal->path);
 127         r = journal_directory_vacuum(w->output, w->metrics.max_use, w->metrics.n_max_files, 0, NULL, /* verbose = */ true);
 128         if (r < 0)
 129                 return r;
 130
 131         log_debug("Retrying write.");
 132         r = journal_file_append_entry(
 133                         w->journal,
 134                         ts,
 135                         boot_id,
 136                         iovw->iovec, iovw->count,
 137                         &w->seqnum,
 138                         /* seqnum_id= */ NULL,
 139                         /* ret_object= */ NULL,
 140                         /* ret_offset= */ NULL);
 141         if (r < 0)
 142                 return r;
 143
 144         if (w->server)
 145                 w->server->event_count += 1;
 146         return 0;
 147 }