-
Notifications
You must be signed in to change notification settings - Fork 2.9k
/
UfsJournalLogWriter.java
477 lines (438 loc) · 19 KB
/
UfsJournalLogWriter.java
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
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
/*
* The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
* (the "License"). You may not use this work except in compliance with the License, which is
* available at www.apache.org/licenses/LICENSE-2.0
*
* This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
* either express or implied, as more fully set forth in the License.
*
* See the NOTICE file distributed with this work for information regarding copyright ownership.
*/
package alluxio.master.journal.ufs;
import alluxio.RuntimeConstants;
import alluxio.conf.PropertyKey;
import alluxio.conf.ServerConfiguration;
import alluxio.exception.ExceptionMessage;
import alluxio.exception.JournalClosedException;
import alluxio.exception.JournalClosedException.IOJournalClosedException;
import alluxio.master.journal.JournalEntryStreamReader;
import alluxio.master.journal.JournalWriter;
import alluxio.proto.journal.Journal.JournalEntry;
import alluxio.underfs.UnderFileSystem;
import alluxio.underfs.options.CreateOptions;
import alluxio.underfs.options.OpenOptions;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.io.ByteStreams;
import com.google.common.io.Closer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.net.URI;
import java.util.ArrayDeque;
import java.util.List;
import java.util.Queue;
import javax.annotation.concurrent.ThreadSafe;
/**
* Class for writing journal edit log entries from the primary master. It marks the current log
* complete (so that it is visible to the secondary masters) when the current log is large enough.
*
* When a new journal writer is created, it also marks the current log complete if there is one.
*
* A journal garbage collector thread is created when the writer is created, and is stopped when the
* writer is closed.
*/
@ThreadSafe
final class UfsJournalLogWriter implements JournalWriter {
private static final Logger LOG = LoggerFactory.getLogger(UfsJournalLogWriter.class);
private final UfsJournal mJournal;
private final UnderFileSystem mUfs;
/** The maximum size in bytes of a log file. */
private final long mMaxLogSize;
/** The next sequence number to use. */
private long mNextSequenceNumber;
/** When mRotateForNextWrite is set, mJournalOutputStream must be closed before the next write. */
private boolean mRotateLogForNextWrite;
/**
* The output stream to write the journal log entries.
* Initially this field is null.
* Also set this field to null when an {@link IOException} is caught.
*/
private JournalOutputStream mJournalOutputStream;
/** The garbage collector. */
private UfsJournalGarbageCollector mGarbageCollector;
/** Whether the journal log writer is closed. */
private boolean mClosed;
/**
* Set mNeedsRecovery to true when an IOException is thrown when trying to write journal entries.
* Clear this flag when {@link #maybeRecoverFromUfsFailures()} successfully recovers.
*/
private boolean mNeedsRecovery = false;
/**
* Journal entries that have been written successfully to the underlying
* {@link DataOutputStream}, but have not been flushed. Should a failure occur
* before flush, {@code UfsJournalLogWriter} is able to retry writing the
* journal entries.
*/
private Queue<JournalEntry> mEntriesToFlush;
/**
* Creates a new instance of {@link UfsJournalLogWriter}.
*
* @param journal the handle to the journal
* @param nextSequenceNumber the sequence number to begin writing at
*/
UfsJournalLogWriter(UfsJournal journal, long nextSequenceNumber) throws IOException {
mJournal = Preconditions.checkNotNull(journal, "journal");
mUfs = mJournal.getUfs();
mNextSequenceNumber = nextSequenceNumber;
mMaxLogSize = ServerConfiguration.getBytes(PropertyKey.MASTER_JOURNAL_LOG_SIZE_BYTES_MAX);
mRotateLogForNextWrite = true;
UfsJournalFile currentLog = UfsJournalSnapshot.getCurrentLog(mJournal);
if (currentLog != null) {
mJournalOutputStream = new JournalOutputStream(currentLog, ByteStreams.nullOutputStream());
}
mGarbageCollector = new UfsJournalGarbageCollector(mJournal);
mEntriesToFlush = new ArrayDeque<>();
}
public synchronized void write(JournalEntry entry) throws IOException, JournalClosedException {
try {
maybeRecoverFromUfsFailures();
maybeRotateLog();
} catch (IOJournalClosedException e) {
throw e.toJournalClosedException();
}
try {
JournalEntry entryToWrite =
entry.toBuilder().setSequenceNumber(mNextSequenceNumber).build();
entryToWrite.writeDelimitedTo(mJournalOutputStream);
LOG.debug("Adding journal entry (seq={}) to retryList with {} entries.",
entryToWrite.getSequenceNumber(), mEntriesToFlush.size());
mEntriesToFlush.add(entryToWrite);
mNextSequenceNumber++;
} catch (IOJournalClosedException e) {
throw e.toJournalClosedException();
} catch (IOException e) {
// Set mNeedsRecovery to true so that {@code maybeRecoverFromUfsFailures}
// can know a UFS failure has occurred.
mNeedsRecovery = true;
throw new IOException(ExceptionMessage.JOURNAL_WRITE_FAILURE
.getMessageWithUrl(RuntimeConstants.ALLUXIO_DEBUG_DOCS_URL,
mJournalOutputStream.currentLog(), e.getMessage()), e);
}
}
/**
* Core logic of UFS journal recovery from UFS failures.
*
* If Alluxio stores its journals in UFS, then Alluxio needs to handle UFS failures.
* When UFS is dead, there is nothing Alluxio can do because Alluxio relies on UFS to
* persist journal entries. Consequently any metadata operation will block because Alluxio
* cannot flush their journal entries.
* Once UFS comes back online, Alluxio needs to perform the following operations:
* 1. Find out the sequence number of the last persisted journal entry, say X. Then the first
* non-persisted entry has sequence number Y = X + 1.
* 2. Check whether there is any missing journal entry between Y (inclusive) and the oldest
* entry in mEntriesToFlush, say Z. If Z > Y, then it means journal entries in [Y, Z) are
* missing, and Alluxio cannot recover. Otherwise, for each journal entry in
* {@link #mEntriesToFlush}, if its sequence number is larger than or equal to Y, retry
* writing it to UFS by calling the {@code UfsJournalLogWriter#write} method.
*/
private void maybeRecoverFromUfsFailures() throws IOException, JournalClosedException {
if (!mNeedsRecovery) {
return;
}
long lastPersistSeq = recoverLastPersistedJournalEntry();
if (lastPersistSeq == -1) {
throw new RuntimeException("Cannot find any journal entry to recover from.");
}
createNewLogFile(lastPersistSeq + 1);
if (!mEntriesToFlush.isEmpty()) {
JournalEntry firstEntryToFlush = mEntriesToFlush.peek();
if (firstEntryToFlush.getSequenceNumber() > lastPersistSeq + 1) {
throw new RuntimeException(ExceptionMessage.JOURNAL_ENTRY_MISSING.getMessageWithUrl(
RuntimeConstants.ALLUXIO_DEBUG_DOCS_URL,
lastPersistSeq + 1, firstEntryToFlush.getSequenceNumber()));
}
long retryEndSeq = lastPersistSeq;
LOG.info("Retry writing unwritten journal entries from seq {}", lastPersistSeq + 1);
for (JournalEntry entry : mEntriesToFlush) {
if (entry.getSequenceNumber() > lastPersistSeq) {
try {
entry.toBuilder().build().writeDelimitedTo(mJournalOutputStream);
retryEndSeq = entry.getSequenceNumber();
} catch (IOJournalClosedException e) {
throw e.toJournalClosedException();
} catch (IOException e) {
throw new IOException(ExceptionMessage.JOURNAL_WRITE_FAILURE
.getMessageWithUrl(RuntimeConstants.ALLUXIO_DEBUG_DOCS_URL,
mJournalOutputStream.currentLog(), e.getMessage()), e);
}
}
}
LOG.info("Finished writing unwritten journal entries from {} to {}.",
lastPersistSeq + 1, retryEndSeq);
if (retryEndSeq != mNextSequenceNumber - 1) {
throw new RuntimeException("Failed to recover all entries to flush, expecting "
+ (mNextSequenceNumber - 1) + " but only found entry " + retryEndSeq);
}
}
mNeedsRecovery = false;
}
/**
* Examine the UFS to determine the most recent journal entry, and return its sequence number.
*
* 1. Locate the most recent incomplete journal file, i.e. journal file that starts with
* a valid sequence number S (hex), and ends with 0x7fffffffffffffff. The journal file
* name encodes this information, i.e. S-0x7fffffffffffffff.
* 2. Sequentially scan the incomplete journal file, and identify the last journal
* entry that has been persisted in UFS. Suppose it is X.
* 3. Rename the incomplete journal file to S-<X+1>. Future journal writes will write to
* a new file named <X+1>-0x7fffffffffffffff.
* 4. If the incomplete journal does not exist or no entry can be found in the incomplete
* journal, check the last complete journal file for the last persisted journal entry.
*
* @return sequence number of the last persisted journal entry, or -1 if no entry can be found
*/
private long recoverLastPersistedJournalEntry() throws IOException {
UfsJournalSnapshot snapshot = UfsJournalSnapshot.getSnapshot(mJournal);
long lastPersistSeq = -1;
UfsJournalFile currentLog = snapshot.getCurrentLog(mJournal);
if (currentLog != null) {
LOG.info("Recovering from previous UFS journal write failure."
+ " Scanning for the last persisted journal entry.");
try (JournalEntryStreamReader reader =
new JournalEntryStreamReader(mUfs.open(currentLog.getLocation().toString(),
OpenOptions.defaults().setRecoverFailedOpen(true)))) {
JournalEntry entry;
while ((entry = reader.readEntry()) != null) {
if (entry.getSequenceNumber() > lastPersistSeq) {
lastPersistSeq = entry.getSequenceNumber();
}
}
} catch (IOException e) {
throw e;
}
if (lastPersistSeq != -1) { // If the current log is an empty file, do not complete with SN: 0
completeLog(currentLog, lastPersistSeq + 1);
}
}
// Search for and scan the latest COMPLETE journal and find out the sequence number of the
// last persisted journal entry, in case no entry has been found in the INCOMPLETE journal.
if (lastPersistSeq < 0) {
// Re-evaluate snapshot because the incomplete journal will be destroyed if
// it does not contain any valid entry.
snapshot = UfsJournalSnapshot.getSnapshot(mJournal);
// journalFiles[journalFiles.size()-1] is the latest complete journal file.
List<UfsJournalFile> journalFiles = snapshot.getLogs();
if (!journalFiles.isEmpty()) {
for (int i = journalFiles.size() - 1; i >= 0; i--) {
UfsJournalFile journal = journalFiles.get(i);
if (!journal.isIncompleteLog()) { // Do not consider incomplete logs (handled above)
lastPersistSeq = journal.getEnd() - 1;
LOG.info("Found last persisted journal entry with seq {} in {}.",
lastPersistSeq, journal.getLocation().toString());
break;
}
}
}
}
return lastPersistSeq;
}
/**
* Closes the current journal output stream and creates a new one.
* The implementation must be idempotent so that it can work when retrying during failures.
*/
private void maybeRotateLog() throws IOException {
if (!mRotateLogForNextWrite) {
return;
}
if (mJournalOutputStream != null) {
mJournalOutputStream.close();
mJournalOutputStream = null;
}
createNewLogFile(mNextSequenceNumber);
mRotateLogForNextWrite = false;
}
private void createNewLogFile(long startSequenceNumber) throws IOException {
URI newLog = UfsJournalFile
.encodeLogFileLocation(mJournal, startSequenceNumber, UfsJournal.UNKNOWN_SEQUENCE_NUMBER);
UfsJournalFile currentLog = UfsJournalFile.createLogFile(newLog, startSequenceNumber,
UfsJournal.UNKNOWN_SEQUENCE_NUMBER);
OutputStream outputStream = mUfs.create(currentLog.getLocation().toString(),
CreateOptions.defaults(ServerConfiguration.global()).setEnsureAtomic(false)
.setCreateParent(true));
mJournalOutputStream = new JournalOutputStream(currentLog, outputStream);
LOG.info("Created current log file: {}", currentLog);
}
/**
* Completes the given log.
*
* If the log is empty, it will be deleted.
*
* This method must be safe to run by multiple masters at the same time. This could happen if a
* primary master loses leadership and takes a while to close its journal. By the time it
* completes the current log, the new primary might be trying to close it as well.
*
* @param currentLog the log to complete
* @param nextSequenceNumber the next sequence number for the log to complete
*/
private void completeLog(UfsJournalFile currentLog, long nextSequenceNumber) throws IOException {
String current = currentLog.getLocation().toString();
if (nextSequenceNumber <= currentLog.getStart()) {
LOG.info("No journal entry found in current journal file {}. Deleting it", current);
if (!mUfs.deleteFile(current)) {
LOG.warn("Failed to delete empty journal file {}", current);
}
return;
}
LOG.info("Completing log {} with next sequence number {}", current, nextSequenceNumber);
String completed = UfsJournalFile
.encodeLogFileLocation(mJournal, currentLog.getStart(), nextSequenceNumber).toString();
if (!mUfs.renameFile(current, completed)) {
// Completes could happen concurrently, check whether another master already did the rename.
if (!mUfs.exists(completed)) {
throw new IOException(
String.format("Failed to rename journal log from %s to %s", current, completed));
}
if (mUfs.exists(current)) {
// Rename is not atomic, so this could happen if we failed partway through a rename.
LOG.info("Deleting current log {}", current);
if (!mUfs.deleteFile(current)) {
LOG.warn("Failed to delete current log file {}", current);
}
}
}
}
public synchronized void flush() throws IOException, JournalClosedException {
maybeRecoverFromUfsFailures();
if (mJournalOutputStream == null || mJournalOutputStream.bytesWritten() == 0) {
// There is nothing to flush.
return;
}
try {
mJournalOutputStream.flush();
// Since flush has succeeded, it's safe to clear the mEntriesToFlush queue
// because they are considered "persisted" in UFS.
mEntriesToFlush.clear();
} catch (IOJournalClosedException e) {
throw e.toJournalClosedException();
} catch (IOException e) { // On next operation, attempt to recover from a UFS failure
mNeedsRecovery = true;
UfsJournalFile currentLog = mJournalOutputStream.currentLog();
mJournalOutputStream = null;
throw new IOException(ExceptionMessage.JOURNAL_FLUSH_FAILURE
.getMessageWithUrl(RuntimeConstants.ALLUXIO_DEBUG_DOCS_URL,
currentLog, e.getMessage()), e);
}
boolean overSize = mJournalOutputStream.bytesWritten() >= mMaxLogSize;
if (overSize || !mUfs.supportsFlush()) {
// (1) The log file is oversize, needs to be rotated. Or
// (2) Underfs is S3 or OSS, flush on S3OutputStream/OSSOutputStream will only flush to
// local temporary file, call close and complete the log to sync the journal entry to S3/OSS.
if (overSize) {
LOG.info("Rotating log file. size: {} maxSize: {}", mJournalOutputStream.bytesWritten(),
mMaxLogSize);
}
mRotateLogForNextWrite = true;
}
}
@Override
public synchronized void close() throws IOException {
Closer closer = Closer.create();
if (mJournalOutputStream != null) {
closer.register(mJournalOutputStream);
}
closer.register(mGarbageCollector);
closer.close();
mClosed = true;
}
/**
* A simple wrapper that wraps a output stream to the current log file. When this stream is
* closed, the log file will be completed.
*
* Many of the methods in this class might throw {@link IOJournalClosedException} if the journal
* writer is closed when they are called. The exception needs to extend IOException because the
* OutputStream API only throws IOException. Callers of these methods should re-throw the
* {@link IOJournalClosedException} as a regular {@link JournalClosedException} so that it will be
* properly handled by callers.
*/
private class JournalOutputStream extends OutputStream {
// Not intended for use outside this inner class.
private final DataOutputStream mOutputStream;
private final UfsJournalFile mCurrentLog;
JournalOutputStream(UfsJournalFile currentLog, OutputStream stream) throws IOException {
mOutputStream = wrapDataOutputStream(stream);
mCurrentLog = currentLog;
}
/**
* @return the number of bytes written to this stream
*/
long bytesWritten() {
if (mOutputStream == null) {
return 0;
}
return mOutputStream.size();
}
/**
* @return the log file being written to by this stream
*/
UfsJournalFile currentLog() {
return mCurrentLog;
}
@Override
public void write(int b) throws IOException {
checkJournalWriterOpen();
mOutputStream.write(b);
}
@Override
public void write(byte[] b) throws IOException {
checkJournalWriterOpen();
mOutputStream.write(b);
}
@Override
public void write(byte[] b, int off, int len) throws IOException {
checkJournalWriterOpen();
mOutputStream.write(b, off, len);
}
@Override
public void flush() throws IOException {
checkJournalWriterOpen();
mOutputStream.flush();
}
/**
* Closes the stream by committing the log. The implementation must be idempotent as this
* close can fail and be retried.
*/
@Override
public void close() throws IOException {
checkJournalWriterOpen();
mOutputStream.close();
LOG.info("Marking {} as complete with log entries within [{}, {}).",
mCurrentLog.getLocation(), mCurrentLog.getStart(), mNextSequenceNumber);
completeLog(mCurrentLog, mNextSequenceNumber);
}
private void checkJournalWriterOpen() throws IOJournalClosedException {
if (mClosed) {
throw new JournalClosedException("Journal writer is closed").toIOException();
}
}
}
private static DataOutputStream wrapDataOutputStream(OutputStream stream) {
if (stream instanceof DataOutputStream) {
return (DataOutputStream) stream;
} else {
return new DataOutputStream(stream);
}
}
/**
* @return the next sequence number to write
*/
public synchronized long getNextSequenceNumber() {
return mNextSequenceNumber;
}
@VisibleForTesting
synchronized JournalOutputStream getJournalOutputStream() {
return mJournalOutputStream;
}
}