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

   1 /*-*- Mode: C; c-basic-offset: 8; indent-tabs-mode: nil -*-*/
   2
   3 /***
   4   This file is part of systemd.
   5
   6   Copyright 2012 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 "journal-remote.h"
  23 #include "journal-remote-write.h"
  24
  25 int iovw_put(struct iovec_wrapper *iovw, void* data, size_t len) {
  26         if (!GREEDY_REALLOC(iovw->iovec, iovw->size_bytes, iovw->count + 1))
  27                 return log_oom();
  28
  29         iovw->iovec[iovw->count++] = (struct iovec) {data, len};
  30         return 0;
  31 }
  32
  33 void iovw_free_contents(struct iovec_wrapper *iovw) {
  34         for (size_t j = 0; j < iovw->count; j++)
  35                 free(iovw->iovec[j].iov_base);
  36         free(iovw->iovec);
  37         iovw->iovec = NULL;
  38         iovw->size_bytes = iovw->count = 0;
  39 }
  40
  41 size_t iovw_size(struct iovec_wrapper *iovw) {
  42         size_t n = 0, i;
  43
  44         for(i = 0; i < iovw->count; i++)
  45                 n += iovw->iovec[i].iov_len;
  46
  47         return n;
  48 }
  49
  50 /**********************************************************************
  51  **********************************************************************
  52  **********************************************************************/
  53
  54 static int do_rotate(JournalFile **f, bool compress, bool seal) {
  55         int r = journal_file_rotate(f, compress, seal);
  56         if (r < 0) {
  57                 if (*f)
  58                         log_error("Failed to rotate %s: %s", (*f)->path,
  59                                   strerror(-r));
  60                 else
  61                         log_error("Failed to create rotated journal: %s",
  62                                   strerror(-r));
  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                 free(w);
  80                 return NULL;
  81         }
  82
  83         w->n_ref = 1;
  84         w->server = server;
  85
  86         return w;
  87 }
  88
  89 Writer* writer_free(Writer *w) {
  90         if (!w)
  91                 return NULL;
  92
  93         if (w->journal) {
  94                 log_debug("Closing journal file %s.", w->journal->path);
  95                 journal_file_close(w->journal);
  96         }
  97
  98         if (w->server) {
  99                 w->server->event_count += w->seqnum;
 100                 if (w->hashmap_key)
 101                         hashmap_remove(w->server->writers, w->hashmap_key);
 102         }
 103
 104         free(w->hashmap_key);
 105
 106         if (w->mmap)
 107                 mmap_cache_unref(w->mmap);
 108
 109         free(w);
 110
 111         return NULL;
 112 }
 113
 114 Writer* writer_unref(Writer *w) {
 115         if (w && (-- w->n_ref <= 0))
 116                 writer_free(w);
 117
 118         return NULL;
 119 }
 120
 121 Writer* writer_ref(Writer *w) {
 122         if (w)
 123                 assert_se(++ w->n_ref >= 2);
 124
 125         return w;
 126 }
 127
 128
 129 int writer_write(Writer *s,
 130                  struct iovec_wrapper *iovw,
 131                  dual_timestamp *ts,
 132                  bool compress,
 133                  bool seal) {
 134         int r;
 135
 136         assert(s);
 137         assert(iovw);
 138         assert(iovw->count > 0);
 139
 140         if (journal_file_rotate_suggested(s->journal, 0)) {
 141                 log_info("%s: Journal header limits reached or header out-of-date, rotating",
 142                          s->journal->path);
 143                 r = do_rotate(&s->journal, compress, seal);
 144                 if (r < 0)
 145                         return r;
 146         }
 147
 148         r = journal_file_append_entry(s->journal, ts, iovw->iovec, iovw->count,
 149                                       &s->seqnum, NULL, NULL);
 150         if (r >= 0)
 151                 return 1;
 152
 153         log_debug("%s: Write failed, rotating: %s", s->journal->path, strerror(-r));
 154         r = do_rotate(&s->journal, compress, seal);
 155         if (r < 0)
 156                 return r;
 157         else
 158                 log_info("%s: Successfully rotated journal", s->journal->path);
 159
 160         log_debug("Retrying write.");
 161         r = journal_file_append_entry(s->journal, ts, iovw->iovec, iovw->count,
 162                                       &s->seqnum, NULL, NULL);
 163         return r < 0 ? r : 1;
 164 }