Merge pull request #333 from reef-technologies/replication-monitoring

Replication monitoring

Author: ppolewicz

CHANGELOG.md

@@ -6,6 +6,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+As in version 1.16.0, the replication API may still be unstable in this release,
+though no known major changes to it are planned at this point. Early adopters may find it
+feasible to rely this implementation already.
+
 ### Added
 * Add included_sources module, for keeping track of included modified third-party libraries
 * Add `include_existing_files` parameter to `ReplicationSetupHelper`
@@ -16,6 +20,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 * Change the per part retry limit from 5 to 20 for data transfer operations. Please note that the retry system is not considered to be a part of the public interface and is subject to be adjusted
 * Do not wait more than 64 seconds between retry attempts (unless server asks for it)
 * On longer failures wait an additional (random, up to 1s) amount of time to prevent client synchronization
+* Flatten `ReplicationConfiguration` interface
+* Reorder actions of `ReplicationSetupHelper` to avoid zombie rules
 
 ### Fixed
 * Fix: downloading compressed files and decompressing them on the fly now does not cause a TruncatedOutput error
@@ -26,6 +32,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Infrastructure
 * Add 3.11.0-beta.1 to CI
 * Change Sphinx major version from 5 to 6
+* Extract folder/bucket scanning into a new `scan` module
 * Enable pip cache in CI
 
 ## [1.16.0] - 2022-04-27
@@ -109,7 +116,7 @@ Expect substantial amount of work on sdk interface:
 
 ### Changed
 * The `importlib-metadata` requirement is less strictly bound now (just >=3.3.0 for python > 3.5).
-* `B2Api` `update_file_legal_hold` and `update_file_retention_setting` now return the set values 
+* `B2Api` `update_file_legal_hold` and `update_file_retention_setting` now return the set values
 
 ### Added
 * `BucketIdNotFound` thrown based on B2 cloud response
@@ -125,7 +132,7 @@ Expect substantial amount of work on sdk interface:
 ## [1.11.0] - 2021-06-24
 
 ### Changed
-* apiver `v2` interface released. `from b2sdk.v2 import ...` is now the recommended import, 
+* apiver `v2` interface released. `from b2sdk.v2 import ...` is now the recommended import,
   but `from b2sdk.v1 import ...` works as before
 
 ## [1.10.0] - 2021-06-23
@@ -151,7 +158,7 @@ Expect substantial amount of work on sdk interface:
 * Old buckets (from past tests) are cleaned up before running integration tests in a single thread
 
 ### Removed
-* Remove deprecated `SyncReport` methods 
+* Remove deprecated `SyncReport` methods
 
 ## [1.9.0] - 2021-06-07
 
@@ -166,7 +173,7 @@ Expect substantial amount of work on sdk interface:
 * `B2Api` unittests for v0, v1 and v2 are now common
 * `B2Api.cancel_large_file` returns a `FileIdAndName` object instead of a `FileVersion` object in v2
 * `FileVersion` has a mandatory `api` parameter in v2
-* `B2Folder` holds a handle to B2Api 
+* `B2Folder` holds a handle to B2Api
 * `Bucket` unit tests for v1 and v2 are now common
 
 ### Fixed
@@ -195,7 +202,7 @@ Expect substantial amount of work on sdk interface:
 * Encryption settings, types and providers are now part of the public API
 
 ### Removed
-* Remove `Bucket.copy_file` and `Bucket.start_large_file` 
+* Remove `Bucket.copy_file` and `Bucket.start_large_file`
 * Remove `FileVersionInfo.format_ls_entry` and `FileVersionInfo.format_folder_ls_entry`
 
 ## [1.7.0] - 2021-04-22

b2sdk/_v3/__init__.py

@@ -164,14 +164,7 @@
 from b2sdk.sync.action import B2HideAction
 from b2sdk.sync.action import B2UploadAction
 from b2sdk.sync.action import LocalDeleteAction
-from b2sdk.sync.exception import EnvironmentEncodingError
 from b2sdk.sync.exception import IncompleteSync
-from b2sdk.sync.exception import InvalidArgument
-from b2sdk.sync.folder import AbstractFolder
-from b2sdk.sync.folder import B2Folder
-from b2sdk.sync.folder import LocalFolder
-from b2sdk.sync.folder_parser import parse_sync_folder
-from b2sdk.sync.path import AbstractSyncPath, B2SyncPath, LocalSyncPath
 from b2sdk.sync.policy import AbstractFileSyncPolicy
 from b2sdk.sync.policy import CompareVersionMode
 from b2sdk.sync.policy import NewerFileSyncMode
@@ -189,27 +182,43 @@
 from b2sdk.sync.policy_manager import POLICY_MANAGER
 from b2sdk.sync.report import SyncFileReporter
 from b2sdk.sync.report import SyncReport
-from b2sdk.sync.scan_policies import DEFAULT_SCAN_MANAGER
-from b2sdk.sync.scan_policies import IntegerRange
-from b2sdk.sync.scan_policies import RegexSet
-from b2sdk.sync.scan_policies import ScanPoliciesManager
-from b2sdk.sync.scan_policies import convert_dir_regex_to_dir_prefix_regex
 from b2sdk.sync.sync import KeepOrDeleteMode
 from b2sdk.sync.sync import Synchronizer
-from b2sdk.sync.sync import zip_folders
 from b2sdk.sync.encryption_provider import AbstractSyncEncryptionSettingsProvider
 from b2sdk.sync.encryption_provider import BasicSyncEncryptionSettingsProvider
 from b2sdk.sync.encryption_provider import ServerDefaultSyncEncryptionSettingsProvider
 from b2sdk.sync.encryption_provider import SERVER_DEFAULT_SYNC_ENCRYPTION_SETTINGS_PROVIDER
 
+# scan
+
+from b2sdk.scan.exception import EnvironmentEncodingError
+from b2sdk.scan.exception import InvalidArgument
+from b2sdk.scan.folder import AbstractFolder
+from b2sdk.scan.folder import B2Folder
+from b2sdk.scan.folder import LocalFolder
+from b2sdk.scan.folder_parser import parse_folder
+from b2sdk.scan.path import AbstractPath, B2Path, LocalPath
+from b2sdk.scan.policies import convert_dir_regex_to_dir_prefix_regex
+from b2sdk.scan.policies import DEFAULT_SCAN_MANAGER
+from b2sdk.scan.policies import IntegerRange
+from b2sdk.scan.policies import RegexSet
+from b2sdk.scan.policies import ScanPoliciesManager
+from b2sdk.scan.report import ProgressReport
+from b2sdk.scan.scan import zip_folders
+from b2sdk.scan.scan import AbstractScanResult
+from b2sdk.scan.scan import AbstractScanReport
+from b2sdk.scan.scan import CountAndSampleScanReport
+
 # replication
 
 from b2sdk.replication.setting import ReplicationConfigurationFactory
 from b2sdk.replication.setting import ReplicationConfiguration
-from b2sdk.replication.setting import ReplicationSourceConfiguration
 from b2sdk.replication.setting import ReplicationRule
-from b2sdk.replication.setting import ReplicationDestinationConfiguration
+from b2sdk.replication.types import ReplicationStatus
 from b2sdk.replication.setup import ReplicationSetupHelper
+from b2sdk.replication.monitoring import ReplicationScanResult
+from b2sdk.replication.monitoring import ReplicationReport
+from b2sdk.replication.monitoring import ReplicationMonitor
 
 # other
 

b2sdk/_v3/exception.py

@@ -71,14 +71,14 @@
 from b2sdk.exception import SSECKeyError
 from b2sdk.exception import WrongEncryptionModeForBucketDefault
 from b2sdk.exception import interpret_b2_error
-from b2sdk.sync.exception import EmptyDirectory
-from b2sdk.sync.exception import EnvironmentEncodingError
 from b2sdk.sync.exception import IncompleteSync
-from b2sdk.sync.exception import InvalidArgument
-from b2sdk.sync.exception import NotADirectory
-from b2sdk.sync.exception import UnableToCreateDirectory
-from b2sdk.sync.exception import UnSyncableFilename
-from b2sdk.sync.exception import check_invalid_argument
+from b2sdk.scan.exception import UnableToCreateDirectory
+from b2sdk.scan.exception import EmptyDirectory
+from b2sdk.scan.exception import EnvironmentEncodingError
+from b2sdk.scan.exception import InvalidArgument
+from b2sdk.scan.exception import NotADirectory
+from b2sdk.scan.exception import UnsupportedFilename
+from b2sdk.scan.exception import check_invalid_argument
 
 __all__ = (
     'AccessDenied',
@@ -144,7 +144,7 @@
     'UnknownHost',
     'UnrecognizedBucketType',
     'UnableToCreateDirectory',
-    'UnSyncableFilename',
+    'UnsupportedFilename',
     'UnsatisfiableRange',
     'UnusableFileName',
     'interpret_b2_error',

b2sdk/file_version.py

@@ -199,6 +199,10 @@ class FileVersion(BaseFileVersion):
         'action',
     ]
 
+    # defined at https://www.backblaze.com/b2/docs/files.html#httpHeaderSizeLimit
+    DEFAULT_HEADERS_LIMIT = 7000
+    ADVANCED_HEADERS_LIMIT = 2048
+
     def __init__(
         self,
         api: 'B2Api',
@@ -282,6 +286,41 @@ def download(
             encryption=encryption,
         )
 
+    def _get_upload_headers(self) -> bytes:
+        """
+        Return encoded http headers, as when sending an upload request to b2 http api.
+        WARNING: the headers do not contain newlines between headers and spaces between
+        key and value. This implementation is in par with ADVANCED_HEADERS_LIMIT
+        and is reasonable only for `has_large_header` method
+        """
+        headers = self.api.raw_api.get_upload_file_headers(
+            upload_auth_token=self.api.account_info.get_account_auth_token(),
+            file_name=self.file_name,
+            content_length=self.size,
+            content_type=self.content_type,
+            content_sha1=self.content_sha1,
+            file_infos=self.file_info,
+            server_side_encryption=self.server_side_encryption,
+            file_retention=self.file_retention,
+            legal_hold=self.legal_hold,
+        )
+
+        headers_str = ''.join(
+            f'{key}{value}' for key, value in headers.items() if value is not None
+        )
+        return headers_str.encode('utf8')
+
+    @property
+    def has_large_header(self) -> bool:
+        """
+        Determine whether FileVersion's info fits header size limit defined by B2.
+        This function makes sense only for "advanced" buckets, i.e. those which
+        have Server-Side Encryption or File Lock enabled.
+
+        See https://www.backblaze.com/b2/docs/files.html#httpHeaderSizeLimit.
+        """
+        return len(self._get_upload_headers()) > self.ADVANCED_HEADERS_LIMIT
+
 
 class DownloadVersion(BaseFileVersion):
     """

b2sdk/raw_api.py

@@ -303,6 +303,42 @@ def update_file_retention(
     ):
         pass
 
+    @classmethod
+    def get_upload_file_headers(
+        cls,
+        upload_auth_token: str,
+        file_name: str,
+        content_length: int,
+        content_type: str,
+        content_sha1: str,
+        file_infos: dict,
+        server_side_encryption: Optional[EncryptionSetting],
+        file_retention: Optional[FileRetentionSetting],
+        legal_hold: Optional[LegalHold],
+    ) -> dict:
+        headers = {
+            'Authorization': upload_auth_token,
+            'Content-Length': str(content_length),
+            'X-Bz-File-Name': b2_url_encode(file_name),
+            'Content-Type': content_type,
+            'X-Bz-Content-Sha1': content_sha1,
+        }
+        for k, v in file_infos.items():
+            headers[FILE_INFO_HEADER_PREFIX + k] = b2_url_encode(v)
+        if server_side_encryption is not None:
+            assert server_side_encryption.mode in (
+                EncryptionMode.NONE, EncryptionMode.SSE_B2, EncryptionMode.SSE_C
+            )
+            server_side_encryption.add_to_upload_headers(headers)
+
+        if legal_hold is not None:
+            legal_hold.add_to_upload_headers(headers)
+
+        if file_retention is not None:
+            file_retention.add_to_to_upload_headers(headers)
+
+        return headers
+
     @abstractmethod
     def upload_file(
         self,
@@ -857,27 +893,17 @@ def upload_file(
         """
         # Raise UnusableFileName if the file_name doesn't meet the rules.
         self.check_b2_filename(file_name)
-        headers = {
-            'Authorization': upload_auth_token,
-            'Content-Length': str(content_length),
-            'X-Bz-File-Name': b2_url_encode(file_name),
-            'Content-Type': content_type,
-            'X-Bz-Content-Sha1': content_sha1,
-        }
-        for k, v in file_infos.items():
-            headers[FILE_INFO_HEADER_PREFIX + k] = b2_url_encode(v)
-        if server_side_encryption is not None:
-            assert server_side_encryption.mode in (
-                EncryptionMode.NONE, EncryptionMode.SSE_B2, EncryptionMode.SSE_C
-            )
-            server_side_encryption.add_to_upload_headers(headers)
-
-        if legal_hold is not None:
-            legal_hold.add_to_upload_headers(headers)
-
-        if file_retention is not None:
-            file_retention.add_to_to_upload_headers(headers)
-
+        headers = self.get_upload_file_headers(
+            upload_auth_token=upload_auth_token,
+            file_name=file_name,
+            content_length=content_length,
+            content_type=content_type,
+            content_sha1=content_sha1,
+            file_infos=file_infos,
+            server_side_encryption=server_side_encryption,
+            file_retention=file_retention,
+            legal_hold=legal_hold,
+        )
         return self.b2_http.post_content_return_json(upload_url, headers, data_stream)
 
     def upload_part(