Refactor session handling in networking-mlnx

Due upstream changes in neutron to provide compatibility with
SQLAlchemy 2.0. [1].[2], and because session.autocommit=False
will be the only value accepted, refactor all session handling using
context manager

[1] https://opendev.org/openstack/neutron-lib/commit/ef1e8abb89141db39e62eaa1cbdb73aca1ceb647
[2] https://opendev.org/openstack/neutron/commit/88fb5416f7169bd3542bca6af0192f935c8cc18e

Change-Id: I8e020a14b681d6e655b225ab764e332e59c13de7

Author: wmousa

networking_mlnx/db/db.py

@@ -26,8 +26,9 @@
 from networking_mlnx.plugins.ml2.drivers.sdn import constants as sdn_const
 
 
-def check_for_pending_or_processing_ops(session, object_uuid, operation=None):
-    q = session.query(sdn_journal_db.SdnJournal).filter(
+@db_api.CONTEXT_READER
+def check_for_pending_or_processing_ops(context, object_uuid, operation=None):
+    q = context.session.query(sdn_journal_db.SdnJournal).filter(
         or_(sdn_journal_db.SdnJournal.state == sdn_const.PENDING,
             sdn_journal_db.SdnJournal.state == sdn_const.PROCESSING),
         sdn_journal_db.SdnJournal.object_uuid == object_uuid)
@@ -36,11 +37,12 @@ def check_for_pending_or_processing_ops(session, object_uuid, operation=None):
             q = q.filter(sdn_journal_db.SdnJournal.operation.in_(operation))
         else:
             q = q.filter(sdn_journal_db.SdnJournal.operation == operation)
-    return session.query(q.exists()).scalar()
+    return context.session.query(q.exists()).scalar()
 
 
-def check_for_pending_delete_ops_with_parent(session, object_type, parent_id):
-    rows = session.query(sdn_journal_db.SdnJournal).filter(
+@db_api.CONTEXT_READER
+def check_for_pending_delete_ops_with_parent(context, object_type, parent_id):
+    rows = context.session.query(sdn_journal_db.SdnJournal).filter(
         or_(sdn_journal_db.SdnJournal.state == sdn_const.PENDING,
             sdn_journal_db.SdnJournal.state == sdn_const.PROCESSING),
         sdn_journal_db.SdnJournal.object_type == object_type,
@@ -54,123 +56,137 @@ def check_for_pending_delete_ops_with_parent(session, object_type, parent_id):
     return False
 
 
-def check_for_older_ops(session, row):
-    q = session.query(sdn_journal_db.SdnJournal).filter(
+@db_api.CONTEXT_READER
+def check_for_older_ops(context, row):
+    q = context.session.query(sdn_journal_db.SdnJournal).filter(
         or_(sdn_journal_db.SdnJournal.state == sdn_const.PENDING,
             sdn_journal_db.SdnJournal.state == sdn_const.PROCESSING),
         sdn_journal_db.SdnJournal.object_uuid == row.object_uuid,
         sdn_journal_db.SdnJournal.created_at < row.created_at,
         sdn_journal_db.SdnJournal.id != row.id)
-    return session.query(q.exists()).scalar()
+    return context.session.query(q.exists()).scalar()
 
 
-def get_all_db_rows(session):
-    return session.query(sdn_journal_db.SdnJournal).all()
+@db_api.CONTEXT_READER
+def get_all_db_rows(context):
+    return context.session.query(sdn_journal_db.SdnJournal).all()
 
 
-def get_all_db_rows_by_state(session, state):
-    return session.query(sdn_journal_db.SdnJournal).filter_by(
+@db_api.CONTEXT_READER
+def get_all_db_rows_by_state(context, state):
+    return context.session.query(sdn_journal_db.SdnJournal).filter_by(
         state=state).all()
 
 
+def _get_row_with_lock(session):
+    row = session.query(sdn_journal_db.SdnJournal).filter_by(
+        state=sdn_const.PENDING).order_by(
+        asc(sdn_journal_db.SdnJournal.last_retried)).with_for_update(
+    ).first()
+    return row
+
+
 # Retry deadlock exception for Galera DB.
 # If two (or more) different threads call this method at the same time, they
 # might both succeed in changing the same row to pending, but at least one
 # of them will get a deadlock from Galera and will have to retry the operation.
 @db_api.retry_db_errors
-def get_oldest_pending_db_row_with_lock(session):
-    with session.begin():
-        row = session.query(sdn_journal_db.SdnJournal).filter_by(
-            state=sdn_const.PENDING).order_by(
-            asc(sdn_journal_db.SdnJournal.last_retried)).with_for_update(
-        ).first()
-        if row:
-            update_db_row_state(session, row, sdn_const.PROCESSING)
-
+@db_api.CONTEXT_WRITER
+def get_oldest_pending_db_row_with_lock(context):
+    row = _get_row_with_lock(context.session)
+    if row:
+        _update_db_row_state(context.session, row, sdn_const.PROCESSING)
     return row
 
 
 @db_api.retry_db_errors
-def get_all_monitoring_db_row_by_oldest(session):
-    with session.begin():
-        rows = session.query(sdn_journal_db.SdnJournal).filter_by(
-            state=sdn_const.MONITORING).order_by(
-            asc(sdn_journal_db.SdnJournal.last_retried)).all()
+@db_api.CONTEXT_READER
+def get_all_monitoring_db_row_by_oldest(context):
+    rows = context.session.query(sdn_journal_db.SdnJournal).filter_by(
+        state=sdn_const.MONITORING).order_by(
+        asc(sdn_journal_db.SdnJournal.last_retried)).all()
     return rows
 
 
 @oslo_db_api.wrap_db_retry(max_retries=db_api.MAX_RETRIES)
-def update_db_row_state(session, row, state):
+@db_api.CONTEXT_WRITER
+def update_db_row_state(context, row, state):
+    _update_db_row_state(context.session, row, state)
+
+
+def _update_db_row_state(session, row, state):
     row.state = state
     session.merge(row)
-    session.flush()
 
 
 @oslo_db_api.wrap_db_retry(max_retries=db_api.MAX_RETRIES)
-def update_db_row_job_id(session, row, job_id):
+@db_api.CONTEXT_WRITER
+def update_db_row_job_id(context, row, job_id):
     row.job_id = job_id
-    session.merge(row)
-    session.flush()
+    context.session.merge(row)
 
 
-def update_pending_db_row_retry(session, row, retry_count):
+@oslo_db_api.wrap_db_retry(max_retries=db_api.MAX_RETRIES)
+@db_api.CONTEXT_WRITER
+def update_pending_db_row_retry(context, row, retry_count):
     if row.retry_count >= retry_count and retry_count != -1:
-        update_db_row_state(session, row, sdn_const.FAILED)
+        _update_db_row_state(context.session, row, sdn_const.FAILED)
     else:
         row.retry_count += 1
-        update_db_row_state(session, row, sdn_const.PENDING)
+        _update_db_row_state(context.session, row, sdn_const.PENDING)
 
 
 # This function is currently not used.
 # Deleted resources are marked as 'deleted' in the database.
 @oslo_db_api.wrap_db_retry(max_retries=db_api.MAX_RETRIES)
-def delete_row(session, row=None, row_id=None):
+@db_api.CONTEXT_WRITER
+def delete_row(context, row=None, row_id=None):
     if row_id:
-        row = session.query(sdn_journal_db.SdnJournal).filter_by(
+        row = context.session.query(sdn_journal_db.SdnJournal).filter_by(
             id=row_id).one()
     if row:
-        session.delete(row)
-        session.flush()
+        context.session.delete(row)
 
 
 @oslo_db_api.wrap_db_retry(max_retries=db_api.MAX_RETRIES)
-def create_pending_row(session, object_type, object_uuid,
+@db_api.CONTEXT_WRITER
+def create_pending_row(context, object_type, object_uuid,
                        operation, data):
     data = jsonutils.dumps(data)
     row = sdn_journal_db.SdnJournal(object_type=object_type,
                                     object_uuid=object_uuid,
                                     operation=operation, data=data,
                                     created_at=func.now(),
                                     state=sdn_const.PENDING)
-    session.add(row)
-    # Keep session flush for unit tests. NOOP for L2/L3 events since calls are
-    # made inside database session transaction with subtransactions=True.
-    session.flush()
+    context.session.add(row)
 
 
-@db_api.retry_db_errors
 def _update_maintenance_state(session, expected_state, state):
-    with session.begin():
-        row = session.query(sdn_maintenance_db.SdnMaintenance).filter_by(
-            state=expected_state).with_for_update().one_or_none()
-        if row is None:
-            return False
+    row = session.query(sdn_maintenance_db.SdnMaintenance).filter_by(
+        state=expected_state).with_for_update().one_or_none()
+    if row is None:
+        return False
 
-        row.state = state
-        return True
+    row.state = state
+    return True
 
 
-def lock_maintenance(session):
-    return _update_maintenance_state(session, sdn_const.PENDING,
+@db_api.retry_db_errors
+@db_api.CONTEXT_WRITER
+def lock_maintenance(context):
+    return _update_maintenance_state(context.session, sdn_const.PENDING,
                                      sdn_const.PROCESSING)
 
 
-def unlock_maintenance(session):
-    return _update_maintenance_state(session, sdn_const.PROCESSING,
+@db_api.retry_db_errors
+@db_api.CONTEXT_WRITER
+def unlock_maintenance(context):
+    return _update_maintenance_state(context.session, sdn_const.PROCESSING,
                                      sdn_const.PENDING)
 
 
-def update_maintenance_operation(session, operation=None):
+@db_api.CONTEXT_WRITER
+def update_maintenance_operation(context, operation=None):
     """Update the current maintenance operation details.
 
     The function assumes the lock is held, so it mustn't be run outside of a
@@ -180,28 +196,28 @@ def update_maintenance_operation(session, operation=None):
     if operation:
         op_text = operation.__name__
 
-    with session.begin():
-        row = session.query(sdn_maintenance_db.SdnMaintenance).one_or_none()
-        row.processing_operation = op_text
-
-
-def delete_rows_by_state_and_time(session, state, time_delta):
-    with session.begin():
-        now = session.execute(func.now()).scalar()
-        session.query(sdn_journal_db.SdnJournal).filter(
-            sdn_journal_db.SdnJournal.state == state,
-            sdn_journal_db.SdnJournal.last_retried < now - time_delta).delete(
-            synchronize_session=False)
-        session.expire_all()
-
-
-def reset_processing_rows(session, max_timedelta):
-    with session.begin():
-        now = session.execute(func.now()).scalar()
-        max_timedelta = datetime.timedelta(seconds=max_timedelta)
-        rows = session.query(sdn_journal_db.SdnJournal).filter(
-            sdn_journal_db.SdnJournal.last_retried < now - max_timedelta,
-            sdn_journal_db.SdnJournal.state == sdn_const.PROCESSING,
-        ).update({'state': sdn_const.PENDING})
+    row = context.session.query(
+        sdn_maintenance_db.SdnMaintenance).one_or_none()
+    row.processing_operation = op_text
+
+
+@db_api.CONTEXT_WRITER
+def delete_rows_by_state_and_time(context, state, time_delta):
+    now = context.session.execute(func.now()).scalar()
+    context.session.query(sdn_journal_db.SdnJournal).filter(
+        sdn_journal_db.SdnJournal.state == state,
+        sdn_journal_db.SdnJournal.last_retried < now - time_delta).delete(
+        synchronize_session=False)
+    context.session.expire_all()
+
+
+@db_api.CONTEXT_WRITER
+def reset_processing_rows(context, max_timedelta):
+    now = context.session.execute(func.now()).scalar()
+    max_timedelta = datetime.timedelta(seconds=max_timedelta)
+    rows = context.session.query(sdn_journal_db.SdnJournal).filter(
+        sdn_journal_db.SdnJournal.last_retried < now - max_timedelta,
+        sdn_journal_db.SdnJournal.state == sdn_const.PROCESSING,
+    ).update({'state': sdn_const.PENDING})
 
     return rows

networking_mlnx/journal/dependency_validations.py

@@ -19,14 +19,14 @@
 from networking_mlnx.plugins.ml2.drivers.sdn import constants as sdn_const
 
 
-def _is_valid_operation(session, row):
+def _is_valid_operation(context, row):
     # Check if there are older updates in the queue
-    if db.check_for_older_ops(session, row):
+    if db.check_for_older_ops(context, row):
         return False
     return True
 
 
-def validate_network_operation(session, row):
+def validate_network_operation(context, row):
     """Validate the network operation based on dependencies.
 
     Validate network operation depending on whether it's dependencies
@@ -36,19 +36,19 @@ def validate_network_operation(session, row):
         # Check for any pending or processing create or update
         # ops on this uuid itself
         if db.check_for_pending_or_processing_ops(
-            session, row.object_uuid, [sdn_const.PUT,
+            context, row.object_uuid, [sdn_const.PUT,
                                        sdn_const.POST]):
             return False
         if db.check_for_pending_delete_ops_with_parent(
-            session, sdn_const.PORT, row.object_uuid):
+            context, sdn_const.PORT, row.object_uuid):
             return False
     elif (row.operation == sdn_const.PUT and
-            not _is_valid_operation(session, row)):
+            not _is_valid_operation(context, row)):
         return False
     return True
 
 
-def validate_port_operation(session, row):
+def validate_port_operation(context, row):
     """Validate port operation based on dependencies.
 
     Validate port operation depending on whether it's dependencies
@@ -59,10 +59,10 @@ def validate_port_operation(session, row):
         network_id = network_dict['network_id']
         # Check for pending or processing network operations
         ops = db.check_for_pending_or_processing_ops(
-            session, network_id, [sdn_const.POST])
+            context, network_id, [sdn_const.POST])
         if ops:
             return False
-    return _is_valid_operation(session, row)
+    return _is_valid_operation(context, row)
 
 
 _VALIDATION_MAP = {
@@ -71,13 +71,13 @@ def validate_port_operation(session, row):
 }
 
 
-def validate(session, row):
+def validate(context, row):
     """Validate resource dependency in journaled operations.
 
-    :param session: db session
+    :param context: context manager
     :param row: entry in journal entry to be validated
     """
-    return _VALIDATION_MAP[row.object_type](session, row)
+    return _VALIDATION_MAP[row.object_type](context, row)
 
 
 def register_validator(object_type, validator):