tutorials/ApacheCNF/automation/healthcheck.py

   1 # ============LICENSE_START=======================================================
   2 # Copyright (C) 2021 Samsung
   3 # ================================================================================
   4 # Licensed under the Apache License, Version 2.0 (the "License");
   5 # you may not use this file except in compliance with the License.
   6 # You may obtain a copy of the License at
   7 #
   8 #      http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15 #
  16 # ============LICENSE_END=========================================================
  17 import logging
  18 import os
  19 import zipfile
  20 from io import BytesIO
  21
  22 from onapsdk.aai.business import Customer
  23 from onapsdk.cds.blueprint import Workflow, Blueprint
  24 from config import Config, VariablesDict
  25
  26 #FIXME remove from global scope
  27 logger = logging.getLogger("")
  28 logger.setLevel(logging.INFO)
  29 fh = logging.StreamHandler()
  30 fh_formatter = logging.Formatter('%(asctime)s %(levelname)s %(lineno)d:%(filename)s(%(process)d) - %(message)s')
  31 fh.setFormatter(fh_formatter)
  32 logger.addHandler(fh)
  33
  34
  35 def resolve_hc_inputs(config: Config):
  36     logger.info("******** Check Customer *******")
  37     customer_id = config.service_instance["customer_id"]
  38     customer = Customer.get_by_global_customer_id(customer_id)
  39     if customer is None:
  40         raise Exception("Customer %s wasn't found in ONAP" % customer_id)
  41     logger.info("******** Check Service Subscription *******")
  42     service_subscription = None
  43     for service_sub in customer.service_subscriptions:
  44         logger.debug("Service subscription %s is found", service_sub.service_type)
  45         if service_sub.service_type == config.service_model["model_name"]:
  46             logger.info("Service %s subscribed", config.service_model["model_name"])
  47             service_subscription = service_sub
  48             break
  49     logger.info("******** Retrieve Service Metadata *******")
  50     service_instance = None
  51     for single_service in service_subscription.service_instances:
  52         if single_service.instance_name == config.service_instance["instance_name"]:
  53             service_instance = single_service
  54             break
  55     service_id = service_instance.instance_id
  56     vnfs = list(service_instance.vnf_instances)
  57     if len(vnfs) > 1:
  58         raise NotImplementedError("Service %s is composed of more than one vnf!" % service_id)
  59     if not vnfs:
  60         raise Exception("Service %s doesn't contain any vnfs" % service_id)
  61     vnf_id = vnfs[0].vnf_id
  62     return service_id, vnf_id
  63
  64 def main():
  65     mypath = os.path.dirname(os.path.realpath(__file__))
  66     config = Config(env_dict=VariablesDict.env_variable)
  67     for vnf in config.service_model["vnfs"]:
  68         file = vnf["vsp"]["vsp_file"]
  69         file_path = os.path.join(mypath, file)
  70         with zipfile.ZipFile(file_path, 'r') as package:
  71             cba_io = BytesIO(package.read("CBA.zip"))
  72             cba_io.seek(0)
  73             blueprint = Blueprint(cba_io.read())
  74
  75             healthcheck: Workflow = blueprint.get_workflow_by_name('health-check')
  76             serv_id, vnf_id = resolve_hc_inputs(config)
  77             cds_input = {"health-check-properties":
  78                 {
  79                     "service-instance-id": serv_id,
  80                     "vnf-id": vnf_id
  81                 }
  82             }
  83
  84             logger.info("Requesting Healthcheck for CBA %s:%s with inputs:\n%s",
  85                     blueprint.metadata.template_name,
  86                     blueprint.metadata.template_version,
  87                     cds_input)
  88             result = healthcheck.execute(cds_input)
  89             logger.info("Healthcheck process completed with result: %s", result)
  90             logger.info("Please check cds-blueprints-processor logs to see exact status")
  91
  92 if __name__ == "__main__":
  93     main()