Stone_Red/dvc
1
0
Fork 0
mirror of https://github.com/Stone-Red-Code/dvc.git synced 2026-11-02 07:18:25 +01:00
Code Issues Packages Projects Releases Wiki Activity

oss: migrate to ossfs (#6307)

Browse Source 
* Migrate to ossfs

fix 6177

* Add verification to object stored on oss

* Change the oss remote's dependency.

pass all of the tests

* Performance optimaztion and ossfs version update

* Remove the argument to set the resumable get/put

1. Experiments shows resumable get/put has little influnence on performance.
2. Just use resumable ones as default

* Use a new api for performance optimization

* Without filesize progress bar version

* With progress bar version

* Bump and add comments

1. Bump ossfs from 2021.6.2 to 2021.7.0
2. add some comments to the verification trigger

* Change the name of kwrags following parent class.

* Bump to new ossfs and rewrite upload_fobj

Because some problem with oss-emulater
1. bump to ossfs 2021.7.1
2. rewrite upload_fobj

* Optimization on rm_files

* Some optimization follow the old version.

* Update dvc/fs/oss.py

Co-authored-by: Ruslan Kuprieiev <[email protected]>

* Update dvc/objects/db/oss.py

Co-authored-by: Ruslan Kuprieiev <[email protected]>

* bump into ossfs 2021.7.2

* Bump into ossfs 2021.7.3OC

* Switch to real_oss_server

* Switch to a real oss server

* Use the oss bucket located in US.

* Remove some emulator related code

* Add envs to ci

* remove some oss pytest XFAIL

Because we didn't use emulator, we can remove some pytest XFAILs

Co-authored-by: Ruslan Kuprieiev <[email protected]>
This commit is contained in:
Gao
2021-07-13 15:27:34 +03:00
committed by GitHub
co-authored by Ruslan Kuprieiev
parent 58f0911234
commit 93f36d9351
8 changed files with 69 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/tests.yaml
+3
View File
@@ -100,6 +100,9 @@ jobs:
GDRIVE_CREDENTIALS_DATA: ${{ secrets.GDRIVE_CREDENTIALS_DATA }}
AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
OSS_ACCESS_KEY_ID: ${{ secrets.OSS_ACCESS_KEY_ID}}
OSS_ACCESS_KEY_SECRET: ${{ secrets.OSS_ACCESS_KEY_SECRET}}
OSS_ENDPOINT: ${{ secrets.OSS_ENDPOINT}}
run: >-
python -m tests -n=4
--cov-report=xml --cov-report=term
dvc/fs/oss.py
+42 -93
View File
@@ -8,124 +8,73 @@ from dvc.path_info import CloudURLInfo
from dvc.progress import Tqdm
from dvc.scheme import Schemes
from .base import BaseFileSystem
from .fsspec_wrapper import ObjectFSWrapper
logger = logging.getLogger(__name__)
class OSSFileSystem(BaseFileSystem): # pylint:disable=abstract-method
"""
oss2 document:
https://www.alibabacloud.com/help/doc-detail/32026.htm
Examples
----------
$ dvc remote add myremote oss://my-bucket/path
Set key id, key secret and endpoint using modify command
$ dvc remote modify myremote oss_key_id my-key-id
$ dvc remote modify myremote oss_key_secret my-key-secret
$ dvc remote modify myremote oss_endpoint endpoint
or environment variables
$ export OSS_ACCESS_KEY_ID="my-key-id"
$ export OSS_ACCESS_KEY_SECRET="my-key-secret"
$ export OSS_ENDPOINT="endpoint"
"""
# pylint:disable=abstract-method
class OSSFileSystem(ObjectFSWrapper):
scheme = Schemes.OSS
PATH_CLS = CloudURLInfo
REQUIRES = {"oss2": "oss2"}
REQUIRES = {"ossfs": "ossfs"}
PARAM_CHECKSUM = "etag"
COPY_POLL_SECONDS = 5
LIST_OBJECT_PAGE_SIZE = 100
DETAIL_FIELDS = frozenset(("etag", "size"))
def __init__(self, **config):
super().__init__(**config)
self.endpoint = config.get("oss_endpoint") or os.getenv("OSS_ENDPOINT")
self.key_id = (
config.get("oss_key_id")
or os.getenv("OSS_ACCESS_KEY_ID")
or "defaultId"
def _prepare_credentials(self, **config):
login_info = {}
login_info["key"] = config.get("oss_key_id") or os.getenv(
"OSS_ACCESS_KEY_ID"
)
self.key_secret = (
config.get("oss_key_secret")
or os.getenv("OSS_ACCESS_KEY_SECRET")
or "defaultSecret"
login_info["secret"] = config.get("oss_key_secret") or os.getenv(
"OSS_ACCESS_KEY_SECRET"
)
login_info["endpoint"] = config.get("oss_endpoint")
return login_info
@wrap_prop(threading.Lock())
@cached_property
def oss_service(self):
import oss2
def fs(self):
from ossfs import OSSFileSystem as _OSSFileSystem
logger.debug(f"key id: {self.key_id}")
logger.debug(f"key secret: {self.key_secret}")
return oss2.Auth(self.key_id, self.key_secret)
def _get_bucket(self, bucket):
import oss2
return oss2.Bucket(self.oss_service, self.endpoint, bucket)
def _generate_download_url(self, path_info, expires=3600):
return self._get_bucket(path_info.bucket).sign_url(
"GET", path_info.path, expires
)
def exists(self, path_info) -> bool:
paths = self._list_paths(path_info)
return any(path_info.path == path for path in paths)
def _list_paths(self, path_info):
import oss2
for blob in oss2.ObjectIterator(
self._get_bucket(path_info.bucket), prefix=path_info.path
):
yield blob.key
def walk_files(self, path_info, **kwargs):
if not kwargs.pop("prefix", False):
path_info = path_info / ""
for fname in self._list_paths(path_info):
if fname.endswith("/"):
continue
yield path_info.replace(path=fname)
return _OSSFileSystem(**self.fs_args)
def remove(self, path_info):
if path_info.scheme != self.scheme:
raise NotImplementedError
logger.debug(f"Removing oss://{path_info}")
self._get_bucket(path_info.bucket).delete_object(path_info.path)
def _upload_fobj(self, fobj, to_info, **kwargs):
self._get_bucket(to_info.bucket).put_object(to_info.path, fobj)
self.fs.rm_file(self._with_bucket(path_info))
def _upload(
self, from_file, to_info, name=None, no_progress_bar=False, **_kwargs
self, from_file, to_info, name=None, no_progress_bar=False, **kwargs
):
with Tqdm(desc=name, disable=no_progress_bar, bytes=True) as pbar:
bucket = self._get_bucket(to_info.bucket)
bucket.put_object_from_file(
to_info.path, from_file, progress_callback=pbar.update_to
total = os.path.getsize(from_file)
with Tqdm(
disable=no_progress_bar,
total=total,
bytes=True,
desc=name,
**kwargs,
) as pbar:
self.fs.put_file(
from_file,
self._with_bucket(to_info),
progress_callback=pbar.update_to,
)
self.fs.invalidate_cache(self._with_bucket(to_info.parent))
def _download(
self, from_info, to_file, name=None, no_progress_bar=False, **_kwargs
self, from_info, to_file, name=None, no_progress_bar=False, **pbar_args
):
with Tqdm(desc=name, disable=no_progress_bar, bytes=True) as pbar:
import oss2
bucket = self._get_bucket(from_info.bucket)
oss2.resumable_download(
bucket,
from_info.path,
total = self.fs.size(self._with_bucket(from_info))
with Tqdm(
disable=no_progress_bar,
total=total,
bytes=True,
desc=name,
**pbar_args,
) as pbar:
self.fs.get_file(
self._with_bucket(from_info),
to_file,
progress_callback=pbar.update_to,
)
dvc/objects/db/__init__.py
+4
View File
@@ -5,6 +5,7 @@ def get_odb(fs, path_info, **config):
from .base import ObjectDB
from .gdrive import GDriveObjectDB
from .local import LocalObjectDB
from .oss import OSSObjectDB
from .ssh import SSHObjectDB
if fs.scheme == Schemes.LOCAL:
@@ -16,6 +17,9 @@ def get_odb(fs, path_info, **config):
if fs.scheme == Schemes.GDRIVE:
return GDriveObjectDB(fs, path_info, **config)
if fs.scheme == Schemes.OSS:
return OSSObjectDB(fs, path_info, **config)
return ObjectDB(fs, path_info, **config)
dvc/objects/db/oss.py
+9
View File
@@ -0,0 +1,9 @@
from .base import ObjectDB
class OSSObjectDB(ObjectDB):
"""
Temporary extra verification
"""
DEFAULT_VERIFY = True
setup.py
+1 -1
View File
@@ -101,7 +101,7 @@ gdrive = ["pydrive2>=1.8.1", "six >= 1.13.0"]
s3 = ["s3fs==2021.6.1", "aiobotocore[boto3]==1.3.0"]
azure = ["adlfs==2021.7.0", "azure-identity>=1.4.0", "knack"]
# https://github.com/Legrandin/pycryptodome/issues/465
oss = ["oss2==2.6.1", "pycryptodome>=3.10"]
oss = ["ossfs==2021.7.3"]
ssh = ["paramiko[invoke]>=2.7.0"]
# Remove the env marker if/when pyarrow is available for Python3.9
tests/func/test_data_cloud.py
+1 -7
View File
@@ -30,14 +30,8 @@ all_clouds = [
"hdfs",
"webdav",
"webhdfs",
"oss",
]
] + [
pytest.param(
pytest.lazy_fixture("oss"),
marks=pytest.mark.xfail(
reason="https://github.com/iterative/dvc/issues/4633"
),
)
]
# Clouds that implement the general methods that can be tested
tests/remotes/oss.py
+6 -13
View File
@@ -9,7 +9,7 @@ from dvc.utils import env2bool
from .base import Base
TEST_OSS_REPO_BUCKET = "dvc-test"
TEST_OSS_REPO_BUCKET = "dvc-test-github"
EMULATOR_OSS_ENDPOINT = "127.0.0.1:{port}"
EMULATOR_OSS_ACCESS_KEY_ID = "AccessKeyID"
EMULATOR_OSS_ACCESS_KEY_SECRET = "AccessKeySecret"
@@ -63,22 +63,15 @@ def oss_server(test_config, docker_compose, docker_services):
@pytest.fixture
def oss(oss_server):
def oss(real_oss):
import oss2
url = OSS.get_url()
ret = OSS(url)
ret.config = {
"url": url,
"oss_key_id": EMULATOR_OSS_ACCESS_KEY_ID,
"oss_key_secret": EMULATOR_OSS_ACCESS_KEY_SECRET,
"oss_endpoint": oss_server,
}
ret = real_oss
auth = oss2.Auth(
EMULATOR_OSS_ACCESS_KEY_ID, EMULATOR_OSS_ACCESS_KEY_SECRET
auth = oss2.Auth(ret.config["oss_key_id"], ret.config["oss_key_secret"])
bucket = oss2.Bucket(
auth, ret.config["oss_endpoint"], TEST_OSS_REPO_BUCKET
)
bucket = oss2.Bucket(auth, oss_server, TEST_OSS_REPO_BUCKET)
try:
bucket.get_bucket_info()
except oss2.exceptions.NoSuchBucket:
tests/unit/remote/test_oss.py
+3 -3
View File
@@ -16,6 +16,6 @@ def test_init(dvc):
"oss_endpoint": endpoint,
}
fs = OSSFileSystem(**config)
assert fs.endpoint == endpoint
assert fs.key_id == key_id
assert fs.key_secret == key_secret
assert fs.fs._endpoint == endpoint
assert fs.fs._auth.id == key_id
assert fs.fs._auth.secret == key_secret