support backup of portal to s3

Author: saimonation

cterasdk/core/query.py

@@ -4,6 +4,10 @@
 from ..common import Object
 
 
+def database(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.database(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

@@ -27,10 +27,7 @@ async def get_object(client, file_id, chunk):
     :returns: Object
     :rtype: bytes
     """
-    message = (
-        f"Downloading block #{chunk.number} "
-        f"(offset={chunk.offset}, length={chunk.length})"
-    )
+    message = f"Downloading block (offset={chunk.offset}, length={chunk.length})"
 
     if file_id:
         message += f" for file ID {file_id}"
@@ -61,10 +58,8 @@ async def get_object(client, file_id, chunk):
         "unknown": "Unknown error"
     }
 
-    message = (
-        f"Failed to download block #{chunk.number} "
-        f"(offset={chunk.offset}, length={chunk.length})"
-    )
+    message = f"Failed to download block (offset={chunk.offset}, length={chunk.length})"
+
     if file_id:
         message = message + f" for file ID {file_id}"
 
@@ -128,17 +123,14 @@ async def process_chunk(client, file_id, chunk, encryption_key, semaphore):
     :rtype: cterasdk.direct.types.Block
     """
     async def process(client, chunk, encryption_key):
-        message = (
-            f"Processing block #{chunk.number} "
-            f"(offset={chunk.offset}, length={chunk.length})"
-        )
+        message = f"Processing block (offset={chunk.offset}, length={chunk.length}) "
         if file_id:
             message = message + f" for file ID {file_id}"
         logger.debug(message)
         encrypted_object = await get_object(client, file_id, chunk)
         decrypted_object = await decrypt_object(file_id, encrypted_object, encryption_key, chunk)
         decompressed_object = await decompress_object(file_id, decrypted_object, chunk)
-        return Block(file_id, chunk.number, chunk.offset, decompressed_object, chunk.length)
+        return Block(file_id, chunk.offset, decompressed_object, chunk.length)
 
     if semaphore is not None:
         async with semaphore:
@@ -188,7 +180,7 @@ def decrypt_encryption_key(file_id, wrapped_key, secret_access_key):
 
 
 @execute_with_retries(retries=3, backoff=1, max_backoff=10)
-async def get_chunks(api, bearer, file_id):
+async def get_chunks(api, bearer, file_id, start=None, end=None):
     """
     Get Chunks.
 
@@ -200,7 +192,8 @@ async def get_chunks(api, bearer, file_id):
     """
     logger.debug('Listing blocks for file ID: %s', file_id)
     try:
-        response = await api.get(f'{file_id}', headers={'Authorization': bearer})
+        params = {k: v for k, v in [('rangeStart', start), ('rangeEnd', end)] if v is not None}
+        response = await api.get(f'{file_id}', params=params, headers={'Authorization': bearer})
         if not response.chunks:
             logger.error('Could not find blocks for file ID: %s.', file_id)
             raise BlocksNotFoundError(file_id)

cterasdk/direct/types.py

@@ -56,89 +56,177 @@ class CompressionLib:
 
 class Chunk(Object):
 
-    def __init__(self, number, offset, url, length):
+    def __init__(self, offset, url, length):
         """
         Initialize a Chunk.
 
-        :param int number: Chunk number.
         :param int offset: Chunk offset.
         :param str url: Signed URL.
         :param int length: Object length.
         """
         super().__init__(
-            number=number,
             offset=offset,
             url=url,
             length=length
         )
 
 
-class Metadata(Object):
+class MetadataPart(Object):
     """
-    CTERA Direct IO File Metadata
+    CTERA Direct I/O File Metadata Part
+
+    :ivar int start: Starting offset.
+    :ivar int end: Ending offset.
+    :ivar int length: Length of the range in bytes.
+    :ivar bool encrypted: Indicates whether the range data is encrypted.
+    :ivar str encryption_key: Encryption key used for the range data.
+    :ivar bool compressed: Indicates whether the range data is compressed.
+    :ivar str compression_alg: Compression algorithm used for the range data.
+    :ivar list[cterasdk.common.object.Object] chunks: List of Chunk Objects.
     """
-
-    def __init__(self, file_id, server_object):
+    def __init__(self, offset, end, encrypted, encryption_key, compressed, compression_alg, chunks):
         """
         Initialize a Direct IO metadata response object.
 
-        :param int file_id: File ID.
-        :param cterasdk.common.object.Object server_object: Response Object.
+        :param int offset: Starting offset.
+        :param int end: Ending offset.
+        :param bool encrypted: Indicates whether the file is encrypted.
+        :param str encryption_key: Encryption key used for the file.
+        :param bool compressed: Indicates whether the file is compressed.
+        :param str compression_alg: Compression algorithm used for the file.
+        :param list[cterasdk.common.object.Object] chunks: List of Chunk Objects.
         """
         super().__init__(
-            file_id=file_id,
-            encrypted=server_object.encrypt_info.data_encrypted,
-            compressed=server_object.compression_type != CompressionLib.Off,
-            chunks=Metadata._format_chunks(server_object.chunks)
+            start=offset,
+            end=end,
+            length=(end - offset) + 1,
+            encrypted=encrypted,
+            encryption_key=encryption_key,
+            compressed=compressed,
+            compression_alg=compression_alg
         )
-        self.encryption_key = server_object.encrypt_info.wrapped_key if self.encrypted else None
-        self.compression_library = server_object.compression_type if self.compressed else None
-        last_chunk = self.chunks[-1]
-        self.size = last_chunk.offset + last_chunk.length
+        for chunk in chunks:
+            self.chunks.append(Chunk(offset, chunk.url, chunk.len))
+            offset = offset + chunk.len
 
     @staticmethod
-    def _format_chunks(server_object):
+    def from_server_object(server_object):
+        compressed = server_object.compression_type != CompressionLib.Off
+        start, end = map(int, server_object.actual_blocks_range.range.split('-'))
+        return MetadataPart(
+            start,
+            end,
+            server_object.encrypt_info.data_encrypted,
+            server_object.encrypt_info.wrapped_key if server_object.encrypt_info.data_encrypted else None,
+            compressed,
+            server_object.compression_type if compressed else None
+        )
+
+    def __repr__(self):
+        return str(self)
+
+    def __str__(self):
+        return (
+            f"{self.__class__.__name__}("
+            f"{{'start': {self.start}, "
+            f"'end': {self.end}, "
+            f"'encrypted': {self.encrypted}, "
+            f"'compressed': {self.compressed}, "
+            f"'length': {self.length}, "
+            f"'chunks': {len(self.chunks)}}})"
+        )
+
+
+class Metadata(Object):
+    """
+    CTERA Direct I/O File Metadata
+
+    :ivar int file_id: File ID.
+    :ivar int file_size: File Size.
+    :ivar list[cterasdk.direct.types.MetadataPart] parts: List of Metadata Parts.
+    :ivar bool encrypted: Indicates whether the range data is encrypted.
+    :ivar str encryption_key: Encryption key used for the range data.
+    :ivar bool compressed: Indicates whether the range data is compressed.
+    :ivar str compression_alg: Compression algorithm used for the range data.
+    """
+    def __init__(self, file_id, *parts):
         """
-        Create Chunks.
+        Initialize a Direct IO metadata response object.
 
         :param int file_id: File ID.
-        :param cterasdk.common.object.Object server_object: Server response.
-        :param list[int] blocks: List of block numbers to retrieve.
-        :returns: Chunk objects
-        :rtype: list[cterasdk.direct.types.Chunk]
+        :param list[cterasdk.direct.types.MetadataPart] parts: List of Metadata Parts.
         """
-        offset = 0
-        chunks = []
-        for number, chunk in enumerate(server_object, 1):
-            chunks.append(Chunk(number, offset, chunk.url, chunk.len))
-            offset = offset + chunk.len
-        return chunks
+        super().__init__(file_id=file_id, parts=parts)
+
+    @property
+    def size(self):
+        return self.parts[0].actual_blocks_range.file_size
+
+    @property
+    def encrypted(self):
+        return self.parts[0].encrypted
+
+    @property
+    def encryption_key(self):
+        return self.parts[0].encryption_key
+
+    @property
+    def compressed(self):
+        return self.parts[0].compressed
+
+    @property
+    def compression_alg(self):
+        return self.parts[0].compression_alg
+
+    @property
+    def start(self):
+        return self.parts[0].start
+
+    @property
+    def end(self):
+        return self.parts[-1].end
 
     def serialize(self):
         """
-        Serialize Direct IO metadata to a dictionary.
+        Serialize Direct I/O metadata to a dictionary.
         """
-        x = copy.deepcopy(self)
-        if self.encrypted:
-            x.encryption_key = utils.utf8_decode(base64.b64encode(self.encryption_key))
-        return x
+        chunks = []
+        for part in self.parts:
+            chunks.extend(part.chunks)
+
+        return {
+            'file_id': self.file_id,
+            'size': self.size,
+            'encrypted': self.encrypted,
+            'encryption_key': utils.utf8_decode(base64.b64encode(self.encryption_key)) if self.encrypted else None,
+            'compressed': self.compressed,
+            'compression_alg': self.compression_alg,
+            'chunks': chunks
+        }
+
+    def __repr__(self):
+        return str(self)
+
+    def __str__(self):
+        d = self.serialize()
+        d.pop('encryption_key')
+        d['chunks'] = len(d['chunks'])
+        return d
 
 
 class Block:
     """Block"""
 
-    def __init__(self, file_id, number, offset, data, length):
+    def __init__(self, file_id, offset, data, length):
         """
         Initialize a Block.
 
         :param int file_id: File ID.
-        :param int number: Block number.
         :param int offset: Block offset.
         :param bytes data: Bytes
         :param int length: Block length.
         """
         self._file_id = file_id
-        self._number = number
         self._offset = offset
         self._data = data
         self._length = length
@@ -147,10 +235,6 @@ def __init__(self, file_id, number, offset, data, length):
     def file_id(self):
         return self._file_id
 
-    @property
-    def number(self):
-        return self._number
-
     @property
     def offset(self):
         return self._offset