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     max_active_runs=1,
  46     # schedule_interval=timedelta(minutes=SCHEDULE_INTERVAL),
  47     schedule_interval=f"*/{SCHEDULE_INTERVAL} * * * *",
  48     start_date=datetime(2022, 1, 1),
  49     tags=["osm", "topology"],
  50 )
  51 def ns_topology():
  52     @task(task_id="get_topology")
  53     def get_topology():
  54         """
  55         Get NS topology from MongoDB and exports it as a metric
  56         to Prometheus
  57         """
  58
  59         # Define Prometheus Metric for NS topology
  60         registry = CollectorRegistry()
  61         metric = Gauge(
  62             PROMETHEUS_METRIC,
  63             PROMETHEUS_METRIC_DESCRIPTION,
  64             labelnames=[
  65                 "ns_id",
  66                 "project_id",
  67                 "vnf_id",
  68                 "vdu_id",
  69                 "vm_id",
  70                 "vim_id",
  71                 "vdu_name",
  72                 "vnf_member_index",
  73             ],
  74             registry=registry,
  75         )
  76
  77         # Getting VNFR list from MongoDB
  78         logger.info("Getting VNFR list from MongoDB")
  79         cfg = Config()
  80         # logger.debug(cfg.conf)
  81         common_db = CommonDbClient(cfg)
  82         vnfr_list = common_db.get_vnfrs()
  83
  84         # Only send topology if ns state is one of the nsAllowedStatesSet
  85         nsAllowedStatesSet = {"INSTANTIATED"}
  86
  87         # For loop to get NS topology.
  88         # For each VDU, a metric sample is created with the appropriate labels
  89         for vnfr in vnfr_list:
  90             # Label ns_id
  91             vnf_id = vnfr["_id"]
  92             # Label ns_id
  93             ns_id = vnfr["nsr-id-ref"]
  94             # Label vnfd_id
  95             vnfd_id = vnfr["vnfd-ref"]
  96             # Label project_id
  97             project_list = vnfr.get("_admin", {}).get("projects_read", [])
  98             project_id = "None"
  99             if project_list:
 100                 project_id = project_list[0]
 101             # Other info
 102             ns_state = vnfr["_admin"]["nsState"]
 103             vnf_membex_index = vnfr["member-vnf-index-ref"]
 104             logger.info(
 105                 f"Read VNFR: id: {vnf_id}, ns_id: {ns_id}, "
 106                 f"state: {ns_state}, vnfd_id: {vnfd_id}, "
 107                 f"vnf_membex_index: {vnf_membex_index}, "
 108                 f"project_id: {project_id}"
 109             )
 110             # Only send topology if ns State is one of the nsAllowedStatesSet
 111             if ns_state not in nsAllowedStatesSet:
 112                 continue
 113
 114             logger.debug("VDU list:")
 115             for vdu in vnfr.get("vdur", []):
 116                 # Label vdu_id
 117                 vdu_id = vdu["_id"]
 118                 # Label vim_id
 119                 vim_info = vdu.get("vim_info")
 120                 if not vim_info:
 121                     logger.info("Error: vim_info not available in vdur")
 122                     continue
 123                 if len(vim_info) != 1:
 124                     logger.info("Error: more than one vim_info in vdur")
 125                     continue
 126                 vim_id = next(iter(vim_info))[4:]
 127                 # TODO: check if it makes sense to use vnfr.vim-account-id as vim_id instead of the vim_info key
 128                 # Label vm_id
 129                 vm_id = vdu.get("vim-id")
 130                 if not vm_id:
 131                     logger.info("Error: vim-id not available in vdur")
 132                     continue
 133                 # Other VDU info
 134                 vdu_name = vdu.get("name", "UNKNOWN")
 135                 logger.debug(
 136                     f"  VDU id: {vdu_id}, name: {vdu_name}, "
 137                     f"vim_id: {vim_id}, vm_id: {vm_id}"
 138                 )
 139                 logger.info(
 140                     f"METRIC SAMPLE: ns_id: {ns_id}, "
 141                     f"project_id: {project_id}, vnf_id: {vnf_id}, "
 142                     f"vdu_id: {vdu_id}, vm_id: {vm_id}, vim_id: {vim_id}"
 143                 )
 144                 metric.labels(
 145                     ns_id,
 146                     project_id,
 147                     vnf_id,
 148                     vdu_id,
 149                     vm_id,
 150                     vim_id,
 151                     vdu_name,
 152                     vnf_membex_index,
 153                 ).set(1)
 154
 155         push_to_gateway(
 156             gateway=PROMETHEUS_PUSHGW, job=PROMETHEUS_JOB, registry=registry
 157         )
 158         return
 159
 160     get_topology()
 161
 162
 163 dag = ns_topology()