-
Notifications
You must be signed in to change notification settings - Fork 190
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Moved Accumulator package to common for #874.
Signed-off-by: mallikagogoi7 <[email protected]>
- Loading branch information
1 parent
fe1c9d3
commit 71bfb78
Showing
12 changed files
with
197 additions
and
105 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
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
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
116 changes: 116 additions & 0 deletions
116
.../common/src/main/java/org/opensearch/dataprepper/plugins/accumulator/LocalFileBuffer.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,116 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.dataprepper.plugins.accumulator; | ||
|
|
||
| import org.apache.commons.lang3.time.StopWatch; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
| import software.amazon.awssdk.core.sync.RequestBody; | ||
| import software.amazon.awssdk.services.s3.S3Client; | ||
| import software.amazon.awssdk.services.s3.model.PutObjectRequest; | ||
|
|
||
| import java.io.BufferedOutputStream; | ||
| import java.io.File; | ||
| import java.io.FileNotFoundException; | ||
| import java.io.OutputStream; | ||
| import java.io.FileOutputStream; | ||
| import java.io.IOException; | ||
|
|
||
| import java.nio.file.Files; | ||
| import java.nio.file.Paths; | ||
| import java.util.concurrent.TimeUnit; | ||
|
|
||
| /** | ||
| * A buffer can hold local file data and flushing it to S3. | ||
| */ | ||
| public class LocalFileBuffer implements Buffer { | ||
|
|
||
| private static final Logger LOG = LoggerFactory.getLogger(LocalFileBuffer.class); | ||
| private final OutputStream outputStream; | ||
| private int eventCount; | ||
| private final StopWatch watch; | ||
| private final File localFile; | ||
|
|
||
| LocalFileBuffer(File tempFile) throws FileNotFoundException { | ||
| localFile = tempFile; | ||
| outputStream = new BufferedOutputStream(new FileOutputStream(tempFile)); | ||
| eventCount = 0; | ||
| watch = new StopWatch(); | ||
| watch.start(); | ||
| } | ||
|
|
||
| @Override | ||
| public long getSize() { | ||
| try { | ||
| outputStream.flush(); | ||
| } catch (IOException e) { | ||
| LOG.error("An exception occurred while flushing data to buffered output stream :", e); | ||
| } | ||
| return localFile.length(); | ||
| } | ||
|
|
||
| @Override | ||
| public int getEventCount() { | ||
| return eventCount; | ||
| } | ||
|
|
||
| @Override | ||
| public long getDuration(){ | ||
| return watch.getTime(TimeUnit.SECONDS); | ||
| } | ||
|
|
||
| /** | ||
| * Upload accumulated data to amazon s3. | ||
| * @param s3Client s3 client object. | ||
| * @param bucket bucket name. | ||
| * @param key s3 object key path. | ||
| */ | ||
| @Override | ||
| public void flushToS3(S3Client s3Client, String bucket, String key) { | ||
| flushAndCloseStream(); | ||
| s3Client.putObject( | ||
| PutObjectRequest.builder().bucket(bucket).key(key).build(), | ||
| RequestBody.fromFile(localFile)); | ||
| removeTemporaryFile(); | ||
| } | ||
|
|
||
| /** | ||
| * write byte array to output stream. | ||
| * @param bytes byte array. | ||
| * @throws IOException while writing to output stream fails. | ||
| */ | ||
| @Override | ||
| public void writeEvent(byte[] bytes) throws IOException { | ||
| outputStream.write(bytes); | ||
| outputStream.write(System.lineSeparator().getBytes()); | ||
| eventCount++; | ||
| } | ||
|
|
||
| /** | ||
| * Flushing the buffered data into the output stream. | ||
| */ | ||
| protected void flushAndCloseStream(){ | ||
| try { | ||
| outputStream.flush(); | ||
| outputStream.close(); | ||
| } catch (IOException e) { | ||
| LOG.error("An exception occurred while flushing data to buffered output stream :", e); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Remove the local temp file after flushing data to s3. | ||
| */ | ||
| protected void removeTemporaryFile() { | ||
| if (localFile != null) { | ||
| try { | ||
| Files.deleteIfExists(Paths.get(localFile.toString())); | ||
| } catch (IOException e) { | ||
| LOG.error("Unable to delete Local file {}", localFile, e); | ||
| } | ||
| } | ||
| } | ||
| } |
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
37 changes: 37 additions & 0 deletions
37
...-plugins/common/src/main/java/org/opensearch/dataprepper/plugins/sink/ThresholdCheck.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,37 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.dataprepper.plugins.sink; | ||
|
|
||
| import org.opensearch.dataprepper.model.types.ByteCount; | ||
| import org.opensearch.dataprepper.plugins.accumulator.Buffer; | ||
|
|
||
| /** | ||
| * Check threshold limits. | ||
| */ | ||
| public class ThresholdCheck { | ||
|
|
||
| private ThresholdCheck() { | ||
| } | ||
|
|
||
| /** | ||
| * Check threshold exceeds. | ||
| * @param currentBuffer current buffer. | ||
| * @param maxEvents maximum event provided by user as threshold. | ||
| * @param maxBytes maximum bytes provided by user as threshold. | ||
| * @param maxCollectionDuration maximum event collection duration provided by user as threshold. | ||
| * @return boolean value whether the threshold are met. | ||
| */ | ||
| public static boolean checkThresholdExceed(final Buffer currentBuffer, final int maxEvents, final ByteCount maxBytes, final long maxCollectionDuration) { | ||
| if (maxEvents > 0) { | ||
| return currentBuffer.getEventCount() + 1 > maxEvents || | ||
| currentBuffer.getDuration() > maxCollectionDuration || | ||
| currentBuffer.getSize() > maxBytes.getBytes(); | ||
| } else { | ||
| return currentBuffer.getDuration() > maxCollectionDuration || | ||
| currentBuffer.getSize() > maxBytes.getBytes(); | ||
| } | ||
| } | ||
| } |
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
Oops, something went wrong.