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 "alloc-util.h"
  23 #include "journal-remote.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         iovw->iovec = mfree(iovw->iovec);
  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_errno(r, "Failed to rotate %s: %m", (*f)->path);
  63                 else
  64                         log_error_errno(r, "Failed to create rotated journal: %m");
  65         }
  66
  67         return r;
  68 }
  69
  70 Writer* writer_new(RemoteServer *server) {
  71         Writer *w;
  72
  73         w = new0(Writer, 1);
  74         if (!w)
  75                 return NULL;
  76
  77         memset(&w->metrics, 0xFF, sizeof(w->metrics));
  78
  79         w->mmap = mmap_cache_new();
  80         if (!w->mmap) {
  81                 free(w);
  82                 return NULL;
  83         }
  84
  85         w->n_ref = 1;
  86         w->server = server;
  87
  88         return w;
  89 }
  90
  91 Writer* writer_free(Writer *w) {
  92         if (!w)
  93                 return NULL;
  94
  95         if (w->journal) {
  96                 log_debug("Closing journal file %s.", w->journal->path);
  97                 journal_file_close(w->journal);
  98         }
  99
 100         if (w->server && w->hashmap_key)
 101                 hashmap_remove(w->server->writers, w->hashmap_key);
 102
 103         free(w->hashmap_key);
 104
 105         if (w->mmap)
 106                 mmap_cache_unref(w->mmap);
 107
 108         free(w);
 109
 110         return NULL;
 111 }
 112
 113 Writer* writer_unref(Writer *w) {
 114         if (w && (-- w->n_ref <= 0))
 115                 writer_free(w);
 116
 117         return NULL;
 118 }
 119
 120 Writer* writer_ref(Writer *w) {
 121         if (w)
 122                 assert_se(++ w->n_ref >= 2);
 123
 124         return w;
 125 }
 126
 127 int writer_write(Writer *w,
 128                  struct iovec_wrapper *iovw,
 129                  dual_timestamp *ts,
 130                  bool compress,
 131                  bool seal) {
 132         int r;
 133
 134         assert(w);
 135         assert(iovw);
 136         assert(iovw->count > 0);
 137
 138         if (journal_file_rotate_suggested(w->journal, 0)) {
 139                 log_info("%s: Journal header limits reached or header out-of-date, rotating",
 140                          w->journal->path);
 141                 r = do_rotate(&w->journal, compress, seal);
 142                 if (r < 0)
 143                         return r;
 144         }
 145
 146         r = journal_file_append_entry(w->journal, ts, iovw->iovec, iovw->count,
 147                                       &w->seqnum, NULL, NULL);
 148         if (r >= 0) {
 149                 if (w->server)
 150                         w->server->event_count += 1;
 151                 return 1;
 152         }
 153
 154         log_debug_errno(r, "%s: Write failed, rotating: %m", w->journal->path);
 155         r = do_rotate(&w->journal, compress, seal);
 156         if (r < 0)
 157                 return r;
 158         else
 159                 log_debug("%s: Successfully rotated journal", w->journal->path);
 160
 161         log_debug("Retrying write.");
 162         r = journal_file_append_entry(w->journal, ts, iovw->iovec, iovw->count,
 163                                       &w->seqnum, NULL, NULL);
 164         if (r < 0)
 165                 return r;
 166
 167         if (w->server)
 168                 w->server->event_count += 1;
 169         return 1;
 170 }