Limiting network traffic rate for uploading stage. #48

MikhailBurdukov · 2023-08-22T13:31:30Z

No description provided.

aalexfvk · 2023-08-22T16:30:55Z

ch_backup/config.py

@@ -101,6 +101,11 @@ def _as_seconds(t: str) -> int:
        "uploading_threads": 4,
        # The maximum number of objects the stage's input queue can hold simultaneously, `0`is unbounded
        "queue_size": 10,
+        # Upper bound of network loading per second on the uploading stage. (bytes/sec)


It is better to move this settings in a sub-dict under a key "rate_limiter": {...}

Let's use human readable notation like 1Mib. Check humanfriendly.parse_size()

How about renaming uploading_traffic_limit_retry_time to retry_interval and uploading_traffic_limit to max_upload_rate

aalexfvk · 2023-08-22T16:33:22Z

ch_backup/storage/async_pipeline/pipeline_builder.py

@@ -134,7 +135,8 @@ def build_uploading_stage(
        buffer_size = stage_config["buffer_size"]
        chunk_size = stage_config["chunk_size"]
        queue_size = stage_config["queue_size"]
-
+        traffic_limit_per_sec = stage_config["uploading_traffic_limit"]


Also suggest renaming in according to key names

aalexfvk · 2023-08-22T16:54:15Z

ch_backup/storage/async_pipeline/pipeline_builder.py

@@ -152,6 +154,12 @@ def build_uploading_stage(
                ),
                maxsize=queue_size,
            ),
+            thread_map(


I think it must be before ChunkingStage which is set first for now.

ch-backup/ch_backup/storage/async_pipeline/pipeline_builder.py

Lines 174 to 177 in aa1e6ac

self.append(

thread_flat_map(ChunkingStage(chunk_size, buffer_size), maxsize=queue_size),

*stages

)

aalexfvk · 2023-08-22T16:56:17Z

ch_backup/storage/async_pipeline/stages/storage/multipart_storage_uploading_stage.py

@@ -83,6 +85,36 @@ def __call__(self, part: UploadingPart, index: int) -> UploadingPart:
        return part


+class TrafficLimitingStage(Handler):


Lets rename to RateLimiterStage. I think we can limit not only traffic by means of it.

aalexfvk · 2023-08-22T16:58:47Z

ch_backup/storage/async_pipeline/stages/storage/multipart_storage_uploading_stage.py

+        get_time_func: Callable = time.time,
+        sleep_func: Callable = time.sleep,


I think we don't need this parameters here. But only in RateLimiter class.

aalexfvk · 2023-08-22T17:03:42Z

ch_backup/storage/async_pipeline/stages/storage/multipart_storage_uploading_stage.py

+        self._sleep_func = sleep_func
+
+    def __call__(self, part: UploadingPart, index: int) -> UploadingPart:
+        while not self._rate_limiter.grant(len(part.data)):


I would propose to return here count of bytes(tokens) that can pass through without exceeding the limit. And pass these bytes further. Pay attention that it will be just bytes, not UploadingPart which is created only by StartMultipartUploadStage

aalexfvk · 2023-08-22T17:09:40Z

ch_backup/util.py

+        if self._limit_per_sec == 0:
+            return True
+        current_time = self._get_time_func()
+        lapse = current_time - self._bucket_last_update


Lets add comments to this code. Like replenish bucket and so on. Or move it to a separate function self._replenish_bucket()

aalexfvk · 2023-08-23T06:57:14Z

ch_backup/util.py

+        self._bucket_tokens = self._limit_per_sec
+        self._bucket_last_update = self._get_time_func()
+
+    def grant(self, tokens=1):


Let's add function that extracts all available tokens. We can use it in RateLimiterStage instead of grant(). But let grant() stay because it can be useful.

aalexfvk · 2023-08-23T07:58:42Z

ch_backup/util.py

+        )
+        if self._bucket_tokens >= tokens:
+            self._bucket_tokens -= tokens
+            self._bucket_last_update = current_time


I think last_update must be updated after every replenish

aalexfvk · 2023-08-23T08:01:53Z

ch_backup/util.py

@@ -371,6 +371,38 @@ def dataclass_from_dict(type_: Type[T], data: dict) -> T:
    return type_(**{k: v for k, v in data.items() if k in class_fields})  # type: ignore[call-arg]


+class RateLimiter:


Let's move this class into a separate import module from util.py

aalexfvk · 2023-08-23T08:12:30Z

tests/integration/features/traffic_limiting.feature

+    And clickhouse01 has test clickhouse data test1
+    And we have executed queries on clickhouse01
+    """
+    CREATE USER test_user IDENTIFIED WITH plaintext_password BY 'password';


Let's test only data table without ACl entites. In my opinion they are complicating test. SImplify as far as possible

aalexfvk · 2023-08-23T08:15:01Z

tests/integration/features/traffic_limiting.feature

+   Given ch-backup configuration on clickhouse01
+    """
+    storage:
+        uploading_traffic_limit_retry_time: 33554432


We can use Examples: clause to parametrize these both tests to one. Have a look at deduplication.feature for example.

aalexfvk · 2023-08-23T08:33:15Z

tests/unit/test_rate_limiter.py

+    chunk_size=st.just(16),
+    buffer_size=st.just(128),
+)
+def test_traffic_rate_limiting(content_size, chunk_size, buffer_size):


Let's write comprehensive unit tests only for RateLimiter class.
Testing on pipeline level looks cumbersome and overloaded with irrilevant details.

MikhailBurdukov · 2023-08-23T15:36:29Z

@aalexfvk, Fixed all requested changes in the last commit.

MikhailBurdukov added 2 commits August 22, 2023 13:30

Limiting network traffic rate for uploading stage.

d4525a9

Resolve merge conflict

aa1e6ac

MikhailBurdukov requested review from Alex-Burmak and aalexfvk August 22, 2023 15:26

aalexfvk requested changes Aug 23, 2023

View reviewed changes

aalexfvk reviewed Aug 23, 2023

View reviewed changes

Review changes

b447b66

MikhailBurdukov added 3 commits August 24, 2023 11:28

Review changes

48bac15

Extended knob descr

44dfbdd

Fix codespell

b18a71b

aalexfvk approved these changes Aug 29, 2023

View reviewed changes

MikhailBurdukov merged commit 00ff3e2 into main Aug 29, 2023
12 checks passed

MikhailBurdukov deleted the MDB-23714_uploading_traffic_rate branch August 29, 2023 12:00

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Limiting network traffic rate for uploading stage. #48

Limiting network traffic rate for uploading stage. #48

MikhailBurdukov commented Aug 22, 2023

aalexfvk Aug 22, 2023

aalexfvk Aug 22, 2023

aalexfvk Aug 22, 2023

aalexfvk Aug 22, 2023

aalexfvk Aug 22, 2023

aalexfvk Aug 22, 2023

aalexfvk Aug 22, 2023 •

edited

Loading

aalexfvk Aug 23, 2023

aalexfvk Aug 23, 2023

aalexfvk Aug 23, 2023

aalexfvk Aug 23, 2023

aalexfvk Aug 23, 2023

aalexfvk Aug 23, 2023

MikhailBurdukov commented Aug 23, 2023

	self.append(
	thread_flat_map(ChunkingStage(chunk_size, buffer_size), maxsize=queue_size),
	*stages
	)

		@@ -83,6 +85,36 @@ def __call__(self, part: UploadingPart, index: int) -> UploadingPart:
		return part


		class TrafficLimitingStage(Handler):

		get_time_func: Callable = time.time,
		sleep_func: Callable = time.sleep,

		@@ -371,6 +371,38 @@ def dataclass_from_dict(type_: Type[T], data: dict) -> T:
		return type_(**{k: v for k, v in data.items() if k in class_fields}) # type: ignore[call-arg]


		class RateLimiter:

Limiting network traffic rate for uploading stage. #48

Limiting network traffic rate for uploading stage. #48

Conversation

MikhailBurdukov commented Aug 22, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aalexfvk Aug 22, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

MikhailBurdukov commented Aug 23, 2023

aalexfvk Aug 22, 2023 •

edited

Loading