Skip to content

[bk-gc] Prevent OOM: Restrict number of entry-loggers extraction #1938

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
rdhabalia wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[bk-gc] Prevent OOM: Restrict number of entry-loggers extraction #1938

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
133 changes: 83 additions & 50 deletions bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/GarbageCollectorThread.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,13 @@ public class GarbageCollectorThread extends SafeRunnable {

// This is how often we want to run the Garbage Collector Thread (in milliseconds).
final long gcWaitTime;
// Max number of entry-logger files should be extracted concurrently while
// performing GC
final int maxEntryLoggersPerScan;
// flag to iterate over chunk of of entry-logger files
boolean moreEntryLoggers = true;
long lastIterationLogId = 0;
private final boolean verifyMetadataOnGc;

// Compaction parameters
boolean enableMinorCompaction = false;
Expand Down Expand Up @@ -146,6 +153,9 @@ public GarbageCollectorThread(ServerConfiguration conf,
this.entryLogger = ledgerStorage.getEntryLogger();
this.ledgerStorage = ledgerStorage;
this.gcWaitTime = conf.getGcWaitTime();

this.maxEntryLoggersPerScan = conf.getMaxEntryLoggersScanOnGc();
this.verifyMetadataOnGc = conf.getVerifyMetadataOnGC();

this.numActiveEntryLogs = 0;
this.totalEntryLogSize = 0L;
Expand Down Expand Up @@ -322,55 +332,64 @@ public void runWithFlags(boolean force, boolean suspendMajor, boolean suspendMin
}
// Recover and clean up previous state if using transactional compaction
compactor.cleanUpAndRecover();

long startTime = System.currentTimeMillis();
boolean isIteration = false;
do {
// Extract all of the ledger ID's that comprise all of the entry
// logs
// (except for the current new one which is still being written to).
entryLogMetaMap = extractMetaFromEntryLogs(entryLogMetaMap, isIteration);

// gc inactive/deleted ledgers
doGcLedgers();

@reddycharan reddycharan Feb 16, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this is very expensive operation. It iterates over all the ledger ranges in the metadata server/all the ledgers in the bookie, this shouldn't be called multiple times

@rdhabalia rdhabalia Feb 16, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes, that's correct. therefore, have provided option maxEntryLoggersPerScan=2000. In normal case, bookie will have < 500 entryLog files so, it will not require multiple iteration but it will target issue where bookie is not able to recover with large number of entrylog files. and therefore, @merlimat suggested to persist entryLogmetadataMap into rocksDB so, I have created PR: #1949


// gc entry logs
doGcEntryLogs();

@reddycharan reddycharan Feb 16, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this method iterates over all the entries in entryLogMetaMap, again calling this multiple times is not sensible.


if (suspendMajor) {
LOG.info("Disk almost full, suspend major compaction to slow down filling disk.");
}
if (suspendMinor) {
LOG.info("Disk full, suspend minor compaction to slow down filling disk.");
}

// Extract all of the ledger ID's that comprise all of the entry logs
// (except for the current new one which is still being written to).
entryLogMetaMap = extractMetaFromEntryLogs(entryLogMetaMap);

// gc inactive/deleted ledgers
doGcLedgers();

// gc entry logs
doGcEntryLogs();

if (suspendMajor) {
LOG.info("Disk almost full, suspend major compaction to slow down filling disk.");
}
if (suspendMinor) {
LOG.info("Disk full, suspend minor compaction to slow down filling disk.");
}

long curTime = System.currentTimeMillis();
if (enableMajorCompaction && (!suspendMajor)
&& (force || curTime - lastMajorCompactionTime > majorCompactionInterval)) {
// enter major compaction
LOG.info("Enter major compaction, suspendMajor {}", suspendMajor);
majorCompacting.set(true);
doCompactEntryLogs(majorCompactionThreshold);
lastMajorCompactionTime = System.currentTimeMillis();
// and also move minor compaction time
lastMinorCompactionTime = lastMajorCompactionTime;
gcStats.getMajorCompactionCounter().inc();
majorCompacting.set(false);
} else if (enableMinorCompaction && (!suspendMinor)
&& (force || curTime - lastMinorCompactionTime > minorCompactionInterval)) {
// enter minor compaction
LOG.info("Enter minor compaction, suspendMinor {}", suspendMinor);
minorCompacting.set(true);
doCompactEntryLogs(minorCompactionThreshold);
lastMinorCompactionTime = System.currentTimeMillis();
gcStats.getMinorCompactionCounter().inc();
minorCompacting.set(false);
}
long curTime = System.currentTimeMillis();
if (enableMajorCompaction && (!suspendMajor)
&& (force || curTime - lastMajorCompactionTime > majorCompactionInterval)) {
// enter major compaction
LOG.info("Enter major compaction, suspendMajor {}", suspendMajor);
majorCompacting.set(true);
doCompactEntryLogs(majorCompactionThreshold);
lastMajorCompactionTime = System.currentTimeMillis();
// and also move minor compaction time
lastMinorCompactionTime = lastMajorCompactionTime;
gcStats.getMajorCompactionCounter().inc();
majorCompacting.set(false);
} else if (enableMinorCompaction && (!suspendMinor)
&& (force || curTime - lastMinorCompactionTime > minorCompactionInterval)) {
// enter minor compaction
LOG.info("Enter minor compaction, suspendMinor {}", suspendMinor);
minorCompacting.set(true);
doCompactEntryLogs(minorCompactionThreshold);
lastMinorCompactionTime = System.currentTimeMillis();
gcStats.getMinorCompactionCounter().inc();
minorCompacting.set(false);
}

if (force) {
if (forceGarbageCollection.compareAndSet(true, false)) {
LOG.info("{} Set forceGarbageCollection to false after force GC to make it forceGC-able again.", Thread
.currentThread().getName());
if (force) {
if (forceGarbageCollection.compareAndSet(true, false)) {
LOG.info("{} Set forceGarbageCollection to false after force GC to make it forceGC-able again.",
Thread.currentThread().getName());
}
}
}
gcStats.getGcThreadRuntime().registerSuccessfulEvent(
MathUtils.nowInNano() - threadStart, TimeUnit.NANOSECONDS);
gcStats.getGcThreadRuntime().registerSuccessfulEvent(MathUtils.nowInNano() - threadStart,
TimeUnit.NANOSECONDS);
isIteration = true;
} while (moreEntryLoggers);

long endTime = System.currentTimeMillis();
LOG.info("Garbage collector completed in {}", TimeUnit.MILLISECONDS.toSeconds(endTime - startTime));
}

/**
Expand Down Expand Up @@ -526,21 +545,35 @@ protected void compactEntryLog(EntryLogMetadata entryLogMeta) {
* Existing EntryLogs to Meta
* @throws IOException
*/
protected Map<Long, EntryLogMetadata> extractMetaFromEntryLogs(Map<Long, EntryLogMetadata> entryLogMetaMap) {
protected Map<Long, EntryLogMetadata> extractMetaFromEntryLogs(Map<Long, EntryLogMetadata> entryLogMetaMap, boolean isIteration) {
moreEntryLoggers = false;
// Extract it for every entry log except for the current one.
// Entry Log ID's are just a long value that starts at 0 and increments
// by 1 when the log fills up and we roll to a new one.
long curLogId = entryLogger.getLeastUnflushedLogId();
boolean hasExceptionWhenScan = false;
for (long entryLogId = scannedLogId; entryLogId < curLogId; entryLogId++) {
// Comb the current entry log file if it has not already been extracted.
int entryLogFileCount = 0;
long entryLogId = isIteration ? lastIterationLogId : scannedLogId;
for (; entryLogId < curLogId; entryLogId++, entryLogFileCount++) {
if (entryLogFileCount > maxEntryLoggersPerScan && verifyMetadataOnGc) {
lastIterationLogId = entryLogId;
moreEntryLoggers = true;
LOG.debug("extraction max-entry-logger {}, next iteration starts from {}", entryLogFileCount,
maxEntryLoggersPerScan, entryLogId);
break;
}
// Comb the current entry log file if it has not already been
// extracted.
if (entryLogMetaMap.containsKey(entryLogId)) {
entryLogFileCount--;
continue;
}

// check whether log file exists or not
// if it doesn't exist, this log file might have been garbage collected.
// if it doesn't exist, this log file might have been garbage
// collected.
if (!entryLogger.logExists(entryLogId)) {
entryLogFileCount--;
continue;
}

Expand Down
20 changes: 20 additions & 0 deletions bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ServerConfiguration.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -104,6 +104,7 @@ public class ServerConfiguration extends AbstractConfiguration<ServerConfigurati
protected static final String GC_OVERREPLICATED_LEDGER_WAIT_TIME = "gcOverreplicatedLedgerWaitTime";
protected static final String USE_TRANSACTIONAL_COMPACTION = "useTransactionalCompaction";
protected static final String VERIFY_METADATA_ON_GC = "verifyMetadataOnGC";
protected static final String MAX_ENTRY_LOGGERS_SCAN_ON_GC = "maxEntryLoggersScanOnGc";
// Scrub Parameters
protected static final String LOCAL_SCRUB_PERIOD = "localScrubInterval";
protected static final String LOCAL_SCRUB_RATE_LIMIT = "localScrubRateLimit";
Expand Down Expand Up @@ -431,6 +432,25 @@ public ServerConfiguration setVerifyMetadataOnGc(boolean verifyMetadataOnGC) {
return this;
}

/**
* Get Max entry-logger files that should be loaded while performing gc.
*
*/
public int getMaxEntryLoggersScanOnGc() {
return this.getInt(MAX_ENTRY_LOGGERS_SCAN_ON_GC, 1000);
}

/**
* Set Max entry-logger files that should be loaded while performing gc.
*
* @return use transactional compaction
*/
public ServerConfiguration setMaxEntryLoggersScanOnGc(int maxEntryLoggerForScan) {
this.setProperty(MAX_ENTRY_LOGGERS_SCAN_ON_GC, maxEntryLoggerForScan);
return this;
}


/**
* Get whether local scrub is enabled.
*
Expand Down
4 changes: 3 additions & 1 deletion bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CompactionTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -138,6 +138,8 @@ public void setUp() throws Exception {
baseConf.setLedgerStorageClass(InterleavedLedgerStorage.class.getName());
baseConf.setIsThrottleByBytes(this.isThrottleByBytes);
baseConf.setIsForceGCAllowWhenNoSpace(false);
baseConf.setVerifyMetadataOnGc(true);
baseConf.setMaxEntryLoggersScanOnGc(1);

super.setUp();
}
Expand Down Expand Up @@ -300,7 +302,7 @@ public void testMinorCompaction() throws Exception {
baseConf.setGcWaitTime(60000);
baseConf.setMinorCompactionInterval(120000);
baseConf.setMajorCompactionInterval(240000);

// restart bookies
restartBookies(baseConf);

Expand Down