Skip to content

Commit

Permalink
feat(controller): support aliyun oss
Browse files Browse the repository at this point in the history
  • Loading branch information
jialeicui committed Sep 9, 2022
1 parent 920db1f commit 79da602
Show file tree
Hide file tree
Showing 16 changed files with 243 additions and 70 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand All @@ -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;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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
}
Expand All @@ -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) {
Expand Down
1 change: 1 addition & 0 deletions server/controller/src/main/resources/application.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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() {

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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() {
Expand Down
15 changes: 15 additions & 0 deletions server/pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -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>
Expand Down Expand Up @@ -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>

Expand Down Expand Up @@ -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>
Expand Down
16 changes: 4 additions & 12 deletions server/storage-access-layer/pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -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>
Expand Down Expand Up @@ -69,18 +73,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>
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,41 @@
/*
* 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;

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();
}

public long getSize() {
return this.size;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,87 @@
/*
* 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.aliyun;

import ai.starwhale.mlops.storage.LengthAbleInputStream;
import ai.starwhale.mlops.storage.StorageAccessService;
import ai.starwhale.mlops.storage.StorageObjectInfo;
import ai.starwhale.mlops.storage.s3.S3Config;
import ai.starwhale.mlops.storage.util.MetaHelper;
import com.aliyun.oss.OSS;
import com.aliyun.oss.OSSClientBuilder;
import com.aliyun.oss.model.GetObjectRequest;
import com.aliyun.oss.model.HeadObjectRequest;
import com.aliyun.oss.model.OSSObjectSummary;
import com.aliyun.oss.model.ObjectMetadata;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.stream.Stream;

public class StorageAccessServiceAliyun implements StorageAccessService {
final String bucket;

final OSS ossClient;

public StorageAccessServiceAliyun(S3Config s3Config) {
this.bucket = s3Config.getBucket();
this.ossClient = new OSSClientBuilder()
.build(s3Config.getEndpoint(), s3Config.getAccessKey(), s3Config.getSecretKey());
}

@Override
public StorageObjectInfo head(String path) throws IOException {
var resp = this.ossClient.headObject(new HeadObjectRequest(this.bucket, path));
return new StorageObjectInfo(true, resp.getContentLength(), MetaHelper.mapToString(resp.getUserMetadata()));
}

@Override
public void put(String path, InputStream inputStream, long size) throws IOException {
var meta = new ObjectMetadata();
meta.setContentLength(size);
this.ossClient.putObject(this.bucket, path, inputStream);
}

@Override
public void put(String path, byte[] body) throws IOException {
this.ossClient.putObject(this.bucket, path, new ByteArrayInputStream(body));
}

@Override
public LengthAbleInputStream get(String path) throws IOException {
var resp = this.ossClient.getObject(this.bucket, path);
return new LengthAbleInputStream(resp.getObjectContent(), resp.getObjectMetadata().getContentLength());
}

@Override
public LengthAbleInputStream get(String path, Long offset, Long size) throws IOException {
var req = new GetObjectRequest(bucket, path).withRange(offset, offset + size - 1);
var resp = this.ossClient.getObject(req);
return new LengthAbleInputStream(resp.getObjectContent(), resp.getObjectMetadata().getContentLength());
}

@Override
public Stream<String> list(String path) throws IOException {
var resp = this.ossClient.listObjects(this.bucket, path);
return resp.getObjectSummaries().stream().map(OSSObjectSummary::getKey);
}

@Override
public void delete(String path) throws IOException {
this.ossClient.deleteObject(this.bucket, path);
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
package ai.starwhale.mlops.storage.configuration;

import ai.starwhale.mlops.storage.StorageAccessService;
import ai.starwhale.mlops.storage.aliyun.StorageAccessServiceAliyun;
import ai.starwhale.mlops.storage.s3.StorageAccessServiceS3;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
Expand All @@ -29,8 +30,13 @@ public class StorageAccessConfig {

@Bean
@ConditionalOnProperty(prefix = "sw.storage", name = "type", havingValue = "s3", matchIfMissing = true)
public StorageAccessService storageAccessService(StorageProperties storageProperties) {
public StorageAccessService s3(StorageProperties storageProperties) {
return new StorageAccessServiceS3(storageProperties.getS3Config());
}

@Bean
@ConditionalOnProperty(prefix = "sw.storage", name = "type", havingValue = "aliyun")
public StorageAccessService aliyun(StorageProperties storageProperties) {
return new StorageAccessServiceAliyun(storageProperties.getS3Config());
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@ public class FileStorageEnv {
public static final String ENV_KEY_PREFIX = "SW_OBJECT_STORE_KEY_PREFIX";

public enum FileSystemEnvType {
S3, HDFS, NFS, LOCAL_FS, REST_RESOURCE, FTP
S3, ALIYUN, HDFS, NFS, LOCAL_FS, REST_RESOURCE, FTP
}

public FileStorageEnv add(String name, String value) {
Expand Down
Loading

0 comments on commit 79da602

Please sign in to comment.