-
Notifications
You must be signed in to change notification settings - Fork 1.7k
/
logrotate.cpp
253 lines (205 loc) · 7.36 KB
/
logrotate.cpp
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
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
#include <unistd.h>
#include <new>
#include <functional>
#include <string>
#include <process/dispatch.hpp>
#include <process/io.hpp>
#include <process/process.hpp>
#include <stout/bytes.hpp>
#include <stout/error.hpp>
#include <stout/exit.hpp>
#include <stout/nothing.hpp>
#include <stout/path.hpp>
#include <stout/stringify.hpp>
#include <stout/try.hpp>
#include <stout/os/shell.hpp>
#include <stout/os/write.hpp>
#include "slave/container_loggers/logrotate.hpp"
using namespace process;
using namespace mesos::internal::logger::rotate;
class LogrotateLoggerProcess : public Process<LogrotateLoggerProcess>
{
public:
LogrotateLoggerProcess(const Flags& _flags)
: flags(_flags),
leading(None()),
bytesWritten(0)
{
// Prepare a buffer for reading from the `incoming` pipe.
length = sysconf(_SC_PAGE_SIZE);
buffer = new char[length];
}
virtual ~LogrotateLoggerProcess()
{
if (buffer != NULL) {
delete[] buffer;
buffer = NULL;
}
if (leading.isSome()) {
os::close(leading.get());
}
}
// Prepares and starts the loop which reads from stdin, writes to the
// leading log file, and manages total log size.
Future<Nothing> run()
{
// Populate the `logrotate` configuration file.
// See `Flags::logrotate_options` for the format.
//
// NOTE: We specify a size of `--max_size - length` because `logrotate`
// has slightly different size semantics. `logrotate` will rotate when the
// max size is *exceeded*. We rotate to keep files *under* the max size.
const std::string config =
"\"" + flags.log_filename.get() + "\" {\n" +
flags.logrotate_options.getOrElse("") + "\n" +
"size " + stringify(flags.max_size.bytes() - length) + "\n" +
"}";
Try<Nothing> result = os::write(
flags.log_filename.get() + CONF_SUFFIX, config);
if (result.isError()) {
return Failure("Failed to write configuration file: " + result.error());
}
// NOTE: This is a prerequisuite for `io::read`.
Try<Nothing> nonblock = os::nonblock(STDIN_FILENO);
if (nonblock.isError()) {
return Failure("Failed to set nonblocking pipe: " + nonblock.error());
}
// NOTE: This does not block.
loop();
return promise.future();
}
// Reads from stdin and writes to the leading log file.
void loop()
{
io::read(STDIN_FILENO, buffer, length)
.then([&](size_t readSize) -> Future<Nothing> {
// Check if EOF has been reached on the input stream.
// This indicates that the container (whose logs are being
// piped to this process) has exited.
if (readSize <= 0) {
promise.set(Nothing());
return Nothing();
}
// Do log rotation (if necessary) and write the bytes to the
// leading log file.
Try<Nothing> result = write(readSize);
if (result.isError()) {
promise.fail("Failed to write: " + result.error());
return Nothing();
}
// Use `dispatch` to limit the size of the call stack.
dispatch(self(), &LogrotateLoggerProcess::loop);
return Nothing();
});
}
// Writes the buffer from stdin to the leading log file.
// When the number of written bytes exceeds `--max_size`, the leading
// log file is rotated. When the number of log files exceed `--max_files`,
// the oldest log file is deleted.
Try<Nothing> write(size_t readSize)
{
// Rotate the log file if it will grow beyond the `--max_size`.
if (bytesWritten + readSize > flags.max_size.bytes()) {
rotate();
}
// If the leading log file is not open, open it.
// NOTE: We open the file in append-mode as `logrotate` may sometimes fail.
if (leading.isNone()) {
Try<int> open = os::open(
flags.log_filename.get(),
O_WRONLY | O_CREAT | O_APPEND | O_CLOEXEC,
S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH);
if (open.isError()) {
return Error(
"Failed to open '" + flags.log_filename.get() +
"': " + open.error());
}
leading = open.get();
}
// Write from stdin to `leading`.
// NOTE: We do not exit on error here since we are prioritizing
// clearing the STDIN pipe (which would otherwise potentially block
// the container on write) over log fidelity.
Try<Nothing> result =
os::write(leading.get(), std::string(buffer, readSize));
if (result.isError()) {
std::cerr << "Failed to write: " << result.error() << std::endl;
}
bytesWritten += readSize;
return Nothing();
}
// Calls `logrotate` on the leading log file and resets the `bytesWritten`.
void rotate()
{
if (leading.isSome()) {
os::close(leading.get());
leading = None();
}
// Call `logrotate` to move around the files.
// NOTE: If `logrotate` fails for whatever reason, we will ignore
// the error and continue logging. In case the leading log file
// is not renamed, we will continue appending to the existing
// leading log file.
os::shell(
flags.logrotate_path +
" --state \"" + flags.log_filename.get() + STATE_SUFFIX + "\" \"" +
flags.log_filename.get() + CONF_SUFFIX + "\"");
// Reset the number of bytes written.
bytesWritten = 0;
}
private:
Flags flags;
// For reading from stdin.
char* buffer;
size_t length;
// For writing and rotating the leading log file.
Option<int> leading;
size_t bytesWritten;
// Used to capture when log rotation has completed because the
// underlying process/input has terminated.
Promise<Nothing> promise;
};
int main(int argc, char** argv)
{
Flags flags;
// Load and validate flags from the environment and command line.
Try<flags::Warnings> load = flags.load(None(), &argc, &argv);
if (load.isError()) {
EXIT(EXIT_FAILURE) << flags.usage(load.error());
}
// Log any flag warnings.
foreach (const flags::Warning& warning, load->warnings) {
LOG(WARNING) << warning.message;
}
// Make sure this process is running in its own session.
// This ensures that, if the parent process (presumably the Mesos agent)
// terminates, this logger process will continue to run.
if (::setsid() == -1) {
EXIT(EXIT_FAILURE)
<< ErrnoError("Failed to put child in a new session").message;
}
// Asynchronously control the flow and size of logs.
LogrotateLoggerProcess process(flags);
spawn(&process);
// Wait for the logging process to finish.
Future<Nothing> status = dispatch(process, &LogrotateLoggerProcess::run);
status.await();
terminate(process);
wait(process);
return status.isReady() ? EXIT_SUCCESS : EXIT_FAILURE;
}