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