diff options
author | efiacor <fiachra.corcoran@est.tech> | 2020-12-11 16:19:47 +0000 |
---|---|---|
committer | efiacor <fiachra.corcoran@est.tech> | 2021-01-12 11:35:09 +0000 |
commit | 5e0e87f7e86f511511d37f90242300296e7b5dc4 (patch) | |
tree | d8ed3fc96a28cbdcd347565837e8a3fb6bcf2742 /components/pm-subscription-handler/pmsh_service | |
parent | 1e6f4d9c7ea75302f4b902dbbe13642b6d20d716 (diff) |
[PMSH] Add retry mech for DELETE_FAILED NFs
Signed-off-by: efiacor <fiachra.corcoran@est.tech>
Change-Id: I3980f0bb43c67e192828172dafe4e7be102dcc98
Issue-ID: DCAEGEN2-2152
Diffstat (limited to 'components/pm-subscription-handler/pmsh_service')
8 files changed, 159 insertions, 65 deletions
diff --git a/components/pm-subscription-handler/pmsh_service/mod/aai_event_handler.py b/components/pm-subscription-handler/pmsh_service/mod/aai_event_handler.py index fd9f6252..46d52f12 100755 --- a/components/pm-subscription-handler/pmsh_service/mod/aai_event_handler.py +++ b/components/pm-subscription-handler/pmsh_service/mod/aai_event_handler.py @@ -77,7 +77,7 @@ def process_aai_events(mr_sub, mr_pub, app, app_conf): def _process_event(action, nf, mr_pub, app_conf): if action == AAIEvent.UPDATE.value: logger.info(f'Update event found for network function {nf.nf_name}') - app_conf.subscription.activate_subscription([nf], mr_pub, app_conf) + app_conf.subscription.create_subscription_on_nfs([nf], mr_pub, app_conf) elif action == AAIEvent.DELETE.value: logger.info(f'Delete event found for network function {nf.nf_name}') NetworkFunction.delete(nf_name=nf.nf_name) diff --git a/components/pm-subscription-handler/pmsh_service/mod/api/db_models.py b/components/pm-subscription-handler/pmsh_service/mod/api/db_models.py index ea9603ef..a9dd6efe 100755 --- a/components/pm-subscription-handler/pmsh_service/mod/api/db_models.py +++ b/components/pm-subscription-handler/pmsh_service/mod/api/db_models.py @@ -63,6 +63,7 @@ class NetworkFunctionModel(db.Model): model_name = Column(String(100)) sdnc_model_name = Column(String(100)) sdnc_model_version = Column(String(100)) + retry_count = Column(Integer) subscriptions = relationship( 'NfSubRelationalModel', @@ -71,7 +72,7 @@ class NetworkFunctionModel(db.Model): def __init__(self, nf_name, ip_address, model_invariant_id, model_version_id, model_name, sdnc_model_name, - sdnc_model_version): + sdnc_model_version, retry_count=0): self.nf_name = nf_name self.ip_address = ip_address self.model_invariant_id = model_invariant_id @@ -79,6 +80,7 @@ class NetworkFunctionModel(db.Model): self.model_name = model_name self.sdnc_model_name = sdnc_model_name self.sdnc_model_version = sdnc_model_version + self.retry_count = retry_count def __repr__(self): return str(self.to_nf()) diff --git a/components/pm-subscription-handler/pmsh_service/mod/exit_handler.py b/components/pm-subscription-handler/pmsh_service/mod/exit_handler.py index fbb8b241..aed86307 100755 --- a/components/pm-subscription-handler/pmsh_service/mod/exit_handler.py +++ b/components/pm-subscription-handler/pmsh_service/mod/exit_handler.py @@ -40,13 +40,15 @@ class ExitHandler: logger.info('Graceful shutdown of PMSH initiated.') logger.debug(f'ExitHandler was called with signal number: {sig_num}.') for thread in self.periodic_tasks: - if thread.name == 'app_conf_thread': + if thread.name == 'aai_event_thread': logger.info(f'Cancelling thread {thread.name}') thread.cancel() current_sub = self.app_conf.subscription if current_sub.administrativeState == AdministrativeState.UNLOCKED.value: try: - current_sub.deactivate_subscription(self.subscription_handler.mr_pub, self.app_conf) + nfs = self.app_conf.subscription.get_network_functions() + current_sub.delete_subscription_from_nfs(nfs, self.subscription_handler.mr_pub, + self.app_conf) except Exception as e: logger.error(f'Failed to shut down PMSH application: {e}', exc_info=True) for thread in self.periodic_tasks: diff --git a/components/pm-subscription-handler/pmsh_service/mod/network_function.py b/components/pm-subscription-handler/pmsh_service/mod/network_function.py index a6d2164f..02656356 100755 --- a/components/pm-subscription-handler/pmsh_service/mod/network_function.py +++ b/components/pm-subscription-handler/pmsh_service/mod/network_function.py @@ -105,6 +105,19 @@ class NetworkFunction: exc_info=True) return not params_set + def increment_retry_count(self): + try: + NetworkFunctionModel.query.filter( + NetworkFunctionModel.nf_name == self.nf_name)\ + .update({'retry_count': NetworkFunctionModel.retry_count + 1}, + synchronize_session='evaluate') + db.session.commit() + except Exception as e: + logger.error(f'Failed to update retry_count of NetworkFunction: {self.nf_name}: {e}', + exc_info=True) + finally: + db.session.remove() + @staticmethod def get(nf_name): """ Retrieves a network function diff --git a/components/pm-subscription-handler/pmsh_service/mod/policy_response_handler.py b/components/pm-subscription-handler/pmsh_service/mod/policy_response_handler.py index 73a5e7e8..09c97047 100644 --- a/components/pm-subscription-handler/pmsh_service/mod/policy_response_handler.py +++ b/components/pm-subscription-handler/pmsh_service/mod/policy_response_handler.py @@ -30,6 +30,10 @@ policy_response_handle_functions = { AdministrativeState.UNLOCKED.value: { 'success': Subscription.update_sub_nf_status, 'failed': Subscription.update_sub_nf_status + }, + AdministrativeState.LOCKING.value: { + 'success': NetworkFunction.delete, + 'failed': Subscription.update_sub_nf_status } } diff --git a/components/pm-subscription-handler/pmsh_service/mod/subscription.py b/components/pm-subscription-handler/pmsh_service/mod/subscription.py index 34753e84..8443c9de 100755 --- a/components/pm-subscription-handler/pmsh_service/mod/subscription.py +++ b/components/pm-subscription-handler/pmsh_service/mod/subscription.py @@ -1,5 +1,5 @@ # ============LICENSE_START=================================================== -# Copyright (C) 2019-2020 Nordix Foundation. +# Copyright (C) 2019-2021 Nordix Foundation. # ============================================================================ # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. @@ -27,26 +27,40 @@ class SubNfState(Enum): CREATED = 'CREATED' PENDING_DELETE = 'PENDING_DELETE' DELETE_FAILED = 'DELETE_FAILED' + DELETED = 'DELETED' class AdministrativeState(Enum): UNLOCKED = 'UNLOCKED' + LOCKING = 'LOCKING' LOCKED = 'LOCKED' - PENDING = 'PENDING' subscription_nf_states = { AdministrativeState.LOCKED.value: { - 'success': SubNfState.CREATED, + 'success': SubNfState.DELETED, 'failed': SubNfState.DELETE_FAILED }, AdministrativeState.UNLOCKED.value: { 'success': SubNfState.CREATED, 'failed': SubNfState.CREATE_FAILED + }, + AdministrativeState.LOCKING.value: { + 'success': SubNfState.DELETED, + 'failed': SubNfState.DELETE_FAILED } } +def _get_nf_objects(nf_sub_relationships): + nfs = [] + for nf_sub_entry in nf_sub_relationships: + nf_model_object = NetworkFunctionModel.query.filter( + NetworkFunctionModel.nf_name == nf_sub_entry.nf_name).one_or_none() + nfs.append(nf_model_object.to_nf()) + return nfs + + class Subscription: def __init__(self, **kwargs): self.subscriptionName = kwargs.get('subscriptionName') @@ -68,7 +82,7 @@ class Subscription: SubscriptionModel.subscription_name == self.subscriptionName).one_or_none()) if existing_subscription is None: new_subscription = SubscriptionModel(subscription_name=self.subscriptionName, - status=AdministrativeState.PENDING.value) + status=AdministrativeState.LOCKED.value) db.session.add(new_subscription) db.session.commit() return new_subscription @@ -109,14 +123,17 @@ class Subscription: """ try: clean_sub = {k: v for k, v in self.__dict__.items() if k != 'nfFilter'} + if self.administrativeState == AdministrativeState.LOCKING.value: + change_type = 'DELETE' + else: + change_type = 'CREATE' sub_event = {'nfName': nf.nf_name, 'ipv4Address': nf.ip_address, 'blueprintName': nf.sdnc_model_name, 'blueprintVersion': nf.sdnc_model_version, 'policyName': app_conf.operational_policy_name, - 'changeType': 'DELETE' - if self.administrativeState == AdministrativeState.LOCKED.value - else 'CREATE', 'closedLoopControlName': app_conf.control_loop_name, + 'changeType': change_type, + 'closedLoopControlName': app_conf.control_loop_name, 'subscription': clean_sub} return sub_event except Exception as e: @@ -182,35 +199,45 @@ class Subscription: db.session.remove() return sub_models - def activate_subscription(self, nfs, mr_pub, app_conf): - logger.info(f'Activate subscription initiated for {self.subscriptionName}.') + def create_subscription_on_nfs(self, nfs, mr_pub, app_conf): + """ Publishes an event to create a Subscription on an nf + + Args: + nfs(list[NetworkFunction]): A list of NetworkFunction Objects. + mr_pub (_MrPub): MR publisher + app_conf (AppConfig): the application configuration. + """ try: existing_nfs = self.get_network_functions() sub_model = self.get() for nf in [new_nf for new_nf in nfs if new_nf not in existing_nfs]: - logger.info(f'Publishing event to activate ' - f'Sub: {self.subscriptionName} for the nf: {nf.nf_name}') + logger.info(f'Publishing event to create ' + f'Sub: {self.subscriptionName} on nf: {nf.nf_name}') mr_pub.publish_subscription_event_data(self, nf, app_conf) self.add_network_function_to_subscription(nf, sub_model) self.update_sub_nf_status(self.subscriptionName, SubNfState.PENDING_CREATE.value, nf.nf_name) except Exception as err: - raise Exception(f'Error publishing activation event to MR: {err}') + raise Exception(f'Error publishing create event to MR: {err}') + + def delete_subscription_from_nfs(self, nfs, mr_pub, app_conf): + """ Publishes an event to delete a Subscription from an nf - def deactivate_subscription(self, mr_pub, app_conf): + Args: + nfs(list[NetworkFunction]): A list of NetworkFunction Objects. + mr_pub (_MrPub): MR publisher + app_conf (AppConfig): the application configuration. + """ try: - nfs = self.get_network_functions() - if nfs: - logger.info(f'Deactivate subscription initiated for {self.subscriptionName}.') - for nf in nfs: - mr_pub.publish_subscription_event_data(self, nf, app_conf) - logger.debug(f'Publishing Event to deactivate ' - f'Sub: {self.subscriptionName} for the nf: {nf.nf_name}') - self.update_sub_nf_status(self.subscriptionName, - SubNfState.PENDING_DELETE.value, - nf.nf_name) + for nf in nfs: + logger.debug(f'Publishing Event to delete ' + f'Sub: {self.subscriptionName} from the nf: {nf.nf_name}') + mr_pub.publish_subscription_event_data(self, nf, app_conf) + self.update_sub_nf_status(self.subscriptionName, + SubNfState.PENDING_DELETE.value, + nf.nf_name) except Exception as err: - raise Exception(f'Error publishing deactivation event to MR: {err}') + raise Exception(f'Error publishing delete event to MR: {err}') @staticmethod def get_all_nfs_subscription_relations(): @@ -245,10 +272,22 @@ class Subscription: def get_network_functions(self): nf_sub_relationships = NfSubRelationalModel.query.filter( NfSubRelationalModel.subscription_name == self.subscriptionName) - nfs = [] - for nf_sub_entry in nf_sub_relationships: - nf_model_object = NetworkFunctionModel.query.filter( - NetworkFunctionModel.nf_name == nf_sub_entry.nf_name).one_or_none() - nfs.append(nf_model_object.to_nf()) + nfs = _get_nf_objects(nf_sub_relationships) + db.session.remove() + return nfs + + def get_delete_failed_nfs(self): + nf_sub_relationships = NfSubRelationalModel.query.filter( + NfSubRelationalModel.subscription_name == self.subscriptionName, + NfSubRelationalModel.nf_sub_status == SubNfState.DELETE_FAILED.value) + nfs = _get_nf_objects(nf_sub_relationships) + db.session.remove() + return nfs + + def get_delete_pending_nfs(self): + nf_sub_relationships = NfSubRelationalModel.query.filter( + NfSubRelationalModel.subscription_name == self.subscriptionName, + NfSubRelationalModel.nf_sub_status == SubNfState.PENDING_DELETE.value) + nfs = _get_nf_objects(nf_sub_relationships) db.session.remove() return nfs diff --git a/components/pm-subscription-handler/pmsh_service/mod/subscription_handler.py b/components/pm-subscription-handler/pmsh_service/mod/subscription_handler.py index 6de702f2..a273a446 100644 --- a/components/pm-subscription-handler/pmsh_service/mod/subscription_handler.py +++ b/components/pm-subscription-handler/pmsh_service/mod/subscription_handler.py @@ -22,12 +22,11 @@ from mod.subscription import AdministrativeState class SubscriptionHandler: - def __init__(self, mr_pub, app, app_conf, aai_event_thread, policy_event_thread): + def __init__(self, mr_pub, app, app_conf, aai_event_thread): self.mr_pub = mr_pub self.app = app self.app_conf = app_conf self.aai_event_thread = aai_event_thread - self.policy_event_thread = policy_event_thread def execute(self): """ @@ -35,37 +34,73 @@ class SubscriptionHandler: the Subscription if a change has occurred """ self.app.app_context().push() + self.app_conf.refresh_config() local_admin_state = self.app_conf.subscription.get_local_sub_admin_state() new_administrative_state = self.app_conf.subscription.administrativeState try: if local_admin_state == new_administrative_state: - logger.info('Administrative State did not change in the Config') + logger.info(f'Administrative State did not change in the app config: ' + f'{new_administrative_state}') else: - if new_administrative_state == AdministrativeState.UNLOCKED.value: - self._activate(local_admin_state, new_administrative_state) - elif local_admin_state == AdministrativeState.PENDING.value: - logger.info('Administrative State is PENDING') - else: - self._deactivate(local_admin_state, new_administrative_state) + self._check_state_change(local_admin_state, new_administrative_state) except Exception as err: logger.error(f'Error occurred during the activation/deactivation process {err}', exc_info=True) - def _activate(self, local_admin_state, new_administrative_state): - logger.info(f'Administrative State has changed from {local_admin_state} ' - f'to {new_administrative_state}.') + def _check_state_change(self, local_admin_state, new_administrative_state): + if local_admin_state == AdministrativeState.LOCKING.value: + self._check_for_failed_nfs() + else: + if new_administrative_state == AdministrativeState.UNLOCKED.value: + logger.info(f'Administrative State has changed from {local_admin_state} ' + f'to {new_administrative_state}.') + self._activate() + elif new_administrative_state == AdministrativeState.LOCKED.value: + logger.info(f'Administrative State has changed from {local_admin_state} ' + f'to {new_administrative_state}.') + self._deactivate() + else: + logger.error(f'Invalid AdministrativeState: {new_administrative_state}') + + def _activate(self): nfs_in_aai = aai.get_pmsh_nfs_from_aai(self.app_conf) - self.app_conf.subscription.activate_subscription(nfs_in_aai, self.mr_pub, - self.app_conf) + self.app_conf.subscription.create_subscription_on_nfs(nfs_in_aai, self.mr_pub, + self.app_conf) self.app_conf.subscription.update_subscription_status() - logger.info('Start listening for new NFs on AAI-EVENT topic in MR.') - self.aai_event_thread.start() - self.policy_event_thread.start() + if not self.aai_event_thread.is_alive(): + logger.info('Start polling for NF info on AAI-EVENT topic on DMaaP MR.') + self.aai_event_thread.start() - def _deactivate(self, local_admin_state, new_administrative_state): - logger.info(f'Administrative State has changed from {local_admin_state} ' - f'to {new_administrative_state}.') - self.aai_event_thread.cancel() - logger.info('Stop listening for NFs on AAI-EVENT topic in MR.') - self.app_conf.subscription.deactivate_subscription(self.mr_pub, self.app_conf) - self.app_conf.subscription.update_subscription_status() + def _deactivate(self): + nfs = self.app_conf.subscription.get_network_functions() + if nfs: + self.aai_event_thread.cancel() + logger.info('Stop listening for NFs events on AAI-EVENT topic in MR.') + self.app_conf.subscription.administrativeState = AdministrativeState.LOCKING.value + logger.info('Subscription is now LOCKING/DEACTIVATING.') + self.app_conf.subscription.delete_subscription_from_nfs(nfs, self.mr_pub, self.app_conf) + self.app_conf.subscription.update_subscription_status() + + def _check_for_failed_nfs(self): + logger.info('Checking for DELETE_FAILED NFs before LOCKING Subscription.') + del_failed_nfs = self.app_conf.subscription.get_delete_failed_nfs() + if del_failed_nfs or self.app_conf.subscription.get_delete_pending_nfs(): + for nf in del_failed_nfs: + nf_model = nf.get(nf.nf_name) + if nf_model.retry_count < 3: + logger.info(f'Retry deletion of subscription ' + f'{self.app_conf.subscription.subscriptionName} ' + f'from NF: {nf.nf_name}') + self.app_conf.subscription.delete_subscription_from_nfs([nf], self.mr_pub, + self.app_conf) + nf.increment_retry_count() + else: + logger.error(f'Failed to delete the subscription ' + f'{self.app_conf.subscription.subscriptionName} ' + f'from NF: {nf.nf_name} after {nf_model.retry_count} ' + f'attempts. Removing NF from DB') + nf.delete(nf_name=nf.nf_name) + else: + logger.info('Proceeding to LOCKED adminState.') + self.app_conf.subscription.administrativeState = AdministrativeState.LOCKED.value + self.app_conf.subscription.update_subscription_status() diff --git a/components/pm-subscription-handler/pmsh_service/pmsh_service_main.py b/components/pm-subscription-handler/pmsh_service/pmsh_service_main.py index f92fdc91..307235db 100755 --- a/components/pm-subscription-handler/pmsh_service/pmsh_service_main.py +++ b/components/pm-subscription-handler/pmsh_service/pmsh_service_main.py @@ -40,26 +40,25 @@ def main(): logger.error(f'Failed to get config and create application: {e}', exc_info=True) sys.exit(e) - app_conf_thread = PeriodicTask(10, app_conf.refresh_config) - app_conf_thread.name = 'app_conf_thread' - app_conf_thread.start() - policy_response_handler = PolicyResponseHandler(policy_mr_sub, app_conf, app) policy_response_handler_thread = PeriodicTask(25, policy_response_handler.poll_policy_topic) policy_response_handler_thread.name = 'policy_event_thread' + logger.info('Start polling PMSH_CL_INPUT topic on DMaaP MR.') + policy_response_handler_thread.start() aai_event_thread = PeriodicTask(20, process_aai_events, args=(aai_event_mr_sub, policy_mr_pub, app, app_conf)) aai_event_thread.name = 'aai_event_thread' + logger.info('Start polling for NF info on AAI-EVENT topic on DMaaP MR.') + aai_event_thread.start() - subscription_handler = SubscriptionHandler(policy_mr_pub, app, app_conf, aai_event_thread, - policy_response_handler_thread) + subscription_handler = SubscriptionHandler(policy_mr_pub, app, app_conf, aai_event_thread) - subscription_handler_thread = PeriodicTask(30, subscription_handler.execute) + subscription_handler_thread = PeriodicTask(20, subscription_handler.execute) subscription_handler_thread.name = 'sub_handler_thread' subscription_handler_thread.start() - periodic_tasks = [app_conf_thread, aai_event_thread, subscription_handler_thread, + periodic_tasks = [aai_event_thread, subscription_handler_thread, policy_response_handler_thread] signal(SIGTERM, ExitHandler(periodic_tasks=periodic_tasks, |