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 int iovw_put(struct iovec_wrapper *iovw, void* data, size_t len) {
  24         if (!GREEDY_REALLOC(iovw->iovec, iovw->size_bytes, iovw->count + 1))
  25                 return log_oom();
  26
  27         iovw->iovec[iovw->count++] = (struct iovec) {data, len};
  28         return 0;
  29 }
  30
  31 void iovw_free_contents(struct iovec_wrapper *iovw) {
  32         iovw->iovec = mfree(iovw->iovec);
  33         iovw->size_bytes = iovw->count = 0;
  34 }
  35
  36 size_t iovw_size(struct iovec_wrapper *iovw) {
  37         size_t n = 0, i;
  38
  39         for (i = 0; i < iovw->count; i++)
  40                 n += iovw->iovec[i].iov_len;
  41
  42         return n;
  43 }
  44
  45 void iovw_rebase(struct iovec_wrapper *iovw, char *old, char *new) {
  46         size_t i;
  47
  48         for (i = 0; i < iovw->count; i++)
  49                 iovw->iovec[i].iov_base = (char*) iovw->iovec[i].iov_base - old + new;
  50 }
  51
  52 /**********************************************************************
  53  **********************************************************************
  54  **********************************************************************/
  55
  56 static int do_rotate(JournalFile **f, bool compress, bool seal) {
  57         int r = journal_file_rotate(f, compress, seal, NULL);
  58         if (r < 0) {
  59                 if (*f)
  60                         log_error_errno(r, "Failed to rotate %s: %m", (*f)->path);
  61                 else
  62                         log_error_errno(r, "Failed to create rotated journal: %m");
  63         }
  64
  65         return r;
  66 }
  67
  68 Writer* writer_new(RemoteServer *server) {
  69         Writer *w;
  70
  71         w = new0(Writer, 1);
  72         if (!w)
  73                 return NULL;
  74
  75         memset(&w->metrics, 0xFF, sizeof(w->metrics));
  76
  77         w->mmap = mmap_cache_new();
  78         if (!w->mmap)
  79                 return mfree(w);
  80
  81         w->n_ref = 1;
  82         w->server = server;
  83
  84         return w;
  85 }
  86
  87 Writer* writer_free(Writer *w) {
  88         if (!w)
  89                 return NULL;
  90
  91         if (w->journal) {
  92                 log_debug("Closing journal file %s.", w->journal->path);
  93                 journal_file_close(w->journal);
  94         }
  95
  96         if (w->server && w->hashmap_key)
  97                 hashmap_remove(w->server->writers, w->hashmap_key);
  98
  99         free(w->hashmap_key);
 100
 101         if (w->mmap)
 102                 mmap_cache_unref(w->mmap);
 103
 104         return mfree(w);
 105 }
 106
 107 Writer* writer_unref(Writer *w) {
 108         if (w && (-- w->n_ref <= 0))
 109                 writer_free(w);
 110
 111         return NULL;
 112 }
 113
 114 Writer* writer_ref(Writer *w) {
 115         if (w)
 116                 assert_se(++ w->n_ref >= 2);
 117
 118         return w;
 119 }
 120
 121 int writer_write(Writer *w,
 122                  struct iovec_wrapper *iovw,
 123                  dual_timestamp *ts,
 124                  bool compress,
 125                  bool seal) {
 126         int r;
 127
 128         assert(w);
 129         assert(iovw);
 130         assert(iovw->count > 0);
 131
 132         if (journal_file_rotate_suggested(w->journal, 0)) {
 133                 log_info("%s: Journal header limits reached or header out-of-date, rotating",
 134                          w->journal->path);
 135                 r = do_rotate(&w->journal, compress, seal);
 136                 if (r < 0)
 137                         return r;
 138         }
 139
 140         r = journal_file_append_entry(w->journal, ts, iovw->iovec, iovw->count,
 141                                       &w->seqnum, NULL, NULL);
 142         if (r >= 0) {
 143                 if (w->server)
 144                         w->server->event_count += 1;
 145                 return 1;
 146         }
 147
 148         log_debug_errno(r, "%s: Write failed, rotating: %m", w->journal->path);
 149         r = do_rotate(&w->journal, compress, seal);
 150         if (r < 0)
 151                 return r;
 152         else
 153                 log_debug("%s: Successfully rotated journal", w->journal->path);
 154
 155         log_debug("Retrying write.");
 156         r = journal_file_append_entry(w->journal, ts, iovw->iovec, iovw->count,
 157                                       &w->seqnum, NULL, NULL);
 158         if (r < 0)
 159                 return r;
 160
 161         if (w->server)
 162                 w->server->event_count += 1;
 163         return 1;
 164 }