src/osm_ngsa/dags/multivim_vim_status.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 import DAG
  21 from airflow.decorators import task
  22 from osm_mon.core.common_db import CommonDbClient
  23 from osm_mon.core.config import Config
  24 from osm_mon.vim_connectors.azure import AzureCollector
  25 from osm_mon.vim_connectors.gcp import GcpCollector
  26 from osm_mon.vim_connectors.openstack import OpenStackCollector
  27 from prometheus_client import CollectorRegistry, Gauge, push_to_gateway
  28
  29
  30 SUPPORTED_VIM_TYPES = ["openstack", "vio", "gcp", "azure"]
  31 PROMETHEUS_PUSHGW = "pushgateway-prometheus-pushgateway:9091"
  32 PROMETHEUS_JOB_PREFIX = "airflow_osm_vim_status_"
  33 PROMETHEUS_METRIC = "osm_vim_status"
  34 PROMETHEUS_METRIC_DESCRIPTION = "VIM status"
  35 SCHEDULE_INTERVAL = 1
  36
  37 # Logging
  38 logger = logging.getLogger("airflow.task")
  39
  40
  41 def get_all_vim():
  42     """Get VIMs from MongoDB"""
  43     logger.info("Getting VIM list")
  44
  45     cfg = Config()
  46     logger.info(cfg.conf)
  47     common_db = CommonDbClient(cfg)
  48     vim_accounts = common_db.get_vim_accounts()
  49     vim_list = []
  50     for vim in vim_accounts:
  51         logger.info(f'Read VIM {vim["_id"]} ({vim["name"]})')
  52         vim_list.append(
  53             {"_id": vim["_id"], "name": vim["name"], "vim_type": vim["vim_type"]}
  54         )
  55
  56     logger.info(vim_list)
  57     logger.info("Getting VIM list OK")
  58     return vim_list
  59
  60
  61 def create_dag(dag_id, dag_number, dag_description, vim_id):
  62     dag = DAG(
  63         dag_id,
  64         catchup=False,
  65         default_args={
  66             "depends_on_past": False,
  67             "retries": 1,
  68             # "retry_delay": timedelta(minutes=1),
  69             "retry_delay": timedelta(seconds=10),
  70         },
  71         description=dag_description,
  72         is_paused_upon_creation=False,
  73         # schedule_interval=timedelta(minutes=SCHEDULE_INTERVAL),
  74         schedule_interval=f"*/{SCHEDULE_INTERVAL} * * * *",
  75         start_date=datetime(2022, 1, 1),
  76         tags=["osm", "vim"],
  77     )
  78
  79     with dag:
  80
  81         def get_vim_collector(vim_account):
  82             """Return a VIM collector for the vim_account"""
  83             vim_type = vim_account["vim_type"]
  84             if "config" in vim_account and "vim_type" in vim_account["config"]:
  85                 vim_type = vim_account["config"]["vim_type"].lower()
  86                 if vim_type == "vio" and "vrops_site" not in vim_account["config"]:
  87                     vim_type = "openstack"
  88             if vim_type == "openstack" or vim_type == "vio":
  89                 return OpenStackCollector(vim_account)
  90             if vim_type == "gcp":
  91                 return GcpCollector(vim_account)
  92             if vim_type == "azure":
  93                 return AzureCollector(vim_account)
  94             logger.info(f"VIM type '{vim_type}' not supported")
  95             return None
  96
  97         @task(task_id="get_vim_status_and_send_to_prometheus")
  98         def get_vim_status_and_send_to_prometheus(vim_id: str):
  99             """Authenticate against VIM and check status"""
 100
 101             # Get VIM account info from MongoDB
 102             logger.info(f"Reading VIM info, id: {vim_id}")
 103             cfg = Config()
 104             common_db = CommonDbClient(cfg)
 105             vim_account = common_db.get_vim_account(vim_account_id=vim_id)
 106             logger.info(vim_account)
 107
 108             # Define Prometheus Metric for NS topology
 109             registry = CollectorRegistry()
 110             metric = Gauge(
 111                 PROMETHEUS_METRIC,
 112                 PROMETHEUS_METRIC_DESCRIPTION,
 113                 labelnames=[
 114                     "vim_account_id",
 115                 ],
 116                 registry=registry,
 117             )
 118             metric.labels(vim_id).set(0)
 119
 120             # Get status of VIM
 121             collector = get_vim_collector(vim_account)
 122             if collector:
 123                 status = collector.is_vim_ok()
 124                 logger.info(f"VIM status: {status}")
 125                 if status:
 126                     metric.labels(vim_id).set(1)
 127             else:
 128                 logger.info("Error creating VIM collector")
 129             # Push to Prometheus
 130             push_to_gateway(
 131                 gateway=PROMETHEUS_PUSHGW,
 132                 job=f"{PROMETHEUS_JOB_PREFIX}{vim_id}",
 133                 registry=registry,
 134             )
 135             return
 136
 137         get_vim_status_and_send_to_prometheus(vim_id)
 138
 139     return dag
 140
 141
 142 vim_list = get_all_vim()
 143 for index, vim in enumerate(vim_list):
 144     vim_type = vim["vim_type"]
 145     if vim_type in SUPPORTED_VIM_TYPES:
 146         vim_id = vim["_id"]
 147         vim_name = vim["name"]
 148         dag_description = f"Dag for VIM {vim_name} status"
 149         dag_id = f"vim_status_{vim_id}"
 150         logger.info(f"Creating DAG {dag_id}")
 151         globals()[dag_id] = create_dag(
 152             dag_id=dag_id,
 153             dag_number=index,
 154             dag_description=dag_description,
 155             vim_id=vim_id,
 156         )
 157     else:
 158         logger.info(f"VIM type '{vim_type}' not supported for monitoring VIM status")