Skip to content

Commit

Permalink
feat(controller): support aliyun oss (#1157)
Browse files Browse the repository at this point in the history
* feat(controller): support aliyun oss

* add aliyun in regex
  • Loading branch information
jialeicui authored Sep 13, 2022
1 parent 5a62e1b commit 3cf2e81
Show file tree
Hide file tree
Showing 17 changed files with 304 additions and 71 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
Original file line number Diff line number Diff line change
Expand Up @@ -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");
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 @@ -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>
Expand Down
Original file line number Diff line number Diff line change
@@ -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;
}
}
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
Loading

0 comments on commit 3cf2e81

Please sign in to comment.