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