Merge pull request #701 from cderici/storage

#701

#### Description

This adds the implementation of storage. In particular, it introduces the following functions:

`unit.add_storage()`
`unit.attach_storage()`
`model.create_storage_pool()`
`unit.detach_storage()`
`model.list_storage()`
`model.list_storage_pools()`
`model.remove_storage()`
`model.remove_storage_pool()`
`model.show_storage()`
`model.update_storage_pool()`

Also adds the `--attach-storage` parameter to the `application.add_unit`.

Fixes #694 and #695 

#### QA Steps

A couple of new tests are added to test the main functionality of storages and pools. In particular, all the following should pass:

```
tox -e integration -- tests/integration/test_model.py::test_add_storage
```

```
tox -e integration -- tests/integration/test_model.py::test_list_storage
```

```
tox -e integration -- tests/integration/test_model.py::test_storage_pools
```

#### Notes & Discussion

`attach_storage` & `detach_storage` seem to work inconsistently on Juju's side, we're currently investigating, possibly a launchpad bug is coming, which makes it difficult/impossible to test their behavior here, so we skip the tests for these at the moment.

Author: jujubot

juju/application.py

@@ -126,10 +126,12 @@ async def relate(self, local_relation, remote_relation):
 
         return await self.model.relate(local_relation, remote_relation)
 
-    async def add_unit(self, count=1, to=None):
+    async def add_unit(self, count=1, to=None, attach_storage=[]):
         """Add one or more units to this application.
 
         :param int count: Number of units to add
+        :param [str] attach_storage: Existing storage to attach to the deployed unit
+        (not available on k8s models)
         :param str to: Placement directive, e.g.::
             '23' - machine 23
             'lxc:7' - new lxc container on machine 7
@@ -153,6 +155,7 @@ async def add_unit(self, count=1, to=None):
             application=self.name,
             placement=parse_placement(to) if to else None,
             num_units=count,
+            attach_storage=attach_storage,
         )
 
         return await jasyncio.gather(*[

juju/model.py

@@ -888,6 +888,123 @@ async def reset(self, force=False):
             lambda: len(self.machines) == 0
         )
 
+    async def create_storage_pool(self, name, provider_type, attributes=""):
+        """Create or define a storage pool.
+
+        :param str name: a pool name
+        :param str provider_type: provider type (defaults to "kubernetes" for
+        Kubernetes models)
+        :param str attributes: attributes for configuration as space-separated pairs,
+        e.g. tags, size, path, etc.
+        :return:
+        """
+        _attrs = [splt.split("=") for splt in attributes.split()]
+
+        storage_facade = client.StorageFacade.from_connection(self.connection())
+        return await storage_facade.CreatePool(pools=[client.StoragePool(
+            name=name,
+            provider=provider_type,
+            attrs=dict(_attrs)
+        )])
+
+    async def remove_storage_pool(self, name):
+        """Remove an existing storage pool.
+
+        :param str name:
+        :return:
+        """
+        storage_facade = client.StorageFacade.from_connection(self.connection())
+        return await storage_facade.RemovePool(pools=[client.StoragePoolDeleteArg(name)])
+
+    async def update_storage_pool(self, name, attributes=""):
+        """ Update storage pool attributes.
+
+        :param name:
+        :param attributes: "key=value key=value ..."
+        :return:
+        """
+        _attrs = dict([splt.split("=") for splt in attributes.split()])
+        if len(_attrs) == 0:
+            raise JujuError("Expected at least one attribute to update")
+
+        storage_facade = client.StorageFacade.from_connection(self.connection())
+        return await storage_facade.UpdatePool(pools=[client.StoragePool(
+            name=name,
+            attrs=_attrs,
+        )])
+
+    async def list_storage(self, filesystem=False, volume=False):
+        """Lists storage details.
+
+        :param bool filesystem: List filesystem storage
+        :param bool volume: List volume storage
+        :return:
+        """
+        storage_facade = client.StorageFacade.from_connection(self.connection())
+
+        if filesystem and volume:
+            raise JujuError("--filesystem and --volume can not be used together")
+        if filesystem:
+            _res = await storage_facade.ListFilesystems(filters=[client.FilesystemFilter()])
+        elif volume:
+            _res = await storage_facade.ListVolumes(filters=[client.VolumeFilter()])
+        else:
+            _res = await storage_facade.ListStorageDetails(filters=[client.StorageFilter()])
+
+        err = _res.results[0].error
+        res = _res.results[0].result
+
+        if err is not None:
+            raise JujuError(err.message)
+
+        return [details.serialize() for details in res]
+
+    async def show_storage_details(self, *storage_ids):
+        """Shows storage instance information.
+
+        :param []str storage_ids:
+        :return:
+        """
+        if not storage_ids:
+            raise JujuError("Expected at least one storage ID")
+
+        storage_facade = client.StorageFacade.from_connection(self.connection())
+        res = await storage_facade.StorageDetails(entities=[client.Entity(tag.storage(s)) for s in storage_ids])
+        return [s.result.serialize() for s in res.results]
+
+    async def list_storage_pools(self):
+        """List storage pools.
+
+        :return:
+        """
+        # TODO (cderici): Filter on pool type, name.
+        storage_facade = client.StorageFacade.from_connection(self.connection())
+        res = await storage_facade.ListPools(filters=[client.StoragePoolFilter()])
+        err = res.results[0].error
+        if err:
+            raise JujuError(err.message)
+        return [p.serialize() for p in res.results[0].storage_pools]
+
+    async def remove_storage(self, *storage_ids, force=False, destroy_storage=False):
+        """Removes storage from the model.
+
+        :param bool force: Remove storage even if it is currently attached
+        :param bool destroy_storage: Remove the storage and destroy it
+        :param []str storage_ids:
+        :return:
+        """
+        if not storage_ids:
+            raise JujuError("Expected at least one storage ID")
+
+        storage_facade = client.StorageFacade.from_connection(self.connection())
+        ret = await storage_facade.Remove(storage=[client.RemoveStorageInstance(
+            destroy_storage=destroy_storage,
+            force=force,
+            tag=s,
+        ) for s in storage_ids])
+        if ret.results[0].error:
+            raise JujuError(ret.results[0].error.message)
+
     async def remove_application(self, app_name, block_until_done=False):
         """Removes the given application from the model.
 

juju/tag.py

@@ -44,6 +44,10 @@ def application(app_name):
     return _prefix('application-', app_name)
 
 
+def storage(app_name):
+    return _prefix('storage-', app_name)
+
+
 def unit(unit_name):
     return _prefix('unit-', unit_name.replace('/', '-'))
 

juju/unit.py

@@ -2,7 +2,7 @@
 
 import pyrfc3339
 
-from juju.errors import JujuAPIError
+from juju.errors import JujuAPIError, JujuError
 
 from . import model, tag
 from .annotationhelper import _get_annotations, _set_annotations
@@ -120,6 +120,72 @@ async def resolved(self, retry=False):
             retry=retry,
             tags={'entities': [{'tag': self.tag}]})
 
+    async def add_storage(self, storage_name, pool=None, count=1, size=1024):
+        """Creates a storage and adds it to this unit.
+
+        :param: str storage_name: Name of the storage
+        :param: str pool: the storage pool to provision storage instances from. Must
+        be a name from 'juju storage-pools'.  The default pool is available via
+        executing 'juju model-config storage-default-block-source'.
+        :param: int count: the number of storage instances to provision from <storage-pool> of
+        <size>. Must be a positive integer. The default count is "1". May be restricted
+        by the charm, which can specify a maximum number of storage instances per unit.
+        :param: int size: the required size of the storage instance, in MiB.
+
+        :return: []str storage_tags
+        """
+        constraints = client.StorageConstraints(count=count)
+        if pool:
+            constraints = client.StorageConstraints(pool=pool, count=count, size=size)
+
+        storage_facade = client.StorageFacade.from_connection(self.connection)
+        res = await storage_facade.AddToUnit(storages=[client.StorageAddParams(
+            name=storage_name,
+            unit=tag.unit(self.name),
+            storage=constraints,
+        )])
+        result = res.results[0]
+        if result.error is not None:
+            raise JujuError("{}".format(result.error))
+        storage_details = result.result
+        return storage_details.storage_tags
+
+    async def attach_storage(self, storage_ids=[]):
+        """Attaches existing storage to this unit.
+
+        :param [str] storage_ids: existing storage ids to attach to the unit
+        :return:
+        """
+        if not storage_ids:
+            raise JujuError("Expected a storage ID to be attached to unit {}".format(self.name))
+
+        storage_facade = client.StorageFacade.from_connection(self.connection)
+        return await storage_facade.Attach(ids=[client.StorageAttachmentId(
+            storage_tag=tag.storage(s_id),
+            unit_tag=self.tag,
+        ) for s_id in storage_ids])
+
+    async def detach_storage(self, *storage_ids, force=False):
+        """Detaches storage from units.
+
+        :param bool force: Forcefully detach storage
+        :param [str] storage_ids:
+        :return:
+        """
+        if not storage_ids:
+            raise JujuError("Expected at least one storage ID")
+
+        storage_facade = client.StorageFacade.from_connection(self.connection)
+        ret = await storage_facade.DetachStorage(
+            force=force,
+            ids=client.StorageAttachmentIds(ids=[client.StorageAttachmentId(
+                storage_tag=tag.storage(s),
+                unit_tag=self.tag,
+            ) for s in storage_ids])
+        )
+        if ret.results[0].error:
+            raise JujuError(ret.results[0].error.message)
+
     async def run(self, command, timeout=None):
         """Run command on this unit.
 

tests/integration/test_model.py

@@ -12,7 +12,7 @@
 
 import pylxd
 import pytest
-from juju import jasyncio
+from juju import jasyncio, tag
 from juju.client import client
 from juju.errors import JujuError, JujuUnitError, JujuConnectionError
 from juju.model import Model, ModelObserver
@@ -1051,3 +1051,77 @@ async def test_model_cache_update(event_loop):
         await model.disconnect()
         await m.disconnect()
         await controller.destroy_models(model_name)
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_add_storage(event_loop):
+    async with base.CleanModel() as model:
+        app = await model.deploy('postgresql')
+        await model.wait_for_idle(status="active")
+        unit = app.units[0]
+        ret = await unit.add_storage("pgdata")
+        assert any([tag.storage("pgdata") in s for s in ret])
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_detach_storage(event_loop):
+    pytest.skip('detach/attach_storage inconsistent on Juju side, unable to test')
+    async with base.CleanModel() as model:
+        app = await model.deploy('postgresql')
+        await model.wait_for_idle(status="active")
+        unit = app.units[0]
+        storage_ids = await unit.add_storage("pgdata")
+        storage_id = storage_ids[0]
+        await jasyncio.sleep(5)
+
+        _storage_details_1 = await model.show_storage_details(storage_id)
+        storage_details_1 = _storage_details_1[0]
+        assert 'unit-postgresql-0' in storage_details_1['attachments']
+
+        await unit.detach_storage(storage_id, force=True)
+        await jasyncio.sleep(20)
+
+        _storage_details_2 = await model.show_storage_details(storage_id)
+        storage_details_2 = _storage_details_2[0]
+        assert ('unit-postgresql-0' not in storage_details_2['attachments']) or \
+            storage_details_2['attachments']['unit-postgresql-0'].life == 'dying'
+
+        # remove_storage
+        await model.remove_storage(storage_id, force=True)
+        await jasyncio.sleep(10)
+        storages = await model.list_storage()
+        assert all([storage_id not in s['storage-tag'] for s in storages])
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_list_storage(event_loop):
+    async with base.CleanModel() as model:
+        app = await model.deploy('postgresql')
+        await model.wait_for_idle(status="active")
+        unit = app.units[0]
+        await unit.add_storage("pgdata")
+        storages = await model.list_storage()
+        await model.list_storage(filesystem=True)
+        await model.list_storage(volume=True)
+
+        assert any([tag.storage("pgdata") in s['storage-tag'] for s in storages])
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_storage_pools(event_loop):
+    async with base.CleanModel() as model:
+        await model.deploy('postgresql')
+        await model.wait_for_idle(status="active")
+
+        await model.create_storage_pool("test-pool", "lxd")
+        pools = await model.list_storage_pools()
+        assert "test-pool" in [p['name'] for p in pools]
+
+        await model.remove_storage_pool("test-pool")
+        await jasyncio.sleep(5)
+        pools = await model.list_storage_pools()
+        assert "test-pool" not in [p['name'] for p in pools]