| # ============LICENSE_START=================================================== |
| # Copyright (C) 2019-2020 Nordix Foundation. |
| # ============================================================================ |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| # |
| # SPDX-License-Identifier: Apache-2.0 |
| # ============LICENSE_END===================================================== |
| import re |
| from enum import Enum |
| |
| import mod.pmsh_logging as logger |
| from mod import db |
| from mod.db_models import SubscriptionModel, NfSubRelationalModel |
| from tenacity import retry, retry_if_exception_type, wait_exponential, stop_after_attempt |
| |
| |
| class SubNfState(Enum): |
| PENDING_CREATE = 'PENDING_CREATE' |
| CREATE_FAILED = 'CREATE_FAILED' |
| CREATED = 'CREATED' |
| PENDING_DELETE = 'PENDING_DELETE' |
| DELETE_FAILED = 'DELETE_FAILED' |
| |
| |
| class AdministrativeState(Enum): |
| UNLOCKED = 'UNLOCKED' |
| LOCKED = 'LOCKED' |
| |
| |
| class Subscription: |
| def __init__(self, **kwargs): |
| self.subscriptionName = kwargs.get('subscriptionName') |
| self.administrativeState = kwargs.get('administrativeState') |
| self.fileBasedGP = kwargs.get('fileBasedGP') |
| self.fileLocation = kwargs.get('fileLocation') |
| self.nfTypeModelInvariantId = kwargs.get('nfTypeModelInvariantId') |
| self.nfFilter = kwargs.get('nfFilter') |
| self.measurementGroups = kwargs.get('measurementGroups') |
| |
| def prepare_subscription_event(self, xnf_name): |
| """Prepare the sub event for publishing |
| |
| Args: |
| xnf_name: the AAI xnf name. |
| |
| Returns: |
| dict: the Subscription event to be published. |
| """ |
| clean_sub = {k: v for k, v in self.__dict__.items() if k != 'nfFilter'} |
| clean_sub.update({'nfName': xnf_name, 'policyName': f'OP-{self.subscriptionName}', |
| 'changeType': 'DELETE' |
| if self.administrativeState == AdministrativeState.LOCKED.value |
| else 'CREATE'}) |
| return clean_sub |
| |
| def create(self): |
| """ Creates a subscription database entry |
| |
| Returns: |
| Subscription object |
| """ |
| existing_subscription = (SubscriptionModel.query.filter( |
| SubscriptionModel.subscription_name == self.subscriptionName).one_or_none()) |
| |
| if existing_subscription is None: |
| new_subscription = SubscriptionModel(subscription_name=self.subscriptionName, |
| status=self.administrativeState) |
| |
| db.session.add(new_subscription) |
| db.session.commit() |
| |
| return new_subscription |
| |
| else: |
| logger.debug(f'Subscription {self.subscriptionName} already exists,' |
| f' returning this subscription..') |
| return existing_subscription |
| |
| def add_network_functions_to_subscription(self, nf_list): |
| """ Associates network functions to a Subscription |
| |
| Args: |
| nf_list : A list of NetworkFunction objects. |
| """ |
| current_sub = self.create() |
| logger.debug(f'Adding network functions to subscription {current_sub.subscription_name}') |
| |
| for nf in nf_list: |
| current_nf = nf.create() |
| |
| existing_entry = NfSubRelationalModel.query.filter( |
| NfSubRelationalModel.subscription_name == current_sub.subscription_name, |
| NfSubRelationalModel.nf_name == current_nf.nf_name).one_or_none() |
| if existing_entry is None: |
| new_nf_sub = NfSubRelationalModel(current_sub.subscription_name, |
| nf.nf_name, SubNfState.PENDING_CREATE.value) |
| new_nf_sub.nf = current_nf |
| logger.debug(current_nf) |
| current_sub.nfs.append(new_nf_sub) |
| |
| db.session.add(current_sub) |
| db.session.commit() |
| |
| @staticmethod |
| def get(subscription_name): |
| """ Retrieves a subscription |
| |
| Args: |
| subscription_name (str): The subscription name |
| |
| Returns: |
| Subscription object else None |
| """ |
| return SubscriptionModel.query.filter( |
| SubscriptionModel.subscription_name == subscription_name).one_or_none() |
| |
| @staticmethod |
| def get_all(): |
| """ Retrieves a list of subscriptions |
| |
| Returns: |
| list: Subscription list else empty |
| """ |
| return SubscriptionModel.query.all() |
| |
| def update_subscription_status(self): |
| """ Updates the status of subscription in subscription table """ |
| SubscriptionModel.query.filter( |
| SubscriptionModel.subscription_name == self.subscriptionName). \ |
| update({SubscriptionModel.status: self.administrativeState}, |
| synchronize_session='evaluate') |
| |
| db.session.commit() |
| |
| def delete_subscription(self): |
| """ Deletes a subscription and all its association from the database. A network function |
| that is only associated with the subscription being removed will also be deleted.""" |
| subscription = SubscriptionModel.query.filter( |
| SubscriptionModel.subscription_name == self.subscriptionName).one_or_none() |
| if subscription: |
| for nf_relationship in subscription.nfs: |
| other_nf_relationship = NfSubRelationalModel.query.filter( |
| NfSubRelationalModel.subscription_name != self.subscriptionName, |
| NfSubRelationalModel.nf_name == nf_relationship.nf_name).one_or_none() |
| if not other_nf_relationship: |
| db.session.delete(nf_relationship.nf) |
| db.session.delete(subscription) |
| db.session.commit() |
| |
| @retry(wait=wait_exponential(multiplier=1, min=30, max=120), stop=stop_after_attempt(3), |
| retry=retry_if_exception_type(Exception)) |
| def process_subscription(self, nfs, mr_pub): |
| action = 'Deactivate' |
| sub_nf_state = SubNfState.PENDING_DELETE.value |
| self.update_subscription_status() |
| |
| if self.administrativeState == AdministrativeState.UNLOCKED.value: |
| action = 'Activate' |
| sub_nf_state = SubNfState.PENDING_CREATE.value |
| |
| try: |
| for nf in nfs: |
| mr_pub.publish_subscription_event_data(self, nf.nf_name) |
| logger.debug(f'Publishing Event to {action} ' |
| f'Sub: {self.subscriptionName} for the nf: {nf.nf_name}') |
| self.add_network_functions_to_subscription(nfs) |
| self.update_sub_nf_status(self.subscriptionName, sub_nf_state, nf.nf_name) |
| except Exception as err: |
| raise Exception(f'Error publishing activation event to MR: {err}') |
| |
| @staticmethod |
| def get_all_nfs_subscription_relations(): |
| """ Retrieves all network function to subscription relations |
| |
| Returns: |
| list: NetworkFunctions per Subscription list else empty |
| """ |
| nf_per_subscriptions = NfSubRelationalModel.query.all() |
| |
| return nf_per_subscriptions |
| |
| @staticmethod |
| def update_sub_nf_status(subscription_name, status, nf_name): |
| """ Updates the status of the subscription for a particular nf |
| |
| Args: |
| subscription_name (str): The subscription name |
| nf_name (str): The network function name |
| status (str): Status of the subscription |
| """ |
| NfSubRelationalModel.query.filter( |
| NfSubRelationalModel.subscription_name == subscription_name, |
| NfSubRelationalModel.nf_name == nf_name). \ |
| update({NfSubRelationalModel.nf_sub_status: status}, synchronize_session='evaluate') |
| |
| db.session.commit() |
| |
| |
| class NetworkFunctionFilter: |
| def __init__(self, **kwargs): |
| self.nf_sw_version = kwargs.get('swVersions') |
| self.nf_names = kwargs.get('nfNames') |
| self.regex_matcher = re.compile('|'.join(raw_regex for raw_regex in self.nf_names)) |
| |
| def is_nf_in_filter(self, nf_name): |
| """Match the nf name against regex values in Subscription.nfFilter.nfNames |
| |
| Args: |
| nf_name: the AAI nf name. |
| |
| Returns: |
| bool: True if matched, else False. |
| """ |
| return self.regex_matcher.search(nf_name) |