diff options
Diffstat (limited to 'apps')
4 files changed, 213 insertions, 239 deletions
diff --git a/apps/placement/optimizers/conductor/remote_opt_processor.py b/apps/placement/optimizers/conductor/remote_opt_processor.py index 3d7a287..2e681be 100644 --- a/apps/placement/optimizers/conductor/remote_opt_processor.py +++ b/apps/placement/optimizers/conductor/remote_opt_processor.py @@ -17,23 +17,26 @@ # ------------------------------------------------------------------------- # -import json from jinja2 import Template +import json from requests import RequestException - import traceback -from osdf.operation.error_handling import build_json_error_body -from osdf.logging.osdf_logging import metrics_log, MH, error_log, debug_log -from osdf.adapters.conductor import conductor + from apps.license.optimizers.simple_license_allocation import license_optim +from osdf.adapters.conductor import conductor +from osdf.logging.osdf_logging import debug_log +from osdf.logging.osdf_logging import error_log +from osdf.logging.osdf_logging import metrics_log +from osdf.logging.osdf_logging import MH +from osdf.operation.error_handling import build_json_error_body from osdf.utils.interfaces import get_rest_client from osdf.utils.mdc_utils import mdc_from_json def conductor_response_processor(conductor_response, req_id, transaction_id): """Build a response object to be sent to client's callback URL from Conductor's response - This includes Conductor's placement optimization response, and required ASDC license artifacts + This includes Conductor's placement optimization response, and required ASDC license artifacts :param conductor_response: JSON response from Conductor :param raw_response: Raw HTTP response corresponding to above :param req_id: Id of a request @@ -60,13 +63,15 @@ def conductor_response_processor(conductor_response, req_id, transaction_id): try: solution['assignmentInfo'].append({"key": name_map.get(key, key), "value": value}) except KeyError: - debug_log.debug("The key[{}] is not mapped and will not be returned in assignment info".format(key)) + debug_log.debug("The key[{}] is not mapped and will not be returned in assignment info" + .format(key)) for key, value in reco[resource]['attributes'].items(): try: solution['assignmentInfo'].append({"key": name_map.get(key, key), "value": value}) except KeyError: - debug_log.debug("The key[{}] is not mapped and will not be returned in assignment info".format(key)) + debug_log.debug("The key[{}] is not mapped and will not be returned in assignment info" + .format(key)) composite_solutions.append(solution) request_status = "completed" if conductor_response['plans'][0]['status'] == "done" \ @@ -91,8 +96,8 @@ def conductor_response_processor(conductor_response, req_id, transaction_id): def conductor_no_solution_processor(conductor_response, request_id, transaction_id, template_placement_response="templates/plc_opt_response.jsont"): """Build a response object to be sent to client's callback URL from Conductor's response - This is for case where no solution is found + This is for case where no solution is found :param conductor_response: JSON response from Conductor :param raw_response: Raw HTTP response corresponding to above :param request_id: request Id associated with the client request (same as conductor response's "name") @@ -108,6 +113,7 @@ def conductor_no_solution_processor(conductor_response, request_id, transaction_ def process_placement_opt(request_json, policies, osdf_config): """Perform the work for placement optimization (e.g. call SDC artifact and make conductor request) + NOTE: there is scope to make the requests to policy asynchronous to speed up overall performance :param request_json: json content from original request :param policies: flattened policies corresponding to this request @@ -115,7 +121,7 @@ def process_placement_opt(request_json, policies, osdf_config): :param prov_status: provStatus retrieved from Subscriber policy :return: None, but make a POST to callback URL """ - + try: mdc_from_json(request_json) rc = get_rest_client(request_json, service="so") @@ -134,7 +140,11 @@ def process_placement_opt(request_json, policies, osdf_config): demands = request_json['placementInfo']['placementDemands'] request_parameters = request_json['placementInfo']['requestParameters'] service_info = request_json['serviceInfo'] - resp = conductor.request(req_info, demands, request_parameters, service_info, True, + template_fields = { + 'location_enabled': True, + 'version': '2017-10-10' + } + resp = conductor.request(req_info, demands, request_parameters, service_info, template_fields, osdf_config, policies) if resp["plans"][0].get("recommendations"): placement_response = conductor_response_processor(resp, req_id, transaction_id) @@ -162,8 +172,7 @@ def process_placement_opt(request_json, policies, osdf_config): return try: - metrics_log.info(MH.calling_back_with_body(req_id, rc.url,placement_response)) + metrics_log.info(MH.calling_back_with_body(req_id, rc.url, placement_response)) rc.request(json=placement_response, noresponse=True) - except RequestException : # can't do much here but log it and move on + except RequestException: # can't do much here but log it and move on error_log.error("Error sending asynchronous notification for {} {}".format(req_id, traceback.format_exc())) - diff --git a/apps/slice_selection/models/api/nsi_selection_request.py b/apps/slice_selection/models/api/nsi_selection_request.py index 943fa56..b395012 100644 --- a/apps/slice_selection/models/api/nsi_selection_request.py +++ b/apps/slice_selection/models/api/nsi_selection_request.py @@ -17,8 +17,14 @@ # from osdf.models.api.common import OSDFModel -from schematics.types import BaseType, StringType, URLType, IntType, BooleanType -from schematics.types.compound import ModelType, ListType, DictType +from schematics.types import BaseType +from schematics.types import BooleanType +from schematics.types.compound import DictType +from schematics.types.compound import ListType +from schematics.types.compound import ModelType +from schematics.types import IntType +from schematics.types import StringType +from schematics.types import URLType class RequestInfo(OSDFModel): @@ -50,7 +56,7 @@ class NSISelectionAPI(OSDFModel): """Request for nsi selection (specific to optimization and additional metadata""" requestInfo = ModelType(RequestInfo, required=True) NSTInfo = ModelType(NxTInfo, required=True) - NSSTInfo = ListType(ModelType(NxTInfo), required=True) + NSSTInfo = ListType(ModelType(NxTInfo), required=False) serviceProfile = DictType(BaseType, required=True) subnetCapabilities = ListType(ModelType(SubnetCapability), required=True) preferReuse = BooleanType() diff --git a/apps/slice_selection/optimizers/conductor/remote_opt_processor.py b/apps/slice_selection/optimizers/conductor/remote_opt_processor.py index 40638fc..c1c6980 100644 --- a/apps/slice_selection/optimizers/conductor/remote_opt_processor.py +++ b/apps/slice_selection/optimizers/conductor/remote_opt_processor.py @@ -20,92 +20,106 @@ Module for processing slice selection request """ -import json -import traceback from requests import RequestException +from threading import Thread +import traceback -from apps.slice_selection.optimizers.conductor.response_processor \ - import conductor_response_processor, conductor_error_response_processor, solution_with_only_slice_profile, get_nsi_selection_response +from apps.slice_selection.optimizers.conductor.response_processor import ResponseProcessor from osdf.adapters.conductor import conductor from osdf.adapters.policy.interface import get_policies -from osdf.adapters.policy.utils import group_policies_gen -from osdf.logging.osdf_logging import error_log, debug_log +from osdf.logging.osdf_logging import debug_log +from osdf.logging.osdf_logging import error_log +from osdf.utils.interfaces import get_rest_client from osdf.utils.mdc_utils import mdc_from_json -def process_nsi_selection_opt(request_json, osdf_config): - """Process the nsi selection request from API layer - :param request_json: api request - :param policies: flattened policies corresponding to this request - :param osdf_config: configuration specific to OSDF app - :return: response as a dictionary - """ - req_info = request_json['requestInfo'] - try: - mdc_from_json(request_json) - - overall_recommendations = dict() - nst_info_map = dict() - new_nsi_solutions = list() - for nst_info in request_json["NSTInfoList"]: - nst_name = nst_info["modelName"] - nst_info_map[nst_name] = {"NSTName": nst_name, - "UUID": nst_info["modelVersionId"], - "invariantUUID": nst_info["modelInvariantId"]} - - if request_json["serviceProfile"]["resourceSharingLevel"] == "non-shared": - new_nsi_solution = solution_with_only_slice_profile(request_json['serviceProfile'], nst_info_map.get(nst_name)) - new_nsi_solutions.append(new_nsi_solution) +class SliceSelectionOptimizer(Thread): + def __init__(self, osdf_config, slice_config, request_json, model_type): + self.osdf_config = osdf_config + self.slice_config = slice_config + self.request_json = request_json + self.model_type = model_type + self.response_processor = ResponseProcessor(request_json['requestInfo'], slice_config) + + def run(self): + self.process_slice_selection_opt() + + def process_slice_selection_opt(self): + """Process the slice selection request from the API layer""" + req_info = self.request_json['requestInfo'] + rc = get_rest_client(self.request_json, service='so') + + try: + if self.model_type == 'NSSI' \ + and self.request_json['sliceProfile'].get('resourceSharingLevel', "") == 'not-shared': + final_response = self.response_processor.get_slice_selection_response([]) + else: - policy_request_json = request_json.copy() - policy_request_json['serviceInfo']['serviceName'] = nst_name - policies = get_policies(policy_request_json, "slice_selection") - - demands = get_slice_demands(nst_name, policies, osdf_config.core) - - request_parameters = request_json.get('serviceProfile',{}) - service_info = {} - req_info['numSolutions'] = 'all' - try: - resp = conductor.request(req_info, demands, request_parameters, service_info, False, - osdf_config, policies) - except RequestException as e: - resp = e.response.json() - error = resp['plans'][0]['message'] - error_log.error('Error from conductor {}'.format(error)) - debug_log.debug("Response from conductor {}".format(str(resp))) - overall_recommendations[nst_name] = resp["plans"][0].get("recommendations") - - if request_json["serviceProfile"]["resourceSharingLevel"] == "non-shared": - solutions = dict() - solutions['newNSISolutions'] = new_nsi_solutions - solutions['sharedNSISolutions'] = [] - return get_nsi_selection_response(req_info, solutions) - else: - return conductor_response_processor(overall_recommendations, nst_info_map, req_info, request_json["serviceProfile"]) - except Exception as ex: - error_log.error("Error for {} {}".format(req_info.get('requestId'), - traceback.format_exc())) - error_message = str(ex) - return conductor_error_response_processor(req_info, error_message) - - -def get_slice_demands(model_name, policies, config): - """ - :param model_name: model name of the slice - :param policies: flattened polcies corresponding to the request - :param config: configuration specific to OSDF app - :return: list of demands for the request - """ - group_policies = group_policies_gen(policies, config) - subscriber_policy_list = group_policies["onap.policies.optimization.service.SubscriberPolicy"] - slice_demands = list() - for subscriber_policy in subscriber_policy_list: - policy_properties = subscriber_policy[list(subscriber_policy.keys())[0]]['properties'] - if model_name in policy_properties["services"]: - for subnet in policy_properties["properties"]["subscriberName"]: - slice_demand = dict() - slice_demand["resourceModuleName"] = subnet - slice_demand['resourceModelInfo'] = {} - slice_demands.append(slice_demand) - return slice_demands + final_response = self.do_slice_selection() + + except Exception as ex: + error_log.error("Error for {} {}".format(req_info.get('requestId'), + traceback.format_exc())) + error_message = str(ex) + final_response = self.response_processor.process_error_response(error_message) + + try: + rc.request(json=final_response, noresponse=True) + except RequestException: + error_log.error("Error sending asynchronous notification for {} {}".format(req_info['request_id'], + traceback.format_exc())) + + def do_slice_selection(self): + req_info = self.request_json['requestInfo'] + app_info = self.slice_config['app_info'][self.model_type] + mdc_from_json(self.request_json) + requirements = self.request_json.get(app_info['requirements_field'], {}) + model_info = self.request_json.get(app_info['model_info']) + model_name = model_info['name'] + policies = self.get_app_policies(model_name, app_info['app_name']) + request_parameters = self.get_request_parameters(requirements) + + demands = [ + { + "resourceModuleName": model_name, + "resourceModelInfo": {} + } + ] + + try: + template_fields = { + 'location_enabled': False, + 'version': '2020-08-13' + } + resp = conductor.request(req_info, demands, request_parameters, {}, template_fields, + self.osdf_config, policies) + except RequestException as e: + resp = e.response.json() + error = resp['plans'][0]['message'] + error_log.error('Error from conductor {}'.format(error)) + return self.response_processor.process_error_response(error) + + debug_log.debug("Response from conductor {}".format(str(resp))) + recommendations = resp["plans"][0].get("recommendations") + subnets = [subnet['domainType'] for subnet in self.request_json['subnetCapabilities']] \ + if self.request_json.get('subnetCapabilities') else [] + return self.response_processor.process_response(recommendations, model_info, subnets) + + def get_request_parameters(self, requirements): + camel_to_snake = self.slice_config['attribute_mapping']['camel_to_snake'] + request_params = {camel_to_snake[key]: value for key, value in requirements.items()} + subnet_capabilities = self.request_json.get('subnetCapabilities') + if subnet_capabilities: + for subnet_capability in subnet_capabilities: + domain_type = f"{subnet_capability['domainType'].lower().replace('-', '_')}_" + capability_details = subnet_capability['capabilityDetails'] + for key, value in capability_details.items(): + request_params[f"{domain_type}{camel_to_snake[key]}"] = value + return request_params + + def get_app_policies(self, model_name, app_name): + policy_request_json = self.request_json.copy() + policy_request_json['serviceInfo'] = {'serviceName': model_name} + if 'preferReuse' in self.request_json: + policy_request_json['preferReuse'] = "reuse" if self.request_json['preferReuse'] else "create_new" + return get_policies(policy_request_json, app_name) diff --git a/apps/slice_selection/optimizers/conductor/response_processor.py b/apps/slice_selection/optimizers/conductor/response_processor.py index a9bdad0..71a350f 100644 --- a/apps/slice_selection/optimizers/conductor/response_processor.py +++ b/apps/slice_selection/optimizers/conductor/response_processor.py @@ -20,144 +20,89 @@ Module for processing response from conductor for slice selection """ -from osdf.logging.osdf_logging import debug_log - - -SLICE_PROFILE_FIELDS = {"latency":"latency", "max_number_of_ues":"maxNumberOfUEs", "coverage_area_ta_list": "coverageAreaTAList", - "ue_mobility_level":"uEMobilityLevel", "resource_sharing_level":"resourceSharingLevel", "exp_data_rate_ul": "expDataRateUL", - "exp_data_rate_dl":"expDataRateDL", "area_traffic_cap_ul":"areaTrafficCapUL", "area_traffic_cap_dl": "areaTrafficCapDL", - "activity_factor":"activityFactor", "e2e_latency":"e2eLatency", "jitter":"jitter", "survival_time": "survivalTime", - "exp_data_rate":"expDataRate", "payload_size":"payloadSize", "traffic_density":"trafficDensity", "conn_density":"connDensity", - "reliability":"reliability", "service_area_dimension":"serviceAreaDimension", "cs_availability": "csAvailability"} - - -def conductor_response_processor(overall_recommendations, nst_info_map, request_info, service_profile): - """Process conductor response to form the response for the API request - :param overall_recommendations: recommendations from conductor - :param nst_info_map: NST info from the request - :param request_info: request info - :return: response json as a dictionary - """ - shared_nsi_solutions = list() - new_nsi_solutions = list() - for nst_name, recommendations in overall_recommendations.items(): - if not (recommendations): - new_nsi_solution = solution_with_only_slice_profile(service_profile, nst_info_map.get(nst_name)) - new_nsi_solutions.append(new_nsi_solution) - continue - - for recommendation in recommendations: - nsi_set = set(values['candidate']['nsi_id'] for key, values in recommendation.items()) - if len(nsi_set) == 1: - nsi_id = nsi_set.pop() - candidate = list(recommendation.values())[0]['candidate'] - debug_log.debug("The NSSIs in the solution belongs to the same NSI {}" - .format(nsi_id)) - shared_nsi_solution = dict() - shared_nsi_solution["NSIId"] = nsi_id - shared_nsi_solution["NSIName"] = candidate.get('nsi_name') - shared_nsi_solution["UUID"] = candidate.get('nsi_model_version_id') - shared_nsi_solution["invariantUUID"] = candidate.get('nsi_model_invariant_id') - - nssi_info_list = get_nssi_solutions(recommendation) - nssis = list() - for nssi_info in nssi_info_list: - nssi = dict() - nssi["NSSIId"] = nssi_info.get("NSSISolution").get("NSSIId") - nssi["NSSIName"] = nssi_info.get("NSSISolution").get("NSSIName") - nssi["UUID"] = "" - nssi["invariantUUID"] = "" - nssi_info.get("sliceProfile").update({"domainType":"cn"}) - nssi["sliceProfile"] = [nssi_info.get("sliceProfile")] - nssis.append(nssi) - - shared_nsi_solution["NSSIs"] = nssis - shared_nsi_solutions.append(shared_nsi_solution) - else: - nssi_solutions = get_nssi_solutions(recommendation) - new_nsi_solution = dict() - new_nsi_solution['matchLevel'] = "" - new_nsi_solution['NSTInfo'] = nst_info_map.get(nst_name) - new_nsi_solution['NSSISolutions'] = nssi_solutions - new_nsi_solutions.append(new_nsi_solution) - - solutions = dict() - solutions['sharedNSISolutions'] = shared_nsi_solutions - solutions['newNSISolutions'] = new_nsi_solutions - return get_nsi_selection_response(request_info, solutions) - - -def solution_with_only_slice_profile(service_profile, nst_info): - nssi_solutions = get_slice_profile_from_service_profile(service_profile) - new_nsi_solution = dict() - new_nsi_solution['matchLevel'] = "" - new_nsi_solution['NSTInfo'] = nst_info - new_nsi_solution['NSSISolutions'] = nssi_solutions - return new_nsi_solution - -def conductor_error_response_processor(request_info, error_message): - """Form response message from the error message - :param request_info: request info - :param error_message: error message while processing the request - :return: response json as dictionary - """ - return {'requestId': request_info['requestId'], - 'transactionId': request_info['transactionId'], - 'requestStatus': 'error', - 'statusMessage': error_message} - - -def get_slice_profile_from_service_profile(service_profile): - nssi_solutions = list() - service_profile["domainType"] = "cn" - nssi_solution = {"sliceProfile": service_profile} - nssi_solutions.append(nssi_solution) - return nssi_solutions - - -def get_nssi_solutions(recommendation): - """Get nssi solutions from recommendation - :param recommendation: recommendation from conductor - :return: new nssi solutions list - """ - nssi_solutions = list() - - for nsst_name, nsst_rec in recommendation.items(): - candidate = nsst_rec['candidate'] - nssi_info, slice_profile = get_solution_from_candidate(candidate) - nsst_info = {"NSSTName": nsst_name} - nssi_solution = {"sliceProfile": slice_profile, - "NSSTInfo": nsst_info, - "NSSISolution": nssi_info} - nssi_solutions.append(nssi_solution) - return nssi_solutions - - -def get_solution_from_candidate(candidate): - """Get nssi info from candidate - :param candidate: Candidate from the recommendation - :return: nssi_info and slice profile derived from candidate - """ - slice_profile = dict() - nssi_info = {"NSSIName": candidate['instance_name'], - "NSSIId": candidate['candidate_id']} - - for field in SLICE_PROFILE_FIELDS: - if candidate[field]: - slice_profile[SLICE_PROFILE_FIELDS[field]] = candidate[field] - - return nssi_info, slice_profile - - -def get_nsi_selection_response(request_info, solutions): - """Get NSI selection response from final solution - :param request_info: request info - :param solutions: final solutions - :return: NSI selection response to send back as dictionary - """ - return {'requestId': request_info['requestId'], - 'transactionId': request_info['transactionId'], - 'requestStatus': 'completed', - 'statusMessage': '', - 'solutions': solutions} - +import re + + +class ResponseProcessor(object): + def __init__(self, request_info, slice_config): + self.request_info = request_info + self.slice_config = slice_config + + def process_response(self, recommendations, model_info, subnets): + """Process conductor response to form the response for the API request + + :param recommendations: recommendations from conductor + :param model_info: model info from the request + :param subnets: list of subnets + :return: response json as a dictionary + """ + if not recommendations: + return self.get_slice_selection_response([]) + model_name = model_info['name'] + solutions = [self.get_solution_from_candidate(rec[model_name]['candidate'], model_info, subnets) + for rec in recommendations] + return self.get_slice_selection_response(solutions) + + def get_solution_from_candidate(self, candidate, model_info, subnets): + if candidate['inventory_type'] == 'nssi': + return { + 'UUID': model_info['UUID'], + 'invariantUUID': model_info['invariantUUID'], + 'NSSIName': candidate['instance_name'], + 'NSSIId': candidate['instance_id'] + } + + elif candidate['inventory_type'] == 'nsi': + return { + 'existingNSI': True, + 'sharedNSISolution': { + 'UUID': model_info['UUID'], + 'invariantUUID': model_info['invariantUUID'], + 'NSIName': candidate['instance_name'], + 'NSIId': candidate['instance_id'] + } + } + + elif candidate['inventory_type'] == 'slice_profiles': + return { + 'existingNSI': False, + 'newNSISolution': { + 'slice_profiles': self.get_slice_profiles_from_candidate(candidate, subnets) + } + } + + def get_slice_profiles_from_candidate(self, candidate, subnets): + slice_profiles = [] + for subnet in subnets: + slice_profile = {self.get_profile_attribute(k, subnet): v for k, v in candidate.items() + if k.startswith(subnet)} + slice_profile['domainType'] = subnet + slice_profiles.append(slice_profile) + return slice_profiles + + def get_profile_attribute(self, attribute, subnet): + snake_to_camel = self.slice_config['attribute_mapping']['snake_to_camel'] + return snake_to_camel[re.sub(f'^{subnet}_', '', attribute)] + + def process_error_response(self, error_message): + """Form response message from the error message + + :param error_message: error message while processing the request + :return: response json as dictionary + """ + return {'requestId': self.request_info['requestId'], + 'transactionId': self.request_info['transactionId'], + 'requestStatus': 'error', + 'statusMessage': error_message} + + def get_slice_selection_response(self, solutions): + """Get NSI selection response from final solution + + :param solutions: final solutions + :return: NSI selection response to send back as dictionary + """ + return {'requestId': self.request_info['requestId'], + 'transactionId': self.request_info['transactionId'], + 'requestStatus': 'completed', + 'statusMessage': '', + 'solutions': solutions} |