-
Notifications
You must be signed in to change notification settings - Fork 123
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Modify and split CompactorMain into logical classes (#3509)
* Modify and split CompactorMain into logical classes Split CorfuStoreCompactorMain into CorfuCompactorCheckpointer and CorfuCompactorControls -The CorfuCompactorControls class holds all the controls related paramters and performs required operation accordingly -The CorfuCompactorCheckpointer starts checkpointing tables if there is an ongoing compaction cycle * Fix codacy issues and modify compactor_runner as required * Simplify the compactor class names
- Loading branch information
1 parent
a571d68
commit d86ad96
Showing
8 changed files
with
335 additions
and
252 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
85 changes: 85 additions & 0 deletions
85
corfudb-tools/src/main/java/org/corfudb/compactor/CompactorCheckpointer.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,85 @@ | ||
| package org.corfudb.compactor; | ||
|
|
||
| import lombok.extern.slf4j.Slf4j; | ||
|
|
||
| import org.corfudb.runtime.CheckpointerBuilder; | ||
| import org.corfudb.runtime.CorfuRuntime; | ||
| import org.corfudb.runtime.DistributedCheckpointer; | ||
| import org.corfudb.runtime.DistributedCheckpointerHelper; | ||
| import org.corfudb.runtime.ServerTriggeredCheckpointer; | ||
| import org.corfudb.runtime.collections.CorfuStore; | ||
|
|
||
| import java.util.Optional; | ||
| import java.util.concurrent.TimeUnit; | ||
|
|
||
| /** | ||
| * Invokes the startCheckpointing() method of DistributedCompactor to checkpoint tables that weren't | ||
| * checkpointed by other clients | ||
| */ | ||
| @Slf4j | ||
| public class CompactorCheckpointer { | ||
| private final DistributedCheckpointerHelper distributedCheckpointerHelper; | ||
| private final DistributedCheckpointer distributedCheckpointer; | ||
| public static final int RETRY_CHECKPOINTING = 5; | ||
| private static final int RETRY_CHECKPOINTING_SLEEP_SECOND = 10; | ||
|
|
||
| public CompactorCheckpointer(String[] args) throws Exception { | ||
| CompactorBaseConfig config = new CompactorBaseConfig(args, "", ""); | ||
|
|
||
| Thread.currentThread().setName("Cmpt-chkpter-" + config.getNodeLocator().getPort()); | ||
| CorfuRuntime cpRuntime = (CorfuRuntime.fromParameters( | ||
| config.getParams())).parseConfigurationString(config.getNodeLocator().toEndpointUrl()).connect(); | ||
| CorfuRuntime corfuRuntime = (CorfuRuntime.fromParameters( | ||
| config.getParams())).parseConfigurationString(config.getNodeLocator().toEndpointUrl()).connect(); | ||
| CorfuStore corfuStore = new CorfuStore(corfuRuntime); | ||
|
|
||
| this.distributedCheckpointerHelper = new DistributedCheckpointerHelper(corfuStore); | ||
| this.distributedCheckpointer = new ServerTriggeredCheckpointer(CheckpointerBuilder.builder() | ||
| .corfuRuntime(corfuRuntime) | ||
| .cpRuntime(Optional.of(cpRuntime)) | ||
| .persistedCacheRoot(config.getPersistedCacheRoot()) | ||
| .isClient(false) | ||
| .build(), corfuStore, distributedCheckpointerHelper.getCompactorMetadataTables()); | ||
| } | ||
|
|
||
| public CompactorCheckpointer(DistributedCheckpointerHelper distributedCheckpointerHelper, | ||
| DistributedCheckpointer distributedCheckpointer) { | ||
| this.distributedCheckpointerHelper = distributedCheckpointerHelper; | ||
| this.distributedCheckpointer = distributedCheckpointer; | ||
| } | ||
|
|
||
| /** | ||
| * Entry point to invoke checkpointing by the client jvm | ||
| * | ||
| * @param args command line argument strings | ||
| */ | ||
| public static void main(String[] args) { | ||
| try { | ||
| CompactorCheckpointer corfuCompactorMain = new CompactorCheckpointer(args); | ||
| corfuCompactorMain.startCheckpointing(); | ||
| } catch (Exception e) { | ||
| log.error("CorfuStoreCompactorMain crashed with error: {}, Exception: ", | ||
| CompactorBaseConfig.CORFU_LOG_CHECKPOINT_ERROR, e); | ||
| } | ||
| log.info("Exiting CorfuStoreCompactor"); | ||
| } | ||
|
|
||
| protected void startCheckpointing() { | ||
| try { | ||
| for (int i = 0; i < RETRY_CHECKPOINTING; i++) { | ||
| if (distributedCheckpointerHelper.hasCompactionStarted()) { | ||
| distributedCheckpointer.checkpointTables(); | ||
| break; | ||
| } | ||
| log.info("Compaction cycle hasn't started yet..."); | ||
| TimeUnit.SECONDS.sleep(RETRY_CHECKPOINTING_SLEEP_SECOND); | ||
| } | ||
| } catch (InterruptedException ie) { | ||
| log.error("Sleep interrupted with exception: ", ie); | ||
| } catch (Exception e) { | ||
| log.error("Exception during checkpointing: {}, StackTrace: {}", e.getMessage(), e.getStackTrace()); | ||
| } finally { | ||
| distributedCheckpointer.shutdown(); | ||
| } | ||
| } | ||
| } |
85 changes: 85 additions & 0 deletions
85
corfudb-tools/src/main/java/org/corfudb/compactor/CompactorController.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,85 @@ | ||
| package org.corfudb.compactor; | ||
|
|
||
| import lombok.extern.slf4j.Slf4j; | ||
|
|
||
| import org.corfudb.runtime.CorfuRuntime; | ||
| import org.corfudb.runtime.DistributedCheckpointerHelper; | ||
| import org.corfudb.runtime.collections.CorfuStore; | ||
|
|
||
| /** | ||
| * Helps with inserting the required compactor controls key into the CompactionControlsTable | ||
| * These keys are used by the manager to make decisions regarding triggering the next cycle | ||
| */ | ||
| @Slf4j | ||
| public class CompactorController { | ||
| private final CorfuRuntime corfuRuntime; | ||
| private final CorfuStore corfuStore; | ||
| private final CompactorControllerConfig config; | ||
| private final DistributedCheckpointerHelper distributedCheckpointerHelper; | ||
| public CompactorController(String[] args) throws Exception { | ||
| this.config = new CompactorControllerConfig(args); | ||
|
|
||
| Thread.currentThread().setName("Cmpt-ctrls-" + config.getNodeLocator().getPort()); | ||
| this.corfuRuntime = (CorfuRuntime.fromParameters( | ||
| config.getParams())).parseConfigurationString(config.getNodeLocator().toEndpointUrl()).connect(); | ||
| this.corfuStore = new CorfuStore(corfuRuntime); | ||
|
|
||
| this.distributedCheckpointerHelper = new DistributedCheckpointerHelper(corfuStore); | ||
| } | ||
|
|
||
| public CompactorController(CompactorControllerConfig config, CorfuRuntime corfuRuntime, CorfuStore corfuStore, | ||
| DistributedCheckpointerHelper distributedCheckpointerHelper) { | ||
| this.config = config; | ||
| this.corfuRuntime = corfuRuntime; | ||
| this.corfuStore = corfuStore; | ||
| this.distributedCheckpointerHelper = distributedCheckpointerHelper; | ||
| } | ||
|
|
||
| /** | ||
| * Entry point to invoke compactor controls operations | ||
| * | ||
| * @param args command line argument strings | ||
| */ | ||
| public static void main(String[] args) { | ||
| try { | ||
| CompactorController corfuStoreCompactorControls = new CompactorController(args); | ||
| corfuStoreCompactorControls.doCompactorAction(); | ||
| } catch (Exception e) { | ||
| log.error("CorfuStoreCompactorMain crashed with error: {}, Exception: ", | ||
| CompactorBaseConfig.CORFU_LOG_CHECKPOINT_ERROR, e); | ||
| } | ||
| log.info("Exiting CorfuStoreCompactor"); | ||
| } | ||
|
|
||
| protected void doCompactorAction() { | ||
| if (config.isFreezeCompaction() || config.isDisableCompaction()) { | ||
| if (config.isDisableCompaction()) { | ||
| log.info("Disabling compaction..."); | ||
| distributedCheckpointerHelper.disableCompaction(); | ||
| } | ||
| if (config.isFreezeCompaction()) { | ||
| log.info("Freezing compaction..."); | ||
| distributedCheckpointerHelper.freezeCompaction(); | ||
| } | ||
| return; | ||
| } | ||
| if (config.isUnfreezeCompaction()) { | ||
| log.info("Unfreezing compaction..."); | ||
| distributedCheckpointerHelper.unfreezeCompaction(); | ||
| } | ||
| if (config.isEnableCompaction()) { | ||
| log.info("Enabling compaction..."); | ||
| distributedCheckpointerHelper.enableCompaction(); | ||
| } | ||
| if (config.isInstantTriggerCompaction()) { | ||
| if (config.isTrim()) { | ||
| log.info("Enabling instant compaction trigger with trim..."); | ||
| distributedCheckpointerHelper.instantTrigger(true); | ||
|
|
||
| } else { | ||
| log.info("Enabling instant compactor trigger..."); | ||
| distributedCheckpointerHelper.instantTrigger(false); | ||
| } | ||
| } | ||
| } | ||
| } |
57 changes: 57 additions & 0 deletions
57
corfudb-tools/src/main/java/org/corfudb/compactor/CompactorControllerConfig.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,57 @@ | ||
| package org.corfudb.compactor; | ||
|
|
||
| import lombok.Getter; | ||
| import lombok.extern.slf4j.Slf4j; | ||
|
|
||
| /** | ||
| * CorfuCompactorControlsConfig class parses the compactor controls specific parameters passed if any | ||
| */ | ||
| @Getter | ||
| @Slf4j | ||
| public class CompactorControllerConfig extends CompactorBaseConfig { | ||
| private final boolean upgradeDescriptorTable; | ||
| private final boolean instantTriggerCompaction; | ||
| private final boolean trim; | ||
| private final boolean freezeCompaction; | ||
| private final boolean unfreezeCompaction; | ||
| private final boolean disableCompaction; | ||
| private final boolean enableCompaction; | ||
|
|
||
| public CompactorControllerConfig(String[] args) { | ||
| super(args, CompactorControlsCmdLineHelper.USAGE_PARAMS, CompactorControlsCmdLineHelper.OPTIONS_PARAMS); | ||
|
|
||
| upgradeDescriptorTable = getOpt("--upgradeDescriptorTable").isPresent(); | ||
| instantTriggerCompaction = getOpt("--instantTriggerCompaction").isPresent(); | ||
| trim = getOpt("--trim").isPresent(); | ||
| freezeCompaction = getOpt("--freezeCompaction").isPresent(); | ||
| unfreezeCompaction = getOpt("--unfreezeCompaction").isPresent(); | ||
| disableCompaction = getOpt("--disableCompaction").isPresent(); | ||
| enableCompaction = getOpt("--enableCompaction").isPresent(); | ||
|
|
||
| if (freezeCompaction && unfreezeCompaction) { | ||
| log.error("Both freeze and unfreeze compaction parameters cannot be passed together"); | ||
| throw new IllegalArgumentException("Both freeze and unfreeze compaction parameters cannot be passed together"); | ||
| } | ||
| if (disableCompaction && enableCompaction) { | ||
| log.error("Both enable and disable compaction parameters cannot be passed together"); | ||
| throw new IllegalArgumentException("Both enable and disable compaction parameters cannot be passed together"); | ||
| } | ||
| } | ||
|
|
||
| public static class CompactorControlsCmdLineHelper { | ||
| public static final String USAGE_PARAMS = " [--trim=<trim>] " + | ||
| "[--upgradeDescriptorTable=<upgradeDescriptorTable>] " + | ||
| "[--instantTriggerCompaction=<instantTriggerCompaction>] " + | ||
| "[--freezeCompaction=<freezeCompaction>] " + | ||
| "[--unfreezeCompaction=<unfreezeCompaction>] " + | ||
| "[--disableCompaction=<disableCompaction>] " + | ||
| "[--enableCompaction=<enableCompaction>]"; | ||
| public static final String OPTIONS_PARAMS = "--trim=<trim> Should trim be performed along with instantTrigger\n" | ||
| + "--upgradeDescriptorTable=<upgradeDescriptorTable> Repopulate descriptor table?\n" | ||
| + "--instantTriggerCompaction=<instantTriggerCompaction> If compactor cycle needs to be triggered instantly\n" | ||
| + "--freezeCompaction=<freezeCompaction> If compaction needs to be frozen\n" | ||
| + "--unfreezeCompaction=<unfreezeCompaction> If compaction needs to be resumed\n" | ||
| + "--disableCompaction=<disableCompaction> If compaction needs to be disabled\n" | ||
| + "--enableCompaction=<enableCompaction> If compaction needs to be enabled"; | ||
| } | ||
| } |
Oops, something went wrong.