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

   1 /* SPDX-License-Identifier: LGPL-2.1+ */
   2 /***
   3   Copyright © 2012 Zbigniew Jędrzejewski-Szmek
   4 ***/
   5
   6 #include "alloc-util.h"
   7 #include "journal-remote.h"
   8
   9 static int do_rotate(JournalFile **f, bool compress, bool seal) {
  10         int r = journal_file_rotate(f, compress, (uint64_t) -1, seal, NULL);
  11         if (r < 0) {
  12                 if (*f)
  13                         log_error_errno(r, "Failed to rotate %s: %m", (*f)->path);
  14                 else
  15                         log_error_errno(r, "Failed to create rotated journal: %m");
  16         }
  17
  18         return r;
  19 }
  20
  21 Writer* writer_new(RemoteServer *server) {
  22         Writer *w;
  23
  24         w = new0(Writer, 1);
  25         if (!w)
  26                 return NULL;
  27
  28         memset(&w->metrics, 0xFF, sizeof(w->metrics));
  29
  30         w->mmap = mmap_cache_new();
  31         if (!w->mmap)
  32                 return mfree(w);
  33
  34         w->n_ref = 1;
  35         w->server = server;
  36
  37         return w;
  38 }
  39
  40 Writer* writer_free(Writer *w) {
  41         if (!w)
  42                 return NULL;
  43
  44         if (w->journal) {
  45                 log_debug("Closing journal file %s.", w->journal->path);
  46                 journal_file_close(w->journal);
  47         }
  48
  49         if (w->server && w->hashmap_key)
  50                 hashmap_remove(w->server->writers, w->hashmap_key);
  51
  52         free(w->hashmap_key);
  53
  54         if (w->mmap)
  55                 mmap_cache_unref(w->mmap);
  56
  57         return mfree(w);
  58 }
  59
  60 Writer* writer_unref(Writer *w) {
  61         if (w && (-- w->n_ref <= 0))
  62                 writer_free(w);
  63
  64         return NULL;
  65 }
  66
  67 Writer* writer_ref(Writer *w) {
  68         if (w)
  69                 assert_se(++ w->n_ref >= 2);
  70
  71         return w;
  72 }
  73
  74 int writer_write(Writer *w,
  75                  struct iovec_wrapper *iovw,
  76                  dual_timestamp *ts,
  77                  bool compress,
  78                  bool seal) {
  79         int r;
  80
  81         assert(w);
  82         assert(iovw);
  83         assert(iovw->count > 0);
  84
  85         if (journal_file_rotate_suggested(w->journal, 0)) {
  86                 log_info("%s: Journal header limits reached or header out-of-date, rotating",
  87                          w->journal->path);
  88                 r = do_rotate(&w->journal, compress, seal);
  89                 if (r < 0)
  90                         return r;
  91         }
  92
  93         r = journal_file_append_entry(w->journal, ts, NULL,
  94                                       iovw->iovec, iovw->count,
  95                                       &w->seqnum, NULL, NULL);
  96         if (r >= 0) {
  97                 if (w->server)
  98                         w->server->event_count += 1;
  99                 return 0;
 100         } else if (r == -EBADMSG)
 101                 return r;
 102
 103         log_debug_errno(r, "%s: Write failed, rotating: %m", w->journal->path);
 104         r = do_rotate(&w->journal, compress, seal);
 105         if (r < 0)
 106                 return r;
 107         else
 108                 log_debug("%s: Successfully rotated journal", w->journal->path);
 109
 110         log_debug("Retrying write.");
 111         r = journal_file_append_entry(w->journal, ts, NULL,
 112                                       iovw->iovec, iovw->count,
 113                                       &w->seqnum, NULL, NULL);
 114         if (r < 0)
 115                 return r;
 116
 117         if (w->server)
 118                 w->server->event_count += 1;
 119         return 0;
 120 }