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

   1 /***
   2   This file is part of systemd.
   3
   4   Copyright 2012 Zbigniew Jędrzejewski-Szmek
   5
   6   systemd is free software; you can redistribute it and/or modify it
   7   under the terms of the GNU Lesser General Public License as published by
   8   the Free Software Foundation; either version 2.1 of the License, or
   9   (at your option) any later version.
  10
  11   systemd is distributed in the hope that it will be useful, but
  12   WITHOUT ANY WARRANTY; without even the implied warranty of
  13   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  14   Lesser General Public License for more details.
  15
  16   You should have received a copy of the GNU Lesser General Public License
  17   along with systemd; If not, see <http://www.gnu.org/licenses/>.
  18 ***/
  19
  20 #include "alloc-util.h"
  21 #include "journal-remote.h"
  22
  23 static int do_rotate(JournalFile **f, bool compress, bool seal) {
  24         int r = journal_file_rotate(f, compress, seal, NULL);
  25         if (r < 0) {
  26                 if (*f)
  27                         log_error_errno(r, "Failed to rotate %s: %m", (*f)->path);
  28                 else
  29                         log_error_errno(r, "Failed to create rotated journal: %m");
  30         }
  31
  32         return r;
  33 }
  34
  35 Writer* writer_new(RemoteServer *server) {
  36         Writer *w;
  37
  38         w = new0(Writer, 1);
  39         if (!w)
  40                 return NULL;
  41
  42         memset(&w->metrics, 0xFF, sizeof(w->metrics));
  43
  44         w->mmap = mmap_cache_new();
  45         if (!w->mmap)
  46                 return mfree(w);
  47
  48         w->n_ref = 1;
  49         w->server = server;
  50
  51         return w;
  52 }
  53
  54 Writer* writer_free(Writer *w) {
  55         if (!w)
  56                 return NULL;
  57
  58         if (w->journal) {
  59                 log_debug("Closing journal file %s.", w->journal->path);
  60                 journal_file_close(w->journal);
  61         }
  62
  63         if (w->server && w->hashmap_key)
  64                 hashmap_remove(w->server->writers, w->hashmap_key);
  65
  66         free(w->hashmap_key);
  67
  68         if (w->mmap)
  69                 mmap_cache_unref(w->mmap);
  70
  71         return mfree(w);
  72 }
  73
  74 Writer* writer_unref(Writer *w) {
  75         if (w && (-- w->n_ref <= 0))
  76                 writer_free(w);
  77
  78         return NULL;
  79 }
  80
  81 Writer* writer_ref(Writer *w) {
  82         if (w)
  83                 assert_se(++ w->n_ref >= 2);
  84
  85         return w;
  86 }
  87
  88 int writer_write(Writer *w,
  89                  struct iovec_wrapper *iovw,
  90                  dual_timestamp *ts,
  91                  bool compress,
  92                  bool seal) {
  93         int r;
  94
  95         assert(w);
  96         assert(iovw);
  97         assert(iovw->count > 0);
  98
  99         if (journal_file_rotate_suggested(w->journal, 0)) {
 100                 log_info("%s: Journal header limits reached or header out-of-date, rotating",
 101                          w->journal->path);
 102                 r = do_rotate(&w->journal, compress, seal);
 103                 if (r < 0)
 104                         return r;
 105         }
 106
 107         r = journal_file_append_entry(w->journal, ts, iovw->iovec, iovw->count,
 108                                       &w->seqnum, NULL, NULL);
 109         if (r >= 0) {
 110                 if (w->server)
 111                         w->server->event_count += 1;
 112                 return 1;
 113         }
 114
 115         log_debug_errno(r, "%s: Write failed, rotating: %m", w->journal->path);
 116         r = do_rotate(&w->journal, compress, seal);
 117         if (r < 0)
 118                 return r;
 119         else
 120                 log_debug("%s: Successfully rotated journal", w->journal->path);
 121
 122         log_debug("Retrying write.");
 123         r = journal_file_append_entry(w->journal, ts, iovw->iovec, iovw->count,
 124                                       &w->seqnum, NULL, NULL);
 125         if (r < 0)
 126                 return r;
 127
 128         if (w->server)
 129                 w->server->event_count += 1;
 130         return 1;
 131 }