private CompletableFuture < Void > takeSnapshots ( boolean rescheduleAfterCompletion , boolean force ) { // If compaction is already in progress, return the existing future and reschedule if this is a scheduled compaction. if ( compactFuture != null ) { if ( rescheduleAfterCompletion ) { compactFuture . whenComplete ( ( r , e ) -> scheduleSnapshots ( ) ) ; } return compactFuture ; } long lastApplied = raft . getLastApplied ( ) ; // Only take snapshots if segments can be removed from the log below the lastApplied index. if ( raft . getLog ( ) . isCompactable ( lastApplied ) && raft . getLog ( ) . getCompactableIndex ( lastApplied ) > lastCompacted ) { // Determine whether the node is running out of disk space. boolean runningOutOfDiskSpace = isRunningOutOfDiskSpace ( ) ; // Determine whether the node is running out of memory. boolean runningOutOfMemory = isRunningOutOfMemory ( ) ; // If compaction is not already being forced... if ( ! force // And the node isn't running out of memory (we need to free up memory if it is)... && ! runningOutOfMemory // And dynamic compaction is enabled (we need to compact immediately if it's disabled)... && raft . getStorage ( ) . dynamicCompaction ( ) // And the node isn't running out of disk space (we need to compact immediately if it is)... && ! runningOutOfDiskSpace // And the server is under high load (we can skip compaction at this point)... && raft . getLoadMonitor ( ) . isUnderHighLoad ( ) ) { // We can skip taking a snapshot for now. logger . debug ( "Skipping compaction due to high load" ) ; if ( rescheduleAfterCompletion ) { scheduleSnapshots ( ) ; } return CompletableFuture . completedFuture ( null ) ; } logger . debug ( "Snapshotting services" ) ; // Update the index at which the log was last compacted. this . lastCompacted = lastApplied ; // We need to ensure that callbacks added to the compaction future are completed in the order in which they // were added in order to preserve the order of retries when appending to the log. compactFuture = new OrderedFuture <> ( ) ; // Wait for snapshots in all state machines to be completed before compacting the log at the last applied index. takeSnapshots ( ) . whenComplete ( ( snapshot , error ) -> { if ( error == null ) { scheduleCompletion ( snapshot . persist ( ) ) ; } } ) ; // Reschedule snapshots after completion if necessary. if ( rescheduleAfterCompletion ) { compactFuture . whenComplete ( ( r , e ) -> scheduleSnapshots ( ) ) ; } return compactFuture ; } // Otherwise, if the log can't be compacted anyways, just reschedule snapshots. else { if ( rescheduleAfterCompletion ) { scheduleSnapshots ( ) ; } return CompletableFuture . completedFuture ( null ) ; } }
Takes a snapshot of all services and compacts logs if the server is not under high load or disk needs to be freed .