Saimon/backup to s3 (#354)

Author: saimonation

cterasdk/core/query.py

@@ -4,6 +4,10 @@
 from ..common import Object
 
 
+def run(core, path, param):
+    return create_callback_function(core, path, callback_response=DefaultResponse)(param)
+
+
 def create_callback_function(core, path, name=None, *, callback_response=None):
     """
     Create a query callback function

cterasdk/core/servers.py

@@ -20,6 +20,7 @@ class Servers(BaseCommand):
     def __init__(self, portal):
         super().__init__(portal)
         self.tasks = Tasks(self._core)
+        self.backup = Backup(self._core)
 
     def _get_entire_object(self, server):
         ref = f'/servers/{server}'
@@ -43,6 +44,19 @@ def get(self, name, include=None):
             raise ObjectNotFoundException(f'/servers/{name}')
         return server
 
+    @property
+    def system_database(self):
+        """
+        Retrieve the main database object
+
+        :returns: Main database object.
+        :rtype: cterasdk.common.object.Object
+        """
+        response = query.run(self._core, '/servers', query.QueryParamBuilder().addFilter(
+            query.FilterBuilder('mainDB').eq(True)
+        ).build())
+        return response.objects[0]
+
     def list_servers(self, include=None):
         """
         Retrieve the servers that comprise CTERA Portal.
@@ -70,7 +84,6 @@ def modify(self, name, server_name=None, app=None, preview=None, enable_public_i
         :param bool,optional enable_replication: Enable or disable database replication
         :param str,optional replica_of: Configure as a replicate of another Portal server. `enable_replication` must be set to `True`
         """
-
         server = self._get_entire_object(name)
         if enable_replication is True and replica_of is not None:
             server.replicationSettings = Object()
@@ -101,6 +114,40 @@ def modify(self, name, server_name=None, app=None, preview=None, enable_public_i
             raise CTERAException(f'Server modification failed: {ref}') from error
 
 
+class Backup(BaseCommand):
+
+    def connected(self):
+        """
+        Verify connectivity to the backup S3 bucket
+        """
+        return self._core.servers.system_database.backupToBucket.status == 'Connected'
+
+    def enable(self, bucket, interval):
+        """
+        Enable Main Database Backup to an S3 Bucket
+
+        :param cterasdk.core.types.Bucket bucket: Storage bucket
+        :param int interval: Backup interval in minutes
+        """
+        database = self._core.servers.system_database
+        database.backupToBucket = Object(enabled=True, exportSchedulePeriod=interval, details=bucket.database_backup_server_object())
+        logger.info("Enabling database backup. %s", {'server': database.name})
+        response = self._core.api.put(f'/servers/{database.name}', database)
+        logger.info("Database backup enabled. %s", {'server': database.name})
+        return response
+
+    def disable(self):
+        """
+        Disable Main Database Backup
+        """
+        database = self._core.servers.system_database
+        database.backupToBucket.enabled = False
+        logger.info("Disabling database backup. %s", {'server': database.name})
+        response = self._core.api.put(f'/servers/{database.name}', database)
+        logger.info("Database backup disabled. %s", {'server': database.name})
+        return response
+
+
 class Tasks(BaseCommand):
 
     def background(self, name):

cterasdk/core/types.py

@@ -331,6 +331,19 @@ def __init__(self, bucket, driver, access_key, secret_key, endpoint, https, dire
     def trust_all_certificates(self):
         return not self.verify_ssl
 
+    def database_backup_server_object(self):
+        return Object(
+            storage=self.driver,
+            bucket=self.bucket,
+            accessKey=self.access_key,
+            secretKey=self.secret_key,
+            endPoint=self.endpoint,
+            useHttps=self.https,
+            trustAllCertificates=self.trust_all_certificates,
+            masterHost=None,
+            usePathStyleAddressing=False
+        )
+
 
 class AzureBlob(HTTPBucket):
 

cterasdk/direct/lib.py

@@ -1,12 +1,11 @@
 import logging
 import asyncio
-import urllib.parse
 
 from ..lib.retries import execute_with_retries
 from .types import Metadata, Block
 from .crypto import decrypt_key, decrypt_block
 from .decompressor import decompress
-from ..exceptions.transport import BadRequest, Unauthorized, Forbidden, Unprocessable, InternalServerError, HTTPError
+from ..exceptions.transport import BadRequest, Unauthorized, Unprocessable, InternalServerError, HTTPError
 from ..exceptions.direct import (
     AuthorizationError, BlockListConnectionError, BlockListTimeout, BlockValidationException, BlocksNotFoundError,
     DecompressBlockError, DecryptBlockError, DecryptKeyError, DirectIOError, DownloadConnectionError,
@@ -73,10 +72,6 @@ async def get_object(client, file_id, chunk):
     raise exception
 
 
-def is_azure_object_storage(chunk):
-    return urllib.parse.urlparse(chunk.url).netloc.endswith('core.windows.net')
-
-
 async def decrypt_object(file_id, encrypted_object, encryption_key, chunk):
     """
     Decrypt Encrypted Object.
@@ -88,7 +83,7 @@ async def decrypt_object(file_id, encrypted_object, encryption_key, chunk):
     :rtype: bytes
     """
     try:
-        return decrypt_block(encrypted_object[16:] if is_azure_object_storage(chunk) else encrypted_object, encryption_key)
+        return decrypt_block(encrypted_object, encryption_key)
     except DirectIOError:
         logger.error('Failed to decrypt block.')
         raise DecryptBlockError(file_id, chunk)
@@ -207,7 +202,7 @@ async def get_chunks(api, bearer, file_id):
         return Metadata(file_id, response)
     except BadRequest as error:
         raise ObjectNotFoundError(file_id) from error
-    except (Unauthorized, Forbidden) as error:
+    except Unauthorized as error:
         raise AuthorizationError(file_id) from error
     except Unprocessable as error:
         raise UnsupportedStorageError(file_id) from error

docs/source/UserGuides/Portal/Administration.rst

@@ -605,6 +605,39 @@ Server Tasks
    for task in admin.servers.tasks.scheduled('database'):
       print(task.name)
 
+
+Main Database Backup to S3
+--------------------------
+
+.. automethod:: cterasdk.core.servers.Backup.enable
+   :noindex:
+
+.. code-block:: python
+
+   name = 'target-s3-bucket-name'
+   access, secret = 'access-key', 'secret-access-key'
+   endpoint = 's3.eu-west-1.amazonaws.com'
+   https = True
+
+   bucket = core_types.AmazonS3(name, access, secret, endpoint, https)  # use verify_ssl=False to trust all certificates
+   user.servers.backup.enable(bucket, 60)  # backup every 60 minutes
+
+
+.. automethod:: cterasdk.core.servers.Backup.disable
+   :noindex:
+
+.. code-block:: python
+
+   user.servers.backup.disable()
+
+.. automethod:: cterasdk.core.servers.Backup.status
+   :noindex:
+
+.. code-block:: python
+
+   is_connected = user.servers.backup.connected()
+
+
 Messaging Service
 =================
 

tests/ut/core/admin/test_servers.py

@@ -4,6 +4,7 @@
 import munch
 from cterasdk.common import Object
 from cterasdk.core import servers
+from cterasdk.core.types import AmazonS3
 from cterasdk import exceptions
 from tests.ut.core.admin import base_admin
 
@@ -110,6 +111,72 @@ def test_modify_success(self):
         self._assert_equal_objects(actual_param, expected_param)
         self.assertEqual(ret, put_response)
 
+    def test_system_database(self):
+        with mock.patch("cterasdk.core.servers.query.run") as query_mock:
+            query_mock.return_value = munch.Munch({
+                'objects': [munch.Munch({'mainDB': True})]
+            })
+            server = servers.Servers(self._global_admin).system_database
+            expected_query_params = base_admin.BaseCoreTest._create_query_params(start_from=0, count_limit=50, filters=[
+                munch.Munch({
+                    'field': 'mainDB',
+                    'restriction': 'eq',
+                    '_classname': 'BooleanFilter',
+                    'value': True
+                })
+            ])
+            actual_query_params = query_mock.call_args[0][2]
+            self._assert_equal_objects(actual_query_params, expected_query_params)
+            self.assertEqual(server.mainDB, True)
+
+    def test_enable_server_backup(self):
+        self._init_global_admin()
+        server_name = 'server'
+        with mock.patch("cterasdk.core.servers.Servers.system_database", new_callable=mock.PropertyMock) as query_mock:
+            query_mock.return_value = munch.Munch({'name': server_name, 'backupToBucket': None})
+            bucket, access, secret, endpoint = 'bucket-name', 'access', 'secret', 'www.endpoint.com'
+            bucket = AmazonS3(bucket, access, secret, endpoint, True, verify_ssl=False)
+            servers.Servers(self._global_admin).backup.enable(bucket, 60)
+            self._global_admin.api.put.assert_called_once_with(f'/servers/{server_name}', mock.ANY)
+            actual_param = self._global_admin.api.put.call_args[0][1]
+            expected_param = munch.Munch({
+                'enabled': True,
+                'exportSchedulePeriod': 60,
+                'details': TestCoreServers._create_database_backup_server_object(bucket)
+            })
+            self._assert_equal_objects(actual_param.backupToBucket, expected_param)
+
+    @staticmethod
+    def _create_database_backup_server_object(bucket):
+        return munch.Munch({
+            'storage': bucket.driver,
+            'bucket': bucket.bucket,
+            'accessKey': bucket.access_key,
+            'secretKey': bucket.secret_key,
+            'endPoint': bucket.endpoint,
+            'useHttps': bucket.https,
+            'trustAllCertificates': bucket.trust_all_certificates,
+            'masterHost': None,
+            'usePathStyleAddressing': False
+        })
+
+    def test_disable_server_backup(self):
+        self._init_global_admin()
+        server_name = 'server'
+        with mock.patch("cterasdk.core.servers.Servers.system_database", new_callable=mock.PropertyMock) as query_mock:
+            query_mock.return_value = munch.Munch({'name': server_name, 'backupToBucket': munch.Munch({'enabled': True})})
+            servers.Servers(self._global_admin).backup.disable()
+            self._global_admin.api.put.assert_called_once_with(f'/servers/{server_name}', mock.ANY)
+            actual_param = self._global_admin.api.put.call_args[0][1]
+            self._assert_equal_objects(actual_param.backupToBucket.enabled, False)
+
+    def test_server_backup_status(self):
+        self._init_global_admin()
+        with mock.patch("cterasdk.core.servers.Servers.system_database", new_callable=mock.PropertyMock) as query_mock:
+            query_mock.return_value = munch.Munch({'backupToBucket': munch.Munch({'status': 'Connected'})})
+            ret = servers.Servers(self._global_admin).backup.connected()
+            self.assertEqual(ret, True)
+
     @staticmethod
     def _create_server_object(name=None, app=None, preview=None, enable_public_ip=None,
                               public_ip=None, allow_user_login=None, enable_replication=None, replica_of=None):