"""Service module.""" # Copyright 2022 Orange, Deutsche Telekom AG # # 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. import base64 import pathlib as Path import time from dataclasses import dataclass, field from enum import Enum from io import BytesIO, TextIOWrapper from os import makedirs from typing import Dict, List, Callable, Iterator, Optional, Type, Union, Any, BinaryIO from zipfile import ZipFile, BadZipFile import oyaml as yaml from requests import Response import onapsdk.constants as const from onapsdk.exceptions import (ParameterError, RequestError, ResourceNotFound, StatusError, ValidationError) from onapsdk.sdc.category_management import ServiceCategory from onapsdk.sdc.properties import NestedInput, Property from onapsdk.sdc.sdc_resource import SdcResource from onapsdk.utils.configuration import components_needing_distribution from onapsdk.utils.headers_creator import headers_sdc_creator, headers_sdc_artifact_upload from onapsdk.utils.jinja import jinja_env @dataclass class VfModule: # pylint: disable=too-many-instance-attributes """VfModule dataclass.""" name: str group_type: str model_name: str model_version_id: str model_invariant_uuid: str model_version: str model_customization_id: str properties: Iterator[Property] @dataclass class NodeTemplate: # pylint: disable=too-many-instance-attributes """Node template dataclass. Base class for Vnf, Pnf and Network classes. """ name: str node_template_type: str model_name: str model_version_id: str model_invariant_id: str model_version: str model_customization_id: str model_instance_name: str component: "Component" @property def properties(self) -> Iterator["Property"]: """Node template properties. Returns: Iterator[Property]: Node template properties iterator """ return self.component.properties @dataclass class Vnf(NodeTemplate): """Vnf dataclass.""" vf_modules: List[VfModule] = field(default_factory=list) @dataclass class Pnf(NodeTemplate): """Pnf dataclass.""" class Network(NodeTemplate): # pylint: disable=too-few-public-methods """Network dataclass.""" class ServiceInstantiationType(Enum): """Service instantiation type enum class. Service can be instantiated using `A-la-carte` or `Macro` flow. It has to be determined during design time. That class stores these two values to set during initialization. """ A_LA_CARTE = "A-la-carte" MACRO = "Macro" class Service(SdcResource): # pylint: disable=too-many-instance-attributes, too-many-public-methods """ ONAP Service Object used for SDC operations. Attributes: name (str): the name of the service. Defaults to "ONAP-test-Service". identifier (str): the unique ID of the service from SDC. status (str): the status of the service from SDC. version (str): the version ID of the service from SDC. uuid (str): the UUID of the Service (which is different from identifier, don't ask why...) distribution_status (str): the status of distribution in the different ONAP parts. distribution_id (str): the ID of the distribution when service is distributed. distributed (bool): True if the service is distributed unique_identifier (str): Yet Another ID, just to puzzle us... """ SERVICE_PATH = "services" def __init__(self, name: str = None, version: str = None, sdc_values: Dict[str, str] = None, # pylint: disable=too-many-arguments resources: List[SdcResource] = None, properties: List[Property] = None, inputs: List[Union[Property, NestedInput]] = None, instantiation_type: Optional[ServiceInstantiationType] = \ None, category: str = None, role: str = "", function: str = "", service_type: str = ""): """ Initialize service object. Args: name (str, optional): the name of the service version (str, optional): the version of the service sdc_values (Dict[str, str], optional): dictionary of values returned by SDC resources (List[SdcResource], optional): list of SDC resources properties (List[Property], optional): list of properties to add to service. None by default. inputs (List[Union[Property, NestedInput]], optional): list of inputs to declare for service. It can be both Property or NestedInput object. None by default. instantiation_type (ServiceInstantiationType, optional): service instantiation type. ServiceInstantiationType.A_LA_CARTE by default category (str, optional): service category name role (str, optional): service role function (str, optional): service function. Empty by default service_type (str, optional): service type. Empty by default """ super().__init__(sdc_values=sdc_values, version=version, properties=properties, inputs=inputs, category=category) self.name: str = name or "ONAP-test-Service" self.distribution_status = None self.category_name: str = category self.role: str = role self.function: str = function self.service_type: str = service_type if sdc_values: self.distribution_status = sdc_values['distributionStatus'] self.category_name = sdc_values["category"] self.resources = resources or [] self._instantiation_type: Optional[ServiceInstantiationType] = instantiation_type self._distribution_id: str = None self._distributed: bool = False self._resource_type: str = "services" self._tosca_model: bytes = None self._tosca_template: str = None self._vnfs: list = None self._pnfs: list = None self._networks: list = None self._vf_modules: list = None @classmethod def get_by_unique_uuid(cls, unique_uuid: str) -> "Service": """Get the service model using unique uuid. Returns: Service: object with provided unique_uuid Raises: ResourceNotFound: No service with given unique_uuid exists """ services: List["Service"] = cls.get_all() for service in services: if service.unique_uuid == unique_uuid: return service raise ResourceNotFound("Service with given unique uuid doesn't exist") def onboard(self) -> None: """Onboard the Service in SDC. Raises: StatusError: service has an invalid status ParameterError: no resources, no properties for service in DRAFT status """ # first Lines are equivalent for all onboard functions but it's more # readable if not self.status: # equivalent step as in onboard-function in sdc_resource self.create() time.sleep(self._time_wait) self.onboard() elif self.status == const.DRAFT: if not any([self.resources, self._properties_to_add]): raise ParameterError("No resources nor properties were given") self.declare_resources_and_properties() self.checkin() time.sleep(self._time_wait) self.onboard() elif self.status == const.CHECKED_IN: self.certify() time.sleep(self._time_wait) self.onboard() elif self.status == const.CERTIFIED: self.distribute() self.onboard() elif self.status == const.DISTRIBUTED: self._logger.info("Service %s onboarded", self.name) else: self._logger.error("Service has invalid status: %s", self.status) raise StatusError(self.status) @property def distribution_id(self) -> str: """Return and lazy load the distribution_id.""" if not self._distribution_id: self.load_metadata() return self._distribution_id @distribution_id.setter def distribution_id(self, value: str) -> None: """Set value for distribution_id.""" self._distribution_id = value @property def distributed(self) -> bool: """Return and lazy load the distributed state.""" if not self._distributed: self._check_distributed() return self._distributed @property def tosca_template(self) -> str: """Service tosca template file. Get tosca template from service tosca model bytes. Returns: str: Tosca template file """ if not self._tosca_template and self.tosca_model: self._unzip_csar_file(BytesIO(self.tosca_model), self._load_tosca_template) return self._tosca_template @property def tosca_model(self) -> bytes: """Service's tosca model file. Send request to get service TOSCA model, Returns: bytes: TOSCA model file bytes """ if not self._tosca_model: url = "{}/services/{}/toscaModel".format(self._base_url(), self.identifier) headers = self.headers.copy() headers["Accept"] = "application/octet-stream" self._tosca_model = self.send_message( "GET", "Download Tosca Model for {}".format(self.name), url, headers=headers).content return self._tosca_model def create_node_template(self, node_template_type: Type[NodeTemplate], component: "Component") -> NodeTemplate: """Create a node template type object. The base of the all node template types objects (Vnf, Pnf, Network) is the same. The difference is only for the Vnf which can have vf modules associated with. Vf modules could have "vf_module_label" property with"base_template_dummy_ignore" value. These vf modules should be ignored/ Args: node_template_type (Type[NodeTemplate]): Node template class type component (Component): Component on which base node template object should be created Returns: NodeTemplate: Node template object created from component """ node_template: NodeTemplate = node_template_type( name=component.name, node_template_type=component.tosca_component_name, model_name=component.component_name, model_version_id=component.sdc_resource.identifier, model_invariant_id=component.sdc_resource.unique_uuid, model_version=component.sdc_resource.version, model_customization_id=component.customization_uuid, model_instance_name=self.name, component=component ) if node_template_type is Vnf: if component.group_instances: for vf_module in component.group_instances: if not any([property_def["name"] == "vf_module_label"] and \ property_def["value"] == "base_template_dummy_ignore" for \ property_def in vf_module["properties"]): node_template.vf_modules.append(VfModule( name=vf_module["name"], group_type=vf_module["type"], model_name=vf_module["groupName"], model_version_id=vf_module["groupUUID"], model_invariant_uuid=vf_module["invariantUUID"], model_version=vf_module["version"], model_customization_id=vf_module["customizationUUID"], properties=( Property( name=property_def["name"], property_type=property_def["type"], description=property_def["description"], value=property_def["value"] ) for property_def in vf_module["properties"] \ if property_def["value"] and not ( property_def["name"] == "vf_module_label" and \ property_def["value"] == "base_template_dummy_ignore" ) ) )) return node_template def __has_component_type(self, origin_type: str) -> bool: """Check if any of Service's component type is provided origin type. In template generation is checked if Service has some types of components, based on that blocks are added to the request template. It's not the best option to get all components to check if at least one with given type exists for conditional statement. Args: origin_type (str): Type to check if any component exists. Returns: bool: True if service has at least one component with given origin type, False otherwise """ return any((component.origin_type == origin_type for component in self.components)) @property def has_vnfs(self) -> bool: """Check if service has at least one VF component.""" return self.__has_component_type("VF") @property def has_pnfs(self) -> bool: """Check if service has at least one PNF component.""" return self.__has_component_type("PNF") @property def has_vls(self) -> bool: """Check if service has at least one VL component.""" return self.__has_component_type("VL") @property def vnfs(self) -> Iterator[Vnf]: """Service Vnfs. Load VNFs from components generator. It creates a generator of the vf modules as well, but without vf modules which has "vf_module_label" property value equal to "base_template_dummy_ignore". Returns: Iterator[Vnf]: Vnf objects iterator """ for component in self.components: if component.origin_type == "VF": yield self.create_node_template(Vnf, component) @property def pnfs(self) -> Iterator[Pnf]: """Service Pnfs. Load PNFS from components generator. Returns: Iterator[Pnf]: Pnf objects generator """ for component in self.components: if component.origin_type == "PNF": yield self.create_node_template(Pnf, component) @property def networks(self) -> Iterator[Network]: """Service networks. Load networks from service's components generator. Returns: Iterator[Network]: Network objects generator """ for component in self.components: if component.origin_type == "VL": yield self.create_node_template(Network, component) @property def deployment_artifacts_url(self) -> str: """Deployment artifacts url. Returns: str: SdcResource Deployment artifacts url """ return (f"{self._base_create_url()}/services/" f"{self.unique_identifier}/filteredDataByParams?include=deploymentArtifacts") @property def add_deployment_artifacts_url(self) -> str: """Add deployment artifacts url. Returns: str: Url used to add deployment artifacts """ return (f"{self._base_create_url()}/services/" f"{self.unique_identifier}/artifacts") @property def properties_url(self) -> str: """Properties url. Returns: str: SdcResource properties url """ return (f"{self._base_create_url()}/services/" f"{self.unique_identifier}/filteredDataByParams?include=properties") @property def metadata_url(self) -> str: """Metadata url. Returns: str: Service metadata url """ return (f"{self._base_create_url()}/services/" f"{self.unique_identifier}/filteredDataByParams?include=metadata") @property def resource_inputs_url(self) -> str: """Service inputs url. Returns: str: Service inputs url """ return (f"{self._base_create_url()}/services/" f"{self.unique_identifier}") @property def set_input_default_value_url(self) -> str: """Url to set input default value. Returns: str: SDC API url used to set input default value """ return (f"{self._base_create_url()}/services/" f"{self.unique_identifier}/update/inputs") @property def origin_type(self) -> str: """Service origin type. Value needed for composition. It's used for adding SDC resource as an another SDC resource component. For Service that value has to be set to "ServiceProxy". Returns: str: Service resource origin type """ return "ServiceProxy" @property def instantiation_type(self) -> ServiceInstantiationType: """Service instantiation type. One of `ServiceInstantiationType` enum value. Returns: ServiceInstantiationType: Service instantiation type """ if not self._instantiation_type: if not self.created(): self._instantiation_type = ServiceInstantiationType.A_LA_CARTE else: response: str = self.send_message_json("GET", f"Get service {self.name} metadata", self.metadata_url)["metadata"]\ ["instantiationType"] self._instantiation_type = ServiceInstantiationType(response) return self._instantiation_type def create(self) -> None: """Create the Service in SDC if not already existing.""" self._create("service_create.json.j2", name=self.name, instantiation_type=self.instantiation_type.value, category=self.category, role=self.role, service_type=self.service_type, function=self.function) def declare_resources_and_properties(self) -> None: """Delcare resources and properties. It declares also inputs. """ for resource in self.resources: self.add_resource(resource) for property_to_add in self._properties_to_add: self.add_property(property_to_add) for input_to_add in self._inputs_to_add: self.declare_input(input_to_add) def checkin(self) -> None: """Checkin Service.""" self._verify_lcm_to_sdc(const.DRAFT, const.CHECKIN) def submit(self) -> None: """Really submit the SDC Service.""" self._verify_lcm_to_sdc(const.CHECKED_IN, const.SUBMIT_FOR_TESTING) def start_certification(self) -> None: """Start Certification on Service.""" headers = headers_sdc_creator(SdcResource.headers) self._verify_lcm_to_sdc(const.CHECKED_IN, const.START_CERTIFICATION, headers=headers) def certify(self) -> None: """Certify Service in SDC.""" headers = headers_sdc_creator(SdcResource.headers) self._verify_lcm_to_sdc(const.CHECKED_IN, const.CERTIFY, headers=headers) def approve(self) -> None: """Approve Service in SDC.""" headers = headers_sdc_creator(SdcResource.headers) self._verify_approve_to_sdc(const.CERTIFIED, const.APPROVE, headers=headers) def distribute(self) -> None: """Apptove Service in SDC.""" headers = headers_sdc_creator(SdcResource.headers) self._verify_distribute_to_sdc(const.CERTIFIED, const.DISTRIBUTE, headers=headers) def redistribute(self) -> None: """Apptove Service in SDC.""" headers = headers_sdc_creator(SdcResource.headers) self._verify_distribute_to_sdc(const.DISTRIBUTED, const.DISTRIBUTE, headers=headers) def get_tosca(self, paths) -> None: """Get Service tosca files and save it.""" url = "{}/services/{}/toscaModel".format(self._base_url(), self.identifier) headers = self.headers.copy() headers["Accept"] = "application/octet-stream" result = self.send_message("GET", "Download Tosca Model for {}".format( self.name), url, headers=headers) if result: self._create_tosca_file(paths, result) def _create_tosca_file(self, paths, result: Response) -> None: """Create Service Tosca files from HTTP response.""" csar_filename = "service-{}-csar.csar".format(self.name) makedirs(paths, exist_ok=True) with open((paths + csar_filename), 'wb') as csar_file: for chunk in result.iter_content(chunk_size=128): csar_file.write(chunk) try: self._unzip_csar_file(paths + csar_filename, self._write_csar_file) except BadZipFile as exc: self._logger.exception(exc) def _check_distributed(self) -> bool: """Check if service is distributed and update status accordingly.""" url = "{}/services/distribution/{}".format(self._base_create_url(), self.distribution_id) headers = headers_sdc_creator(SdcResource.headers) status = {} for component in components_needing_distribution(): status[component] = False try: result = self.send_message_json("GET", "Check distribution for {}".format( self.name), url, headers=headers) except ResourceNotFound: msg = f"No distributions found for {self.name} of {self.__class__.__name__}." self._logger.debug(msg) else: status = self._update_components_status(status, result) for state in status.values(): if not state: self._distributed = False return self._distributed = True def _update_components_status(self, status: Dict[str, bool], result: Response) -> Dict[str, bool]: """Update components distribution status.""" distrib_list = result['distributionStatusList'] self._logger.debug("[SDC][Get Distribution] distrib_list = %s", distrib_list) for elt in distrib_list: status = self._parse_components_status(status, elt) return status def _parse_components_status(self, status: Dict[str, bool], element: Dict[str, Any]) -> Dict[str, bool]: """Parse components distribution status.""" for key in status: if ((key in element['omfComponentID']) and (const.DOWNLOAD_OK in element['status'])): status[key] = True self._logger.info(("[SDC][Get Distribution] Service " "distributed in %s"), key) return status def load_metadata(self) -> None: """Load Metada of Service and retrieve informations.""" url = "{}/services/{}/distribution".format(self._base_create_url(), self.identifier) headers = headers_sdc_creator(SdcResource.headers) result = self.send_message_json("GET", "Get Metadata for {}".format( self.name), url, headers=headers) if ('distributionStatusOfServiceList' in result and len(result['distributionStatusOfServiceList']) > 0): # API changed and the latest distribution is not added to the end # of distributions list but inserted as the first one. dist_status = result['distributionStatusOfServiceList'][0] self._distribution_id = dist_status['distributionID'] @classmethod def _get_all_url(cls) -> str: """ Get URL for all elements in SDC. Returns: str: the url """ return "{}/{}".format(cls._base_url(), cls._sdc_path()) def _really_submit(self) -> None: """Really submit the SDC Service in order to enable it.""" result = self._action_to_sdc(const.CERTIFY, action_type="lifecycleState") if result: self.load() def _specific_copy(self, obj: 'Service') -> None: """ Copy specific properties from object. Args: obj (Service): the object to "copy" """ super()._specific_copy(obj) self.category_name = obj.category_name self.role = obj.role def _verify_distribute_to_sdc(self, desired_status: str, desired_action: str, **kwargs) -> None: self._verify_action_to_sdc(desired_status, desired_action, "distribution", **kwargs) def _verify_approve_to_sdc(self, desired_status: str, desired_action: str, **kwargs) -> None: self._verify_action_to_sdc(desired_status, desired_action, "distribution-state", **kwargs) def _verify_lcm_to_sdc(self, desired_status: str, desired_action: str, **kwargs) -> None: self._verify_action_to_sdc(desired_status, desired_action, "lifecycleState", **kwargs) def _verify_action_to_sdc(self, desired_status: str, desired_action: str, action_type: str, **kwargs) -> None: """ Verify action to SDC. Verify that object is in right state before launching the action on SDC. Raises: StatusError: if current status is not the desired status. Args: desired_status (str): the status the object should be desired_action (str): the action we want to perform action_type (str): the type of action ('distribution-state' or 'lifecycleState') **kwargs: any specific stuff to give to requests """ self._logger.info("attempting to %s Service %s in SDC", desired_action, self.name) if self.status == desired_status and self.created(): self._action_to_sdc(desired_action, action_type=action_type, **kwargs) self.load() elif not self.created(): self._logger.warning("Service %s in SDC is not created", self.name) elif self.status != desired_status: msg = (f"Service {self.name} in SDC is in status {self.status} " f"and it should be in status {desired_status}") raise StatusError(msg) @staticmethod def _unzip_csar_file(zip_file: Union[str, BytesIO], function: Callable[[str, TextIOWrapper], None]) -> None: """ Unzip Csar File and perform an action on the file. Raises: ValidationError: CSAR file has no service template """ folder = "Definitions" prefix = "service-" suffix = "-template.yml" with ZipFile(zip_file) as myzip: service_template = None for name in myzip.namelist(): if (name[-13:] == suffix and name[:20] == f"{folder}/{prefix}"): service_template = name if not service_template: msg = (f"CSAR file has no service template. " f"Valid path: {folder}/{prefix}*{suffix}") raise ValidationError(msg) with myzip.open(service_template) as template_file: function(service_template, template_file) @staticmethod def _write_csar_file(service_template: str, template_file: TextIOWrapper) -> None: """Write service temple into a file.""" with open(service_template[12:], 'wb') as file: file.write(template_file.read()) # _service_template is not used but function generation is generic # pylint: disable-unused-argument def _load_tosca_template(self, _service_template: str, template_file: TextIOWrapper) -> None: """Load Tosca template.""" self._tosca_template = yaml.safe_load(template_file.read()) @classmethod def _sdc_path(cls) -> None: """Give back the end of SDC path.""" return cls.SERVICE_PATH def get_nf_unique_id(self, nf_name: str) -> str: """ Get nf (network function) uniqueID. Get nf uniqueID from service nf in sdc. Args: nf_name (str): the nf from which we extract the unique ID Returns: the nf unique ID Raises: ResourceNotFound: Couldn't find NF by name. """ url = f"{self._base_create_url()}/services/{self.unique_identifier}" request_return = self.send_message_json('GET', 'Get nf unique ID', url) for instance in filter(lambda x: x["componentName"] == nf_name, request_return["componentInstances"]): return instance["uniqueId"] raise ResourceNotFound(f"NF '{nf_name}'") def add_artifact_to_vf(self, vnf_name: str, artifact_type: str, artifact_name: str, artifact: BinaryIO = None): """ Add artifact to vf. Add artifact to vf using payload data. Raises: RequestError: file upload (POST request) for an artifact fails. Args: vnf_name (str): the vnf which we want to add the artifact artifact_type (str): all SDC artifact types are supported (DCAE_*, HEAT_*, ...) artifact_name (str): the artifact file name including its extension artifact (str): binary data to upload """ missing_identifier = self.get_nf_unique_id(vnf_name) url = (f"{self._base_create_url()}/services/{self.unique_identifier}/" f"resourceInstance/{missing_identifier}/artifacts") template = jinja_env().get_template("add_artifact_to_vf.json.j2") data = template.render(artifact_name=artifact_name, artifact_label=f"sdk{Path.PurePosixPath(artifact_name).stem}", artifact_type=artifact_type, b64_artifact=base64.b64encode(artifact).decode('utf-8')) headers = headers_sdc_artifact_upload(base_header=self.headers, data=data) try: self.send_message('POST', 'Add artifact to vf', url, headers=headers, data=data) except RequestError as exc: self._logger.error(("an error occured during file upload for an Artifact" "to VNF %s"), vnf_name) raise exc def get_component_properties_url(self, component: "Component") -> str: """Url to get component's properties. This method is here because component can have different url when it's a component of another SDC resource type, eg. for service and for VF components have different urls. Also for VL origin type components properties url is different than for the other types. Args: component (Component): Component object to prepare url for Returns: str: Component's properties url """ if component.origin_type == "VL": return super().get_component_properties_url(component) return (f"{self.resource_inputs_url}/" f"componentInstances/{component.unique_id}/{component.actual_component_uid}/inputs") def get_component_properties_value_set_url(self, component: "Component") -> str: """Url to set component property value. This method is here because component can have different url when it's a component of another SDC resource type, eg. for service and for VF components have different urls. Also for VL origin type components properties url is different than for the other types. Args: component (Component): Component object to prepare url for Returns: str: Component's properties url """ if component.origin_type == "VL": return super().get_component_properties_value_set_url(component) return (f"{self.resource_inputs_url}/" f"resourceInstance/{component.unique_id}/inputs") def get_category_for_new_resource(self) -> ServiceCategory: """Get category for service not created in SDC yet. If no category values are provided default category is going to be used. Returns: ServiceCategory: Category of the new service """ if not self._category_name: return ServiceCategory.get(name="Network Service") return ServiceCategory.get(name=self._category_name)