openstack
/
octavia
Code Issues Proposed changes

Ignore DELETED amphorae when performing certificate rotation 

Deleted amphorae should not be included when housekeeping service is
performing certificate rotation. Otherwise, deleted amphorae will
become ERROR.

Story: #2007635
Task: #39681

Change-Id: Ie2c748ff69c8fd736e9c67d686068a33d803222e
This commit is contained in:
Lingxian Kong 2020-09-24 00:24:49 +12:00
parent ba2b9482e7
commit f3eff29ce1
2 changed files with 19 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
octavia/db/repositories.py
View File

@ -30,6 +30,7 @@ from oslo_utils import uuidutils
from sqlalchemy.orm import joinedload
from sqlalchemy.orm import noload
from sqlalchemy.orm import subqueryload
from sqlalchemy.sql.expression import false
from sqlalchemy.sql import func
from octavia.common import constants as consts
@ -1417,9 +1418,12 @@ class AmphoraRepository(BaseRepository):
seconds=expired_seconds)
with session.begin(subtransactions=True):
amp = session.query(self.model_class).with_for_update().filter_by(
cert_busy=False).filter(
self.model_class.cert_expiration < expired_date).first()
amp = session.query(self.model_class).with_for_update().filter(
self.model_class.status.notin_(
[consts.DELETED, consts.PENDING_DELETE]),
self.model_class.cert_busy == false(),
self.model_class.cert_expiration < expired_date
).first()
if amp is None:
return None

12
octavia/tests/functional/db/test_repositories.py
View File

@ -3999,6 +3999,18 @@ class AmphoraRepositoryTest(BaseRepositoryTest):
self.assertEqual(cert_expired_amphora.cert_expiration, expiration)
self.assertEqual(cert_expired_amphora.id, amphora2.id)
def test_get_cert_expired_amphora_deleted(self):
amphora = self.create_amphora(self.FAKE_UUID_3)
expiration = datetime.datetime.utcnow() + datetime.timedelta(seconds=1)
self.amphora_repo.update(self.session, amphora.id,
status=constants.DELETED,
cert_expiration=expiration)
cert_expired_amphora = self.amphora_repo.get_cert_expiring_amphora(
self.session)
self.assertIsNone(cert_expired_amphora)
def test_get_lb_for_health_update(self):
amphora1 = self.create_amphora(self.FAKE_UUID_1)
amphora2 = self.create_amphora(self.FAKE_UUID_3)