feat(controller): support aliyun oss

server/controller/src/main/java/ai/starwhale/mlops/datastore/ObjectStore.java

@@ -24,8 +24,6 @@
 import java.io.IOException;
 import java.util.Iterator;
 import org.springframework.stereotype.Component;
-import software.amazon.awssdk.core.ResponseInputStream;
-import software.amazon.awssdk.services.s3.model.GetObjectResponse;
 
 @Component
 public class ObjectStore {
@@ -49,14 +47,7 @@ public void put(String name, SwBuffer buf) throws IOException {
 
     public SwBuffer get(String name) throws IOException {
         try (var is = this.storageAccessService.get(name)) {
-            int length;
-            if (is instanceof FileInputStream) {
-                length = (int) ((FileInputStream) is).getChannel().size();
-            } else {
-                //noinspection unchecked
-                length = ((ResponseInputStream<GetObjectResponse>) is).response().contentLength().intValue();
-            }
-            var ret = this.bufferManager.allocate(length);
+            var ret = this.bufferManager.allocate(Math.toIntExact(is.getSize()));
             int read = is.readNBytes(ret.asByteBuffer().array(), 0, ret.capacity());
             assert read == ret.capacity();
             return ret;

server/controller/src/main/java/ai/starwhale/mlops/domain/swds/objectstore/StorageAccessParser.java

@@ -21,8 +21,8 @@
 import ai.starwhale.mlops.exception.SwValidationException;
 import ai.starwhale.mlops.exception.SwValidationException.ValidSubject;
 import ai.starwhale.mlops.storage.StorageAccessService;
+import ai.starwhale.mlops.storage.aliyun.StorageAccessServiceAliyun;
 import ai.starwhale.mlops.storage.fs.FileStorageEnv;
-import ai.starwhale.mlops.storage.fs.FileStorageEnv.FileSystemEnvType;
 import ai.starwhale.mlops.storage.s3.S3Config;
 import ai.starwhale.mlops.storage.s3.StorageAccessServiceS3;
 import java.util.Map;
@@ -40,13 +40,13 @@ public class StorageAccessParser {
     ConcurrentHashMap<String, StorageAccessService> storageAccessServicePool = new ConcurrentHashMap<>();
 
     public StorageAccessParser(StorageAccessService defaultStorageAccessService,
-            SwDatasetVersionMapper swDatasetVersionMapper) {
+                               SwDatasetVersionMapper swDatasetVersionMapper) {
         this.defaultStorageAccessService = defaultStorageAccessService;
         this.swDatasetVersionMapper = swDatasetVersionMapper;
     }
 
     public StorageAccessService getStorageAccessServiceFromAuth(Long datasetId, String uri,
-            String authName) {
+                                                                String authName) {
         if (StringUtils.hasText(authName)) {
             authName = authName.toUpperCase(); // env vars are uppercase always
         }
@@ -67,15 +67,20 @@ public StorageAccessService getStorageAccessServiceFromAuth(Long datasetId, Stri
         if (null == env) {
             return defaultStorageAccessService;
         }
-        if (env.getEnvType() != FileSystemEnvType.S3) {
-            throw new SwValidationException(ValidSubject.SWDS).tip(
-                    "file system not supported yet: " + env.getEnvType());
+
+        switch (env.getEnvType()) {
+            case S3:
+                var s3 = new StorageAccessServiceS3(env2S3Config(new StorageUri(uri), env, authName));
+                storageAccessServicePool.putIfAbsent(formatKey(datasetId, authName), s3);
+                return s3;
+            case ALIYUN:
+                var aliyun = new StorageAccessServiceAliyun(env2S3Config(new StorageUri(uri), env, authName));
+                storageAccessServicePool.putIfAbsent(formatKey(datasetId, authName), aliyun);
+                return aliyun;
+            default:
+                throw new SwValidationException(ValidSubject.SWDS).tip(
+                        "file system not supported yet: " + env.getEnvType());
         }
-        StorageAccessServiceS3 storageAccessServiceS3 = new StorageAccessServiceS3(
-                env2S3Config(new StorageUri(uri), env, authName));
-        storageAccessServicePool.putIfAbsent(formatKey(datasetId, authName),
-                storageAccessServiceS3);
-        return storageAccessServiceS3;
     }
 
     String formatKey(Long datasetId, String authName) {

server/controller/src/main/java/ai/starwhale/mlops/domain/swds/objectstore/StorageAuths.java

@@ -34,7 +34,7 @@ public class StorageAuths {
     static final String NAME_DEFAULT = "";
 
     static final Pattern LINE_PATTERN = Pattern.compile(
-            "^(USER\\.(S3|HDFS|WEBHDFS|LOCALFS|NFS|FTP|SFTP|HTTP|HTTPS)\\.((\\w+)\\.)?(\\w+))=(\\w*)$");
+            "^(USER\\.(S3|ALIYUN|HDFS|WEBHDFS|LOCALFS|NFS|FTP|SFTP|HTTP|HTTPS)\\.((\\w+)\\.)?(\\w+))=(\\w*)$");
 
     public StorageAuths(String authsText) {
         String[] lines = authsText.split("\n");

server/controller/src/main/resources/application.yaml

@@ -33,6 +33,7 @@ sw:
       host-path-for-cache: ${SW_K8S_HOST_PATH_FOR_CACHE:/mnt/data}
       job-template-path: ${SW_K8S_JOB_TEMPLATE_PATH:}
   storage:
+    type: ${SW_STORAGE_TYPE:}
     path-prefix: ${SW_STORAGE_PREFIX:starwhale}
     fs-root-dir: ${SW_STORAGE_FS_ROOT_DIR:/usr/local/starwhale}
     s3-config:

server/controller/src/test/java/ai/starwhale/mlops/domain/swds/SwdsUploaderTest.java

@@ -50,6 +50,7 @@
 import ai.starwhale.mlops.domain.user.UserService;
 import ai.starwhale.mlops.domain.user.bo.User;
 import ai.starwhale.mlops.exception.SwValidationException;
+import ai.starwhale.mlops.storage.LengthAbleInputStream;
 import ai.starwhale.mlops.storage.StorageAccessService;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import java.io.ByteArrayInputStream;
@@ -131,7 +132,12 @@ public void testSwdsUploader() throws IOException {
                 .build();
         when(swdsVersionMapper.findByDsIdAndVersionNameForUpdate(1L, dsVersionId)).thenReturn(mockedEntity);
         when(swdsVersionMapper.findByDsIdAndVersionName(1L, dsVersionId)).thenReturn(mockedEntity);
-        when(storageAccessService.get(anyString())).thenReturn(new ByteArrayInputStream(index_file_content.getBytes()));
+        when(storageAccessService.get(anyString())).thenReturn(
+                new LengthAbleInputStream(
+                        new ByteArrayInputStream(index_file_content.getBytes()),
+                        index_file_content.getBytes().length
+                )
+        );
         HttpServletResponse httpResponse = mock(HttpServletResponse.class);
         ServletOutputStream mockOutPutStream = new ServletOutputStream() {
 

server/controller/src/test/java/ai/starwhale/mlops/domain/swds/objectstore/TestDsFileGetter.java

@@ -25,6 +25,7 @@
 import ai.starwhale.mlops.datastore.ColumnType;
 import ai.starwhale.mlops.domain.swds.mapper.SwDatasetVersionMapper;
 import ai.starwhale.mlops.domain.swds.po.SwDatasetVersionEntity;
+import ai.starwhale.mlops.storage.LengthAbleInputStream;
 import ai.starwhale.mlops.storage.StorageAccessService;
 import ai.starwhale.mlops.storage.StorageObjectInfo;
 import java.io.ByteArrayInputStream;
@@ -40,7 +41,7 @@ public void testFileGetter() throws IOException {
         StorageAccessService storageAccessService = mock(
                 StorageAccessService.class);
         when(storageAccessService.get(eq("bdc/bdcsd"), anyLong(), anyLong())).thenReturn(
-                new ByteArrayInputStream("abc".getBytes()));
+                new LengthAbleInputStream(new ByteArrayInputStream("abc".getBytes()), 3));
         when(storageAccessService.head("bdcsd")).thenReturn(new StorageObjectInfo(false, 1L, null));
         when(storageAccessService.head("bdc/bdcsd")).thenReturn(new StorageObjectInfo(true, 1L, null));
         when(storageAccessParser.getStorageAccessServiceFromAuth(anyLong(), anyString(), anyString())).thenReturn(

server/controller/src/test/java/ai/starwhale/mlops/resulting/ResultQuerierTest.java

@@ -24,6 +24,7 @@
 import ai.starwhale.mlops.domain.job.status.JobStatus;
 import ai.starwhale.mlops.exception.SwProcessException;
 import ai.starwhale.mlops.exception.SwValidationException;
+import ai.starwhale.mlops.storage.LengthAbleInputStream;
 import ai.starwhale.mlops.storage.StorageAccessService;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import java.io.ByteArrayInputStream;
@@ -122,12 +123,13 @@ private StorageAccessService mockStorageAccessService() throws IOException {
         return storageAccessService;
     }
 
-    private InputStream mockInputStream() {
-        return new ByteArrayInputStream(OBJECT.getBytes());
+    private LengthAbleInputStream mockInputStream() {
+        return new LengthAbleInputStream(new ByteArrayInputStream(OBJECT.getBytes()), OBJECT.getBytes().length);
     }
 
-    private InputStream mockResultInputStream() {
-        return new ByteArrayInputStream(MOCK_RESULT.getBytes());
+    private LengthAbleInputStream mockResultInputStream() {
+        return new LengthAbleInputStream(
+                new ByteArrayInputStream(MOCK_RESULT.getBytes()), MOCK_RESULT.getBytes().length);
     }
 
     JobMapper mockJobMapper() {

server/pom.xml

@@ -33,6 +33,8 @@
     <protoc-jar-maven-plugin.version>3.11.4</protoc-jar-maven-plugin.version>
     <maven-checkstyle-plugin.version>3.2.0</maven-checkstyle-plugin.version>
     <testcontainers.version>1.17.3</testcontainers.version>
+    <awssdk.version>2.17.159</awssdk.version>
+    <aliyunoss.version>3.15.1</aliyunoss.version>
     <starwhale.version>0.1.0-SNAPSHOT</starwhale.version>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
@@ -152,6 +154,18 @@
         <artifactId>testcontainers</artifactId>
         <version>${testcontainers.version}</version>
       </dependency>
+      <dependency>
+        <groupId>software.amazon.awssdk</groupId>
+        <artifactId>bom</artifactId>
+        <version>${awssdk.version}</version>
+        <type>pom</type>
+        <scope>import</scope>
+      </dependency>
+      <dependency>
+        <groupId>com.aliyun.oss</groupId>
+        <artifactId>aliyun-sdk-oss</artifactId>
+        <version>${aliyunoss.version}</version>
+      </dependency>
     </dependencies>
   </dependencyManagement>
 
@@ -263,6 +277,7 @@
               <exclude>**/ai/starwhale/mlops/datastore/*Wal.*</exclude>
               <exclude>**/ai/starwhale/mlops/datastore/*Wal$*.*</exclude>
               <exclude>**/ai/starwhale/mlops/datastore/Wal.*</exclude>
+              <exclude>**/ai/starwhale/mlops/storage/aliyun</exclude>
             </excludes>
           </configuration>
           <version>0.8.6</version>

server/storage-access-layer/pom.xml

@@ -24,6 +24,10 @@
       <groupId>software.amazon.awssdk</groupId>
       <artifactId>s3</artifactId>
     </dependency>
+      <dependency>
+        <groupId>com.aliyun.oss</groupId>
+        <artifactId>aliyun-sdk-oss</artifactId>
+      </dependency>
     <dependency>
       <groupId>org.projectlombok</groupId>
       <artifactId>lombok</artifactId>
@@ -63,18 +67,6 @@
     </dependency>
   </dependencies>
 
-  <dependencyManagement>
-    <dependencies>
-      <dependency>
-        <groupId>software.amazon.awssdk</groupId>
-        <artifactId>bom</artifactId>
-        <version>2.17.159</version>
-        <type>pom</type>
-        <scope>import</scope>
-      </dependency>
-    </dependencies>
-  </dependencyManagement>
-
   <build>
     <plugins>
       <plugin>

server/storage-access-layer/src/main/java/ai/starwhale/mlops/storage/LengthAbleInputStream.java

@@ -0,0 +1,101 @@
+/*
+ * Copyright 2022 Starwhale, Inc. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package ai.starwhale.mlops.storage;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+public class LengthAbleInputStream extends InputStream {
+    private final InputStream inputStream;
+
+    private final long size;
+
+    public LengthAbleInputStream(InputStream inputStream, long size) {
+        this.inputStream = inputStream;
+        this.size = size;
+    }
+
+    @Override
+    public int read() throws IOException {
+        return this.inputStream.read();
+    }
+
+    @Override
+    public int read(byte[] b) throws IOException {
+        return inputStream.read(b);
+    }
+
+    @Override
+    public int read(byte[] b, int off, int len) throws IOException {
+        return inputStream.read(b, off, len);
+    }
+
+    @Override
+    public byte[] readAllBytes() throws IOException {
+        return inputStream.readAllBytes();
+    }
+
+    @Override
+    public byte[] readNBytes(int len) throws IOException {
+        return inputStream.readNBytes(len);
+    }
+
+    @Override
+    public int readNBytes(byte[] b, int off, int len) throws IOException {
+        return inputStream.readNBytes(b, off, len);
+    }
+
+    @Override
+    public long skip(long n) throws IOException {
+        return inputStream.skip(n);
+    }
+
+    @Override
+    public int available() throws IOException {
+        return inputStream.available();
+    }
+
+    @Override
+    public void close() throws IOException {
+        inputStream.close();
+    }
+
+    @Override
+    public void mark(int readlimit) {
+        inputStream.mark(readlimit);
+    }
+
+    @Override
+    public void reset() throws IOException {
+        inputStream.reset();
+    }
+
+    @Override
+    public boolean markSupported() {
+        return inputStream.markSupported();
+    }
+
+    @Override
+    public long transferTo(OutputStream out) throws IOException {
+        return inputStream.transferTo(out);
+    }
+
+    public long getSize() {
+        return this.size;
+    }
+}

server/storage-access-layer/src/main/java/ai/starwhale/mlops/storage/StorageAccessService.java

@@ -31,9 +31,9 @@ public interface StorageAccessService {
 
     void put(String path, byte[] body) throws IOException;
 
-    InputStream get(String path) throws IOException;
+    LengthAbleInputStream get(String path) throws IOException;
 
-    InputStream get(String path, Long offset, Long size) throws IOException;
+    LengthAbleInputStream get(String path, Long offset, Long size) throws IOException;
 
     Stream<String> list(String path) throws IOException;