1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
|
/* SPDX-License-Identifier: LGPL-2.1+ */
#include "alloc-util.h"
#include "journal-remote.h"
static int do_rotate(JournalFile **f, bool compress, bool seal) {
int r = journal_file_rotate(f, compress, (uint64_t) -1, seal, NULL);
if (r < 0) {
if (*f)
log_error_errno(r, "Failed to rotate %s: %m", (*f)->path);
else
log_error_errno(r, "Failed to create rotated journal: %m");
}
return r;
}
Writer* writer_new(RemoteServer *server) {
Writer *w;
w = new0(Writer, 1);
if (!w)
return NULL;
memset(&w->metrics, 0xFF, sizeof(w->metrics));
w->mmap = mmap_cache_new();
if (!w->mmap)
return mfree(w);
w->n_ref = 1;
w->server = server;
return w;
}
static Writer* writer_free(Writer *w) {
if (!w)
return NULL;
if (w->journal) {
log_debug("Closing journal file %s.", w->journal->path);
journal_file_close(w->journal);
}
if (w->server && w->hashmap_key)
hashmap_remove(w->server->writers, w->hashmap_key);
free(w->hashmap_key);
if (w->mmap)
mmap_cache_unref(w->mmap);
return mfree(w);
}
DEFINE_TRIVIAL_REF_UNREF_FUNC(Writer, writer, writer_free);
int writer_write(Writer *w,
struct iovec_wrapper *iovw,
dual_timestamp *ts,
bool compress,
bool seal) {
int r;
assert(w);
assert(iovw);
assert(iovw->count > 0);
if (journal_file_rotate_suggested(w->journal, 0)) {
log_info("%s: Journal header limits reached or header out-of-date, rotating",
w->journal->path);
r = do_rotate(&w->journal, compress, seal);
if (r < 0)
return r;
}
r = journal_file_append_entry(w->journal, ts, NULL,
iovw->iovec, iovw->count,
&w->seqnum, NULL, NULL);
if (r >= 0) {
if (w->server)
w->server->event_count += 1;
return 0;
} else if (r == -EBADMSG)
return r;
log_debug_errno(r, "%s: Write failed, rotating: %m", w->journal->path);
r = do_rotate(&w->journal, compress, seal);
if (r < 0)
return r;
else
log_debug("%s: Successfully rotated journal", w->journal->path);
log_debug("Retrying write.");
r = journal_file_append_entry(w->journal, ts, NULL,
iovw->iovec, iovw->count,
&w->seqnum, NULL, NULL);
if (r < 0)
return r;
if (w->server)
w->server->event_count += 1;
return 0;
}
|
