openstack/keystone
Code Issues Proposed changes

Merge "Purge soft-deleted trusts"

Browse Source
This commit is contained in:
Zuul 2018-10-04 06:20:17 +00:00 committed by Gerrit Code Review
commit f4e052b8bf
4 changed files with 92 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
keystone/cmd/cli.py
View File

@ -676,7 +676,7 @@ class TokenFlush(BaseApp):
class TrustFlush(BaseApp):
"""Flush expired trusts from the backend."""
"""Flush expired and non-expired soft deleted trusts from the backend."""
name = 'trust_flush'
@ -699,7 +699,7 @@ class TrustFlush(BaseApp):
def main(cls):
drivers = backends.load_backends()
trust_manager = drivers['trust_api']
trust_manager.flush_expired_trusts(
trust_manager.flush_expired_and_soft_deleted_trusts(
project_id=CONF.command.project_id,
trustor_user_id=CONF.command.trustor_user_id,
trustee_user_id=CONF.command.trustee_user_id)

78
keystone/tests/unit/trust/test_backends.py
View File

@ -206,7 +206,7 @@ class TrustTests(object):
trust_ref2, roles)
self.assertIsNotNone(trust_data)
PROVIDERS.trust_api.flush_expired_trusts()
PROVIDERS.trust_api.flush_expired_and_soft_deleted_trusts()
trusts = self.trust_api.list_trusts()
self.assertEqual(len(trusts), 1)
self.assertEqual(trust_ref2['id'], trusts[0]['id'])
@ -233,7 +233,7 @@ class TrustTests(object):
trust_ref2, roles)
self.assertIsNotNone(trust_data)
PROVIDERS.trust_api.flush_expired_trusts(
PROVIDERS.trust_api.flush_expired_and_soft_deleted_trusts(
project_id=self.tenant_bar['id'],
trustor_user_id=self.user_foo['id'],
trustee_user_id=self.user_two['id'])
@ -263,7 +263,7 @@ class TrustTests(object):
trust_ref2, roles)
self.assertIsNotNone(trust_data)
PROVIDERS.trust_api.flush_expired_trusts(
PROVIDERS.trust_api.flush_expired_and_soft_deleted_trusts(
trustor_user_id=self.user_foo['id'],
trustee_user_id=self.user_two['id'])
trusts = self.trust_api.list_trusts()
@ -292,7 +292,7 @@ class TrustTests(object):
trust_ref2, roles)
self.assertIsNotNone(trust_data)
PROVIDERS.trust_api.flush_expired_trusts(
PROVIDERS.trust_api.flush_expired_and_soft_deleted_trusts(
project_id=self.tenant_bar['id'],
trustee_user_id=self.user_two['id'])
trusts = self.trust_api.list_trusts()
@ -321,7 +321,7 @@ class TrustTests(object):
trust_ref2, roles)
self.assertIsNotNone(trust_data)
PROVIDERS.trust_api.flush_expired_trusts(
PROVIDERS.trust_api.flush_expired_and_soft_deleted_trusts(
project_id=self.tenant_bar['id'],
trustor_user_id=self.user_foo['id'])
trusts = self.trust_api.list_trusts()
@ -350,7 +350,7 @@ class TrustTests(object):
trust_ref2, roles)
self.assertIsNotNone(trust_data)
PROVIDERS.trust_api.flush_expired_trusts(
PROVIDERS.trust_api.flush_expired_and_soft_deleted_trusts(
project_id=self.tenant_bar['id'])
trusts = self.trust_api.list_trusts()
self.assertEqual(len(trusts), 1)
@ -377,7 +377,7 @@ class TrustTests(object):
trust_data = PROVIDERS.trust_api.create_trust(trust_ref2['id'],
trust_ref2, roles)
self.assertIsNotNone(trust_data)
PROVIDERS.trust_api.flush_expired_trusts(
PROVIDERS.trust_api.flush_expired_and_soft_deleted_trusts(
trustee_user_id=self.user_two['id'])
trusts = self.trust_api.list_trusts()
self.assertEqual(len(trusts), 1)
@ -405,8 +405,70 @@ class TrustTests(object):
trust_ref2, roles)
self.assertIsNotNone(trust_data)
PROVIDERS.trust_api.flush_expired_trusts(
PROVIDERS.trust_api.flush_expired_and_soft_deleted_trusts(
trustor_user_id=self.user_foo['id'])
trusts = self.trust_api.list_trusts()
self.assertEqual(len(trusts), 1)
self.assertEqual(trust_ref2['id'], trusts[0]['id'])
def test_non_expired_soft_deleted_trusts(self):
roles = [{"id": "member"},
{"id": "other"},
{"id": "browser"}]
trust_ref1 = core.new_trust_ref(
self.user_foo['id'], self.user_two['id'],
project_id=self.tenant_bar['id'])
trust_ref1['expires_at'] = \
timeutils.utcnow() + datetime.timedelta(minutes=10)
trust_ref2 = core.new_trust_ref(
self.user_two['id'], self.user_two['id'],
project_id=self.tenant_bar['id'])
trust_ref2['expires_at'] = \
timeutils.utcnow() + datetime.timedelta(minutes=5)
trust_data = PROVIDERS.trust_api.create_trust(trust_ref1['id'],
trust_ref1, roles)
self.assertIsNotNone(trust_data)
trust_data = PROVIDERS.trust_api.create_trust(trust_ref2['id'],
trust_ref2, roles)
self.assertIsNotNone(trust_data)
PROVIDERS.trust_api.delete_trust(trust_ref2['id'])
PROVIDERS.trust_api.flush_expired_and_soft_deleted_trusts()
trusts = self.trust_api.list_trusts()
self.assertEqual(len(trusts), 1)
self.assertEqual(trust_ref1['id'], trusts[0]['id'])
def test_non_expired_non_deleted_trusts(self):
roles = [{"id": "member"},
{"id": "other"},
{"id": "browser"}]
trust_ref1 = core.new_trust_ref(
self.user_foo['id'], self.user_two['id'],
project_id=self.tenant_bar['id'])
trust_ref1['expires_at'] = \
timeutils.utcnow() + datetime.timedelta(minutes=10)
trust_ref2 = core.new_trust_ref(
self.user_two['id'], self.user_two['id'],
project_id=self.tenant_bar['id'])
trust_ref2['expires_at'] = \
timeutils.utcnow() + datetime.timedelta(minutes=5)
trust_ref3 = core.new_trust_ref(
self.user_two['id'], self.user_foo['id'],
project_id=self.tenant_bar['id'])
trust_ref3['expires_at'] = None
trust_data = PROVIDERS.trust_api.create_trust(trust_ref1['id'],
trust_ref1, roles)
self.assertIsNotNone(trust_data)
trust_data = PROVIDERS.trust_api.create_trust(trust_ref2['id'],
trust_ref2, roles)
self.assertIsNotNone(trust_data)
PROVIDERS.trust_api.delete_trust(trust_ref2['id'])
trust_data = PROVIDERS.trust_api.create_trust(trust_ref3['id'],
trust_ref3, roles)
self.assertIsNotNone(trust_data)
PROVIDERS.trust_api.flush_expired_and_soft_deleted_trusts()
trusts = self.trust_api.list_trusts()
self.assertEqual(len(trusts), 2)

19
keystone/trust/backends/sql.py
View File

@ -14,6 +14,8 @@
from oslo_utils import timeutils
from six.moves import range
import sqlalchemy
from sqlalchemy.ext.hybrid import hybrid_property
from keystone.common import sql
@ -189,17 +191,22 @@ class Trust(base.TrustDriverBase):
for trust_ref in trusts:
trust_ref.deleted_at = timeutils.utcnow()
def flush_expired_trusts(self, project_id=None, trustor_user_id=None,
trustee_user_id=None):
def flush_expired_and_soft_deleted_trusts(self, project_id=None,
trustor_user_id=None,
trustee_user_id=None):
with sql.session_for_write() as session:
query = session.query(TrustModel)
query = query.\
filter(sqlalchemy.or_(TrustModel.deleted_at.isnot(None),
sqlalchemy.and_(
TrustModel.expires_at.isnot(None),
TrustModel.expires_at <
timeutils.utcnow())))
if project_id:
query = query.filter_by(project_id=project_id)
if trustor_user_id:
query = query.filter_by(trustor_user_id=trustor_user_id)
if trustee_user_id:
query = query.filter_by(trustee_user_id=trustee_user_id)
for ref in query:
if ref.expires_at is not None:
if ref.expires_at < timeutils.utcnow():
session.delete(ref)
query.delete(synchronize_session=False)
session.flush()

13
releasenotes/notes/bug-1473292-c21481e6aec29ec2.yaml
View File

@ -2,9 +2,10 @@
feature:
- |
[`Bug 1473292 <https://bugs.launchpad.net/keystone/+bug/1473292>`_]
As trusts created by user are stored in database resulting db it to grow
larger as trusts that are expired are not automatically purged by keystone.
Thus this implements TrustFlush via keystone-manage to delete expired trusts.
As trusts created by user are stored in database resulting it to grow larger
as trusts that are expired and soft deleted non-expired are not automatically
purged by keystone.Thus this implements TrustFlush via keystone-manage to delete
expired and soft deleted non-expired trusts.
Command:
@ -12,9 +13,9 @@ feature:
Options:
project-id <string>: To purge expired trusts of given project-id.
trustor-user-id <string>: To purge expired trusts of given trustor-id.
trustee-user-id <string>: To purge expired trusts of given trustee-id.
project-id <string>: To purge trusts of given project-id.
trustor-user-id <string>: To purge trusts of given trustor-id.
trustee-user-id <string>: To purge trusts of given trustee-id.