Repository: iterative/dvc
Pull request: GS progress for push & pull
- [x] upload with progress
- [x] download with progress
- [x] add totals, filenames, disable, etc
- [x] test
- fixes #1566

Problem statement:
gs: support progress callback
I may have missed an issue concerning this, but when I am `dvc push`ing to a google cloud remote, the progress bars that are currently displayed don't get updated progressively (They go from 0 when starting to 100 when finished).

It would be nice to have a dynamic progressbar with upload speed and remaining time estimation, especially when push huge sets of data.

Hints from issue discussion:
Hi @mhham !

Indeed, callback for google cloud storate push/pull is not implemented yet. Let's keep this issue open as a reminder. Thank you for the feedback! 🙂 
Problem is there's no current API.

`google.cloud.storage.bucket.Bucket.`:
- [`copy_blob`](https://googleapis.github.io/google-cloud-python/latest/storage/buckets.html#google.cloud.storage.bucket.Bucket.copy_blob)
- [`upload_from_filename`](https://googleapis.github.io/google-cloud-python/latest/storage/blobs.html#google.cloud.storage.blob.Blob.upload_from_filename)
- [`download_to_filename`](https://googleapis.github.io/google-cloud-python/latest/storage/blobs.html#google.cloud.storage.blob.Blob.download_to_filename)

Possibly could work around this by using [`upload_from_file`](https://googleapis.github.io/google-cloud-python/latest/storage/blobs.html#google.cloud.storage.blob.Blob.upload_from_file) and [`download_to_file`](https://googleapis.github.io/google-cloud-python/latest/storage/blobs.html#google.cloud.storage.blob.Blob.download_to_file) with our own custom file objects which have progress for `read()` and `write()`
Looks good, let me look at it.
Let's focus on the downloading first.. .
> with our own custom file objects

What do you mean @casperdcl ? Could you provide a link to that? I only found this
https://github.com/iterative/dvc/blob/38c210068ba5aeb70ad92a8c22e85832640d8ebd/dvc/api.py#L56
which doesn't seems to be helpful.

I think the more appropriate approach would be to use this [google-resumable-media](https://googleapis.github.io/google-resumable-media-python/latest/google.resumable_media.requests.html#chunked-downloads) as it's recommended 

> For more fine-grained control over the download process, check out google-resumable-media. For example, this library allows downloading parts of a blob rather than the whole thing.

[here](https://googleapis.dev/python/storage/latest/blobs.html#google.cloud.storage.blob.Blob.download_to_file).

I will check if using `download_to_file` with `start` & `end` gives the similar result to `google-resumable-media`. If so, we could avoid adding a new dependency. 

> .. with our own custom file objects which have progress for read() and write()

Ok, got it. I think @casperdcl meant the custom file object is to be written as a part of this task.
Based on the resumable media link it seems like the proposed `download_to_file` approach would be just as complicated to implement really. If you have issues though feel free to open an incomplete PR and I can take a look.
My main problem is testing - @efiop how would you recommended setting up local testing for google cloud storage push/pull?
@casperdcl I'm not aware of any local implementation that you could run on your machine 🙁 So far I was just using real GS to test things out.
@casperdcl in meantime i started to play a bit with your concept of custom file object and did this very limited PoC (since it's PoC I used a simple file in the root of the dir &  `unittests`)
https://github.com/iterative/dvc/compare/master...xliiv:progress?expand=1

It's for `StringIO`, so some work needed to adjust it to our needs, but looks promising.

Maybe `tqdm` should expose objects like the `ProgressStringIO`?
@xliiv A side note: Please use pytest-style tests for new tests. We do have some unittest tests as a legacy, but all new ones are written with pytest.

Also, why StringIO and not BytestIO?

Commit message: neaten default chunks

Fixes https://github.com/iterative/dvc/pull/2809#discussion_r347178103

Patch to review:
diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py
index c2fb4efc1c..3091c628dc 100644
--- a/dvc/remote/gs.py
+++ b/dvc/remote/gs.py
@@ -1,14 +1,17 @@
-from __future__ import unicode_literals
+from __future__ import unicode_literals, division
 
 import logging
 from datetime import timedelta
 from functools import wraps
+import io
+import os.path
 
 from funcy import cached_property
 
 from dvc.config import Config
 from dvc.exceptions import DvcException
 from dvc.path_info import CloudURLInfo
+from dvc.progress import Tqdm
 from dvc.remote.base import RemoteBASE
 from dvc.scheme import Schemes
 from dvc.utils.compat import FileNotFoundError  # skipcq: PYL-W0622
@@ -20,25 +23,24 @@ def dynamic_chunk_size(func):
     @wraps(func)
     def wrapper(*args, **kwargs):
         import requests
-        from google.cloud.storage.blob import Blob, _DEFAULT_CHUNKSIZE
+        from google.cloud.storage.blob import _DEFAULT_CHUNKSIZE
 
-        # Default chunk size for gs is 100M, which might be too much for
-        # particular network (see [1]). So if we are getting ConnectionError,
-        # we should try lowering the chunk size until we reach the minimum
-        # allowed chunk size of 256K. Also note that `chunk_size` must be a
-        # multiple of 256K per the API specification.
+        # `ConnectionError` may be due to too large `chunk_size`
+        # (see [#2572]) so try halving on error.
+        # Note: start with 40 * [default: 256K] = 10M.
+        # Note: must be multiple of 256K.
         #
-        # [1] https://github.com/iterative/dvc/issues/2572
+        # [#2572]: https://github.com/iterative/dvc/issues/2572
 
         # skipcq: PYL-W0212
-        multiplier = int(_DEFAULT_CHUNKSIZE / Blob._CHUNK_SIZE_MULTIPLE)
+        multiplier = 40
         while True:
             try:
                 # skipcq: PYL-W0212
-                chunk_size = Blob._CHUNK_SIZE_MULTIPLE * multiplier
+                chunk_size = _DEFAULT_CHUNKSIZE * multiplier
                 return func(*args, chunk_size=chunk_size, **kwargs)
             except requests.exceptions.ConnectionError:
-                multiplier = int(multiplier / 2)
+                multiplier //= 2
                 if not multiplier:
                     raise
 
@@ -46,9 +48,22 @@ def wrapper(*args, **kwargs):
 
 
 @dynamic_chunk_size
-def _upload_to_bucket(bucket, from_file, to_info, **kwargs):
-    blob = bucket.blob(to_info.path, **kwargs)
-    blob.upload_from_filename(from_file)
+def _upload_to_bucket(bucket, from_file, to_info, chunk_size=None, **kwargs):
+    blob = bucket.blob(to_info.path, chunk_size=chunk_size, **kwargs)
+    with Tqdm(
+        desc=to_info.path, total=os.path.getsize(from_file), bytes=True
+    ) as pbar:
+        with io.open(from_file, mode="rb") as fd:
+            raw_read = fd.read
+
+            def read(size=chunk_size):
+                res = raw_read(size)
+                if res:
+                    pbar.update(len(res))
+                return res
+
+            fd.read = read
+            blob.upload_from_file(fd)
 
 
 class RemoteGS(RemoteBASE):
@@ -130,7 +145,16 @@ def _upload(self, from_file, to_info, **_kwargs):
     def _download(self, from_info, to_file, **_kwargs):
         bucket = self.gs.bucket(from_info.bucket)
         blob = bucket.get_blob(from_info.path)
-        blob.download_to_filename(to_file)
+        with Tqdm(desc=from_info.path, total=blob.size, bytes=True) as pbar:
+            with io.open(to_file, mode="wb") as fd:
+                raw_write = fd.write
+
+                def write(bytes):
+                    raw_write(bytes)
+                    pbar.update(len(bytes))
+
+                fd.write = write
+                blob.download_to_file(fd)
 
     def _generate_download_url(self, path_info, expires=3600):
         expiration = timedelta(seconds=int(expires))


Code review comments:
diff_hunk: @@ -20,35 +23,47 @@ def dynamic_chunk_size(func):
     @wraps(func)
     def wrapper(*args, **kwargs):
         import requests
-        from google.cloud.storage.blob import Blob, _DEFAULT_CHUNKSIZE
+        from google.cloud.storage.blob import _DEFAULT_CHUNKSIZE
 
-        # Default chunk size for gs is 100M, which might be too much for
-        # particular network (see [1]). So if we are getting ConnectionError,
-        # we should try lowering the chunk size until we reach the minimum
-        # allowed chunk size of 256K. Also note that `chunk_size` must be a
-        # multiple of 256K per the API specification.
+        # `ConnectionError` may be due to too large `chunk_size`
+        # (see [#2572]) so try halving on error.
+        # Note: start with 40 * [default: 256K] = 10M.
+        # Note: must be multiple of 256K.
         #
-        # [1] https://github.com/iterative/dvc/issues/2572
+        # [#2572]: https://github.com/iterative/dvc/issues/2572
 
         # skipcq: PYL-W0212
-        multiplier = int(_DEFAULT_CHUNKSIZE / Blob._CHUNK_SIZE_MULTIPLE)
+        multiplier = 40
         while True:
             try:
                 # skipcq: PYL-W0212
-                chunk_size = Blob._CHUNK_SIZE_MULTIPLE * multiplier
+                chunk_size = _DEFAULT_CHUNKSIZE * multiplier
path: dvc/remote/gs.py
text: @user1:
_DEFAULT_CHUNKSIZE is 100M, so you are setting it to 400M. Probably not what you meant :)

@author:
4,000M actually :D

diff_hunk: @@ -20,35 +23,47 @@ def dynamic_chunk_size(func):
     @wraps(func)
     def wrapper(*args, **kwargs):
         import requests
-        from google.cloud.storage.blob import Blob, _DEFAULT_CHUNKSIZE
+        from google.cloud.storage.blob import _DEFAULT_CHUNKSIZE
 
-        # Default chunk size for gs is 100M, which might be too much for
-        # particular network (see [1]). So if we are getting ConnectionError,
-        # we should try lowering the chunk size until we reach the minimum
-        # allowed chunk size of 256K. Also note that `chunk_size` must be a
-        # multiple of 256K per the API specification.
+        # `ConnectionError` may be due to too large `chunk_size`
+        # (see [#2572]) so try halving on error.
+        # Note: start with 40 * [default: 256K] = 10M.
+        # Note: must be multiple of 256K.
         #
-        # [1] https://github.com/iterative/dvc/issues/2572
+        # [#2572]: https://github.com/iterative/dvc/issues/2572
 
         # skipcq: PYL-W0212
-        multiplier = int(_DEFAULT_CHUNKSIZE / Blob._CHUNK_SIZE_MULTIPLE)
+        multiplier = 40
         while True:
             try:
                 # skipcq: PYL-W0212
-                chunk_size = Blob._CHUNK_SIZE_MULTIPLE * multiplier
+                chunk_size = _DEFAULT_CHUNKSIZE * multiplier
                 return func(*args, chunk_size=chunk_size, **kwargs)
             except requests.exceptions.ConnectionError:
-                multiplier = int(multiplier / 2)
+                multiplier //= 2
                 if not multiplier:
                     raise
 
     return wrapper
 
 
 @dynamic_chunk_size
-def _upload_to_bucket(bucket, from_file, to_info, **kwargs):
-    blob = bucket.blob(to_info.path, **kwargs)
-    blob.upload_from_filename(from_file)
+def _upload_to_bucket(bucket, from_file, to_info, chunk_size=None, **kwargs):
+    blob = bucket.blob(to_info.path, chunk_size=chunk_size, **kwargs)
+    with Tqdm(
+        desc=to_info.path, total=os.path.getsize(from_file), bytes=True
+    ) as pbar:
+        with io.open(from_file, mode="rb") as fd:
+            raw_read = fd.read
+
+            def read(size=chunk_size):
+                res = raw_read(size)
+                if res:
+                    pbar.update(len(res))
+                return res
+
+            fd.read = read
path: dvc/remote/gs.py
text: @user1:
I would wrap it instead of monkey-patching:
```python
with io.open(from_file, mode="rb") as raw_fd, \
     Tqdm.wrap(raw_fd, total=..., desc=...) as fd:
    blob.upload_from_file(fd)
```
This will allow us reusing this. We have the same in [GDrive PR](https://github.com/iterative/dvc/pull/2551).

@author:
yes I thought of wrapping but opted to patch like this instead since

1. It's easier to see what's happening/debug
2. It's shorter/requires less code to implement
3. I didn't think it would be reused.

There are a few minor issues I have with the GDrive PR but I'd rather both this PR and the GDrive one are merged as is and then I can unify both of them nicely using wrappers.

@author:
actually the wrapping of file-like objects is a nice feature request from https://github.com/tqdm/tqdm/issues/84 which dropped off my radar

**edit** FOUR years ago? Really? Ergh.

@author:
Feel free to give feedback at https://github.com/tqdm/tqdm/pull/844 (going to make this easier). 

@user2:
@author, so probably after https://github.com/tqdm/tqdm/pull/844 gets merged, we should create a ticket to wrapattrs here, right?

@user1:
The issue with this code is that it can't be reused without copy paste. And that is desirable.

See `TrackFileReadProgress` in [GDrive PR](https://github.com/iterative/dvc/pull/2551/files).

@author:
yes, of course, but wrapping things to make them reusable without knowing that they actually need to be reused is a waste. In this case it turns out there is indeed another place for use - but it's in a PR in tandem (#2551) so it's best to wait for both PRs to be merged and then fix later. Past that there's the upstream tqdm/tqdm#844 which will fix this.

@user1:
Wrapping things for abstraction is not waste, now a function does two things uploading/downloading and some low-level stuff, which should be abstracted at some point. So it's not only about reuse.

I agree that this can wait. So my resolution:
- create a ticket about wrapping this, 
- mention both GS and GDrive code there, 
- resolve this conversation

@author:
created #2829