forked from raystack/firehose
-
Notifications
You must be signed in to change notification settings - Fork 4
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
a376678
commit 6409a7e
Showing
8 changed files
with
404 additions
and
3 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
49 changes: 49 additions & 0 deletions
49
src/main/java/com/gotocompany/firehose/config/ObjectStorageServiceConfig.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,49 @@ | ||
| package com.gotocompany.firehose.config; | ||
|
|
||
| import org.aeonbits.owner.Config; | ||
|
|
||
| public interface ObjectStorageServiceConfig extends Config { | ||
|
|
||
| @Key("${OSS_TYPE}_OSS_ENDPOINT") | ||
| String getOssEndpoint(); | ||
|
|
||
| @Key("${OSS_TYPE}_OSS_REGION") | ||
| String getOssRegion(); | ||
|
|
||
| @Key("${OSS_TYPE}_OSS_ACCESS_ID") | ||
| String getOssAccessId(); | ||
|
|
||
| @Key("${OSS_TYPE}_OSS_ACCESS_KEY") | ||
| String getOssAccessKey(); | ||
|
|
||
| @Key("${OSS_TYPE}_OSS_BUCKET_NAME") | ||
| String getOssBucketName(); | ||
|
|
||
| @Key("${OSS_TYPE}_OSS_DIRECTORY_PREFIX") | ||
| String getOssDirectoryPrefix(); | ||
|
|
||
| @Key("${OSS_TYPE}_OSS_SOCKET_TIMEOUT_MS") | ||
| @DefaultValue("50000") | ||
| Integer getOssSocketTimeoutMs(); | ||
|
|
||
| @Key("${OSS_TYPE}_OSS_CONNECTION_TIMEOUT_MS") | ||
| @DefaultValue("50000") | ||
| Integer getOssConnectionTimeoutMs(); | ||
|
|
||
| @Key("${OSS_TYPE}_OSS_CONNECTION_REQUEST_TIMEOUT_MS") | ||
| @DefaultValue("-1") | ||
| Integer getOssConnectionRequestTimeoutMs(); | ||
|
|
||
| @Key("${OSS_TYPE}_OSS_REQUEST_TIMEOUT_MS") | ||
| @DefaultValue("300000") | ||
| Integer getOssRequestTimeoutMs(); | ||
|
|
||
| @Key("${OSS_TYPE}_OSS_RETRY_ENABLED") | ||
| @DefaultValue("true") | ||
| boolean isRetryEnabled(); | ||
|
|
||
| @Key("${OSS_TYPE}_OSS_MAX_RETRY_ATTEMPTS") | ||
| @DefaultValue("3") | ||
| int getOssMaxRetryAttempts(); | ||
|
|
||
| } |
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -2,5 +2,6 @@ | |
|
|
||
| public enum BlobStorageType { | ||
| GCS, | ||
| S3 | ||
| S3, | ||
| OSS | ||
| } | ||
108 changes: 108 additions & 0 deletions
108
src/main/java/com/gotocompany/firehose/sink/common/blobstorage/oss/ObjectStorageService.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,108 @@ | ||
| package com.gotocompany.firehose.sink.common.blobstorage.oss; | ||
|
|
||
| import com.aliyun.oss.ClientBuilderConfiguration; | ||
| import com.aliyun.oss.ClientException; | ||
| import com.aliyun.oss.OSS; | ||
| import com.aliyun.oss.OSSClientBuilder; | ||
| import com.aliyun.oss.OSSException; | ||
| import com.aliyun.oss.common.auth.DefaultCredentialProvider; | ||
| import com.aliyun.oss.common.comm.NoRetryStrategy; | ||
| import com.aliyun.oss.common.comm.SignVersion; | ||
| import com.aliyun.oss.model.BucketList; | ||
| import com.aliyun.oss.model.ListBucketsRequest; | ||
| import com.aliyun.oss.model.PutObjectRequest; | ||
| import com.gotocompany.firehose.config.ObjectStorageServiceConfig; | ||
| import com.gotocompany.firehose.sink.common.blobstorage.BlobStorage; | ||
| import com.gotocompany.firehose.sink.common.blobstorage.BlobStorageException; | ||
| import lombok.extern.slf4j.Slf4j; | ||
|
|
||
| import java.io.ByteArrayInputStream; | ||
| import java.io.File; | ||
| import java.util.Optional; | ||
|
|
||
| @Slf4j | ||
| public class ObjectStorageService implements BlobStorage { | ||
|
|
||
| private final OSS oss; | ||
| private final ObjectStorageServiceConfig objectStorageServiceConfig; | ||
|
|
||
| public ObjectStorageService(ObjectStorageServiceConfig objectStorageServiceConfig) { | ||
| this(objectStorageServiceConfig, initializeOss(objectStorageServiceConfig)); | ||
| } | ||
|
|
||
| public ObjectStorageService(ObjectStorageServiceConfig objectStorageServiceConfig, OSS oss) { | ||
| this.oss = oss; | ||
| this.objectStorageServiceConfig = objectStorageServiceConfig; | ||
| checkBucket(); | ||
| } | ||
|
|
||
| private static OSS initializeOss(ObjectStorageServiceConfig objectStorageServiceConfig) { | ||
| ClientBuilderConfiguration clientBuilderConfiguration = new ClientBuilderConfiguration(); | ||
| clientBuilderConfiguration.setSignatureVersion(SignVersion.V4); | ||
| clientBuilderConfiguration.setSocketTimeout(objectStorageServiceConfig.getOssSocketTimeoutMs()); | ||
| clientBuilderConfiguration.setConnectionTimeout(objectStorageServiceConfig.getOssConnectionTimeoutMs()); | ||
| clientBuilderConfiguration.setConnectionRequestTimeout(objectStorageServiceConfig.getOssConnectionRequestTimeoutMs()); | ||
| clientBuilderConfiguration.setRequestTimeout(objectStorageServiceConfig.getOssRequestTimeoutMs()); | ||
| if (objectStorageServiceConfig.isRetryEnabled()) { | ||
| clientBuilderConfiguration.setMaxErrorRetry(objectStorageServiceConfig.getOssMaxRetryAttempts()); | ||
| } else { | ||
| clientBuilderConfiguration.setRetryStrategy(new NoRetryStrategy()); | ||
| } | ||
| return OSSClientBuilder.create() | ||
| .endpoint(objectStorageServiceConfig.getOssEndpoint()) | ||
| .region(objectStorageServiceConfig.getOssRegion()) | ||
| .credentialsProvider(new DefaultCredentialProvider(objectStorageServiceConfig.getOssAccessId(), | ||
| objectStorageServiceConfig.getOssAccessKey())) | ||
| .clientConfiguration(clientBuilderConfiguration) | ||
| .build(); | ||
| } | ||
|
|
||
| @Override | ||
| public void store(String objectName, String filePath) throws BlobStorageException { | ||
| PutObjectRequest putObjectRequest = new PutObjectRequest( | ||
| objectStorageServiceConfig.getOssBucketName(), | ||
| buildObjectPath(objectName), | ||
| new File(filePath) | ||
| ); | ||
| putObject(putObjectRequest); | ||
| } | ||
|
|
||
| @Override | ||
| public void store(String objectName, byte[] content) throws BlobStorageException { | ||
| PutObjectRequest putObjectRequest = new PutObjectRequest( | ||
| objectStorageServiceConfig.getOssBucketName(), | ||
| buildObjectPath(objectName), | ||
| new ByteArrayInputStream(content) | ||
| ); | ||
| putObject(putObjectRequest); | ||
| } | ||
|
|
||
| private void putObject(PutObjectRequest putObjectRequest) throws BlobStorageException { | ||
| try { | ||
| oss.putObject(putObjectRequest); | ||
| } catch (ClientException e) { | ||
| log.error("Failed to put object to OSS", e); | ||
| throw new BlobStorageException("client_error", e.getMessage(), e); | ||
| } catch (OSSException e) { | ||
| log.error("Failed to put object to OSS requestID:{} hostID:{}", e.getRequestId(), e.getHostId()); | ||
| throw new BlobStorageException(e.getErrorCode(), e.getErrorMessage(), e); | ||
| } | ||
| } | ||
|
|
||
| private String buildObjectPath(String objectName) { | ||
| return Optional.ofNullable(objectStorageServiceConfig.getOssDirectoryPrefix()) | ||
| .map(prefix -> prefix + "/" + objectName) | ||
| .orElse(objectName); | ||
| } | ||
|
|
||
| private void checkBucket() { | ||
| BucketList bucketList = oss.listBuckets(new ListBucketsRequest(objectStorageServiceConfig.getOssBucketName(), | ||
| null, 1)); | ||
| if (bucketList.getBucketList().isEmpty()) { | ||
| log.error("Bucket does not exist:{}", objectStorageServiceConfig.getOssBucketName()); | ||
| log.error("Please create OSS bucket before running firehose: {}", objectStorageServiceConfig.getOssBucketName()); | ||
| throw new IllegalArgumentException("Bucket does not exist"); | ||
| } | ||
| } | ||
|
|
||
| } |
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.