feat: add oss storage type in blob sink

build.gradle

@@ -33,7 +33,7 @@ lombok {
 }
 
 group 'com.gotocompany'
-version '0.10.7'
+version '0.10.9'
 
 def projName = "firehose"
 
@@ -103,6 +103,7 @@ dependencies {
     implementation group: 'com.gotocompany', name: 'depot', version: '0.9.2'
     implementation group: 'com.networknt', name: 'json-schema-validator', version: '1.0.59' exclude group: 'org.slf4j'
     implementation 'dev.cel:cel:0.5.2'
+    implementation group: 'com.aliyun.oss', name: 'aliyun-sdk-oss', version: '3.18.1'
 
     testImplementation group: 'junit', name: 'junit', version: '4.11'
     testImplementation 'org.hamcrest:hamcrest-all:1.3'

docs/docs/sinks/blob-sink.md

@@ -262,3 +262,72 @@ The amount of time to allow the client to complete the execution of an API call.
 - Example value: `40000`
 - Type: `optional`
 - Default value : `40000`
+
+### OSS Configuration
+
+When using OSS as the blob storage type (`SINK_BLOB_STORAGE_TYPE=OSS`), the following configurations are required:
+
+### `SINK_BLOB_OSS_ENDPOINT`
+
+The endpoint of your OSS bucket.
+
+- Example value: `oss-cn-hangzhou.aliyuncs.com`
+- Type: `required`
+
+### `SINK_BLOB_OSS_ACCESS_KEY_ID`
+
+The access key ID for OSS authentication.
+
+- Type: `required`
+
+### `SINK_BLOB_OSS_ACCESS_KEY_SECRET`
+
+The access key secret for OSS authentication.
+
+- Type: `required`
+
+### `SINK_BLOB_OSS_BUCKET_NAME`
+
+The name of the OSS bucket.
+
+- Type: `required`
+
+### `SINK_BLOB_OSS_DIRECTORY_PREFIX`
+
+The directory prefix in the OSS bucket where files will be uploaded.
+
+- Example value: `data/raw`
+- Type: `optional`
+- Default value: ``
+
+### `SINK_BLOB_OSS_MAX_CONNECTIONS`
+
+Maximum number of concurrent connections to OSS.
+
+- Example value: `1024`
+- Type: `optional`
+- Default value: `1024`
+
+### `SINK_BLOB_OSS_SOCKET_TIMEOUT`
+
+Socket timeout in milliseconds.
+
+- Example value: `50000`
+- Type: `optional`
+- Default value: `50000`
+
+### `SINK_BLOB_OSS_CONNECTION_TIMEOUT`
+
+Connection timeout in milliseconds.
+
+- Example value: `50000`
+- Type: `optional`
+- Default value: `50000`
+
+### `SINK_BLOB_OSS_MAX_ERROR_RETRY`
+
+Maximum number of retry attempts for failed operations.
+
+- Example value: `3`
+- Type: `optional`
+- Default value: `3`

src/main/java/com/gotocompany/firehose/config/OSSConfig.java

@@ -0,0 +1,37 @@
+package com.gotocompany.firehose.config;
+
+import org.aeonbits.owner.Config;
+
+public interface OSSConfig extends Config {
+
+    @Key("${OSS_TYPE}_OSS_ENDPOINT")
+    String getOSSEndpoint();
+
+    @Key("${OSS_TYPE}_OSS_ACCESS_KEY_ID")
+    String getOSSAccessKeyId();
+
+    @Key("${OSS_TYPE}_OSS_ACCESS_KEY_SECRET")
+    String getOSSAccessKeySecret();
+
+    @Key("${OSS_TYPE}_OSS_BUCKET_NAME")
+    String getOSSBucketName();
+
+    @Key("${OSS_TYPE}_OSS_DIRECTORY_PREFIX")
+    String getOSSDirectoryPrefix();
+
+    @Key("${OSS_TYPE}_OSS_MAX_CONNECTIONS")
+    @DefaultValue("1024")
+    Integer getOSSMaxConnections();
+
+    @Key("${OSS_TYPE}_OSS_SOCKET_TIMEOUT")
+    @DefaultValue("50000")
+    Integer getOSSSocketTimeout();
+
+    @Key("${OSS_TYPE}_OSS_CONNECTION_TIMEOUT")
+    @DefaultValue("50000")
+    Integer getOSSConnectionTimeout();
+
+    @Key("${OSS_TYPE}_OSS_MAX_ERROR_RETRY")
+    @DefaultValue("3")
+    Integer getOSSMaxErrorRetry();
+}

src/main/java/com/gotocompany/firehose/sink/blob/BlobSinkFactory.java

@@ -71,10 +71,12 @@ public static BlobStorage createSinkObjectStorage(BlobSinkConfig sinkConfig, Map
             case S3:
                 configuration.put("S3_TYPE", "SINK_BLOB");
                 break;
+            case OSS:
+                configuration.put("OSS_TYPE", "SINK_BLOB");
+                break;
             default:
                 throw new IllegalArgumentException("Sink Blob Storage type " + sinkConfig.getBlobStorageType() + "is not supported");
         }
         return BlobStorageFactory.createObjectStorage(sinkConfig.getBlobStorageType(), configuration);
-
     }
 }

src/main/java/com/gotocompany/firehose/sink/common/blobstorage/BlobStorageFactory.java

@@ -1,8 +1,10 @@
 package com.gotocompany.firehose.sink.common.blobstorage;
 
 import com.gotocompany.firehose.config.GCSConfig;
+import com.gotocompany.firehose.config.OSSConfig;
 import com.gotocompany.firehose.config.S3Config;
 import com.gotocompany.firehose.sink.common.blobstorage.gcs.GoogleCloudStorage;
+import com.gotocompany.firehose.sink.common.blobstorage.oss.ObjectStorageService;
 import com.gotocompany.firehose.sink.common.blobstorage.s3.S3;
 import org.aeonbits.owner.ConfigFactory;
 
@@ -24,10 +26,16 @@ public static BlobStorage createObjectStorage(BlobStorageType storageType, Map<S
                 try {
                     S3Config s3Config = ConfigFactory.create(S3Config.class, config);
                     return new S3(s3Config);
-                 } catch (Exception e) {
+                } catch (Exception e) {
                     throw new IllegalArgumentException("Exception while creating S3 Storage", e);
                 }
-
+            case OSS:
+                try {
+                    OSSConfig ossConfig = ConfigFactory.create(OSSConfig.class, config);
+                    return new ObjectStorageService(ossConfig);
+                } catch (Exception e) {
+                    throw new IllegalArgumentException("Exception while creating OSS Storage", e);
+                }
             default:
                 throw new IllegalArgumentException("Blob Storage Type " + storageType + " is not supported");
         }

src/main/java/com/gotocompany/firehose/sink/common/blobstorage/BlobStorageType.java

@@ -2,5 +2,6 @@
 
 public enum BlobStorageType {
     GCS,
-    S3
+    S3,
+    OSS
 }

src/main/java/com/gotocompany/firehose/sink/common/blobstorage/oss/ObjectStorageService.java

@@ -0,0 +1,55 @@
+package com.gotocompany.firehose.sink.common.blobstorage.oss;
+
+import com.aliyun.oss.OSS;
+import com.aliyun.oss.OSSClientBuilder;
+import com.aliyun.oss.OSSException;
+import com.aliyun.oss.model.ObjectMetadata;
+import com.gotocompany.firehose.config.OSSConfig;
+import com.gotocompany.firehose.sink.common.blobstorage.BlobStorage;
+import com.gotocompany.firehose.sink.common.blobstorage.BlobStorageException;
+
+import java.io.ByteArrayInputStream;
+import java.io.File;
+import java.nio.file.Paths;
+
+public class ObjectStorageService implements BlobStorage {
+    private final OSS ossClient;
+    private final String bucketName;
+    private final String directoryPrefix;
+
+    public ObjectStorageService(OSSConfig config) {
+        this.ossClient = createOSSClient(
+                config.getOSSEndpoint(),
+                config.getOSSAccessKeyId(),
+                config.getOSSAccessKeySecret()
+        );
+        this.bucketName = config.getOSSBucketName();
+        this.directoryPrefix = config.getOSSDirectoryPrefix();
+    }
+
+    protected OSS createOSSClient(String endpoint, String accessKeyId, String accessKeySecret) {
+        return new OSSClientBuilder().build(endpoint, accessKeyId, accessKeySecret);
+    }
+
+    @Override
+    public void store(String objectName, String localPath) throws BlobStorageException {
+        try {
+            String fullPath = Paths.get(directoryPrefix, objectName).toString();
+            ossClient.putObject(bucketName, fullPath, new File(localPath));
+        } catch (OSSException e) {
+            throw new BlobStorageException(e.getErrorCode(), "OSS Upload failed", e);
+        }
+    }
+
+    @Override
+    public void store(String objectName, byte[] content) throws BlobStorageException {
+        try {
+            String fullPath = Paths.get(directoryPrefix, objectName).toString();
+            ObjectMetadata metadata = new ObjectMetadata();
+            metadata.setContentLength(content.length);
+            ossClient.putObject(bucketName, fullPath, new ByteArrayInputStream(content), metadata);
+        } catch (OSSException e) {
+            throw new BlobStorageException(e.getErrorCode(), "OSS Upload failed", e);
+        }
+    }
+}