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