17465: Adds KeepClient storage classes support to Collection class.

Arvados-DCO-1.1-Signed-off-by: Lucas Di Pentima <lucas.dipentima@curii.com>

diff --git a/sdk/python/arvados/arvfile.py b/sdk/python/arvados/arvfile.py
index 6893b94bf78b7b16a1da1a802fdabe419563eb0d..7c6b732d36fcefc5a88896382ed99b61b83f9d4d 100644 (file)
--- a/sdk/python/arvados/arvfile.py
+++ b/sdk/python/arvados/arvfile.py
@@ -481,7 +481,7 @@ class _BlockManager(object):
     DEFAULT_PUT_THREADS = 2
     DEFAULT_GET_THREADS = 2
 
-    def __init__(self, keep, copies=None, put_threads=None, num_retries=None):
+    def __init__(self, keep, copies=None, put_threads=None, num_retries=None, storage_classes=[]):
         """keep: KeepClient object to use"""
         self._keep = keep
         self._bufferblocks = collections.OrderedDict()
@@ -497,6 +497,7 @@ class _BlockManager(object):
             self.num_put_threads = _BlockManager.DEFAULT_PUT_THREADS
         self.num_get_threads = _BlockManager.DEFAULT_GET_THREADS
         self.copies = copies
+        self.storage_classes = storage_classes
         self._pending_write_size = 0
         self.threads_lock = threading.Lock()
         self.padding_block = None
@@ -555,9 +556,9 @@ class _BlockManager(object):
                     return
 
                 if self.copies is None:
-                    loc = self._keep.put(bufferblock.buffer_view[0:bufferblock.write_pointer].tobytes(), num_retries=self.num_retries)
+                    loc = self._keep.put(bufferblock.buffer_view[0:bufferblock.write_pointer].tobytes(), num_retries=self.num_retries, classes=self.storage_classes)
                 else:
-                    loc = self._keep.put(bufferblock.buffer_view[0:bufferblock.write_pointer].tobytes(), num_retries=self.num_retries, copies=self.copies)
+                    loc = self._keep.put(bufferblock.buffer_view[0:bufferblock.write_pointer].tobytes(), num_retries=self.num_retries, copies=self.copies, classes=self.storage_classes)
                 bufferblock.set_state(_BufferBlock.COMMITTED, loc)
             except Exception as e:
                 bufferblock.set_state(_BufferBlock.ERROR, e)
@@ -726,9 +727,9 @@ class _BlockManager(object):
         if sync:
             try:
                 if self.copies is None:
-                    loc = self._keep.put(block.buffer_view[0:block.write_pointer].tobytes(), num_retries=self.num_retries)
+                    loc = self._keep.put(block.buffer_view[0:block.write_pointer].tobytes(), num_retries=self.num_retries, classes=self.storage_classes)
                 else:
-                    loc = self._keep.put(block.buffer_view[0:block.write_pointer].tobytes(), num_retries=self.num_retries, copies=self.copies)
+                    loc = self._keep.put(block.buffer_view[0:block.write_pointer].tobytes(), num_retries=self.num_retries, copies=self.copies, classes=self.storage_classes)
                 block.set_state(_BufferBlock.COMMITTED, loc)
             except Exception as e:
                 block.set_state(_BufferBlock.ERROR, e)

diff --git a/sdk/python/arvados/collection.py b/sdk/python/arvados/collection.py
index 26902931582244142054d69bc2e52fe596927de3..846de0115e0866dc2499a429aa0dda1e0c41918e 100644 (file)
--- a/sdk/python/arvados/collection.py
+++ b/sdk/python/arvados/collection.py
@@ -1261,6 +1261,7 @@ class Collection(RichCollectionBase):
                  apiconfig=None,
                  block_manager=None,
                  replication_desired=None,
+                 storage_classes_desired=None,
                  put_threads=None):
         """Collection constructor.
 
@@ -1293,12 +1294,18 @@ class Collection(RichCollectionBase):
           configuration applies. If not None, this value will also be used
           for determining the number of block copies being written.
 
+        :storage_classes_desired:
+          A list of storage class names where to upload the data. If None,
+          the keepstores are expected to store the data into their default
+          storage class.
+
         """
         super(Collection, self).__init__(parent)
         self._api_client = api_client
         self._keep_client = keep_client
         self._block_manager = block_manager
         self.replication_desired = replication_desired
+        self.storage_classes_desired = storage_classes_desired
         self.put_threads = put_threads
 
         if apiconfig:
@@ -1410,7 +1417,8 @@ class Collection(RichCollectionBase):
             copies = (self.replication_desired or
                       self._my_api()._rootDesc.get('defaultCollectionReplication',
                                                    2))
-            self._block_manager = _BlockManager(self._my_keep(), copies=copies, put_threads=self.put_threads, num_retries=self.num_retries)
+            classes = self.storage_classes_desired or []
+            self._block_manager = _BlockManager(self._my_keep(), copies=copies, put_threads=self.put_threads, num_retries=self.num_retries, storage_classes=classes)
         return self._block_manager
 
     def _remember_api_response(self, response):
@@ -1573,8 +1581,14 @@ class Collection(RichCollectionBase):
         body={}
         if properties:
             body["properties"] = properties
-        if storage_classes:
-            body["storage_classes_desired"] = storage_classes
+        desired_classes = storage_classes
+        # Instance level storage_classes takes precedence over argument.
+        if self.storage_classes_desired:
+            if desired_classes and self.storage_classes_desired != desired_classes:
+                _logger.warning("Storage classes already set to {}".format(self.storage_classes_desired))
+            desired_classes = self.storage_classes_desired
+        if desired_classes:
+            body["storage_classes_desired"] = desired_classes
         if trash_at:
             t = trash_at.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
             body["trash_at"] = t
@@ -1692,8 +1706,14 @@ class Collection(RichCollectionBase):
                 body["owner_uuid"] = owner_uuid
             if properties:
                 body["properties"] = properties
-            if storage_classes:
-                body["storage_classes_desired"] = storage_classes
+            desired_classes = storage_classes
+            # Instance level storage_classes takes precedence over argument.
+            if self.storage_classes_desired:
+                if desired_classes and self.storage_classes_desired != desired_classes:
+                    _logger.warning("Storage classes already set to {}".format(self.storage_classes_desired))
+                desired_classes = self.storage_classes_desired
+            if desired_classes:
+                body["storage_classes_desired"] = desired_classes
             if trash_at:
                 t = trash_at.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
                 body["trash_at"] = t

diff --git a/sdk/python/arvados/keep.py b/sdk/python/arvados/keep.py
index e3ef642c961184c7380a90196a19799c81ee48a9..2f20132aecdf17ee9b8d6da2d043037508e989ce 100644 (file)
--- a/sdk/python/arvados/keep.py
+++ b/sdk/python/arvados/keep.py
@@ -1272,7 +1272,7 @@ class KeepClient(object):
                 "failed to write {} after {} (wanted {} copies but wrote {})".format(
                     data_hash, loop.attempts_str(), (copies, classes), writer_pool.done()), service_errors, label="service")
 
-    def local_store_put(self, data, copies=1, num_retries=None):
+    def local_store_put(self, data, copies=1, num_retries=None, classes=[]):
         """A stub for put().
 
         This method is used in place of the real put() method when

diff --git a/sdk/python/tests/test_arvfile.py b/sdk/python/tests/test_arvfile.py
index 086fa542a2b28b2112f92eb49e21934247cf2710..0b8e7c8f8bf2d4615209e0dbdbfd81e6e54b32af 100644 (file)
--- a/sdk/python/tests/test_arvfile.py
+++ b/sdk/python/tests/test_arvfile.py
@@ -7,10 +7,7 @@ from builtins import hex
 from builtins import str
 from builtins import range
 from builtins import object
-import bz2
 import datetime
-import gzip
-import io
 import mock
 import os
 import unittest
@@ -19,7 +16,7 @@ import time
 import arvados
 from arvados._ranges import Range
 from arvados.keep import KeepLocator
-from arvados.collection import Collection, CollectionReader
+from arvados.collection import Collection
 from arvados.arvfile import ArvadosFile, ArvadosFileReader
 
 from . import arvados_testutil as tutil
@@ -36,7 +33,7 @@ class ArvadosFileWriterTestCase(unittest.TestCase):
         def get_from_cache(self, locator):
             self.requests.append(locator)
             return self.blocks.get(locator)
-        def put(self, data, num_retries=None, copies=None):
+        def put(self, data, num_retries=None, copies=None, classes=[]):
             pdh = tutil.str_keep_locator(data)
             self.blocks[pdh] = bytes(data)
             return pdh
@@ -173,8 +170,6 @@ class ArvadosFileWriterTestCase(unittest.TestCase):
             self.assertEqual("zzzzz-4zz18-mockcollection0", c.manifest_locator())
             self.assertFalse(c.modified())
 
-
-
     def test_write_to_end(self):
         keep = ArvadosFileWriterTestCase.MockKeep({
             "781e5e245d69b566979b86e28d23f2c7+10": b"0123456789",
@@ -366,7 +361,6 @@ class ArvadosFileWriterTestCase(unittest.TestCase):
             self.assertEqual("zzzzz-4zz18-mockcollection0", c.manifest_locator())
             self.assertFalse(c.modified())
 
-
     def test_large_write(self):
         keep = ArvadosFileWriterTestCase.MockKeep({})
         api = ArvadosFileWriterTestCase.MockApi({}, {})
@@ -400,7 +394,6 @@ class ArvadosFileWriterTestCase(unittest.TestCase):
 
             self.assertEqual(c.manifest_text(), ". 7f614da9329cd3aebf59b91aadc30bf0+67108864 781e5e245d69b566979b86e28d23f2c7+10 0:2:count.txt 67108864:10:count.txt\n")
 
-
     def test_sparse_write2(self):
         keep = ArvadosFileWriterTestCase.MockKeep({})
         api = ArvadosFileWriterTestCase.MockApi({}, {})
@@ -417,7 +410,6 @@ class ArvadosFileWriterTestCase(unittest.TestCase):
 
             self.assertEqual(c.manifest_text(), ". 7f614da9329cd3aebf59b91aadc30bf0+67108864 781e5e245d69b566979b86e28d23f2c7+10 0:67108864:count.txt 0:67108864:count.txt 0:2:count.txt 67108864:10:count.txt\n")
 
-
     def test_sparse_write3(self):
         keep = ArvadosFileWriterTestCase.MockKeep({})
         api = ArvadosFileWriterTestCase.MockApi({}, {})
@@ -448,7 +440,6 @@ class ArvadosFileWriterTestCase(unittest.TestCase):
                 writer.seek(0)
                 self.assertEqual(writer.read(), b"000000000011111111112222222222\x00\x00\x00\x00\x00\x00\x00\x00\x00\x004444444444")
 
-
     def test_rewrite_on_empty_file(self):
         keep = ArvadosFileWriterTestCase.MockKeep({})
         with Collection('. ' + arvados.config.EMPTY_BLOCK_LOCATOR + ' 0:0:count.txt',
@@ -876,7 +867,6 @@ class BlockManagerTest(unittest.TestCase):
             blockmanager.commit_bufferblock(bufferblock, True)
             self.assertEqual(bufferblock.state(), arvados.arvfile._BufferBlock.COMMITTED)
 
-
     def test_bufferblock_commit_with_error(self):
         mockkeep = mock.MagicMock()
         mockkeep.put.side_effect = arvados.errors.KeepWriteError("fail")

diff --git a/sdk/python/tests/test_collections.py b/sdk/python/tests/test_collections.py
index 66f062c167250b2873ea6a047554b5986f17ee46..e1c6fc2bbca5aeb76ecc438c7b07739040a843ad 100644 (file)
--- a/sdk/python/tests/test_collections.py
+++ b/sdk/python/tests/test_collections.py
@@ -9,11 +9,9 @@ import arvados
 import copy
 import mock
 import os
-import pprint
 import random
 import re
 import sys
-import tempfile
 import datetime
 import ciso8601
 import time