src/osm_ngsa/dags/ns_topology.py

   1 #######################################################################################
   2 # Copyright ETSI Contributors and Others.
   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
  13 # implied.
  14 # See the License for the specific language governing permissions and
  15 # limitations under the License.
  16 #######################################################################################
  17 from datetime import datetime, timedelta
  18 import logging
  19
  20 from airflow.decorators import dag, task
  21 from osm_mon.core.common_db import CommonDbClient
  22 from osm_mon.core.config import Config
  23 from prometheus_client import CollectorRegistry, Gauge, push_to_gateway
  24
  25
  26 PROMETHEUS_PUSHGW = "pushgateway-prometheus-pushgateway:9091"
  27 PROMETHEUS_JOB = "airflow_osm_ns_topology"
  28 PROMETHEUS_METRIC = "ns_topology"
  29 PROMETHEUS_METRIC_DESCRIPTION = "Network services topology"
  30 SCHEDULE_INTERVAL = 2
  31
  32 # Logging
  33 logger = logging.getLogger("airflow.task")
  34
  35
  36 @dag(
  37     catchup=False,
  38     default_args={
  39         "depends_on_past": False,
  40         "retries": 1,
  41         "retry_delay": timedelta(seconds=10),
  42     },
  43     description="NS topology",
  44     is_paused_upon_creation=False,
  45     # schedule_interval=timedelta(minutes=SCHEDULE_INTERVAL),
  46     schedule_interval=f"*/{SCHEDULE_INTERVAL} * * * *",
  47     start_date=datetime(2022, 1, 1),
  48     tags=["osm", "topology"],
  49 )
  50 def ns_topology():
  51     @task(task_id="get_topology")
  52     def get_topology():
  53         """
  54         Get NS topology from MongoDB and exports it as a metric
  55         to Prometheus
  56         """
  57
  58         # Define Prometheus Metric for NS topology
  59         registry = CollectorRegistry()
  60         metric = Gauge(
  61             PROMETHEUS_METRIC,
  62             PROMETHEUS_METRIC_DESCRIPTION,
  63             labelnames=[
  64                 "ns_id",
  65                 "project_id",
  66                 "vnf_id",
  67                 "vdu_id",
  68                 "vm_id",
  69                 "vim_id",
  70                 "vdu_name",
  71                 "vnf_member_index",
  72             ],
  73             registry=registry,
  74         )
  75
  76         # Getting VNFR list from MongoDB
  77         logger.info("Getting VNFR list from MongoDB")
  78         cfg = Config()
  79         # logger.debug(cfg.conf)
  80         common_db = CommonDbClient(cfg)
  81         vnfr_list = common_db.get_vnfrs()
  82
  83         # Only send topology if ns state is one of the nsAllowedStatesSet
  84         nsAllowedStatesSet = {"INSTANTIATED"}
  85
  86         # For loop to get NS topology.
  87         # For each VDU, a metric sample is created with the appropriate labels
  88         for vnfr in vnfr_list:
  89             # Label ns_id
  90             vnf_id = vnfr["_id"]
  91             # Label ns_id
  92             ns_id = vnfr["nsr-id-ref"]
  93             # Label vnfd_id
  94             vnfd_id = vnfr["vnfd-ref"]
  95             # Label project_id
  96             project_list = vnfr.get("_admin", {}).get("projects_read", [])
  97             project_id = "None"
  98             if project_list:
  99                 project_id = project_list[0]
 100             # Other info
 101             ns_state = vnfr["_admin"]["nsState"]
 102             vnf_membex_index = vnfr["member-vnf-index-ref"]
 103             logger.info(
 104                 f"Read VNFR: id: {vnf_id}, ns_id: {ns_id}, "
 105                 f"state: {ns_state}, vnfd_id: {vnfd_id}, "
 106                 f"vnf_membex_index: {vnf_membex_index}, "
 107                 f"project_id: {project_id}"
 108             )
 109             # Only send topology if ns State is one of the nsAllowedStatesSet
 110             if ns_state not in nsAllowedStatesSet:
 111                 continue
 112
 113             logger.debug("VDU list:")
 114             for vdu in vnfr.get("vdur", []):
 115                 # Label vdu_id
 116                 vdu_id = vdu["_id"]
 117                 # Label vim_id
 118                 vim_info = vdu.get("vim_info")
 119                 if not vim_info:
 120                     logger.info("Error: vim_info not available in vdur")
 121                     continue
 122                 if len(vim_info) != 1:
 123                     logger.info("Error: more than one vim_info in vdur")
 124                     continue
 125                 vim_id = next(iter(vim_info))[4:]
 126                 # TODO: check if it makes sense to use vnfr.vim-account-id as vim_id instead of the vim_info key
 127                 # Label vm_id
 128                 vm_id = vdu.get("vim-id")
 129                 if not vm_id:
 130                     logger.info("Error: vim-id not available in vdur")
 131                     continue
 132                 # Other VDU info
 133                 vdu_name = vdu.get("name", "UNKNOWN")
 134                 logger.debug(
 135                     f"  VDU id: {vdu_id}, name: {vdu_name}, "
 136                     f"vim_id: {vim_id}, vm_id: {vm_id}"
 137                 )
 138                 logger.info(
 139                     f"METRIC SAMPLE: ns_id: {ns_id}, "
 140                     f"project_id: {project_id}, vnf_id: {vnf_id}, "
 141                     f"vdu_id: {vdu_id}, vm_id: {vm_id}, vim_id: {vim_id}"
 142                 )
 143                 metric.labels(
 144                     ns_id,
 145                     project_id,
 146                     vnf_id,
 147                     vdu_id,
 148                     vm_id,
 149                     vim_id,
 150                     vdu_name,
 151                     vnf_membex_index,
 152                 ).set(1)
 153
 154         push_to_gateway(
 155             gateway=PROMETHEUS_PUSHGW, job=PROMETHEUS_JOB, registry=registry
 156         )
 157         return
 158
 159     get_topology()
 160
 161
 162 dag = ns_topology()