and this project adheres to [Semantic Versioning](http://semver.org/).
+## [1.4.0]
+### Changed
+* Update filter to allow updates in realtime (DCAEGEN2-2531)
+
## [1.3.1]
### Changed
* Updated Subscription object retrieving key from App Config data (DCAEGEN2-2713)
"type": "string",
"description": "The name of the nf in A&AI."
},
- "ipv4Address": {
+ "ipAddress": {
"type": "string",
- "description": "The ipv4address of the nf being targeted."
+ "description": "The ip address of the nf being targeted."
},
"policyName": {
"type": "string",
},
"required": [
"nfName",
- "ipv4Address",
+ "ipAddress",
"policyName",
"closedLoopControlName",
"blueprintName",
handlers:
onap_log_handler:
class: logging.handlers.RotatingFileHandler
- filename: /var/log/ONAP/dcaegen2/services/pmsh/application.log
+ filename: ./application.log
mode: a
maxBytes: 10000000
backupCount: 10
# ============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.
# SPDX-License-Identifier: Apache-2.0
# ============LICENSE_END=====================================================
-from sqlalchemy import Column, Integer, String, ForeignKey
+from sqlalchemy import Column, Integer, String, ForeignKey, JSON
from sqlalchemy.orm import relationship
from mod import db
__tablename__ = 'subscriptions'
id = Column(Integer, primary_key=True, autoincrement=True)
subscription_name = Column(String(100), unique=True)
+ nfFilter = Column(JSON)
status = Column(String(20))
nfs = relationship(
cascade='all, delete-orphan',
backref='subscription')
- def __init__(self, subscription_name, status):
+ def __init__(self, subscription_name, nfFilter, status):
self.subscription_name = subscription_name
+ self.nfFilter = nfFilter
self.status = status
def __repr__(self):
- return f'subscription_name: {self.subscription_name}, status: {self.status}'
+ return f'subscription_name: {self.subscription_name}, ' \
+ f'nfFilter: {self.nfFilter}, ' \
+ f'status: {self.status}'
def __eq__(self, other):
if isinstance(self, other.__class__):
sub_nfs = NfSubRelationalModel.query.filter(
NfSubRelationalModel.subscription_name == self.subscription_name).all()
db.session.remove()
- return {'subscription_name': self.subscription_name, 'subscription_status': self.status,
+ return {'subscription_name': self.subscription_name,
+ 'nfFilter': self.nfFilter,
+ 'subscription_status': self.status,
'network_functions': [sub_nf.serialize_nf() for sub_nf in sub_nfs]}
# ============LICENSE_START===================================================
-# Copyright (C) 2020 Nordix Foundation.
+# Copyright (C) 2020-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.
'success': Subscription.update_sub_nf_status,
'failed': Subscription.update_sub_nf_status
},
+ AdministrativeState.FILTERING.value: {
+ 'success': NetworkFunction.delete,
+ 'failed': Subscription.update_sub_nf_status
+ },
AdministrativeState.LOCKING.value: {
'success': NetworkFunction.delete,
'failed': Subscription.update_sub_nf_status
}
+def _set_sub_nf_status(administrative_state, response_message):
+ """
+
+ Args:
+ administrative_state (str): The administrative state of the subscription
+ response_message (str): The message in the response regarding the state (success|failed)
+
+ Returns:
+ str: sub_nf_status
+ """
+ return subscription_nf_states[administrative_state][response_message].value;
+
+
class PolicyResponseHandler:
def __init__(self, mr_sub, app_conf, app):
self.mr_sub = mr_sub
== self.app_conf.subscription.subscriptionName:
nf_name = data['status']['nfName']
response_message = data['status']['message']
+ change_type = data['status']['changeType']
self._handle_response(self.app_conf.subscription.subscriptionName,
- administrative_state, nf_name, response_message)
+ administrative_state, nf_name, response_message, change_type)
except Exception as err:
logger.error(f'Error trying to poll policy response topic on MR: {err}', exc_info=True)
@staticmethod
- def _handle_response(subscription_name, administrative_state, nf_name, response_message):
+ def _handle_response(subscription_name, administrative_state, nf_name, response_message, change_type):
"""
Handles the response from Policy, updating the DB
logger.info(f'Response from MR: Sub: {subscription_name} for '
f'NF: {nf_name} received, updating the DB')
try:
+ if administrative_state == AdministrativeState.UNLOCKED.value and change_type == "DELETED":
+ administrative_state = AdministrativeState.FILTERING.value
sub_nf_status = subscription_nf_states[administrative_state][response_message].value
policy_response_handle_functions[administrative_state][response_message](
- subscription_name=subscription_name, status=sub_nf_status, nf_name=nf_name)
+ subscription_name=subscription_name, status=sub_nf_status, nf_name=nf_name)
except Exception as err:
logger.error(f'Error changing nf_sub status in the DB: {err}')
raise
+
+
+
{
"enum":[
"UNLOCKED",
- "LOCKED"
+ "LOCKED",
+ "FILTERING"
]
}
]
#
# SPDX-License-Identifier: Apache-2.0
# ============LICENSE_END=====================================================
+import json
from enum import Enum
from mod import db, logger
UNLOCKED = 'UNLOCKED'
LOCKING = 'LOCKING'
LOCKED = 'LOCKED'
+ FILTERING = 'FILTERING'
subscription_nf_states = {
AdministrativeState.LOCKING.value: {
'success': SubNfState.DELETED,
'failed': SubNfState.DELETE_FAILED
+ },
+ AdministrativeState.FILTERING.value: {
+ 'success': SubNfState.DELETED,
+ 'failed': SubNfState.DELETE_FAILED
}
}
return nfs
+def get_nfs_for_creation_and_deletion(existing_nfs, new_nfs, action, mrpub, app_conf):
+ """ Finds new/old nfs for creation/deletion from subscription """
+ for existing_nf in existing_nfs:
+ _compare_nfs(action, app_conf, existing_nf, mrpub, new_nfs)
+
+
+def _compare_nfs(action, app_conf, existing_nf, mrpub, new_nfs):
+ """ Compares old nfs list to existing nfs list"""
+ for new_nf in new_nfs:
+ if existing_nf.nf_name != new_nf.nf_name:
+ _apply_action_to_nfs(action, app_conf, existing_nf, mrpub, new_nf)
+
+
+def _apply_action_to_nfs(action, app_conf, existing_nf, mrpub, new_nf):
+ """ Performs create/delete of nf from subscription as required"""
+ if action == 'create':
+ app_conf.subscription.create_subscription_on_nfs([new_nf], mrpub, app_conf)
+ elif action == 'delete':
+ app_conf.subscription.delete_subscription_from_nfs([existing_nf], mrpub, app_conf)
+
+
class Subscription:
def __init__(self, **kwargs):
self.subscriptionName = kwargs.get('subscriptionName')
Returns:
Subscription object
"""
+ existing_subscription = (SubscriptionModel.query.filter(
+ SubscriptionModel.subscription_name == self.subscriptionName).one_or_none())
+ if existing_subscription is None:
+ return self.create_new_sub()
+ else:
+ if existing_subscription.nfFilter and \
+ self._filter_diff(existing_subscription.nfFilter) and \
+ existing_subscription.status == AdministrativeState.UNLOCKED.value:
+ return self.update_existing_sub(existing_subscription)
+
+ def update_existing_sub(self, existing_subscription):
+ """Update subscription status
+
+ Args:
+ existing_subscription: the current subscription
+
+ Returns:
+ Subscription: updated subscription
+ """
+ self.administrativeState = \
+ AdministrativeState.FILTERING.value
+ self.nfFilter = existing_subscription.nfFilter
+ self.update_subscription_status()
+ logger.debug(f'Subscription {self.subscriptionName} already exists,'
+ f' returning this subscription..')
+ return existing_subscription
+
+ def create_new_sub(self):
try:
- 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=AdministrativeState.LOCKED.value)
- 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
+ new_subscription = SubscriptionModel(subscription_name=self.subscriptionName,
+ nfFilter=self.nfFilter,
+ status=AdministrativeState.LOCKED.value)
+ db.session.add(new_subscription)
+ db.session.commit()
+ return new_subscription
except Exception as e:
logger.error(f'Failed to create subscription {self.subscriptionName} in the DB: {e}',
exc_info=True)
finally:
db.session.remove()
+ def _filter_diff(self, existing_subscription_filter):
+ existing_subscription_filter, nf_filter = \
+ json.dumps(existing_subscription_filter, sort_keys=True), \
+ json.dumps(self.nfFilter, sort_keys=True)
+ return existing_subscription_filter != nf_filter
+
def update_subscription_status(self):
""" Updates the status of subscription in subscription table """
try:
SubscriptionModel.query.filter(
- SubscriptionModel.subscription_name == self.subscriptionName)\
+ SubscriptionModel.subscription_name == self.subscriptionName) \
.update({SubscriptionModel.status: self.administrativeState},
synchronize_session='evaluate')
finally:
db.session.remove()
+ def update_subscription_filter(self):
+ """ Updates the filter of subscription in subscription table """
+ try:
+ SubscriptionModel.query.filter(
+ SubscriptionModel.subscription_name == self.subscriptionName) \
+ .update({SubscriptionModel.nfFilter: self.nfFilter},
+ synchronize_session='evaluate')
+
+ db.session.commit()
+ except Exception as e:
+ logger.error(f'Failed to update status of subscription: {self.subscriptionName}: {e}',
+ exc_info=True)
+ finally:
+ db.session.remove()
+
def prepare_subscription_event(self, nf, app_conf):
"""Prepare the sub event for publishing
dict: the Subscription event to be published.
"""
try:
- clean_sub = {k: v for k, v in self.__dict__.items() if k != 'nfFilter'}
+ clean_sub = {k: v for k, v in self.__dict__.items()
+ if k != 'nfFilter' and k != 'current_filter'}
if self.administrativeState == AdministrativeState.LOCKING.value:
change_type = 'DELETE'
else:
change_type = 'CREATE'
sub_event = {'nfName': nf.nf_name,
- 'ipv4Address': nf.ip_address,
+ 'ipAddress': nf.ip_address,
'blueprintName': nf.sdnc_model_name,
'blueprintVersion': nf.sdnc_model_version,
'policyName': app_conf.operational_policy_name,
# ============LICENSE_START===================================================
-# Copyright (C) 2020-2021 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.
from jsonschema import ValidationError
from mod import logger, aai_client
+from mod.aai_client import _filter_nf_data, get_pmsh_nfs_from_aai
from mod.aai_event_handler import process_aai_events
from mod.network_function import NetworkFunctionFilter
from mod.pmsh_utils import PeriodicTask
-from mod.subscription import AdministrativeState
+from mod.subscription import AdministrativeState, Subscription, get_nfs_for_creation_and_deletion
class SubscriptionHandler:
else:
self.app_conf.refresh_config()
self.app_conf.validate_sub_schema()
- new_administrative_state = self.app_conf.subscription.administrativeState
- if local_admin_state == new_administrative_state:
- logger.info(f'Administrative State did not change in the app config: '
- f'{new_administrative_state}')
- else:
- self._check_state_change(local_admin_state, new_administrative_state)
+ local_admin_state = self.apply_subscription_changes()
+ self.compare_admin_state(local_admin_state)
except (ValidationError, TypeError) as err:
logger.error(f'Error occurred during validation of subscription schema {err}',
exc_info=True)
logger.error(f'Error occurred during the activation/deactivation process {err}',
exc_info=True)
+ def apply_subscription_changes(self):
+ """ Applies changes to subscription
+
+ Args:
+ local_admin_state(enum): The local adminstrative state
+
+ Returns:
+ Enum: Updated administrative state
+ """
+ local_admin_state = self.app_conf.subscription.get_local_sub_admin_state()
+ if local_admin_state == AdministrativeState.FILTERING.value:
+ existing_nfs = self.app_conf.subscription.get_network_functions()
+ self.app_conf.nf_filter = \
+ NetworkFunctionFilter(**self.app_conf.subscription.nfFilter)
+ new_nfs = get_pmsh_nfs_from_aai(self.app_conf)
+ self.app_conf.subscription.update_subscription_filter()
+ get_nfs_for_creation_and_deletion(existing_nfs, new_nfs, 'delete',
+ self.mr_pub, self.app_conf)
+ get_nfs_for_creation_and_deletion(existing_nfs, new_nfs, 'create',
+ self.mr_pub, self.app_conf)
+
+ return local_admin_state
+
+ def compare_admin_state(self, local_admin_state):
+ """ Check for changes in administrative state
+
+ Args:
+ local_admin_state(enum):
+
+ """
+ new_administrative_state = self.app_conf.subscription.administrativeState
+ if local_admin_state == new_administrative_state:
+ logger.info(f'Administrative State did not change in the app config: '
+ f'{new_administrative_state}')
+ else:
+ self._check_state_change(local_admin_state, new_administrative_state)
+
+ def fetch_aai_nf_data(self):
+ """ Fetchs AAI data
+
+ Returns:
+ dict: the json response from AAI query after filter is applied
+ """
+ aai_nf_data = aai_client._get_all_aai_nf_data(self.app_conf)
+ if aai_nf_data:
+ new_nfs = _filter_nf_data(aai_nf_data, self.app_conf)
+ else:
+ raise RuntimeError('Failed to get data from AAI')
+ return new_nfs
+
def _check_state_change(self, local_admin_state, new_administrative_state):
if new_administrative_state == AdministrativeState.UNLOCKED.value:
logger.info(f'Administrative State has changed from {local_admin_state} '
<groupId>org.onap.dcaegen2.services</groupId>
<artifactId>pmsh</artifactId>
<name>dcaegen2-services-pm-subscription-handler</name>
- <version>1.3.1-SNAPSHOT</version>
+ <version>1.4.0-SNAPSHOT</version>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<sonar.sources>.</sonar.sources>
setup(
name="pm_subscription_handler",
- version="1.3.1",
+ version="1.4.0",
packages=find_packages(),
author="lego@est.tech",
author_email="lego@est.tech",
"policyName":"pmsh-operational-policy",\r
"changeType":"CREATE",\r
"closedLoopControlName":"pmsh-control-loop",\r
- "ipv4Address": "1.2.3.4",\r
+ "ipAddress": "1.2.3.4",\r
"subscription":{\r
"subscriptionName":"ExtraPM-All-gNB-R2B",\r
"administrativeState":"UNLOCKED",\r
# ============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.
from unittest.mock import patch
from mod.api.db_models import SubscriptionModel
-from mod.network_function import NetworkFunction
+from mod.network_function import NetworkFunction, NetworkFunctionFilter
from mod.policy_response_handler import PolicyResponseHandler, policy_response_handle_functions
from mod.subscription import AdministrativeState, SubNfState
from tests.base_setup import BaseClassSetup
self.policy_response_handler = PolicyResponseHandler(self.mock_policy_mr_sub,
self.app_conf,
mock_app)
+ self.nfFilter = NetworkFunctionFilter(**{
+ "nfNames": [
+ "^pnf.*"
+ ],
+ "modelInvariantIDs": [
+
+ ],
+ "modelVersionIDs": [
+
+ ],
+ "modelNames": [
+
+ ]
+ })
def tearDown(self):
super().tearDown()
self.policy_response_handler._handle_response(
self.app_conf.subscription.subscriptionName,
AdministrativeState.LOCKED.value,
- self.nf.nf_name, 'success')
+ self.nf.nf_name, 'success', 'DELETED')
mock_delete.assert_called()
self.policy_response_handler._handle_response(
self.app_conf.subscription.subscriptionName,
AdministrativeState.LOCKED.value,
- self.nf.nf_name, 'failed')
+ self.nf.nf_name, 'failed','DELETE_FAILED')
mock_update_sub_nf.assert_called_with(
subscription_name=self.app_conf.subscription.subscriptionName,
status=SubNfState.DELETE_FAILED.value, nf_name=self.nf.nf_name)
self.policy_response_handler._handle_response(
self.app_conf.subscription.subscriptionName,
AdministrativeState.UNLOCKED.value,
- self.nf.nf_name, 'success')
+ self.nf.nf_name, 'success', 'CREATED')
mock_update_sub_nf.assert_called_with(
subscription_name=self.app_conf.subscription.subscriptionName,
status=SubNfState.CREATED.value, nf_name=self.nf.nf_name)
self.policy_response_handler._handle_response(
self.app_conf.subscription.subscriptionName,
AdministrativeState.UNLOCKED.value,
- self.nf.nf_name, 'failed')
+ self.nf.nf_name, 'failed', 'CREATE_FAILED')
mock_update_sub_nf.assert_called_with(
subscription_name=self.app_conf.subscription.subscriptionName,
status=SubNfState.CREATE_FAILED.value, nf_name=self.nf.nf_name)
@patch('mod.subscription.Subscription.get')
def test_poll_policy_topic_calls_methods_correct_sub(self, mock_get_sub, mock_handle_response):
response_data = ['{"name": "ResponseEvent","status": { "subscriptionName": '
- '"ExtraPM-All-gNB-R2B", "nfName": "pnf300", "message": "success" } }']
+ '"ExtraPM-All-gNB-R2B", "nfName": "pnf300", "message": "success", "changeType": "CREATED" } }']
self.mock_policy_mr_sub.get_from_topic.return_value = response_data
- mock_get_sub.return_value = SubscriptionModel(subscription_name='ExtraPM-All-gNB-R2B',
- status=AdministrativeState.UNLOCKED.value)
+ mock_get_sub.return_value = \
+ SubscriptionModel(subscription_name='ExtraPM-All-gNB-R2B',
+ nfFilter=self.nfFilter,
+ status=AdministrativeState.UNLOCKED.value)
self.policy_response_handler.poll_policy_topic()
self.mock_policy_mr_sub.get_from_topic.assert_called()
mock_handle_response.assert_called_with(self.app_conf.subscription.subscriptionName,
AdministrativeState.UNLOCKED.value, 'pnf300',
- 'success')
+ 'success', "CREATED")
@patch('mod.policy_response_handler.PolicyResponseHandler._handle_response')
@patch('mod.subscription.Subscription.get')
response_data = ['{"name": "ResponseEvent","status": { "subscriptionName": '
'"Different_Subscription", "nfName": "pnf300", "message": "success" } }']
self.mock_policy_mr_sub.get_from_topic.return_value = response_data
- mock_get_sub.return_value = SubscriptionModel(subscription_name='ExtraPM-All-gNB-R2B',
- status=AdministrativeState.UNLOCKED.value)
+ mock_get_sub.return_value = \
+ SubscriptionModel(subscription_name='ExtraPM-All-gNB-R2B',
+ nfFilter=self.nfFilter,
+ status=AdministrativeState.UNLOCKED.value)
self.policy_response_handler.poll_policy_topic()
self.mock_policy_mr_sub.get_from_topic.assert_called()
@patch('mod.subscription.Subscription.get')
def test_poll_policy_topic_exception(self, mock_get_sub, mock_logger):
self.mock_policy_mr_sub.get_from_topic.return_value = 'wrong_return'
- mock_get_sub.return_value = SubscriptionModel(subscription_name='ExtraPM-All-gNB-R2B',
- status=AdministrativeState.UNLOCKED.value)
+ mock_get_sub.return_value = \
+ SubscriptionModel(subscription_name='ExtraPM-All-gNB-R2B',
+ nfFilter=self.nfFilter,
+ status=AdministrativeState.UNLOCKED.value)
self.policy_response_handler.poll_policy_topic()
mock_logger.assert_called()
import os
from unittest.mock import patch, Mock
+from pmsh_service.mod.network_function import NetworkFunctionFilter
from requests import Session
import mod.aai_client as aai_client
mock_session_put.return_value.text = self.aai_response_data
with open(os.path.join(os.path.dirname(__file__), 'data/aai_model_info.json'), 'r') as data:
self.aai_model_data = data.read()
+ with open(os.path.join(os.path.dirname(__file__), 'data/cbs_data_1.json'), 'r') as data:
+ self.cbs_data = json.load(data)
+ self.nfFilter = NetworkFunctionFilter(**self.cbs_data['config']['pmsh_policy']['subscription']['nfFilter'])
mock_session_get.return_value.status_code = 200
mock_session_get.return_value.text = self.aai_model_data
self.mock_mr_sub = Mock()
self.assertEqual(3, len(nfs))
self.assertIsInstance(nfs[0], NetworkFunction)
+
+ def test_filter_diff_with_difference(self):
+ print(self.nfFilter)
+ networkFunction = '{"nfNames":["^pnf.*","^vnf.*"],"modelInvariantIDs": ["Extra Data"],"modelVersionIDs": ["Extra Data"],"modelNames": ["Extra Data""]}'
+ self.assertTrue(self.app_conf.subscription._filter_diff(networkFunction));
+
+ def test_filter_diff_without_difference(self):
+ networkFunction = '{"nfNames":["^pnf.*","^vnf.*"],"modelInvariantIDs": [],"modelVersionIDs": [],"modelNames": []}'
+ self.assertTrue(self.app_conf.subscription._filter_diff(networkFunction));
+ self.assertIsNotNone(self.app_conf.subscription)
+
+ def test_update_subscription_filter(self):
+ original_filter = self.app_conf.subscription.nfFilter
+ self.app_conf.subscription.nfFilter = '{"nfNames":["^pnf.*","^vnf.*"],"modelInvariantIDs": ["Extra Data"],"modelVersionIDs": ["Extra Data"],"modelNames": ["Extra Data""]}'
+ self.app_conf.subscription.update_subscription_filter()
+ updated_subscription = (self.app_conf.subscription.get())
+ self.assertTrue(updated_subscription.nfFilter == self.app_conf.subscription.nfFilter)
+ self.assertFalse(updated_subscription == original_filter)
+ print(self.app_conf.subscription.get_network_functions())
+
major=1
-minor=3
-patch=1
+minor=4
+patch=0
base_version=${major}.${minor}.${patch}
release_version=${base_version}
snapshot_version=${base_version}-SNAPSHOT
Code Review / dcaegen2 / services.git / commitdiff