-
Notifications
You must be signed in to change notification settings - Fork 5
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* refactor http request as a stand alone class * fixes for pr * fix PR * changing import * * fixing spaces * adding queue impl * add sender builder * m * g * adding default queue function * get logs buffer interface as parameter * changing tests to abstract to fit both queues * adding long run test for in memory q * optimizing imports and editing README * adding a thread to check for disk space making the disk queue fully async * update release notes * adding null checks * deleting author + fixing tests * fixing long run tests log message * fix print Successfully sent bulk... when error happend and shouldRetry returns false * update readme * refactor http request as a stand alone class * fixes for pr * fix PR * changing import * * adding queue impl * fixing spaces * add sender builder * m * g * adding default queue function * get logs buffer interface as parameter * changing tests to abstract to fit both queues * adding long run test for in memory q * optimizing imports and editing README * adding a thread to check for disk space making the disk queue fully async * update release notes * adding null checks * deleting author + fixing tests * fixing long run tests log message * fix print Successfully sent bulk... when error happend and shouldRetry returns false * update readme * after PR - change style, selling, add constants, and renaming in memory class
- Loading branch information
Showing
17 changed files
with
797 additions
and
265 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
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
158 changes: 158 additions & 0 deletions
158
logzio-sender/src/main/java/io/logz/sender/DiskQueue.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,158 @@ | ||
| package io.logz.sender; | ||
|
|
||
| import com.bluejeans.common.bigqueue.BigQueue; | ||
| import io.logz.sender.exceptions.LogzioParameterErrorException; | ||
|
|
||
| import java.io.File; | ||
| import java.util.concurrent.ScheduledExecutorService; | ||
| import java.util.concurrent.TimeUnit; | ||
|
|
||
| public class DiskQueue implements LogzioLogsBufferInterface{ | ||
| private final BigQueue logsBuffer; | ||
| private final File queueDirectory; | ||
| private final boolean dontCheckEnoughDiskSpace; | ||
| private final int fsPercentThreshold; | ||
| private final SenderStatusReporter reporter; | ||
| private final ScheduledExecutorService diskSpaceTasks; | ||
| private volatile boolean isEnoughSpace; | ||
|
|
||
| private DiskQueue(File bufferDir, boolean dontCheckEnoughDiskSpace, int fsPercentThreshold, | ||
| int gcPersistedQueueFilesIntervalSeconds, SenderStatusReporter reporter, | ||
| int checkDiskSpaceInterval, ScheduledExecutorService diskSpaceTasks) | ||
| throws LogzioParameterErrorException { | ||
|
|
||
| this.reporter = reporter; | ||
| queueDirectory = bufferDir; | ||
| validateParameters(); | ||
| // divide bufferDir to dir and queue name | ||
| String dir = bufferDir.getAbsoluteFile().getParent(); | ||
| String queueNameDir = bufferDir.getName(); | ||
| if (dir == null || queueNameDir.isEmpty() ) { | ||
| throw new LogzioParameterErrorException("bufferDir", " value is empty: " + bufferDir.getAbsolutePath()); | ||
| } | ||
| logsBuffer = new BigQueue(dir, queueNameDir); | ||
| this.dontCheckEnoughDiskSpace = dontCheckEnoughDiskSpace; | ||
| this.fsPercentThreshold = fsPercentThreshold; | ||
| this.diskSpaceTasks = diskSpaceTasks; | ||
| this.isEnoughSpace = true; | ||
| diskSpaceTasks.scheduleWithFixedDelay(this::gcBigQueue, 0, gcPersistedQueueFilesIntervalSeconds, TimeUnit.SECONDS); | ||
| diskSpaceTasks.scheduleWithFixedDelay(this::isEnoughSpace, 0, checkDiskSpaceInterval, TimeUnit.MILLISECONDS); | ||
| } | ||
|
|
||
| private void validateParameters() throws LogzioParameterErrorException { | ||
| if (queueDirectory == null) { | ||
| throw new LogzioParameterErrorException("bufferDir", "value is null."); | ||
| } | ||
| if (reporter == null) { | ||
| throw new LogzioParameterErrorException("reporter", "value is null."); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void enqueue(byte[] log) { | ||
| if (isEnoughSpace) { | ||
| logsBuffer.enqueue(log); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public byte[] dequeue() { | ||
| return logsBuffer.dequeue(); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean isEmpty() { | ||
| return logsBuffer.isEmpty(); | ||
| } | ||
|
|
||
| private void isEnoughSpace() { | ||
| if (dontCheckEnoughDiskSpace) { | ||
| return; | ||
| } | ||
| int actualUsedFsPercent = 100 - ((int) (((double) queueDirectory.getUsableSpace() / queueDirectory.getTotalSpace()) * 100)); | ||
| if (actualUsedFsPercent >= fsPercentThreshold) { | ||
| reporter.warning(String.format("Logz.io: Dropping logs, as FS used space on %s is %d percent, and the drop threshold is %d percent", | ||
| queueDirectory.getAbsolutePath(), actualUsedFsPercent, fsPercentThreshold)); | ||
| isEnoughSpace = false; | ||
| } else { | ||
| isEnoughSpace = true; | ||
| } | ||
| } | ||
|
|
||
| private void gcBigQueue() { | ||
| try { | ||
| logsBuffer.gc(); | ||
| } catch (Throwable e) { | ||
| // We cant throw anything out, or the task will stop, so just swallow all | ||
| reporter.error("Uncaught error from BigQueue.gc()", e); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void close() { | ||
| gcBigQueue(); | ||
| } | ||
|
|
||
| public static class Builder { | ||
| private boolean dontCheckEnoughDiskSpace = false; | ||
| private int fsPercentThreshold = 98; | ||
| private int gcPersistedQueueFilesIntervalSeconds = 30; | ||
| private int checkDiskSpaceInterval = 1000; | ||
| private File bufferDir; | ||
| private SenderStatusReporter reporter; | ||
| private ScheduledExecutorService diskSpaceTasks; | ||
| private LogzioSender.Builder context; | ||
|
|
||
| Builder(LogzioSender.Builder context, ScheduledExecutorService diskSpaceTasks) { | ||
| this.context = context; | ||
| this.diskSpaceTasks = diskSpaceTasks; | ||
| } | ||
|
|
||
| public Builder setFsPercentThreshold(int fsPercentThreshold) { | ||
| this.fsPercentThreshold = fsPercentThreshold; | ||
| if (fsPercentThreshold == -1) { | ||
| dontCheckEnoughDiskSpace = true; | ||
| } | ||
| return this; | ||
| } | ||
|
|
||
| public Builder setGcPersistedQueueFilesIntervalSeconds(int gcPersistedQueueFilesIntervalSeconds) { | ||
| this.gcPersistedQueueFilesIntervalSeconds = gcPersistedQueueFilesIntervalSeconds; | ||
| return this; | ||
| } | ||
|
|
||
| public Builder setCheckDiskSpaceInterval(int checkDiskSpaceInterval) { | ||
| this.checkDiskSpaceInterval = checkDiskSpaceInterval; | ||
| return this; | ||
| } | ||
|
|
||
| public Builder setBufferDir(File bufferDir) { | ||
| this.bufferDir = bufferDir; | ||
| return this; | ||
| } | ||
|
|
||
| Builder setReporter(SenderStatusReporter reporter) { | ||
| this.reporter = reporter; | ||
| return this; | ||
| } | ||
|
|
||
| Builder setDiskSpaceTasks(ScheduledExecutorService diskSpaceTasks) { | ||
| this.diskSpaceTasks = diskSpaceTasks; | ||
| return this; | ||
| } | ||
|
|
||
| public LogzioSender.Builder EndDiskQueue() { | ||
| context.setDiskQueueBuilder(this); | ||
| return context; | ||
| } | ||
|
|
||
| DiskQueue build() throws LogzioParameterErrorException { | ||
| return new DiskQueue(bufferDir, dontCheckEnoughDiskSpace, fsPercentThreshold, | ||
| gcPersistedQueueFilesIntervalSeconds, reporter, checkDiskSpaceInterval, diskSpaceTasks); | ||
| } | ||
| } | ||
|
|
||
| public static Builder builder(LogzioSender.Builder context, ScheduledExecutorService diskSpaceTasks){ | ||
| return new Builder(context, diskSpaceTasks); | ||
| } | ||
| } |
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
Oops, something went wrong.